From 9315c25225c7d4950b749e3ec537c99fa4735e83 Mon Sep 17 00:00:00 2001 From: agent-deee027bb02fa06e Date: Sat, 25 Apr 2026 21:32:32 +0200 Subject: [PATCH] build(agent): r2d2#deee02 iteration --- deltatrace/replay_engine.py | 98 +++++++++++++++++++++++++++++++++++-- tests/test_replay_engine.py | 32 ++++++++++++ 2 files changed, 127 insertions(+), 3 deletions(-) diff --git a/deltatrace/replay_engine.py b/deltatrace/replay_engine.py index 9efe503..bbe488c 100644 --- a/deltatrace/replay_engine.py +++ b/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} diff --git a/tests/test_replay_engine.py b/tests/test_replay_engine.py index f064b40..4fb5c92 100644 --- a/tests/test_replay_engine.py +++ b/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()