stormlog.phases

Structured phase telemetry helpers for trackers and analysis.

class stormlog.phases.PhaseAttribution(phase_resolution, phase_source=None, phase_path=None, phase_paths=<factory>, scope_id=None, thread_id=None, thread_name=None, phase_summary=None)[source]

Bases: object

Resolved workload phase attribution for an anomaly or report item.

Parameters:

  • phase_resolution (str)

  • phase_source (str | None)

  • phase_path (str | None)

  • phase_paths (list[str])

  • scope_id (str | None)

  • thread_id (int | None)

  • thread_name (str | None)

  • phase_summary (PhaseSummary | None)

phase_resolution: str
phase_source: str | None = None
phase_path: str | None = None
phase_paths: list[str]
scope_id: str | None = None
thread_id: int | None = None
thread_name: str | None = None
phase_summary: PhaseSummary | None = None
class stormlog.phases.PhaseSummary(phase_path, source)[source]

Bases: object

Optional presentation-oriented phase winner when canonical attribution stays ambiguous.

Parameters:

  • phase_path (str)

  • source (str)

phase_path: str
source: str
stormlog.phases.attribute_active_spans(spans, *, strategy='deepest_per_thread', origin_thread_id=None, origin_phase_scope_id=None)[source]

Collapse active spans into a unique or ambiguity-preserving attribution.

Parameters:

  • spans (Sequence['PhaseSpan'])

  • strategy (str)

  • origin_thread_id (int | None)

  • origin_phase_scope_id (str | None)

Return type:

PhaseAttribution | None

class stormlog.phases.PhaseBoundary(event_type, context, metadata, scope_id, path)[source]

Bases: object

Structured boundary payload emitted into tracker telemetry events.

Parameters:

  • event_type (str)

  • context (str)

  • metadata (dict[str, Any])

  • scope_id (str)

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

event_type: str
context: str
metadata: dict[str, Any]
scope_id: str
path: tuple[str, ...]
class stormlog.phases.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]')[source]

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.PhaseHandle(*, scope_id, name, path, close_callback)[source]

Bases: object

A closeable tracker phase handle returned by enter_phase().

Parameters:

  • scope_id (str)

  • name (str)

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

  • close_callback (Callable[[], Any])

property phase_path: str

Return the formatted phase path.

property closed: bool

Return True once the handle has been closed.

close()[source]

Close the phase handle once.

Return type:

Any

exception stormlog.phases.PhaseProtocolError[source]

Bases: RuntimeError

Raised when phase handles are closed incorrectly.

class stormlog.phases.PhaseRecorder[source]

Bases: object

Per-tracker phase nesting state and boundary payload generation.

reset()[source]

Drop all active phase scopes for a new tracker session.

Return type:

None

enter(*, session_id, rank, name, attrs=None)[source]

Primary runtime API returning both a token and the emitted boundary.

Parameters:

  • session_id (str)

  • rank (int)

  • name (str)

  • attrs (Mapping[str, Any] | None)

Return type:

tuple[PhaseToken, PhaseBoundary]

enter_phase(*, session_id, rank, name, metadata=None)[source]

Register one nested phase enter transition.

Parameters:

  • session_id (str)

  • rank (int)

  • name (str)

  • metadata (Mapping[str, Any] | None)

Return type:

PhaseBoundary

exit(token)[source]

Primary runtime API for strict token-based exit.

Parameters:

token (PhaseToken)

Return type:

PhaseBoundary

exit_phase(*, session_id, rank, scope_id, thread_id)[source]

Register one nested phase exit transition.

Parameters:

  • session_id (str)

  • rank (int)

  • scope_id (str)

  • thread_id (int)

Return type:

PhaseBoundary

class stormlog.phases.PhaseReplayIndex(intervals_by_group)[source]

Bases: object

Replay phase boundaries into queryable active spans.

Parameters:

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

classmethod from_events(events)[source]

Build a replay index from telemetry events.

Parameters:

events (Sequence[Any])

Return type:

PhaseReplayIndex

spans_for(*, session_id, rank=None)[source]

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)[source]

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)[source]

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)[source]

Resolve one event-like object against the replay index.

Parameters:

event (Any)

Return type:

Any

class stormlog.phases.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)[source]

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
class stormlog.phases.PhaseToken(scope_id, session_id, rank, thread_id, name)[source]

Bases: object

Opaque runtime token used for strict phase exit semantics.

Parameters:

  • scope_id (str)

  • session_id (str)

  • rank (int)

  • thread_id (int)

  • name (str)

scope_id: str
session_id: str
rank: int
thread_id: int
name: str
stormlog.phases.phase_attribution_to_payload(attribution)[source]

Serialize a phase attribution without emitting redundant summary fields.

Parameters:

attribution (PhaseAttribution | None)

Return type:

dict[str, Any] | None

stormlog.phases.resolve_phase_for_event(index, event)[source]

Resolve a phase attribution for one event-like object using replay data.

Parameters:

  • index (ReplayPhaseReplayIndex)

  • event (Any)

Return type:

PhaseAttribution | None

stormlog.phases.format_phase_path(path)[source]

Return a human-readable phase path label.

Parameters:

path (Sequence[str])

Return type:

str

stormlog.phases.is_phase_boundary_event(event)[source]

Return True when the event is a structured phase boundary.

Parameters:

event (Any)

Return type:

bool

stormlog.phases.merge_phase_attributions(first, second)[source]

Merge two attribution candidates without inventing a false unique path.

Parameters:
Return type:

PhaseAttribution | None

stormlog.phases.parse_phase_boundary(event)[source]

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

Parameters:

event (Any)

Return type:

PhaseBoundaryRecord | None

stormlog.phases.summarize_phase_attribution(attribution)[source]

Return a user-facing summary string for one phase attribution.

Parameters:

attribution (PhaseAttribution | None)

Return type:

str | None

stormlog.phases.summarize_phase_resolution(*, phase_resolution, phase_path=None, phase_paths=None)[source]

Render one phase resolution without hiding ambiguity semantics.

Parameters:

  • phase_resolution (str | None)

  • phase_path (str | None)

  • phase_paths (Sequence[str] | None)

Return type:

str | None

Modules

policy

Phase attribution policy and formatting helpers.

replay

Phase boundary parsing and replay index helpers.

runtime

Runtime phase state tracking and boundary payload emission.