UPDE — PHA-C Event Timeline¶

PHACTimelineRecord is the trajectory-level PHA-C evidence surface. It consumes phase and axial-position matrices from a moving-frame run, threads the single-sample PHACHandoffRecord contract through time, and emits a compact lock/loss/reset timeline for downstream review lanes.

The timeline is intentionally review-only. It never writes to actuators, changes coupling, schedules hardware, or mutates supervisor state.

For the full PHA-C chain from spatial modulation through moving-frame dynamics and timeline hashing, use PHACAcceptanceRecord.

Use cases¶

Use the PHA-C event timeline when a downstream lane needs event evidence across an entire trajectory instead of a single sample:

MIF/FRC review where the first lock time, lock losses, and reset count must be carried into chamber or field-reconstruction analysis;
Studio panels that need a stable summary of when a moving-frame run entered or left a reviewed merge window;
replay ledgers where every sample handoff hash must roll into one canonical trajectory hash;
safety reviews that need to separate observed lock evidence from any later policy or operator action;
polyglot parity gates that compare Python, Rust, Go, Julia, and Mojo source-contract behavior for the same trajectory.

Contract¶

Inputs are two real finite matrices with shape (time, oscillator):

phases_by_step[t, i]    = wrapped or unwrapped phase for oscillator i at sample t
positions_by_step[t, i] = axial position for oscillator i at sample t
times[t]                = strictly increasing sample time

For each sample, the builder calls build_pha_c_handoff_record(...) and carries consecutive_lock_samples forward. The final timeline records:

first_lock_index and first_lock_time for the first achieved consecutive joint lock;
final_lock_achieved for the last sample;
phase-lock, spatial-lock, full-lock, lock-loss, and reset counts;
maximum phase dispersion, maximum spatial dispersion, minimum Kuramoto order parameter, maximum distance to the reference position, and minimum signed phase/spatial margins over the trajectory;
replay validation that each minimum signed margin equals the resolved tolerance minus the corresponding maximum dispersion under PHA_C_TIMELINE_MARGIN_REPLAY_TOLERANCE;
tolerance profile name, multiplier, resolved phase tolerance, and resolved spatial tolerance;
SHA-256 digests for the time vector, all sample records, transition table, and final timeline;
execution_disabled=True, actuating=False, and the fixed claim boundary pha_c_event_timeline_review_only.

Minimal example¶

import numpy as np
from scpn_phase_orchestrator.upde.pha_c_timeline import (
    build_pha_c_event_timeline,
    verify_pha_c_event_timeline,
)

phases = np.array(
    [
        [-0.02, 0.0, 0.02],
        [-0.002, 0.0, 0.002],
        [-0.0015, 0.0, 0.0015],
        [-0.001, 0.0, 0.001],
        [-0.02, 0.0, 0.02],
    ]
)
positions = np.array(
    [
        [-0.003, 0.0, 0.003],
        [-0.0005, 0.0, 0.0005],
        [-0.0004, 0.0, 0.0004],
        [-0.0003, 0.0, 0.0003],
        [-0.003, 0.0, 0.003],
    ]
)
times = np.arange(phases.shape[0]) * 0.5

timeline = build_pha_c_event_timeline(
    phases,
    positions,
    times=times,
    phase_tol_rad=0.01,
    spatial_tol_m=0.002,
    required_consecutive_samples=3,
    tolerance_profile="baseline_1x",
)

assert timeline.first_lock_index == 3
assert timeline.lock_loss_count == 1
assert timeline.execution_disabled
assert not timeline.actuating
evidence_payload = timeline.to_dict()
verify_pha_c_event_timeline(timeline)

Use verify_pha_c_event_timeline(...) when replaying a stored trajectory record. It rechecks timeline counts, first-lock semantics, transition-count bounds, review-only flags, signed margin equations, SHA-256 fields, and the canonical timeline hash without requiring raw trajectory matrices. The signed margin replay rejects records whose positive-looking phase or spatial margin no longer matches tolerance - maximum_dispersion.

The Rust, Go, Julia, and Mojo source-contract rows are validated before canonical dictionary projection. Raw PHACTimelineRecord fields must keep their declared domains: numeric evidence must be finite real non-boolean scalars, counts and indexes must be integers, observation/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.

Handoff versus timeline¶

Surface	Scope	Main output	Use when
`PHACHandoffRecord`	one phase/position sample	scalar lock evidence plus sample hashes	a downstream lane needs one reviewed event-state atom
`PHACTimelineRecord`	complete trajectory	lock acquisition, loss, reset, profile, and trajectory hashes	a downstream lane needs replayable event history

