Generate matched scalar controls for coupled trial attribution

experiments/make_coupled_scalar_controls.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Generate projected scalar controls from completed frozen coupled-trial cases.
+
+This performs saved-vector algebra only. Replaying x+t*p through the existing
+full operator distinguishes scalar step selection from transverse SOC benefit.
+"""
+import argparse
+import csv
+import datetime
+import hashlib
+import json
+import math
+from pathlib import Path
+import sys
+
+PREFIX = 'coupled_trial_'
+SHARED = ('base_normalized.txt', 'base_physical.txt', 'base_residual.txt', 'direction.txt',
+          'state_factors.txt', 'residual_factors.txt', 'layout.csv')
+
+
+def digest(path):
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def vector(path):
+    words = path.read_text().split()
+    values = [float(word) for word in words[1:]]
+    if not words or int(words[0]) != len(values) or not values or not all(map(math.isfinite, values)):
+        raise ValueError(f'Invalid finite vector: {path}')
+    return values
+
+
+def rows(path):
+    with path.open() as stream:
+        return list(csv.DictReader(stream))
+
+
+def norm(values):
+    return math.sqrt(math.fsum(value * value for value in values))
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('screen', type=Path)
+    parser.add_argument('output', type=Path)
+    args = parser.parse_args()
+    screen, output = args.screen.resolve(), args.output.resolve()
+    if output.exists() or screen in output.parents:
+        parser.error('Use a fresh output directory outside the immutable source screen.')
+    source_provenance = screen / 'provenance.json'
+    campaign = json.loads(source_provenance.read_text())
+    input_hashes = {str(source_provenance): digest(source_provenance)}
+    expected_shared = None
+    generated, cases, observations = [], [], []
+    for source_case, label in (('late_raw_coupled', 'raw'), ('late_curvature_coupled', 'gravity')):
+        folder, record_path = screen / source_case, screen / (source_case + '_run.json')
+        record = json.loads(record_path.read_text())
+        if record.get('exit_code') != 0 or not record.get('finished_utc') or record.get('timed_out', False):
+            raise ValueError(f'Source case has not completed successfully: {source_case}')
+        input_hashes[str(record_path)] = digest(record_path)
+        names = [PREFIX + name for name in SHARED] + ['metadata.txt'] + [
+            PREFIX + name for name in ('summary.csv', 'trials.csv', 'selected_state_a2.txt')]
+        for name in names:
+            path = folder / name
+            value = digest(path)
+            if record.get('output_hashes', {}).get(name) != value:
+                raise ValueError(f'Source artifact does not match its completed-run hash: {path}')
+            input_hashes[str(path)] = value
+        shared = {name: input_hashes[str(folder / (PREFIX + name))] for name in SHARED}
+        if expected_shared is not None and shared != expected_shared:
+            raise ValueError('Source cases do not share exactly the same base, direction, normalization and layout.')
+        expected_shared = shared
+        meta = dict(line.split('=', 1) for line in (folder / 'metadata.txt').read_text().splitlines() if '=' in line)
+        if float(meta['n']) != 1.5 or float(meta['J']) != 0. or meta['central_phase'] != '0':
+            raise ValueError('These controls require the saved n=1.5, J=0 model without a central-density constraint.')
+        summary = {row['metric']: float(row['value']) for row in rows(folder / (PREFIX + 'summary.csv'))}
+        if summary['probe_exception'] != 0 or summary['outer_verified_gate'] != 1:
+            raise ValueError('Source probe did not complete with a verified outer Newton direction.')
+        matches = [row for row in rows(folder / (PREFIX + 'trials.csv')) if row['sample'] == '2']
+        if len(matches) != 1 or float(matches[0]['alpha']) != 1 / 256 or matches[0]['selected_accepted'] != '1' or \
+                int(matches[0]['soc_status']) != int(summary['soc_status_kept']):
+            raise ValueError('Expected a kept, Armijo-accepted coupled correction at sample 2, alpha=1/256.')
+        x, p, selected = [vector(folder / (PREFIX + name)) for name in
+                          ('base_normalized.txt', 'direction.txt', 'selected_state_a2.txt')]
+        if not len(x) == len(p) == len(selected):
+            raise ValueError('Source vectors have incompatible dimensions.')
+        displacement = [y - b for y, b in zip(selected, x)]
+        pp = math.fsum(value * value for value in p)
+        if not math.isfinite(pp) or pp <= 0.:
+            raise ValueError('Projection requires a finite nonzero direction norm.')
+        t = math.fsum(step * direction for step, direction in zip(displacement, p)) / pp
+        candidate = [math.fsum((b, t * direction)) for b, direction in zip(x, p)]
+        perpendicular = [math.fsum((step, -t * direction)) for step, direction in zip(displacement, p)]
+        if not math.isfinite(t) or t <= 0. or not all(map(math.isfinite, candidate)):
+            raise ValueError('Projected control is nonfinite or does not advance along the supplied direction.')
+        candidate_path = output / (label + '_projected_normalized.txt')
+        generated.append((candidate_path, candidate))
+        cases.append({'id': 'scalar_match_' + label + '_soc', 'arguments': [
+            '--index', '1.5', '--angular-momentum', '0', '--no-central', '--max-newton', '0',
+            '--strategy', 'baseline', '--preconditioner-variant', 'independent',
+            '--probe-state', str(folder / (PREFIX + 'base_physical.txt')),
+            '--replay-candidate', str(candidate_path)]})
+        observations.append({'source_case': source_case, 'sample': 2, 'nominal_alpha': 1 / 256, 't': t,
+            'selected_displacement_norm': norm(displacement), 'scalar_displacement_norm': t * math.sqrt(pp),
+            'perpendicular_norm': norm(perpendicular), 'perpendicular_dot_direction':
+                math.fsum(value * direction for value, direction in zip(perpendicular, p)),
+            'source_selected_residual_norm': float(matches[0]['selected_norm']), 'candidate': str(candidate_path)})
+
+    # Validate everything before creating output; never replace campaign inputs.
+    output.mkdir(parents=True)
+    for path, values in generated:
+        path.write_text(str(len(values)) + '\n' + ''.join(format(value, '.17g') + '\n' for value in values))
+    (output / 'cases.json').write_text(json.dumps(cases, indent=2, allow_nan=False) + '\n')
+    generator = Path(__file__).resolve()
+    (output / 'generator.py').write_bytes(generator.read_bytes())
+    provenance = {'created_utc': datetime.datetime.now(datetime.timezone.utc).isoformat(),
+        'generator': str(generator), 'generator_sha256': digest(generator), 'python': sys.version,
+        'source_screen': str(screen), 'source_commit': campaign['commit'], 'input_hashes': input_hashes,
+        'output_hashes': {path.name: digest(path) for path in output.iterdir()}, 'controls': observations,
+        'scope': 'Projection controls only; no FEM evaluations, acceptance conclusions, or linear solves performed.'}
+    (output / 'generator_provenance.json').write_text(json.dumps(provenance, indent=2, allow_nan=False) + '\n')
+    print(output / 'cases.json')
+
+
+if __name__ == '__main__':
+    main()