How Event Sourcing Works in Q00/ouroboros: BaseEvent and Aggregate Types Explained

Event Sourcing in Q00/ouroboros records every state change as an immutable BaseEvent object, grouping events by aggregate_type and aggregate_id to enable deterministic replay via a lightweight SQLite backend.

The architecture treats the event store as the single source of truth, capturing the complete lifecycle of entities like sessions, executions, and ontologies through append-only, past-tense event records.

The BaseEvent Model

Every event in the system inherits from BaseEvent, defined in src/ouroboros/events/base.py. This Pydantic model enforces immutability while providing a consistent schema for the event store.

Core Event Schema

A BaseEvent contains the following fields:

• id: UUID that uniquely identifies the event
• type: Dot-notation, past-tense name (e.g., orchestrator.session.started)
• timestamp: UTC time when the event was created
• aggregate_type: The domain of the entity (e.g., session, execution, ontology)
• aggregate_id: The unique identifier of that specific entity
• data: Payload specific to the event
• consensus_id: Optional grouping marker for consensus-driven events

class BaseEvent(BaseModel, frozen=True):
    id: UUID
    type: str
    timestamp: datetime
    aggregate_type: str
    aggregate_id: str
    data: dict[str, Any] = Field(default_factory=dict)
    consensus_id: Optional[str] = None

Immutability Guarantees

The frozen=True parameter makes every event instance immutable after creation. This prevents accidental mutation of historical records, ensuring audit trails remain trustworthy and replay operations remain deterministic.

The EventStore Implementation

The EventStore class in src/ouroboros/persistence/event_store.py provides the persistence layer, operating over an async SQLite database using sqlite+aiosqlite.

Storing Events

The store provides atomic operations for both single and batch writes:

• append(event) inserts a single event using event.to_db_dict()
• append_batch(events) writes multiple events in one transaction, ensuring atomicity

Both methods serialize the Pydantic model into a dictionary format optimized for database insertion.

Database Schema and Indexing

The underlying schema in src/ouroboros/persistence/schema.py optimizes retrieval speed through strategic indexing:

• ix_events_aggregate_type: Enables quick filtering by domain
• ix_events_aggregate_type_id: A composite index on (aggregate_type, aggregate_id) that accelerates aggregate-specific queries

When replaying an aggregate's history, the store executes:

SELECT * FROM events
WHERE aggregate_type = :aggregate_type
  AND aggregate_id   = :aggregate_id
ORDER BY timestamp, id;

The composite index ensures this query remains performant even as the event store grows.

Aggregate-Centric Replay

An aggregate represents any logical entity whose complete lifecycle can be reconstructed from its events. Common aggregates in Ouroboros include sessions, executions, and lineage traces.

Reconstructing State

To rebuild an aggregate's current state, the system:

1. Calls EventStore.replay(aggregate_type, aggregate_id) to fetch chronologically ordered events
2. Receives a list of BaseEvent objects reconstructed via BaseEvent.from_db_row
3. Feeds the events into a domain-specific projector (e.g., SessionProjector or LineageProjector) that folds them into a mutable state object

Because events are immutable and ordered by timestamp plus ID, the replay is deterministic and can be repeated for debugging, audit trails, or session resumption.

Why Aggregate Types Matter

The aggregate_type field serves critical architectural purposes:

• Separation of concerns: Each domain maintains its own event stream, preventing type collisions across different contexts
• Efficient queries: The composite index allows the store to fetch only relevant event slices rather than scanning entire tables
• Clear replay boundaries: Projectors know exactly when an aggregate's event stream ends, simplifying state reconstruction logic

Creating Domain Events

Domain modules provide factory functions that pre-populate type, aggregate_type, and aggregate_id to ensure consistency. For the orchestrator domain in src/ouroboros/orchestrator/events.py:

