stormlog.phases.replay

Phase boundary parsing and replay index helpers.

Functions

is_phase_boundary_event(event)	Return True when the event is a structured phase boundary.
parse_phase_boundary(event)	Extract one normalized phase payload from an event-like object.

Classes

PhaseBoundaryRecord(action, name, path, ...)
PhaseReplayIndex(intervals_by_group)	Replay phase boundaries into queryable active spans.
PhaseSpan(session_id, rank, thread_id, ...)

class stormlog.phases.replay.PhaseBoundaryRecord(action: 'str', name: 'str', path: 'tuple[str, ...]', depth: 'int', scope_id: 'str', parent_scope_id: 'str | None', thread_id: 'int', thread_name: 'str', sequence: 'int', session_id: 'str', timestamp_ns: 'int', attributes: 'dict[str, Any]')

Bases: object

Parameters:

• action (str)

• name (str)

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

• depth (int)

• scope_id (str)

• parent_scope_id (str | None)

• thread_id (int)

• thread_name (str)

• sequence (int)

• session_id (str)

• timestamp_ns (int)

• attributes (dict[str, Any])

action: str

name: str

path: tuple[str, ...]

depth: int

scope_id: str

parent_scope_id: str | None

thread_id: int

thread_name: str

sequence: int

session_id: str

timestamp_ns: int

attributes: dict[str, Any]

class stormlog.phases.replay.PhaseReplayIndex(intervals_by_group)

Bases: object

Replay phase boundaries into queryable active spans.

Parameters:

intervals_by_group (Mapping[tuple[str, int], Sequence[PhaseSpan]])

classmethod from_events(events)

Build a replay index from telemetry events.

Parameters:

events (Sequence[Any])

Return type:

PhaseReplayIndex

spans_for(*, session_id, rank=None)

Return all reconstructed spans for one session/rank selection.

Parameters:

• session_id (str)

• rank (int | None)

Return type:

list[PhaseSpan]

active_spans(*, timestamp_ns, session_id, rank=None)

Return active spans at a timestamp before attribution policy is applied.

Parameters:

• timestamp_ns (int)

• session_id (str)

• rank (int | None)

Return type:

list[PhaseSpan]

resolve(*, timestamp_ns, session_id, rank=None, origin_thread_id=None, origin_phase_scope_id=None)

Resolve one timestamp against the replay index.

Parameters:

• timestamp_ns (int)

• session_id (str | None)

• rank (int | None)

• origin_thread_id (int | None)

• origin_phase_scope_id (str | None)

Return type:

Any

resolve_for_event(event)

Resolve one event-like object against the replay index.

Parameters:

event (Any)

Return type:

Any

class stormlog.phases.replay.PhaseSpan(session_id: 'str', rank: 'int', thread_id: 'int', thread_name: 'str', scope_id: 'str', path: 'tuple[str, ...]', start_ns: 'int', end_ns: 'int', sequence: 'int', synthetic_end: 'bool' = False)

Bases: object

Parameters:

• session_id (str)

• rank (int)

• thread_id (int)

• thread_name (str)

• scope_id (str)

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

• start_ns (int)

• end_ns (int)

• sequence (int)

• synthetic_end (bool)

session_id: str

rank: int

thread_id: int

thread_name: str

scope_id: str

path: tuple[str, ...]

start_ns: int

end_ns: int

sequence: int

synthetic_end: bool = False

property depth: int

stormlog.phases.replay.is_phase_boundary_event(event)

Return True when the event is a structured phase boundary.

Parameters:

event (Any)

Return type:

bool

stormlog.phases.replay.parse_phase_boundary(event)

Extract one normalized phase payload from an event-like object.

Parameters:

event (Any)

Return type:

PhaseBoundaryRecord | None