The turn is the atom of the runtime. Everything else — sessions, coordinators, stores, workflow
nodes — exists to support one thing: taking a prompt, driving the model loop to idle, and producing
a result. v1 and v2 run the same inner loop. The difference is what wraps it. v1's envelope is an
await: the turn is a promise, and if the process dies mid-turn, the promise disappears with it.
v2's envelope is a four-phase journal: the turn is a state machine, and if the process dies at any
phase, the next process knows exactly where it was.
This chapter follows the turn from its shared atom through both envelopes, showing where the code diverges and why.
Both v1 and v2 share a function called drive_run. Chapter 1 showed its extraction from Session as evidence that the correct turn shape was reachable; here we read it as the atom around which the two envelopes are built. The v1 header describes it as "the settle-and-compact engine extracted verbatim from the Session, so the root and a subagent share ONE path." The function operates on an injected RunLoop — a thin interface over pi's Agent — rather than constructing the loop itself.
Here is the actual implementation from apps/runtime-v1/src/core/run-agent.ts:
async def drive_run(loop: RunLoop, opts: DriveRunOptions) -> None:
await opts.start() # agent.prompt(text, images) or agent.resume()
await loop.wait_for_idle() # pi drives the model; tool calls happen here
await opts.checkpoint() # persist the new messages to the durable store
trailing = trailing_assistant(loop.state.messages)
if trailing and is_context_overflow(trailing, loop.state.model.context_window or 0):
# Reactive compaction: compact regardless of threshold, then continue.
outcome = await opts.compactor.compact(loop.state.messages, loop.state.model)
if outcome.summarized and not outcome.needs_new_session:
loop.state.messages = outcome.messages
await loop.resume()
await loop.wait_for_idle()
await opts.checkpoint()
return
# Proactive threshold compaction after a settled turn.
if (
trailing
and trailing.stop_reason != 'aborted'
and trailing.stop_reason != 'error'
and opts.compactor.should_compact(loop.state.messages, loop.state.model)
):
outcome = await opts.compactor.compact(loop.state.messages, loop.state.model)
if outcome.summarized and not outcome.needs_new_session:
loop.state.messages = outcome.messages
# proactive compaction is best-effort — not re-checkpointed;
# on restart the settled turn re-compacts idempotently.
if opts.on_compacted is not None:
opts.on_compacted(len(outcome.messages))Three phases, always in this order: start, idle, checkpoint. Then two optional compaction paths. The first is reactive: if the model stopped because the context window overflowed, compact now and continue. The second is proactive: if the transcript has grown past a threshold, compact after the settled turn (without stopping the current result).
The RunLoop interface that drive_run accepts is a Pick of pi's Agent['state']:
# apps/runtime-v1/src/core/run-agent.ts
class RunLoop(Protocol):
@property
def state(self) -> AgentState: ... # Pick[Agent.state, 'messages' | 'model']
async def wait_for_idle(self) -> None: ...
async def resume(self) -> None: ...The interface is deliberately thin. drive_run does not know whether the loop is backed by a
persistent durable Session or a fresh ephemeral agent for a subagent run. It does not know whether
the checkpoint persists to Postgres or to an in-memory map. It drives whatever loop it is given, and
calls whatever checkpoint function it is given. The seam boundary is the injection point.
This is the atom. Everything below is the envelope.
In v1, a turn is driven by Session.run_turn. The Session tracks two pieces of in-process state
across turns: a boolean flag and a live task.
# apps/runtime-v1 Session (simplified from the source)
class Session:
def __init__(self) -> None:
self._turn_in_flight: bool = False
self._current_turn: asyncio.Task[None] | None = None
async def run_turn(self, submission: AgentSubmission) -> None:
self._turn_in_flight = True
self._current_turn = asyncio.create_task(self._execute_turn(submission))
try:
await self._current_turn
finally:
self._turn_in_flight = False
self._current_turn = None_turn_in_flight is a boolean. _current_turn is a task. The turn is alive as long as the task
is pending. If the process dies — OOM, restart, SIGKILL — the task dies with it. The store has
the checkpoint from the last completed opts.checkpoint() call, but the runtime has no record of
where in the turn the process was when it died.
When the runtime restarts, it faces a question: was the submission completed, or not? If
opts.checkpoint() ran, the messages are in the store. If the process died between
loop.wait_for_idle() and opts.checkpoint() — a window that can be several seconds on an expensive
tool call — the messages are gone. The submission appears pending in the store (it was never marked
complete), so a new process will try to run it again. It will start a fresh turn, not resume the
interrupted one.
This is safe: the agent redoes the work. It is not durable: the model call is repeated, tool calls that had side effects may fire twice, and the user may receive two responses. For short in-process turns with cheap model calls, the failure is rare and the cost of a retry is low. For turns that involve long tool executions or large model calls, the consequences are meaningful.
The _turn_in_flight flag also reveals a subtle constraint. v1's Session uses it to detect and
reject concurrent turn attempts on the same session. This is correct behavior — you should not
drive two turns concurrently on the same agent loop — but the enforcement is purely in-process. A
second process that holds a reference to the same session key does not see _turn_in_flight = True.
The in-process flag is not a distributed lock.
v2 wraps drive_run in a four-phase turn journal. The journal is a record in AgentExecutionStore
that tracks exactly where in the turn the session was when the process stopped. The phases are:
before_provider → the session has begun the turn; no model call yet
provider_started → the model stream has opened; a streamKey is recorded
tool_request_recorded → a tool call is in-flight; the request is persisted
committed → the turn is complete; messages are checkpointed
The state machine transitions:
[*] → before_provider: beginTurnJournal
before_provider → provider_started:
updateTurnJournalPhase (+streamKey)
provider_started → tool_request_recorded:
updateTurnJournalPhase (+toolRequest, checkpointLeafId)
provider_started → committed:
commitTurnJournal (text-only turn, no tool call)
tool_request_recorded → committed:
commitTurnJournal (+committedLeafId)
committed → [*]
When the model responds with text and no tool call, the journal skips tool_request_recorded and
commits directly; tool_request and checkpoint_leaf_id stay unset, which is how recovery
distinguishes a text turn from a tool turn.
Each transition is a write to the store. The journal entry carries identifiers that let the next
process reconstruct what happened. Here is the AgentTurnJournal type from
apps/runtime/src/durable/types.ts:
@dataclass(frozen=True)
class AgentTurnJournalBase:
submission_id: str
session_key: str
kind: Literal['dispatch', 'direct']
attempt_id: str
operation_id: str
turn_id: str
revision: int
created_at: float
@dataclass(frozen=True)
class BeforeProvider(AgentTurnJournalBase):
phase: Literal['before_provider'] = 'before_provider'
@dataclass(frozen=True)
class ProviderStarted(AgentTurnJournalBase):
phase: Literal['provider_started'] = 'provider_started'
stream_key: str = ''
@dataclass(frozen=True)
class ToolRequestRecorded(AgentTurnJournalBase):
phase: Literal['tool_request_recorded'] = 'tool_request_recorded'
tool_request: ToolRequest = field(default_factory=ToolRequest)
checkpoint_leaf_id: str = ''
@dataclass(frozen=True)
class Committed(AgentTurnJournalBase):
phase: Literal['committed'] = 'committed'
committed_leaf_id: str = ''
type AgentTurnJournal = BeforeProvider | ProviderStarted | ToolRequestRecorded | CommittedThe union is the invariant — pattern matching on phase narrows the type, so mypy only allows
accessing stream_key in the ProviderStarted branch.
One friction point in the Python translation: each variant defaults phase so the discriminator auto-sets, and Python's dataclass rule then requires every field declared after it to have a default too. That is why stream_key, tool_request, and checkpoint_leaf_id carry empty defaults — not because empty values are meaningful, but because a defaulted field followed by a required one is rejected. (The base's required fields are fine — it's defaulted-then-required that Python rejects.) In production code, a __post_init__ validator or a factory function that enforces required arguments at construction time would restore the real requiredness.
The revision field is a generation counter. When the Coordinator recovers and tries a new
attempt on the same submission, it calls replace_turn_journal_attempt, which re-points the slot to a
new attempt_id while incrementing revision. A write from a stale attempt (one holding the old
attempt_id) returns False — the journal is single-writer per attempt. This is how v2 prevents
two processes from simultaneously driving the same turn: the journal slot enforces it in the store,
not in process memory.
The Coordinator is the component that drives submissions to settlement. It is v2's answer to v1's
_current_turn task — but where the task is ephemeral (in-process only), the Coordinator's
state is durable (in the store, readable by any process).
When a turn arrives via Session.deliver(prompt) for a live session, or via
coordinator.resume_session for a recovered one, the Coordinator creates or resumes a submission and
drives it through the journal phases:
Coordinator.runAttempt(submission):
1. beginTurnJournal(submissionId, 'before_provider')
2. open model stream
→ updateTurnJournalPhase('provider_started', { streamKey })
3. tool call arrives
→ updateTurnJournalPhase('tool_request_recorded', { toolRequest, checkpointLeafId })
4. driveRun completes (waitForIdle + checkpoint)
→ commitTurnJournal({ committedLeafId })
5. mark submission settled
If the process dies between steps 2 and 3, the next process finds phase provider_started in the
journal. The streamKey and markStreamConsumed primitives exist in the store for future
stream-segment reconstruction, but the current recovery path re-drives the turn rather than
replaying captured chunks. If it dies between 3 and 4, it finds tool_request_recorded and knows
which tool call was in-flight — it can inject the "tool execution was interrupted" marker and
continue from there.
The journal transforms a question ("did this turn complete?") into a precise answer ("it reached phase X"). The Coordinator does not retry blindly; it repairs from the known state.
The repair paths are where the journal earns its complexity. Consider the provider_started phase:
the model stream was open, and then the process died.
On restart, the Coordinator reads the journal, finds provider_started, and re-drives the turn.
The streamKey and markStreamConsumed store primitives exist to support future stream-segment
reconstruction (read back captured chunks, inject a partial assistant message, skip the re-query),
but the current recovery path does not yet implement reconstruction — it re-queries the provider.
The journal's value at this phase is recording that the provider was reached, so the recovery
path can make an informed decision rather than guessing.
v1 crash during model stream:
transcript: [user: prompt] (no assistant message recorded)
recovery: retry the full model call (no record of whether provider was reached)
v2 crash during model stream:
journal: phase = provider_started, streamKey recorded
recovery: currently re-drives the turn (stream reconstruction planned but not yet wired)
For tool calls, the repair is different. The tool call is in-flight — an external system may have
executed it, or may not have. The Coordinator does not know which. It injects an
INTERRUPTED_TOOL_MESSAGE as the tool result and lets the model decide what to do: retry the tool,
assume it succeeded, or give up. The runtime does not pretend the tool succeeded, and it does not
pretend it failed. It surfaces the interruption honestly as information the model can act on.
The tool_request_recorded phase carries the full tool_request — the tool name, the parameters,
the call ID. This means the Coordinator can log exactly which tool was interrupted, even if the tool
result never arrived. A future version can use this to idempotency-check retries against external
systems.
The separation between drive_run and the journal is intentional. drive_run does not know about
journals. It takes a RunLoop and options; it calls start, wait_for_idle, checkpoint. The
journal is constructed around it by Session.run_attempt, which wraps each drive_run call with
journal phase transitions.
v1 turn structure:
Session.run_turn(submission)
└─ drive_run(loop, { start, checkpoint, compactor })
↑ task-level durability only
v2 turn structure:
Coordinator.run_attempt(submission)
├─ begin_turn_journal('before_provider')
├─ update_turn_journal_phase('provider_started')
├─ drive_run(loop, { start, checkpoint, compactor })
│ ↑ identical inner loop
└─ commit_turn_journal('committed')
The inner loop is identical. The outer structure is the rewrite. This separation has a practical
consequence for testing: drive_run can be tested with a fake RunLoop and a no-op checkpoint
function, with no journal infrastructure. The journal machinery can be tested separately with a fake
session and a scripted model stream. The two concerns evolve independently.
It also means that compaction strategy changes affect drive_run but not the journal. If a future
version changes the proactive threshold check, or adds a new compaction mode, the journal phase
transitions do not change. The turn's boundary conditions (when does the Coordinator consider a turn
to have started? to have committed?) are stable even when the inner behavior changes.
One more thing v2 adds outside drive_run: the AgentSubmission. In v1, a turn is started by
calling Session.prompt directly. In v2, a turn is represented as a durable submission — a record
in the store that exists before the turn begins:
@dataclass(frozen=True)
class AgentSubmission:
submission_id: str # stable handle for the unit of work
session_key: str # which session this belongs to
kind: Literal['direct', 'dispatch']
status: Literal['queued', 'running', 'settled']
input: DirectAgentPayload | DispatchInput
lease_owner: str # the process id that currently holds the work
lease_expiry: float # when the lease times out (another process can claim)
retry_budget: int # how many more attempts are allowedThe lease mechanism is the distributed lock v1's _turn_in_flight boolean was not. If a process
claims a submission (status = 'running', lease_owner = process_id) and then dies, the lease
expires. Another process running list_runnable_submissions() finds the submission — its lease is
expired and its status is still running — and claims it. The journal tells the new process exactly
where to resume.
The pair of submission + journal gives end-to-end durability:
submission record created → prompt is not dropped (even if process dies before turn starts)
journal phase written → turn progress is not lost (even if process dies mid-turn)
journal committed → turn completion is not lost (even if process dies before settled)
submission marked settled → work is done; no future process will retry it
No prompt is silently dropped. No turn is silently abandoned. Every unit of work has a record, and every record has a state. This is what v2 adds over v1: not a different model call, not a different compaction algorithm, not a different session format — but the guarantee that the infrastructure around the model call is recoverable at every boundary.
The contracts from Chapter 3 named the seams: AgentContext carries session, sandbox, chat,
model, store. The construction root from Chapter 4 assembled those seams into a running system,
with SessionFactory as the single point where a component graph becomes a session. This chapter
followed a single turn through that system.
At the top level: a prompt arrives, the Coordinator creates an AgentSubmission, and writes the
first journal entry. At the middle level: Session.run_attempt wraps drive_run with journal phase
transitions, updating the store at each boundary. At the bottom level: drive_run calls
opts.start(), loop.wait_for_idle(), and opts.checkpoint() — the same three steps in both v1 and
v2, operating on a RunLoop that is the pi model loop, regardless of what surrounds it.
Chapters 3–5: The Primitive
Ch 3: Contracts Ch 4: Construction Ch 5: The Turn
┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐
│ AgentContext │ │ SessionFactory │ │ drive_run │
│ ModelProvider │──via──▶│ (single seam) │──into─▶│ (shared loop) │
│ SandboxApi │ │ │ │ │
│ AgentExecStore │ │ Wires contracts │ │ v1: await │
│ EventContract │ │ into live edges │ │ v2: journal + │
└──────────────────┘ └──────────────────┘ │ resume │
└──────────────────┘
"What seams exist" "Who wires them" "How they execute"
The rewrite did not change what the model does. It changed what the system knows about what the model did.
You now understand the turn from atom to envelope: drive_run is the shared inner loop that both
v1 and v2 run identically, and the four-phase journal is what v2 wraps around it to make turns
survive process death. The primitive did not change. The boundary conditions did.
Part II builds on this foundation. Chapter 6 details how each of the four journal phases maps to a
precise point of no return — why before_provider, provider_started, tool_request_recorded, and
committed are the right decomposition and not three phases or five. Chapter 7 shows
classify_submission_state: a pure function that reads the journal and returns one of eight recovery
states. The journal makes "did this turn complete?" answerable; the classifier makes the answer
actionable without retrying work that already happened.