File size: 1,527 Bytes
42e32ed a196e34 42e32ed a196e34 42e32ed a196e34 42e32ed a2ca0e0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | from __future__ import annotations
from collections.abc import Iterable
from src import observability as obs
from src.core.events import Event
class Ledger:
"""Append-only in-memory ledger for the first vertical slice."""
def __init__(self) -> None:
self._events: list[Event] = []
self._seen_ids: set[str] = set()
@property
def events(self) -> tuple[Event, ...]:
return tuple(self._events)
def append(self, event: Event) -> Event:
if event.id in self._seen_ids:
return event
self._events.append(event)
self._seen_ids.add(event.id)
obs.log("ledger.append", level="debug", id=event.id, kind=event.kind, actor=event.actor, turn=event.turn)
obs.incr("ledger.events", 1, kind=event.kind)
return event
def extend(self, events: Iterable[Event]) -> None:
for event in events:
self.append(event)
def reset(self) -> None:
obs.log("ledger.reset", level="debug", events=len(self._events))
self._events.clear()
self._seen_ids.clear()
def events_for_run(self, run_id: str) -> tuple[Event, ...]:
"""Return the events of *run_id* in append/offset order."""
return tuple(e for e in self._events if e.run_id == run_id)
def runs(self) -> tuple[str, ...]:
"""Return the distinct run_ids in first-seen order."""
seen: dict[str, None] = {}
for e in self._events:
seen.setdefault(e.run_id, None)
return tuple(seen)
|