event_recorder

event_recorder

Event recording system for agent execution telemetry.

Classes

EventType

Bases: str, Enum

Supported event types for agent telemetry.

AgentEvent dataclass

AgentEvent(event_type: str, timestamp: float, metadata: Dict[str, Any] = dict())

Single event recorded during agent execution.

EventRecorder

EventRecorder()

Thread-safe recorder for agent execution events.

Records events with timestamps for later correlation with energy telemetry. All operations are thread-safe for use in concurrent agent execution.

Example: >>> recorder = EventRecorder() >>> recorder.record('tool_call_start', tool='calculator') >>> recorder.record('tool_call_end', tool='calculator') >>> events = recorder.get_events() >>> len(events) 2

Initialize the event recorder.

Source code in src/openjarvis/evals/core/event_recorder.py

def __init__(self) -> None:
    """Initialize the event recorder."""
    self._events: List[AgentEvent] = []
    self._lock = threading.Lock()

Functions

record

record(event_type: str, **metadata: Any) -> None

Record an event with current timestamp.

Args: event_type: Type of event (e.g., 'tool_call_start', 'lm_inference_end') **metadata: Additional metadata to attach to the event

Source code in src/openjarvis/evals/core/event_recorder.py

def record(self, event_type: str, **metadata: Any) -> None:
    """Record an event with current timestamp.

    Args:
        event_type: Type of event (e.g., 'tool_call_start', 'lm_inference_end')
        **metadata: Additional metadata to attach to the event
    """
    event = AgentEvent(
        event_type=event_type,
        timestamp=time.time(),
        metadata=metadata,
    )
    with self._lock:
        self._events.append(event)

get_events

get_events() -> List[AgentEvent]

Return a copy of all recorded events.

Returns: List of all recorded events in chronological order.

Source code in src/openjarvis/evals/core/event_recorder.py

def get_events(self) -> List[AgentEvent]:
    """Return a copy of all recorded events.

    Returns:
        List of all recorded events in chronological order.
    """
    with self._lock:
        return list(self._events)

clear

clear() -> None

Clear all recorded events.

Source code in src/openjarvis/evals/core/event_recorder.py

def clear(self) -> None:
    """Clear all recorded events."""
    with self._lock:
        self._events.clear()