mixle.experimental.typed_runtime.replay module

Deterministic and tolerance replay for proposal/commit logs.

class ReplayEntry(batch, expected_receipt, expected_state=None)[source]

Bases: object

One immutable batch, expected receipt, and optional resulting state.

Parameters:

  • batch (ProposalBatch)

  • expected_receipt (CommitReceipt)

  • expected_state (Any)

as_dict()[source]

Return metadata without serializing arbitrary expected state.

Return type:

dict[str, Any]

class ReplayLog(entries=())[source]

Bases: object

Append-only sequence of proposal batches and terminal commit receipts.

record() deep-copies its inputs so a later caller mutation cannot rewrite history – but the list holding them was public, so log.entries[0] = other and log.entries.clear() rewrote exactly the history the copying protects, and replay_log would then replay a sequence that no coordinator ever committed (MXR-080-1905). entries is a detached tuple now; indexing into it (log.entries[0].batch) is unchanged.

Parameters:

entries (Sequence[ReplayEntry])

property entries: tuple[ReplayEntry, ...]

The recorded replay entries, oldest first, as a detached tuple.

record(batch, receipt, *, expected_state=None)[source]

Record detached replay inputs so later caller mutation cannot rewrite history.

Parameters:

  • batch (ProposalBatch)

  • receipt (CommitReceipt)

  • expected_state (Any)

Return type:

None

as_dict()[source]

Return a JSON-compatible replay manifest.

Return type:

dict[str, Any]

class ReplayMode(*values)[source]

Bases: StrEnum

State comparison guarantee expected from a backend.

class ReplayReport(mode, absolute_tolerance, relative_tolerance, steps)[source]

Bases: object

Whole-log replay verdict with explicit numeric tolerance.

Parameters:

  • mode (ReplayMode)

  • absolute_tolerance (float)

  • relative_tolerance (float)

  • steps (tuple[ReplayStepReceipt, ...])

property status: ReplayStatus

Distinguish successful evidence from mismatch and absent evidence.

property ran: bool

Whether at least one replay step was attempted.

property matched: bool

Whether every recorded transaction replayed under the selected guarantee.

as_dict()[source]

Return a JSON-compatible replay report.

Return type:

dict[str, Any]

class ReplayStatus(*values)[source]

Bases: StrEnum

Whole-log replay outcome; an empty log is explicitly not run.

class ReplayStepReceipt(index, commit_id, matched, mismatches, actual_receipt=None)[source]

Bases: object

Comparison result for one replayed transaction.

Parameters:

  • index (int)

  • commit_id (str)

  • matched (bool)

  • mismatches (tuple[str, ...])

  • actual_receipt (CommitReceipt | None)

as_dict()[source]

Return a JSON-compatible replay comparison.

Return type:

dict[str, Any]

replay_log(log, coordinator, *, mode=ReplayMode.BITWISE, state_probe=None, absolute_tolerance=0.0, relative_tolerance=0.0)[source]

Replay every batch and compare semantic receipts plus resulting state.

Parameters:

  • log (ReplayLog)

  • coordinator (TransactionalCoordinator)

  • mode (ReplayMode)

  • state_probe (Callable[[], Any] | None)

  • absolute_tolerance (float)

  • relative_tolerance (float)

Return type:

ReplayReport