Skip to content

UPDE — PHA-C Acceptance Chain

PHACAcceptanceRecord is the end-to-end PHA-C review gate. It advances a moving-frame trajectory through the same physics surfaces used by the PHA-C lane, then converts the result into timeline evidence and a deterministic acceptance hash.

The acceptance record is review-only. It does not write to actuators, change coupling policy, schedule hardware, or mutate supervisor state.

Use cases

Use the PHA-C acceptance chain when reviewers need evidence that the whole lane was exercised rather than a single isolated module:

  • MIF/FRC readiness checks where spatial coupling, Doppler correction, moving-frame propagation, merge-window locking, handoff hashing, and timeline hashing must be reviewed together;
  • release gates that need one compact record proving the PHA-C chain can run from initial state to final lock evidence;
  • benchmark snapshots that aggregate the per-module Rust, Go, Julia, Mojo, and Python parity gates while labelling local timings as non-isolated regression evidence;
  • Studio or replay panels that need an acceptance hash without exposing raw trajectory arrays.

Contract

Inputs are the initial state plus schedules:

phases_t0[i]              = initial oscillator phase
positions_t0[i]           = initial axial position
omega_schedule[t, i]      = natural frequency at schedule row t
velocity_schedule[t, i]   = axial velocity at schedule row t
knm[i, j]                 = zero-diagonal base coupling matrix

For each schedule row, the builder:

  1. resolves distance-modulated coupling through SpatialCouplingModulator;
  2. computes the graph-weighted Doppler correction;
  3. advances one moving-frame UPDE step;
  4. records phase and position trajectory rows;
  5. computes the signed moving-frame kinematic residual max |z[t+1] - (z[t] + v[t] * dt)|;
  6. builds a PHACTimelineRecord over the trajectory;
  7. can project the verified acceptance payload into a Lean PHACKinematicProofObligation for SPOFormal.Kinematic; and
  8. hashes schedules, trajectories, spatial couplings, Doppler trace, timeline, proof-obligation references, and final acceptance payload.

The record carries:

  • first-lock index/time and final-lock state;
  • lock sample, lock-loss, reset, and maximum consecutive-lock counts;
  • maximum absolute Doppler correction and spatial coupling;
  • maximum moving-frame kinematic residual, maximum absolute velocity, and maximum per-oscillator axial path length;
  • final-position, maximum-velocity, and path-length equation replay under PHA_C_ACCEPTANCE_KINEMATIC_SUMMARY_REPLAY_TOLERANCE;
  • fixed-point Lean proof-obligation compatibility through build_pha_c_kinematic_proof_obligation(...);
  • maximum phase/spatial dispersion, minimum signed phase/spatial margins, minimum Kuramoto order parameter, and maximum distance to reference;
  • replay validation that each signed margin equals its tolerance minus the corresponding maximum dispersion under PHA_C_ACCEPTANCE_MARGIN_REPLAY_TOLERANCE;
  • resolved tolerance profile provenance;
  • execution_disabled=True, actuating=False, and the fixed claim boundary pha_c_end_to_end_acceptance_review_only.

Minimal example

import numpy as np
from scpn_phase_orchestrator.upde.pha_c_acceptance import (
    build_pha_c_acceptance_record,
    verify_pha_c_acceptance_record,
)
from scpn_phase_orchestrator.upde.pha_c_formal_obligation import (
    build_pha_c_kinematic_proof_obligation,
    verify_pha_c_kinematic_proof_obligation,
)

n = 5
phases = np.linspace(-0.002, 0.002, n)
positions = np.linspace(-0.0006, 0.0006, n)
omega = np.zeros((4, n))
knm = np.full((n, n), 0.04)
np.fill_diagonal(knm, 0.0)
velocities = np.vstack([np.linspace(0.10, 0.12, n) for _ in range(4)])

record = build_pha_c_acceptance_record(
    phases,
    positions,
    omega,
    knm,
    velocities,
    dt=1.0e-3,
    required_consecutive_samples=3,
    tolerance_profile="baseline_1x",
    backend="python",
)

assert record.first_lock_index == 2
assert record.final_lock_achieved
assert record.kinematic_residual_max_m <= 1.0e-12
assert record.execution_disabled
assert not record.actuating
acceptance_payload = record.to_dict()
verify_pha_c_acceptance_record(record)
obligation = build_pha_c_kinematic_proof_obligation(record)
assert obligation.proof_obligations_discharged
verify_pha_c_kinematic_proof_obligation(obligation)

Use verify_pha_c_acceptance_record(...) when replaying a stored acceptance record. It rechecks sample/step consistency, first-lock semantics, review-only flags, signed margin equations, the moving-frame kinematic residual bound, kinematic equation replay flags and tolerance, SHA-256 fields, the timeline digest reference, and the canonical acceptance hash without requiring the original schedules or trajectories. The signed margin replay rejects records whose positive-looking phase or spatial margin no longer matches tolerance - maximum_dispersion. Use verify_pha_c_kinematic_proof_obligation(...) when release review also needs the accepted runtime envelope bound to Lean. The release-facing boundary is the combined KinematicBounds.acceptanceCertificate predicate and acceptance_certificate_discharges_runtime_preconditions theorem; that certificate includes the spatial budget theorem, phase-budget theorem, and acceptance kinematic-equation replay certificate.

Relationship to other PHA-C records

Surface Scope Use when
PHACHandoffRecord one phase/position sample a downstream lane needs one reviewed event-state atom
PHACTimelineRecord complete phase/position trajectory a downstream lane needs lock/loss/reset history
PHACAcceptanceRecord full PHA-C physics chain plus timeline a release, MIF/FRC, or benchmark gate needs end-to-end evidence
PHACKinematicProofObligation fixed-point Lean assumptions derived from verified acceptance a formal gate or downstream lane needs a signed theorem-specific obligation

Polyglot parity and benchmark snapshot

The benchmark gate records Rust, Mojo, Julia, Go, and Python source-contract slots for the acceptance builder, then aggregates the existing PHA-C subgates: spatial modulation, time-varying omega, Doppler, moving-frame, merge window, handoff, and timeline. Acceptance rows also publish the minimum signed margins copied from the timeline plus the moving-frame kinematic residual, so release evidence exposes both the distance to the reviewed merge envelope and the mechanical validity of the axial schedule. The benchmark row now also records phase_margin_equation_validated, spatial_margin_equation_validated, signed_margin_equations_validated, and margin_replay_tolerance for every backend slot. It also records final_position_equation_validated, max_abs_velocity_equation_validated, path_length_equation_validated, kinematic_equations_validated, and kinematic_summary_replay_tolerance for every backend slot so the aggregate acceptance record preserves the moving-frame summary equations rather than only scalar residuals. The same benchmark now builds and verifies a Lean kinematic proof-obligation manifest for every backend row. The acceptance gate fails unless the manifest names SPOFormal.Kinematic, targets budget_certificate_discharges_budget, replays its canonical hash, replays the finite-horizon Gronwall budget trace, and discharges the fixed-point merge-window and phase-margin checks. The final formal row must also name KinematicBounds.acceptanceCertificate, name acceptance_certificate_discharges_runtime_preconditions, and discharge the combined spatial-budget, phase-budget, and acceptance-replay certificate. The same manifest publishes sampled continuous-rate fields for dt, horizon time, velocity-rate bounds, and residual-rate bounds so Rust, Mojo, Julia, Go, and Python source-contract rows carry the same time-normalised formal assumptions. The rows now also publish the SPOFormal.Continuous horizon theorem, continuous drive-rate sum, continuous horizon-drive replay, continuous budget, continuous margin, and a Boolean continuous-envelope discharge flag. The formal manifest now also publishes formal_obligation_acceptance_kinematic_equations_validated and formal_obligation_acceptance_kinematic_summary_replay_tolerance, binding the Lean-facing proof obligation to the verified acceptance record's moving-frame summary equations. The same benchmark row now publishes formal_obligation_acceptance_replay_certificate_discharged, formal_obligation_acceptance_certificate_discharged, formal_obligation_acceptance_certificate_predicate, and formal_obligation_acceptance_certificate_theorem, and the parity gate fails unless every backend row discharges the combined Lean acceptance certificate. Before calculating parity error, the source-contract validator now compares raw PHACAcceptanceRecord fields rather than the canonical to_dict() payload, so numeric strings, non-finite values, boolean aliases, malformed integer fields, non-plain lock flags, malformed string/hash fields, and typed field divergence fail closed before benchmark rows are published. Imported subgate payload flags are also parsed as strict integers so malformed benchmark evidence cannot be accepted through Python string coercion.

uv run python benchmarks/pha_c_acceptance_benchmark.py \
  --parity-gate \
  --calls 1 \
  --output benchmarks/results/pha_c_acceptance.json

