Event Sourcing

Event-driven state management: immutable event log, projections, replay, and audit.

Why Event Sourcing?

Core Concepts

Events = Facts

@dataclass
class Event:
    event_id: str              # UUID
    event_type: str            # "AgentStarted", "ToolCalled", "DecisionMade"
    aggregate_id: str          # Thread/agent ID
    timestamp: datetime        # UTC
    payload: dict              # Event data (JSON-serializable)
    metadata: EventMetadata

@dataclass
class EventMetadata:
    causation_id: str | None   # Event that caused this
    correlation_id: str | None # Business correlation
    user_id: str | None
    version: int = 1           # Event schema version

Event Types for Agents

Event Store Interface

class EventStore:
    async def append(self, events: list[Event]) -> None:
        """Append events atomically"""

    async def get_events(self, aggregate_id: str, 
                         from_version: int = 0) -> list[Event]:
        """Get all events for aggregate"""

    async def get_events_by_correlation(self, correlation_id: str) -> list[Event]:
        """Get events across aggregates"""

    async def subscribe(self, event_types: list[str]) -> AsyncIterator[Event]:
        """Real-time event stream"""

Aggregates (State from Events)

Aggregate Base

class Aggregate:
    def __init__(self, aggregate_id: str):
        self.aggregate_id = aggregate_id
        self.version = 0
        self.uncommitted_events: list[Event] = []

    def apply(self, event: Event):
        """Apply event to update state"""
        handler = getattr(self, f"_apply_{event.event_type}", None)
        if handler:
            handler(event.payload)
        self.version = event.version

    def raise_event(self, event_type: str, payload: dict):
        event = Event(
            event_id=str(uuid4()),
            event_type=event_type,
            aggregate_id=self.aggregate_id,
            timestamp=datetime.utcnow(),
            payload=payload,
            version=self.version + 1
        )
        self.apply(event)
        self.uncommitted_events.append(event)

    def commit(self) -> list[Event]:
        events = self.uncommitted_events
        self.uncommitted_events = []
        return events

Agent Aggregate

class AgentAggregate(Aggregate):
    def __init__(self, agent_id: str):
        super().__init__(agent_id)
        self.state = AgentState(agent_id=agent_id)

    # Command handlers
    def start_task(self, goal: str, user_id: str):
        self.raise_event("AgentTaskStarted", {
            "goal": goal,
            "user_id": user_id,
            "timestamp": datetime.utcnow().isoformat()
        })

    def call_tool(self, tool_name: str, args: dict):
        self.raise_event("ToolCallStarted", {
            "tool_name": tool_name,
            "args": args,
            "call_id": str(uuid4())
        })

    def complete_tool(self, call_id: str, result: dict):
        self.raise_event("ToolCallCompleted", {
            "call_id": call_id,
            "result": result
        })

    def request_human_approval(self, context: dict):
        self.raise_event("HumanApprovalRequested", {
            "context": context,
            "request_id": str(uuid4())
        })

    # Event handlers
    def _apply_AgentTaskStarted(self, payload):
        self.state.goal = payload["goal"]
        self.state.user_id = payload["user_id"]
        self.state.status = "running"

    def _apply_ToolCallCompleted(self, payload):
        self.state.tool_results[payload["call_id"]] = payload["result"]
        self.state.current_step += 1

Projections (Read Models)

Projection = Derived View

class Projection:
    def __init__(self, name: str, event_store: EventStore):
        self.name = name
        self.event_store = event_store
        self.checkpoint_version = 0

    async def project(self, events: list[Event]):
        for event in events:
            handler = getattr(self, f"on_{event.event_type}", None)
            if handler:
                await handler(event)
        self.checkpoint_version = max(e.version for e in events) if events else self.checkpoint_version

    async def run(self, from_version: int = 0):
        async for event in self.event_store.get_events_from_version(from_version):
            await self.project([event])

Example Projections

# 1. Conversation History (for UI)
class ConversationProjection(Projection):
    def __init__(self, db_pool):
        super().__init__("conversation", ...)
        self.db = db_pool

    async def on_UserMessageReceived(self, event):
        await self.db.execute("""
            INSERT INTO messages (thread_id, role, content, created_at)
            VALUES ($1, $2, $3, $4)
        """, event.aggregate_id, "user", event.payload["content"], event.timestamp)

    async def on_AssistantMessageSent(self, event):
        await self.db.execute("""
            INSERT INTO messages (thread_id, role, content, created_at)
            VALUES ($1, $2, $3, $4)
        """, event.aggregate_id, "assistant", event.payload["content"], event.timestamp)

# 2. Analytics (for dashboards)
class AnalyticsProjection(Projection):
    async def on_AgentTaskStarted(self, event):
        await self.increment("tasks_started", event.payload["user_id"])

    async def on_ToolCallCompleted(self, event):
        await self.increment("tool_calls", event.payload["tool_name"])
        await self.record_latency("tool_latency", event.payload["duration_ms"])

# 3. Audit Log (immutable)
class AuditProjection(Projection):
    async def on_Event(self, event):
        await self.audit_log.write({
            "event_id": event.event_id,
            "type": event.event_type,
            "aggregate": event.aggregate_id,
            "timestamp": event.timestamp,
            "user": event.metadata.user_id,
            "payload": event.payload
        })

Replay & Rebuild

Rebuild Aggregate from Events

class AggregateRepository:
    def __init__(self, event_store: EventStore):
        self.event_store = event_store

    async def load(self, aggregate_id: str, aggregate_class) -> Aggregate:
        aggregate = aggregate_class(aggregate_id)
        events = await self.event_store.get_events(aggregate_id)
        for event in events:
            aggregate.apply(event)
        return aggregate

    async def save(self, aggregate: Aggregate):
        events = aggregate.commit()
        if events:
            await self.event_store.append(events)

