UPDE — PHA-C Handoff

PHACHandoffRecord is the PHA-C event/state bridge between moving-frame simulation, merge-window monitoring, replay, MIF import, and Studio review. It converts one phase-plus-position sample into deterministic scalar evidence: phase dispersion, spatial dispersion, signed phase/spatial safety margins, order parameter, lock status, source digests, and a canonical record hash.

The handoff is intentionally review-only. It does not write to actuators, modify coupling, schedule hardware, or mutate supervisor state.

For trajectory-level evidence, use PHACTimelineRecord. The timeline consumes the same handoff contract across all samples and reports first lock, lock loss, reset counts, tolerance-profile provenance, and a canonical trajectory hash.

Use cases

Use the PHA-C handoff when a downstream lane needs a compact, verifiable event record instead of raw phase and position arrays:

• MIF/FRC merger review where lock evidence must be carried into a separate field-reconstruction or chamber-analysis lane;
• Studio dashboards that show whether a moving-frame run reached a reviewed phase-space merge window without exposing full state vectors;
• deterministic replay where the same phase/position sample must regenerate the same event hash;
• operator evidence streams where the event can be logged before a human or policy layer decides whether any later action is allowed;
• polyglot benchmark gates that compare Python, Rust, Go, Julia, and Mojo source-contract behavior without making throughput claims.

Contract

The handoff first evaluates MergeWindowMonitor semantics:

phase_locked   = max_i |wrap(theta_i - theta_ref)| <= phase_tol_rad
spatial_locked = max_i |z_i - z_ref| <= spatial_tol_m
phase_margin   = phase_tol_rad - phase_dispersion_rad
spatial_margin = spatial_tol_m - spatial_dispersion_m
lock_achieved  = consecutive joint locks >= required_consecutive_samples

It then adds:

• signed margins that expose the distance to the reviewed phase and spatial envelopes, with negative values for failed predicates;
• phase_order_parameter = |mean(exp(i theta_i))|;
• distance_to_reference_max_m = max_i |z_i - z_ref|;
• the tolerance profile name and multiplier when baseline_1x, buffer_3x, or review_5x is used;
• SHA-256 digests for the phase vector, position vector, merge report, source-chain, and final record;
• execution_disabled=True and actuating=False;
• the fixed claim boundary pha_c_event_state_handoff_review_only.

The record contains scalar evidence and hashes only. It avoids serialising full phase or position vectors into public evidence records.

Minimal example

import numpy as np
from scpn_phase_orchestrator.upde.pha_c_handoff import (
    build_pha_c_handoff_record,
    verify_pha_c_handoff_record,
)

record = build_pha_c_handoff_record(
    np.array([0.0, 0.003, -0.004]),
    np.array([0.0, 0.0005, -0.0008]),
    t=4.0,
    phase_tol_rad=0.01,
    spatial_tol_m=0.002,
    required_consecutive_samples=3,
    prior_consecutive_lock_samples=2,
    tolerance_profile="baseline_1x",
)

assert record.lock_achieved
assert record.execution_disabled
assert not record.actuating
event_payload = record.to_dict()
verify_pha_c_handoff_record(record)

Use verify_pha_c_handoff_record(...) when replaying a stored record. It rechecks the review-only claim boundary, non-actuating flags, SHA-256 field formats, scalar lock invariants, signed margin equations, and canonical record hash without requiring the original phase or position vectors. The signed margin replay tolerance is published as PHA_C_HANDOFF_MARGIN_REPLAY_TOLERANCE; both phase and spatial margins must replay as tolerance - dispersion inside that bound.

The Rust, Go, Julia, and Mojo source-contract rows are validated before canonical dictionary projection. Raw PHACHandoffRecord fields must keep their declared domains: numeric evidence must be finite real non-boolean scalars, counts must be integers, lock and non-actuation flags must be plain booleans, and provenance/hash fields must be strings. Numeric strings, boolean aliases, and bytes-like string substitutes are rejected before to_dict() can coerce them into a matching canonical payload.

Polyglot parity

The benchmark gate records Rust, Mojo, Julia, Go, and Python source-contract slots. The current handoff path is evidence construction, not a numerical hot loop, so the non-Python slots validate parity against the Python reference contract. If native kernels are later added, they must preserve the same hashes signed margins, signed-margin equations, and fail-closed input boundaries. The benchmark's maximum-error helper uses the same strict raw-field parser as the source-contract validator, so malformed fields cannot be hidden by canonical payload coercion.

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