Committed benchmark JSON is local regression evidence only. It is not a production timing claim unless rerun under the benchmark-isolation protocol.

Failure boundaries

The acceptance builder fails closed on:

  • empty, non-finite, complex, object-dtype, or boolean state vectors;
  • mismatched phase and position shapes;
  • malformed omega or velocity schedules;
  • non-square, non-finite, or non-zero-diagonal coupling matrices;
  • invalid moving-frame backend names;
  • non-positive dt, tolerances, or integration controls;
  • unknown tolerance profile names.

The polyglot source-contract validator also fails closed on acceptance records whose raw dataclass fields do not preserve the expected numeric, integer, boolean, and string/hash domains, even if their canonical dictionary projection would coerce to the reference value.

PHACAcceptanceRecord dataclass

PHACAcceptanceRecord(
    sample_count: int,
    step_count: int,
    oscillator_count: int,
    start_time: float,
    end_time: float,
    dt: float,
    first_lock_index: int,
    first_lock_time: float,
    first_lock_observed: bool,
    final_lock_achieved: bool,
    lock_sample_count: int,
    lock_loss_count: int,
    reset_count: int,
    max_consecutive_lock_samples: int,
    max_abs_doppler_term: float,
    max_abs_spatial_coupling: float,
    max_phase_dispersion_rad: float,
    max_spatial_dispersion_m: float,
    kinematic_residual_max_m: float,
    max_abs_velocity_m_per_s: float,
    path_length_max_m: float,
    final_position_equation_validated: bool,
    max_abs_velocity_equation_validated: bool,
    path_length_equation_validated: bool,
    kinematic_equations_validated: bool,
    kinematic_summary_replay_tolerance: float,
    min_phase_margin_rad: float,
    min_spatial_margin_m: float,
    min_phase_order_parameter: float,
    max_distance_to_reference_m: float,
    reference_phase: float,
    reference_point: float,
    phase_tol_rad: float,
    spatial_tol_m: float,
    tolerance_profile_name: str,
    tolerance_profile_multiplier: float,
    required_consecutive_samples: int,
    moving_frame_backend_request: str,
    claim_boundary: str,
    evidence_kind: str,
    execution_disabled: bool,
    actuating: bool,
    omega_schedule_sha256: str,
    velocity_schedule_sha256: str,
    phase_trajectory_sha256: str,
    position_trajectory_sha256: str,
    initial_spatial_coupling_sha256: str,
    final_spatial_coupling_sha256: str,
    doppler_trace_sha256: str,
    timeline_sha256: str,
    acceptance_sha256: str,
)

Audit-ready acceptance evidence for the complete PHA-C chain.

Methods:

to_dict

to_dict() -> dict[str, float | int | bool | str]

Return a JSON-safe canonical representation.

Returns

dict[str, float | int | bool | str] Return a JSON-safe canonical representation.

Source code in src/scpn_phase_orchestrator/upde/pha_c_acceptance.py
def to_dict(self) -> dict[str, float | int | bool | str]:
    """Return a JSON-safe canonical representation.

    Returns
    -------
    dict[str, float | int | bool | str]
        Return a JSON-safe canonical representation.
    """
    return pha_c_acceptance_record_to_dict(self)

build_pha_c_acceptance_record

build_pha_c_acceptance_record(
    phases_t0: ArrayLike,
    positions_t0: ArrayLike,
    omega_schedule: ArrayLike,
    knm: ArrayLike,
    velocity_schedule: ArrayLike,
    *,
    alpha: object = 0.0,
    spatial_modulator: SpatialCouplingModulator
    | None = None,
    doppler_strength: object = 0.001,
    doppler_epsilon: object = 1e-09,
    zeta: object = 0.0,
    psi: object = 0.0,
    dt: object = 0.001,
    method: str = "rk4",
    n_substeps: object = 1,
    atol: object = 1e-09,
    rtol: object = 1e-09,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
    phase_tol_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_tol_m: object = DEFAULT_SPATIAL_TOL_M,
    required_consecutive_samples: object = 3,
    tolerance_profile: object | None = "baseline_1x",
    backend: object = "python",
) -> PHACAcceptanceRecord

Build deterministic end-to-end PHA-C acceptance evidence.

The builder advances a moving-frame trajectory one schedule row at a time, recording spatially modulated coupling and Doppler traces before converting the trajectory into a PHA-C event timeline. The output remains review-only; it is evidence for downstream gates and never permits actuation.

Parameters

phases_t0 : ArrayLike Initial oscillator phases in radians, shape (N,). positions_t0 : ArrayLike Initial axial coordinates per oscillator, shape (N,). omega_schedule : ArrayLike Per-step natural-frequency vectors, shape (n_steps, N). knm : ArrayLike Coupling matrix K_nm, shape (N, N). velocity_schedule : ArrayLike Per-step axial velocity vectors, shape (n_steps, N). alpha : object Phase-lag matrix in radians, shape (N, N), or None for no lag. spatial_modulator : SpatialCouplingModulator | None Configured spatial coupling modulator. doppler_strength : object Doppler coupling-correction strength. doppler_epsilon : object Numerical floor guarding the Doppler denominator. zeta : object External drive strength ζ. psi : object External drive reference phase Ψ in radians. dt : object Integration step size. method : str Integration method (euler, rk4, or rk45). n_substeps : object Number of inner substeps per outer step. atol : object Absolute tolerance for the adaptive (rk45) integrator. rtol : object Relative tolerance for the adaptive (rk45) integrator. reference_phase : object Reference phase for the lock criterion, in radians. reference_point : object Reference axial coordinate for the spatial-margin criterion. phase_tol_rad : object Phase lock tolerance in radians. spatial_tol_m : object Spatial lock tolerance in metres. required_consecutive_samples : object Consecutive in-tolerance samples required to declare lock. tolerance_profile : object | None Named tolerance profile, or None for the baseline profile. backend : object Name of the compute backend to run.

Returns

PHACAcceptanceRecord The deterministic end-to-end PHA-C acceptance record.

Raises

ValueError If any input is invalid or the acceptance replay fails.

