Event Sourcing for AI Agents: State That Doesn't Lie
Your agent is processing an invoice: it fetches the client, validates amounts, calls three external tools, builds the payload — and crashes on a 30-second timeout with one step left.
What do you do? Restart from scratch and hope idempotency holds? Read the database to infer what ran and what didn't? Or honestly admit you have no idea exactly where it left off?
That "I don't know exactly where it stopped" is the problem event sourcing solves. And it's a problem that surfaces precisely when agent flows get complex enough to matter.
Why agent state lies by default
When an agent updates its state through tool calls and external APIs, the value you see in the database at any given moment is a snapshot, not a history. You know where the process is now, but not how it got there.
This causes three real problems in production:
You can't resume precisely. If the agent crashed after fetch_client but before create_invoice, your database might show status: "processing" with no further context. Restarting from scratch risks duplicating fetch_client; jumping straight to create_invoice risks working with data that was never retrieved.
Debugging is archaeology. When a flow produces a wrong result three hours after execution, what tools did the agent call? In what order? What did each one return? Without a structured action log, you're staring at the final state and reconstructing backwards.
Audit trails don't exist. For regulated processes — invoicing, contracts, customer data — you need to prove what the agent did, when, and with what data. An updated_at field in the database is not an audit trail.
What event sourcing means for AI agents
The idea is simple: instead of storing the agent's current state, you store each action as an immutable event. The current state is always a projection derived from those events.
type AgentEvent =
| { type: "agent.started"; taskId: string; input: unknown; seq: number; ts: number }
| { type: "tool.called"; tool: string; args: unknown; seq: number; ts: number }
| { type: "tool.result"; tool: string; result: unknown; seq: number; ts: number }
| { type: "agent.crashed"; reason: string; seq: number; ts: number }
| { type: "task.completed"; output: unknown; seq: number; ts: number };
interface EventStore {
append(taskId: string, event: Omit<AgentEvent, "seq">): Promise<AgentEvent>;
list(taskId: string, fromSeq?: number): Promise<AgentEvent[]>;
}
The seq field is the sequence number. It's what makes replay possible: fetch all events from sequence 2 and reconstruct exactly what state the agent had at that point.
Building the event store in TypeScript
A minimal implementation over PostgreSQL (or your existing database) has three pieces:
// The table in your DB
// CREATE TABLE agent_events (
// id SERIAL PRIMARY KEY,
// task_id TEXT NOT NULL,
// seq INT NOT NULL,
// type TEXT NOT NULL,
// payload JSONB NOT NULL,
// ts BIGINT NOT NULL,
// UNIQUE (task_id, seq)
// );
async function appendEvent(
db: Pool,
taskId: string,
type: string,
payload: unknown,
seq: number
): Promise<void> {
await db.query(
`INSERT INTO agent_events (task_id, seq, type, payload, ts)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (task_id, seq) DO NOTHING`,
[taskId, seq, type, JSON.stringify(payload), Date.now()]
);
}
async function getEvents(
db: Pool,
taskId: string,
fromSeq = 0
): Promise<AgentEvent[]> {
const rows = await db.query(
`SELECT seq, type, payload, ts FROM agent_events
WHERE task_id = $1 AND seq >= $2 ORDER BY seq ASC`,
[taskId, fromSeq]
);
return rows.rows.map(r => ({ ...r.payload, type: r.type, seq: r.seq, ts: r.ts }));
}
The ON CONFLICT DO NOTHING on insert is the key to idempotency: if the agent restarts and tries to write the same event again, it silently skips it.
Projecting state from events
Once you have the event log, deriving the current state is a pure reduction:
interface TaskState {
taskId: string;
status: "running" | "crashed" | "completed";
lastSeq: number;
toolResults: Record<string, unknown>;
output?: unknown;
}
function projectState(events: AgentEvent[]): TaskState {
return events.reduce<TaskState>(
(state, event) => {
switch (event.type) {
case "agent.started":
return { ...state, status: "running", lastSeq: event.seq };
case "tool.result":
return {
...state,
lastSeq: event.seq,
toolResults: { ...state.toolResults, [event.tool]: event.result },
};
case "agent.crashed":
return { ...state, status: "crashed", lastSeq: event.seq };
case "task.completed":
return { ...state, status: "completed", lastSeq: event.seq, output: event.output };
default:
return { ...state, lastSeq: event.seq };
}
},
{ taskId: "", status: "running", lastSeq: 0, toolResults: {} }
);
}
When the agent restarts after a crash, it calls getEvents(taskId), projects the state, and knows exactly which tools it already called and what results it got. Nothing repeats.
Wiring it into your agent loop
Inside the main agent loop, each tool call becomes two events:
async function callToolWithEvents(
store: EventStore,
taskId: string,
tool: string,
args: unknown,
seq: number
): Promise<unknown> {
// If result exists in projected state, skip the call and return cached result
const events = await store.list(taskId);
const state = projectState(events);
if (state.toolResults[tool]) return state.toolResults[tool];
await store.append(taskId, { type: "tool.called", tool, args, seq, ts: Date.now() });
const result = await callTool(tool, args); // your actual implementation
await store.append(taskId, { type: "tool.result", tool, result, seq: seq + 1, ts: Date.now() });
return result;
}
With this pattern, if the agent crashes between tool.called and tool.result — right during the external call — it restarts and retries the call (no cached result). If it crashes after tool.result, it finds the cached result and skips it. Either way, the behavior is deterministic.
When to use event sourcing — and when not to
Event sourcing has a real cost: more database writes, more agent logic, state that lives in two places (event + projection). It's not for every flow.
It makes sense when:
- The flow is long (more than 3–4 sequential tool calls)
- Any tool call has external side effects that are hard to reverse (sending an email, creating a CRM record, calling a payment API)
- You need an audit trail by contract or regulation
- The cost of replaying from scratch is high (time, money, risk of duplicates)
You don't need it when the flow is short, all operations are idempotent by design, or the agent can simply retry without consequences.
How this differs from observability and from Saga
Event sourcing is not the same as OpenTelemetry tracing. Tracing records what happened so you can read it. Event sourcing records what happened so the agent itself can read it and continue.
It's also not the Saga pattern, which manages compensating transactions when something fails — undoing what was done. Event sourcing is neutral on compensation: it simply guarantees you have the full history to make any decision, including the decision to compensate.
They're complementary patterns. A robust production agent typically uses both: event sourcing for state, Saga for compensation logic when needed. At DAILYMP, this is the architecture we apply for agent flows that touch critical external systems.
The symptom to watch for
If your agent logs show any variant of "I don't know where the previous run stopped, restarting from scratch", you have the problem event sourcing solves. It's not always obvious because silent restarts from scratch usually work — until they don't.
The cost of catching it late is high: duplicate CRM records, invoices sent twice, inconsistent state that support spends hours diagnosing.
If you're building agents that touch critical business processes — billing, contracts, customer support, data pipelines — the question isn't whether you need event sourcing. It's when the absence of it will hurt.
Want to see how this applies to a real flow in your stack? In AI integration for businesses we build the resilience layer from the start, not as a retroactive patch.