Committed benchmark JSON is local regression evidence only. It is not a production timing claim unless rerun under the benchmark-isolation protocol. The payload exposes phase_margin_equation_validated, spatial_margin_equation_validated, signed_margin_equations_validated, and margin_replay_tolerance; the parity gate fails unless every declared backend row proves the phase and spatial margin equations.

Failure boundaries

The handoff fails closed on:

• empty, non-finite, complex, object-dtype, or boolean phase/position vectors;
• mismatched phase and position lengths;
• non-finite timestamps or references;
• negative tolerances;
• invalid consecutive-sample controls.
• source-contract records with numeric strings, non-finite raw numeric fields, boolean aliases, malformed integer fields, non-plain booleans, or non-string provenance/hash fields.

PHACHandoffRecord dataclass

PHACHandoffRecord(
    t: float,
    oscillator_count: int,
    phase_dispersion_rad: float,
    spatial_dispersion_m: float,
    phase_margin_rad: float,
    spatial_margin_m: float,
    phase_locked: bool,
    spatial_locked: bool,
    lock_achieved: bool,
    consecutive_lock_samples: int,
    phase_order_parameter: float,
    distance_to_reference_max_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,
    claim_boundary: str,
    evidence_kind: str,
    execution_disabled: bool,
    actuating: bool,
    phase_state_sha256: str,
    position_state_sha256: str,
    merge_report_sha256: str,
    source_chain_sha256: str,
    record_sha256: str,
)

Audit-ready downstream handoff for a PHA-C moving-frame sample.

The record intentionally carries only scalar evidence and SHA-256 digests of the sampled phase/position vectors. It is suitable for replay and review lanes, but it never authorises actuation.

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_handoff.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_handoff_record_to_dict(self)

build_pha_c_handoff_record

build_pha_c_handoff_record(
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    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,
    prior_consecutive_lock_samples: object = 0,
    tolerance_profile: object | None = None,
) -> PHACHandoffRecord

Build a deterministic PHA-C event/state handoff record.

The handoff consumes the same phase/position sample as :func:evaluate_merge_window, mirrors its fail-closed validation, adds Kuramoto order-parameter and source-chain digests, then returns a non-actuating record with a canonical hash.

Parameters

phases : ArrayLike Oscillator phases in radians, shape (N,). positions : ArrayLike Absolute axial coordinates per oscillator, shape (N,). t : object Absolute time of the sample in seconds. 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. prior_consecutive_lock_samples : object Consecutive lock-sample count carried in from a prior window. tolerance_profile : object | None Named tolerance profile, or None for the baseline profile.

Returns

PHACHandoffRecord The deterministic PHA-C event/state handoff record.

Raises

ValueError If any input is invalid.

Source code in src/scpn_phase_orchestrator/upde/pha_c_handoff.py

