orxtra v0.13.0 /trace.src.orxtra.trace._trace_sink
Updated
On this page

TraceSink adapts OverseerEvent dataclasses (RunStarted, TaskEscalated, etc.) into trace events: converts the class name to snake_case as event_type and serializes fields via write_event.

#trace.src.orxtra.trace._trace_sink

#trace.src.orxtra.trace._trace_sink

TraceSink -- EventSink[OverseerEvent] that writes events to trace.

Replaces the old event_callback pattern on TraceWriter/PgBackend/InMemoryBackend.

DESIGN NOTE: The removed event_callback was a notification mechanism that fired AFTER trace writes happened (write_event, transition_run, transition_task). It propagated trace-level events to listeners with signature (event_id, run_id, event_type, data) -> Awaitable[None].

TraceSink serves the opposite direction: it RECEIVES OverseerEvent objects (RunStarted, TaskEscalated, etc.) and WRITES them to trace as new events. The old event_callback was unused in production (no caller ever passed it), so removing it has no behavioral impact.

#_to_snake_case

python
def _to_snake_case(name: str) -> str

Convert CamelCase to snake_case.

#_serialize_event

python
def _serialize_event(event: OverseerEvent) -> dict[str, Any]

Serialize a dataclass event to a JSON-safe dict.

#TraceSink

EventSink[OverseerEvent] that persists events via TraceWriter.

Transforms each OverseerEvent into a write_event call with:

  • event_type: snake_case of the class name (e.g., RunStarted -> run_started)
  • data: serialized fields of the event dataclass

#on_event

python
async def on_event(self, event: OverseerEvent) -> None

Write the OverseerEvent to trace storage, attributed to the run.

Search