The timeline is built from handoff records. If native accelerator kernels are added later, they must preserve both the per-sample handoff hashes and the final timeline hash, including the minimum signed margins.

Polyglot parity¶

The benchmark gate records Rust, Mojo, Julia, Go, and Python source-contract slots. The current timeline 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 payload publishes phase_margin_equation_validated, spatial_margin_equation_validated, signed_margin_equations_validated, and margin_replay_tolerance for every backend row. 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_timeline_benchmark.py \
  --parity-gate \
  --calls 1 \
  --output benchmarks/results/pha_c_timeline.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 timeline fails closed on:

empty, non-finite, complex, object-dtype, or boolean trajectory matrices;
non-matrix phase or position inputs;
mismatched phase and position shapes;
missing, non-finite, non-vector, wrong-length, or non-increasing time vectors;
negative tolerances;
invalid consecutive-sample controls;
unknown tolerance profile names.
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.

PHACTimelineRecord `dataclass` ¶

PHACTimelineRecord(
    sample_count: int,
    oscillator_count: int,
    start_time: float,
    end_time: float,
    duration_s: float,
    first_lock_index: int,
    first_lock_time: float,
    first_lock_observed: bool,
    final_lock_achieved: bool,
    lock_sample_count: int,
    phase_lock_sample_count: int,
    spatial_lock_sample_count: int,
    lock_loss_count: int,
    reset_count: int,
    max_consecutive_lock_samples: int,
    max_phase_dispersion_rad: float,
    max_spatial_dispersion_m: 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,
    claim_boundary: str,
    evidence_kind: str,
    execution_disabled: bool,
    actuating: bool,
    time_state_sha256: str,
    sample_records_sha256: str,
    transition_table_sha256: str,
    timeline_sha256: str,
)

Audit-ready PHA-C trajectory event timeline.

The record stores scalar trajectory evidence plus SHA-256 digests of the time vector, per-sample handoff records, transition table, and final timeline. It is designed for review and replay lanes, not for direct control output.

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_timeline.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_event_timeline_to_dict(self)

build_pha_c_event_timeline ¶

build_pha_c_event_timeline(
    phases_by_step: ArrayLike,
    positions_by_step: ArrayLike,
    *,
    times: ArrayLike | None = None,
    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 = None,
) -> PHACTimelineRecord

Build deterministic PHA-C lock/loss timeline evidence.

Parameters mirror :func:build_pha_c_handoff_record, but consume complete trajectory matrices with shape (time, oscillator). The consecutive-lock counter is threaded through every sample so downstream consumers can review acquisition, loss, and reset events without raw state arrays.

Parameters¶

phases_by_step : ArrayLike Phase history, shape (n_steps, N). positions_by_step : ArrayLike Position history, shape (n_steps, N). times : ArrayLike | None Per-step timestamps in seconds, or None for unit spacing. 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.

Returns¶

PHACTimelineRecord The deterministic PHA-C lock/loss timeline record.

Raises¶

ValueError If any input is invalid.

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

