The event store: the run becomes a ledger
Run the Chapter 7 agent and close the terminal: everything it did is gone. The transcript lived in a Python list, and lists die with their process. Before this book can talk about crashes, resumption, audits, or fleets, the run needs a durable home, and the shape of that home is the single most consequential storage decision in the platform. This chapter makes the decision (an append-only event log) and implements it in about thirty lines.
The ledger, not the balance
The concepts chapter introduced the idea with a bank account: the bank does not store your balance and overwrite it; it stores the transactions, and the balance is computed. Event sourcing applies the same discipline to the agent. We never store "the current state of the run" as a thing to update. We store what happened, append-only, and compute any state we need by reading the history.
The alternative (a mutable runs row with status and last_step
columns, updated in place) always looks simpler on day one, and it
quietly discards the history every update. Then the questions arrive that
only history can answer. What did the agent actually see before it did
that? Where exactly were we when the process died? Can we re-run
yesterday's incident? With a ledger, each of those is a read. With a
mutable row, each is a shrug.
Concretely, an event is three fields:
{"seq": 4, "type": "tool.returned",
"data": {"id": "t1", "name": "query_metrics", "output": "..."}}
seq is a counter giving every event an address ("resume after 6" needs
addresses). type is a past-tense name, and the tense is a design
statement: events record facts that already occurred, never intentions.
Five types cover Part 1: run.created, model.responded, tool.called,
tool.returned, run.completed. One deliberate omission: no wall-clock
timestamps in the demo, because Chapter 10 will demand
byte-identical replays, and timestamps are nondeterminism smuggled in as
metadata. Real systems record them, and exclude them from replay
comparisons; the lab simply keeps the lesson pure.
Why a JSONL file
The storage engine for Part 1 is a JSON Lines file: one JSON object per line, appended to the end. This is not a toy compromise; it is the smallest thing with the right properties, and naming those properties tells you what to demand from the grown-up version later:
- Append is the safest write there is. No seeking, no rewriting existing bytes. A crash mid-write can only damage the final line, and a torn final line fails to parse, which makes corruption detectable: truncate the tail and you are consistent again.
- It is inspectable with nothing.
grep,tail, and your eyes work. For a debugging-centric design, that is a feature with compound interest. - It maps one-to-one onto the production version. In Part 7 the same
design lands on DynamoDB (run id as partition key,
seqas sort key, conditional writes preventing two writers from claiming the sameseq). Nothing conceptual changes; only the durability and concurrency story grows up.
The store itself is barely code, and that is the point:
class EventStore:
def append(self, etype: str, data: dict) -> dict:
self._seq += 1
event = {"seq": self._seq, "type": etype, "data": data}
with self.path.open("a") as f:
f.write(json.dumps(event) + "\n")
return event
State is a fold
If we never store state, how do we ever have state? By folding: start
from an empty state, apply events one at a time, each nudging the state
forward. (The name comes from functional programming, where fold means
exactly "combine a sequence into one value"; Python's sum is a fold
over numbers.)
def fold(events, state=None):
s = state or {"status": "empty", "turns": 0, "tool_calls": 0, ...}
for e in events:
if e["type"] == "run.created": s["status"] = "running"
elif e["type"] == "model.responded": s["turns"] += 1
elif e["type"] == "tool.returned": s["tool_calls"] += 1
elif e["type"] == "run.completed": s["status"] = "completed"
s["applied_through"] = e["seq"]
return s
Because state is derived, it cannot drift from history; there is no
second copy to disagree with the first. And because fold accepts a
starting state, we get checkpoints almost for free. A checkpoint is a
saved fold result plus the seq it covers; resuming means folding only
the events after it. Same answer, less reading. That equality (snapshot
plus suffix equals full fold) is the entire mathematical content of
"resume from checkpoint," and the demo below asserts it rather than
asking you to believe it.
The remaining piece is instrumenting the loop, and it is deliberately
anticlimactic: run_agent_logged is Chapter 7's loop with an append
next to each step that matters. Model responded: append. Tool called,
tool returned: append, append. The loop's logic does not change, which is
the property to preserve as the recorder follows us into fleets: the
flight recorder observes; it never steers.
Run it
python3 event_store.py
--- the ledger: 11 events in events.jsonl ---
1 run.created {"task": "Checkout p99 latency spiked overnight. Find th...
2 model.responded {"stop_reason": "tool_use", "text": "Starting with the l...
3 tool.called {"id": "t1", "name": "query_metrics", "args": {"service"...
4 tool.returned {"id": "t1", "name": "query_metrics", "output": "{\"serv...
5 model.responded {"stop_reason": "tool_use", "text": "p99 jumps from ~210...
6 tool.called {"id": "t2", "name": "list_deploys", "args": {"service":...
7 tool.returned {"id": "t2", "name": "list_deploys", "output": "{\"servi...
8 tool.called {"id": "t3", "name": "search_logs", "args": {"query": "p...
9 tool.returned {"id": "t3", "name": "search_logs", "output": "{\"query\...
10 model.responded {"stop_reason": "end_turn", "text": "Likely cause: deplo...
11 run.completed {"answer": "Likely cause: deploy v841 at 01:55 shrank th...
--- state, folded from the ledger (never stored) ---
status completed
task Checkout p99 latency spiked overnight. Find the likely cause...
turns 3
tool_calls 3
answer Likely cause: deploy v841 at 01:55 shrank the payments-db co...
applied_through 11
--- the checkpoint property ---
snapshot at seq 6, then fold the rest: equals full fold? True
Eleven events tell the investigation's whole story, in order, with
addresses. The folded state is a summary that was computed just now, from
nothing but the file; delete the state, nothing is lost. And the
checkpoint line is the quiet star: folding a snapshot taken at seq 6
through the remaining five events lands on exactly the state that
folding all eleven produces. When Part 6 kills a five-hundred-agent fleet
mid-run and resumes it, this True is the property doing the heavy
lifting; only the numbers get bigger.
Don't be confused: the event log vs the message list. The loop still maintains its
messageslist; the model still receives it every call. The two records answer to different masters. Messages are what the model sees, shaped by the API and eventually pruned or compacted to fit a context window. Events are what the system did, shaped by our need to audit and resume, and never pruned. The direction matters: you can rebuild the message list from the event log (it captures every response and result), but not the reverse once messages are trimmed. The ledger is the source of truth; the message list is a view.
Whether the file needs stronger guarantees than "the OS will probably
flush it" depends on stakes: the honest durability ladder runs from this
lab's buffered writes, through forcing bytes to disk before
acknowledging (fsync), to replicated storage where a machine can die
without losing the ledger. Part 7 climbs to the top rung; the design
stays identical the whole way, which is why it was worth getting right at
thirty lines.
Full source
"""Lab 1.3: the event store. The agent's history as an append-only ledger.
Three pieces:
1. EventStore: append-only JSON-lines file with sequence numbers,
2. fold(): current state computed from events, never stored,
3. a logged variant of the Lab 1.1 loop that records every step.
The demo runs the checkout investigation from Lab 1.1 silently, then shows
what the recorder captured: the raw ledger, the state folded from it, and
the checkpoint property (folding from a snapshot equals folding from
scratch), which is what makes cheap resumption possible.
Standard library only. Deterministic: same run, same events.
"""
from __future__ import annotations
import json
from pathlib import Path
from loop_agent import INVESTIGATION_SCRIPT, TOOLS, ScriptedModel, Tool
# ---------------------------------------------------------------------------
# 1. The store: append-only JSON lines, one event per line
# ---------------------------------------------------------------------------
class EventStore:
def __init__(self, path: str):
self.path = Path(path)
self.path.write_text("") # fresh ledger for the demo
self._seq = 0
def append(self, etype: str, data: dict) -> dict:
self._seq += 1
event = {"seq": self._seq, "type": etype, "data": data}
with self.path.open("a") as f:
f.write(json.dumps(event) + "\n")
return event
def events(self) -> list[dict]:
out = []
for line in self.path.read_text().splitlines():
out.append(json.loads(line)) # a torn last line would fail here
return out
# ---------------------------------------------------------------------------
# 2. State is a fold over events, not a thing we store
# ---------------------------------------------------------------------------
def fold(events: list[dict], state: dict | None = None) -> dict:
s = state or {"status": "empty", "task": None, "turns": 0,
"tool_calls": 0, "answer": None, "applied_through": 0}
s = dict(s)
for e in events:
data = e["data"]
if e["type"] == "run.created":
s["status"], s["task"] = "running", data["task"]
elif e["type"] == "model.responded":
s["turns"] += 1
elif e["type"] == "tool.returned":
s["tool_calls"] += 1
elif e["type"] == "run.completed":
s["status"], s["answer"] = "completed", data["answer"]
s["applied_through"] = e["seq"]
return s
# ---------------------------------------------------------------------------
# 3. The Lab 1.1 loop, instrumented: same logic, every step recorded
# ---------------------------------------------------------------------------
def run_agent_logged(model, tools: list[Tool], task: str,
store: EventStore, max_turns: int = 10) -> str:
toolbox = {t.name: t for t in tools}
messages: list[dict] = [{"role": "user", "content": task}]
store.append("run.created", {"task": task})
for turn in range(1, max_turns + 1):
reply = model.respond(messages)
store.append("model.responded",
{"stop_reason": reply.stop_reason, "text": reply.text,
"tool_calls": [vars(c) for c in reply.tool_calls]})
messages.append({"role": "assistant", "content": reply.text,
"tool_calls": [vars(c) for c in reply.tool_calls]})
if reply.stop_reason == "end_turn":
store.append("run.completed", {"answer": reply.text,
"turns": turn})
return reply.text
results = []
for call in reply.tool_calls:
store.append("tool.called", {"id": call.id, "name": call.name,
"args": call.args})
output = toolbox[call.name].fn(**call.args)
store.append("tool.returned", {"id": call.id, "name": call.name,
"output": output})
results.append({"tool_call_id": call.id, "content": output})
messages.append({"role": "tool", "results": results})
raise RuntimeError(f"gave up after {max_turns} turns")
if __name__ == "__main__":
store = EventStore("events.jsonl")
run_agent_logged(
model=ScriptedModel(INVESTIGATION_SCRIPT),
tools=TOOLS,
task="Checkout p99 latency spiked overnight. Find the likely cause.",
store=store,
)
events = store.events()
print(f"--- the ledger: {len(events)} events in events.jsonl ---")
for e in events:
summary = json.dumps(e["data"])
if len(summary) > 56:
summary = summary[:56] + "..."
print(f" {e['seq']:>2} {e['type']:<16} {summary}")
print("\n--- state, folded from the ledger (never stored) ---")
state = fold(events)
for key, value in state.items():
text = str(value)
print(f" {key:<15} {text if len(text) <= 60 else text[:60] + '...'}")
print("\n--- the checkpoint property ---")
snapshot = fold(events[:6]) # checkpoint: state through seq 6
resumed = fold(events[6:], snapshot) # resume folding from the snapshot
print(f" snapshot at seq 6, then fold the rest: equals full fold? "
f"{resumed == fold(events)}")
👉 Next: replay. The ledger records everything the run did, and the model interface lets us substitute what the model says. Point those two facts at each other and you get a time machine.