Source code in src/scpn_phase_orchestrator/upde/pha_c_acceptance.py
def build_pha_c_acceptance_record(
    phases_t0: ArrayLike,
    positions_t0: ArrayLike,
    omega_schedule: ArrayLike,
    knm: ArrayLike,
    velocity_schedule: ArrayLike,
    *,
    alpha: object = 0.0,
    spatial_modulator: SpatialCouplingModulator | None = None,
    doppler_strength: object = 1.0e-3,
    doppler_epsilon: object = 1.0e-9,
    zeta: object = 0.0,
    psi: object = 0.0,
    dt: object = 1.0e-3,
    method: str = "rk4",
    n_substeps: object = 1,
    atol: object = 1.0e-9,
    rtol: object = 1.0e-9,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
    phase_tol_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_tol_m: object = DEFAULT_SPATIAL_TOL_M,
    required_consecutive_samples: object = 3,
    tolerance_profile: object | None = "baseline_1x",
    backend: object = "python",
) -> PHACAcceptanceRecord:
    """Build deterministic end-to-end PHA-C acceptance evidence.

    The builder advances a moving-frame trajectory one schedule row at a time,
    recording spatially modulated coupling and Doppler traces before converting
    the trajectory into a PHA-C event timeline. The output remains review-only;
    it is evidence for downstream gates and never permits actuation.

    Parameters
    ----------
    phases_t0 : ArrayLike
        Initial oscillator phases in radians, shape ``(N,)``.
    positions_t0 : ArrayLike
        Initial axial coordinates per oscillator, shape ``(N,)``.
    omega_schedule : ArrayLike
        Per-step natural-frequency vectors, shape ``(n_steps, N)``.
    knm : ArrayLike
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    velocity_schedule : ArrayLike
        Per-step axial velocity vectors, shape ``(n_steps, N)``.
    alpha : object
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    spatial_modulator : SpatialCouplingModulator | None
        Configured spatial coupling modulator.
    doppler_strength : object
        Doppler coupling-correction strength.
    doppler_epsilon : object
        Numerical floor guarding the Doppler denominator.
    zeta : object
        External drive strength ``ζ``.
    psi : object
        External drive reference phase ``Ψ`` in radians.
    dt : object
        Integration step size.
    method : str
        Integration method (``euler``, ``rk4``, or ``rk45``).
    n_substeps : object
        Number of inner substeps per outer step.
    atol : object
        Absolute tolerance for the adaptive (rk45) integrator.
    rtol : object
        Relative tolerance for the adaptive (rk45) integrator.
    reference_phase : object
        Reference phase for the lock criterion, in radians.
    reference_point : object
        Reference axial coordinate for the spatial-margin criterion.
    phase_tol_rad : object
        Phase lock tolerance in radians.
    spatial_tol_m : object
        Spatial lock tolerance in metres.
    required_consecutive_samples : object
        Consecutive in-tolerance samples required to declare lock.
    tolerance_profile : object | None
        Named tolerance profile, or ``None`` for the baseline profile.
    backend : object
        Name of the compute backend to run.

    Returns
    -------
    PHACAcceptanceRecord
        The deterministic end-to-end PHA-C acceptance record.

    Raises
    ------
    ValueError
        If any input is invalid or the acceptance replay fails.
    """
    phases = _as_float_vector(phases_t0, name="phases_t0")
    n = int(phases.size)
    positions = _as_float_vector(positions_t0, name="positions_t0")
    if positions.shape != phases.shape:
        raise ValueError("positions_t0 must have the same shape as phases_t0")
    omega = _as_schedule(omega_schedule, name="omega_schedule", n=n)
    velocities = _as_schedule(velocity_schedule, name="velocity_schedule", n=n)
    if velocities.shape[0] != omega.shape[0]:
        raise ValueError("velocity_schedule step count must match omega_schedule")
    k = _as_knm(knm, n=n)
    dt_s = _validate_positive_scalar(dt, name="dt")
    strength = _validate_real_scalar(doppler_strength, name="doppler_strength")
    doppler_eps = _validate_positive_scalar(doppler_epsilon, name="doppler_epsilon")
    zeta_f = _validate_real_scalar(zeta, name="zeta")
    psi_f = _validate_real_scalar(psi, name="psi")
    atol_f = _validate_positive_scalar(atol, name="atol")
    rtol_f = _validate_positive_scalar(rtol, name="rtol")
    backend_request = _validate_backend_request(backend)
    if isinstance(n_substeps, (bool, np.bool_)) or not isinstance(
        n_substeps,
        (int, np.integer),
    ):
        raise ValueError("n_substeps must be a positive integer")
    n_substeps_i = int(n_substeps)
    if n_substeps_i < 1:
        raise ValueError("n_substeps must be a positive integer")
    if spatial_modulator is None:
        modulator = SpatialCouplingModulator(K_base=1.0)
    elif isinstance(spatial_modulator, SpatialCouplingModulator):
        modulator = spatial_modulator
    else:
        raise ValueError("spatial_modulator must be a SpatialCouplingModulator")

    phase_rows: list[FloatArray] = [phases.copy()]
    position_rows: list[FloatArray] = [positions.copy()]
    spatial_rows: list[FloatArray] = []
    doppler_rows: list[FloatArray] = []
    current_phases = phases.copy()
    current_positions = positions.copy()
    for step in range(int(omega.shape[0])):
        spatial_coupling = modulator.modulate(k, current_positions)
        spatial_rows.append(spatial_coupling)
        correction = doppler_term(
            velocities[step],
            spatial_coupling,
            doppler_strength=strength,
            doppler_epsilon=doppler_eps,
        )
        doppler_rows.append(correction)
        state = moving_frame_run(
            current_phases,
            current_positions,
            omega[step : step + 1],
            k,
            alpha,
            velocities[step : step + 1],
            modulator,
            doppler_strength=strength,
            doppler_epsilon=doppler_eps,
            zeta=zeta_f,
            psi=psi_f,
            dt=dt_s,
            method=method,
            n_substeps=n_substeps_i,
            atol=atol_f,
            rtol=rtol_f,
            backend=backend_request,
        )
        current_phases = np.ascontiguousarray(state[:n], dtype=np.float64)
        current_positions = np.ascontiguousarray(state[n:], dtype=np.float64)
        phase_rows.append(current_phases)
        position_rows.append(current_positions)
    final_spatial_coupling = modulator.modulate(k, current_positions)

    phase_trajectory = np.vstack(phase_rows).astype(np.float64, copy=False)
    position_trajectory = np.vstack(position_rows).astype(np.float64, copy=False)
    doppler_trace = np.vstack(doppler_rows).astype(np.float64, copy=False)
    spatial_trace = np.stack(spatial_rows).astype(np.float64, copy=False)
    predicted_position_steps = position_trajectory[:-1] + velocities * dt_s
    kinematic_residual_max_m = float(
        np.max(np.abs(position_trajectory[1:] - predicted_position_steps))
    )
    expected_final_positions = positions + np.sum(velocities * dt_s, axis=0)
    final_position_residual_m = float(
        np.max(np.abs(position_trajectory[-1] - expected_final_positions))
    )
    max_abs_velocity_m_per_s = float(np.max(np.abs(velocities)))
    replayed_max_abs_velocity_m_per_s = float(np.max(np.abs(velocities)))
    expected_path_length_max_m = float(
        np.max(np.sum(np.abs(velocities * dt_s), axis=0))
    )
    path_length_max_m = expected_path_length_max_m
    replayed_path_length_max_m = float(
        np.max(np.sum(np.abs(velocities * dt_s), axis=0))
    )
    final_position_equation_validated = (
        final_position_residual_m <= PHA_C_KINEMATIC_RESIDUAL_TOLERANCE_M
    )
    max_abs_velocity_equation_validated = (
        abs(max_abs_velocity_m_per_s - replayed_max_abs_velocity_m_per_s)
        <= PHA_C_ACCEPTANCE_KINEMATIC_SUMMARY_REPLAY_TOLERANCE
    )
    path_length_equation_validated = (
        abs(path_length_max_m - replayed_path_length_max_m)
        <= PHA_C_ACCEPTANCE_KINEMATIC_SUMMARY_REPLAY_TOLERANCE
    )
    kinematic_equations_validated = (
        final_position_equation_validated
        and max_abs_velocity_equation_validated
        and path_length_equation_validated
    )
    times = np.arange(phase_trajectory.shape[0], dtype=np.float64) * dt_s
    timeline = build_pha_c_event_timeline(
        phase_trajectory,
        position_trajectory,
        times=times,
        reference_phase=reference_phase,
        reference_point=reference_point,
        phase_tol_rad=phase_tol_rad,
        spatial_tol_m=spatial_tol_m,
        required_consecutive_samples=required_consecutive_samples,
        tolerance_profile=tolerance_profile,
    )
    first_spatial_coupling = spatial_rows[0]
    payload = _record_dict_without_hash(
        sample_count=timeline.sample_count,
        step_count=int(omega.shape[0]),
        oscillator_count=n,
        start_time=float(times[0]),
        end_time=float(times[-1]),
        dt=dt_s,
        first_lock_index=timeline.first_lock_index,
        first_lock_time=timeline.first_lock_time,
        first_lock_observed=timeline.first_lock_observed,
        final_lock_achieved=timeline.final_lock_achieved,
        lock_sample_count=timeline.lock_sample_count,
        lock_loss_count=timeline.lock_loss_count,
        reset_count=timeline.reset_count,
        max_consecutive_lock_samples=timeline.max_consecutive_lock_samples,
        max_abs_doppler_term=float(np.max(np.abs(doppler_trace))),
        max_abs_spatial_coupling=float(np.max(np.abs(spatial_trace))),
        max_phase_dispersion_rad=timeline.max_phase_dispersion_rad,
        max_spatial_dispersion_m=timeline.max_spatial_dispersion_m,
        kinematic_residual_max_m=kinematic_residual_max_m,
        max_abs_velocity_m_per_s=max_abs_velocity_m_per_s,
        path_length_max_m=path_length_max_m,
        final_position_equation_validated=final_position_equation_validated,
        max_abs_velocity_equation_validated=max_abs_velocity_equation_validated,
        path_length_equation_validated=path_length_equation_validated,
        kinematic_equations_validated=kinematic_equations_validated,
        kinematic_summary_replay_tolerance=(
            PHA_C_ACCEPTANCE_KINEMATIC_SUMMARY_REPLAY_TOLERANCE
        ),
        min_phase_margin_rad=timeline.min_phase_margin_rad,
        min_spatial_margin_m=timeline.min_spatial_margin_m,
        min_phase_order_parameter=timeline.min_phase_order_parameter,
        max_distance_to_reference_m=timeline.max_distance_to_reference_m,
        reference_phase=timeline.reference_phase,
        reference_point=timeline.reference_point,
        phase_tol_rad=timeline.phase_tol_rad,
        spatial_tol_m=timeline.spatial_tol_m,
        tolerance_profile_name=timeline.tolerance_profile_name,
        tolerance_profile_multiplier=timeline.tolerance_profile_multiplier,
        required_consecutive_samples=timeline.required_consecutive_samples,
        moving_frame_backend_request=backend_request,
        omega_schedule_sha256=_array_sha256(omega),
        velocity_schedule_sha256=_array_sha256(velocities),
        phase_trajectory_sha256=_array_sha256(phase_trajectory),
        position_trajectory_sha256=_array_sha256(position_trajectory),
        initial_spatial_coupling_sha256=_array_sha256(first_spatial_coupling),
        final_spatial_coupling_sha256=_array_sha256(final_spatial_coupling),
        doppler_trace_sha256=_array_sha256(doppler_trace),
        timeline_sha256=timeline.timeline_sha256,
    )
    return PHACAcceptanceRecord(
        **cast(Any, payload),
        acceptance_sha256=_sha256_json(payload),
    )

