build(agent): r2d2#deee02 iteration

deltatrace/replay_engine.py

@@ -34,7 +34,13 @@ class ReplayEngine:
         self._handlers.setdefault(event_type, []).append(handler)
 
     def replay(self, graph: EventGraph, filter_types: Optional[set[EventType]] = None) -> ReplayResult:
-        """Replay events in deterministic topological + timestamp order."""
+        """Replay events in deterministic topological + timestamp order.
+
+        Optional behavior:
+        - If `filter_types` is provided, only events of those types are replayed.
+        - If `reference_event_log` is provided (list of event dicts), compute a
+          fidelity score in [0.0, 1.0] comparing this run to the reference.
+        """
         ordered = graph.topological_order()
 
         if filter_types:
@@ -68,7 +74,93 @@ class ReplayEngine:
             event_log=event_log,
         )
 
-    def replay_from_snapshot(self, snapshot: dict) -> ReplayResult:
+    def replay(self, graph: EventGraph, filter_types: Optional[set[EventType]] = None,
+               reference_event_log: Optional[list[dict]] = None) -> ReplayResult:
+        """Replay events and optionally score fidelity against a reference log.
+
+        Backwards-compatible: if called without `reference_event_log`, behavior
+        is unchanged and fidelity_score is 1.0.
+        """
+        ordered = graph.topological_order()
+
+        if filter_types:
+            ordered = [e for e in ordered if e.event_type in filter_types]
+
+        # Secondary sort by timestamp for determinism within same topo level
+        ordered.sort(key=lambda e: e.timestamp_ns)
+
+        event_log = []
+        replayed = 0
+        min_ts = ordered[0].timestamp_ns if ordered else 0
+        max_ts = ordered[-1].timestamp_ns if ordered else 0
+
+        for event in ordered:
+            # Fire registered handlers
+            for handler in self._handlers.get(event.event_type, []):
+                handler(event)
+
+            event_log.append({
+                "id": event.id,
+                "type": event.event_type.value,
+                "timestamp_ns": event.timestamp_ns,
+                "payload": event.payload,
+            })
+            replayed += 1
+
+        # Compute a simple fidelity score vs a provided reference log: fraction of
+        # matching ids at the same indices (normalized by the longer of the two).
+        fidelity = 1.0
+        if reference_event_log is not None:
+            ref_ids = [e.get("id") for e in reference_event_log]
+            cur_ids = [e.get("id") for e in event_log]
+            matches = sum(1 for i in range(min(len(ref_ids), len(cur_ids))) if ref_ids[i] == cur_ids[i])
+            denom = max(len(ref_ids), len(cur_ids)) if max(len(ref_ids), len(cur_ids)) > 0 else 1
+            fidelity = matches / denom
+
+        return ReplayResult(
+            events_replayed=replayed,
+            total_latency_ns=max_ts - min_ts,
+            fidelity_score=fidelity,
+            event_log=event_log,
+        )
+
+    def replay_from_snapshot(self, snapshot: dict, reference_event_log: Optional[list[dict]] = None) -> ReplayResult:
         """Replay from a serialized event graph snapshot."""
         graph = EventGraph.from_dict(snapshot)
-        return self.replay(graph)
+        return self.replay(graph, reference_event_log=reference_event_log)
+
+    def create_checkpoint(self, graph: EventGraph, upto_event_id: Optional[str] = None) -> dict:
+        """Create a serialized checkpoint of the event graph.
+
+        If `upto_event_id` is provided, include events in topological order up to
+        and including that event. Otherwise returns full snapshot (graph.to_dict()).
+        """
+        if upto_event_id is None:
+            return graph.to_dict()
+
+        ordered = graph.topological_order()
+        included_ids = []
+        for e in ordered:
+            included_ids.append(e.id)
+            if e.id == upto_event_id:
+                break
+
+        # Build snapshot consisting of included events and edges between them
+        events = [
+            {
+                "id": e.id,
+                "type": e.event_type.value,
+                "timestamp_ns": e.timestamp_ns,
+                "payload": e.payload,
+                "source_adapter": e.source_adapter,
+            }
+            for e in ordered if e.id in included_ids
+        ]
+
+        edges = [
+            {"from": edge.from_id, "to": edge.to_id, "latency_ns": edge.latency_ns, "label": edge.label}
+            for edge in graph.edges
+            if edge.from_id in included_ids and edge.to_id in included_ids
+        ]
+
+        return {"events": events, "edges": edges}

tests/test_replay_engine.py

@@ -61,6 +61,38 @@ class TestReplayEngine(unittest.TestCase):
         result = engine.replay_from_snapshot(snapshot)
         self.assertEqual(result.events_replayed, 4)
 
+    def test_replay_checkpoint_resume(self):
+        graph = self._build_graph()
+        engine = ReplayEngine()
+
+        # Create a checkpoint up to the SIGNAL event and replay just that prefix
+        signal = next(e for e in graph.topological_order() if e.event_type == EventType.SIGNAL)
+        checkpoint = engine.create_checkpoint(graph, upto_event_id=signal.id)
+        res = engine.replay_from_snapshot(checkpoint)
+        # tick + signal expected
+        self.assertEqual(res.events_replayed, 2)
+
+    def test_determinism_score(self):
+        graph = self._build_graph()
+        engine = ReplayEngine()
+
+        # Reference run
+        r1 = engine.replay(graph)
+        ref = r1.event_log
+
+        # Mutate timestamps in a snapshot to force a different replay ordering
+        snapshot = graph.to_dict()
+        # Find the signal event and make its timestamp far later
+        for e in snapshot["events"]:
+            if e["type"] == "signal":
+                e["timestamp_ns"] = e["timestamp_ns"] + 10_000_000_000
+                break
+
+        mutated = EventGraph.from_dict(snapshot)
+        r2 = engine.replay(mutated, reference_event_log=ref)
+        # Expect fidelity to be lower than perfect (not equal to 1.0)
+        self.assertLess(r2.fidelity_score, 1.0)
+
 
 if __name__ == "__main__":
     unittest.main()