tboudreaux/MeanField
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Independently audit frozen nonlinear path vectors and provenance

Browse Source
This commit is contained in:
Emily Boudreaux
2026-09-18 08:37:54 -04:00
parent ba057b3983
commit b8d1c6a5ab
2 changed files with 751 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

499
experiments/check_nonlinear_step.py Normal file
View File

@@ -0,0 +1,499 @@
#!/usr/bin/env python3
"""Audit frozen nonlinear paths using saved vectors, not scalar curvature fits.
Artifact validity, independently verified linear accuracy, observed nonlinear
improvement, and physical validity are different conclusions. Remainder floors
are heuristic estimates, never certified FEM assembly bounds.
"""
import argparse
import csv
import hashlib
import json
import math
from pathlib import Path
import sys
P = 'nonlinear_step_'
DIRECTIONS = ('full', 'shape', 'fields')
PATHS = ('raw', 'curvature')
EPS = sys.float_info.epsilon
GR = {'gravity_gradient_relation', 'poisson_balance'}
GV = {'gravity_gradient', 'gravity_potential'}
BASE_FILES = {'metadata.txt', P + 'layout.csv'} | {P + n + '.csv' for n in ('summary', 'linear', 'paths', 'blocks', 'mixed', 'preparations', 'inner', 'restorations')} | {
P + n + '.txt' for n in ('physical_base', 'normalized_base', 'residual', 'rhs', 'state_factors', 'residual_factors', 'restored_residual')
} | {P + k + '_' + d + '.txt' for k in ('direction', 'action', 'restored_action') for d in DIRECTIONS}
def digest(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
def rows(path):
with path.open() as stream:
return list(csv.DictReader(stream))
def vector(path):
tokens = path.read_text().split()
count = int(tokens[0])
values = [float(t) for t in tokens[1:]]
if count <= 0 or count != len(values) or not all(math.isfinite(v) for v in values):
raise ValueError(f'Invalid finite vector: {path}')
return values
def norm(v):
return math.sqrt(math.fsum(x * x for x in v))
def dot(a, b):
if len(a) != len(b): raise ValueError('Incompatible vectors')
return math.fsum(x * y for x, y in zip(a, b))
def minus(a, b):
if len(a) != len(b): raise ValueError('Incompatible vectors')
return [x - y for x, y in zip(a, b)]
def close(a, b, scale=0.0):
return math.isfinite(a) and math.isfinite(b) and abs(a - b) <= 5e-10 * max(abs(a), abs(b)) + 512 * EPS * abs(scale) + 1e-28
def nonnegative(x):
return math.isfinite(x) and x >= 0
def sanitize(value):
if isinstance(value, float) and not math.isfinite(value): return None
if isinstance(value, dict): return {k: sanitize(v) for k, v in value.items()}
if isinstance(value, (list, tuple)): return [sanitize(v) for v in value]
return value
def flag(row, key, check, prefix):
check(prefix + key + ':boolean', row[key] in ('0', '1'))
return row[key] == '1'
def layout_for(layout, kind, count):
result, covered = {}, []
entries = [r for r in layout if r['kind'] == kind]
for row in entries:
start, size = int(row['offset']), int(row['size'])
if start < 0 or size < 0: raise ValueError('Invalid layout')
result[row['block']] = (start, size)
covered.extend(range(start, start + size))
if len(result) != len(entries) or covered != list(range(count)):
raise ValueError(f'{kind} blocks do not partition the vector')
return result
def indices(layout, names):
return sorted(i for name in names for i in range(layout[name][0], sum(layout[name])))
def measurements(base, action, trial, alpha):
prediction = [f + alpha * j for f, j in zip(base, action)]
remainder = minus(trial, prediction)
fn, jn, ln, tn, rn = map(norm, (base, action, prediction, trial, remainder))
projection = dot(prediction, remainder)
identity = abs(tn * tn - ln * ln - 2 * projection - rn * rn)
return {'base_norm': fn, 'action_norm': jn, 'linear_prediction_norm': ln, 'residual_norm': tn,
'remainder_norm': rn, 'remainder_over_alpha': rn / alpha, 'remainder_over_alpha_squared': rn / alpha ** 2,
'linear_dot_remainder': projection, 'norm_identity_error': identity}, prediction, remainder
def check_measurements(row, values, check, prefix):
a = float(row['alpha'])
scale = values['base_norm'] + a * values['action_norm'] + values['residual_norm']
for name, value in values.items():
reported = float(row[name])
if name == 'norm_identity_error':
check(prefix + name, math.isfinite(reported) and max(abs(reported), value) <= 2048 * EPS * scale ** 2 + 1e-28)
else:
rounding_scale = scale / a if name == 'remainder_over_alpha' else scale / a ** 2 if name == 'remainder_over_alpha_squared' else scale ** 2 if name == 'linear_dot_remainder' else scale
check(prefix + name, close(value, reported, rounding_scale))
def vector_orders(observations, linear_passed):
groups = {}
for row in observations: groups.setdefault((row['path'], row['direction'], row['block']), []).append(row)
output = []
for key, entries in groups.items():
entries.sort(key=lambda r: r['alpha'], reverse=True)
intervals = []
for first, second in zip(entries, entries[1:]):
resolved = all(r['usable'] and r['increment_resolved'] and r['remainder_norm'] > 10 * r['floor'] for r in (first, second))
if resolved:
intervals.append({'alpha_large': first['alpha'], 'alpha_small': second['alpha'],
'vector_remainder_order': math.log(first['remainder_norm'] / second['remainder_norm']) / math.log(first['alpha'] / second['alpha']),
'derivative_error_ratio': second['remainder_over_alpha'] / first['remainder_over_alpha']})
output.append(dict(zip(('path', 'direction', 'block'), key)) | {'resolved_intervals': intervals,
'status': 'resolved_vector_observations' if intervals else 'unresolved',
'eligible_for_newton_forcing_comparison': linear_passed})
return output
def inspect_case(folder):
try: return inspect_case_artifacts(Path(folder))
except (OSError, ValueError, KeyError, TypeError, IndexError, OverflowError, ZeroDivisionError) as error:
return {'passed': False, 'checks': [{'check': 'complete_readable_artifacts', 'passed': False, 'error': str(error)}],
'observations_valid': False, 'vector_order_observations': [], 'required_artifacts': sorted(BASE_FILES)}
def inspect_case_artifacts(folder):
checks = []
def check(name, passed): checks.append({'check': name, 'passed': bool(passed)})
summary_rows = rows(folder / (P + 'summary.csv'))
summary = {r['metric']: float(r['value']) for r in summary_rows}
check('unique_summary', len(summary) == len(summary_rows))
meta = dict(line.split('=', 1) for line in (folder / 'metadata.txt').read_text().splitlines() if '=' in line)
saved = {name: vector(folder / (P + name + '.txt')) for name in ('physical_base', 'normalized_base', 'residual', 'rhs', 'state_factors', 'residual_factors')}
x, physical, f = saved['normalized_base'], saved['physical_base'], saved['residual']
sf, rf = saved['state_factors'], saved['residual_factors']
check('state_dimensions', len(x) == len(physical) == len(sf))
check('residual_dimensions', len(f) == len(saved['rhs']) == len(rf))
check('positive_factors', all(v > 0 for v in sf + rf))
check('normalized_physical_consistency', all(close(a, b * c, abs(a)) for a, b, c in zip(x, physical, sf)))
check('rhs_is_minus_F', saved['rhs'] == [-v for v in f])
layout = rows(folder / (P + 'layout.csv'))
vl, rl = layout_for(layout, 'value', len(x)), layout_for(layout, 'residual', len(f))
gi, gr, si = indices(vl, GV), indices(rl, GR), set(indices(vl, {'surface_deformation'}))
retained = sorted(set(range(len(x))) - set(gi))
dirs = {d: vector(folder / (P + 'direction_' + d + '.txt')) for d in DIRECTIONS}
acts = {d: vector(folder / (P + 'action_' + d + '.txt')) for d in DIRECTIONS}
for d in DIRECTIONS: check(d + ':dimensions', len(dirs[d]) == len(x) and len(acts[d]) == len(f))
check('shape_split_exact', dirs['shape'] == [v if i in si else 0.0 for i, v in enumerate(dirs['full'])])
check('fields_split_exact', dirs['fields'] == [0.0 if i in si else v for i, v in enumerate(dirs['full'])])
split_error = norm([a - b - c for a, b, c in zip(acts['full'], acts['shape'], acts['fields'])])
check('action_split', split_error <= 512 * EPS * sum(norm(v) for v in acts.values()) + 1e-28)
lr = rows(folder / (P + 'linear.csv'))
check('one_linear_report', len(lr) == 1)
linear = lr[0]
rhs_norm, defect = norm(saved['rhs']), norm([a + b for a, b in zip(f, acts['full'])])
target = max(summary['linear_absolute_tolerance'], float(meta['linear_tol']) * rhs_norm)
check('linear_control_matches_metadata', summary['linear_relative_tolerance'] == float(meta['linear_tol']) and summary['linear_maximum_iterations'] == int(meta['max_linear']) and summary['linear_restart_length'] == int(meta['restart']))
gate = defect <= target
for key, value in [('rhs_norm', rhs_norm), ('verified_true_norm', defect), ('verified_relative_true', defect / rhs_norm), ('threshold', target)]:
check('linear:' + key, close(value, float(linear[key]), 1.0 if key == 'verified_relative_true' else rhs_norm))
check('summary_linear_gate', summary['linear_verified_gate'] == int(gate))
check('honest_linear_gate', flag(linear, 'verified_gate', check, 'linear:') == gate)
converged = flag(linear, 'converged', check, 'linear:')
check('linear_status', converged == (linear['status'] == '0'))
check('linear_finite_reported_defect', nonnegative(float(linear['reported_true_norm'])))
check('linear_budget', 0 <= int(linear['iterations']) <= int(meta['max_linear']))
check('linear_time', nonnegative(float(linear['seconds'])))
path_rows, block_rows, mixed_rows = (rows(folder / (P + n + '.csv')) for n in ('paths', 'blocks', 'mixed'))
pm = {(int(r['sample']), r['direction'], r['path']): r for r in path_rows}
bm = {(int(r['sample']), r['direction'], r['path'], r['block']): r for r in block_rows}
mm = {(int(r['sample']), r['path'], r['block']): r for r in mixed_rows}
check('unique_paths', len(pm) == len(path_rows)); check('unique_blocks', len(bm) == len(block_rows)); check('unique_mixed', len(mm) == len(mixed_rows))
samples = sorted({key[0] for key in pm})
check('eight_samples', samples == list(range(8)))
check('complete_path_grid', set(pm) == {(i, d, p) for i in samples for d in DIRECTIONS for p in PATHS})
layouts = {'all': (0, len(f)), **rl}
check('complete_block_grid', set(bm) == {(i,d,p,b) for i in samples for d in DIRECTIONS for p in PATHS for b in layouts})
check('complete_mixed_grid', set(mm) == {(i,p,b) for i in samples for p in PATHS for b in layouts})
restoration_error = validate_restoration(folder, summary, f, acts, layouts, check)
inner_rows = rows(folder / (P + 'inner.csv'))
inner_map = {(int(r['sample']), r['direction']): r for r in inner_rows}
check('unique_inner_reports', len(inner_map) == len(inner_rows))
check('inner_coverage', set(inner_map) == {(i,d) for i in samples for d in DIRECTIONS if pm[(i,d,'raw')]['prepared'] == '1'})
validate_work(folder, summary, linear, inner_rows, path_rows, check)
cancellation = []
for block, (start, size) in layouts.items():
an = {d:norm(acts[d][start:start+size]) for d in DIRECTIONS}
denominator = an['shape'] * an['fields']
cancellation.append({'block':block, 'action_norms':an,
'shape_fields_cosine':dot(acts['shape'][start:start+size],acts['fields'][start:start+size])/denominator if denominator else None,
'full_over_sum_split_norms':an['full']/(an['shape']+an['fields']) if an['shape']+an['fields'] else None})
required, observations, outcomes = set(BASE_FILES), [], []
for sample in samples:
alphas = {float(pm[(sample, d, p)]['alpha']) for d in DIRECTIONS for p in PATHS}
check(f'{sample}:common_positive_alpha', len(alphas) == 1 and all(math.isfinite(a) and a > 0 for a in alphas))
a, data = next(iter(alphas)), {}
check(f'{sample}:recorded_alpha', a == summary['alpha_' + str(sample)])
for d in DIRECTIONS:
pred_name = P + 'prediction_' + d + '_a' + str(sample) + '.txt'
required.add(pred_name)
saved_prediction = vector(folder / pred_name)
check(f'{sample}:{d}:prediction_vector', len(saved_prediction) == len(f) and all(close(v,f0+a*j,abs(f0)+a*abs(j)) for v,f0,j in zip(saved_prediction,f,acts[d])))
for path in PATHS:
row, label = pm[(sample, d, path)], f'{sample}:{d}:{path}:'
flags = {k: flag(row, k, check, label) for k in ('prepared', 'inner_converged', 'prepared_gate', 'usable', 'increment_resolved')}
check(label + 'usable_definition', flags['usable'] == (flags['prepared'] and flags['inner_converged'] and flags['prepared_gate']))
check(label + 'seconds', nonnegative(float(row['seconds'])))
correction = [0.0] * len(x)
if path == 'curvature' and flags['inner_converged']:
correction_name = P + 'correction_curvature_' + d + '_a' + str(sample) + '.txt'
required.add(correction_name); correction = vector(folder / correction_name)
check(label + 'correction_only_gravity', len(correction) == len(x) and all(correction[i] == 0 for i in retained))
if path == 'curvature' and (sample,d) in inner_map:
ir = inner_map[(sample,d)]
inner_converged = ir['converged'] == '1'
check(label + 'inner_matches', flags['inner_converged'] == inner_converged and inner_converged == (ir['status'] == '0') and float(ir['alpha']) == a)
inner_target = max(summary['gravity_absolute_tolerance'],summary['gravity_relative_tolerance']*float(ir['rhs_norm']))
check(label + 'inner_target', close(float(ir['threshold']),inner_target,inner_target))
if inner_converged: check(label + 'inner_gate', nonnegative(float(ir['true_norm'])) and float(ir['true_norm']) <= inner_target)
if path == 'raw': check(label + 'vacuous_flags', flags['inner_converged'] and flags['prepared_gate'])
if not flags['prepared']:
check(label + 'not_usable', not flags['usable'])
check(label + 'unprepared_metrics_unavailable', all(math.isnan(float(row[k])) for k in ('residual_norm','merit','remainder_norm','linear_dot_remainder','norm_identity_error','heuristic_roundoff_scale')))
for block in layouts:
br = bm[(sample,d,path,block)]
check(label + block + ':unprepared_unavailable', all(math.isnan(float(br[k])) for k in ('residual_norm','remainder_norm','linear_dot_remainder','norm_identity_error')))
outcomes.append({'sample':sample,'alpha':a,'direction':d,'path':path,**flags,
'status':'raw_preparation_failed' if path=='raw' else 'raw_preparation_blocked' if pm[(sample,d,'raw')]['prepared']=='0' else 'inner_solve_failed' if not flags['inner_converged'] else 'corrected_preparation_failed'})
data[(d, path)] = None; continue
files = {k: P + k + '_' + path + '_' + d + '_a' + str(sample) + '.txt' for k in ('state', 'residual')}
required.update(files.values())
state, trial = vector(folder / files['state']), vector(folder / files['residual'])
check(label + 'dimensions', len(state) == len(x) and len(trial) == len(f))
values, prediction, remainder = measurements(f, acts[d], trial, a)
check_measurements(row, values, check, label)
check(label + 'merit', close(.5 * values['residual_norm'] ** 2, float(row['merit']), values['residual_norm'] ** 2))
check(label + 'slope', close(dot(f, acts[d]), float(row['slope']), norm(f) * norm(acts[d])))
increment = minus(state, x)
intended = [a * v + delta for v,delta in zip(dirs[d],correction)]
addition_error = norm(minus(increment,intended))
expected_norm = norm(intended)
check(label + 'expected_step_norm', close(expected_norm,float(row['expected_step_norm']),expected_norm))
check(label + 'addition_error_norm', close(addition_error,float(row['addition_error_norm']),norm(x)))
resolved = expected_norm > 64*EPS*(1+norm(x)) and norm(increment)>0 and addition_error <= .01*expected_norm
check(label + 'honest_increment_resolution', flags['increment_resolved'] == resolved)
derivative = values['remainder_over_alpha']/values['action_norm'] if values['action_norm']>0 else math.nan
check(label + 'relative_derivative_error', close(derivative,float(row['relative_derivative_error']),derivative) if math.isfinite(derivative) else math.isnan(float(row['relative_derivative_error'])))
check(label + 'actual_step_norm', close(norm(increment), float(row['actual_step_norm']), norm(x)))
if path == 'raw':
intended = [a * v for v in dirs[d]]
error = norm(minus(increment, intended))
check(label + 'raw_state_path', error <= 512 * EPS * (norm(x) + norm(intended)) + 1e-28)
check(label + 'vacuous_gravity_flags', flags['inner_converged'] and flags['prepared_gate'])
else:
raw = data[(d, 'raw')]
check(label + 'prepared_raw_required', raw is not None)
delta = minus(state, raw['state'])
check(label + 'saved_correction_matches_state', norm(minus(delta,correction)) <= 512*EPS*(norm(state)+norm(correction))+1e-28)
check(label + 'retained_coordinates_unchanged', all(delta[i] == 0 for i in retained))
check(label + 'gravity_correction_norm', close(norm(delta), float(row['gravity_correction_norm']), norm(state)))
target_defect = norm(remainder[i] for i in gr)
check(label + 'gravity_affine_target_defect', close(target_defect, float(row['gravity_target_defect']), values['base_norm'] + a * values['action_norm']))
allowance, threshold = float(row['gravity_arithmetic_allowance']), float(row['gravity_target_threshold'])
check(label + 'threshold_is_inner_defect_plus_allowance', close(threshold,float(inner_map[(sample,d)]['true_norm'])+allowance,threshold))
check(label + 'gravity_thresholds', nonnegative(allowance) and nonnegative(threshold) and threshold >= allowance)
if flags['prepared_gate']:
check(label + 'gravity_target_gate', target_defect <= threshold + 512 * EPS * values['base_norm'])
check(label + 'gravity_affine_gate', nonnegative(float(row['gravity_affine_disagreement'])) and float(row['gravity_affine_disagreement']) <= allowance)
data[(d, path)] = {'state': state, 'trial': trial, 'prediction': prediction, 'remainder': remainder, 'flags': flags}
outcomes.append({'sample': sample, 'alpha': a, 'direction': d, 'path': path, **flags, 'status':'usable' if flags['usable'] else 'prepared_gravity_gate_failed',
'actual_gravity_residual_norm':norm(trial[i] for i in gr),
'gravity_affine_target_defect':float(row['gravity_target_defect']) if path=='curvature' else None,
'actual_norm': values['residual_norm'], 'predicted_norm': values['linear_prediction_norm'],
'actual_merit_decrease': .5 * (values['base_norm'] ** 2 - values['residual_norm'] ** 2),
'predicted_merit_decrease': .5 * (values['base_norm'] ** 2 - values['linear_prediction_norm'] ** 2)})
for block, (start, size) in layouts.items():
br = bm[(sample, d, path, block)]
check(label + block + ':layout', int(br['offset']) == start and int(br['size']) == size and float(br['alpha']) == a)
part, _, _ = measurements(f[start:start + size], acts[d][start:start + size], trial[start:start + size], a)
check_measurements(br, part, check, label + block + ':')
roundoff = 128 * EPS * (1+part['base_norm'] + a * part['action_norm'] + part['residual_norm'])
floor = max(8*restoration_error,roundoff)
reported_floor = float(br['heuristic_roundoff_scale'])
check(label + block + ':heuristic_floor', math.isfinite(reported_floor) and roundoff*(1-1e-12) <= reported_floor <= floor*(1+1e-12))
if block == 'all': check(label+'path_floor_matches',close(float(row['heuristic_roundoff_scale']),reported_floor,reported_floor))
observations.append({'sample': sample, 'alpha': a, 'path': path, 'direction': d, 'block': block,
'usable': flags['usable'], 'increment_resolved': flags['increment_resolved'], **part, 'floor': floor})
for path in PATHS:
usable = all(data[(d, path)] is not None and data[(d, path)]['flags']['usable'] for d in DIRECTIONS)
for block, (start, size) in layouts.items():
row, label = mm[(sample, path, block)], f'{sample}:{path}:mixed:{block}:'
check(label + 'layout', int(row['offset']) == start and int(row['size']) == size and float(row['alpha']) == a)
check(label + 'honest_usable', flag(row, 'usable', check, label) == usable)
all_prepared = all(data[(d,path)] is not None for d in DIRECTIONS)
if not all_prepared:
check(label+'unavailable_metrics',all(math.isnan(float(row[k])) for k in ('full_remainder_norm','mixed_remainder_norm','linear_dot_mixed','norm_identity_error')))
continue
remainders = {d: data[(d, path)]['remainder'][start:start + size] for d in DIRECTIONS}
remainders['mixed'] = [u - v - w for u, v, w in zip(remainders['full'], remainders['shape'], remainders['fields'])]
prediction = data[('full', path)]['prediction'][start:start + size]
scale = norm(prediction) + sum(norm(v) for v in remainders.values())
projections = {d: dot(prediction, v) for d, v in remainders.items()}
for d, vals in remainders.items():
check(label + d + ':norm', close(norm(vals), float(row[d + '_remainder_norm']), scale))
check(label + d + ':projection', close(projections[d], float(row['linear_dot_' + d]), scale ** 2))
error = abs(projections['full'] - projections['shape'] - projections['fields'] - projections['mixed'])
check(label + 'projection_identity', error <= 2048 * EPS * scale ** 2 + 1e-28 and math.isfinite(float(row['projection_identity_error'])) and abs(float(row['projection_identity_error'])) <= 2048*EPS*scale**2+1e-28)
identity = norm(data[('full',path)]['trial'][start:start+size])**2 - norm(prediction)**2 - 2*(projections['shape']+projections['fields']+projections['mixed']) - norm(remainders['full'])**2
check(label+'full_merit_identity',abs(identity)<=2048*EPS*scale**2+1e-28 and math.isfinite(float(row['norm_identity_error'])) and abs(float(row['norm_identity_error']))<=2048*EPS*scale**2+1e-28)
validate_summary(summary, check)
for obs in observations:
if obs['path'] == 'curvature': obs['floor'] += float(pm[(obs['sample'], obs['direction'], obs['path'])]['gravity_arithmetic_allowance'])
passed = all(r['passed'] for r in checks)
return sanitize({'passed': passed, 'checks': checks, 'observations_valid': passed, 'required_artifacts': sorted(required),
'metadata': meta, 'summary': summary, 'direction_quality': {'raw_linear_gate': gate, 'backend_converged': converged,
'verified_relative_defect': defect / rhs_norm, 'passed': gate and converged}, 'paths': outcomes, 'shape_fields_action_cancellation': cancellation,
'preparation_failures':[r for r in rows(folder/(P+'preparations.csv')) if r['prepared']=='0'],
'gravity_assessment_scope':'Saved trial vectors independently verify the affine-target defect, not the underlying gravity action used by the reported affine-disagreement allowance. Actual gravity residual norms are reported separately.',
'vector_order_observations': vector_orders(observations, gate and converged) if passed else [],
'scope': 'Artifact validity is separate from linear convergence, nonlinear improvement, admissibility and physical validation.',
'order_scope': 'Orders require usable paths, finite verified J actions, resolved increments and vector remainders exceeding10 times the heuristic floor. Floor=max(8 times maximum restoration drift,128eps times(1+residual/action scales))+gravity arithmetic allowance for corrected paths. Normalized block norms use the configured diagonal Riesz factors, not exact physical L2. Noninteger superlinear order can be consistent with the EOS; derivative-error decline is reported without an order-two pass requirement. Failed outer forcing gates do not invalidate directional Taylor measurements; they preclude Newton-forcing comparisons and eta-based descent guarantees. This is not a certified assembly bound or universal order claim.'})
def validate_summary(summary, check):
ones = {'accepted_buffers_bitwise_identical','normalization_factors_bitwise_identical',
'roundoff_scale_is_subtraction_heuristic','path_seconds_include_shared_raw_preparation'}
zeroes = {'accepted_state_change','accepted_residual_change','accepted_physical_change',
'accepted_commits','polynomial_direction_repairs','linear_gate_failure_skips_trials'}
check('required_summary_invariants', ones | zeroes | {'maximum_original_restoration_error','maximum_restored_action_difference'} <= summary.keys())
for key in ones: check(key,summary.get(key)==1)
for key in zeroes: check(key,summary.get(key)==0)
check('supported_schema',summary.get('schema_version')==1 and summary.get('machine_epsilon')==EPS)
def validate_restoration(folder, summary, f, actions, layouts, check):
restored = vector(folder/(P+'restored_residual.txt'))
difference = norm(minus(restored,f))
records = rows(folder/(P+'restorations.csv'))
grouped = {}
for row in records: grouped.setdefault((row['sequence'],row['kind']),[]).append(row)
maximum = 0.0
for key,group in grouped.items():
by_block = {r['block']:r for r in group}
check('restoration:'+str(key)+':complete_blocks',len(by_block)==len(group) and set(by_block)==set(layouts))
for block,(start,size) in layouts.items():
row=by_block[block]
check('restoration:'+str(key)+':'+block, int(row['offset'])==start and int(row['size'])==size and close(float(row['base_norm']),norm(f[start:start+size])) and nonnegative(float(row['difference_norm'])))
block_norm = norm(float(by_block[b]['difference_norm']) for b in layouts if b!='all')
check('restoration:'+str(key)+':norm_partition',close(block_norm,float(by_block['all']['difference_norm']),norm(f)))
maximum=max(maximum,float(by_block['all']['difference_norm']))
check('restoration_records_present',bool(grouped))
check('maximum_restoration_recorded',close(maximum,summary['maximum_original_restoration_error'],norm(f)))
check('final_residual_restored',difference<=maximum+512*EPS*norm(f)+1e-28)
changes=[]
for d in DIRECTIONS:
final=vector(folder/(P+'restored_action_'+d+'.txt'))
changes.append(norm(minus(final,actions[d])))
check('maximum_action_change_recorded',close(max(changes),summary['maximum_restored_action_difference'],max(map(norm,actions.values()))))
# Restoration may be nonzero roundoff, but cannot be arbitrarily large and
# still certify that these finite differences use one frozen operator.
check('base_restoration_small',maximum<=2048*EPS*(1+norm(f)))
check('action_restoration_small',max(changes)<=2048*EPS*(1+max(map(norm,actions.values()))))
return maximum
def validate_work(folder, summary, linear, inner, paths, check):
preparations=rows(folder/(P+'preparations.csv'))
check('preparation_count',len(preparations)==summary['preparations'])
check('preparation_sequence', [int(r['sequence']) for r in preparations]==list(range(1,len(preparations)+1)))
check('preparation_times',all(nonnegative(float(r['seconds'])) for r in preparations))
check('preparation_total_time',close(math.fsum(float(r['seconds']) for r in preparations),summary['preparation_seconds']))
check('prepared_path_count',sum(r['prepared']=='1' for r in paths)==summary['prepared_paths'])
for direction in DIRECTIONS:
for path in PATHS:
calls=[r for r in preparations if r['kind']==path and r['direction']==direction]
wanted=[r for r in paths if r['path']==path and r['direction']==direction and (path=='raw' or r['inner_converged']=='1')]
check(direction+':'+path+':preparation_coverage',len(calls)==len(wanted) and {(float(r['alpha']),r['prepared']) for r in calls}=={(float(r['alpha']),r['prepared']) for r in wanted})
check('gravity_solve_count',len(inner)==summary['gravity_solves'])
check('gravity_iteration_total',sum(int(r['iterations']) for r in inner)==summary['gravity_iterations'])
check('gravity_failure_count',sum(r['converged']!='1' for r in inner)==summary['gravity_failures'])
check('gravity_solve_time',close(math.fsum(float(r['seconds']) for r in inner),summary['gravity_solve_seconds']))
check('linear_time_matches',close(float(linear['seconds']),summary['linear_solve_seconds']))
# Raw and corrected path timers overlap; do not add them to form a cost.
check('total_time_scope',nonnegative(summary['total_seconds']) and summary['total_seconds']+1e-6>=summary['preparation_seconds']+summary['linear_solve_seconds']+summary['gravity_solve_seconds'])
def inspect_campaign(campaign):
campaign, checks, results = Path(campaign).resolve(), [], {}
def check(name, passed): checks.append({'check': name, 'passed': bool(passed)})
try:
provenance = json.loads((campaign / 'provenance.json').read_text())
check('immutable_executable', digest(campaign / 'plateau_diagnostic') == provenance['binary_sha256'])
check('mesh_hash', digest(Path(provenance['mesh'])) == provenance['mesh_sha256'])
cases = provenance['cases']
check('unique_nonempty_cases', bool(cases) and len(cases) == len({c['id'] for c in cases}))
check('manifest_snapshot', json.loads((campaign / 'cases.json').read_text()) == cases)
for case in cases:
name = case['id']; folder = campaign / name
record = json.loads((campaign / (name + '_run.json')).read_text())
check(name + ':completed', bool(record.get('finished_utc')) and record.get('exit_code') == 0 and not record.get('timed_out', False))
args = case['arguments']
command = [str(campaign / 'plateau_diagnostic'), *args, '--mesh', provenance['mesh'], '--output', str(folder)]
check(name + ':command', record.get('command') == command and record.get('id') == name)
expected = set()
for i, arg in enumerate(args):
if arg == '--probe-state':
p = Path(args[i + 1]); p = p if p.is_absolute() else Path(provenance['source']) / p
expected.add(str(p.resolve()))
check(name + ':required_input_hashes', expected <= record.get('input_state_hashes', {}).keys())
for p, value in record.get('input_state_hashes', {}).items(): check(name + ':input:' + p, digest(Path(p)) == value)
for p, value in record.get('output_hashes', {}).items(): check(name + ':output:' + p, digest(folder / p) == value)
result = inspect_case(folder)
check(name + ':required_output_hashes', set(result['required_artifacts']) <= record.get('output_hashes', {}).keys())
bindings={'--linear-tol':'linear_tol','--max-linear':'max_linear','--restart':'restart',
'--index':'n','--angular-momentum':'J','--gravity-tol':'gravity_tolerance',
'--gravity-iterations':'gravity_iterations','--gravity-restart':'gravity_restart',
'--schur-sweeps':'context_gravity_schur_sweeps','--schur-relaxation':'context_gravity_schur_relaxation'}
metadata=result.get('metadata',{})
check(name+':expert_probe_selected','--nonlinear-step-probe' in args and metadata.get('nonlinear_step_probe')=='1')
for option,key in bindings.items():
if option in args:check(name+':requested_'+key,key in metadata and float(metadata[key])==float(args[args.index(option)+1]))
if '--preconditioner-variant' in args:check(name+':requested_inverse',metadata.get('context_preconditioner_variant')==args[args.index('--preconditioner-variant')+1])
if '--no-central' in args:check(name+':no_central_constraint',metadata.get('central_phase')=='0')
results[name] = result
passed = all(c['passed'] for c in checks) and all(r['passed'] for r in results.values())
if not all(c['passed'] for c in checks):
for r in results.values(): r.update(passed=False, observations_valid=False, vector_order_observations=[])
return {'passed': passed, 'source_commit': provenance['commit'], 'provenance_sha256': digest(campaign / 'provenance.json'),
'checker_sha256': digest(Path(__file__)), 'provenance_checks': checks, 'cases': results,
'scope': 'Independent vector/provenance audit. A failed numerical gate is a valid observation, not a successful model.'}
except (OSError, ValueError, KeyError, TypeError, IndexError) as error:
for r in results.values(): r.update(passed=False, observations_valid=False, vector_order_observations=[])
return {'passed': False, 'provenance_checks': checks + [{'check': 'complete_readable_campaign', 'passed': False, 'error': str(error)}], 'cases': results}
def compare_same_state(left_folder, left_review, right_folder, right_review):
"""Compare sampled directions only after caller-provided artifact audits.
A small action of their difference is evidence about this one sampled
response. It is neither a nullspace construction nor a condition estimate.
"""
left_folder,right_folder=Path(left_folder),Path(right_folder)
names=('physical_base','normalized_base','residual','rhs','state_factors','residual_factors')
identity={name:digest(left_folder/(P+name+'.txt'))==digest(right_folder/(P+name+'.txt')) for name in names}
identity['layout']=digest(left_folder/(P+'layout.csv'))==digest(right_folder/(P+'layout.csv'))
available=left_review['passed'] and right_review['passed'] and all(identity.values())
result={'available':available,'same_input_hashes':identity,
'eligible_for_newton_forcing_comparison':available and left_review.get('direction_quality',{}).get('passed',False) and right_review.get('direction_quality',{}).get('passed',False),
'scope':'w=p_right-p_left and Jw=action_right-action_left use identical saved state/RHS/factors/layout. One sampled weak response is not proof of a nullspace, gauge or condition number. Norms use diagonal normalization, not exact physical L2.'}
if not available:return result
pa,pb=(vector(folder/(P+'direction_full.txt')) for folder in (left_folder,right_folder))
ja,jb=(vector(folder/(P+'action_full.txt')) for folder in (left_folder,right_folder))
w,jw=minus(pb,pa),minus(jb,ja)
result['direction_difference_norm']=norm(w);result['action_difference_norm']=norm(jw)
result['sampled_action_over_direction_difference']=norm(jw)/norm(w) if norm(w)>0 else None
result['direction_cosine']=dot(pa,pb)/(norm(pa)*norm(pb)) if norm(pa)*norm(pb)>0 else None
result['differences_by_block']={}
layout=rows(left_folder/(P+'layout.csv'))
for kind,values in [('value',w),('residual',jw)]:
partition=layout_for(layout,kind,len(values))
result['differences_by_block'][kind]={block:norm(values[start:start+size]) for block,(start,size) in partition.items()}
return result
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('campaign', type=Path)
parser.add_argument('--output', type=Path, required=True)
args = parser.parse_args()
if args.campaign.resolve() in args.output.resolve().parents:
parser.error('Write the review outside the immutable campaign.')
result = inspect_campaign(args.campaign)
args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + '\n')
print('PASS' if result['passed'] else 'FAIL', args.output)
for name, case in result['cases'].items(): print(name, 'artifacts=', case['passed'], 'raw_linear=', case.get('direction_quality', {}).get('raw_linear_gate'))
return 0 if result['passed'] else 1
if __name__ == '__main__':
raise SystemExit(main())