def create_session_started_event(session_id, execution_id, seed_id, seed_goal) -> BaseEvent:
    return BaseEvent(
        type="orchestrator.session.started",
        aggregate_type="session",
        aggregate_id=session_id,
        data={
            "execution_id": execution_id,
            "seed_id": seed_id,
            "seed_goal": seed_goal,
        },
    )

Similar factories exist for ontology, lineage, and evaluation aggregates, each hardcoding the appropriate aggregate_type to maintain domain boundaries.

Practical Implementation Examples

Creating and Persisting a Session Event

from ouroboros.orchestrator.events import create_session_started_event
from ouroboros.persistence.event_store import EventStore
import asyncio

async def demo():
    # Build the event using the domain factory

    ev = create_session_started_event(
        session_id="sess_001",
        execution_id="exec_001",
        seed_id="seed_42",
        seed_goal="Generate a report"
    )

    # Persist to the event store

    store = EventStore("sqlite+aiosqlite:///tmp/ouroboros.db")
    await store.initialize()
    await store.append(ev)
    await store.close()

asyncio.run(demo())

Replaying an Aggregate's Event Stream

from ouroboros.persistence.event_store import EventStore
import asyncio

async def replay_session():
    store = EventStore("sqlite+aiosqlite:///tmp/ouroboros.db")
    await store.initialize()

    # Retrieve all events for a specific session aggregate

    events = await store.replay(aggregate_type="session", aggregate_id="sess_001")
    for ev in events:
        print(f"{ev.type}: {ev.data}")

    await store.close()

asyncio.run(replay_session())

Batch Appending Multiple Events

from ouroboros.orchestrator.events import (
    create_progress_event,
    create_task_completed_event,
)
from ouroboros.persistence.event_store import EventStore
import asyncio

async def batch_demo():
    store = EventStore()
    await store.initialize()

    batch = [
        create_progress_event(
            session_id="sess_001",
            message_type="assistant",
            content_preview="Analyzing requirements..."
        ),
        create_task_completed_event(
            session_id="sess_001",
            acceptance_criterion="AC-001",
            success=True,
            result_summary="Requirement analysis finished."
        ),
    ]

    # Atomic batch insertion

    await store.append_batch(batch)
    await store.close()

asyncio.run(batch_demo())

Summary

• BaseEvent in src/ouroboros/events/base.py provides an immutable, frozen Pydantic model for all domain events
• The aggregate_type and aggregate_id pair partitions the event store by logical entity domains, enabling efficient replay
• EventStore in src/ouroboros/persistence/event_store.py handles atomic append operations and deterministic replay via SQLite
• Composite indexes on (aggregate_type, aggregate_id) ensure replay queries remain performant at scale
• Factory functions in domain modules standardize event creation while enforcing correct aggregate classification

Frequently Asked Questions

What is the difference between aggregate_type and aggregate_id in Ouroboros?

The aggregate_type identifies the domain category (e.g., session, ontology, execution), while aggregate_id uniquely identifies a specific instance within that category (e.g., sess_001). This pairing allows the EventStore to isolate event streams by both domain and entity, ensuring projectors only receive relevant events during replay.

How does Ouroboros ensure events remain immutable?

Events inherit from BaseEvent, which sets frozen=True in its Pydantic configuration. This prevents any modification to event fields after instantiation, guaranteeing that historical records cannot be altered accidentally. The frozen model also makes events hashable and thread-safe for concurrent processing.

Can I replay events across multiple aggregate types simultaneously?

The replay() method in EventStore requires both aggregate_type and aggregate_id parameters, querying the composite index for a specific entity slice. To replay across types, you would need to issue separate replay calls for each aggregate or implement a custom query that filters only by timestamp, though this would bypass the performance optimizations designed for aggregate-centric reconstruction.

What database backend does Ouroboros use for event storage?

Ouroboros uses SQLite with aiosqlite as its default backend, specified via the connection string sqlite+aiosqlite. The EventStore class initializes this connection asynchronously and manages schema creation including the critical composite indexes on aggregate_type and aggregate_id for efficient query performance.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}