def build_pha_c_event_timeline(
    phases_by_step: ArrayLike,
    positions_by_step: ArrayLike,
    *,
    times: ArrayLike | None = None,
    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 = None,
) -> PHACTimelineRecord:
    """Build deterministic PHA-C lock/loss timeline evidence.

    Parameters mirror :func:`build_pha_c_handoff_record`, but consume complete
    trajectory matrices with shape ``(time, oscillator)``. The consecutive-lock
    counter is threaded through every sample so downstream consumers can review
    acquisition, loss, and reset events without raw state arrays.

    Parameters
    ----------
    phases_by_step : ArrayLike
        Phase history, shape ``(n_steps, N)``.
    positions_by_step : ArrayLike
        Position history, shape ``(n_steps, N)``.
    times : ArrayLike | None
        Per-step timestamps in seconds, or ``None`` for unit spacing.
    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.

    Returns
    -------
    PHACTimelineRecord
        The deterministic PHA-C lock/loss timeline record.

    Raises
    ------
    ValueError
        If any input is invalid.
    """
    phase_matrix = _as_float_matrix(phases_by_step, name="phases_by_step")
    position_matrix = _as_float_matrix(positions_by_step, name="positions_by_step")
    if position_matrix.shape != phase_matrix.shape:
        raise ValueError("positions_by_step must have the same shape as phases_by_step")

    sample_count, oscillator_count = phase_matrix.shape
    time_vector = _as_time_vector(times, sample_count=sample_count)

    records: list[PHACHandoffRecord] = []
    consecutive = 0
    for index, timestamp in enumerate(time_vector):
        record = build_pha_c_handoff_record(
            phase_matrix[index],
            position_matrix[index],
            t=float(timestamp),
            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,
            prior_consecutive_lock_samples=consecutive,
            tolerance_profile=tolerance_profile,
        )
        consecutive = record.consecutive_lock_samples
        records.append(record)

    first = records[0]
    first_lock_index = next(
        (index for index, record in enumerate(records) if record.lock_achieved),
        -1,
    )
    first_lock_observed = first_lock_index >= 0
    first_lock_time = (
        float(time_vector[first_lock_index]) if first_lock_observed else 0.0
    )
    transition_rows = _transition_table(records)
    record_dicts = [record.to_dict() for record in records]

    sample_records_sha256 = _sha256_json(record_dicts)
    time_state_sha256 = _vector_sha256(time_vector)
    transition_table_sha256 = _sha256_json(transition_rows)
    timeline_payload = _timeline_dict_without_hash(
        sample_count=sample_count,
        oscillator_count=oscillator_count,
        start_time=float(time_vector[0]),
        end_time=float(time_vector[-1]),
        duration_s=float(time_vector[-1] - time_vector[0]),
        first_lock_index=first_lock_index,
        first_lock_time=first_lock_time,
        first_lock_observed=first_lock_observed,
        final_lock_achieved=records[-1].lock_achieved,
        lock_sample_count=sum(record.lock_achieved for record in records),
        phase_lock_sample_count=sum(record.phase_locked for record in records),
        spatial_lock_sample_count=sum(record.spatial_locked for record in records),
        lock_loss_count=sum(row["lock_lost"] for row in transition_rows),
        reset_count=sum(row["reset"] for row in transition_rows),
        max_consecutive_lock_samples=max(
            record.consecutive_lock_samples for record in records
        ),
        max_phase_dispersion_rad=max(record.phase_dispersion_rad for record in records),
        max_spatial_dispersion_m=max(record.spatial_dispersion_m for record in records),
        min_phase_margin_rad=min(record.phase_margin_rad for record in records),
        min_spatial_margin_m=min(record.spatial_margin_m for record in records),
        min_phase_order_parameter=min(
            record.phase_order_parameter for record in records
        ),
        max_distance_to_reference_m=max(
            record.distance_to_reference_max_m for record in records
        ),
        reference_phase=first.reference_phase,
        reference_point=first.reference_point,
        phase_tol_rad=first.phase_tol_rad,
        spatial_tol_m=first.spatial_tol_m,
        tolerance_profile_name=first.tolerance_profile_name,
        tolerance_profile_multiplier=first.tolerance_profile_multiplier,
        required_consecutive_samples=first.required_consecutive_samples,
        time_state_sha256=time_state_sha256,
        sample_records_sha256=sample_records_sha256,
        transition_table_sha256=transition_table_sha256,
    )
    return PHACTimelineRecord(
        **cast(Any, timeline_payload),
        timeline_sha256=_sha256_json(timeline_payload),
    )

pha_c_event_timeline_to_dict ¶

pha_c_event_timeline_to_dict(
    timeline: PHACTimelineRecord,
) -> dict[str, float | int | bool | str]

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

Parameters¶

timeline : PHACTimelineRecord The PHA-C event-timeline record to operate on.

Returns¶

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

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

def pha_c_event_timeline_to_dict(
    timeline: PHACTimelineRecord,
) -> dict[str, float | int | bool | str]:
    """Return the canonical JSON-safe PHA-C timeline payload.

    Parameters
    ----------
    timeline : PHACTimelineRecord
        The PHA-C event-timeline record to operate on.

    Returns
    -------
    dict[str, float | int | bool | str]
        The canonical JSON-safe PHA-C timeline payload.
    """
    payload = _timeline_dict_without_hash(
        sample_count=timeline.sample_count,
        oscillator_count=timeline.oscillator_count,
        start_time=timeline.start_time,
        end_time=timeline.end_time,
        duration_s=timeline.duration_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,
        phase_lock_sample_count=timeline.phase_lock_sample_count,
        spatial_lock_sample_count=timeline.spatial_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_phase_dispersion_rad=timeline.max_phase_dispersion_rad,
        max_spatial_dispersion_m=timeline.max_spatial_dispersion_m,
        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,
        time_state_sha256=timeline.time_state_sha256,
        sample_records_sha256=timeline.sample_records_sha256,
        transition_table_sha256=timeline.transition_table_sha256,
    )
    payload["timeline_sha256"] = timeline.timeline_sha256
    return payload