Replay for Debugging

class EventReplayer:
    def __init__(self, event_store: EventStore):
        self.event_store = event_store

    async def replay_aggregate(self, aggregate_id: str) -> list[StateSnapshot]:
        """Return state at each version"""
        aggregate = Aggregate(aggregate_id)
        snapshots = []

        events = await self.event_store.get_events(aggregate_id)
        for event in events:
            aggregate.apply(event)
            snapshots.append(StateSnapshot(
                version=event.version,
                timestamp=event.timestamp,
                event_type=event.event_type,
                state=copy.deepcopy(aggregate.state)
            ))
        return snapshots

    async def replay_from_timestamp(self, timestamp: datetime) -> list[Event]:
        """Replay all events after timestamp"""
        return await self.event_store.get_events_after(timestamp)

Rebuild Projections

class ProjectionRebuilder:
    def __init__(self, event_store: EventStore, projections: list[Projection]):
        self.event_store = event_store
        self.projections = projections

    async def rebuild_all(self, from_version: int = 0):
        # Reset all projections
        for proj in self.projections:
            await proj.reset()

        # Replay all events
        async for event in self.event_store.get_events_from_version(0):
            for proj in self.projections:
                await proj.project([event])

Event Schema Evolution

Versioning

# Event payloads versioned
@dataclass
class Event:
    version: int = 1  # Event schema version
    payload: dict

# Upcasters: v1 → v2
class Upcaster:
    def upcast(self, event: Event) -> Event:
        if event.version == 1 and event.event_type == "ToolCallCompleted":
            # v1 had "output", v2 has "result" + "metadata"
            if "output" in event.payload:
                event.payload["result"] = event.payload.pop("output")
                event.payload["metadata"] = {}
                event.version = 2
        return event

# Registry
class UpcasterRegistry:
    def __init__(self):
        self.upcasters = {}

    def register(self, event_type: str, upcaster: Upcaster):
        self.upcasters[event_type] = upcaster

    def upcast(self, event: Event) -> Event:
        while event.version < CURRENT_VERSION:
            upcaster = self.upcasters.get(event.event_type)
            if not upcaster:
                break
            event = upcaster.upcast(event)
        return event

Snapshotting (Performance)

Snapshot = State + Version

@dataclass
class Snapshot:
    aggregate_id: str
    version: int
    state: dict
    timestamp: datetime

class SnapshotStore:
    def __init__(self, store, snapshot_interval=100):
        self.store = store
        self.interval = snapshot_interval

    async def maybe_snapshot(self, aggregate: Aggregate):
        if aggregate.version % self.interval == 0:
            snapshot = Snapshot(
                aggregate_id=aggregate.aggregate_id,
                version=aggregate.version,
                state=asdict(aggregate.state),
                timestamp=datetime.utcnow()
            )
            await self.store.save(snapshot)

    async def load_snapshot(self, aggregate_id: str) -> Snapshot | None:
        return await self.store.get_latest(aggregate_id)

Load with Snapshot

class SnapshottingRepository:
    def __init__(self, event_store, snapshot_store, aggregate_class):
        self.event_store = event_store
        self.snapshot_store = snapshot_store
        self.aggregate_class = aggregate_class

    async def load(self, aggregate_id: str) -> Aggregate:
        # 1. Try snapshot
        snapshot = await self.snapshot_store.load_snapshot(aggregate_id)

        if snapshot:
            aggregate = self.aggregate_class(aggregate_id)
            aggregate.state = snapshot.state
            aggregate.version = snapshot.version
            from_version = snapshot.version
        else:
            aggregate = self.aggregate_class(aggregate_id)
            from_version = 0

        # 2. Replay events after snapshot
        events = await self.event_store.get_events(aggregate_id, from_version)
        for event in events:
            aggregate.apply(event)

        return aggregate

CQRS (Command Query Responsibility Segregation)

Commands (Write)

@dataclass
class Command:
    command_id: str
    command_type: str
    aggregate_id: str
    payload: dict
    metadata: CommandMetadata

@dataclass
class CommandMetadata:
    user_id: str
    correlation_id: str
    timestamp: datetime

class CommandHandler:
    def __init__(self, repository: AggregateRepository):
        self.repository = repository

    async def handle(self, command: Command):
        aggregate = await self.repository.load(command.aggregate_id)
        handler = getattr(aggregate, f"handle_{command.command_type}")
        handler(command.payload)
        await self.repository.save(aggregate)

Queries (Read)

@dataclass
class Query:
    query_type: str
    payload: dict

class QueryHandler:
    def __init__(self, read_model):
        self.read_model = read_model

    async def handle(self, query: Query) -> Any:
        handler = getattr(self, f"handle_{query.query_type}")
        return await handler(query.payload)

# Example read model
class AgentReadModel:
    async def get_active_agents(self) -> list[AgentSummary]:
        # Query projection DB
        pass

    async def get_thread_history(self, thread_id: str) -> list[Message]:
        # Query conversation projection
        pass

Related Notes

• Checkpointing — Checkpointing vs event sourcing
• Session Management — Sessions as aggregates
• Distributed State — Distributed event sourcing
• Versioning Migration — Event schema evolution
• Memory Systems — Events as memory
• LangGraph — LangGraph event streaming
• Observability — Event-based observability
• Compliance — Audit trails
• Research Directions — Event-native architectures

Sources: "Event Sourcing" (Fowler), "Domain-Driven Design" (Evans), "Event Sourcing" (Young), temporal.io, Akka Persistence, EventStoreDB, Axon Framework, Kafka, CDC patterns, Greg Young's ES talks.