Metadata-Version: 2.4
Name: dex-agents
Version: 0.2.0
Summary: Durable agent sessions for Dex, shared by the Dex agent framework adapters
Author: Super Durable
License-Expression: LicenseRef-Sustainable-Use-1.0
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Typing :: Typed
Requires-Dist: dex-python-sdk==1.35.0
Requires-Dist: litellm>=1.104,<1.106 ; extra == 'litellm'
Requires-Dist: httpx2>=2.10,<3 ; extra == 'mcp'
Requires-Dist: mcp>=2.3,<3 ; extra == 'mcp'
Requires-Dist: opentelemetry-api>=1.20,<2 ; extra == 'otel'
Requires-Python: >=3.11
Project-URL: Homepage, https://github.com/superdurable/dex/tree/main/agents/python/core
Project-URL: Repository, https://github.com/superdurable/dex
Project-URL: Changelog, https://github.com/superdurable/dex/blob/main/agents/CHANGELOG.md
Provides-Extra: litellm
Provides-Extra: mcp
Provides-Extra: otel
Description-Content-Type: text/markdown

# dex-agents

`dex-agents` (import `dex_agents`) is the Python core of the Dex agent
framework adapters. It provides one durable agent session Flow,
`AgentSessionFlow`, and the `AgentFrameworkRunner` contract that each
framework adapter implements. The plan is
[agent-framework-adapters.md](https://github.com/superdurable/dex/blob/main/docs/design/plan/agent-framework-adapters.md).

One Flow execution is one agent session; the Flow ID is the session ID. The
session runs in **Step mode**: every model call is one `RunModel` Step, and
every tool call, READ included, is its own Step. The tool call Steps of one
model response run in parallel, and `ResumeTurn` waits for their results
before the next model call. Each call therefore has its own retry policy,
timeout, failure route, history entry, and Dex Web node, and a crash repeats
at most the one call that was running.

## Installation

Applications usually install an adapter, which brings in this package. To
build a runner of your own, install the core directly. It needs Python 3.11 or
later and requires the released `dex-python-sdk==1.35.0`:

```bash
pip install dex-agents
```

`dex_agents.testing` ships in the same wheel, so an adapter's integration
tests can start a private engine and Worker processes. So does
`dex_agents.ui_stream`, the [browser streams](#browser-streams).
`dex_agents.mcp` offers the tools of MCP servers as agent tools; install it
with `pip install "dex-agents[mcp]"`, which adds the MCP SDK (`mcp>=2.3,<3`).
See [MCP tools](#mcp-tools-dex_agentsmcp).

For OpenTelemetry spans, install the `otel` extra, which adds only
`opentelemetry-api`; see [Tracing](#tracing).

The package has these modules. Optional dependencies are extras, so no
module needs a new package name:

| Module | Install | Contents |
| --- | --- | --- |
| `dex_agents` | `dex-agents` | The session Flow, the runner contract, built-in tools, `AgentSessionClient`, and tracing options |
| `dex_agents.stepwise` | `dex-agents` | Stepwise mode: `AgentDefinition`, function tools, tool sources, the `ModelClient` protocol, `ScriptedModelClient`, and the runner |
| `dex_agents.litellm` | `dex-agents[litellm]` | `LiteLLMModelClient`, a `ModelClient` on LiteLLM |
| `dex_agents.mcp` | `dex-agents[mcp]` | `McpToolSource`, MCP servers' tools as agent tools |
| `dex_agents.ui_stream` | `dex-agents` | Server-Sent Event frames and snapshot JSON for UIs |
| `dex_agents.testing` | `dex-agents` | Private engines, Worker processes, gates, Step history, and `ScriptedModelClient` (the same class as `dex_agents.stepwise`'s) |

Spans need the `otel` extra. Importing `dex_agents.litellm` without its extra
raises `ImportError` that names `dex-agents[litellm]`.

Develop the package inside this repository with uv 0.12.2. The adapters in
this repository resolve it from `../core` by path through
`[tool.uv.sources]`; their published wheels require it from PyPI instead:

```bash
cd agents/python/core
uv sync --locked
DEXCLI=/path/to/dexcli ./ci.sh
```

An adapter requires `dex-agents==<same version>`. All agents packages share
one version line; see the
[compatibility table](https://github.com/superdurable/dex/blob/main/agents/README.md#versions-and-compatibility).

## Running an agent

```python
from dex import AsyncClient, AsyncWorker, BlobCacheConfig, ClientOptions, Registry
from dex import WorkerOptions, open_blob_cache
from dex_agents import AgentSessionClient, AgentSessionFlow, AgentSessionOptions
from dex_agents import ToolEffect, ToolPolicy

flow = AgentSessionFlow(
    MyFrameworkRunner(agent),
    flow_type="TriageAgent",
    options=AgentSessionOptions(
        tool_policies={"search_tickets": ToolPolicy(ToolEffect.READ)},
    ),
)
registry = Registry([flow], allow_async_handlers=True)
cache = open_blob_cache(BlobCacheConfig(".dex-cache", 64 * 1024 * 1024))
worker = AsyncWorker(registry, cache, WorkerOptions(server_address="localhost:8801"))
client = AsyncClient(registry, cache, ClientOptions(server_address="localhost:8801"))

sessions = AgentSessionClient(client, flow)
await sessions.start_session("session-42", "Find my open tickets")
snapshot = await sessions.get_agent_snapshot("session-42")
```

Register one `AgentSessionFlow` per agent, each with its own `flow_type`. All
Steps are `async`, so the Registry needs `allow_async_handlers=True` and an
`AsyncWorker` serves it. Tool calls run as parallel Steps, so give the Worker
enough task slots (`PullWorkerOptions.max_concurrent_tasks`) for the largest
batch of tool calls a model makes. A session never completes by itself; it
waits for the next user message after every turn. Stop it with
`AsyncClient.stop_flow`.

## Stepwise mode

Stepwise mode is the built-in agent loop, for agents defined in Dex terms
without a framework. `StepwiseAgentRunner` implements the runner contract
below, so the session Flow, its Steps, approvals, and events are the ones
every adapter uses. Each `RunModel` makes exactly one model call through a
`ModelClient`, and every tool call is its own Step.

```python
from dex import Registry
from dex_agents import BuiltInTool, ToolEffect
from dex_agents.litellm import LiteLLMModelClient
from dex_agents.stepwise import AgentDefinition, ToolCallContext, function_tool
from dex_agents.stepwise import stepwise_agent_session_flow


@function_tool(effect=ToolEffect.READ)
async def lookup_order(order_id: str) -> dict[str, str]:
    """Look up an order by its ID."""
    return await orders.get(order_id)


@function_tool(effect=ToolEffect.NON_IDEMPOTENT_WRITE)
async def refund_order(order_id: str, amount_cents: int, context: ToolCallContext) -> str:
    """Refund part of an order."""
    return await payments.refund(order_id, amount_cents, idempotency_key=context.idempotency_key)


support_agent = AgentDefinition(
    name="support",
    instructions="You help customers with their orders.",
    model=LiteLLMModelClient("openai/gpt-5-mini"),
    tools=[lookup_order, refund_order],
    built_in_tools=frozenset({BuiltInTool.ASK_USER, BuiltInTool.DURABLE_WAIT}),
)
flow = stepwise_agent_session_flow(support_agent, flow_type="SupportAgent")
registry = Registry([flow], allow_async_handlers=True)
```

`AgentDefinition` fields: `name`, `instructions` (a string, or a function of
`InstructionContext` evaluated in every `RunModel`), `model`, `model_settings`,
`tools`, `tool_sources` (see [Tool sources](#tool-sources)), `tool_policies`,
`default_tool_effect`, `built_in_tools`,
`max_model_calls_per_turn` (100; after that many calls in one turn the next
call offers no tools, and a tool call the model makes anyway gets the result
`The tool call was not run: this turn reached its limit of <n> model calls.`
while the turn ends), `interrupt_deadline`, `event_stream_capacity_bytes`, and
`plan_mode_enabled` (see [Plans and plan mode](#plans-and-plan-mode)).
A function tool may not share a name with an enabled built-in tool:
`AgentDefinition` refuses it, and so does `run_model_step`, before any model
request, when only the session options enable that built-in.
`stepwise_agent_session_flow` derives the `AgentSessionOptions` from it.

`@function_tool(effect=..., requires_approval=None, name=None,
description=None, parameters_schema=None)` turns a plain sync or async
function into a tool. The name defaults to the function name, the description
to the docstring, and the JSON Schema is derived from the type hints with the
standard library: `str`, `int`, `float`, `bool`, `None`, `list`, `dict`,
`Literal`, `Enum`, unions, dataclasses, `TypedDict`, and
`Annotated[T, "description"]`. A parameter typed `ToolCallContext` receives the
idempotency key, attempt, Flow ID, turn ID, tool call ID, event writer,
heartbeat, and the answers to the tool's questions. A sync function runs in a
worker thread. Arguments that break the schema, and an unknown tool name,
raise `ToolExecutionError`, which the model sees: arguments that are not JSON
give `invalid arguments for <tool>: arguments are not valid JSON`. A `str` result is the
output; any other result is serialized as JSON.

In `RunModel`, the runner builds the model input from committed state only:
the instructions, every stored item, the plan note from `build_plan_note`
before the turn's items when a plan exists or the turn plans, and the results
of the previous model response from the final records. Every stored tool call is paired with a
result; a missing one becomes the outcome-unknown text, so an orphaned call
never reaches a provider. It stores items as they appear
(`ToolCallsRequested.new_messages`), so `FrameworkRunState` holds only the
turn's model call count.

### Tool sources

`AgentDefinition.tool_sources` takes tools that are listed at run time, such
as an MCP server's tools. Any object with a `source_name` and `list_tools`,
`deferred_tool_call`, `execute_deferred_tool_call`, and `poll_tool_task` is a
`ToolSource`; `McpToolSource` is one:

```python
from dex_agents.mcp import McpStdioServerConfig, McpToolSource

files = McpToolSource(
    [McpStdioServerConfig(name="files", command="mcp-server-files", args=["/srv/reports"])],
    resource_and_prompt_tools_enabled=True,
)
agent = AgentDefinition(
    name="reports",
    instructions="You manage reports.",
    model=LiteLLMModelClient("openai/gpt-5-mini"),
    tools=[lookup_order],
    tool_sources=[files],
    built_in_tools=frozenset({BuiltInTool.ASK_USER}),
)
```

In every `RunModel` of a chat or executing turn, the runner awaits each
source's `list_tools`, for at most `tool_source_listing_timeout` (10 seconds),
and offers the tools after the function tools and before the built-in tools.
A source whose listing raises or takes longer offers no tools in that model
call, with a warning log, so a down server never fails a turn. A planning
turn, and a call after the model call limit, list nothing.

A call of a listed tool becomes the source's `deferred_tool_call`, so the
source's policy, effect class, and `tool_source_reference` apply: approvals,
`CallTool` or `CallSingleAttemptTool`, the idempotency key, and tool tasks
work as for any tool. The runner passes the definition it offered as
`listed_tool`, so a listing refreshed during the model call cannot change it.
It stores the source's reference as `{"stepwise_tool_source": <source_name>,
"reference": ...}`, so the tool Step and `PollToolTask` reach the source that
built the call, with its own reference, on any Worker, without a listing, and
whatever the order of `tool_sources`. Source names must be unique and stable
across deploys: `McpToolSource` defaults its `source_name` to `"mcp"`, which
adding or removing a server keeps, and takes `source_name=`. Two
`McpToolSource`s in one agent must therefore name at least one of them. A stored call whose source name this Worker does not have raises
`LookupError`, so its effect class decides the outcome. A name that no listing
offered, such as a tool of a server that was down, fails as `unknown tool
<name>` without an approval (`unclassified`, no approval): it never reaches a
source.

A listed name that repeats a function tool, an enabled built-in tool, or
another source's tool fails the `RunModel` attempt with `ValueError` before any
model request: `tool <name> uses the name of an enabled built-in tool`, `tool
<name> of a tool source uses the name of a function tool`, or `two tool sources
list a tool named <name>`. A source may define `declared_tool_names()`, names
it always lists, and `AgentDefinition` checks those when it is built; two
`McpToolSource`s with the resource and prompt tools are refused there. A
source may also define `cached_tools()`, its last listing without I/O, which
the synchronous `estimate_input_tokens` counts; otherwise it counts the
runner's last listing of that source.

### Model clients

```python
class ModelClient(Protocol):
    async def call_model(self, request: ModelRequest, output: ModelOutputWriter) -> ModelResponse: ...
    def estimate_input_tokens(self, request: ModelRequest) -> int: ...
```

Every client makes one request per call with no retries of its own, sends the
whole input every time with no provider-side conversation state, and returns
each output item provider-exact. `ModelOutputItem.item` is the provider's JSON
text: opaque fields such as encrypted reasoning content and thinking
signatures are kept byte for byte and replayed unchanged to the same provider
and API. An item from another provider or API replays as its text and tool
calls, and its reasoning is dropped. User text and tool results are stored in
a provider-neutral form.

`LiteLLMModelClient(model, api=None, api_base=None, api_key_provider=None,
request_options=None)` uses the Responses API for OpenAI models, with
`store=False` and `include=["reasoning.encrypted_content"]`, and streaming
Chat Completions for every other provider, keeping Anthropic `thinking_blocks`
with their `signature`. It streams text and reasoning summaries to
`text_delta` and `reasoning_delta` events and reports usage. Credentials come
from the Worker's environment, as LiteLLM reads them, or from
`api_key_provider`; never from Flow input or Attributes.

`ScriptedModelClient(respond)` from `dex_agents.stepwise` answers each call
from a function of the `ModelRequest`, so an agent runs without a model key: a
local demo mode, a tutorial, or a test. `respond` returns a
`ScriptedModelReply` with `text`, `tool_calls` (`ScriptedToolCall`), and
`reasoning`, sync or async. `ScriptedModelClient.user_texts(request)` and
`ScriptedModelClient.tool_results(request)` read the request. `requests` keeps
the latest `max_recorded_requests` requests, 100 by default. Keep the script
deterministic, because a crashed `RunModel` reruns it. `dex_agents.testing`
exports the same class.

```python
from dex_agents.stepwise import ModelRequest, ScriptedModelClient, ScriptedModelReply


def reply_as_demo_model(request: ModelRequest) -> ScriptedModelReply:
    results = ScriptedModelClient.tool_results(request)
    if results:
        return ScriptedModelReply(text="; ".join(result.output for result in results))
    return ScriptedModelReply(text=f"Demo reply: {ScriptedModelClient.user_texts(request)[-1]}")


model = ScriptedModelClient(reply_as_demo_model) if demo else LiteLLMModelClient(model_name)
```

## Built-in tools: questions and durable waits

Built-in tools are session features, so they work under every runner. Enable
them in `AgentSessionOptions.built_in_tools`. The runner offers the enabled
definitions from `BUILT_IN_TOOL_DEFINITIONS` (filter by
`turn.enabled_built_in_tools`) to its model, and `RunModel` routes their calls
to their own Steps with no approval and the `read` effect. `write_todos` is
described in [Plans and plan mode](#plans-and-plan-mode). An enabled name is
reserved; a disabled one is an ordinary tool name. Invalid arguments record
`failed` with `invalid <tool> arguments: <reason>` at once. `null` for an
optional argument means the argument is absent, because providers in
strict-schema mode send it.

- `ask_user` `{question, choices?, allows_free_text?}` raises an interrupt of
  kind `question` with the tool call ID as its interrupt ID and waits in
  `AwaitQuestionAnswer`. `answer_question(interrupt_id, answer)` answers it;
  the model sees `{"status":"answered","answer":"<text>"}` as the tool result
  and continues the same turn. A second answer, a blank answer, or an answer
  outside `choices` when free text is not allowed returns `False`.
  `resolve_interrupt(..., approved=False)` declines it (`The user declined to
  answer.`), and `interrupt_deadline` applies (`question deadline passed`).
- `durable_wait` `{duration_seconds, reason}` records `waiting_for_timer` with
  `wait_until` and waits in `AwaitTimer` on a Timer. No model call, coroutine,
  or Worker holds the wait. The model then sees
  `{"status":"completed","waited_seconds":<n>,"reason":"<reason>"}`. Durations
  are whole seconds from 1 to `max_durable_wait` (365 days by default).

A tool can ask a question too: `execute_deferred_tool_call` returns
`ToolInputRequested(question, choices, allows_free_text, continuation)` instead
of a result, only when the call has taken no effect yet. The tool Step stores
the question on its record and moves to `AwaitQuestionAnswer`, whose Execute
raises the `question` interrupt `<tool_call_id>-input-<n>` under the
`PendingInterrupts` lock, so parallel tool calls can ask at once. If another
call of the turn already uses that ID, for its record or a pending interrupt,
the asking call records `failed` instead. After the
answer, the session runs the same tool Step type again with `DeferredToolExecution.input_answers` and
`tool_call.input_continuation`. A function tool returns it from the function
and reads the answers from `ToolCallContext`.

## Plans and plan mode

The `write_todos` built-in replaces the session's plan, the `AgentPlan`
Attribute, as a whole. Its arguments are `{todos: [{content, status}]}`, with
a status of `pending`, `in_progress`, or `completed`. `RunModel` records the
call in its own commit, publishes its result, and still moves to `ResumeTurn`;
no tool Step runs. Each call that succeeds writes revision + 1: no tasks is
`cleared`, a planning turn writes a `draft`, every task completed is
`completed`, and anything else is `active`. The model sees
`{"status":"updated","revision":<n>,"task_count":<k>}` or
`{"status":"cleared","revision":<n>}`, and the session writes `plan_updated`,
which it also writes when an approval marks a draft `active`.
A response with more than one `write_todos` call fails each of them with
`write_todos was called more than once in one response`; invalid arguments fail
with `invalid todos: <reason>`.

Plan mode reviews a plan before it runs. Enable it with
`AgentSessionOptions(plan_mode_enabled=True)`, which needs `write_todos` and a
runner that is a `PlanModeRunner` (`supports_plan_mode` is `True`). The
Stepwise runner is one; `AgentDefinition(plan_mode_enabled=True)` sets the
option. `AgentDefinition.planning_tool_choice` is `"auto"` by default: a
planning turn offers only `write_todos`, the plan note tells the model to write
the plan with it, and a turn that answers in text, such as with a clarifying
question, ends without a draft. `"required"` forces `write_todos` on the first
model call of a planning turn while no draft exists. Claude Opus 5.5, Claude
Sonnet 5.5, and Anthropic extended thinking refuse a forced tool choice, so
`AgentDefinition` raises `ValueError` for `"required"` when the model client's
`supports_forced_tool_choice(settings)` returns `False`, as
`LiteLLMModelClient`'s does for them. For a model LiteLLM routes to Anthropic,
or to Claude on Bedrock or Vertex AI, it counts a `thinking` option that is not
disabled, or a `reasoning_effort` other than `"none"`, as extended thinking.
Gemini and OpenAI models accept a forced tool choice with thinking or reasoning.

```python
sessions = AgentSessionClient(client, flow)
await sessions.send_user_message("session-42", "Plan the migration", plan_mode=True)
snapshot = await sessions.get_agent_snapshot("session-42")
if snapshot.plan is not None and snapshot.plan.status is PlanStatus.DRAFT:
    await sessions.execute_plan("session-42", snapshot.plan.revision)
```

- A message with `plan_mode`, or any message while a `draft` plan exists,
  starts a turn whose `AgentState.interaction_mode` is `planning`. The runner
  reads it from `AgentTurnContext.interaction_mode` and offers only
  `write_todos`.
  `RunModel` rejects any other call of a planning turn with `The tool call was
  not run: a planning turn can only call write_todos.`. Feedback on a draft is
  an ordinary message that revises it.
- `send_user_message` returns `False` for a `plan_mode` message when plan mode
  is off.
- `execute_plan(ExecutePlanRequest(revision))` returns `True` only while the
  session waits for a user message, the plan has that revision and is `draft`
  or `active`, no execution of it is pending, and nothing is queued, steered, or pending.
  It records the approval in `PendingPlanExecution` and publishes it to the
  `PlanExecutions` ChannelMap instance of that revision. It locks
  `PendingPlanExecution`, so a concurrent call gets `RpcLockConflictError`
  (`AgentSessionClient` retries it) and then `False`.
- `AwaitUserMessage` waits on the approval of the current revision only, after
  queued messages, so an approval of a revision that a later turn replaced
  never runs. It marks the plan `active`, writes `plan_execution_started`, and
  starts an `executing` turn with every tool, whose user message is `[Dex] The
  user approved plan revision <n>. Execute it and keep the task statuses
  current with write_todos.`.
- `build_plan_note(plan, interaction_mode)` returns the note that shows the
  model the plan. Stepwise adds it as user text before the turn's items in
  every model call, so the model always sees the latest plan.

## Steering and turn cancellation

A user message is queued or steered. `send_user_message` queues it: it waits
until the running turn ends, and the user can still delete or steer it.
`steer_user_message` steers the running turn with new text or with a queued
message, and `cancel_turn` stops it with no new turn.

```python
sessions = AgentSessionClient(client, flow)
await sessions.send_user_message("session-42", "Also check order 8")
snapshot = await sessions.get_agent_snapshot("session-42")
queued = snapshot.queued_messages[0]
await sessions.steer_user_message("session-42", message_id=queued.message_id)
await sessions.steer_user_message("session-42", text="Use the staging account instead")
await sessions.delete_queued_message("session-42", snapshot.queued_messages[1].message_id)
await sessions.cancel_turn("session-42", snapshot.state.active_turn_id)
```

A steered turn ends at its next safe Step boundary. Steering never interrupts
a model call or a tool call that has started: a running model call finishes,
and a final answer it returns completes the turn as usual; a tool that has
started runs to the end, and so does a tool task it started, questions
included. `cancel_turn` stops the turn now instead: it cancels a running model
call or compaction, whose answer is dropped, and every wait for an approval, an
answer, a durable wait's Timer, or a tool task's next poll. A tool call Step
that has started still runs to the end, and the turn ends after it. A started
tool task is abandoned: its call records `outcome_unknown` with
`cancellation_reason` `turn_cancelled`, and a runner that implements
`ToolTaskCanceller` is asked to cancel the task, best-effort; a result the
task returns later is dropped. A call that has not started records
`rejected` with `cancellation_reason` `steered` or `turn_cancelled`, and the
model later sees
`{"status":"cancelled","reason":"superseded_by_steered_user_message"}` or
`{"status":"cancelled","reason":"turn_cancelled_by_user"}` as its result. That
includes approvals, questions, and durable waits that were still waiting. A
call that is settled first keeps its own result: an `ask_user` answer that
commits first, a durable wait whose Timer fires first, and a built-in call with
invalid arguments, which fails at once. A UI can briefly show an approval or a
question of a turn that is already steered; answering it then returns `False`
or has no effect. `ApplyTurnControl` then ends the turn. When the runner
implements `build_unfinished_turn_messages`, it stores the turn's exact
transcript; otherwise it stores a note that lists the calls that ran, as after
a failed turn. A steered turn is followed by a new turn whose user messages are
every steered message, in order. That turn is a `planning` turn when any of
its messages has `plan_mode` or a `draft` plan exists, as for a queued message;
a queued `plan_mode` message moved by its ID keeps `plan_mode`. A `write_todos`
call settles in `RunModel`'s own commit, like a built-in call with invalid
arguments, so one that a model call made while the user steered still replaces
the plan.

While no turn is active, there is nothing to steer: `steer_user_message` with
`text` queues the message like `send_user_message`, so it starts the next turn
at once.

- `steer_user_message` and `delete_queued_message` take the Channel message ID
  from `AgentSnapshot.queued_messages`. Both return `False` for a message that
  is no longer queued. When the session consumes the message while the RPC
  runs, the RPC raises `ChannelMessageNotFoundError` and changes nothing;
  reload the snapshot and let the user choose again.
- `cancel_turn` returns `True` when the request is recorded. The turn may still
  complete first, when its model call committed the final answer before the
  RPC committed. A second call for the same turn returns `True` and changes
  nothing.
- `steer_user_message` and `delete_queued_message` lock nothing. `cancel_turn`
  locks `TurnCancellationRequest` and `PendingInterrupts`, which only short
  Steps lock, so concurrent cancels apply in order and none of them waits for
  a model call; `AgentSessionClient.cancel_turn` retries the rare
  `RpcLockConflictError`.
- `resolve_interrupt` or `answer_question` racing a steer either commits
  first or returns `False`. An approval or a tool's answer that commits first
  moves its call on, and the tool Step then records it cancelled before the
  tool starts; an `ask_user` answer that commits first is the call's result.
- `AgentSnapshot.steered_messages` lists the steered messages not applied yet.
- A turn that a Worker without steering started or advanced cannot be steered
  or cancelled: `AgentState.accepts_turn_control` is `False`, and both RPCs
  return `False`. It runs to its end as before. Call the RPCs only after every
  Worker of the Flow type runs this release and every turn that was active
  during the rollout has ended, because such a turn can hold a wait that an
  earlier Worker began, which a steer never wakes.

## Message storage and retention

> **Upgrade rule: stop every 0.1.0 Worker of a Flow type before any 0.2.0
> Worker of it starts.** Once a 0.2.0 Worker stores a message in a 0.1.0
> session, it moves the older messages out of `AgentMessages`. A 0.1.0 Worker
> that then runs a Step of the session fails the turn, because it cannot read
> the moved messages.

The newest messages are the window: one `AgentMessages` instance per
message, so storing a message writes only that message. When the window holds
more than `message_window_size` messages (64), its oldest messages move into
immutable `AgentMessageChunks` instances in the same commit. A chunk holds at
most `message_chunk_size` messages (32) and `message_chunk_max_bytes` of
encoded JSON (1 MiB, at most 8 MiB), so it stays well under the Server's
`values.maxValueBytes`. A larger message skips the window and goes straight
into its own chunk in the Step that stores it, so a message too large for the
Server fails that Step's turn at once instead of a later rollover. One commit
moves at most 4 chunks and 8 MiB, so the window can briefly hold more than its
size. The `AgentMessageIndex` Attribute lists each kept chunk's first and last
sequence, where the window starts, and the oldest stored message; it changes
only when chunks are written or deleted. `TurnReplySequence` records where the
latest turn's replies start. Every Step that stores messages loads the whole window.
With compaction, `RunModel` loads only the chunks after the summary, through
per-movement load options. Without compaction, the model input is every
message, so it loads every chunk.

With compaction, retention keeps Flow state bounded however long the session
runs. It deletes the oldest chunk while every message in it is at or before
`ContextSummary.last_sequence` and at least `retained_message_limit` messages
(1,000) are stored after it, so the newest `retained_message_limit` messages
and every message after the summary always stay. A deleted message is gone
from `list_agent_messages` and `get_agent_snapshot`; paging ends at
`AgentMessageIndex.first_sequence`, and every page's
`deleted_before_sequence` says where the deleted messages end. An application
that needs the whole transcript copies new messages out with
`list_agent_messages` before `retained_message_limit` more are stored after
them, such as after each `turn_completed`, or sets
`retained_message_limit=None`. A runner without `ContextCompactionRunner` on
a session with deleted messages logs a warning on each model call, because
its input starts after them with no summary.

```python
from dataclasses import replace

options = replace(
    stepwise_agent_session_options(definition),
    message_window_size=64,
    message_chunk_size=32,
    message_chunk_max_bytes=1024 * 1024,
    retained_message_limit=5_000,
)
```

A session that dex-agents 0.1.0 stored has every message in `AgentMessages`
and no index. This release reads it as one large window, and each Step that
stores messages moves at most 4 chunks, so the older messages leave the
window over several commits.

## Compaction

Compaction keeps the model input under the context window. Set
`AgentSessionOptions.compaction` (or `AgentDefinition.compaction` for
Stepwise) to a `CompactionPolicy`:

```python
from dex_agents import AgentSessionOptions, CompactionPolicy

options = AgentSessionOptions(
    compaction=CompactionPolicy(context_window_tokens=400_000),  # trigger 0.85, keep 0.3
)
```

Before each model call, `RunModel` asks the runner for
`estimate_input_tokens(turn)`. At `trigger_ratio` of the window it asks for
`select_compaction_cutoff(turn, retained_token_budget)`, the last message to
summarize; when that sequence is past the current summary, `RunModel` sets
status `compacting_context` and moves to `CompactContext` without a model
call. `CompactContext` has `RunModel`'s timeouts, retries, and failure route.
It calls `summarize_context(ContextSummaryRequest)` once, stores the
`ContextSummary` Attribute (generation, `first_sequence` 1, `last_sequence`,
text, `created_at`), records the call's usage, writes `compaction_started` and
`compaction_finished`, and moves back to `RunModel`. A crash reruns only the
summary call. From then on the runner receives the messages after the summary
in `turn.messages` and the summary in `turn.context_summary`. A second estimate
that is still high finds no new cutoff, so the pair cannot loop.

Summarized messages stay stored until retention deletes them; see
[Message storage and retention](#message-storage-and-retention). The runner must implement
`ContextCompactionRunner`, and `AgentSessionFlow` raises `TypeError` for a
compaction policy otherwise. `StepwiseAgentRunner` implements it: its input
starts with `CONTEXT_SUMMARY_NOTE_HEADER`
(`[Dex] Summary of the earlier conversation:`) and the summary, it cuts only
between model responses whose tool calls all have stored results, so a turn
with many tool calls compacts mid-turn without separating a call from its
result, and it summarizes with `CONTEXT_SUMMARY_INSTRUCTIONS` and no tools.
`ContextSummaryRequest.input_token_budget` carries the policy's
`context_window_tokens`. When the summary request plus the definition's
`max_output_tokens`, if set, would exceed it, the runner leaves out the oldest
messages, up to a cut that keeps each model response and each tool call with
its result whole, and says so with
`build_context_summary_omission_note(omitted_count)`
(`[Dex] <n> earlier messages did not fit in this request and are left out.`).
Each `CompactContext` attempt emits a `chat` span for the summary call, as
`RunModel` does; return `ContextSummaryResult.model_call_metadata` to name it.

## Session settings

Session settings let one Flow type serve sessions that differ, such as by
model, without a Worker pool per model. They are one JSON object that the
application sets when it starts the session and can replace later:

```python
models = {name: LiteLLMModelClient(name) for name in ("openai/gpt-5-mini", "openai/gpt-5")}
definition = AgentDefinition(
    name="support",
    instructions="You help customers with their orders.",
    model=models["openai/gpt-5-mini"],
    # Choose only from clients the Worker built; the settings come from the application.
    resolve_model=lambda settings: models.get(
        str(settings.get("model")), models["openai/gpt-5-mini"]
    ),
)

await sessions.start_session("session-42", "Hello", session_settings={"model": "openai/gpt-5"})
await sessions.update_session_settings("session-42", {"model": "openai/gpt-5-mini"})
```

- `StartAgentSession` stores `AgentSessionStartInput.session_settings` in the
  `SessionSettings` Attribute. `update_session_settings` replaces it; nothing
  is merged, and `{}` clears it. It returns `False` before the session
  initialized.
- A change takes effect when the next turn starts, the same boundary as
  `plan_mode`. The Step that starts a turn copies the settings into
  `TurnSessionSettings`, tagged with the turn, and the turn's model calls,
  retries, and compaction read only that copy, so a running turn keeps the
  settings it started with, even across a Worker crash. A copy that an earlier
  turn left reads as no settings.
- A runner reads `AgentTurnContext.session_settings`, an empty mapping when
  there are none, such as in a session started on 0.1.0.
  `ContextSummaryRequest.session_settings` carries the same value. Stepwise's
  `AgentDefinition.resolve_model` maps it to the `ModelClient` of each model
  call, including summary calls and token estimates, and
  `InstructionContext.session_settings` passes it to an instructions function.
- `get_agent_snapshot` returns the next turn's settings as
  `AgentSnapshot.session_settings`, not the running turn's: right after an
  update, the running turn may still use the earlier settings.
- Only model calls see the settings. Tool calls do not receive them, and the
  compaction policy's `context_window_tokens` is one value for every resolved
  model.
- **Never put a secret, or anything a browser must not see, in settings.**
  They are Flow state that `get_agent_snapshot` returns, and
  `dex_agents.ui_stream`'s browser snapshot JSON includes them as stored. Put
  a reference, such as a model or credential name, and let the Worker resolve
  it. An exception that `resolve_model` raises becomes the turn's
  `last_turn_error`, which snapshots also show, so keep the settings out of
  its message.
- The compact JSON text, UTF-8 with non-ASCII unescaped, is at most
  `MAX_SESSION_SETTINGS_BYTES` (16 KiB). `validate_session_settings` and
  `session_settings_size` check it, and `AgentSessionClient` checks before it
  sends. The cores write some floats and U+2028/U+2029 differently, so the
  session's own checks accept `SESSION_SETTINGS_SIZE_TOLERANCE_BYTES` (1 KiB)
  more: `update_session_settings` returns `False` beyond that, or for `None`,
  and `StartAgentSession` fails the session at once, with no retry, with a
  reason that starts with `INVALID_START_SESSION_SETTINGS_REASON`.
- Start sessions with settings, and call `update_session_settings`, only after
  every Worker of the Flow type runs 0.2.0. A 0.1.0 Worker drops the start
  settings and starts turns without them.

## The runner contract

An adapter implements three methods. The session calls them from Steps, so
each call may run again after a crash or retry. Keep the runner stateless
between calls and derive everything from the arguments.

```python
class AgentFrameworkRunner(Protocol):
    def build_user_message(self, text: str) -> FrameworkMessage: ...
    async def run_model_step(self, turn: AgentTurnContext) -> AgentTurnOutcome: ...
    async def execute_deferred_tool_call(self, call: DeferredToolExecution) -> DeferredToolOutcome: ...

class UnfinishedTurnMessageBuilder(Protocol):  # optional
    def build_unfinished_turn_messages(self, turn: AgentTurnContext) -> Sequence[FrameworkMessage]: ...
```

`DeferredToolOutcome` is `ToolExecutionResult | ToolInputRequested`.
`AgentFrameworkRunner` is generic in what `execute_deferred_tool_call` returns,
and a bare `AgentFrameworkRunner` means `AgentFrameworkRunner[ToolExecutionResult]`,
as in 0.1.0. A runner whose tools ask the user implements
`AgentFrameworkRunner[DeferredToolOutcome]`.

`AgentTurnOutcome` is `TurnCompleted | ToolCallsRequested`.

### `build_user_message(text)`

Return `FrameworkMessage(role=MessageRole.USER, framework_item=..., text=text)`.
The session stores it as the turn's first message before the first
`RunModel`. `RecordTurnFailure` also calls it for the turn-failure note. Any
other role fails the Step.

### `run_model_step(turn)`

The framework makes **at most one model call** per `run_model_step`, and
stops before it runs any tool it can defer. `AgentTurnContext` carries
committed state:

| Field | Meaning |
| --- | --- |
| `flow_id`, `turn_id`, `attempt` | Session ID, `"turn-<n>"`, and the one-based `RunModel` attempt. |
| `messages` | Every stored `AgentMessage` in sequence order. The last user message belongs to this turn. |
| `framework_run_state` | `None` for the turn's first model call; otherwise the value from this turn's last `ToolCallsRequested`. `is_resuming` tests it. |
| `tool_calls` | Every `AgentToolCall` record of this turn by tool call ID. When `RunModel` runs, every record is final. `recorded_tool_call(id)` reads one. |
| `resolve_tool_policy(name, framework_effect=None)` | The policy for a tool: explicit policy, else framework metadata, else `default_tool_effect`. |
| `events` | `AgentEventWriter` bound to the turn, Step execution, and attempt: `write_text_delta`, `write_reasoning_delta`, `write_tool_call_started`, `write_tool_call_finished`, `write_event`. Safe to call from any thread; writes after the call returns are dropped. |
| `heartbeat()` | Records liveness. The session already heartbeats every 20 seconds in the background. |
| `enabled_built_in_tools` | The built-in tools to offer to the model, from `AgentSessionOptions.built_in_tools`. |

A runner must:

1. Without run state, rebuild framework input from each message's
   `framework_item`. With run state, restore the framework from it and give
   every tool call the result from its final record instead of running the
   tool. `describe_tool_call_outcome(record)` renders the model-visible text.
   A framework with a native rejection API may pass `record.rejection_message`
   there instead.
2. Let the framework make one model call.
3. If the response asks for tools, return `ToolCallsRequested(framework_run_state,
   tool_calls, usage)` with one `DeferredToolCall(tool_call_id, tool_name,
   arguments, policy)` per call, in model order. Get each `policy` from
   `turn.resolve_tool_policy(name, framework_effect)`; pass `framework_effect`
   when the framework exposes metadata, such as MCP `readOnlyHint` (`READ`) or
   `idempotentHint` (`IDEMPOTENT_WRITE`). `framework_run_state` must be
   non-empty and must continue the turn after the tools. Tool call IDs must be
   non-empty, contain no `/`, and be unique within one turn; IDs may repeat in
   later turns, because the idempotency key includes the turn ID.
4. If the response is the final answer, return `TurnCompleted(new_messages,
   usage)`. `new_messages` holds the turn's items the session has not stored
   yet, in order: model output, tool requests, tool results, and the final
   answer. The session appends them in one commit. A runner may instead store
   items as they appear with `ToolCallsRequested.new_messages`, which the
   session appends in the commit that records the tool calls; such a runner
   implements `build_unfinished_turn_messages`, which `RecordTurnFailure`
   calls to store the results of the turn's tool calls before its note. If it
   raises or returns anything but `FrameworkMessage` values, the session logs a
   warning and stores only the note. A runner that sets `new_messages` without
   the method fails its turn after that one model call, with no retry.
5. Report `usage` for this model call only. The session adds it to the session
   total. Optionally return `model_call_metadata=ModelCallMetadata(
   request_model, response_model, provider_name, response_id, finish_reasons)`
   in either outcome. The session never stores it; it names the `chat` span
   (see [Tracing](#tracing)). Leave a field `None` when the framework does not
   expose it.
6. Never run a tool inside `run_model_step`, unless the framework cannot defer
   it. Document such tools as at-least-once: a `RunModel` retry runs them
   again.

Any exception fails the attempt. `RunModel` retries up to three attempts
within 30 minutes, rerunning only this one model call, then routes to
`RecordTurnFailure`. Each attempt has a 15 minute timeout.

When a failed turn already has final tool call records, `RecordTurnFailure`
stores one note before it deletes them, built with `build_user_message`, so
the next turn's model does not repeat a write. `build_turn_failure_note`
renders it, one line per record in `deferral_order`:

```text
[Dex] The previous turn failed after these tool calls ran:
- <tool_name> <arguments> -> <status>: <describe_tool_call_outcome(record)>
Do not repeat these tool calls unless the user asks again.
```

### `execute_deferred_tool_call(call)`

`DeferredToolExecution` carries `flow_id`, the approved `tool_call` record,
`idempotency_key` (`"<flow_id>/<turn_id>/<tool_call_id>"`, the same on every
attempt), `attempt`, `events`, `heartbeat`, `heartbeat_checkpoint`,
`save_heartbeat_checkpoint`, and `input_answers`. Run the real tool once and
return `ToolExecutionResult(output)`, `ToolInputRequested` when the tool
needs an answer before it takes any effect, or `ToolTaskStarted` when the tool
started remote work that the session polls (see [Tool tasks](#tool-tasks)).
Tool calls of one model response run in parallel Steps, so this method may run
concurrently with itself.

`save_heartbeat_checkpoint(text)` records a heartbeat that carries `text`, and
every later heartbeat of the attempt, the session's background heartbeats
included, carries it too. A retried attempt reads it as `heartbeat_checkpoint`
and can resume work the earlier attempt started, such as polling a job,
instead of starting it again. A `DeferredToolCall` may also carry
`tool_source_reference`, opaque text stored on the call's `AgentToolCall`
record (wire field `tool_source_reference`), so the tool's source can resolve
the tool in the tool Step without listing it again.

Raise `ToolExecutionError(message)` for a failure the model should see; the
session records `failed` with that message and does not retry. Any other
exception is a Step failure. `read` and `idempotent_write` calls retry up to
three attempts, and then `RecordToolFailure` records `failed` for a read and
`outcome_unknown` for a write, because the write may have taken effect.
Single-attempt calls do not retry; `RecordToolOutcomeUnknown` records
`outcome_unknown`. Both read the recovery error's `attempt_progress`. When it
is `NOT_STARTED`, no attempt reached a Worker and the tool never ran, so both
record `failed` with an `error` that starts with
`TOOL_CALL_NOT_RUN_ERROR_PREFIX` (`the tool did not run, so it is safe to call
again`), then `": "` and the recovery detail. The model may call the tool
again. A call whose tool already asked a question ran before, so it keeps the
status of its effect.

Before calling the runner, the tool Step writes `tool_call_started` and awaits
a heartbeat. A Worker that disconnects after it received the task fails the
attempt, and the call records `outcome_unknown`. The heartbeat matters because
Dex treats a stream that half-closes before any output as one whose task never
ran: the attempt returns to pending with the same attempt number, uses no
retry, and is handed out again. After output, a half-close is a disconnect
that fails the attempt with `NOT_COMPLETED`. So a tool that started is never
returned as not run at the same attempt number, and never reported later as
`NOT_STARTED`.

### Tool tasks

Some tools start remote work and get a handle back, such as an MCP task. A
tool Step must not hold a connection while that work runs, so
`execute_deferred_tool_call` returns
`ToolTaskStarted(task_id, poll_interval_seconds)`, and the runner implements
`ToolTaskPoller.poll_tool_task(poll)`:

```python
class ReportRunner:  # also implements AgentFrameworkRunner[DeferredToolOutcome]
    async def execute_deferred_tool_call(self, call):
        saved = call.heartbeat_checkpoint
        if saved is not None:  # an earlier attempt started the job
            return ToolTaskStarted(task_id=saved, poll_interval_seconds=5)
        job_id = await reports.start(call.idempotency_key)
        await call.save_heartbeat_checkpoint(job_id)
        return ToolTaskStarted(task_id=job_id, poll_interval_seconds=5)

    async def poll_tool_task(self, poll: ToolTaskPoll) -> ToolTaskPollOutcome:
        job = await reports.get(poll.tool_call.task_id)
        answered = poll.tool_call.input_continuation  # ID of the question answered last
        resolved = poll.pending_answer  # its resolution, on the poll right after it
        if resolved is not None and job.question_id == answered:
            # Keyed by the question ID, so a retried poll never answers a later question.
            if resolved.resolution is QuestionResolution.ANSWERED:
                job = await reports.answer(job.id, answered, resolved.answer)
            else:  # DECLINED or DEADLINE_PASSED
                job = await reports.skip(job.id, answered)
        if job.question_id is not None and job.question_id != answered:
            return ToolInputRequested(question=job.question, choices=job.choices,
                                      continuation=job.question_id)
        if not job.done:
            # Keep the answered ID only while the job still shows that question.
            kept = answered if job.question_id == answered else None
            return ToolTaskPending(poll_interval_seconds=10, continuation=kept)
        return ToolExecutionResult(output=job.summary)
```

The tool Step records `waiting_for_task` with `task_id` and
`task_poll_interval_seconds`, writes `tool_task_started`, and ends.
`PollToolTask` waits on a Timer for the interval, rounded up to whole seconds
from 1 to a day, and calls `poll_tool_task` once per execution:

- `ToolExecutionResult` records `succeeded`, and `ToolExecutionError` records
  `failed`.
- `ToolTaskPending(poll_interval_seconds, continuation=None)` stores the next
  interval, stores `continuation` as `input_continuation`, and polls again;
  `None` clears it.
- `ToolInputRequested` asks the user through `AwaitQuestionAnswer`, with the
  interrupt ID `<tool_call_id>-input-<n>`. The task has started, so the
  question never ends the call: an answer, a decline, and a passed deadline
  each go back to the task as a `QuestionAnswer` whose `resolution` is
  `ANSWERED`, `DECLINED`, or `DEADLINE_PASSED`. `PollToolTask` polls at once
  and hands it over as `poll.pending_answer`, only on that first poll; a
  retried attempt of it gets it again. That poll reads the question's
  continuation in `tool_call.input_continuation`, and later polls read what
  the last `ToolTaskPending` returned. Keep the answered question's
  continuation only while the task may still show that question, so that a
  later identical question reads as new. Deliver idempotently, keyed by a
  question ID when the task has one. When a retried attempt cannot tell
  whether an earlier one delivered, return the question again instead: the
  user answers twice rather than a stale answer reaching a later question.
  When the task cannot take a decline, cancel it best-effort and raise an
  exception other than `ToolExecutionError`; the call is then recorded by its
  effect.
- Any other exception fails the poll attempt. A poll gets three attempts; then
  `RecordToolFailure` records `failed` for a `read` call and
  `outcome_unknown` for a write, and `RecordToolOutcomeUnknown` records
  `outcome_unknown` for a single-attempt call, because the task may still run.

No Step holds a connection while the work runs, and a Worker restart only
delays the next poll. A crash during a poll repeats only that poll, so a poll
must only read the task, or deliver an answer idempotently. A `CallTool`
attempt that dies after the remote work started runs again: save the handle
with `save_heartbeat_checkpoint` before you return it, and return the saved
handle when `heartbeat_checkpoint` holds one. Dex may drop a heartbeat
checkpoint, so an attempt that dies before its checkpoint is saved and before
the tool Step commits can still start the work twice; pass the idempotency key
on so the remote system can drop the duplicate. `PollToolTask` owns the record
until it is final and takes no lock. Steering waits for a started task, its
questions included, and a call with `task_id` is never recorded `rejected`.
`cancel_turn` abandons the task instead, so Stop never waits for it: it
cancels `PollToolTask`, drops whatever a running poll returns, and records the
call `outcome_unknown` with `cancellation_reason` `turn_cancelled` in
`CancelToolTask`. Implement `ToolTaskCanceller.cancel_tool_task` to stop the
remote work best-effort; make it idempotent by `task_id`, because a crash or a
failed attempt calls it again. When it keeps failing, the call is still
recorded and the turn ends:

```python
    async def cancel_tool_task(self, cancellation: ToolTaskCancellation) -> None:
        await reports.cancel(cancellation.tool_call.task_id)
```

`DeferredToolExecution.supports_tool_tasks` tells a tool whether the runner polls tasks. A runner without `poll_tool_task`
that returns `ToolTaskStarted` anyway gets `failed` for a `read` call and
`outcome_unknown` otherwise, with `MISSING_TOOL_TASK_POLLER`.

A long task polls many times, and each poll is a Step in the session's
history, so poll at the slowest interval the task allows.

## Effects and policies

| Effect | Step | Retry | Crash mid-call | Default approval |
| --- | --- | --- | --- | --- |
| `read` | `CallTool` | Up to 3 attempts; then `failed` | Runs again | No |
| `idempotent_write` | `CallTool` | Up to 3 attempts, key `<flowId>/<turnId>/<toolCallId>`; then `outcome_unknown` | Runs again with the same key | No |
| `non_idempotent_write` | `CallSingleAttemptTool` | One attempt; then `outcome_unknown` | Outcome unknown | Yes |
| `unclassified` | `CallSingleAttemptTool` | One attempt; then `outcome_unknown` | Outcome unknown | Yes |

`ToolPolicy(effect, requires_approval=None)` overrides the defaults per tool
name; `None` uses the effect's default approval, which `is_approval_required`
reports. `AgentSessionOptions` also sets `default_tool_effect` (default
`unclassified`), the `AgentEvents` Stream capacity (default 10 MiB), and an
optional `interrupt_deadline`, `built_in_tools`, `max_durable_wait`, and
`plan_mode_enabled`. A tool
whose policy requires approval runs only after `AwaitToolApproval` approves it.

## Flow shape

```text
StartAgentSession -> AwaitUserMessage -> RunModel
AwaitUserMessage -> steer -> ApplyTurnControl
RunModel -> steered or cancelled -> ApplyTurnControl
cancel_turn -> cancels RunModel, CompactContext, and the waiting Steps -> FinishTurnCancellation(turn_id)
FinishTurnCancellation -> ApplyTurnControl, or CancelToolTask(id) per abandoned task
RunModel -> TurnCompleted -> AwaitUserMessage
RunModel -> CompactContext(through_sequence) -> RunModel             with a compaction policy
RunModel -> ToolCallsRequested -> go_to_many(
              AwaitToolApproval(id) -> CallTool(id) or CallSingleAttemptTool(id)   per call needing approval
              CallTool(id) or CallSingleAttemptTool(id)                            per other call
              AwaitQuestionAnswer(id)                                              per ask_user call
              AwaitTimer(id)                                                       per durable_wait call
              ResumeTurn(expected_result_count))
CallTool(id) -> ToolInputRequested -> AwaitQuestionAnswer(id) -> CallTool(id)
CallTool(id) -> ToolTaskStarted -> PollToolTask(id) -> PollToolTask(id) ...
PollToolTask(id) -> ToolInputRequested -> AwaitQuestionAnswer(id) -> PollToolTask(id)
ResumeTurn -> RunModel, or ApplyTurnControl when steered or cancelled
ApplyTurnControl -> RunModel with steered messages, else AwaitUserMessage
```

| Step (input) | What it does | Next |
| --- | --- | --- |
| `StartAgentSession` (`AgentSessionStartInput`) | Start Step. Initializes state, usage, interrupts, and `SessionSettings`. | `RunModel` with a non-blank `initial_message`, else `AwaitUserMessage` |
| `AwaitUserMessage` (none) | Waits for any of, in this order, `TurnControlRequests`, `QueuedUserMessages.for_one()`, and, while the plan is `draft` or `active`, `PlanExecutions.for_one(<revision>)`. Locks `TurnCancellationRequest` in Execute, so no turn starts while `cancel_turn` decides. A steer: moves to `ApplyTurnControl`. A message: stores it and starts a `chat` or `planning` turn. An approval: clears `PendingPlanExecution`, marks the plan `active`, and starts an `executing` turn. | `ApplyTurnControl`, `RunModel`, or itself for a stale approval |
| `RunModel` (none) | Ends the turn and moves to `ApplyTurnControl` without a model call when the turn is steered or cancelled, or else to `CompactContext` when a compaction policy's threshold is crossed. Otherwise calls `run_model_step`. 15 min timeout, 2 min heartbeat timeout, 3 attempts in 30 min, failure route `RecordTurnFailure`. On `ToolCallsRequested`, stores the run state, one `AgentToolCalls` record per call, and the interrupts; records `write_todos` calls, and every call but `write_todos` of a planning turn, final in this commit. | `AwaitUserMessage`, `ApplyTurnControl`, `CompactContext`, or `go_to_many` with one Step per tool call plus `ResumeTurn` |
| `CompactContext` (`CompactContextInput`) | Calls `summarize_context` once and stores `ContextSummary`. `RunModel`'s timeouts, attempts, and failure route. | `RunModel` |
| `AwaitToolApproval` (tool call ID) | Waits for `ToolCallCancellations.for_one(id)` or `InterruptResolutions.for_one(id)`, plus a Timer at the deadline, rounded up to whole seconds, while the interrupt is pending. Approved: marks the record `approved`. Rejected or expired: records `rejected` and publishes `ToolCallResults`. A Timer that fires before the deadline waits again. A cancellation records `rejected` with its reason. | `CallTool` or `CallSingleAttemptTool`, or ends |
| `AwaitQuestionAnswer` (tool call ID) | Waits for `ToolCallCancellations.for_one(id)` or `InterruptResolutions.for_one(interrupt_id)`, plus a deadline Timer while the question is pending, or a zero Timer while a tool's question is not raised yet. Locks `PendingInterrupts` in Execute, where it raises a tool's question. Records the answer, decline, or deadline and publishes `ToolCallResults`, or returns a tool question's call to its tool Step, or a task's question to `PollToolTask`. Steering never cancels a task's question; `cancel_turn` cancels the wait and abandons the task. | Ends, the tool Step, `PollToolTask`, `FinishTurnCancellation`, or itself |
| `AwaitTimer` (tool call ID) | Waits on a Timer for the whole seconds until `wait_until`, or `ToolCallCancellations.for_one(id)`. Records `succeeded`, or `rejected` when cancelled, and publishes `ToolCallResults`. | Ends |
| `CallTool` (tool call ID) | Runs one `read` or `idempotent_write` call, unless the turn is steered or cancelled before `tool_call_started`. 3 attempts, failure route `RecordToolFailure`. Records `succeeded`, or `failed` for `ToolExecutionError`, and publishes `ToolCallResults`. A `ToolInputRequested` result raises a question instead, and a `ToolTaskStarted` result records `waiting_for_task` and writes `tool_task_started`. | Ends, `AwaitQuestionAnswer`, or `PollToolTask` |
| `CallSingleAttemptTool` (tool call ID) | Runs one `non_idempotent_write` or `unclassified` call. 1 attempt, failure route `RecordToolOutcomeUnknown`. | Ends, `AwaitQuestionAnswer`, or `PollToolTask` |
| `PollToolTask` (tool call ID) | Waits on a Timer for `task_poll_interval_seconds`, or a zero Timer while `task_answer_pending`, then calls `poll_tool_task`. 3 attempts, failure route `RecordToolFailure` for `read` and `idempotent_write` calls and `RecordToolOutcomeUnknown` otherwise. No lock: steering waits for the task's result, and `cancel_turn` cancels the poll and drops its result. Records the result and publishes `ToolCallResults`, stores the next interval and continuation, or records the task's question. | Ends, itself, `AwaitQuestionAnswer`, or `FinishTurnCancellation` |
| `RecordToolFailure` (tool call ID) | Records `failed` for a read and `outcome_unknown` for a write, with the recovery detail in `error`, and publishes `ToolCallResults`. A call whose tool Step never started (`attempt_progress` `NOT_STARTED`) records `failed` with `TOOL_CALL_NOT_RUN_ERROR_PREFIX` instead. | Ends |
| `RecordToolOutcomeUnknown` (tool call ID) | Records `outcome_unknown` with the recovery detail in `error`, and publishes `ToolCallResults`. A call whose tool Step never started records `failed` with `TOOL_CALL_NOT_RUN_ERROR_PREFIX` instead. | Ends |
| `ResumeTurn` (expected result count) | Waits for `ToolCallResults.for_n(count)` or `TurnControlRequests`. Locks `PendingInterrupts` in Execute. A steer wake cancels the pending interrupts of the turn's approvals and of questions no task asked, and wakes each such approval, question, and durable wait through `ToolCallCancellations`, then waits again. | `RunModel`, `ApplyTurnControl` when steered or cancelled, or itself |
| `ApplyTurnControl` (none) | Consumes every steered message and stale wake. Ends an active steered or cancelled turn with the runner's unfinished-turn messages or a note, and writes `turn_steered` or `turn_cancelled`. | `RunModel` with the steered messages, else `AwaitUserMessage` |
| `RecordTurnFailure` (the failing Step's input, ignored) | Records `last_turn_error`. Appends the runner's unfinished-turn messages when it implements `build_unfinished_turn_messages`. When the turn has final tool call records, appends one turn-failure note. Clears the turn's records and run state. | `AwaitUserMessage` |
| `FinishTurnCancellation` (turn ID) | Locks `TurnCancellationRequest` and `PendingInterrupts`. When `cancel_turn` cancelled the model call or compaction, moves to `ApplyTurnControl`. While tool calls run, records each call whose wait was cancelled as `rejected`, hands each started task to `CancelToolTask`, and ends; `ResumeTurn` then ends the turn. | Ends, `ApplyTurnControl`, `CancelToolTask`, or a waiting Step whose answer committed first |
| `CancelToolTask` (tool call ID) | Calls `cancel_tool_task` when the runner implements `ToolTaskCanceller`, then records `outcome_unknown` and publishes `ToolCallResults`. 3 attempts in 2 min, 30 s timeout, failure route `RecordToolOutcomeUnknown`. | Ends |

Every Step sets `wait_for` and `execute` durability to SYNC explicitly, so a
`FlowConfig(step_durability=ASYNC)` never replays a Step result. Each tool
call Step and its failure route receive the tool call ID as input, so parallel
Steps never share state. Every tool call publishes exactly one `ToolCallResults`
message, when its record becomes final, and a tool Step whose record is already
final ends without running or publishing. `ResumeTurn` goes back to waiting if
a consumed result left a record of the turn unfinished. `AwaitToolApproval`
and the tool Steps load only their own `AgentToolCalls` instance, through
per-movement `StepOptions`.

Persistence, shared by every agent Flow type and exported from `dex_agents`:

| Name | Kind | Value |
| --- | --- | --- |
| `AgentState` | Attribute `AGENT_STATE` | `AgentState` |
| `AgentMessages` | AttributeMap `AGENT_MESSAGES`, key `message_sequence_key(n)` (12-digit zero-padded) | `AgentMessage`; only the window, from `AgentMessageIndex.window_first_sequence` |
| `AgentMessageChunks` | AttributeMap `AGENT_MESSAGE_CHUNKS`, key `message_sequence_key(first sequence)` | `AgentMessageChunk`, immutable, consecutive messages listed in `AgentMessageIndex.chunks` |
| `AgentMessageIndex` | Attribute `AGENT_MESSAGE_INDEX` | `AgentMessageIndex`; unset until a Step first writes a chunk |
| `TurnReplySequence` | Attribute `TURN_REPLY_SEQUENCE` | `TurnReplySequence`, written by the Step that starts a turn |
| `FrameworkRunState` | Attribute `FRAMEWORK_RUN_STATE` | string, set only while a turn waits for tool calls |
| `AgentToolCalls` | AttributeMap `AGENT_TOOL_CALLS`, key = tool call ID | `AgentToolCall`, deleted when the turn ends |
| `PendingInterrupts` | Attribute `PENDING_INTERRUPTS` | `list[AgentInterrupt]` |
| `AgentUsage` | Attribute `AGENT_USAGE` | `AgentUsage` |
| `QueuedUserMessages` | Channel `QUEUED_USER_MESSAGES` | `AgentUserMessage` |
| `InterruptResolutions` | ChannelMap `INTERRUPT_RESOLUTIONS`, key = interrupt ID | `AgentInterruptResolution` |
| `ToolCallResults` | Channel `TOOL_CALL_RESULTS`, internal | `ToolCallCompleted` |
| `SteeredUserMessages` | Channel `STEERED_USER_MESSAGES` | `AgentUserMessage` |
| `TurnControlRequests` | Channel `TURN_CONTROL_REQUESTS`, internal | `TurnControlRequest` |
| `TurnCancellationRequest` | Attribute `TURN_CANCELLATION_REQUEST` | `TurnCancellationRequest`, written only by `cancel_turn` |
| `ToolCallCancellations` | ChannelMap `TOOL_CALL_CANCELLATIONS`, internal, key = tool call ID | `ToolCallCancellation` |
| `AgentTraceContext` | Attribute `AGENT_TRACE_CONTEXT` | string, the W3C `traceparent` of the current activation's root span when its context is valid, sampled or not; deleted otherwise |
| `AgentPlan` | Attribute `AGENT_PLAN` | `AgentPlan`, unset until `write_todos` first succeeds |
| `PlanExecutions` | ChannelMap `PLAN_EXECUTIONS`, key `plan_execution_key(revision)` (decimal) | `PlanExecutionRequest` |
| `PendingPlanExecution` | Attribute `PENDING_PLAN_EXECUTION` | `PlanExecutionRequest` that `execute_plan` accepted and `AwaitUserMessage` has not consumed |
| `ContextSummary` | Attribute `CONTEXT_SUMMARY` | `ContextSummary`, set once the session compacts |
| `AgentEvents` | Stream `flow.agent_events`, one per Flow type | `AgentEvent` |

`AgentState.status` is `waiting_for_user_message`, `running_model`,
`running_tools`, or `compacting_context`, and `AgentState.interaction_mode` is `chat`, `planning`, or
`executing`. An `AgentToolCall` record is `awaiting_approval`, `approved`,
`awaiting_answer`, `waiting_for_timer`, or `waiting_for_task` while it waits,
and `rejected`, `succeeded`, `failed`, or `outcome_unknown` when final; runners
only see final records, and `poll_tool_task` a `waiting_for_task` one.

RPCs:

| RPC | Input | Output | Notes |
| --- | --- | --- | --- |
| `send_user_message` | `AgentUserMessage` | `bool` | Transactional; `False` for blank text, and for a `plan_mode` message when plan mode is off. |
| `execute_plan` | `ExecutePlanRequest` | `bool` | Locks `PendingPlanExecution`; `True` only while the session waits for a user message, the plan has this revision and is `draft` or `active`, no execution of it is pending, and nothing is queued, steered, or pending. `AgentSessionClient` retries lock conflicts. |
| `steer_user_message` | `SteerUserMessageRequest` | `bool` | Transactional, locks nothing. Moves a queued message by `message_id`, or steers with new `text`; while no turn is active, queues `text`. `False` for both or neither, blank text, a message no longer queued, or a turn that does not accept steering; `ChannelMessageNotFoundError` when it was consumed while the RPC ran. |
| `delete_queued_message` | `DeleteQueuedMessageRequest` | `bool` | Transactional, locks nothing. `False` for a message no longer queued. |
| `cancel_turn` | `CancelTurnRequest` | `bool` | Locks `TurnCancellationRequest` and `PendingInterrupts`. Cancels the turn's model call, compaction, and waiting Steps, and moves to `FinishTurnCancellation`. `False` when no turn is active, `turn_id` names another turn, or the turn does not accept cancellation. |
| `resolve_interrupt` | `ResolveInterruptRequest` | `bool` | Locks `PendingInterrupts`; `False` when not pending or past its deadline; removes the interrupt and publishes the answer in one commit. On a question, `approved=False` declines it and `approved=True` returns `False`. A concurrent Step can raise `RpcLockConflictError`; `AgentSessionClient` retries it for 5 seconds. |
| `answer_question` | `AnswerQuestionRequest` | `bool` | Locks `PendingInterrupts`; `False` when the interrupt is not a pending question, is past its deadline, or the answer is blank or not an allowed choice; removes it and publishes the answer in one commit. `AgentSessionClient.answer_question` retries lock conflicts like `resolve_interrupt`. |
| `get_agent_snapshot` | none | `AgentSnapshot` | Loads the window and every kept chunk. State, last 200 stored messages, pending interrupts, the active turn's tool calls, usage, the queued message count, the plan, and the queued and steered messages with their IDs. Serves retained closed sessions. |
| `list_agent_messages` | `ListAgentMessagesRequest` | `AgentMessagesPage` | Loads the window and every kept chunk. Oldest first; limit clamped to 1..200; the last page starts at `AgentMessageIndex.first_sequence`, and `deleted_before_sequence` is set once retention deleted earlier messages. |

Events on `AgentEvents` (`AgentEventType`): `turn_started` `{"sequence"}`,
`turn_completed` `{"message_count"}`, `turn_failed` `{"error"}`,
`turn_steered` `{"steered_message_count", "message_count"}`,
`turn_cancelled` `{"message_count"}`, `text_delta`
and `reasoning_delta` `{"text"}`, `tool_call_started` `{"tool_call_id",
"tool_name"}`, `tool_call_finished` `{"tool_call_id", "tool_name", "status"}`
(once per tool call, including rejections), `tool_call_deferred`
`{"tool_call_id", "tool_name", "effect", "requires_approval"}`,
`interrupt_raised` `{"interrupt_id", "tool_call_id", "tool_name", "deadline",
"kind", "question", "choices", "allows_free_text"}`, `interrupt_resolved`
`{"interrupt_id", "approved", "rejection_message", "kind", "answered",
"cancellation_reason"}` (`steered` or `turn_cancelled` when steering or
cancellation removed the interrupt), `timer_started` `{"tool_call_id",
"wait_until", "reason"}`, `plan_updated` `{"revision", "status",
"task_count"}` (after `write_todos`, or when an approval marks a draft
`active`), `plan_execution_started` `{"revision"}`, `tool_task_started`
`{"tool_call_id", "tool_name", "task_id"}`, `usage_recorded`
`{"recorded", "session"}`, `compaction_started` `{"through_sequence"}`, and
`compaction_finished` `{"generation", "through_sequence"}`.

Every event carries `turn_id`, `step_execution_id` (shared by all attempts of
one Step execution), and the Step `attempt`. A UI groups live text by
`(turn_id, step_execution_id, attempt)`: within one `step_execution_id` it
keeps only the highest `attempt` seen and drops the partial text of lower
attempts. Different `step_execution_id` values of a turn are separate model
call and tool call Steps, and each keeps its text. After `turn_completed`,
`turn_failed`, `turn_steered`, or `turn_cancelled`, replace the turn's live
text with the stored messages. Streams
are best effort; rebuild from `get_agent_snapshot`.

## MCP tools (`dex_agents.mcp`)

`dex_agents.mcp` turns the tools of MCP servers into tools of an agent
session. Every MCP tool call runs in its own tool Step, with an effect class
from the tool's annotations, so a crash never repeats a write silently.
Install the MCP SDK with the `mcp` extra; the core itself never imports it:

```bash
pip install "dex-agents[mcp]"
```

Configure the servers on the Worker, never from Flow input, a user message, or
a browser: a stdio server's command runs with the Worker's privileges.

```python
from dex_agents.mcp import McpStdioServerConfig, McpStreamableHttpServerConfig, McpToolSource

tools = McpToolSource([
    McpStdioServerConfig(name="files", command="mcp-server-files", args=["/srv/reports"]),
    McpStreamableHttpServerConfig(
        name="tickets",
        url="https://tickets.example.com/mcp",
        headers={"Authorization": f"Bearer {os.environ['TICKETS_TOKEN']}"},
    ),
])
```

A Stepwise agent takes the source in `AgentDefinition.tool_sources`; see
[Tool sources](#tool-sources). Another runner offers the tools and defers
every call; the tool Step runs it:

```python
class MyRunner:
    async def run_model_step(self, turn: AgentTurnContext) -> AgentTurnOutcome:
        offered = await tools.list_tools()  # McpTool: name, description, input_schema
        response = await call_my_model(turn.messages, offered)
        if not response.tool_calls:
            return TurnCompleted(new_messages=..., usage=...)
        return ToolCallsRequested(
            framework_run_state=serialize(response),
            tool_calls=[
                tools.deferred_tool_call(
                    turn, tool_call_id=call.id, tool_name=call.name, arguments=call.arguments
                )
                for call in response.tool_calls
            ],
            usage=...,
        )

    async def execute_deferred_tool_call(self, call: DeferredToolExecution) -> DeferredToolOutcome:
        handlers = {**self._local_handlers, **await tools.deferred_tool_handlers()}
        return await handlers[call.tool_call.tool_name](call)

    async def poll_tool_task(self, poll: ToolTaskPoll) -> ToolTaskPollOutcome:
        return await tools.poll_tool_task(poll)
```

`list_tools` lists each server on its own and caches its tools per Worker.
The first listing of a server is awaited; after `listing_ttl` (5 minutes) a
server is listed again in the background while its earlier tools are
returned. A server that cannot be listed is left out with a warning log and
is not tried again for `listing_failure_backoff` (30 seconds), then in the
background, so one bad server never hides the others' tools or slows every
model call. One listing per server runs at a time. `list_tools(refresh=True)`
lists every server again now. `deferred_tool_call` stores the
server and the tool's MCP name on the call's record, so the tool Step calls
that server directly, without a listing; `call_tool` finds the server through
the listing. Tool names are `<server>__<tool>`, with characters other than
letters, digits, `_`, and `-` replaced by `_` and names over 64 characters
shortened with a hash. When two tools get the same name, the first in server
order is offered and the other is left out with a warning log. Key
`ToolPolicy` entries by that name.

### Resources and prompts

With `resource_and_prompt_tools_enabled=True`, `list_tools` also offers three
`READ` tools after the servers' tools, when at least one server is
configured. Each takes a configured server's name as `server`, an `enum` of
the names, and opens a new connection to that server with the source's
transport, headers, and `connection_timeout`:

| Tool | Arguments | Output |
| --- | --- | --- |
| `mcp_list_resources_and_prompts` | `server` | `{"prompts": [...], "resource_templates": [...], "resources": [...]}`, every page |
| `mcp_read_resource` | `server`, `uri` | `{"contents": [...]}` |
| `mcp_get_prompt` | `server`, `name`, optional `arguments` (strings by name) | `{"description": ..., "messages": [...]}` |

Outputs are the server's items as it sent them, in JSON with sorted keys, so
both cores show the same text. A server that does not offer resources or
prompts lists none. A connection failure is retried `connection_attempts`
times. An MCP error answer, a server that stays unreachable, and an
unconfigured server raise `ToolExecutionError`. Enable the tools on one
source of an agent only. Resource subscriptions and list-change notifications are
not followed. The names are `MCP_LIST_RESOURCES_AND_PROMPTS_TOOL_NAME`,
`MCP_READ_RESOURCE_TOOL_NAME`, and `MCP_GET_PROMPT_TOOL_NAME`; key a
`ToolPolicy` by them to require approval.

### Effects

| MCP annotations | Effect | Tool Step | Approval by default |
| --- | --- | --- | --- |
| `readOnlyHint: true` | `read` | `CallTool` | No |
| `idempotentHint: true` | `idempotent_write` | `CallTool` | No |
| anything else, or none | `default_tool_effect`, `unclassified` by default | `CallSingleAttemptTool` | Yes |

`idempotentHint` makes a tool `idempotent_write` whatever `destructiveHint`
says, in both cores. A `ToolPolicy` in `AgentSessionOptions.tool_policies` wins
over the annotations. `trusts_tool_annotations` defaults to `True`, so a server
with a wrong `readOnlyHint` on a write gets that write run without approval and
retried up to three times. Set `trusts_tool_annotations=False` on a server
whose hints you do not trust; its tools then keep the session default unless a
policy names them.

### What a call does

Every request opens a new connection: a new process for a stdio server, a new
HTTP connection for a Streamable HTTP server. Connecting and each handshake
request are bounded by `connection_timeout` (30 seconds by default), and a
`tools/call` by `request_timeout` (10 minutes). The source speaks MCP 2026-07-28,
which has no protocol session, and falls back to the `initialize` handshake for
older servers, so a server restart between listing and calling is harmless.
Every `tools/call` carries the session's idempotency key,
`<flowId>/<turnId>/<toolCallId>`, in `_meta` under
`IDEMPOTENCY_KEY_META_KEY` (`io.superdurable.dex/idempotencyKey`). The MCP SDK
adds `traceparent` to `_meta` when OpenTelemetry has an active trace.

| What happens | Recorded | Re-issued? |
| --- | --- | --- |
| A result | `succeeded`: text blocks joined with newlines, other content as JSON | - |
| An `isError` result, or a JSON-RPC error response other than `-32603` | `failed` with the text (`ToolExecutionError`) | No |
| The call was never sent: its server is not configured, or 3 connection attempts failed | `failed` | No |
| The stream broke, the server exited, or the request timed out after it was sent; an internal error (`-32603`), an HTTP error status, or a body that is not JSON-RPC (`-32700`), which a gateway can return after the server ran the call; a server that keeps asking to repeat the call | `McpToolCallInterruptedError`, a Step failure | `read` and `idempotent_write`: `CallTool` sends a new request, same idempotency key, up to 3 attempts. Other effects: `outcome_unknown`, never sent again |
| The Worker dies during the call | as for any tool Step | Same as above |

MCP 2026-07-28 removed stream resumption, so a broken stream is never resumed;
the effect class alone decides whether the call is sent again.

`notifications/progress` become `mcp.tool_progress` events on `AgentEvents`
(`{"tool_call_id", "tool_name", "progress", "total", "message"}`).

### Long-running MCP tasks

The source declares the tasks extension, `io.modelcontextprotocol/tasks`, on
every request. A server may then answer `tools/call` with a task handle. MCP
2026-07-28 has no protocol session and lets the server choose the task TTL, so
any connection can poll the task.

A Streamable HTTP server's task becomes a [tool task](#tool-tasks): the tool
Step saves the task ID as its heartbeat checkpoint, returns `ToolTaskStarted`,
and ends. `PollToolTask` then calls `poll_tool_task` on a Timer, which polls
`tasks/get` once and writes an `mcp.task_status` event when the status
changed. The next interval is
the server's `pollIntervalMs`, rounded up to whole seconds, or, when that is
shorter, double the previous interval, up to a minute, so a task that runs for
hours adds a few Steps a minute at most. A completed task is the tool's
result; a failed or cancelled task records `failed`. No Step holds a
connection while the task runs, a Worker restart only delays the next poll,
and a task can run for days. A `CallTool` attempt that dies after `tools/call`
returned the task reads the checkpoint on its retry and returns the same task
without calling the tool again, once the checkpoint reached Dex; a lost
checkpoint leaves only the idempotency key to drop a second task. A runner
without `poll_tool_task` gets every task polled inside the tool Step, as
below. A server that no longer knows the task, or
cannot be reached for three poll attempts, records `failed` for a `read` tool
and `outcome_unknown` otherwise; the source never calls the tool again from a
poll. `cancel_turn` abandons the task, and `cancel_tool_task` sends
`tasks/cancel`; forward it from your runner's `cancel_tool_task`, as the
Stepwise runner does. Any answer from the server ends the cancellation, so a
retried one is harmless.

A stdio task lives in its server process, which ends with the request. The
tool Step polls it on the call's own connection, with a heartbeat per poll,
for at most `task_wait_timeout` (10 minutes by default) per attempt, holding a
Worker slot. When the wait passes, or the Worker dies, `CallTool` retries a
`read` or `idempotent_write` call and the retry calls the tool again; other
effects record `outcome_unknown`. `call_tool`, the framework adapter path,
polls every task this way.

### Input requests

A task in `input_required` that asks one form `elicitation/create` whose
schema has a single `string` field, with or without `enum`, or a single
`boolean` field becomes a session question: `poll_tool_task` returns
`ToolInputRequested` with the elicitation's message, the `enum` values or
`yes` and `no` as choices, and the request key and a fingerprint of the
request in its continuation. `AwaitQuestionAnswer` raises the `question`
interrupt `<tool_call_id>-input-<n>`. The first poll after the question is
resolved, which runs at once, sends the resolution with `tasks/update`: an
`accept` input response with the answer, `decline` for a declined question,
or `cancel` for a passed deadline. It sends it only while the task still
lists that same request, so it never answers a different later request
under the same key, and only on its first attempt, after a heartbeat. With
output sent, a half-close of the stream fails the attempt instead of returning
it as not run, so once the send may have happened, Dex retries that attempt
rather than handing it out again under the same attempt number.
A retried attempt cannot tell whether the first one sent it, and the task may
since have asked the identical request again, so it asks the user again
instead: the user may answer twice, but an answer never reaches a later
question.

Right after `tasks/update`, the task may still list the answered request until
it takes the answer in. For 10 seconds after the send, polls run a second apart
and read that request as working; the record's continuation carries the send
time. Once a poll sees the task leave `input_required`, or after those 10
seconds, an identical request is a new question, so a task that asks the same
thing once per batch is asked each time. The task's own result then ends the
call. When `tasks/update` fails for a decline, or the task is cancelled after
one, the source cancels it best-effort and fails the poll, so the call is
recorded `failed` for a `read` tool and `outcome_unknown` otherwise.

Any other input request records `failed` with `the MCP server asked for input
that this agent session cannot provide yet (...)`, naming each request, such as
`an answer to 'Which account?'`, and the waiting task is cancelled. A stdio
task's input request fails this way too, because its process ends with the
request. An `input_required` answer to `tools/call` itself, outside a task,
records `failed` the same way; mapping it to a question is a follow-up. A
state-only `input_required` answer, which only asks for the same request
again, is retried at once.

### With a framework adapter

The runner above is the core's runner-agnostic path. A framework adapter uses
the source when it accepts tools defined by a JSON Schema: wrap each `McpTool`
as a framework tool whose function calls `tools.call_tool(name, arguments,
idempotency_key=...)` with the key from the adapter's tool context, and pass
`{**tools.tool_policies(), **your_policies}` as the adapter's `tool_policies`.
No adapter package changes for this.

| Adapter | Schema-only tool | Key inside the tool |
| --- | --- | --- |
| `dex-agents-openai-agents` | `FunctionTool(params_json_schema=..., on_invoke_tool=...)` | `get_current_deferred_tool_execution()`, the whole call |
| `dex-agents-pydantic-ai` | `Tool.from_schema(...)` | `deferred_tool_idempotency_key(ctx)` |
| `dex-agents-langgraph` | `StructuredTool` with a JSON Schema `args_schema` | `get_tool_call_idempotency_key(config)` |
| `dex-agents-google-adk` | a `BaseTool` with its own `FunctionDeclaration` | `idempotency_key_for_tool_call(tool_context)` |
| `dex-agents-claude-agent-sdk` | `claude_agent_sdk.tool(name, description, schema)` | `current_deferred_tool_execution()`; write calls only, reads run inside the harness |

Pydantic AI, LangGraph, and Google ADK also defer their own MCP client tools
into tool Steps; use the source there for the shared idempotency key and
annotation handling. The OpenAI Agents SDK runs its own MCP servers inline in
`RunModel`, so a wrapper is its only path to tool Steps.

## Browser streams

`dex_agents.ui_stream` lets a browser render a live session from its
`AgentEvents` Stream and recover after a reconnect. It needs no store besides
Dex and no web framework. An application mounts two endpoints:

- **Snapshot.** `read_agent_event_stream_snapshot(client, flow, session_id)`
  returns the `get_agent_snapshot` result plus the Dex resume token to open the
  event stream from. Serve its `to_json()`: `{"snapshot": ..., "resume_token":
  ...}` in the wire format. While a turn is active, the token replays that
  turn's events, so the browser can rebuild its live text. The snapshot leaves
  out the fields only the session's Steps use: `framework_item` on messages,
  the framework's replay data, which can hold the agent's instructions, and
  `tool_source_reference`, `trace_context`, `input_continuation`,
  `input_answers`, and `task_answer_pending` on tool calls. Read them with `get_agent_snapshot` on the
  server when needed. It includes `session_settings` as stored, so keep
  anything a browser must not see out of session settings, or remove the key.
- **Events.** `await stream_agent_event_frames(client, flow, session_id,
  resume_token=...)` checks the token with one short read, then returns a
  generator of Server-Sent Event frames. Each frame's `id` is the event's Dex
  resume token and its `data` is the `AgentEvent` wire JSON, with `payload` as
  JSON text. A reconnecting `EventSource` sends the last `id` as
  `Last-Event-ID`; pass it as `resume_token` and the stream continues with the
  next event, with no repeated or missing event while the Stream retains them.
  After each empty long poll it yields `KEEP_ALIVE_FRAME`, an SSE comment. A
  token that Dex rejects, such as one of another session, raises
  `InvalidAgentEventResumeTokenError` before the response starts; answer 400,
  which stops the `EventSource`, and let the browser load the snapshot again.

The `_sync` variants, `read_agent_event_stream_snapshot_sync` and
`stream_agent_event_frames_sync`, take a synchronous `Client` for WSGI
servers. Send `AGENT_EVENT_STREAM_HEADERS` with the event response. The JSON
matches the TypeScript core's `@superdurable/dex-agents/ui-stream`, which also
offers an AI SDK UI message stream for `useChat`.

**Authorize every route.** The session ID and the resume token come from the
browser, so each route must check that the caller may read the session it
names before it calls a helper.

FastAPI or Starlette, with an `AsyncClient`:

```python
from typing import Any

from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse

from dex_agents.ui_stream import (
    AGENT_EVENT_STREAM_HEADERS,
    InvalidAgentEventResumeTokenError,
    read_agent_event_stream_snapshot,
    stream_agent_event_frames,
)

app = FastAPI()


@app.get("/sessions/{session_id}/snapshot")
async def get_session_snapshot(session_id: str, request: Request) -> dict[str, Any]:
    await authorize_session(request, session_id)
    return (await read_agent_event_stream_snapshot(client, flow, session_id)).to_json()


@app.get("/sessions/{session_id}/events")
async def get_session_events(session_id: str, request: Request) -> Response:
    await authorize_session(request, session_id)
    resume_token = request.headers.get("last-event-id") or request.query_params.get(
        "resume_token", ""
    )
    try:
        frames = await stream_agent_event_frames(
            client, flow, session_id, resume_token=resume_token
        )
    except InvalidAgentEventResumeTokenError:
        return Response(status_code=400)
    return StreamingResponse(frames, headers=dict(AGENT_EVENT_STREAM_HEADERS))
```

Starlette without FastAPI uses the same calls in `Route` handlers, returning
`JSONResponse(value.to_json())` and `StreamingResponse(frames, headers=...)`.

Quart, with an `AsyncClient`. Turn off the response timeout, because the
stream stays open:

```python
from quart import Quart, make_response, request

app = Quart(__name__)


@app.get("/sessions/<session_id>/snapshot")
async def get_session_snapshot(session_id: str):
    await authorize_session(request, session_id)
    return (await read_agent_event_stream_snapshot(client, flow, session_id)).to_json()


@app.get("/sessions/<session_id>/events")
async def get_session_events(session_id: str):
    await authorize_session(request, session_id)
    resume_token = request.headers.get("Last-Event-ID") or request.args.get("resume_token", "")
    try:
        frames = await stream_agent_event_frames(
            client, flow, session_id, resume_token=resume_token
        )
    except InvalidAgentEventResumeTokenError:
        return "", 400
    response = await make_response(frames, dict(AGENT_EVENT_STREAM_HEADERS))
    response.timeout = None
    return response
```

Flask, with a synchronous `Client`. Each open stream holds one worker thread,
so serve it from a threaded or greenlet server:

```python
from flask import Flask, Response, request

from dex_agents.ui_stream import (
    AGENT_EVENT_STREAM_HEADERS,
    InvalidAgentEventResumeTokenError,
    read_agent_event_stream_snapshot_sync,
    stream_agent_event_frames_sync,
)

app = Flask(__name__)


@app.get("/sessions/<session_id>/snapshot")
def get_session_snapshot(session_id: str):
    authorize_session(request, session_id)
    return read_agent_event_stream_snapshot_sync(sync_client, flow, session_id).to_json()


@app.get("/sessions/<session_id>/events")
def get_session_events(session_id: str) -> Response:
    authorize_session(request, session_id)
    resume_token = request.headers.get("Last-Event-ID") or request.args.get("resume_token", "")
    try:
        frames = stream_agent_event_frames_sync(
            sync_client, flow, session_id, resume_token=resume_token
        )
    except InvalidAgentEventResumeTokenError:
        return Response(status=400)
    return Response(frames, headers=dict(AGENT_EVENT_STREAM_HEADERS))
```

In the browser, load the snapshot, then open the event stream from its token:

```js
const { snapshot, resume_token } = await (await fetch(`/sessions/${sessionId}/snapshot`)).json();
render(snapshot);
const events = new EventSource(
  `/sessions/${sessionId}/events?resume_token=${encodeURIComponent(resume_token)}`,
);
events.onmessage = (message) => applyAgentEvent(JSON.parse(message.data));
```

On a reconnect the browser repeats the URL and adds `Last-Event-ID`, which the
routes above prefer to the query parameter. The browser applies the event
rules in [Flow shape](#flow-shape): within one `step_execution_id` it keeps
only the highest `attempt`, so a `RunModel` retried after a crash shows only
its new attempt's text. After `turn_completed`, `turn_failed`, `turn_steered`,
or `turn_cancelled`, and after any long gap, it reloads the snapshot. Approve or reject a pending interrupt from
the snapshot with `AgentSessionClient.resolve_interrupt`.

Limits:

- Stream events live in the memory of the active Dex Server process, so a
  restart or failover loses them, and trimming can remove old events. With the
  Stream Store disabled (`streamStore.backend: disabled`), the snapshot still works, with an empty token, and the
  event stream yields only keep-alive comments; poll the snapshot instead.
- The event stream never ends by itself; the web framework closes it when the
  browser disconnects. An async generator stops when the framework cancels
  it; a synchronous one first finishes its current long poll, up to
  `poll_timeout`.
- Stopping in the browser only closes the response. To stop the turn, the
  application's stop route calls `cancel_turn`.

## Tracing

The session emits OpenTelemetry GenAI spans through the OpenTelemetry API. The
`otel` extra installs only `opentelemetry-api`; the application brings the SDK
and an exporter:

```bash
pip install "dex-agents[otel]" opentelemetry-sdk opentelemetry-exporter-otlp
```

Tracing is on by default and uses the global tracer provider, so an
application that configures the SDK gets spans without further setup. Without
the API, the session never imports OpenTelemetry. With the API but no SDK, the
spans are no-ops and the session stores nothing extra. `AgentTracingOptions`,
passed as `AgentSessionOptions.tracing`, changes the defaults:

```python
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from dex_agents import AgentSessionFlow, AgentSessionOptions, AgentTracingOptions

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
flow = AgentSessionFlow(
    MyFrameworkRunner(agent),
    flow_type="TriageAgent",
    options=AgentSessionOptions(
        tracing=AgentTracingOptions(tracer_provider=provider, agent_name="Triage"),
    ),
)
```

| Option | Default | Meaning |
| --- | --- | --- |
| `is_enabled` | `True` | `False` emits nothing and never imports OpenTelemetry. |
| `tracer_provider` | `None` | The provider to emit with; `None` uses the global provider. |
| `agent_name` | the Flow type | `gen_ai.agent.name` and the `invoke_agent` span name. |
| `should_capture_message_content` | `False` | Put message text, tool arguments, and tool results on spans. |
| `semantic_conventions_version` | `V1_42_0_DEV` | The GenAI semantic conventions the spans follow. |

### One trace per activation

An activation is a burst of Steps between durable waits. It starts when a Step
consumes an input from outside the session: `StartAgentSession`,
`AwaitUserMessage`, or `ApplyTurnControl` with a user message, or
`AwaitToolApproval` with an interrupt answer or an expired deadline. It ends when every Step it started has
ended or waits again. Each activation is one trace:

| Span | Kind | Emitted by | Parent |
| --- | --- | --- | --- |
| `invoke_agent {agent name}` | `INTERNAL` | Each attempt of the Step that starts the activation | None. It is the trace root, with a span link to the previous activation's root. |
| `chat`, or `chat {model}` when the runner returns `ModelCallMetadata` | `CLIENT`; `INTERNAL` when the runner reports `model_calls` other than 1 | Each `RunModel` and `CompactContext` attempt, when it ends | The activation's root |
| `execute_tool {tool name}` | `INTERNAL` | Each `CallTool`, `CallSingleAttemptTool`, and `PollToolTask` attempt | The root of the activation that deferred or approved the call |

A turn without approvals is one trace. Each approval answer starts another,
linked to the one before, and so does each new turn. Every span ends inside its
Step attempt, so no span stays open across a durable wait, and every backend
sees a closed root.

The `invoke_agent` root ends with the Step that starts the activation, so its
duration is near zero, and the activation's later spans start after it ended.
That is on purpose. A span must end in the process that started it, and the
OpenTelemetry API cannot give a span an ID before it starts, so a root that
covered the whole activation would have to stay open across Steps, Workers,
and possibly days. Tracing backends drop or hide such traces. A closed root
with later children is a valid trace that every backend shows.

While a tool call runs, its `execute_tool` span is the current span. While a
model call runs, the activation's root is current, because the `chat` span
starts when the attempt ends: its kind depends on the `model_calls` the runner
reports. Framework and HTTP client instrumentation therefore nests under the
activation.

The Steps of one activation can run in different Worker processes. The root
therefore stores its W3C `traceparent`, about 55 bytes, in the
`AgentTraceContext` Attribute in the same commit as its Step. `RunModel` reads
it to parent its `chat` span and copies it onto each tool call record it
creates, as `AgentToolCall.trace_context`. An approval replaces the record's
copy with its own root. Each tool Step parents its spans from its record, so
overlapping approvals never swap tool spans between traces. The next
activation's root reads the Attribute for its link.

The Attribute is written whenever the root has a valid span context, sampled
or not. A sampled-out root stores its unsampled context, so under a
parent-based sampler the activation's other spans are dropped with it instead
of starting traces of their own. When a root has no valid context, because
its Worker has tracing disabled or has no OpenTelemetry API or SDK, that Step
deletes the stored value. The activation's later spans then start traces of
their own, instead of joining the previous activation's trace. The schema
always registers `AgentTraceContext`, so Workers with and without tracing can
serve one Flow type.

Every attempt is its own span with `dex.step.attempt`, and all attempts of one
Step execution share `dex.step.execution_id`, so a retry shows as a second span
of the same Step. A failed attempt has status `ERROR`, `error.type` set to the
exception class name, and the exception as an event. A `ToolExecutionError`
sets `error.type` to `ToolExecutionError`. An attempt whose Worker died exports
no span. `ResumeTurn`, `RecordToolFailure`, `RecordToolOutcomeUnknown`, and
`RecordTurnFailure` emit no span: the failed attempts before them already have
error spans.

### Attributes

| Span | Attribute | Value |
| --- | --- | --- |
| All | `gen_ai.conversation.id`, `session.id`, `dex.flow.id` | The Flow ID, which is the session ID |
| All | `dex.run.id` | The Flow run ID |
| All | `dex.step.type`, `dex.step.execution_id`, `dex.step.attempt` | The Step that emitted the span |
| All | `dex.agent.turn.id` | `turn-<n>` |
| All | `gen_ai.operation.name` | `invoke_agent`, `chat`, or `execute_tool` |
| `invoke_agent`, `execute_tool` | `gen_ai.agent.name` | `agent_name` |
| `invoke_agent` | `dex.agent.activation.trigger` | `user_message`, `interrupt_resolution`, or `interrupt_deadline` |
| `invoke_agent` | `dex.agent.user_message.key` | `message_sequence_key` of the user message |
| `invoke_agent` | `dex.agent.interrupt.id`, `dex.agent.interrupt.approved` | The interrupt and its decision |
| `chat` | `gen_ai.request.model`, `gen_ai.response.model`, `gen_ai.provider.name`, `gen_ai.response.id`, `gen_ai.response.finish_reasons` | From `ModelCallMetadata`, when the runner returns it |
| `chat` | `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens` | From `usage`, when `model_calls` is 1 and the input or output count is not zero |
| `chat` | `gen_ai.usage.cache_read.input_tokens`, `gen_ai.usage.reasoning.output_tokens` | `cached_input_tokens` and `reasoning_tokens`, under the same condition, when not zero |
| `chat` | `dex.agent.input_messages.last_key` | `message_sequence_key` of the newest stored message the call received; for `CompactContext`, the last message it summarized |
| `chat` | `dex.agent.output_messages.keys` | `message_sequence_key` values of what this call produced: the `TurnCompleted` messages after the turn's last tool result |
| `chat` | `dex.agent.requested_tool_call.ids` | `AgentToolCalls` keys that a `ToolCallsRequested` created |
| `execute_tool` | `gen_ai.tool.name`, `gen_ai.tool.call.id` | The tool and its `AgentToolCalls` key |
| `execute_tool` | `dex.agent.tool.effect`, `dex.agent.tool_call.status` | The effect and the final status |

The `chat` span is named `chat {request_model}` when the runner returns
`ModelCallMetadata.request_model`, and `chat` otherwise. Each framework
adapter in this repository returns `ModelCallMetadata` with what its framework
reports for the call and leaves the rest unset. Its README lists the fields it
sets and where they come from.

A runner that reports `model_calls` other than 1, such as a whole-turn harness
that makes several model calls in one `RunModel`, gets an `INTERNAL` `chat`
span without usage attributes. Instrumentation of the provider client
typically traces each of those calls with its own usage, so a summed count on
the session's span would count the tokens twice. Usage still reaches the
`AgentUsage` Attribute.

A single model call can still be counted twice. When instrumentation of the
provider client, such as an OpenAI or Anthropic instrumentation library, also
traces the call, both its span and the session's `chat` span carry usage. Sum
usage from one instrumentation scope in the tracing backend, or remove the
session's usage attributes, for example with an OpenTelemetry Collector
transform processor:

```yaml
processors:
  transform/drop_dex_agents_usage:
    trace_statements:
      - context: span
        statements:
          - delete_matching_keys(attributes, "^gen_ai\\.usage\\.") where instrumentation_scope.name == "dex_agents"
```

### Tracing failures

Tracing never fails or retries a Step. Every OpenTelemetry call is guarded:
starting, annotating, and ending spans, changing the current context, and
serializing content. An exception there, such as one from a span processor or
exporter, is logged on the `dex_agents.tracing` logger and that span is
skipped. The first failure in a process logs a warning with its traceback;
later ones log at debug level. The Step's own result is unchanged, so a broken
processor can never send a write to `outcome_unknown` or rerun a tool.

### Message content

Spans carry content by reference by default: the Flow ID, Step execution ID,
message sequence keys, and tool call IDs above, because the content is already
durable in Dex. Read it with `get_agent_snapshot`, `list_agent_messages`, or
`AsyncClient.get_attributes`. With `should_capture_message_content=True`, spans
also carry content as JSON text in the GenAI message format:

- `invoke_agent`: `gen_ai.input.messages`, the user message.
- `chat`: `gen_ai.input.messages`, the stored messages plus the turn's tool
  calls and their recorded results, and `gen_ai.output.messages`, what this
  model call produced. For `CompactContext`, the messages it summarized and
  the summary text.
- `execute_tool`: `gen_ai.tool.call.arguments` and `gen_ai.tool.call.result`.

Content comes from the `AgentMessage` projections (`text` and the `tool_*`
fields), not from `framework_item`. System instructions and tool definitions
are not captured. Content can be sensitive and large; capture it only where the
tracing backend may store it.

The spans set no `gen_ai.input.messages_ref` or `gen_ai.output.messages_ref`.
Those attributes take an absolute URI, and Dex has no URI for a stored
message. Use the `dex.agent.*` keys with the Flow ID instead.

### Semantic conventions version

The GenAI semantic conventions are still marked Development, so the session
pins a version and sets its schema URL on the tracer.
`GenAISemanticConventionsVersion.V1_42_0_DEV`, the default, follows
[`open-telemetry/semantic-conventions-genai`](https://github.com/open-telemetry/semantic-conventions-genai)
at commit [`dc3571562603552dc555c667dfa327a638d94ab2`](https://github.com/open-telemetry/semantic-conventions-genai/tree/dc3571562603552dc555c667dfa327a638d94ab2)
(2026-11-01). Its schema URL is
`https://opentelemetry.io/schemas/gen-ai-dev/1.42.0-dev`; the repository has no
release tag yet.
`V1_36_0` follows OpenTelemetry semantic conventions 1.36.0 for backends that
still read the older names: `chat` spans carry `gen_ai.system` instead of
`gen_ai.provider.name`, omit the cache and reasoning token counts, and never
carry content, because 1.36.0 recorded content as events.

## Testing an adapter

`dex_agents.testing` runs integration tests against a private `dexcli dev
--engine zd` engine. It reads `DEXCLI` and expects `dex-server-zd` next to it
(or `DEX_SERVER_ZD_BINARY`). The engine runs with
`dev.durabilityMode: processCrashSafe`, which skips fsync and survives a
process kill, through a `-zd-config` file that also works with `dexcli`
releases that predate `--durability`. A
`server_zd_config` file of your own replaces it, so set `dev.durabilityMode`
there too. The Worker runs in a child process in its own process group, so a
test can SIGKILL it at an exact moment.

```python
# tests/my_worker.py: the Worker child.
from dex_agents.testing import run_agent_worker
from tests.my_agent import build_flow

if __name__ == "__main__":
    run_agent_worker([build_flow()])
```

```python
# tests/test_approvals.py: the test process builds the same Flow for its Client.
from pathlib import Path
from dex_agents.testing import EXECUTE_COMPLETED_EVENT, AgentTestStack, wait_for_agent_snapshot
from tests.my_agent import build_flow

PACKAGE_DIR = Path(__file__).resolve().parents[1]

async def test_write_survives_a_worker_kill(tmp_path: Path) -> None:
    flow = build_flow()
    async with AgentTestStack(tmp_path, [flow], "tests.my_worker",
                              worker_cwd=PACKAGE_DIR,
                              worker_env={"MY_STATE_DIR": str(tmp_path)}) as stack:
        sessions = stack.sessions(flow)
        worker = await stack.start_worker()
        await sessions.start_session("session-1", "refund order 7")
        interrupt = await wait_for_agent_snapshot(
            sessions, "session-1",
            lambda snapshot: next(iter(snapshot.pending_interrupts), None),
            description="a pending approval")
        worker.kill()                    # SIGKILL while the session waits
        await stack.start_worker()       # replacement Worker
        assert await sessions.resolve_interrupt(
            "session-1", interrupt.interrupt_id, approved=True)
        history = await stack.read_step_history("session-1")
        executed = [entry.step_type for entry in history
                    if entry.event_type == EXECUTE_COMPLETED_EVENT]
```

Run tests with `pytest-asyncio` in auto mode, as `pyproject.toml` here does.

Inside the Worker child, a tool marks a `FileGate(state_dir, name)` and awaits
`wait_for_release()`; the test awaits `wait_until_reached()` and kills the
Worker at that exact point. Two gates reached at once prove that two tool
Steps run in parallel. `InvocationLog(path)` counts real tool and model calls
across Worker processes. `read_step_history(server_address, flow_id)` lists
completed Steps (`StepHistoryEntry` with `event_type`, `step_type`,
`step_execution_id`, `final_attempt`) through `dexcli flow history`.
`wait_until(description, probe)` polls with a deadline instead of sleeping.
The core's own tests in `tests/` use a scripted runner and show every pattern.

Each package's `ci.sh` installs locked dependencies, runs lint and type
checks, and runs the integration tests with `DEXCLI` set. `Agents CI`
discovers every `agents/*/*/ci.sh`.

The 2,000-turn compaction session behind the agent deployment guide's storage
numbers runs for minutes. `pytest` skips it unless
`DEX_AGENTS_LONG_TESTS=1`, and `ci.sh` runs an 800-turn guard with a retention
limit of 100 instead. The
`Agents Long Tests Nightly` workflow runs it every night. Run it by hand with:

```bash
cd agents/python/core
DEX_AGENTS_LONG_TESTS=1 DEXCLI=/path/to/dexcli uv run --frozen pytest -s tests/test_compaction.py -k 2000_turn
```

## Design decisions

The plan records each of these decisions too.

- Step mode: one `RunModel` Step per model call and one Step per tool call,
  READ included. No framework loop runs several model calls in one Step.
- Tool call Steps and their failure routes take the tool call ID as input.
  There is no shared "current tool call" in `AgentState`.
- `RecordToolFailure` records `failed` for a read and `outcome_unknown` for a
  write: it is reached only through crashes, timeouts, or exceptions other
  than `ToolExecutionError`, so a write may have taken effect.
- A failed tool Step whose recovery error reports `attempt_progress`
  `NOT_STARTED` never ran the tool, so its call records `failed` with
  `TOOL_CALL_NOT_RUN_ERROR_PREFIX` rather than a new status. The model sees an
  ordinary failure and can call the tool again.
- `AgentToolCall.deferral_order` keeps model order for display and for the
  turn-failure note; AttributeMap keys alone do not keep order.
- By default nothing from a turn is stored until it completes: `TurnCompleted`
  carries all of the turn's messages and `ToolCallsRequested` keeps them in the
  framework run state. A runner may store them incrementally with
  `ToolCallsRequested.new_messages`; Stepwise does.
- Built-in tools are routed by the session, not the runner, so they work under
  every adapter that offers their definitions.
- A failed turn that ran tools leaves a note message instead of appending
  nothing, so its writes are not repeated by the next turn.
- Every Step sets SYNC durability explicitly.
- Message sequences start at 1. A blank `initial_message` starts no turn.
  `last_turn_error` is cleared when a later turn completes.
- `resolve_interrupt` drops `rejection_message` on approval. When it resolves
  an interrupt just as its deadline Timer fires, `AwaitToolApproval` waits
  again and applies the answer.
- `AgentSnapshot.state` is `None` until `StartAgentSession` commits.
- Status, role, effect, and record status values are `StrEnum` types in Python;
  the wire carries the plain strings.

## Limits

- Without compaction, `RunModel` loads every stored message, and nothing is
  deleted. `get_agent_snapshot`, `list_agent_messages`, and a
  `RecordTurnFailure` of a runner that builds unfinished-turn messages load
  every kept chunk; retention bounds that load.
- Each model call and tool call adds a Step round trip.
- Steering never stops a model call or a tool call that has started; it
  applies at the next safe Step boundary. `cancel_turn` stops a model call
  but never a tool call Step that has started.
- Interrupt deadlines use the Worker clock.
- One `text_delta` call writes one Stream message; batch very small chunks.
- Tool call IDs that are empty or contain `/` fail the turn.
- The tool Steps use SYNC durability, and Dex never hands one attempt number
  to a second Worker, so a single-attempt tool does not run twice after a
  Worker crash. Cancellation is cooperative, though: a cancelled run may keep
  running while the failure route starts.
- `attempt_progress` `NOT_COMPLETED` is conservative, so a call whose tool
  never ran can still record `outcome_unknown`, such as when Dex Server
  crashed after it recorded the attempt start but before the task reached the
  Worker.
- A tool the framework cannot defer runs inside `RunModel` and repeats on a
  `RunModel` retry.
- Whether turn records cleared at turn end still count against the Server's
  `limits.maxFlowStateBytes` run budget is an open question in the parent
  design.
- `dex_agents.mcp` polls a stdio server's MCP task inside its tool Step for at
  most `task_wait_timeout`, holding a Worker slot, because the task ends with
  the server process.
- Each `PollToolTask` poll is a Step in the session's history; a task that
  runs for days adds up to a Step a minute while the server's interval is
  shorter.
- An MCP task that asks the identical request again within 10 seconds of an
  answer, without leaving `input_required` in between, waits until those 10
  seconds pass before the user is asked. A server slower than 10 seconds to
  take an answer in has its question asked twice.
- A retried `PollToolTask` attempt asks the user again rather than send an
  answer that its first attempt may have sent.
- An MCP `input_required` answer to `tools/call` outside a task records
  `failed`; mapping it to a session question interrupt is a follow-up.
- `dex_agents.mcp` starts a new stdio process, or opens a new HTTP connection,
  for every request. An unreachable server is tried again, with its connection
  attempts, on every `list_tools` call until it answers.
- MCP annotations are hints from the server and are trusted by default. A
  wrong `readOnlyHint` on a write makes that write run without approval and
  retry; set `trusts_tool_annotations=False` or a `ToolPolicy` for servers you
  do not control.
- Tracing covers the session's Steps only. The Dex SDK emits no spans, so Dex
  Server and Worker internals do not appear in the traces.
- The adapters report only the model information their framework exposes, so
  some `chat` spans carry no provider, response ID, or finish reasons. Each
  adapter's README lists what it sets.