pha_c_acceptance_record_to_dict

pha_c_acceptance_record_to_dict(
    record: PHACAcceptanceRecord,
) -> dict[str, float | int | bool | str]

Return the canonical JSON-safe PHA-C acceptance payload.

Parameters

record : PHACAcceptanceRecord The PHA-C record to operate on.

Returns

dict[str, float | int | bool | str] The canonical JSON-safe PHA-C acceptance payload.

Source code in src/scpn_phase_orchestrator/upde/pha_c_acceptance.py
def pha_c_acceptance_record_to_dict(
    record: PHACAcceptanceRecord,
) -> dict[str, float | int | bool | str]:
    """Return the canonical JSON-safe PHA-C acceptance payload.

    Parameters
    ----------
    record : PHACAcceptanceRecord
        The PHA-C record to operate on.

    Returns
    -------
    dict[str, float | int | bool | str]
        The canonical JSON-safe PHA-C acceptance payload.
    """
    payload = _record_dict_without_hash(
        sample_count=record.sample_count,
        step_count=record.step_count,
        oscillator_count=record.oscillator_count,
        start_time=record.start_time,
        end_time=record.end_time,
        dt=record.dt,
        first_lock_index=record.first_lock_index,
        first_lock_time=record.first_lock_time,
        first_lock_observed=record.first_lock_observed,
        final_lock_achieved=record.final_lock_achieved,
        lock_sample_count=record.lock_sample_count,
        lock_loss_count=record.lock_loss_count,
        reset_count=record.reset_count,
        max_consecutive_lock_samples=record.max_consecutive_lock_samples,
        max_abs_doppler_term=record.max_abs_doppler_term,
        max_abs_spatial_coupling=record.max_abs_spatial_coupling,
        max_phase_dispersion_rad=record.max_phase_dispersion_rad,
        max_spatial_dispersion_m=record.max_spatial_dispersion_m,
        kinematic_residual_max_m=record.kinematic_residual_max_m,
        max_abs_velocity_m_per_s=record.max_abs_velocity_m_per_s,
        path_length_max_m=record.path_length_max_m,
        final_position_equation_validated=record.final_position_equation_validated,
        max_abs_velocity_equation_validated=(
            record.max_abs_velocity_equation_validated
        ),
        path_length_equation_validated=record.path_length_equation_validated,
        kinematic_equations_validated=record.kinematic_equations_validated,
        kinematic_summary_replay_tolerance=(record.kinematic_summary_replay_tolerance),
        min_phase_margin_rad=record.min_phase_margin_rad,
        min_spatial_margin_m=record.min_spatial_margin_m,
        min_phase_order_parameter=record.min_phase_order_parameter,
        max_distance_to_reference_m=record.max_distance_to_reference_m,
        reference_phase=record.reference_phase,
        reference_point=record.reference_point,
        phase_tol_rad=record.phase_tol_rad,
        spatial_tol_m=record.spatial_tol_m,
        tolerance_profile_name=record.tolerance_profile_name,
        tolerance_profile_multiplier=record.tolerance_profile_multiplier,
        required_consecutive_samples=record.required_consecutive_samples,
        moving_frame_backend_request=record.moving_frame_backend_request,
        omega_schedule_sha256=record.omega_schedule_sha256,
        velocity_schedule_sha256=record.velocity_schedule_sha256,
        phase_trajectory_sha256=record.phase_trajectory_sha256,
        position_trajectory_sha256=record.position_trajectory_sha256,
        initial_spatial_coupling_sha256=record.initial_spatial_coupling_sha256,
        final_spatial_coupling_sha256=record.final_spatial_coupling_sha256,
        doppler_trace_sha256=record.doppler_trace_sha256,
        timeline_sha256=record.timeline_sha256,
    )
    payload["acceptance_sha256"] = record.acceptance_sha256
    return payload

verify_pha_c_acceptance_record

verify_pha_c_acceptance_record(
    record: PHACAcceptanceRecord,
) -> PHACAcceptanceRecord

Replay and validate a complete PHA-C acceptance record.

Parameters

record : PHACAcceptanceRecord The PHA-C record to operate on.

Returns

PHACAcceptanceRecord The same record after fail-closed replay and validation.

Raises

ValueError If the record fails fail-closed replay or hash validation.

