# Event Sourcing

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

---

## Why Event Sourcing?

| Traditional (State) | Event Sourcing |
|---|---|
| Current state only | Full history |
| Hard to audit | Complete audit trail |
| Hard to debug | Time-travel debugging |
| Schema changes break | Events immutable, project to new schema |
| Hard to replicate | Natural replication via event log |
| Lost intent | "Why" preserved in events |

---

## Core Concepts

### Events = Facts
```python
@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
| Category | Events |
|---|---|
| **Lifecycle** | AgentStarted, AgentPaused, AgentResumed, AgentCompleted, AgentFailed |
| **Conversation** | UserMessageReceived, AssistantMessageSent, SystemPromptSet |
| **Planning** | PlanCreated, PlanUpdated, PlanStepCompleted, PlanFailed |
| **Tools** | ToolCallStarted, ToolCallCompleted, ToolCallFailed, ToolCallRetried |
| **Memory** | MemoryStored, MemoryRetrieved, MemoryConsolidated, MemoryForgotten |
| **Reasoning** | ThoughtGenerated, DecisionMade, HypothesisFormed, VerificationDone |
| **HITL** | HumanApprovalRequested, HumanApprovalGranted, HumanInputReceived |
| **Errors** | ErrorOccurred, ErrorRecovered, RetryAttempted, CircuitBreakerOpened |

### Event Store Interface
```python
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
```python
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
```python
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
```python
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
```python
# 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
```python
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
```python
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
```python
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
```python
# 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
```python
@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
```python
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)
```python
@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)
```python
@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

- [[StateManagement/Checkpointing.md]] — Checkpointing vs event sourcing
- [[StateManagement/Session-Management.md]] — Sessions as aggregates
- [[StateManagement/Distributed-State.md]] — Distributed event sourcing
- [[StateManagement/Versioning-Migration.md]] — Event schema evolution
- [[ContextEngineering/Memory-Systems.md]] — Events as memory
- [[Frameworks/LangGraph.md]] — LangGraph event streaming
- [[Production/Observability.md]] — Event-based observability
- [[Production/Compliance.md]] — Audit trails
- [[Frontiers/Research-Directions.md]] — 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.*