stormlog.infer.correlation_capture

Optional server evidence capture using existing inference and run artifacts.

Functions

append_inference_capture(artifact_path, *, ...)

Append optional evidence and index its raw traces in a run envelope.

link_trace_activities(engine_events, ...)

Join only unambiguous scoped runtime/CUDA IDs, never timestamps alone.

Classes

CaptureCapabilities([supported, enabled, ...])

Features a component supports, enabled, and actually collected.

EngineAdapter(*args, **kwargs)

Optional source of server request, iteration, stage, and membership data.

EngineCapture(capabilities[, events, summary])

Engine evidence; summary is recorded on the adapter's capability event.

TraceAttachment(attachment_id, title[, ...])

A raw trace to register through the existing run envelope catalog.

TraceCapture(capabilities[, events, ...])

Trace evidence; summary is recorded on the collector's capability event.

TraceCollector(*args, **kwargs)

Optional source of GPU activities and raw trace attachments.
class stormlog.infer.correlation_capture.CaptureCapabilities(supported=(), enabled=(), collected=())[source]

Bases: object

Features a component supports, enabled, and actually collected.

Parameters:

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

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

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

supported: tuple[str, ...] = ()
enabled: tuple[str, ...] = ()
collected: tuple[str, ...] = ()
class stormlog.infer.correlation_capture.EngineAdapter(*args, **kwargs)[source]

Bases: Protocol

Optional source of server request, iteration, stage, and membership data.

collect(*, run_id, session_id)[source]
Parameters:

  • run_id (str)

  • session_id (str)

Return type:

EngineCapture

class stormlog.infer.correlation_capture.EngineCapture(capabilities, events=(), summary=None)[source]

Bases: object

Engine evidence; summary is recorded on the adapter’s capability event.

Parameters:
capabilities: CaptureCapabilities
events: tuple[CorrelationEvent, ...] = ()
summary: dict[str, Any] | None = None
class stormlog.infer.correlation_capture.TraceAttachment(attachment_id, title, path=None, url=None, storage='reference', metadata=None)[source]

Bases: object

A raw trace to register through the existing run envelope catalog.

Parameters:

  • attachment_id (str)

  • title (str)

  • path (Path | None)

  • url (str | None)

  • storage (Literal['reference', 'copy'])

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

attachment_id: str
title: str
path: Path | None = None
url: str | None = None
storage: Literal['reference', 'copy'] = 'reference'
metadata: dict[str, Any] | None = None
class stormlog.infer.correlation_capture.TraceCapture(capabilities, events=(), attachments=(), summary=None)[source]

Bases: object

Trace evidence; summary is recorded on the collector’s capability event.

Parameters:
capabilities: CaptureCapabilities
events: tuple[CorrelationEvent, ...] = ()
attachments: tuple[TraceAttachment, ...] = ()
summary: dict[str, Any] | None = None
class stormlog.infer.correlation_capture.TraceCollector(*args, **kwargs)[source]

Bases: Protocol

Optional source of GPU activities and raw trace attachments.

collect(*, run_id, session_id)[source]
Parameters:

  • run_id (str)

  • session_id (str)

Return type:

TraceCapture

stormlog.infer.correlation_capture.append_inference_capture(artifact_path, *, run_id, session, engine_adapter=None, trace_collector=None, envelope_path=None)[source]

Append optional evidence and index its raw traces in a run envelope.

Collection and validation finish before either artifact is changed. An absent component contributes an explicit unavailable capability record.

Parameters:
Return type:

None

Join only unambiguous scoped runtime/CUDA IDs, never timestamps alone.

Parameters:
Return type:

tuple[CorrelationEvent, …]