def build_pha_c_handoff_record(
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    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,
    prior_consecutive_lock_samples: object = 0,
    tolerance_profile: object | None = None,
) -> PHACHandoffRecord:
    """Build a deterministic PHA-C event/state handoff record.

    The handoff consumes the same phase/position sample as
    :func:`evaluate_merge_window`, mirrors its fail-closed validation, adds
    Kuramoto order-parameter and source-chain digests, then returns a
    non-actuating record with a canonical hash.

    Parameters
    ----------
    phases : ArrayLike
        Oscillator phases in radians, shape ``(N,)``.
    positions : ArrayLike
        Absolute axial coordinates per oscillator, shape ``(N,)``.
    t : object
        Absolute time of the sample in seconds.
    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.
    prior_consecutive_lock_samples : object
        Consecutive lock-sample count carried in from a prior window.
    tolerance_profile : object | None
        Named tolerance profile, or ``None`` for the baseline profile.

    Returns
    -------
    PHACHandoffRecord
        The deterministic PHA-C event/state handoff record.

    Raises
    ------
    ValueError
        If any input is invalid.
    """
    phase_vector = _as_float_vector(phases, name="phases")
    position_vector = _as_float_vector(positions, name="positions")
    if position_vector.shape != phase_vector.shape:
        raise ValueError("positions must have the same one-dimensional shape as phases")

    timestamp = _validate_real_scalar(t, name="t")
    phase_reference = _validate_real_scalar(reference_phase, name="reference_phase")
    spatial_reference = _validate_real_scalar(reference_point, name="reference_point")
    tolerance_profile_name = "explicit"
    tolerance_profile_multiplier = 1.0
    if tolerance_profile is None:
        phase_tol = _validate_tolerance(phase_tol_rad, name="phase_tol_rad")
        spatial_tol = _validate_tolerance(spatial_tol_m, name="spatial_tol_m")
    else:
        profile = resolve_merge_window_tolerance_profile(
            tolerance_profile,
            phase_baseline_rad=phase_tol_rad,
            spatial_baseline_m=spatial_tol_m,
        )
        phase_tol = profile.phase_tol_rad
        spatial_tol = profile.spatial_tol_m
        tolerance_profile_name = profile.name
        tolerance_profile_multiplier = profile.multiplier
    required = _validate_sample_count(
        required_consecutive_samples,
        name="required_consecutive_samples",
        minimum=1,
    )
    prior = _validate_sample_count(
        prior_consecutive_lock_samples,
        name="prior_consecutive_lock_samples",
        minimum=0,
    )

    report = evaluate_merge_window(
        phase_vector,
        position_vector,
        t=timestamp,
        reference_phase=phase_reference,
        reference_point=spatial_reference,
        phase_tol_rad=phase_tol,
        spatial_tol_m=spatial_tol,
        required_consecutive_samples=required,
        prior_consecutive_lock_samples=prior,
    )
    phase_hash = _vector_sha256(phase_vector)
    position_hash = _vector_sha256(position_vector)
    merge_hash = _merge_report_sha256(report)
    source_chain_hash = _sha256_json(
        {
            "phase_state_sha256": phase_hash,
            "position_state_sha256": position_hash,
            "merge_report_sha256": merge_hash,
        }
    )
    order_parameter = _phase_order_parameter(phase_vector)
    max_distance = _distance_to_reference_max(position_vector, spatial_reference)
    record_payload = _record_dict_without_hash(
        report=report,
        oscillator_count=int(phase_vector.size),
        phase_order_parameter=order_parameter,
        distance_to_reference_max_m=max_distance,
        reference_phase=phase_reference,
        reference_point=spatial_reference,
        phase_tol_rad=phase_tol,
        spatial_tol_m=spatial_tol,
        tolerance_profile_name=tolerance_profile_name,
        tolerance_profile_multiplier=tolerance_profile_multiplier,
        required_consecutive_samples=required,
        phase_state_sha256=phase_hash,
        position_state_sha256=position_hash,
        merge_report_sha256=merge_hash,
        source_chain_sha256=source_chain_hash,
    )
    record_hash = _sha256_json(record_payload)
    return PHACHandoffRecord(
        **cast(Any, record_payload),
        record_sha256=record_hash,
    )

pha_c_handoff_record_to_dict

pha_c_handoff_record_to_dict(
    record: PHACHandoffRecord,
) -> dict[str, float | int | bool | str]

Convert a :class:PHACHandoffRecord into a JSON-safe dictionary.

Parameters

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

Returns

dict[str, float | int | bool | str] The JSON-safe handoff record dictionary.

Source code in src/scpn_phase_orchestrator/upde/pha_c_handoff.py

