"""Independent finite-response controls with an unknown common clock factor. All derivatives are evaluated from the complete twelve-node rows. No trajectory or reserved response is generated by this contract suite. """ import json from dataclasses import replace from fractions import Fraction as Q import mpmath as mp import networkx as nx import pytest from tnfr.dynamics.relational import RelationalExchangeModel from tnfr.mathematics._rational_interval import I from tnfr.physics.relational_bridge_discrimination import ( assess_bridge_clock_law_discrimination, ) from tnfr.physics.relational_sine_comparison import bound_relational_sine_exchange from tnfr.sdk import export_to_json, relational_report_to_dict LEFT, RIGHT = tuple(range(6)), tuple(range(6, 12)) TARGET = tuple(Q(i % 6, 6) for i in range(12)) BUDGET = dict( amplitude=Q(1, 10), sampling_duration=Q(1, 50), clock_scale_bounds=(Q(9, 10), Q(11, 10)), preparation_error=Q(1, 10**10), observation_error=Q(1, 10**8), ) def _mp(value): value = Q(value) return mp.mpf(value.numerator) / value.denominator @pytest.fixture(scope="module") def source(): graph = nx.disjoint_union(nx.cycle_graph(6), nx.cycle_graph(6)) graph.add_edge(0, 6) graph.graph["GAMMA"] = {"type": "none"} for node in graph: graph.nodes[node].update(EPI=0, theta=0, nu_f=1) return bound_relational_sine_exchange( graph, reference_model=RelationalExchangeModel( 1, epi_weight=0, phase_weight=1, phase_domain="regular" ), ) def _assess(source, **changes): arguments = dict( left_cycle=LEFT, right_cycle=RIGHT, target_phase_turns=TARGET, **BUDGET ) arguments.update(changes) return assess_bridge_clock_law_discrimination(source, **arguments) @pytest.fixture(scope="module") def report(source): return _assess(source) def _matrices(source): neighbors = [[] for _ in source.nodes] for i, j in source.edges: neighbors[i].append(j) neighbors[j].append(i) laplacian = mp.matrix(len(neighbors)) for i, row in enumerate(neighbors): laplacian[i, i] = 1 for j in row: laplacian[i, j] = -mp.mpf(1) / len(row) return neighbors, laplacian def _current_and_derivative(neighbors, phase, eta): current = mp.matrix(len(neighbors), 1) derivative = mp.matrix(len(neighbors)) for i, row in enumerate(neighbors): for j in row: gap = phase[j] - phase[i] sine, cosine = mp.sin(gap), mp.cos(gap) current[i] += (sine + eta * sine**3) / len(row) weight = cosine * (1 + 3 * eta * sine**2) / len(row) derivative[i, j] += weight derivative[i, i] -= weight return current, derivative def test_finite_phase_curvature_and_fourth_jet_use_all_twenty_four_rows(source, report): with mp.workdps(85): neighbors, operator = _matrices(source) a = _mp(BUDGET["amplitude"]) p = tuple(mp.mpf(-1 if i < 6 else 1) / 2 for i in range(12)) target = tuple(2 * mp.pi * _mp(turn) for turn in TARGET) phase = tuple(angle + a * value for angle, value in zip(target, p)) for index, eta in enumerate((mp.mpf(0), mp.mpf(1) / 100)): current, derivative = _current_and_derivative(neighbors, phase, eta) # At x=0 the first phase derivative vanishes; the full nonlinear # second and fourth derivatives are A f and A Df A f. acceleration = operator * current fourth = operator * derivative * acceleration curvature = -(acceleration[6] - acceleration[0]) / a expected = mp.mpf(8) / 9 * (mp.sin(a) + eta * mp.sin(a) ** 3) / a assert abs(curvature - expected) < mp.mpf("1e-78") enclosed = report.ideal_phase_curvature_bounds[index] assert _mp(enclosed.lo) <= curvature <= _mp(enclosed.hi) assert abs(fourth[6] - fourth[0]) <= _mp( report.fourth_phase_bridge_derivative_upper_bound ) assert acceleration[0] != acceleration[1] # At finite amplitude the phase channel is not its tangent value. assert abs(curvature - mp.mpf(8) / 9) > mp.mpf("1e-4") def test_phase_preparation_is_fixed_by_the_full_field_reversor(source): with mp.workdps(75): neighbors, operator = _matrices(source) phase = mp.matrix([mp.mpf(i * i - 2) / 17 for i in range(12)]) form = mp.matrix([mp.mpf(i - 4) / 29 for i in range(12)]) for eta in (mp.mpf(0), mp.mpf(1) / 100): current, _ = _current_and_derivative(neighbors, phase, eta) reversed_current, _ = _current_and_derivative(neighbors, phase, eta) assert current == reversed_current assert operator * (-form) == -(operator * form) # R fixes (0,theta). Uniqueness therefore makes the nominal phase # response even in time; this does not identify noisy preparations. zero_form = mp.matrix(12, 1) assert operator * zero_form == zero_form def test_complete_endpoint_budgets_cover_the_clock_interval(report): assert report.discrimination_certified and report.denominator_positive assert report.phase_chamber_certified assert report.separation_margin_lower_bound > Q(1, 2000) with mp.workdps(85): a = _mp(BUDGET["amplitude"]) rho = _mp(BUDGET["preparation_error"]) sigma = _mp(BUDGET["observation_error"]) lower, upper = (Q(9, 500), Q(11, 500)) growth = mp.exp(2 * _mp(upper)) radius = a * growth / 2 form_fourth = ( 128 * mp.mpf(43) / 40 * radius**3 + 192 * mp.mpf(103) / 100 * radius**2 + 32 * radius ) phase_fourth = 64 * mp.mpf(103) / 100 * radius**2 + 32 * radius common = (4 * rho * growth + 2 * sigma) / a for derivative, retained, endpoint_errors in ( ( form_fourth, report.form_response_error_upper_bound, report.form_clock_endpoint_error_bounds, ), ( phase_fourth, report.phase_response_error_upper_bound, report.phase_clock_endpoint_error_bounds, ), ): coefficients = derivative / (12 * a), common for h, error in zip((lower, upper), endpoint_errors): actual_bound = ( coefficients[0] * _mp(h) ** 2 + coefficients[1] / _mp(h) ** 2 ) assert _mp(error) >= actual_bound assert retained == max(endpoint_errors) # The midpoint in h^2 is a convex combination of the two complete # endpoint totals, not independent worst-case time/noise choices. z = (_mp(lower) ** 2 + _mp(upper) ** 2) / 2 assert coefficients[0] * z + coefficients[1] / z <= _mp(retained) assert _mp(report.phase_chamber_margin_lower_bound) <= mp.pi / 12 - 2 * ( radius + rho * growth ) def test_clock_cancelled_ratios_contain_centers_and_all_reading_corners(report): a = BUDGET["amplitude"] sigma = BUDGET["observation_error"] assert report.ratio_prediction_bounds[0].hi < report.ratio_prediction_bounds[1].lo with mp.workdps(85): for eta, bound in zip((Q(0), Q(1, 100)), report.ratio_prediction_bounds): form = mp.mpf(2) / 3 + _mp(eta) / 2 phase = ( mp.mpf(8) / 9 * (mp.sin(_mp(a)) + _mp(eta) * mp.sin(_mp(a)) ** 3) / _mp(a) ) assert _mp(bound.lo) <= form / phase <= _mp(bound.hi) for h in (Q(9, 500), Q(11, 500)): noise = _mp(2 * sigma / (a * h**2)) for numerator_sign in (-1, 1): for denominator_sign in (-1, 1): ratio = (form + numerator_sign * noise) / ( phase + denominator_sign * noise ) assert _mp(bound.lo) <= ratio <= _mp(bound.hi) def test_change_of_sampling_units_preserves_the_same_structural_durations( source, report ): changed_units = _assess( source, sampling_duration=Q(1, 25), clock_scale_bounds=(Q(9, 20), Q(11, 20)), ) assert changed_units.structural_duration_bounds == report.structural_duration_bounds assert changed_units.ratio_prediction_bounds == report.ratio_prediction_bounds assert ( changed_units.phase_response_error_upper_bound == report.phase_response_error_upper_bound ) def test_signed_integer_lifts_keep_the_physical_bridge_deviation(source, report): # Each node may have its own integral lift. A nonzero bridge target lift # must be subtracted before using the phase statistic 2*(a-v). shifts = tuple(Q(i * i - 4) for i in range(12)) lifted = _assess( source, target_phase_turns=tuple(a + b for a, b in zip(TARGET, shifts)) ) assert lifted.bridge_target_turn_offset == shifts[6] - shifts[0] assert lifted.ratio_prediction_bounds == report.ratio_prediction_bounds with mp.workdps(85): a = _mp(BUDGET["amplitude"]) for i, enclosure in enumerate(lifted.phase_initial_phase_bounds): expected = 2 * mp.pi * _mp(TARGET[i] + shifts[i]) + a * ( mp.mpf(-1 if i < 6 else 1) / 2 ) assert _mp(enclosure.lo) <= expected <= _mp(enclosure.hi) def test_denominator_uncertainty_does_not_produce_an_infinite_or_passing_ratio(source): unreadable = _assess(source, observation_error=Q(1, 100)) assert unreadable.phase_chamber_certified assert not unreadable.denominator_positive assert not unreadable.discrimination_certified assert unreadable.ratio_prediction_bounds is None assert unreadable.separation_margin_lower_bound is None def test_positive_denominator_can_still_leave_the_laws_unresolved(source): noisy = _assess(source, observation_error=Q(1, 10**7)) assert noisy.denominator_positive and noisy.phase_chamber_certified assert not noisy.discrimination_certified assert noisy.separation_margin_lower_bound <= 0 def test_positive_denominator_does_not_replace_the_phase_branch_guard(source): outside = _assess(source, amplitude=Q(1, 2)) assert outside.denominator_positive assert not outside.phase_chamber_certified assert not outside.discrimination_certified assert outside.phase_chamber_margin_lower_bound < 0 def test_sub_grid_exact_amplitude_has_a_regular_sinc_denominator(source): amplitude = Q(1, 2**512) tiny = _assess( source, amplitude=amplitude, preparation_error=0, observation_error=0 ) assert tiny.amplitude == amplitude assert tiny.phase_chamber_certified and tiny.denominator_positive assert tiny.discrimination_certified assert tiny.positive_initial_form[6] == amplitude / 2 assert tiny.sinc_amplitude_bounds.lo > Q(99, 100) assert all(value > Q(4, 5) for value in tiny.denominator_lower_bounds) assert tiny.preparation_error_coefficient == tiny.observation_error_coefficient == 0 @pytest.mark.parametrize( "changes", ( {"clock_scale_bounds": (0, 1)}, {"clock_scale_bounds": (-1, 1)}, {"clock_scale_bounds": (2, 1)}, {"clock_scale_bounds": (True, 2)}, {"clock_scale_bounds": (1, float("inf"))}, {"clock_scale_bounds": (1,)}, {"clock_scale_bounds": (1, 2, 3)}, {"clock_scale_bounds": {1, 2}}, {"sampling_duration": 0}, {"sampling_duration": True}, {"sampling_duration": Q(1, 2)}, {"amplitude": Q(1001, 1000)}, {"amplitude": True}, {"amplitude": 0}, {"amplitude": -1}, {"preparation_error": -1}, {"preparation_error": True}, {"observation_error": -1}, {"observation_error": True}, ), ) def test_invalid_clock_or_duration_is_rejected(source, changes): with pytest.raises((ValueError, TypeError, OverflowError)): _assess(source, **changes) def test_ordered_clock_generator_and_poisoned_caches_preserve_predictions( source, report ): poisoned = replace( source, storage=I(-99), form_rates=(I(88),) * 12, phase_rates=(I(-77),) * 12, ) rebuilt = _assess(poisoned, clock_scale_bounds=iter(BUDGET["clock_scale_bounds"])) assert rebuilt.ratio_prediction_bounds == report.ratio_prediction_bounds assert rebuilt.ideal_phase_curvature_bounds == report.ideal_phase_curvature_bounds with pytest.raises((TypeError, ValueError)): _assess(replace(source, capacity=(True,) + source.capacity[1:])) def test_three_preparations_and_unknown_common_clock_survive_sdk_projection( report, tmp_path ): assert len(report.phase_initial_box) == 24 assert report.phase_initial_form == (0,) * 12 assert report.source.phase == (0,) * 12 projected = relational_report_to_dict(report) destination = tmp_path / "clock-discrimination.json" export_to_json(projected, destination) saved = json.loads(destination.read_text()) assert saved == projected data = saved["report"] assert data["sampling_duration"] == {"numerator": 1, "denominator": 50} assert len(data["ratio_prediction_bounds"]) == 2 assert len(data["phase_initial_box"]) == 24 assert data["discrimination_certified"]