252
experiments/test_nonlinear_step_checker.py Normal file
View File

@@ -0,0 +1,252 @@
#!/usr/bin/env python3
"""Synthetic vector maps catch false Taylor-order and successful-model claims."""
import csv
import json
import math
from pathlib import Path
import tempfile
import unittest
import check_nonlinear_step as c
def csv_file(path, data):
with path.open('w',newline='') as stream:
writer=csv.DictWriter(stream,fieldnames=list(data[0]));writer.writeheader();writer.writerows(data)
def vec_file(path, data):
path.write_text(str(len(data))+'\n'+'\n'.join(repr(v) for v in data)+'\n')
def measure(f,j,t,a):
linear=[x+a*y for x,y in zip(f,j)];r=[x-y for x,y in zip(t,linear)]
fn,jn,ln,tn,rn=[math.sqrt(math.fsum(x*x for x in v)) for v in (f,j,linear,t,r)]
projection=math.fsum(x*y for x,y in zip(linear,r))
return {'base_norm':fn,'action_norm':jn,'linear_prediction_norm':ln,'residual_norm':tn,
'remainder_norm':rn,'remainder_over_alpha':rn/a,'remainder_over_alpha_squared':rn/a**2,
'linear_dot_remainder':projection,'norm_identity_error':tn**2-ln**2-2*projection-rn**2},linear,r
class NonlinearStepCheckerTests(unittest.TestCase):
def setUp(self):
self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
self.campaign=Path(self.temp.name).resolve();self.folder=self.campaign/'synthetic';self.folder.mkdir()
self.make_case()
def make_case(self,eta=.01,order=2):
folder=self.folder
f=[.3,.4,.2,.1,.05];x=[2.,3.,4.,5.,6.]
p=[-(1-eta)*v for v in f];dirs={'full':p,'shape':[0,0,p[2],0,0],'fields':[p[0],p[1],0,p[3],p[4]]}
layout=[{'kind':'value','block':b,'offset':i,'size':1} for i,b in enumerate(['gravity_gradient','gravity_potential','surface_deformation','density','specific_enthalpy'])]
layout += [{'kind':'residual','block':b,'offset':i,'size':1} for i,b in enumerate(['gravity_gradient_relation','poisson_balance','material','surface','hydrostatic'])]
csv_file(folder/(c.P+'layout.csv'),layout)
blocks={'all':(0,5),**{r['block']:(r['offset'],r['size']) for r in layout if r['kind']=='residual'}}
for name,v in {'physical_base':[v/2 for v in x],'normalized_base':x,'residual':f,'rhs':[-v for v in f],
'state_factors':[2.]*5,'residual_factors':[3.]*5,'restored_residual':f}.items():vec_file(folder/(c.P+name+'.txt'),v)
for d,v in dirs.items():
for kind in ('direction','action','restored_action'):vec_file(folder/(c.P+kind+'_'+d+'.txt'),v)
rhs=c.norm(f);true=c.norm([a+b for a,b in zip(f,p)]);passed=true<=.03*rhs
linear={'status':0 if passed else 1,'converged':int(passed),'iterations':3,'rhs_norm':rhs,'reported_true_norm':true,
'verified_true_norm':true,'verified_relative_true':true/rhs,'threshold':.03*rhs,'verified_gate':int(passed),
'operator_applications':4,'inverse_applications':3,'seconds':1.}
csv_file(folder/(c.P+'linear.csv'),[linear])
paths,brs,mixed,inner,preparations=[],[],[],[],[]
def prep(kind,direction,a,n):
preparations.append({'sequence':len(preparations)+1,'kind':kind,'direction':direction,'alpha':a,'prepared':1,
'rejection_reason':-1,'rejection_stage':-1,'seconds':.01,'residual_norm':n})
prep('initial_restore','base',0,rhs)
alphas=[4.**(-i) for i in range(8)]
for sample,a in enumerate(alphas):
samples={}
for d,j in dirs.items():
predicted=[v+a*q for v,q in zip(f,j)]
vec_file(folder/(c.P+'prediction_'+d+'_a'+str(sample)+'.txt'),predicted)
# The map has shape/field interaction. Its first two equations
# can be corrected exactly without changing retained state.
nonlinear=[j[2]**2+j[0]*j[2],j[2]*j[3],2*j[3]**2+j[2]*j[3],j[2]**2,.2*j[4]**2]
raw=[v+a**order*b for v,b in zip(predicted,nonlinear)]
correction=[-a**order*nonlinear[i] if i<2 else 0. for i in range(5)]
vec_file(folder/(c.P+'correction_curvature_'+d+'_a'+str(sample)+'.txt'),correction)
inner_rhs=c.norm([a**order*v for v in nonlinear[:2]])
inner.append({'sample':sample,'alpha':a,'direction':d,'status':0,'converged':1,'iterations':1,'rhs_norm':inner_rhs,
'true_norm':0,'relative_true':0,'threshold':max(1e-14,1e-10*inner_rhs),'operator_applications':2,'inverse_applications':1,
'seconds':.02,'compact_operator':1,'compact_inverse':1})
for path in c.PATHS:
delta=correction if path=='curvature' else [0.]*5
trial=raw[:] if path=='raw' else predicted[:2]+raw[2:]
state=[v+a*q+dc for v,q,dc in zip(x,j,delta)]
intended=[a*q+dc for q,dc in zip(j,delta)];actual=[v-u for v,u in zip(state,x)]
addition=c.norm([v-u for v,u in zip(actual,intended)])
values,pred,remainder=measure(f,j,trial,a)
floor=128*c.EPS*(1+values['base_norm']+a*values['action_norm']+values['residual_norm'])
record={'sample':sample,'alpha':a,'direction':d,'path':path,'prepared':1,'inner_converged':1,'prepared_gate':1,'usable':1,
**values,'merit':.5*c.norm(trial)**2,'slope':sum(v*q for v,q in zip(f,j)),
'relative_derivative_error':values['remainder_over_alpha']/c.norm(j),
'expected_step_norm':c.norm(intended),'actual_step_norm':c.norm(actual),'addition_error_norm':addition,'increment_resolved':1,
'gravity_correction_norm':c.norm(correction) if path=='curvature' else math.nan,
'gravity_target_defect':0 if path=='curvature' else math.nan,'gravity_affine_disagreement':0 if path=='curvature' else math.nan,
'gravity_arithmetic_allowance':1e-12 if path=='curvature' else math.nan,
'gravity_target_threshold':1e-12 if path=='curvature' else math.nan,'seconds':.04 if path=='curvature' else .01,
'heuristic_roundoff_scale':floor}
paths.append(record);prep(path,d,a,c.norm(trial));samples[(d,path)]=(trial,pred,remainder)
for kind,v in [('state',state),('residual',trial)]:vec_file(folder/(c.P+kind+'_'+path+'_'+d+'_a'+str(sample)+'.txt'),v)
for block,(start,size) in blocks.items():
values,_,_=measure(f[start:start+size],j[start:start+size],trial[start:start+size],a)
brs.append({'sample':sample,'alpha':a,'direction':d,'path':path,'block':block,'offset':start,'size':size,**values,
'heuristic_roundoff_scale':128*c.EPS*(1+values['base_norm']+a*values['action_norm']+values['residual_norm'])})
for path in c.PATHS:
for block,(start,size) in blocks.items():
remainders={d:samples[(d,path)][2][start:start+size] for d in c.DIRECTIONS}
remainders['mixed']=[u-v-w for u,v,w in zip(remainders['full'],remainders['shape'],remainders['fields'])]
pred=samples[('full',path)][1][start:start+size];trial=samples[('full',path)][0][start:start+size]
projections={d:sum(a*b for a,b in zip(pred,v)) for d,v in remainders.items()}
mixed.append({'sample':sample,'alpha':a,'path':path,'block':block,'offset':start,'size':size,'usable':1,
**{d+'_remainder_norm':c.norm(v) for d,v in remainders.items()},**{'linear_dot_'+d:v for d,v in projections.items()},
'projection_identity_error':projections['full']-projections['shape']-projections['fields']-projections['mixed'],
'norm_identity_error':c.norm(trial)**2-c.norm(pred)**2-2*(projections['shape']+projections['fields']+projections['mixed'])-c.norm(remainders['full'])**2})
prep('final_restore','base',0,rhs)
restorations=[{'sequence':seq,'kind':kind,'block':block,'offset':start,'size':size,'base_norm':c.norm(f[start:start+size]),'difference_norm':0}
for seq,kind in [(1,'initial_restore'),(len(preparations),'final_restore')] for block,(start,size) in blocks.items()]
summary={'schema_version':1,'machine_epsilon':c.EPS,'linear_relative_tolerance':.03,'linear_absolute_tolerance':0,
'linear_maximum_iterations':10,'linear_restart_length':10,'gravity_relative_tolerance':1e-10,'gravity_absolute_tolerance':1e-14,
'gravity_maximum_iterations':100,'gravity_restart_length':10,'linear_verified_gate':int(passed),'linear_solve_seconds':1.,
'maximum_original_restoration_error':0,'maximum_restored_action_difference':0,'accepted_state_change':0,'accepted_residual_change':0,
'accepted_physical_change':0,'accepted_buffers_bitwise_identical':1,'normalization_factors_bitwise_identical':1,
'accepted_commits':0,'polynomial_direction_repairs':0,'linear_gate_failure_skips_trials':0,
'roundoff_scale_is_subtraction_heuristic':1,'path_seconds_include_shared_raw_preparation':1,
'preparations':len(preparations),'preparation_seconds':.01*len(preparations),'prepared_paths':len(paths),
'gravity_solves':len(inner),'gravity_iterations':len(inner),'gravity_failures':0,'gravity_solve_seconds':.02*len(inner),'total_seconds':3.}
summary.update({'alpha_'+str(i):a for i,a in enumerate(alphas)})
for name,data in [('paths',paths),('blocks',brs),('mixed',mixed),('inner',inner),('preparations',preparations),('restorations',restorations),
('summary',[{'metric':k,'value':v} for k,v in summary.items()])]:csv_file(folder/(c.P+name+'.csv'),data)
(folder/'metadata.txt').write_text('linear_tol=0.03\nmax_linear=10\nrestart=10\nnonlinear_step_probe=1\n')
snapshot=self.campaign/'snapshot.txt';vec_file(snapshot,[v/2 for v in x])
binary=self.campaign/'plateau_diagnostic';binary.write_bytes(b'not an executable')
mesh=self.campaign/'mesh.smesh';mesh.write_text('synthetic')
cases=[{'id':'synthetic','arguments':['--nonlinear-step-probe','--probe-state',str(snapshot)]}]
provenance={'commit':'synthetic','source':str(self.campaign),'mesh':str(mesh),'mesh_sha256':c.digest(mesh),
'binary_sha256':c.digest(binary),'cases':cases}
(self.campaign/'provenance.json').write_text(json.dumps(provenance));(self.campaign/'cases.json').write_text(json.dumps(cases))
record={'id':'synthetic','finished_utc':'2026-09-18T00:00:00Z','exit_code':0,
'command':[str(binary),*cases[0]['arguments'],'--mesh',str(mesh),'--output',str(folder)],
'input_state_hashes':{str(snapshot):c.digest(snapshot)},'output_hashes':{p.name:c.digest(p) for p in folder.iterdir()}}
(self.campaign/'synthetic_run.json').write_text(json.dumps(record))
def assert_valid(self,result):
self.assertTrue(result['passed'],[r for r in result.get('checks',result.get('provenance_checks',[])) if not r['passed']])
def alter(self,name,change):
path=self.folder/(c.P+name+'.csv');data=c.rows(path);change(data);csv_file(path,data)
def test_complete_vector_identities_and_provenance(self):
self.assert_valid(c.inspect_campaign(self.campaign))
def test_failed_raw_linear_gate_preserves_taylor_evidence_without_newton_claim(self):
self.make_case(eta=.1)
result=c.inspect_case(self.folder);self.assert_valid(result)
self.assertFalse(result['direction_quality']['passed'])
self.assertTrue(any(r['resolved_intervals'] for r in result['vector_order_observations']))
self.assertTrue(all(not r['eligible_for_newton_forcing_comparison'] for r in result['vector_order_observations']))
def test_noninteger_superlinear_order_is_not_a_failure(self):
self.make_case(order=1.5);result=c.inspect_case(self.folder);self.assert_valid(result)
entry=next(r for r in result['vector_order_observations'] if r['path']=='raw' and r['direction']=='full' and r['block']=='all')
self.assertTrue(entry['resolved_intervals'])
self.assertTrue(all(abs(r['vector_remainder_order']-1.5)<1e-5 for r in entry['resolved_intervals']))
def test_first_order_vector_error_is_not_misreported_as_quadratic(self):
self.make_case(order=1);result=c.inspect_case(self.folder);self.assert_valid(result)
entry=next(r for r in result['vector_order_observations'] if r['path']=='raw' and r['direction']=='full' and r['block']=='all')
self.assertTrue(all(abs(r['vector_remainder_order']-1)<1e-5 for r in entry['resolved_intervals']))
def test_corrupt_saved_residual_cannot_match_scalar_csv(self):
path=self.folder/(c.P+'residual_raw_full_a0.txt');values=c.vector(path);values[2]+=.001;vec_file(path,values)
self.assertFalse(c.inspect_case(self.folder)['passed'])
def test_mixed_projection_sign_is_checked(self):
self.alter('mixed',lambda data:data[0].update(linear_dot_mixed=.2))
self.assertFalse(c.inspect_case(self.folder)['passed'])
def test_missing_invariant_is_not_success(self):
self.alter('summary',lambda data:data.__setitem__(slice(None),[r for r in data if r['metric']!='accepted_buffers_bitwise_identical']))
self.assertFalse(c.inspect_case(self.folder)['passed'])
def test_changed_retained_coordinate_is_detected(self):
path=self.folder/(c.P+'state_curvature_full_a0.txt');v=c.vector(path);v[4]+=.1;vec_file(path,v)
self.assertFalse(c.inspect_case(self.folder)['passed'])
def test_unhashed_saved_vector_cannot_support_campaign_result(self):
p=self.campaign/'synthetic_run.json';record=json.loads(p.read_text());del record['output_hashes'][c.P+'action_full.txt'];p.write_text(json.dumps(record))
result=c.inspect_campaign(self.campaign);self.assertFalse(result['passed'])
self.assertFalse(result['cases']['synthetic']['observations_valid'])
def test_incomplete_producer_is_not_a_completed_result(self):
p=self.campaign/'synthetic_run.json';record=json.loads(p.read_text());del record['finished_utc'];p.write_text(json.dumps(record))
self.assertFalse(c.inspect_campaign(self.campaign)['passed'])
def unavailable_trial(self, raw_failure):
affected={'raw','curvature'} if raw_failure else {'curvature'}
path=self.folder/(c.P+'paths.csv');data=c.rows(path)
for row in data:
if row['sample']=='0' and row['direction']=='full' and row['path'] in affected:
row.update(prepared=0,usable=0)
for key in ('residual_norm','merit','remainder_norm','remainder_over_alpha','remainder_over_alpha_squared','relative_derivative_error','linear_dot_remainder','norm_identity_error','heuristic_roundoff_scale','gravity_correction_norm','gravity_target_defect','gravity_affine_disagreement','gravity_arithmetic_allowance','gravity_target_threshold'):
row[key]=math.nan
if row['path']=='curvature':row.update(inner_converged=0,prepared_gate=0,increment_resolved=0,expected_step_norm=math.nan,actual_step_norm=math.nan,addition_error_norm=math.nan)
csv_file(path,data)
for name in ('blocks','mixed'):
path=self.folder/(c.P+name+'.csv');data=c.rows(path)
for row in data:
if row['sample']=='0' and row['path'] in affected and (name=='mixed' or row['direction']=='full'):
if name=='mixed':
row['usable']=0
for key in row:
if key.endswith('_norm') or key.startswith('linear_dot_') or key.endswith('_error'):row[key]=math.nan
else:
for key in ('residual_norm','remainder_norm','remainder_over_alpha','remainder_over_alpha_squared','linear_dot_remainder','norm_identity_error','heuristic_roundoff_scale'):row[key]=math.nan
csv_file(path,data)
for path_kind in affected:
for kind in ('state','residual'):(self.folder/(c.P+kind+'_'+path_kind+'_full_a0.txt')).unlink()
(self.folder/(c.P+'correction_curvature_full_a0.txt')).unlink()
path=self.folder/(c.P+'inner.csv');data=c.rows(path)
if raw_failure:data=[r for r in data if not(r['sample']=='0' and r['direction']=='full')]
else:
row=next(r for r in data if r['sample']=='0' and r['direction']=='full')
row.update(status=1,converged=0,iterations=100,true_norm=2*float(row['threshold']),relative_true=2*float(row['threshold'])/float(row['rhs_norm']))
csv_file(path,data);inner=data
path=self.folder/(c.P+'preparations.csv');data=c.rows(path)
data=[r for r in data if not(r['kind']=='curvature' and r['direction']=='full' and float(r['alpha'])==1)]
for i,row in enumerate(data):
row['sequence']=i+1
if raw_failure and row['kind']=='raw' and row['direction']=='full' and float(row['alpha'])==1:row.update(prepared=0,rejection_reason=2,rejection_stage=6,residual_norm=math.nan)
csv_file(path,data)
count=len(data)
self.alter('restorations',lambda records:[r.update(sequence=count) for r in records if r['kind']=='final_restore'])
updates={'prepared_paths':48-len(affected),'preparations':count,'preparation_seconds':.01*count,
'gravity_solves':len(inner),'gravity_iterations':sum(int(r['iterations']) for r in inner),
'gravity_failures':0 if raw_failure else 1,'gravity_solve_seconds':.02*len(inner)}
self.alter('summary',lambda records:[r.update(value=updates[r['metric']]) for r in records if r['metric'] in updates])
def test_preparation_failure_is_honest_unavailable_observation(self):
self.unavailable_trial(True);result=c.inspect_case(self.folder);self.assert_valid(result)
self.assertEqual(len(result['preparation_failures']),1)
self.assertTrue(any(r['status']=='raw_preparation_failed' for r in result['paths']))
def test_failed_inner_solve_does_not_invent_corrected_trial(self):
self.unavailable_trial(False);result=c.inspect_case(self.folder);self.assert_valid(result)
self.assertTrue(any(r['status']=='inner_solve_failed' for r in result['paths']))
def test_failed_prepared_gate_retains_finite_vectors_but_not_usable_path(self):
self.alter('paths',lambda data:[r.update(prepared_gate=0,usable=0,gravity_affine_disagreement=2e-12) for r in data if r['sample']=='0' and r['direction']=='full' and r['path']=='curvature'])
self.alter('mixed',lambda data:[r.update(usable=0) for r in data if r['sample']=='0' and r['path']=='curvature'])
result=c.inspect_case(self.folder);self.assert_valid(result)
self.assertTrue(any(r['status']=='prepared_gravity_gate_failed' for r in result['paths']))
def test_below_floor_order_is_unresolved(self):
observations=[{'path':'raw','direction':'full','block':'all','alpha':a,'usable':True,'increment_resolved':True,
'remainder_norm':a*a*1e-15,'remainder_over_alpha':a*1e-15,'floor':1e-14} for a in (1,.25)]
self.assertFalse(c.vector_orders(observations,True)[0]['resolved_intervals'])
if __name__=='__main__':unittest.main(verbosity=2)