Source code in src/scpn_phase_orchestrator/upde/pha_c_acceptance.py
def verify_pha_c_acceptance_record(
    record: PHACAcceptanceRecord,
) -> PHACAcceptanceRecord:
    """Replay and validate a complete PHA-C acceptance record.

    Parameters
    ----------
    record : PHACAcceptanceRecord
        The PHA-C record to operate on.

    Returns
    -------
    PHACAcceptanceRecord
        The same record after fail-closed replay and validation.

    Raises
    ------
    ValueError
        If the record fails fail-closed replay or hash validation.
    """
    if not isinstance(record, PHACAcceptanceRecord):
        raise ValueError("record must be a PHACAcceptanceRecord")
    for field in (
        "omega_schedule_sha256",
        "velocity_schedule_sha256",
        "phase_trajectory_sha256",
        "position_trajectory_sha256",
        "initial_spatial_coupling_sha256",
        "final_spatial_coupling_sha256",
        "doppler_trace_sha256",
        "timeline_sha256",
    ):
        _validate_sha256_hex(getattr(record, field), name=field)
    acceptance_hash = _validate_sha256_hex(
        record.acceptance_sha256,
        name="acceptance_sha256",
    )
    if record.claim_boundary != PHA_C_ACCEPTANCE_CLAIM_BOUNDARY:
        raise ValueError("claim_boundary must be the PHA-C acceptance review boundary")
    if record.evidence_kind != PHA_C_ACCEPTANCE_EVIDENCE_KIND:
        raise ValueError("evidence_kind must be deterministic acceptance evidence")
    if (
        _validate_record_bool(
            record.execution_disabled,
            name="execution_disabled",
        )
        is not True
    ):
        raise ValueError("execution_disabled must be true")
    if _validate_record_bool(record.actuating, name="actuating") is not False:
        raise ValueError("actuating must be false")
    if (
        not isinstance(record.moving_frame_backend_request, str)
        or not record.moving_frame_backend_request
    ):
        raise ValueError("moving_frame_backend_request must be a non-empty string")

    sample_count = _validate_record_int(
        record.sample_count,
        name="sample_count",
        minimum=2,
    )
    step_count = _validate_record_int(record.step_count, name="step_count", minimum=1)
    if sample_count != step_count + 1:
        raise ValueError("sample_count must equal step_count + 1")
    _validate_record_int(record.oscillator_count, name="oscillator_count", minimum=1)
    required = _validate_record_int(
        record.required_consecutive_samples,
        name="required_consecutive_samples",
        minimum=1,
    )
    count_fields = (
        "lock_sample_count",
        "lock_loss_count",
        "reset_count",
        "max_consecutive_lock_samples",
    )
    counts = {
        field: _validate_record_int(getattr(record, field), name=field, minimum=0)
        for field in count_fields
    }
    if counts["lock_sample_count"] > sample_count:
        raise ValueError("lock_sample_count cannot exceed sample_count")
    if counts["max_consecutive_lock_samples"] > sample_count:
        raise ValueError("max_consecutive_lock_samples cannot exceed sample_count")
    for field in ("lock_loss_count", "reset_count"):
        if counts[field] > step_count:
            raise ValueError(f"{field} cannot exceed step_count")
    final_lock_achieved = _validate_record_bool(
        record.final_lock_achieved,
        name="final_lock_achieved",
    )
    if counts["max_consecutive_lock_samples"] < required and final_lock_achieved:
        raise ValueError("final_lock_achieved requires the consecutive threshold")

    start_time = _validate_real_scalar(record.start_time, name="start_time")
    end_time = _validate_real_scalar(record.end_time, name="end_time")
    dt = _validate_positive_scalar(record.dt, name="dt")
    if end_time < start_time:
        raise ValueError("end_time must be greater than or equal to start_time")
    if abs(end_time - (start_time + dt * step_count)) > 1.0e-12:
        raise ValueError("end_time must equal start_time + dt * step_count")
    first_lock_index = _validate_record_int(
        record.first_lock_index,
        name="first_lock_index",
        minimum=-1,
    )
    first_lock_observed = _validate_record_bool(
        record.first_lock_observed,
        name="first_lock_observed",
    )
    first_lock_time = _validate_real_scalar(
        record.first_lock_time,
        name="first_lock_time",
    )
    if first_lock_observed:
        if first_lock_index < 0 or first_lock_index >= sample_count:
            raise ValueError("first_lock_index must refer to an observed sample")
        if first_lock_time < start_time or first_lock_time > end_time:
            raise ValueError("first_lock_time must be inside the acceptance range")
    else:
        if first_lock_index != -1:
            raise ValueError("first_lock_index must be -1 when no lock is observed")
        if first_lock_time != 0.0:
            raise ValueError("first_lock_time must be 0.0 when no lock is observed")

    for field in (
        "max_abs_doppler_term",
        "max_abs_spatial_coupling",
        "max_phase_dispersion_rad",
        "max_spatial_dispersion_m",
        "kinematic_residual_max_m",
        "max_abs_velocity_m_per_s",
        "path_length_max_m",
        "max_distance_to_reference_m",
        "phase_tol_rad",
        "spatial_tol_m",
    ):
        _validate_nonnegative_record_scalar(getattr(record, field), name=field)
    max_phase_dispersion = _validate_nonnegative_record_scalar(
        record.max_phase_dispersion_rad,
        name="max_phase_dispersion_rad",
    )
    max_spatial_dispersion = _validate_nonnegative_record_scalar(
        record.max_spatial_dispersion_m,
        name="max_spatial_dispersion_m",
    )
    kinematic_residual = _validate_nonnegative_record_scalar(
        record.kinematic_residual_max_m,
        name="kinematic_residual_max_m",
    )
    if kinematic_residual > PHA_C_KINEMATIC_RESIDUAL_TOLERANCE_M:
        raise ValueError(
            "kinematic_residual_max_m must not exceed "
            f"{PHA_C_KINEMATIC_RESIDUAL_TOLERANCE_M} m"
        )
    for field in (
        "final_position_equation_validated",
        "max_abs_velocity_equation_validated",
        "path_length_equation_validated",
        "kinematic_equations_validated",
    ):
        if _validate_record_bool(getattr(record, field), name=field) is not True:
            raise ValueError(f"{field} must be true")
    kinematic_replay_tolerance = _validate_positive_scalar(
        record.kinematic_summary_replay_tolerance,
        name="kinematic_summary_replay_tolerance",
    )
    if (
        kinematic_replay_tolerance
        != PHA_C_ACCEPTANCE_KINEMATIC_SUMMARY_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "kinematic_summary_replay_tolerance must match the acceptance constant"
        )
    phase_tol = _validate_nonnegative_record_scalar(
        record.phase_tol_rad,
        name="phase_tol_rad",
    )
    spatial_tol = _validate_nonnegative_record_scalar(
        record.spatial_tol_m,
        name="spatial_tol_m",
    )
    min_phase_margin = _validate_real_scalar(
        record.min_phase_margin_rad,
        name="min_phase_margin_rad",
    )
    min_spatial_margin = _validate_real_scalar(
        record.min_spatial_margin_m,
        name="min_spatial_margin_m",
    )
    if (
        abs(min_phase_margin - (phase_tol - max_phase_dispersion))
        > PHA_C_ACCEPTANCE_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "min_phase_margin_rad must equal phase_tol_rad - max_phase_dispersion_rad"
        )
    if (
        abs(min_spatial_margin - (spatial_tol - max_spatial_dispersion))
        > PHA_C_ACCEPTANCE_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "min_spatial_margin_m must equal spatial_tol_m - max_spatial_dispersion_m"
        )
    order_parameter = _validate_nonnegative_record_scalar(
        record.min_phase_order_parameter,
        name="min_phase_order_parameter",
    )
    if order_parameter > 1.0 + 1.0e-12:
        raise ValueError("min_phase_order_parameter must be inside [0, 1]")
    multiplier = _validate_real_scalar(
        record.tolerance_profile_multiplier,
        name="tolerance_profile_multiplier",
    )
    if multiplier <= 0.0:
        raise ValueError("tolerance_profile_multiplier must be positive")
    if (
        not isinstance(record.tolerance_profile_name, str)
        or not record.tolerance_profile_name
    ):
        raise ValueError("tolerance_profile_name must be a non-empty string")
    for field in ("reference_phase", "reference_point"):
        _validate_real_scalar(getattr(record, field), name=field)

    payload = pha_c_acceptance_record_to_dict(record)
    replay_payload = dict(payload)
    replay_payload.pop("acceptance_sha256")
    if _sha256_json(replay_payload) != acceptance_hash:
        raise ValueError(
            "acceptance_sha256 does not match the canonical acceptance payload",
        )
    return record

API documentation

pha_c_acceptance

End-to-end PHA-C acceptance evidence.

The PHA-C work lane has several reviewed modules: spatial coupling, time-varying frequencies, Doppler correction, moving-frame integration, merge-window monitoring, handoff records, and event timelines. This module binds those surfaces into one deterministic, non-actuating acceptance record so review lanes can prove the complete chain was exercised rather than only one slice.

Classes

PHACAcceptanceRecord dataclass

PHACAcceptanceRecord(
    sample_count: int,
    step_count: int,
    oscillator_count: int,
    start_time: float,
    end_time: float,
    dt: float,
    first_lock_index: int,
    first_lock_time: float,
    first_lock_observed: bool,
    final_lock_achieved: bool,
    lock_sample_count: int,
    lock_loss_count: int,
    reset_count: int,
    max_consecutive_lock_samples: int,
    max_abs_doppler_term: float,
    max_abs_spatial_coupling: float,
    max_phase_dispersion_rad: float,
    max_spatial_dispersion_m: float,
    kinematic_residual_max_m: float,
    max_abs_velocity_m_per_s: float,
    path_length_max_m: float,
    final_position_equation_validated: bool,
    max_abs_velocity_equation_validated: bool,
    path_length_equation_validated: bool,
    kinematic_equations_validated: bool,
    kinematic_summary_replay_tolerance: float,
    min_phase_margin_rad: float,
    min_spatial_margin_m: float,
    min_phase_order_parameter: float,
    max_distance_to_reference_m: float,
    reference_phase: float,
    reference_point: float,
    phase_tol_rad: float,
    spatial_tol_m: float,
    tolerance_profile_name: str,
    tolerance_profile_multiplier: float,
    required_consecutive_samples: int,
    moving_frame_backend_request: str,
    claim_boundary: str,
    evidence_kind: str,
    execution_disabled: bool,
    actuating: bool,
    omega_schedule_sha256: str,
    velocity_schedule_sha256: str,
    phase_trajectory_sha256: str,
    position_trajectory_sha256: str,
    initial_spatial_coupling_sha256: str,
    final_spatial_coupling_sha256: str,
    doppler_trace_sha256: str,
    timeline_sha256: str,
    acceptance_sha256: str,
)

Audit-ready acceptance evidence for the complete PHA-C chain.

Methods:
to_dict
to_dict() -> dict[str, float | int | bool | str]

Return a JSON-safe canonical representation.

Returns

dict[str, float | int | bool | str] Return a JSON-safe canonical representation.

Source code in src/scpn_phase_orchestrator/upde/pha_c_acceptance.py
def to_dict(self) -> dict[str, float | int | bool | str]:
    """Return a JSON-safe canonical representation.

    Returns
    -------
    dict[str, float | int | bool | str]
        Return a JSON-safe canonical representation.
    """
    return pha_c_acceptance_record_to_dict(self)

Functions:

build_pha_c_acceptance_record

build_pha_c_acceptance_record(
    phases_t0: ArrayLike,
    positions_t0: ArrayLike,
    omega_schedule: ArrayLike,
    knm: ArrayLike,
    velocity_schedule: ArrayLike,
    *,
    alpha: object = 0.0,
    spatial_modulator: SpatialCouplingModulator
    | None = None,
    doppler_strength: object = 0.001,
    doppler_epsilon: object = 1e-09,
    zeta: object = 0.0,
    psi: object = 0.0,
    dt: object = 0.001,
    method: str = "rk4",
    n_substeps: object = 1,
    atol: object = 1e-09,
    rtol: object = 1e-09,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
    phase_tol_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_tol_m: object = DEFAULT_SPATIAL_TOL_M,
    required_consecutive_samples: object = 3,
    tolerance_profile: object | None = "baseline_1x",
    backend: object = "python",
) -> PHACAcceptanceRecord

Build deterministic end-to-end PHA-C acceptance evidence.

The builder advances a moving-frame trajectory one schedule row at a time, recording spatially modulated coupling and Doppler traces before converting the trajectory into a PHA-C event timeline. The output remains review-only; it is evidence for downstream gates and never permits actuation.

Parameters

