stormlog.session

Shared session identity and lifecycle helpers.

Functions

create_session_summary(*, source[, status, ...])

Create a normalized session summary.

finalize_session_summary(summary, *[, ...])

Finalize a session without overwriting degraded terminal states.

infer_session_summary_from_events(*, ...[, ...])

Infer a session summary from event timestamps and lifecycle markers.

new_session_id()

Create a new opaque session identifier.

normalize_session_status(value)

Validate and normalize a session status string.

now_ns()

Return the current wall-clock timestamp in nanoseconds.

select_default_loaded_session(sessions)

Pick the default loaded session, preferring non-empty event groups.

select_default_session(sessions)

Pick the default session using lifecycle priority and recency.

session_summary_from_dict(payload)

Deserialize a session summary mapping.

session_summary_to_dict(summary)

Serialize a session summary into a plain JSON-safe mapping.

sort_session_summaries(summaries)

Sort sessions by selection priority and recency.

stable_legacy_session_id(*parts)

Build a deterministic synthetic session id for legacy artifacts.

update_session_summary(summary, *[, status, ...])

Return a copy of a session summary with lifecycle updates applied.

Classes

SessionSummary(session_id, status, ...[, ...])

Identity and lifecycle metadata for one capture session.
class stormlog.session.SessionSummary(session_id, status, started_at_ns, ended_at_ns, host, pid, job_id=None, rank=0, local_rank=0, world_size=1, source='stormlog.session')[source]

Bases: object

Identity and lifecycle metadata for one capture session.

Parameters:

  • session_id (str)

  • status (str)

  • started_at_ns (int)

  • ended_at_ns (int | None)

  • host (str)

  • pid (int)

  • job_id (str | None)

  • rank (int)

  • local_rank (int)

  • world_size (int)

  • source (str)

session_id: str
status: str
started_at_ns: int
ended_at_ns: int | None
host: str
pid: int
job_id: str | None = None
rank: int = 0
local_rank: int = 0
world_size: int = 1
source: str = 'stormlog.session'
stormlog.session.create_session_summary(*, source, status='running', session_id=None, started_at_ns=None, ended_at_ns=None, host=None, pid=None, job_id=None, rank=0, local_rank=0, world_size=1)[source]

Create a normalized session summary.

Parameters:

  • source (str)

  • status (str)

  • session_id (str | None)

  • started_at_ns (int | None)

  • ended_at_ns (int | None)

  • host (str | None)

  • pid (int | None)

  • job_id (str | None)

  • rank (int)

  • local_rank (int)

  • world_size (int)

Return type:

SessionSummary

stormlog.session.finalize_session_summary(summary, *, ended_at_ns=None)[source]

Finalize a session without overwriting degraded terminal states.

Parameters:
Return type:

SessionSummary

stormlog.session.infer_session_summary_from_events(*, session_id, events, source, fallback_status='incomplete')[source]

Infer a session summary from event timestamps and lifecycle markers.

Parameters:

  • session_id (str)

  • events (Iterable[Any])

  • source (str)

  • fallback_status (str)

Return type:

SessionSummary

stormlog.session.new_session_id()[source]

Create a new opaque session identifier.

Return type:

str

stormlog.session.normalize_session_status(value)[source]

Validate and normalize a session status string.

Parameters:

value (object)

Return type:

str

stormlog.session.now_ns()[source]

Return the current wall-clock timestamp in nanoseconds.

Return type:

int

stormlog.session.select_default_loaded_session(sessions)[source]

Pick the default loaded session, preferring non-empty event groups.

Parameters:

sessions (Iterable[Any])

Return type:

Any | None

stormlog.session.select_default_session(sessions)[source]

Pick the default session using lifecycle priority and recency.

Parameters:

sessions (Iterable[_SessionLikeT])

Return type:

_SessionLikeT | None

stormlog.session.session_summary_from_dict(payload)[source]

Deserialize a session summary mapping.

Parameters:

payload (Mapping[str, Any])

Return type:

SessionSummary

stormlog.session.session_summary_to_dict(summary)[source]

Serialize a session summary into a plain JSON-safe mapping.

Parameters:

summary (SessionSummary)

Return type:

dict[str, Any]

stormlog.session.sort_session_summaries(summaries)[source]

Sort sessions by selection priority and recency.

Parameters:

summaries (Iterable[SessionSummary])

Return type:

list[SessionSummary]

stormlog.session.stable_legacy_session_id(*parts)[source]

Build a deterministic synthetic session id for legacy artifacts.

Parameters:

parts (object)

Return type:

str

stormlog.session.update_session_summary(summary, *, status=None, ended_at_ns=<object object>, source=None)[source]

Return a copy of a session summary with lifecycle updates applied.

Parameters:

  • summary (SessionSummary)

  • status (str | None)

  • ended_at_ns (int | None | object)

  • source (str | None)

Return type:

SessionSummary