diff --git a/deltatrace/capture.py b/deltatrace/capture.py new file mode 100644 index 0000000..5f6e1fd --- /dev/null +++ b/deltatrace/capture.py @@ -0,0 +1,76 @@ +"""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) diff --git a/tests/test_capture.py b/tests/test_capture.py new file mode 100644 index 0000000..427887c --- /dev/null +++ b/tests/test_capture.py @@ -0,0 +1,25 @@ +from deltatrace.capture import create_reproducible_envelope, envelope_to_json +from deltatrace.event_graph import Event, EventType, EventGraph +from deltatrace.replay_engine import ReplayEngine + + +def test_reproducible_envelope_contains_snapshot_and_replays(): + g = EventGraph() + t1 = Event.create(EventType.MD_TICK, {"v": 1}) + s1 = Event.create(EventType.SIGNAL, {"v": 2}) + g.add_event(t1) + g.add_event(s1) + g.add_edge(t1.id, s1.id) + + # create checkpoint/snapshot + snapshot = g.to_dict() + + env = create_reproducible_envelope(snapshot, adapters={"fix": "fix-adapter@0.1"}, rng_seed=42, container_image="sha256:deadbeef", cmdline="demo --run") + + j = envelope_to_json(env) + assert isinstance(j, str) and 'snapshot' in j + + # replay from snapshot inside envelope + engine = ReplayEngine() + res = engine.replay_from_snapshot(env.snapshot) + assert res.events_replayed == len(g.events)