def pha_c_handoff_record_to_dict(
    record: PHACHandoffRecord,
) -> dict[str, float | int | bool | str]:
    """Convert a :class:`PHACHandoffRecord` into a JSON-safe dictionary.

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

    Returns
    -------
    dict[str, float | int | bool | str]
        The JSON-safe handoff record dictionary.
    """
    return {
        "t": float(record.t),
        "oscillator_count": int(record.oscillator_count),
        "phase_dispersion_rad": float(record.phase_dispersion_rad),
        "spatial_dispersion_m": float(record.spatial_dispersion_m),
        "phase_margin_rad": float(record.phase_margin_rad),
        "spatial_margin_m": float(record.spatial_margin_m),
        "phase_locked": bool(record.phase_locked),
        "spatial_locked": bool(record.spatial_locked),
        "lock_achieved": bool(record.lock_achieved),
        "consecutive_lock_samples": int(record.consecutive_lock_samples),
        "phase_order_parameter": float(record.phase_order_parameter),
        "distance_to_reference_max_m": float(record.distance_to_reference_max_m),
        "reference_phase": float(record.reference_phase),
        "reference_point": float(record.reference_point),
        "phase_tol_rad": float(record.phase_tol_rad),
        "spatial_tol_m": float(record.spatial_tol_m),
        "tolerance_profile_name": str(record.tolerance_profile_name),
        "tolerance_profile_multiplier": float(record.tolerance_profile_multiplier),
        "required_consecutive_samples": int(record.required_consecutive_samples),
        "claim_boundary": str(record.claim_boundary),
        "evidence_kind": str(record.evidence_kind),
        "execution_disabled": bool(record.execution_disabled),
        "actuating": bool(record.actuating),
        "phase_state_sha256": str(record.phase_state_sha256),
        "position_state_sha256": str(record.position_state_sha256),
        "merge_report_sha256": str(record.merge_report_sha256),
        "source_chain_sha256": str(record.source_chain_sha256),
        "record_sha256": str(record.record_sha256),
    }

verify_pha_c_handoff_record

verify_pha_c_handoff_record(
    record: PHACHandoffRecord,
) -> PHACHandoffRecord

Replay and validate a PHA-C handoff record hash and safety boundary.

The verifier is intentionally independent of the builder path: it checks the scalar invariants carried by an existing record, validates all SHA-256 digest fields, and recomputes the canonical payload hash from to_dict(). This lets benchmark gates, replay ledgers, and downstream MIF/FRC lanes reject tampered evidence without access to the original phase and position vectors.

Parameters

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

Returns

PHACHandoffRecord The same record after replay and safety-boundary validation.

Raises

ValueError If the record fails replay or safety-boundary validation.

Source code in src/scpn_phase_orchestrator/upde/pha_c_handoff.py