phases_t0 : ArrayLike Initial oscillator phases in radians, shape (N,). positions_t0 : ArrayLike Initial axial coordinates per oscillator, shape (N,). omega_schedule : ArrayLike Per-step natural-frequency vectors, shape (n_steps, N). knm : ArrayLike Coupling matrix K_nm, shape (N, N). velocity_schedule : ArrayLike Per-step axial velocity vectors, shape (n_steps, N). alpha : object Phase-lag matrix in radians, shape (N, N), or None for no lag. spatial_modulator : SpatialCouplingModulator | None Configured spatial coupling modulator. doppler_strength : object Doppler coupling-correction strength. doppler_epsilon : object Numerical floor guarding the Doppler denominator. zeta : object External drive strength ζ. psi : object External drive reference phase Ψ in radians. dt : object Integration step size. method : str Integration method (euler, rk4, or rk45). n_substeps : object Number of inner substeps per outer step. atol : object Absolute tolerance for the adaptive (rk45) integrator. rtol : object Relative tolerance for the adaptive (rk45) integrator. reference_phase : object Reference phase for the lock criterion, in radians. reference_point : object Reference axial coordinate for the spatial-margin criterion. phase_tol_rad : object Phase lock tolerance in radians. spatial_tol_m : object Spatial lock tolerance in metres. required_consecutive_samples : object Consecutive in-tolerance samples required to declare lock. tolerance_profile : object | None Named tolerance profile, or None for the baseline profile. backend : object Name of the compute backend to run.

Returns

PHACAcceptanceRecord The deterministic end-to-end PHA-C acceptance record.

Raises

ValueError If any input is invalid or the acceptance replay fails.

Source code in src/scpn_phase_orchestrator/upde/pha_c_acceptance.py
def build_pha_c_acceptance_record(
    phases_t0: ArrayLike,
    positions_t0: ArrayLike,
    omega_schedule: ArrayLike,
    knm: ArrayLike,
    velocity_schedule: ArrayLike,
    *,
    alpha: object = 0.0,
    spatial_modulator: SpatialCouplingModulator | None = None,
    doppler_strength: object = 1.0e-3,
    doppler_epsilon: object = 1.0e-9,
    zeta: object = 0.0,
    psi: object = 0.0,
    dt: object = 1.0e-3,
    method: str = "rk4",
    n_substeps: object = 1,
    atol: object = 1.0e-9,
    rtol: object = 1.0e-9,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
    phase_tol_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_tol_m: object = DEFAULT_SPATIAL_TOL_M,
    required_consecutive_samples: object = 3,
    tolerance_profile: object | None = "baseline_1x",
    backend: object = "python",
) -> PHACAcceptanceRecord:
    """Build deterministic end-to-end PHA-C acceptance evidence.

    The builder advances a moving-frame trajectory one schedule row at a time,
    recording spatially modulated coupling and Doppler traces before converting
    the trajectory into a PHA-C event timeline. The output remains review-only;
    it is evidence for downstream gates and never permits actuation.

    Parameters
    ----------
    phases_t0 : ArrayLike
        Initial oscillator phases in radians, shape ``(N,)``.
    positions_t0 : ArrayLike
        Initial axial coordinates per oscillator, shape ``(N,)``.
    omega_schedule : ArrayLike
        Per-step natural-frequency vectors, shape ``(n_steps, N)``.
    knm : ArrayLike
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    velocity_schedule : ArrayLike
        Per-step axial velocity vectors, shape ``(n_steps, N)``.
    alpha : object
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    spatial_modulator : SpatialCouplingModulator | None
        Configured spatial coupling modulator.
    doppler_strength : object
        Doppler coupling-correction strength.
    doppler_epsilon : object
        Numerical floor guarding the Doppler denominator.
    zeta : object
        External drive strength ``ζ``.
    psi : object
        External drive reference phase ``Ψ`` in radians.
    dt : object
        Integration step size.
    method : str
        Integration method (``euler``, ``rk4``, or ``rk45``).
    n_substeps : object
        Number of inner substeps per outer step.
    atol : object
        Absolute tolerance for the adaptive (rk45) integrator.
    rtol : object
        Relative tolerance for the adaptive (rk45) integrator.
    reference_phase : object
        Reference phase for the lock criterion, in radians.
    reference_point : object
        Reference axial coordinate for the spatial-margin criterion.
    phase_tol_rad : object
        Phase lock tolerance in radians.
    spatial_tol_m : object
        Spatial lock tolerance in metres.
    required_consecutive_samples : object
        Consecutive in-tolerance samples required to declare lock.
    tolerance_profile : object | None
        Named tolerance profile, or ``None`` for the baseline profile.
    backend : object
        Name of the compute backend to run.

    Returns
    -------
    PHACAcceptanceRecord
        The deterministic end-to-end PHA-C acceptance record.

    Raises
    ------
    ValueError
        If any input is invalid or the acceptance replay fails.
    """
    phases = _as_float_vector(phases_t0, name="phases_t0")
    n = int(phases.size)
    positions = _as_float_vector(positions_t0, name="positions_t0")
    if positions.shape != phases.shape:
        raise ValueError("positions_t0 must have the same shape as phases_t0")
    omega = _as_schedule(omega_schedule, name="omega_schedule", n=n)
    velocities = _as_schedule(velocity_schedule, name="velocity_schedule", n=n)
    if velocities.shape[0] != omega.shape[0]:
        raise ValueError("velocity_schedule step count must match omega_schedule")
    k = _as_knm(knm, n=n)
    dt_s = _validate_positive_scalar(dt, name="dt")
    strength = _validate_real_scalar(doppler_strength, name="doppler_strength")
    doppler_eps = _validate_positive_scalar(doppler_epsilon, name="doppler_epsilon")
    zeta_f = _validate_real_scalar(zeta, name="zeta")
    psi_f = _validate_real_scalar(psi, name="psi")
    atol_f = _validate_positive_scalar(atol, name="atol")
    rtol_f = _validate_positive_scalar(rtol, name="rtol")
    backend_request = _validate_backend_request(backend)
    if isinstance(n_substeps, (bool, np.bool_)) or not isinstance(
        n_substeps,
        (int, np.integer),
    ):
        raise ValueError("n_substeps must be a positive integer")
    n_substeps_i = int(n_substeps)
    if n_substeps_i < 1:
        raise ValueError("n_substeps must be a positive integer")
    if spatial_modulator is None:
        modulator = SpatialCouplingModulator(K_base=1.0)
    elif isinstance(spatial_modulator, SpatialCouplingModulator):
        modulator = spatial_modulator
    else:
        raise ValueError("spatial_modulator must be a SpatialCouplingModulator")

    phase_rows: list[FloatArray] = [phases.copy()]
    position_rows: list[FloatArray] = [positions.copy()]
    spatial_rows: list[FloatArray] = []
    doppler_rows: list[FloatArray] = []
    current_phases = phases.copy()
    current_positions = positions.copy()
    for step in range(int(omega.shape[0])):
        spatial_coupling = modulator.modulate(k, current_positions)
        spatial_rows.append(spatial_coupling)
        correction = doppler_term(
            velocities[step],
            spatial_coupling,
            doppler_strength=strength,
            doppler_epsilon=doppler_eps,
        )
        doppler_rows.append(correction)
        state = moving_frame_run(
            current_phases,
            current_positions,
            omega[step : step + 1],
            k,
            alpha,
            velocities[step : step + 1],
            modulator,
            doppler_strength=strength,
            doppler_epsilon=doppler_eps,
            zeta=zeta_f,
            psi=psi_f,
            dt=dt_s,
            method=method,
            n_substeps=n_substeps_i,
            atol=atol_f,
            rtol=rtol_f,
            backend=backend_request,
        )
        current_phases = np.ascontiguousarray(state[:n], dtype=np.float64)
        current_positions = np.ascontiguousarray(state[n:], dtype=np.float64)
        phase_rows.append(current_phases)
        position_rows.append(current_positions)
    final_spatial_coupling = modulator.modulate(k, current_positions)

    phase_trajectory = np.vstack(phase_rows).astype(np.float64, copy=False)
    position_trajectory = np.vstack(position_rows).astype(np.float64, copy=False)
    doppler_trace = np.vstack(doppler_rows).astype(np.float64, copy=False)
    spatial_trace = np.stack(spatial_rows).astype(np.float64, copy=False)
    predicted_position_steps = position_trajectory[:-1] + velocities * dt_s
    kinematic_residual_max_m = float(
        np.max(np.abs(position_trajectory[1:] - predicted_position_steps))
    )
    expected_final_positions = positions + np.sum(velocities * dt_s, axis=0)
    final_position_residual_m = float(
        np.max(np.abs(position_trajectory[-1] - expected_final_positions))
    )
    max_abs_velocity_m_per_s = float(np.max(np.abs(velocities)))
    replayed_max_abs_velocity_m_per_s = float(np.max(np.abs(velocities)))
    expected_path_length_max_m = float(
        np.max(np.sum(np.abs(velocities * dt_s), axis=0))
    )
    path_length_max_m = expected_path_length_max_m
    replayed_path_length_max_m = float(
        np.max(np.sum(np.abs(velocities * dt_s), axis=0))
    )
    final_position_equation_validated = (
        final_position_residual_m <= PHA_C_KINEMATIC_RESIDUAL_TOLERANCE_M
    )
    max_abs_velocity_equation_validated = (
        abs(max_abs_velocity_m_per_s - replayed_max_abs_velocity_m_per_s)
        <= PHA_C_ACCEPTANCE_KINEMATIC_SUMMARY_REPLAY_TOLERANCE
    )
    path_length_equation_validated = (
        abs(path_length_max_m - replayed_path_length_max_m)
        <= PHA_C_ACCEPTANCE_KINEMATIC_SUMMARY_REPLAY_TOLERANCE
    )
    kinematic_equations_validated = (
        final_position_equation_validated
        and max_abs_velocity_equation_validated
        and path_length_equation_validated
    )
    times = np.arange(phase_trajectory.shape[0], dtype=np.float64) * dt_s
    timeline = build_pha_c_event_timeline(
        phase_trajectory,
        position_trajectory,
        times=times,
        reference_phase=reference_phase,
        reference_point=reference_point,
        phase_tol_rad=phase_tol_rad,
        spatial_tol_m=spatial_tol_m,
        required_consecutive_samples=required_consecutive_samples,
        tolerance_profile=tolerance_profile,
    )
    first_spatial_coupling = spatial_rows[0]
    payload = _record_dict_without_hash(
        sample_count=timeline.sample_count,
        step_count=int(omega.shape[0]),
        oscillator_count=n,
        start_time=float(times[0]),
        end_time=float(times[-1]),
        dt=dt_s,
        first_lock_index=timeline.first_lock_index,
        first_lock_time=timeline.first_lock_time,
        first_lock_observed=timeline.first_lock_observed,
        final_lock_achieved=timeline.final_lock_achieved,
        lock_sample_count=timeline.lock_sample_count,
        lock_loss_count=timeline.lock_loss_count,
        reset_count=timeline.reset_count,
        max_consecutive_lock_samples=timeline.max_consecutive_lock_samples,
        max_abs_doppler_term=float(np.max(np.abs(doppler_trace))),
        max_abs_spatial_coupling=float(np.max(np.abs(spatial_trace))),
        max_phase_dispersion_rad=timeline.max_phase_dispersion_rad,
        max_spatial_dispersion_m=timeline.max_spatial_dispersion_m,
        kinematic_residual_max_m=kinematic_residual_max_m,
        max_abs_velocity_m_per_s=max_abs_velocity_m_per_s,
        path_length_max_m=path_length_max_m,
        final_position_equation_validated=final_position_equation_validated,
        max_abs_velocity_equation_validated=max_abs_velocity_equation_validated,
        path_length_equation_validated=path_length_equation_validated,
        kinematic_equations_validated=kinematic_equations_validated,
        kinematic_summary_replay_tolerance=(
            PHA_C_ACCEPTANCE_KINEMATIC_SUMMARY_REPLAY_TOLERANCE
        ),
        min_phase_margin_rad=timeline.min_phase_margin_rad,
        min_spatial_margin_m=timeline.min_spatial_margin_m,
        min_phase_order_parameter=timeline.min_phase_order_parameter,
        max_distance_to_reference_m=timeline.max_distance_to_reference_m,
        reference_phase=timeline.reference_phase,
        reference_point=timeline.reference_point,
        phase_tol_rad=timeline.phase_tol_rad,
        spatial_tol_m=timeline.spatial_tol_m,
        tolerance_profile_name=timeline.tolerance_profile_name,
        tolerance_profile_multiplier=timeline.tolerance_profile_multiplier,
        required_consecutive_samples=timeline.required_consecutive_samples,
        moving_frame_backend_request=backend_request,
        omega_schedule_sha256=_array_sha256(omega),
        velocity_schedule_sha256=_array_sha256(velocities),
        phase_trajectory_sha256=_array_sha256(phase_trajectory),
        position_trajectory_sha256=_array_sha256(position_trajectory),
        initial_spatial_coupling_sha256=_array_sha256(first_spatial_coupling),
        final_spatial_coupling_sha256=_array_sha256(final_spatial_coupling),
        doppler_trace_sha256=_array_sha256(doppler_trace),
        timeline_sha256=timeline.timeline_sha256,
    )
    return PHACAcceptanceRecord(
        **cast(Any, payload),
        acceptance_sha256=_sha256_json(payload),
    )