verify_pha_c_event_timeline ¶

verify_pha_c_event_timeline(
    timeline: PHACTimelineRecord,
) -> PHACTimelineRecord

Replay and validate a PHA-C event timeline hash and safety boundary.

Parameters¶

timeline : PHACTimelineRecord The PHA-C event-timeline record to operate on.

Returns¶

PHACTimelineRecord The same timeline after replay and safety-boundary validation.

Raises¶

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

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

def verify_pha_c_event_timeline(
    timeline: PHACTimelineRecord,
) -> PHACTimelineRecord:
    """Replay and validate a PHA-C event timeline hash and safety boundary.

    Parameters
    ----------
    timeline : PHACTimelineRecord
        The PHA-C event-timeline record to operate on.

    Returns
    -------
    PHACTimelineRecord
        The same timeline after replay and safety-boundary validation.

    Raises
    ------
    ValueError
        If the timeline fails replay or safety-boundary validation.
    """
    if not isinstance(timeline, PHACTimelineRecord):
        raise ValueError("timeline must be a PHACTimelineRecord")
    _validate_sha256_hex(timeline.time_state_sha256, name="time_state_sha256")
    _validate_sha256_hex(
        timeline.sample_records_sha256,
        name="sample_records_sha256",
    )
    _validate_sha256_hex(
        timeline.transition_table_sha256,
        name="transition_table_sha256",
    )
    timeline_hash = _validate_sha256_hex(
        timeline.timeline_sha256,
        name="timeline_sha256",
    )
    if timeline.claim_boundary != PHA_C_TIMELINE_CLAIM_BOUNDARY:
        raise ValueError("claim_boundary must be the PHA-C timeline review boundary")
    if timeline.evidence_kind != PHA_C_TIMELINE_EVIDENCE_KIND:
        raise ValueError("evidence_kind must be deterministic timeline evidence")
    if (
        _validate_record_bool(
            timeline.execution_disabled,
            name="execution_disabled",
        )
        is not True
    ):
        raise ValueError("execution_disabled must be true")
    if _validate_record_bool(timeline.actuating, name="actuating") is not False:
        raise ValueError("actuating must be false")

    sample_count = _validate_record_int(
        timeline.sample_count,
        name="sample_count",
        minimum=1,
    )
    _validate_record_int(timeline.oscillator_count, name="oscillator_count", minimum=1)
    required = _validate_record_int(
        timeline.required_consecutive_samples,
        name="required_consecutive_samples",
        minimum=1,
    )
    count_fields = (
        "lock_sample_count",
        "phase_lock_sample_count",
        "spatial_lock_sample_count",
        "lock_loss_count",
        "reset_count",
        "max_consecutive_lock_samples",
    )
    counts = {
        field: _validate_record_int(getattr(timeline, field), name=field, minimum=0)
        for field in count_fields
    }
    for field in (
        "lock_sample_count",
        "phase_lock_sample_count",
        "spatial_lock_sample_count",
        "max_consecutive_lock_samples",
    ):
        if counts[field] > sample_count:
            raise ValueError(f"{field} cannot exceed sample_count")
    for field in ("lock_loss_count", "reset_count"):
        if counts[field] > max(sample_count - 1, 0):
            raise ValueError(f"{field} cannot exceed the transition count")
    final_lock_achieved = _validate_record_bool(
        timeline.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(timeline.start_time, name="start_time")
    end_time = _validate_real_scalar(timeline.end_time, name="end_time")
    duration_s = _validate_nonnegative_record_scalar(
        timeline.duration_s,
        name="duration_s",
    )
    if end_time < start_time:
        raise ValueError("end_time must be greater than or equal to start_time")
    if abs(duration_s - (end_time - start_time)) > 1.0e-12:
        raise ValueError("duration_s must equal end_time - start_time")
    first_lock_index = _validate_record_int(
        timeline.first_lock_index,
        name="first_lock_index",
        minimum=-1,
    )
    first_lock_observed = _validate_record_bool(
        timeline.first_lock_observed,
        name="first_lock_observed",
    )
    first_lock_time = _validate_real_scalar(
        timeline.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 timeline 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_phase_dispersion_rad",
        "max_spatial_dispersion_m",
        "max_distance_to_reference_m",
        "phase_tol_rad",
        "spatial_tol_m",
    ):
        _validate_nonnegative_record_scalar(getattr(timeline, field), name=field)
    max_phase_dispersion = _validate_nonnegative_record_scalar(
        timeline.max_phase_dispersion_rad,
        name="max_phase_dispersion_rad",
    )
    max_spatial_dispersion = _validate_nonnegative_record_scalar(
        timeline.max_spatial_dispersion_m,
        name="max_spatial_dispersion_m",
    )
    phase_tol = _validate_nonnegative_record_scalar(
        timeline.phase_tol_rad,
        name="phase_tol_rad",
    )
    spatial_tol = _validate_nonnegative_record_scalar(
        timeline.spatial_tol_m,
        name="spatial_tol_m",
    )
    min_phase_margin = _validate_real_scalar(
        timeline.min_phase_margin_rad,
        name="min_phase_margin_rad",
    )
    min_spatial_margin = _validate_real_scalar(
        timeline.min_spatial_margin_m,
        name="min_spatial_margin_m",
    )
    if (
        abs(min_phase_margin - (phase_tol - max_phase_dispersion))
        > PHA_C_TIMELINE_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_TIMELINE_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "min_spatial_margin_m must equal spatial_tol_m - max_spatial_dispersion_m"
        )
    order_parameter = _validate_nonnegative_record_scalar(
        timeline.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(
        timeline.tolerance_profile_multiplier,
        name="tolerance_profile_multiplier",
    )
    if multiplier <= 0.0:
        raise ValueError("tolerance_profile_multiplier must be positive")
    if (
        not isinstance(timeline.tolerance_profile_name, str)
        or not timeline.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(timeline, field), name=field)

    payload = pha_c_event_timeline_to_dict(timeline)
    replay_payload = dict(payload)
    replay_payload.pop("timeline_sha256")
    if _sha256_json(replay_payload) != timeline_hash:
        raise ValueError(
            "timeline_sha256 does not match the canonical timeline payload",
        )
    return timeline

API documentation¶

pha_c_timeline ¶

Deterministic PHA-C event timelines over moving-frame trajectories.

Single-sample PHA-C handoff records are useful review atoms. This module builds the trajectory-level event timeline that downstream replay, MIF/FRC review, and Studio panels need: first lock, lock loss, reset counts, tolerance profile provenance, and stable hashes over every sample record. The timeline is review-only and never enables actuation.

Classes¶

PHACTimelineRecord `dataclass` ¶

PHACTimelineRecord(
    sample_count: int,
    oscillator_count: int,
    start_time: float,
    end_time: float,
    duration_s: float,
    first_lock_index: int,
    first_lock_time: float,
    first_lock_observed: bool,
    final_lock_achieved: bool,
    lock_sample_count: int,
    phase_lock_sample_count: int,
    spatial_lock_sample_count: int,
    lock_loss_count: int,
    reset_count: int,
    max_consecutive_lock_samples: int,
    max_phase_dispersion_rad: float,
    max_spatial_dispersion_m: 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,
    claim_boundary: str,
    evidence_kind: str,
    execution_disabled: bool,
    actuating: bool,
    time_state_sha256: str,
    sample_records_sha256: str,
    transition_table_sha256: str,
    timeline_sha256: str,
)

Audit-ready PHA-C trajectory event timeline.

The record stores scalar trajectory evidence plus SHA-256 digests of the time vector, per-sample handoff records, transition table, and final timeline. It is designed for review and replay lanes, not for direct control output.

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_timeline.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_event_timeline_to_dict(self)

Functions:¶

build_pha_c_event_timeline ¶

build_pha_c_event_timeline(
    phases_by_step: ArrayLike,
    positions_by_step: ArrayLike,
    *,
    times: ArrayLike | None = None,
    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 = None,
) -> PHACTimelineRecord

Build deterministic PHA-C lock/loss timeline evidence.

Parameters mirror :func:build_pha_c_handoff_record, but consume complete trajectory matrices with shape (time, oscillator). The consecutive-lock counter is threaded through every sample so downstream consumers can review acquisition, loss, and reset events without raw state arrays.

Parameters¶

phases_by_step : ArrayLike Phase history, shape (n_steps, N). positions_by_step : ArrayLike Position history, shape (n_steps, N). times : ArrayLike | None Per-step timestamps in seconds, or None for unit spacing. 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.

Returns¶

PHACTimelineRecord The deterministic PHA-C lock/loss timeline record.

Raises¶

ValueError If any input is invalid.

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

def build_pha_c_event_timeline(
    phases_by_step: ArrayLike,
    positions_by_step: ArrayLike,
    *,
    times: ArrayLike | None = None,
    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 = None,
) -> PHACTimelineRecord:
    """Build deterministic PHA-C lock/loss timeline evidence.

    Parameters mirror :func:`build_pha_c_handoff_record`, but consume complete
    trajectory matrices with shape ``(time, oscillator)``. The consecutive-lock
    counter is threaded through every sample so downstream consumers can review
    acquisition, loss, and reset events without raw state arrays.

    Parameters
    ----------
    phases_by_step : ArrayLike
        Phase history, shape ``(n_steps, N)``.
    positions_by_step : ArrayLike
        Position history, shape ``(n_steps, N)``.
    times : ArrayLike | None
        Per-step timestamps in seconds, or ``None`` for unit spacing.
    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.

    Returns
    -------
    PHACTimelineRecord
        The deterministic PHA-C lock/loss timeline record.

    Raises
    ------
    ValueError
        If any input is invalid.
    """
    phase_matrix = _as_float_matrix(phases_by_step, name="phases_by_step")
    position_matrix = _as_float_matrix(positions_by_step, name="positions_by_step")
    if position_matrix.shape != phase_matrix.shape:
        raise ValueError("positions_by_step must have the same shape as phases_by_step")

    sample_count, oscillator_count = phase_matrix.shape
    time_vector = _as_time_vector(times, sample_count=sample_count)

    records: list[PHACHandoffRecord] = []
    consecutive = 0
    for index, timestamp in enumerate(time_vector):
        record = build_pha_c_handoff_record(
            phase_matrix[index],
            position_matrix[index],
            t=float(timestamp),
            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,
            prior_consecutive_lock_samples=consecutive,
            tolerance_profile=tolerance_profile,
        )
        consecutive = record.consecutive_lock_samples
        records.append(record)

    first = records[0]
    first_lock_index = next(
        (index for index, record in enumerate(records) if record.lock_achieved),
        -1,
    )
    first_lock_observed = first_lock_index >= 0
    first_lock_time = (
        float(time_vector[first_lock_index]) if first_lock_observed else 0.0
    )
    transition_rows = _transition_table(records)
    record_dicts = [record.to_dict() for record in records]

    sample_records_sha256 = _sha256_json(record_dicts)
    time_state_sha256 = _vector_sha256(time_vector)
    transition_table_sha256 = _sha256_json(transition_rows)
    timeline_payload = _timeline_dict_without_hash(
        sample_count=sample_count,
        oscillator_count=oscillator_count,
        start_time=float(time_vector[0]),
        end_time=float(time_vector[-1]),
        duration_s=float(time_vector[-1] - time_vector[0]),
        first_lock_index=first_lock_index,
        first_lock_time=first_lock_time,
        first_lock_observed=first_lock_observed,
        final_lock_achieved=records[-1].lock_achieved,
        lock_sample_count=sum(record.lock_achieved for record in records),
        phase_lock_sample_count=sum(record.phase_locked for record in records),
        spatial_lock_sample_count=sum(record.spatial_locked for record in records),
        lock_loss_count=sum(row["lock_lost"] for row in transition_rows),
        reset_count=sum(row["reset"] for row in transition_rows),
        max_consecutive_lock_samples=max(
            record.consecutive_lock_samples for record in records
        ),
        max_phase_dispersion_rad=max(record.phase_dispersion_rad for record in records),
        max_spatial_dispersion_m=max(record.spatial_dispersion_m for record in records),
        min_phase_margin_rad=min(record.phase_margin_rad for record in records),
        min_spatial_margin_m=min(record.spatial_margin_m for record in records),
        min_phase_order_parameter=min(
            record.phase_order_parameter for record in records
        ),
        max_distance_to_reference_m=max(
            record.distance_to_reference_max_m for record in records
        ),
        reference_phase=first.reference_phase,
        reference_point=first.reference_point,
        phase_tol_rad=first.phase_tol_rad,
        spatial_tol_m=first.spatial_tol_m,
        tolerance_profile_name=first.tolerance_profile_name,
        tolerance_profile_multiplier=first.tolerance_profile_multiplier,
        required_consecutive_samples=first.required_consecutive_samples,
        time_state_sha256=time_state_sha256,
        sample_records_sha256=sample_records_sha256,
        transition_table_sha256=transition_table_sha256,
    )
    return PHACTimelineRecord(
        **cast(Any, timeline_payload),
        timeline_sha256=_sha256_json(timeline_payload),
    )

pha_c_event_timeline_to_dict ¶

pha_c_event_timeline_to_dict(
    timeline: PHACTimelineRecord,
) -> dict[str, float | int | bool | str]

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

Parameters¶

timeline : PHACTimelineRecord The PHA-C event-timeline record to operate on.

Returns¶

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

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

def pha_c_event_timeline_to_dict(
    timeline: PHACTimelineRecord,
) -> dict[str, float | int | bool | str]:
    """Return the canonical JSON-safe PHA-C timeline payload.

    Parameters
    ----------
    timeline : PHACTimelineRecord
        The PHA-C event-timeline record to operate on.

    Returns
    -------
    dict[str, float | int | bool | str]
        The canonical JSON-safe PHA-C timeline payload.
    """
    payload = _timeline_dict_without_hash(
        sample_count=timeline.sample_count,
        oscillator_count=timeline.oscillator_count,
        start_time=timeline.start_time,
        end_time=timeline.end_time,
        duration_s=timeline.duration_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,
        phase_lock_sample_count=timeline.phase_lock_sample_count,
        spatial_lock_sample_count=timeline.spatial_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_phase_dispersion_rad=timeline.max_phase_dispersion_rad,
        max_spatial_dispersion_m=timeline.max_spatial_dispersion_m,
        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,
        time_state_sha256=timeline.time_state_sha256,
        sample_records_sha256=timeline.sample_records_sha256,
        transition_table_sha256=timeline.transition_table_sha256,
    )
    payload["timeline_sha256"] = timeline.timeline_sha256
    return payload

verify_pha_c_event_timeline ¶

verify_pha_c_event_timeline(
    timeline: PHACTimelineRecord,
) -> PHACTimelineRecord

Replay and validate a PHA-C event timeline hash and safety boundary.

Parameters¶

timeline : PHACTimelineRecord The PHA-C event-timeline record to operate on.

Returns¶

PHACTimelineRecord The same timeline after replay and safety-boundary validation.

Raises¶

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

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

def verify_pha_c_event_timeline(
    timeline: PHACTimelineRecord,
) -> PHACTimelineRecord:
    """Replay and validate a PHA-C event timeline hash and safety boundary.

    Parameters
    ----------
    timeline : PHACTimelineRecord
        The PHA-C event-timeline record to operate on.

    Returns
    -------
    PHACTimelineRecord
        The same timeline after replay and safety-boundary validation.

    Raises
    ------
    ValueError
        If the timeline fails replay or safety-boundary validation.
    """
    if not isinstance(timeline, PHACTimelineRecord):
        raise ValueError("timeline must be a PHACTimelineRecord")
    _validate_sha256_hex(timeline.time_state_sha256, name="time_state_sha256")
    _validate_sha256_hex(
        timeline.sample_records_sha256,
        name="sample_records_sha256",
    )
    _validate_sha256_hex(
        timeline.transition_table_sha256,
        name="transition_table_sha256",
    )
    timeline_hash = _validate_sha256_hex(
        timeline.timeline_sha256,
        name="timeline_sha256",
    )
    if timeline.claim_boundary != PHA_C_TIMELINE_CLAIM_BOUNDARY:
        raise ValueError("claim_boundary must be the PHA-C timeline review boundary")
    if timeline.evidence_kind != PHA_C_TIMELINE_EVIDENCE_KIND:
        raise ValueError("evidence_kind must be deterministic timeline evidence")
    if (
        _validate_record_bool(
            timeline.execution_disabled,
            name="execution_disabled",
        )
        is not True
    ):
        raise ValueError("execution_disabled must be true")
    if _validate_record_bool(timeline.actuating, name="actuating") is not False:
        raise ValueError("actuating must be false")

    sample_count = _validate_record_int(
        timeline.sample_count,
        name="sample_count",
        minimum=1,
    )
    _validate_record_int(timeline.oscillator_count, name="oscillator_count", minimum=1)
    required = _validate_record_int(
        timeline.required_consecutive_samples,
        name="required_consecutive_samples",
        minimum=1,
    )
    count_fields = (
        "lock_sample_count",
        "phase_lock_sample_count",
        "spatial_lock_sample_count",
        "lock_loss_count",
        "reset_count",
        "max_consecutive_lock_samples",
    )
    counts = {
        field: _validate_record_int(getattr(timeline, field), name=field, minimum=0)
        for field in count_fields
    }
    for field in (
        "lock_sample_count",
        "phase_lock_sample_count",
        "spatial_lock_sample_count",
        "max_consecutive_lock_samples",
    ):
        if counts[field] > sample_count:
            raise ValueError(f"{field} cannot exceed sample_count")
    for field in ("lock_loss_count", "reset_count"):
        if counts[field] > max(sample_count - 1, 0):
            raise ValueError(f"{field} cannot exceed the transition count")
    final_lock_achieved = _validate_record_bool(
        timeline.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(timeline.start_time, name="start_time")
    end_time = _validate_real_scalar(timeline.end_time, name="end_time")
    duration_s = _validate_nonnegative_record_scalar(
        timeline.duration_s,
        name="duration_s",
    )
    if end_time < start_time:
        raise ValueError("end_time must be greater than or equal to start_time")
    if abs(duration_s - (end_time - start_time)) > 1.0e-12:
        raise ValueError("duration_s must equal end_time - start_time")
    first_lock_index = _validate_record_int(
        timeline.first_lock_index,
        name="first_lock_index",
        minimum=-1,
    )
    first_lock_observed = _validate_record_bool(
        timeline.first_lock_observed,
        name="first_lock_observed",
    )
    first_lock_time = _validate_real_scalar(
        timeline.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 timeline 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_phase_dispersion_rad",
        "max_spatial_dispersion_m",
        "max_distance_to_reference_m",
        "phase_tol_rad",
        "spatial_tol_m",
    ):
        _validate_nonnegative_record_scalar(getattr(timeline, field), name=field)
    max_phase_dispersion = _validate_nonnegative_record_scalar(
        timeline.max_phase_dispersion_rad,
        name="max_phase_dispersion_rad",
    )
    max_spatial_dispersion = _validate_nonnegative_record_scalar(
        timeline.max_spatial_dispersion_m,
        name="max_spatial_dispersion_m",
    )
    phase_tol = _validate_nonnegative_record_scalar(
        timeline.phase_tol_rad,
        name="phase_tol_rad",
    )
    spatial_tol = _validate_nonnegative_record_scalar(
        timeline.spatial_tol_m,
        name="spatial_tol_m",
    )
    min_phase_margin = _validate_real_scalar(
        timeline.min_phase_margin_rad,
        name="min_phase_margin_rad",
    )
    min_spatial_margin = _validate_real_scalar(
        timeline.min_spatial_margin_m,
        name="min_spatial_margin_m",
    )
    if (
        abs(min_phase_margin - (phase_tol - max_phase_dispersion))
        > PHA_C_TIMELINE_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_TIMELINE_MARGIN_REPLAY_TOLERANCE
    ):
        raise ValueError(
            "min_spatial_margin_m must equal spatial_tol_m - max_spatial_dispersion_m"
        )
    order_parameter = _validate_nonnegative_record_scalar(
        timeline.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(
        timeline.tolerance_profile_multiplier,
        name="tolerance_profile_multiplier",
    )
    if multiplier <= 0.0:
        raise ValueError("tolerance_profile_multiplier must be positive")
    if (
        not isinstance(timeline.tolerance_profile_name, str)
        or not timeline.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(timeline, field), name=field)

    payload = pha_c_event_timeline_to_dict(timeline)
    replay_payload = dict(payload)
    replay_payload.pop("timeline_sha256")
    if _sha256_json(replay_payload) != timeline_hash:
        raise ValueError(
            "timeline_sha256 does not match the canonical timeline payload",
        )
    return timeline

UPDE — PHA-C Event Timeline¶

Use cases¶

Contract¶

Minimal example¶

Handoff versus timeline¶

Polyglot parity¶

Failure boundaries¶

PHACTimelineRecord dataclass ¶

Methods:¶

to_dict ¶

Returns¶

build_pha_c_event_timeline ¶

Parameters¶

Returns¶

Raises¶

pha_c_event_timeline_to_dict ¶

Parameters¶

Returns¶

verify_pha_c_event_timeline ¶

Parameters¶

Returns¶

Raises¶

API documentation¶

pha_c_timeline ¶

Classes¶

PHACTimelineRecord dataclass ¶

Methods:¶

to_dict ¶

Returns¶

Functions:¶

build_pha_c_event_timeline ¶

Parameters¶

Returns¶

Raises¶

pha_c_event_timeline_to_dict ¶

Parameters¶

Returns¶

verify_pha_c_event_timeline ¶

Parameters¶

Returns¶

Raises¶

PHACTimelineRecord `dataclass` ¶

PHACTimelineRecord `dataclass` ¶