def verify_pha_c_handoff_record(record: PHACHandoffRecord) -> PHACHandoffRecord:
    """Replay and validate a PHA-C handoff record hash and safety boundary.

    The verifier is intentionally independent of the builder path: it checks the
    scalar invariants carried by an existing record, validates all SHA-256 digest
    fields, and recomputes the canonical payload hash from ``to_dict()``. This
    lets benchmark gates, replay ledgers, and downstream MIF/FRC lanes reject
    tampered evidence without access to the original phase and position vectors.

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

    Returns
    -------
    PHACHandoffRecord
        The same record after replay and safety-boundary validation.

    Raises
    ------
    ValueError
        If the record fails replay or safety-boundary validation.
    """
    if not isinstance(record, PHACHandoffRecord):
        raise ValueError("record must be a PHACHandoffRecord")
    _validate_sha256_hex(record.phase_state_sha256, name="phase_state_sha256")
    _validate_sha256_hex(record.position_state_sha256, name="position_state_sha256")
    _validate_sha256_hex(record.merge_report_sha256, name="merge_report_sha256")
    _validate_sha256_hex(record.source_chain_sha256, name="source_chain_sha256")
    record_hash = _validate_sha256_hex(record.record_sha256, name="record_sha256")
    if record.claim_boundary != PHA_C_HANDOFF_CLAIM_BOUNDARY:
        raise ValueError("claim_boundary must be the PHA-C handoff review boundary")
    if record.evidence_kind != PHA_C_HANDOFF_EVIDENCE_KIND:
        raise ValueError("evidence_kind must be deterministic handoff 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")

    _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,
    )
    consecutive = _validate_record_int(
        record.consecutive_lock_samples,
        name="consecutive_lock_samples",
        minimum=0,
    )
    phase_locked = _validate_record_bool(record.phase_locked, name="phase_locked")
    spatial_locked = _validate_record_bool(record.spatial_locked, name="spatial_locked")
    lock_achieved = _validate_record_bool(record.lock_achieved, name="lock_achieved")
    if lock_achieved and (not phase_locked or not spatial_locked):
        raise ValueError("lock_achieved requires phase and spatial locks")
    if lock_achieved and consecutive < required:
        raise ValueError("lock_achieved requires the consecutive-sample threshold")
    if (not phase_locked or not spatial_locked) and consecutive != 0:
        raise ValueError("unlocked samples must reset consecutive_lock_samples")

    for field in (
        "phase_dispersion_rad",
        "spatial_dispersion_m",
        "distance_to_reference_max_m",
        "phase_tol_rad",
        "spatial_tol_m",
    ):
        _validate_nonnegative_record_scalar(getattr(record, field), name=field)
    phase_dispersion = _validate_nonnegative_record_scalar(
        record.phase_dispersion_rad,
        name="phase_dispersion_rad",
    )
    spatial_dispersion = _validate_nonnegative_record_scalar(
        record.spatial_dispersion_m,
        name="spatial_dispersion_m",
    )
    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",
    )
    phase_margin = _validate_real_scalar(
        record.phase_margin_rad,
        name="phase_margin_rad",
    )
    spatial_margin = _validate_real_scalar(
        record.spatial_margin_m,
        name="spatial_margin_m",
    )
    if (
        abs(phase_margin - (phase_tol - phase_dispersion))
        > PHA_C_HANDOFF_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "phase_margin_rad must equal phase_tol_rad - phase_dispersion_rad"
        )
    if (
        abs(spatial_margin - (spatial_tol - spatial_dispersion))
        > PHA_C_HANDOFF_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "spatial_margin_m must equal spatial_tol_m - spatial_dispersion_m"
        )
    if phase_locked and phase_margin < -PHA_C_HANDOFF_MARGIN_REPLAY_TOLERANCE:
        raise ValueError("phase_locked requires a non-negative phase_margin_rad")
    if not phase_locked and phase_margin >= 0.0:
        raise ValueError("phase-unlocked records require a negative phase_margin_rad")
    if spatial_locked and spatial_margin < -PHA_C_HANDOFF_MARGIN_REPLAY_TOLERANCE:
        raise ValueError("spatial_locked requires a non-negative spatial_margin_m")
    if not spatial_locked and spatial_margin >= 0.0:
        raise ValueError("spatial-unlocked records require a negative spatial_margin_m")
    order_parameter = _validate_nonnegative_record_scalar(
        record.phase_order_parameter,
        name="phase_order_parameter",
    )
    if order_parameter > 1.0 + 1.0e-12:
        raise ValueError("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 ("t", "reference_phase", "reference_point"):
        _validate_real_scalar(getattr(record, field), name=field)

    payload = pha_c_handoff_record_to_dict(record)
    replay_payload = dict(payload)
    replay_payload.pop("record_sha256")
    if _sha256_json(replay_payload) != record_hash:
        raise ValueError("record_sha256 does not match the canonical handoff payload")
    return record

API documentation

pha_c_handoff

Deterministic PHA-C event/state handoff records.

The PHA-C moving-frame lane produces phases and axial positions. The merge window monitor decides whether that phase-space state is inside the reviewed merge tolerance. This module binds those two surfaces into a non-actuating, hash-stable handoff record for downstream replay, Studio review, MIF import, or operator evidence streams.

Classes

PHACHandoffRecord dataclass

PHACHandoffRecord(
    t: float,
    oscillator_count: int,
    phase_dispersion_rad: float,
    spatial_dispersion_m: float,
    phase_margin_rad: float,
    spatial_margin_m: float,
    phase_locked: bool,
    spatial_locked: bool,
    lock_achieved: bool,
    consecutive_lock_samples: int,
    phase_order_parameter: float,
    distance_to_reference_max_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,
    claim_boundary: str,
    evidence_kind: str,
    execution_disabled: bool,
    actuating: bool,
    phase_state_sha256: str,
    position_state_sha256: str,
    merge_report_sha256: str,
    source_chain_sha256: str,
    record_sha256: str,
)

Audit-ready downstream handoff for a PHA-C moving-frame sample.

The record intentionally carries only scalar evidence and SHA-256 digests of the sampled phase/position vectors. It is suitable for replay and review lanes, but it never authorises actuation.

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_handoff.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_handoff_record_to_dict(self)

Functions:

build_pha_c_handoff_record

build_pha_c_handoff_record(
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    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,
    prior_consecutive_lock_samples: object = 0,
    tolerance_profile: object | None = None,
) -> PHACHandoffRecord

Build a deterministic PHA-C event/state handoff record.

The handoff consumes the same phase/position sample as :func:evaluate_merge_window, mirrors its fail-closed validation, adds Kuramoto order-parameter and source-chain digests, then returns a non-actuating record with a canonical hash.

Parameters

phases : ArrayLike Oscillator phases in radians, shape (N,). positions : ArrayLike Absolute axial coordinates per oscillator, shape (N,). t : object Absolute time of the sample in seconds. 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. prior_consecutive_lock_samples : object Consecutive lock-sample count carried in from a prior window. tolerance_profile : object | None Named tolerance profile, or None for the baseline profile.

Returns

PHACHandoffRecord The deterministic PHA-C event/state handoff record.

Raises

ValueError If any input is invalid.

Source code in src/scpn_phase_orchestrator/upde/pha_c_handoff.py

def build_pha_c_handoff_record(
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    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,
    prior_consecutive_lock_samples: object = 0,
    tolerance_profile: object | None = None,
) -> PHACHandoffRecord:
    """Build a deterministic PHA-C event/state handoff record.

    The handoff consumes the same phase/position sample as
    :func:`evaluate_merge_window`, mirrors its fail-closed validation, adds
    Kuramoto order-parameter and source-chain digests, then returns a
    non-actuating record with a canonical hash.

    Parameters
    ----------
    phases : ArrayLike
        Oscillator phases in radians, shape ``(N,)``.
    positions : ArrayLike
        Absolute axial coordinates per oscillator, shape ``(N,)``.
    t : object
        Absolute time of the sample in seconds.
    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.
    prior_consecutive_lock_samples : object
        Consecutive lock-sample count carried in from a prior window.
    tolerance_profile : object | None
        Named tolerance profile, or ``None`` for the baseline profile.

    Returns
    -------
    PHACHandoffRecord
        The deterministic PHA-C event/state handoff record.

    Raises
    ------
    ValueError
        If any input is invalid.
    """
    phase_vector = _as_float_vector(phases, name="phases")
    position_vector = _as_float_vector(positions, name="positions")
    if position_vector.shape != phase_vector.shape:
        raise ValueError("positions must have the same one-dimensional shape as phases")

    timestamp = _validate_real_scalar(t, name="t")
    phase_reference = _validate_real_scalar(reference_phase, name="reference_phase")
    spatial_reference = _validate_real_scalar(reference_point, name="reference_point")
    tolerance_profile_name = "explicit"
    tolerance_profile_multiplier = 1.0
    if tolerance_profile is None:
        phase_tol = _validate_tolerance(phase_tol_rad, name="phase_tol_rad")
        spatial_tol = _validate_tolerance(spatial_tol_m, name="spatial_tol_m")
    else:
        profile = resolve_merge_window_tolerance_profile(
            tolerance_profile,
            phase_baseline_rad=phase_tol_rad,
            spatial_baseline_m=spatial_tol_m,
        )
        phase_tol = profile.phase_tol_rad
        spatial_tol = profile.spatial_tol_m
        tolerance_profile_name = profile.name
        tolerance_profile_multiplier = profile.multiplier
    required = _validate_sample_count(
        required_consecutive_samples,
        name="required_consecutive_samples",
        minimum=1,
    )
    prior = _validate_sample_count(
        prior_consecutive_lock_samples,
        name="prior_consecutive_lock_samples",
        minimum=0,
    )

    report = evaluate_merge_window(
        phase_vector,
        position_vector,
        t=timestamp,
        reference_phase=phase_reference,
        reference_point=spatial_reference,
        phase_tol_rad=phase_tol,
        spatial_tol_m=spatial_tol,
        required_consecutive_samples=required,
        prior_consecutive_lock_samples=prior,
    )
    phase_hash = _vector_sha256(phase_vector)
    position_hash = _vector_sha256(position_vector)
    merge_hash = _merge_report_sha256(report)
    source_chain_hash = _sha256_json(
        {
            "phase_state_sha256": phase_hash,
            "position_state_sha256": position_hash,
            "merge_report_sha256": merge_hash,
        }
    )
    order_parameter = _phase_order_parameter(phase_vector)
    max_distance = _distance_to_reference_max(position_vector, spatial_reference)
    record_payload = _record_dict_without_hash(
        report=report,
        oscillator_count=int(phase_vector.size),
        phase_order_parameter=order_parameter,
        distance_to_reference_max_m=max_distance,
        reference_phase=phase_reference,
        reference_point=spatial_reference,
        phase_tol_rad=phase_tol,
        spatial_tol_m=spatial_tol,
        tolerance_profile_name=tolerance_profile_name,
        tolerance_profile_multiplier=tolerance_profile_multiplier,
        required_consecutive_samples=required,
        phase_state_sha256=phase_hash,
        position_state_sha256=position_hash,
        merge_report_sha256=merge_hash,
        source_chain_sha256=source_chain_hash,
    )
    record_hash = _sha256_json(record_payload)
    return PHACHandoffRecord(
        **cast(Any, record_payload),
        record_sha256=record_hash,
    )

pha_c_handoff_record_to_dict

pha_c_handoff_record_to_dict(
    record: PHACHandoffRecord,
) -> dict[str, float | int | bool | str]

Convert a :class:PHACHandoffRecord into a JSON-safe dictionary.

Parameters

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

Returns

dict[str, float | int | bool | str] The JSON-safe handoff record dictionary.

Source code in src/scpn_phase_orchestrator/upde/pha_c_handoff.py

def pha_c_handoff_record_to_dict(
    record: PHACHandoffRecord,
) -> dict[str, float | int | bool | str]:
    """Convert a :class:`PHACHandoffRecord` into a JSON-safe dictionary.

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

    Returns
    -------
    dict[str, float | int | bool | str]
        The JSON-safe handoff record dictionary.
    """
    return {
        "t": float(record.t),
        "oscillator_count": int(record.oscillator_count),
        "phase_dispersion_rad": float(record.phase_dispersion_rad),
        "spatial_dispersion_m": float(record.spatial_dispersion_m),
        "phase_margin_rad": float(record.phase_margin_rad),
        "spatial_margin_m": float(record.spatial_margin_m),
        "phase_locked": bool(record.phase_locked),
        "spatial_locked": bool(record.spatial_locked),
        "lock_achieved": bool(record.lock_achieved),
        "consecutive_lock_samples": int(record.consecutive_lock_samples),
        "phase_order_parameter": float(record.phase_order_parameter),
        "distance_to_reference_max_m": float(record.distance_to_reference_max_m),
        "reference_phase": float(record.reference_phase),
        "reference_point": float(record.reference_point),
        "phase_tol_rad": float(record.phase_tol_rad),
        "spatial_tol_m": float(record.spatial_tol_m),
        "tolerance_profile_name": str(record.tolerance_profile_name),
        "tolerance_profile_multiplier": float(record.tolerance_profile_multiplier),
        "required_consecutive_samples": int(record.required_consecutive_samples),
        "claim_boundary": str(record.claim_boundary),
        "evidence_kind": str(record.evidence_kind),
        "execution_disabled": bool(record.execution_disabled),
        "actuating": bool(record.actuating),
        "phase_state_sha256": str(record.phase_state_sha256),
        "position_state_sha256": str(record.position_state_sha256),
        "merge_report_sha256": str(record.merge_report_sha256),
        "source_chain_sha256": str(record.source_chain_sha256),
        "record_sha256": str(record.record_sha256),
    }

verify_pha_c_handoff_record

verify_pha_c_handoff_record(
    record: PHACHandoffRecord,
) -> PHACHandoffRecord

Replay and validate a PHA-C handoff record hash and safety boundary.

The verifier is intentionally independent of the builder path: it checks the scalar invariants carried by an existing record, validates all SHA-256 digest fields, and recomputes the canonical payload hash from to_dict(). This lets benchmark gates, replay ledgers, and downstream MIF/FRC lanes reject tampered evidence without access to the original phase and position vectors.

Parameters

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

Returns

PHACHandoffRecord The same record after replay and safety-boundary validation.

Raises

ValueError If the record fails replay or safety-boundary validation.

Source code in src/scpn_phase_orchestrator/upde/pha_c_handoff.py

def verify_pha_c_handoff_record(record: PHACHandoffRecord) -> PHACHandoffRecord:
    """Replay and validate a PHA-C handoff record hash and safety boundary.

    The verifier is intentionally independent of the builder path: it checks the
    scalar invariants carried by an existing record, validates all SHA-256 digest
    fields, and recomputes the canonical payload hash from ``to_dict()``. This
    lets benchmark gates, replay ledgers, and downstream MIF/FRC lanes reject
    tampered evidence without access to the original phase and position vectors.

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

    Returns
    -------
    PHACHandoffRecord
        The same record after replay and safety-boundary validation.

    Raises
    ------
    ValueError
        If the record fails replay or safety-boundary validation.
    """
    if not isinstance(record, PHACHandoffRecord):
        raise ValueError("record must be a PHACHandoffRecord")
    _validate_sha256_hex(record.phase_state_sha256, name="phase_state_sha256")
    _validate_sha256_hex(record.position_state_sha256, name="position_state_sha256")
    _validate_sha256_hex(record.merge_report_sha256, name="merge_report_sha256")
    _validate_sha256_hex(record.source_chain_sha256, name="source_chain_sha256")
    record_hash = _validate_sha256_hex(record.record_sha256, name="record_sha256")
    if record.claim_boundary != PHA_C_HANDOFF_CLAIM_BOUNDARY:
        raise ValueError("claim_boundary must be the PHA-C handoff review boundary")
    if record.evidence_kind != PHA_C_HANDOFF_EVIDENCE_KIND:
        raise ValueError("evidence_kind must be deterministic handoff 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")

    _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,
    )
    consecutive = _validate_record_int(
        record.consecutive_lock_samples,
        name="consecutive_lock_samples",
        minimum=0,
    )
    phase_locked = _validate_record_bool(record.phase_locked, name="phase_locked")
    spatial_locked = _validate_record_bool(record.spatial_locked, name="spatial_locked")
    lock_achieved = _validate_record_bool(record.lock_achieved, name="lock_achieved")
    if lock_achieved and (not phase_locked or not spatial_locked):
        raise ValueError("lock_achieved requires phase and spatial locks")
    if lock_achieved and consecutive < required:
        raise ValueError("lock_achieved requires the consecutive-sample threshold")
    if (not phase_locked or not spatial_locked) and consecutive != 0:
        raise ValueError("unlocked samples must reset consecutive_lock_samples")

    for field in (
        "phase_dispersion_rad",
        "spatial_dispersion_m",
        "distance_to_reference_max_m",
        "phase_tol_rad",
        "spatial_tol_m",
    ):
        _validate_nonnegative_record_scalar(getattr(record, field), name=field)
    phase_dispersion = _validate_nonnegative_record_scalar(
        record.phase_dispersion_rad,
        name="phase_dispersion_rad",
    )
    spatial_dispersion = _validate_nonnegative_record_scalar(
        record.spatial_dispersion_m,
        name="spatial_dispersion_m",
    )
    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",
    )
    phase_margin = _validate_real_scalar(
        record.phase_margin_rad,
        name="phase_margin_rad",
    )
    spatial_margin = _validate_real_scalar(
        record.spatial_margin_m,
        name="spatial_margin_m",
    )
    if (
        abs(phase_margin - (phase_tol - phase_dispersion))
        > PHA_C_HANDOFF_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "phase_margin_rad must equal phase_tol_rad - phase_dispersion_rad"
        )
    if (
        abs(spatial_margin - (spatial_tol - spatial_dispersion))
        > PHA_C_HANDOFF_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "spatial_margin_m must equal spatial_tol_m - spatial_dispersion_m"
        )
    if phase_locked and phase_margin < -PHA_C_HANDOFF_MARGIN_REPLAY_TOLERANCE:
        raise ValueError("phase_locked requires a non-negative phase_margin_rad")
    if not phase_locked and phase_margin >= 0.0:
        raise ValueError("phase-unlocked records require a negative phase_margin_rad")
    if spatial_locked and spatial_margin < -PHA_C_HANDOFF_MARGIN_REPLAY_TOLERANCE:
        raise ValueError("spatial_locked requires a non-negative spatial_margin_m")
    if not spatial_locked and spatial_margin >= 0.0:
        raise ValueError("spatial-unlocked records require a negative spatial_margin_m")
    order_parameter = _validate_nonnegative_record_scalar(
        record.phase_order_parameter,
        name="phase_order_parameter",
    )
    if order_parameter > 1.0 + 1.0e-12:
        raise ValueError("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 ("t", "reference_phase", "reference_point"):
        _validate_real_scalar(getattr(record, field), name=field)

    payload = pha_c_handoff_record_to_dict(record)
    replay_payload = dict(payload)
    replay_payload.pop("record_sha256")
    if _sha256_json(replay_payload) != record_hash:
        raise ValueError("record_sha256 does not match the canonical handoff payload")
    return record