pha_c_acceptance_record_to_dict

pha_c_acceptance_record_to_dict(
    record: PHACAcceptanceRecord,
) -> dict[str, float | int | bool | str]

Return the canonical JSON-safe PHA-C acceptance payload.

Parameters

record : PHACAcceptanceRecord The PHA-C record to operate on.

Returns

dict[str, float | int | bool | str] The canonical JSON-safe PHA-C acceptance payload.

Source code in src/scpn_phase_orchestrator/upde/pha_c_acceptance.py
def pha_c_acceptance_record_to_dict(
    record: PHACAcceptanceRecord,
) -> dict[str, float | int | bool | str]:
    """Return the canonical JSON-safe PHA-C acceptance payload.

    Parameters
    ----------
    record : PHACAcceptanceRecord
        The PHA-C record to operate on.

    Returns
    -------
    dict[str, float | int | bool | str]
        The canonical JSON-safe PHA-C acceptance payload.
    """
    payload = _record_dict_without_hash(
        sample_count=record.sample_count,
        step_count=record.step_count,
        oscillator_count=record.oscillator_count,
        start_time=record.start_time,
        end_time=record.end_time,
        dt=record.dt,
        first_lock_index=record.first_lock_index,
        first_lock_time=record.first_lock_time,
        first_lock_observed=record.first_lock_observed,
        final_lock_achieved=record.final_lock_achieved,
        lock_sample_count=record.lock_sample_count,
        lock_loss_count=record.lock_loss_count,
        reset_count=record.reset_count,
        max_consecutive_lock_samples=record.max_consecutive_lock_samples,
        max_abs_doppler_term=record.max_abs_doppler_term,
        max_abs_spatial_coupling=record.max_abs_spatial_coupling,
        max_phase_dispersion_rad=record.max_phase_dispersion_rad,
        max_spatial_dispersion_m=record.max_spatial_dispersion_m,
        kinematic_residual_max_m=record.kinematic_residual_max_m,
        max_abs_velocity_m_per_s=record.max_abs_velocity_m_per_s,
        path_length_max_m=record.path_length_max_m,
        final_position_equation_validated=record.final_position_equation_validated,
        max_abs_velocity_equation_validated=(
            record.max_abs_velocity_equation_validated
        ),
        path_length_equation_validated=record.path_length_equation_validated,
        kinematic_equations_validated=record.kinematic_equations_validated,
        kinematic_summary_replay_tolerance=(record.kinematic_summary_replay_tolerance),
        min_phase_margin_rad=record.min_phase_margin_rad,
        min_spatial_margin_m=record.min_spatial_margin_m,
        min_phase_order_parameter=record.min_phase_order_parameter,
        max_distance_to_reference_m=record.max_distance_to_reference_m,
        reference_phase=record.reference_phase,
        reference_point=record.reference_point,
        phase_tol_rad=record.phase_tol_rad,
        spatial_tol_m=record.spatial_tol_m,
        tolerance_profile_name=record.tolerance_profile_name,
        tolerance_profile_multiplier=record.tolerance_profile_multiplier,
        required_consecutive_samples=record.required_consecutive_samples,
        moving_frame_backend_request=record.moving_frame_backend_request,
        omega_schedule_sha256=record.omega_schedule_sha256,
        velocity_schedule_sha256=record.velocity_schedule_sha256,
        phase_trajectory_sha256=record.phase_trajectory_sha256,
        position_trajectory_sha256=record.position_trajectory_sha256,
        initial_spatial_coupling_sha256=record.initial_spatial_coupling_sha256,
        final_spatial_coupling_sha256=record.final_spatial_coupling_sha256,
        doppler_trace_sha256=record.doppler_trace_sha256,
        timeline_sha256=record.timeline_sha256,
    )
    payload["acceptance_sha256"] = record.acceptance_sha256
    return payload

verify_pha_c_acceptance_record

verify_pha_c_acceptance_record(
    record: PHACAcceptanceRecord,
) -> PHACAcceptanceRecord

Replay and validate a complete PHA-C acceptance record.

Parameters

record : PHACAcceptanceRecord The PHA-C record to operate on.

Returns

PHACAcceptanceRecord The same record after fail-closed replay and validation.

Raises

