9b1c94704c
* feat(protocol): add protocol package exports * feat(protocol): add canonical serialization and hashing * feat(protocol): add CTP typed records * feat(protocol): add CTP envelope * feat(protocol): add CTP event constructors * feat(protocol): export CTP constructors * feat(protocol): add JSONL event IO * feat(protocol): add CTP replay verification * feat(protocol): export JSONL and replay helpers * test(protocol): pin CTP canonical replay contracts * docs(protocol): ratify CORE Trace Protocol v0
31 lines
1,012 B
Python
31 lines
1,012 B
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
|
|
from .envelope import CtpEnvelope
|
|
|
|
|
|
class ReplayViolation(ValueError):
|
|
"""Raised when a CTP event or chain cannot be replay-verified."""
|
|
|
|
|
|
def verify_event(event: CtpEnvelope) -> None:
|
|
try:
|
|
event.validate()
|
|
except ValueError as exc:
|
|
raise ReplayViolation(str(exc)) from exc
|
|
|
|
|
|
def verify_chain(events: Sequence[CtpEnvelope]) -> None:
|
|
previous_id = ""
|
|
previous_sequence = -1
|
|
for idx, event in enumerate(events):
|
|
verify_event(event)
|
|
if event.sequence < previous_sequence:
|
|
raise ReplayViolation("CTP event sequence regressed")
|
|
if idx > 0 and event.causation_id and event.causation_id != previous_id:
|
|
raise ReplayViolation(
|
|
f"CTP causation break at index {idx}: expected {previous_id}, got {event.causation_id}"
|
|
)
|
|
previous_id = event.message_id or event.computed_message_id()
|
|
previous_sequence = event.sequence
|