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

> Discover how Event Sourcing works in Q00/ouroboros using BaseEvent and aggregate types. Learn about immutable event recording and deterministic replay with SQLite.

- Repository: [Q00/ouroboros](https://github.com/Q00/ouroboros)
- Tags: deep-dive
- Published: 2026-03-14

---

**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`](https://github.com/Q00/ouroboros/blob/main/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

```python
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`](https://github.com/Q00/ouroboros/blob/main/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`](https://github.com/Q00/ouroboros/blob/main/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:

```sql
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`](https://github.com/Q00/ouroboros/blob/main/src/ouroboros/orchestrator/events.py):

```python
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

```python
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

```python
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

```python
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`](https://github.com/Q00/ouroboros/blob/main/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`](https://github.com/Q00/ouroboros/blob/main/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.