77 lines
2.2 KiB
Python
77 lines
2.2 KiB
Python
"""Capture utilities for reproducible trace envelopes.
|
|
|
|
Provides a small helper to bundle a serialized event-graph snapshot with
|
|
immutable runtime metadata so a trace can be archived and later replayed
|
|
in a self-contained manner.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import platform
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
@dataclass
|
|
class ReproducibleEnvelope:
|
|
"""A self-contained capture bundle for deterministic replays.
|
|
|
|
Fields intentionally minimal: snapshot (event graph), adapters metadata,
|
|
deterministic seeds, runtime snapshot (python/platform), and a CLI/config
|
|
snapshot string.
|
|
"""
|
|
|
|
snapshot: Dict[str, Any]
|
|
adapters: Dict[str, str]
|
|
rng_seed: Optional[int]
|
|
container_image: Optional[str]
|
|
cmdline: Optional[str]
|
|
timestamp_ns: int
|
|
runtime: Dict[str, str]
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"snapshot": self.snapshot,
|
|
"adapters": self.adapters,
|
|
"rng_seed": self.rng_seed,
|
|
"container_image": self.container_image,
|
|
"cmdline": self.cmdline,
|
|
"timestamp_ns": self.timestamp_ns,
|
|
"runtime": self.runtime,
|
|
}
|
|
|
|
|
|
def create_reproducible_envelope(
|
|
snapshot: Dict[str, Any],
|
|
adapters: Optional[Dict[str, str]] = None,
|
|
rng_seed: Optional[int] = None,
|
|
container_image: Optional[str] = None,
|
|
cmdline: Optional[str] = None,
|
|
) -> ReproducibleEnvelope:
|
|
"""Create a reproducible envelope bundling snapshot + metadata.
|
|
|
|
This is deliberately lightweight: callers should pass adapter version
|
|
strings (e.g. package@version or binary digest) and any available
|
|
runtime metadata. The envelope is JSON-serializable via `to_dict()`.
|
|
"""
|
|
|
|
adapters = adapters or {}
|
|
return ReproducibleEnvelope(
|
|
snapshot=snapshot,
|
|
adapters=adapters,
|
|
rng_seed=rng_seed,
|
|
container_image=container_image,
|
|
cmdline=cmdline,
|
|
timestamp_ns=time.time_ns(),
|
|
runtime={
|
|
"python": platform.python_version(),
|
|
"platform": platform.platform(),
|
|
},
|
|
)
|
|
|
|
|
|
def envelope_to_json(env: ReproducibleEnvelope) -> str:
|
|
return json.dumps(env.to_dict(), sort_keys=True)
|