Skip to content

Trace contract

The models in risansym.schemas are the advanced API for consumers that read or validate trace files. They are deliberately not re-exported from the package root.

TraceOutput is the authoritative Pydantic model for schema version 1.0. The generated JSON Schema lives at shared/schema/trace.schema.json. Compatible additions require optional fields. Any change that alters required fields or their meaning must introduce a new schema_version and update both validators and shared fixtures in the same delivery.

risansym.schemas

AppLogEvent

Bases: BaseModel

Recorded when a node emits an application-level log message.

Source code in core/src/risansym/schemas.py 
49
50
51
52
53
54
55
56
57
class AppLogEvent(BaseModel):
    """Recorded when a node emits an application-level log message."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    action: Literal["APP_LOG"] = "APP_LOG"
    clock: NonNegativeTime
    source: NodeId
    message: NonEmptyString

ReceiveEvent

Bases: BaseModel

Recorded when a node processes an incoming message.

Source code in core/src/risansym/schemas.py 
35
36
37
38
39
40
41
42
43
44
45
46
class ReceiveEvent(BaseModel):
    """Recorded when a node processes an incoming message."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    action: Literal["RECEIVE"] = "RECEIVE"
    clock: NonNegativeTime = Field(description="Time at which the node processes the event")
    source: NodeId
    target: NodeId
    name: NonEmptyString
    payload: JsonPayload
    node_state: JsonPayload | None = None

TraceCapture

Bases: BaseModel

Describe trace retention and make truncation machine-readable.

Source code in core/src/risansym/schemas.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class TraceCapture(BaseModel):
    """Describe trace retention and make truncation machine-readable."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    max_events: int = Field(gt=0, le=1_000_000)
    recorded_events: int = Field(ge=0)
    dropped_events: int = Field(ge=0)
    truncated: bool

    @model_validator(mode="after")
    def validate_counts(self) -> "TraceCapture":
        """Keep the truncation flag and counters internally consistent."""
        if self.truncated != (self.dropped_events > 0):
            raise ValueError("truncated must be true exactly when dropped_events is positive")
        return self

validate_counts()

Keep the truncation flag and counters internally consistent.

Source code in core/src/risansym/schemas.py 
74
75
76
77
78
79
@model_validator(mode="after")
def validate_counts(self) -> "TraceCapture":
    """Keep the truncation flag and counters internally consistent."""
    if self.truncated != (self.dropped_events > 0):
        raise ValueError("truncated must be true exactly when dropped_events is positive")
    return self

TraceMetadata

Bases: BaseModel

Metadata attached to a complete simulation trace.

Source code in core/src/risansym/schemas.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
class TraceMetadata(BaseModel):
    """Metadata attached to a complete simulation trace."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    schema_version: Literal["1.0"] = "1.0"
    algorithm: NonEmptyString
    topology: NonEmptyString
    tag: str | None = None
    execution_date: datetime.datetime
    parameters: JsonPayload
    metrics: JsonPayload
    capture: TraceCapture

TraceOutput

Bases: BaseModel

Top-level container for a simulation trace file.

Source code in core/src/risansym/schemas.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
class TraceOutput(BaseModel):
    """Top-level container for a simulation trace file."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    metadata: TraceMetadata
    trace: Annotated[list[TraceEvent], Field(max_length=1_000_000)]

    @model_validator(mode="after")
    def validate_recorded_count(self) -> "TraceOutput":
        """Ensure capture metadata describes the serialized trace."""
        if self.metadata.capture.recorded_events != len(self.trace):
            raise ValueError("capture.recorded_events must equal trace length")
        return self

validate_recorded_count()

Ensure capture metadata describes the serialized trace.

Source code in core/src/risansym/schemas.py 
105
106
107
108
109
110
@model_validator(mode="after")
def validate_recorded_count(self) -> "TraceOutput":
    """Ensure capture metadata describes the serialized trace."""
    if self.metadata.capture.recorded_events != len(self.trace):
        raise ValueError("capture.recorded_events must equal trace length")
    return self

TransmitEvent

Bases: BaseModel

Recorded when a node schedules a message for transmission.

Source code in core/src/risansym/schemas.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class TransmitEvent(BaseModel):
    """Recorded when a node schedules a message for transmission."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    action: Literal["TRANSMIT"] = "TRANSMIT"
    clock: NonNegativeTime = Field(description="Time at which the sender dispatched the event")
    event_time: NonNegativeTime = Field(description="Computed arrival time at the target node")
    source: NodeId
    target: NodeId
    name: NonEmptyString
    payload: JsonPayload
    node_state: JsonPayload | None = None

    @model_validator(mode="after")
    def validate_causality(self) -> "TransmitEvent":
        """Ensure the recorded arrival cannot precede transmission."""
        if self.event_time < self.clock:
            raise ValueError("event_time cannot be earlier than clock")
        return self

validate_causality()

Ensure the recorded arrival cannot precede transmission.

Source code in core/src/risansym/schemas.py 
27
28
29
30
31
32
@model_validator(mode="after")
def validate_causality(self) -> "TransmitEvent":
    """Ensure the recorded arrival cannot precede transmission."""
    if self.event_time < self.clock:
        raise ValueError("event_time cannot be earlier than clock")
    return self