ValueError If the record fails fail-closed replay or hash validation.

Source code in src/scpn_phase_orchestrator/upde/pha_c_acceptance.py
def verify_pha_c_acceptance_record(
    record: PHACAcceptanceRecord,
) -> PHACAcceptanceRecord:
    """Replay and validate a complete PHA-C acceptance record.

    Parameters
    ----------
    record : PHACAcceptanceRecord
        The PHA-C record to operate on.

    Returns
    -------
    PHACAcceptanceRecord
        The same record after fail-closed replay and validation.

    Raises
    ------
    ValueError
        If the record fails fail-closed replay or hash validation.
    """
    if not isinstance(record, PHACAcceptanceRecord):
        raise ValueError("record must be a PHACAcceptanceRecord")
    for field in (
        "omega_schedule_sha256",
        "velocity_schedule_sha256",
        "phase_trajectory_sha256",
        "position_trajectory_sha256",
        "initial_spatial_coupling_sha256",
        "final_spatial_coupling_sha256",
        "doppler_trace_sha256",
        "timeline_sha256",
    ):
        _validate_sha256_hex(getattr(record, field), name=field)
    acceptance_hash = _validate_sha256_hex(
        record.acceptance_sha256,
        name="acceptance_sha256",
    )
    if record.claim_boundary != PHA_C_ACCEPTANCE_CLAIM_BOUNDARY:
        raise ValueError("claim_boundary must be the PHA-C acceptance review boundary")
    if record.evidence_kind != PHA_C_ACCEPTANCE_EVIDENCE_KIND:
        raise ValueError("evidence_kind must be deterministic acceptance evidence")
    if (
        _validate_record_bool(
            record.execution_disabled,
            name="execution_disabled",
        )
        is not True
    ):
        raise ValueError("execution_disabled must be true")
    if _validate_record_bool(record.actuating, name="actuating") is not False:
        raise ValueError("actuating must be false")
    if (
        not isinstance(record.moving_frame_backend_request, str)
        or not record.moving_frame_backend_request
    ):
        raise ValueError("moving_frame_backend_request must be a non-empty string")

    sample_count = _validate_record_int(
        record.sample_count,
        name="sample_count",
        minimum=2,
    )
    step_count = _validate_record_int(record.step_count, name="step_count", minimum=1)
    if sample_count != step_count + 1:
        raise ValueError("sample_count must equal step_count + 1")
    _validate_record_int(record.oscillator_count, name="oscillator_count", minimum=1)
    required = _validate_record_int(
        record.required_consecutive_samples,
        name="required_consecutive_samples",
        minimum=1,
    )
    count_fields = (
        "lock_sample_count",
        "lock_loss_count",
        "reset_count",
        "max_consecutive_lock_samples",
    )
    counts = {
        field: _validate_record_int(getattr(record, field), name=field, minimum=0)
        for field in count_fields
    }
    if counts["lock_sample_count"] > sample_count:
        raise ValueError("lock_sample_count cannot exceed sample_count")
    if counts["max_consecutive_lock_samples"] > sample_count:
        raise ValueError("max_consecutive_lock_samples cannot exceed sample_count")
    for field in ("lock_loss_count", "reset_count"):
        if counts[field] > step_count:
            raise ValueError(f"{field} cannot exceed step_count")
    final_lock_achieved = _validate_record_bool(
        record.final_lock_achieved,
        name="final_lock_achieved",
    )
    if counts["max_consecutive_lock_samples"] < required and final_lock_achieved:
        raise ValueError("final_lock_achieved requires the consecutive threshold")

    start_time = _validate_real_scalar(record.start_time, name="start_time")
    end_time = _validate_real_scalar(record.end_time, name="end_time")
    dt = _validate_positive_scalar(record.dt, name="dt")
    if end_time < start_time:
        raise ValueError("end_time must be greater than or equal to start_time")
    if abs(end_time - (start_time + dt * step_count)) > 1.0e-12:
        raise ValueError("end_time must equal start_time + dt * step_count")
    first_lock_index = _validate_record_int(
        record.first_lock_index,
        name="first_lock_index",
        minimum=-1,
    )
    first_lock_observed = _validate_record_bool(
        record.first_lock_observed,
        name="first_lock_observed",
    )
    first_lock_time = _validate_real_scalar(
        record.first_lock_time,
        name="first_lock_time",
    )
    if first_lock_observed:
        if first_lock_index < 0 or first_lock_index >= sample_count:
            raise ValueError("first_lock_index must refer to an observed sample")
        if first_lock_time < start_time or first_lock_time > end_time:
            raise ValueError("first_lock_time must be inside the acceptance range")
    else:
        if first_lock_index != -1:
            raise ValueError("first_lock_index must be -1 when no lock is observed")
        if first_lock_time != 0.0:
            raise ValueError("first_lock_time must be 0.0 when no lock is observed")

    for field in (
        "max_abs_doppler_term",
        "max_abs_spatial_coupling",
        "max_phase_dispersion_rad",
        "max_spatial_dispersion_m",
        "kinematic_residual_max_m",
        "max_abs_velocity_m_per_s",
        "path_length_max_m",
        "max_distance_to_reference_m",
        "phase_tol_rad",
        "spatial_tol_m",
    ):
        _validate_nonnegative_record_scalar(getattr(record, field), name=field)
    max_phase_dispersion = _validate_nonnegative_record_scalar(
        record.max_phase_dispersion_rad,
        name="max_phase_dispersion_rad",
    )
    max_spatial_dispersion = _validate_nonnegative_record_scalar(
        record.max_spatial_dispersion_m,
        name="max_spatial_dispersion_m",
    )
    kinematic_residual = _validate_nonnegative_record_scalar(
        record.kinematic_residual_max_m,
        name="kinematic_residual_max_m",
    )
    if kinematic_residual > PHA_C_KINEMATIC_RESIDUAL_TOLERANCE_M:
        raise ValueError(
            "kinematic_residual_max_m must not exceed "
            f"{PHA_C_KINEMATIC_RESIDUAL_TOLERANCE_M} m"
        )
    for field in (
        "final_position_equation_validated",
        "max_abs_velocity_equation_validated",
        "path_length_equation_validated",
        "kinematic_equations_validated",
    ):
        if _validate_record_bool(getattr(record, field), name=field) is not True:
            raise ValueError(f"{field} must be true")
    kinematic_replay_tolerance = _validate_positive_scalar(
        record.kinematic_summary_replay_tolerance,
        name="kinematic_summary_replay_tolerance",
    )
    if (
        kinematic_replay_tolerance
        != PHA_C_ACCEPTANCE_KINEMATIC_SUMMARY_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "kinematic_summary_replay_tolerance must match the acceptance constant"
        )
    phase_tol = _validate_nonnegative_record_scalar(
        record.phase_tol_rad,
        name="phase_tol_rad",
    )
    spatial_tol = _validate_nonnegative_record_scalar(
        record.spatial_tol_m,
        name="spatial_tol_m",
    )
    min_phase_margin = _validate_real_scalar(
        record.min_phase_margin_rad,
        name="min_phase_margin_rad",
    )
    min_spatial_margin = _validate_real_scalar(
        record.min_spatial_margin_m,
        name="min_spatial_margin_m",
    )
    if (
        abs(min_phase_margin - (phase_tol - max_phase_dispersion))
        > PHA_C_ACCEPTANCE_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "min_phase_margin_rad must equal phase_tol_rad - max_phase_dispersion_rad"
        )
    if (
        abs(min_spatial_margin - (spatial_tol - max_spatial_dispersion))
        > PHA_C_ACCEPTANCE_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "min_spatial_margin_m must equal spatial_tol_m - max_spatial_dispersion_m"
        )
    order_parameter = _validate_nonnegative_record_scalar(
        record.min_phase_order_parameter,
        name="min_phase_order_parameter",
    )
    if order_parameter > 1.0 + 1.0e-12:
        raise ValueError("min_phase_order_parameter must be inside [0, 1]")
    multiplier = _validate_real_scalar(
        record.tolerance_profile_multiplier,
        name="tolerance_profile_multiplier",
    )
    if multiplier <= 0.0:
        raise ValueError("tolerance_profile_multiplier must be positive")
    if (
        not isinstance(record.tolerance_profile_name, str)
        or not record.tolerance_profile_name
    ):
        raise ValueError("tolerance_profile_name must be a non-empty string")
    for field in ("reference_phase", "reference_point"):
        _validate_real_scalar(getattr(record, field), name=field)

    payload = pha_c_acceptance_record_to_dict(record)
    replay_payload = dict(payload)
    replay_payload.pop("acceptance_sha256")
    if _sha256_json(replay_payload) != acceptance_hash:
        raise ValueError(
            "acceptance_sha256 does not match the canonical acceptance payload",
        )
    return record