Keep default run_id and conversation_id stable when a durable run is re-executed (#9701)

Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
Douwe Maan
2026-10-09 09:42:00 -06:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 8a1c21aa8f
commit f87fcea390
25 changed files with 359 additions and 86 deletions
+2
View File
@@ -185,6 +185,8 @@ All other agents and toolsets are supported.
By default, DBOS checkpoints workflow inputs/outputs and step outputs into a database using [`pickle`](https://docs.python.org/3/library/pickle.html). But you can optionally supply a [custom serializer](https://docs.dbos.dev/python/reference/contexts#custom-serialization) through DBOS configuration. This means you need to make sure the [dependencies](../dependencies.md) object provided to [`Agent.run()`][pydantic_ai.agent.Agent.run] / [`Agent.run_sync()`][pydantic_ai.agent.Agent.run_sync], and tool outputs can be serialized.
You may also want to keep the inputs and outputs small (under \~2 MB). PostgreSQL and SQLite support up to 1 GB per field, but large objects may impact performance.
If you don't pass `run_id` or `conversation_id` to [`Agent.run()`][pydantic_ai.agent.Agent.run] in a workflow, the defaults are derived from the workflow ID rather than generated at random, so a recovered workflow keeps the IDs its completed steps already saw.
### Workspaces
Attach the [workspace](../workspace.md) capability, such as `LocalWorkspace`, when you construct the agent, and use `ctx.workspace` as in any run. Each workspace call made in workflow code, including from function tools, which DBOS runs in the workflow, is a step, so file contents and command output count toward the [size guidance above](#agent-run-context-and-dependencies).
+2
View File
@@ -327,6 +327,8 @@ Pydantic AI and provider API clients have their own retry logic. When using Pref
This prevents requests from being retried multiple times at different layers. The layers *multiply*: see [Retry multiplication](../retries.md#retry-multiplication) for the arithmetic.
If you don't pass `run_id` or `conversation_id` to [`Agent.run()`][pydantic_ai.agent.Agent.run] in a flow, the defaults are derived from the flow run rather than generated at random, so a flow retry gives each agent run the same IDs as the attempt before it.
## Caching and Idempotency
Prefect 3.0 provides built-in caching and transactional semantics. Tasks with identical inputs will not re-execute if their results are already cached, making workflows naturally idempotent and resilient to failures.
+2
View File
@@ -236,6 +236,8 @@ If you need one or more of these attributes to be available inside activities, y
The activity's `RunContext` is rebuilt from the serialized payload, so its fields are copies: mutating them inside an activity does not affect the run. In particular, `usage` is a snapshot of the run's usage at the time the activity was scheduled. If a tool [delegates to another agent](../multi-agent-applications.md#agent-delegation) with `usage=ctx.usage`, the delegate's tokens and requests stay behind in the activity: they're missing from the parent run's [`result.usage`][pydantic_ai.agent.AgentRunResult.usage] and are never charged against its [usage limits](../agent.md#usage-limits). To account for delegate usage, carry it yourself: return the delegate's [`result.usage`][pydantic_ai.agent.AgentRunResult.usage] from the tool, or record it in an external store your `deps` can reach. Temporal loses these mutations unconditionally. [DBOS](dbos.md) and [Prefect](prefect.md) pass the live `RunContext` into their in-process durable units, so mutations do accrue while a step or task body actually runs — but they're lost there too whenever the body doesn't run because its recorded result is replayed (DBOS workflow recovery) or reused (a Prefect task cache hit), which makes the same code account differently from one run to the next. Don't rely on the in-process engines' behavior; a return channel that works for all three is under discussion in [pydantic-ai#6886](https://github.com/pydantic/pydantic-ai/issues/6886).
If you don't pass `run_id` or `conversation_id` to [`Agent.run()`][pydantic_ai.agent.Agent.run] in a workflow, the defaults are derived from the workflow run rather than generated at random, so they stay the same when Temporal replays the workflow on another worker, and every activity of the run sees the same IDs as the run's result. Workflows started before Pydantic AI derived these defaults for agents without a workspace keep random IDs, so that their histories replay unchanged.
A tool's [`prepare`](../tools-advanced.md#tool-prepare) function is not affected by these limitations: for tools in a [`FunctionToolset`][pydantic_ai.toolsets.FunctionToolset] (including those defined on the agent itself), it runs in workflow code with the complete `RunContext`, once per run step like outside a workflow. The tool definition it returns is sent to the tool-call activity, which uses it as-is, so the tool the model saw is the tool that runs, down to its [`timeout`](../tools-advanced.md#tool-timeout). Tools from a `DynamicToolset` are the exception: as the toolset is re-resolved inside activities, their `prepare` functions run there as well and see the limited `RunContext`.
A `native=` factory on [`XSearch`][pydantic_ai.capabilities.XSearch] or [`ImageGeneration`][pydantic_ai.capabilities.ImageGeneration] is resolved twice, on either side of the boundary: once in workflow code to configure the native tool, and again inside the fallback subagent's tool-call activity, where it sees the limited `RunContext`. Read `ctx.deps` there, not `ctx.messages`.
+1 -1
View File
@@ -369,7 +369,7 @@ activities.
tool uses it; retries, replays and recovery reattach to that same environment. Plain runs stay
lazy. If a worker dies after creating the environment but before that is recorded, the retry can
create a second one.
- Without an explicit `run_id`, a run of an agent with a workspace capability inside a workflow or flow gets an ID derived from the Temporal
- Without an explicit `run_id`, a run inside a workflow or flow gets an ID derived from the Temporal
execution run ID, the DBOS workflow ID or the Prefect flow run ID, so its workspace state stays
addressable after a worker restart or flow retry.
- On DBOS, a run with a workspace runs its tool calls one at a time: DBOS numbers steps as they
+11 -9
View File
@@ -283,28 +283,29 @@ NEW_CONVERSATION: Literal['new'] = 'new'
def resolve_conversation_id(
explicit: str | None,
message_history: Sequence[_messages.ModelMessage] | None,
*,
default: str | None = None,
) -> str:
"""Resolve the `conversation_id` to use for an agent run.
Priority:
1. `explicit == 'new'` → fresh UUID7 (forks a conversation off the supplied history).
1. `explicit == 'new'` → fresh id (forks a conversation off the supplied history).
2. Explicit string → used as-is.
3. Most recent non-`None` `conversation_id` on `message_history` (scanned from the end).
4. Fresh UUID7.
4. Fresh id.
A fresh UUID7 is intentionally distinct from the run's `run_id`, so callers can
A fresh id is `default` when given (a durable engine's replay-stable id), and a UUID7
otherwise. Either way it is distinct from the run's `run_id`, so callers can
treat the two identifiers as independent.
"""
if explicit == NEW_CONVERSATION:
return str(uuid7())
if explicit is not None:
if explicit is not None and explicit != NEW_CONVERSATION:
return explicit
if message_history:
if explicit is None and message_history:
for message in reversed(message_history):
if (cid := message.conversation_id) is not None:
return cid
return str(uuid7())
return default if default is not None else str(uuid7())
def resolve_conversation(
@@ -355,7 +356,8 @@ def resolve_run_id(
Priority:
1. Explicit string → used as-is (raises `UserError` if empty, or if that id already
appears on `message_history`).
appears on `message_history`). `Agent` passes a durable engine's replay-stable
default here when the caller gave none.
2. Fresh UUID7.
"""
if explicit is not None:
+8 -2
View File
@@ -237,13 +237,19 @@ class RunContext(Generic[RunContextAgentDepsT]):
partial_output: bool = False
"""Whether the output passed to an output validator is partial."""
run_id: str | None = None
""""Unique identifier for the agent run."""
"""Unique identifier for the agent run.
The `run_id` argument to `Agent.run` (etc.), or a fresh UUID7. Inside a durable workflow or flow,
the default is derived from the workflow or flow run instead, so it stays the same when the engine
re-executes the run.
"""
conversation_id: str | None = None
"""Unique identifier for the conversation this run belongs to.
A conversation spans potentially multiple agent runs that share message history.
Resolved at the start of `Agent.run` (etc.) from the explicit `conversation_id`
argument, the most recent `conversation_id` on `message_history`, or a fresh UUID7.
argument, the most recent `conversation_id` on `message_history`, or a fresh UUID7
(inside a durable workflow or flow, a UUID derived from `run_id`).
"""
metadata: dict[str, Any] | None = None
"""Metadata associated with this agent run, if configured."""
+12 -6
View File
@@ -1409,8 +1409,8 @@ class Agent(AbstractAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -1667,17 +1667,23 @@ class Agent(AbstractAgent[AgentDepsT, OutputDataT]):
# Build the graph
graph = _agent_graph.build_agent_graph(self.name, self._deps_type, output_type_)
# Build the initial state
# Build the initial state. Inside a durable workflow or flow, the durability capability supplies
# the default ids, so a re-execution of this code gets the same ones.
resolved_run_id = _agent_graph.resolve_run_id(
run_id if run_id is not None else bootstrap_capability._default_run_id(), # pyright: ignore[reportPrivateUsage]
message_history,
)
state = _agent_graph.GraphAgentState(
message_history=list(message_history) if message_history else [],
usage=usage,
output_retries_used=0,
run_step=0,
run_id=_agent_graph.resolve_run_id(
run_id if run_id is not None else bootstrap_capability._default_run_id(), # pyright: ignore[reportPrivateUsage]
run_id=resolved_run_id,
conversation_id=_agent_graph.resolve_conversation_id(
conversation_id,
message_history,
default=bootstrap_capability._default_conversation_id(resolved_run_id), # pyright: ignore[reportPrivateUsage]
),
conversation_id=_agent_graph.resolve_conversation_id(conversation_id, message_history),
)
historical_response = next(
(message for message in reversed(state.message_history) if isinstance(message, _messages.ModelResponse)),
+12 -12
View File
@@ -580,8 +580,8 @@ class AbstractAgent(Generic[AgentDepsT, OutputDataT], ABC):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -791,8 +791,8 @@ class AbstractAgent(Generic[AgentDepsT, OutputDataT], ABC):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -966,8 +966,8 @@ class AbstractAgent(Generic[AgentDepsT, OutputDataT], ABC):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -1305,8 +1305,8 @@ class AbstractAgent(Generic[AgentDepsT, OutputDataT], ABC):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
deps: Optional dependencies to use for this run.
model_settings: Optional settings to use for this model's request, or a callable
@@ -1495,8 +1495,8 @@ class AbstractAgent(Generic[AgentDepsT, OutputDataT], ABC):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -1709,8 +1709,8 @@ class AbstractAgent(Generic[AgentDepsT, OutputDataT], ABC):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -295,8 +295,8 @@ class WrapperAgent(AbstractAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -559,6 +559,11 @@ class AbstractCapability(ABC, Generic[AgentDepsT]):
return None
def _default_run_id(self) -> str | None:
"""A replay-stable `run_id` for a run started without one, or `None` to generate a UUID7."""
return None
def _default_conversation_id(self, run_id: str) -> str | None:
"""A replay-stable `conversation_id` for a run that has none to inherit, or `None` to generate a UUID7."""
return None
def get_model(self) -> AgentModel[AgentDepsT] | None:
@@ -496,10 +496,15 @@ class CombinedCapability(AbstractCapability[AgentDepsT]):
return tool_defs
def _default_run_id(self) -> str | None:
if self._has_get_workspace:
for capability in reversed(self.capabilities):
if (run_id := capability._default_run_id()) is not None:
return run_id
for capability in reversed(self.capabilities):
if (run_id := capability._default_run_id()) is not None:
return run_id
return None
def _default_conversation_id(self, run_id: str) -> str | None:
for capability in reversed(self.capabilities):
if (conversation_id := capability._default_conversation_id(run_id)) is not None:
return conversation_id
return None
# --- Run lifecycle hooks ---
@@ -267,6 +267,12 @@ class WrapperCapability(AbstractCapability[AgentDepsT]):
def _prepare_workspace(self, ctx: RunContext[AgentDepsT], workspace: Workspace, *, explicit: bool) -> Workspace:
return self.wrapped._prepare_workspace(ctx, workspace, explicit=explicit)
def _default_run_id(self) -> str | None:
return self.wrapped._default_run_id()
def _default_conversation_id(self, run_id: str) -> str | None:
return self.wrapped._default_conversation_id(run_id)
async def prepare_tools(
self,
ctx: RunContext[AgentDepsT],
@@ -1,6 +1,7 @@
from __future__ import annotations
import copy
import uuid
from abc import abstractmethod
from collections.abc import (
AsyncGenerator,
@@ -174,6 +175,18 @@ class _RestrictedRunContext(Protocol):
MODEL_RESPONSE_STREAM_EVENT_TYPES = get_union_args(ModelResponseStreamEvent)
_CONVERSATION_ID_NAMESPACE = uuid.UUID('b4a5470b-7dc5-4744-b6d4-27f7c81a3775')
"""Namespace for the UUID5 a durable run derives its default `conversation_id` from."""
def conversation_id_from_run_id(run_id: str) -> str:
"""Derive a run's default `conversation_id` from its `run_id`.
For engines whose default `run_id` is replay-stable, so a re-execution of the run gets the same
conversation ID without another draw. The derivation is also stable for a caller-supplied `run_id`.
"""
return str(uuid.uuid5(_CONVERSATION_ID_NAMESPACE, run_id))
class _BoundModelOperations(NamedTuple):
request: BoundDurableOperation[ModelRequestParams, Any, ModelResponse]
@@ -496,8 +496,8 @@ class DBOSAgent(WrapperAgent[AgentDepsT, OutputDataT], DBOSConfiguredInstance):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -660,8 +660,8 @@ class DBOSAgent(WrapperAgent[AgentDepsT, OutputDataT], DBOSConfiguredInstance):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -822,8 +822,8 @@ class DBOSAgent(WrapperAgent[AgentDepsT, OutputDataT], DBOSConfiguredInstance):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -1001,8 +1001,8 @@ class DBOSAgent(WrapperAgent[AgentDepsT, OutputDataT], DBOSConfiguredInstance):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -1192,8 +1192,8 @@ class DBOSAgent(WrapperAgent[AgentDepsT, OutputDataT], DBOSConfiguredInstance):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -4,6 +4,7 @@ from collections.abc import Mapping
from contextvars import ContextVar
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, cast
from weakref import WeakKeyDictionary
from dbos import DBOS
from dbos._error import DBOSWorkflowCancelledError, DBOSWorkflowConflictIDError
@@ -11,7 +12,7 @@ from dbos._error import DBOSWorkflowCancelledError, DBOSWorkflowConflictIDError
from pydantic_ai.agent import EventStreamHandler, ParallelExecutionMode
from pydantic_ai.agent.abstract import AbstractAgent
from pydantic_ai.capabilities.abstract import WrapRunHandler
from pydantic_ai.durable_exec._base import BaseDurabilityCapability
from pydantic_ai.durable_exec._base import BaseDurabilityCapability, conversation_id_from_run_id
from pydantic_ai.durable_exec._codec import IDENTITY_CODEC
from pydantic_ai.durable_exec._operation import ToolsetKind
from pydantic_ai.durable_exec._spec import DurabilityEngineSpec
@@ -26,7 +27,14 @@ from ._operation_backend import DBOSBoundOperation, DBOSOperationBackend, DBOSOp
from ._utils import StepConfig, guard_enqueue_in_workflow
if TYPE_CHECKING:
pass
from dbos._context import DBOSContext
_default_run_ids_by_step: WeakKeyDictionary[DBOSContext, dict[int, int]] = WeakKeyDictionary()
"""Agent runs that drew a default `run_id` at each step position, per workflow execution.
Runs started concurrently (e.g. with `asyncio.gather`) resolve their IDs before either starts a step,
so they share a position and need the count to tell them apart.
"""
@dataclass(init=False, kw_only=True)
@@ -220,7 +228,15 @@ class DBOSDurability(BaseDurabilityCapability[AgentDepsT]):
context = get_local_dbos_context()
assert context is not None and context.workflow_id is not None
return f'{context.workflow_id}:{context.function_id + 1}'
step = context.function_id + 1
runs = _default_run_ids_by_step.setdefault(context, {})
index = runs.get(step, 0)
runs[step] = index + 1
# The first run at a position keeps the ID it had before concurrent runs were told apart.
return f'{context.workflow_id}:{step}' if index == 0 else f'{context.workflow_id}:{step}:{index}'
def _default_conversation_id(self, run_id: str) -> str | None:
return conversation_id_from_run_id(run_id) if self.in_durable_context else None
def _durable_run_context(self, ctx: RunContext[AgentDepsT]) -> RunContext[AgentDepsT]:
# A DBOS step degrades to a plain inline call outside a workflow, where enqueueing is
@@ -358,8 +358,8 @@ class PrefectAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -532,8 +532,8 @@ class PrefectAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -707,8 +707,8 @@ class PrefectAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -886,8 +886,8 @@ class PrefectAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -1103,8 +1103,8 @@ class PrefectAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
deps: Optional dependencies to use for this run.
instructions: Optional additional instructions to use for this run.
@@ -7,7 +7,7 @@ from typing import Any, Literal, cast
from prefect.context import FlowRunContext
from pydantic_ai.agent import EventStreamHandler
from pydantic_ai.durable_exec._base import BaseDurabilityCapability
from pydantic_ai.durable_exec._base import BaseDurabilityCapability, conversation_id_from_run_id
from pydantic_ai.durable_exec._codec import IDENTITY_CODEC
from pydantic_ai.durable_exec._operation import (
DurableOperationId,
@@ -114,12 +114,15 @@ class PrefectDurability(BaseDurabilityCapability[AgentDepsT]):
if context is None:
return None
assert context.flow_run is not None
key = 'pydantic_ai:workspace_run_id'
key = 'pydantic_ai:default_run_id'
sequence = context.task_run_dynamic_keys.get(key, 0)
assert isinstance(sequence, int)
context.task_run_dynamic_keys[key] = sequence + 1
return f'{context.flow_run.id}:{sequence}'
def _default_conversation_id(self, run_id: str) -> str | None:
return conversation_id_from_run_id(run_id) if self.in_durable_context else None
def get_durable_operation_backend(self) -> DurableOperationBackend[TaskConfig]:
def tool_config(
operation_id: DurableOperationId, tool: object | None, tool_name: str
@@ -488,8 +488,8 @@ class TemporalAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
Inside workflows, only registered model instances, registered names, or provider strings are valid.
instructions: Optional additional instructions to use for this run.
@@ -655,8 +655,8 @@ class TemporalAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -815,8 +815,8 @@ class TemporalAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -994,8 +994,8 @@ class TemporalAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -1210,8 +1210,8 @@ class TemporalAgent(WrapperAgent[AgentDepsT, OutputDataT]):
`conversation_id` separately. Passing both raises `UserError`.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -19,6 +19,7 @@ from pydantic_ai.capabilities.abstract import AbstractCapability, CapabilityOrde
from pydantic_ai.durable_exec._base import (
MODEL_RESPONSE_STREAM_EVENT_TYPES,
BaseDurabilityCapability,
conversation_id_from_run_id,
)
from pydantic_ai.durable_exec._capability_operation import CapabilityMethodDeclaration
from pydantic_ai.durable_exec._codec import IDENTITY_CODEC
@@ -79,6 +80,9 @@ from ._transports import (
_WorkspaceCallTransport,
)
_STABLE_DEFAULT_RUN_ID_PATCH = 'pydantic_ai:stable_default_run_id'
"""`workflow.patched()` ID marking histories in which every agent run drew its default `run_id`."""
_DEFAULT_MODEL_HEARTBEAT_TIMEOUT = timedelta(seconds=30)
"""Default `heartbeat_timeout` for the model-request activities.
@@ -493,8 +497,23 @@ class TemporalDurability(BaseDurabilityCapability[AgentDepsT]):
def _default_run_id(self) -> str | None:
if not self.in_durable_context:
return None
# `workflow.uuid4()` draws from the workflow's random sequence. Histories recorded before every
# durable run got a stable default only drew here when the root capability supplied workspaces
# and reached this capability directly (not through a wrapper), so the rest replay without the
# draw and keep the sequence (and the random run ID) they were recorded with.
root = self._agent.root_capability if self._agent is not None else None
drew_before = (
root is not None
and root._has_get_workspace # pyright: ignore[reportPrivateUsage]
and any(capability is self for capability in root.capabilities)
)
if not drew_before and not workflow.patched(_STABLE_DEFAULT_RUN_ID_PATCH):
return None
return f'{workflow.info().run_id}:{workflow.uuid4()}'
def _default_conversation_id(self, run_id: str) -> str | None:
return conversation_id_from_run_id(run_id) if self.in_durable_context else None
async def wrap_run(
self,
ctx: RunContext[AgentDepsT],
+6 -6
View File
@@ -598,8 +598,8 @@ class UIAdapter(ABC, Generic[RunInputT, MessageT, EventT, AgentDepsT, OutputData
output validators since output validators would expect an argument that matches the agent's output type.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -713,8 +713,8 @@ class UIAdapter(ABC, Generic[RunInputT, MessageT, EventT, AgentDepsT, OutputData
output validators since output validators would expect an argument that matches the agent's output type.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
@@ -803,8 +803,8 @@ class UIAdapter(ABC, Generic[RunInputT, MessageT, EventT, AgentDepsT, OutputData
output validators since output validators would expect an argument that matches the agent's output type.
message_history: History of the conversation so far.
deferred_tool_results: Optional results for deferred tool calls in the message history.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a workspace capability, run inside a Temporal workflow, DBOS workflow or Prefect flow, gets one derived from the workflow or flow run so its workspace state survives worker recovery.
conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the `run_id` so it stays the same when the engine re-executes the run.
run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated, except that an agent with a `TemporalDurability`, `DBOSDurability` or `PrefectDurability` capability, run inside a workflow or flow, gets one derived from the workflow or flow run so it stays the same when the engine re-executes the run.
model: Optional model to use for this run, required if `model` was not set when creating the agent.
instructions: Optional additional instructions to use for this run.
deps: Optional dependencies to use for this run.
+105 -1
View File
@@ -559,6 +559,106 @@ async def test_durability_run_context_in_durable_context(client: Client):
assert result.output == 'workflow: False, activity: False'
# --- Default `run_id` and `conversation_id` ---
def _stable_ids_model_fn(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
returns = sum(isinstance(part, ToolReturnPart) for message in messages for part in message.parts)
if returns < 2:
return ModelResponse(parts=[ToolCallPart(tool_name='record_ids', args='{}', tool_call_id=f'call-{returns}')])
return ModelResponse(parts=[TextPart(content='done')])
_stable_ids_seen: list[tuple[str | None, str | None]] = []
async def record_ids(ctx: RunContext[object]) -> str:
_stable_ids_seen.append((ctx.run_id, ctx.conversation_id))
return 'ok'
_stable_ids_agent = Agent(
FunctionModel(_stable_ids_model_fn),
name='durability_stable_ids',
toolsets=[FunctionToolset(tools=[record_ids], id='stable_ids')],
capabilities=[TemporalDurability(activity_config=BASE_ACTIVITY_CONFIG)],
)
_stable_ids_later_draws: list[str] = []
@workflow.defn
class StableIdsWorkflow:
@workflow.run
async def run(self, prompt: str) -> list[str]:
result = await _stable_ids_agent.run(prompt)
# What the workflow's own code draws next depends on how many draws the agent run made.
_stable_ids_later_draws.append(str(workflow.uuid4()))
return [result.run_id, result.conversation_id]
async def test_durability_default_ids_survive_replay(client: Client):
"""Without `run_id=` or `conversation_id=`, every replay of the workflow resolves the same IDs.
With workflow caching off, the worker replays the workflow from history for each activity result,
so before the IDs were replay-stable, each tool activity saw a different `run_id`.
"""
_stable_ids_seen.clear()
async with Worker(
client,
task_queue=TASK_QUEUE,
workflows=[StableIdsWorkflow],
plugins=[AgentPlugin(_stable_ids_agent)],
max_cached_workflows=0,
):
handle = await client.start_workflow(
StableIdsWorkflow.run, args=['Hello'], id=f'stable-ids-{uuid.uuid4()}', task_queue=TASK_QUEUE
)
run_id, conversation_id = await handle.result()
assert _stable_ids_seen == [(run_id, conversation_id)] * 2
assert run_id.startswith(f'{handle.result_run_id}:')
assert conversation_id != run_id
async def test_durability_replays_history_recorded_before_stable_default_ids(
client: Client, monkeypatch: pytest.MonkeyPatch
):
"""A history recorded before every run drew a default `run_id` replays without drawing one.
Unsandboxed, so the module-level list sees the workflow's draws in both the original run and the replay.
"""
_stable_ids_later_draws.clear()
# Without the patch marker, the workflow runs the way it did before: no draw from the workflow's
# random sequence, and a random `run_id`.
def unpatched(patch_id: str) -> bool:
return False
monkeypatch.setattr(workflow, 'patched', unpatched)
async with Worker(
client,
task_queue=TASK_QUEUE,
workflows=[StableIdsWorkflow],
plugins=[AgentPlugin(_stable_ids_agent)],
workflow_runner=UnsandboxedWorkflowRunner(),
):
handle = await client.start_workflow(
StableIdsWorkflow.run, args=['Hello'], id=f'legacy-ids-{uuid.uuid4()}', task_queue=TASK_QUEUE
)
run_id, _ = await handle.result()
assert ':' not in run_id
monkeypatch.undo()
history = await handle.fetch_history()
replay = await Replayer(
workflows=[StableIdsWorkflow], plugins=[PydanticAIPlugin()], workflow_runner=UnsandboxedWorkflowRunner()
).replay_workflow(history)
assert replay.replay_failure is None
# The replay drew nothing for the run's ID, so the workflow's next draw matches the recorded one.
assert len(_stable_ids_later_draws) == 2 and _stable_ids_later_draws[0] == _stable_ids_later_draws[1]
# --- A capability operation called from the capability's own tool ---
@@ -1522,7 +1622,11 @@ async def test_durability_validates_only_resolved_runtime_capability_layers():
capabilities=[base_factory, WrapperCapability(wrapped=TemporalDurability())],
)
with patch('pydantic_ai.durable_exec.temporal._durability.workflow.in_workflow', return_value=True):
# Only `in_workflow` is faked, so keep the default-ID hook off the real workflow APIs.
with (
patch('pydantic_ai.durable_exec.temporal._durability.workflow.in_workflow', return_value=True),
patch('pydantic_ai.durable_exec.temporal._durability.workflow.patched', return_value=False),
):
result = await agent.run('hello', capabilities=[Instrumentation(InstrumentationSettings())])
assert result.output == 'skipped'
@@ -253,6 +253,20 @@ async def test_workspace_calls_from_workflow_code_are_activities_and_replay_disp
assert provider.log == log
async def test_workspace_runs_draw_their_run_id_without_the_stable_id_patch(client: Client) -> None:
"""Workspace agents drew a default `run_id` before every durable run did, so they draw without the patch.
Their histories from before therefore replay unchanged, and new ones carry no patch marker.
"""
provider.reset()
(generated, _, engine_id), history = await _execute(client, 'run_id')
assert generated.startswith(f'{engine_id.split(":")[1]}:')
assert not any(event.HasField('marker_recorded_event_attributes') for event in history.events)
replay = await Replayer(workflows=[ScenarioWorkflow], plugins=[PydanticAIPlugin()]).replay_workflow(history)
assert replay.replay_failure is None
# --- The activity side on its own ---------------------------------------------------------------
+41
View File
@@ -4838,3 +4838,44 @@ async def test_dbos_decide_span_nests_under_chat(
lineage, attributes = decide_span_lineage(capfire.exporter.exported_spans_as_dict())
assert lineage[:3] == snapshot(['dbos_decide__model.request', 'chat ship-it', 'invoke_agent dbos_decide'])
assert attributes['pydantic_ai.decision.state'] == 'The migration is reviewed and the tests pass.'
stable_ids_agent = Agent(TestModel(), name='stable_ids_agent', capabilities=[DBOSDurability()])
stable_ids_seen: list[tuple[str | None, str | None]] = []
class _ProcessCrash(BaseException):
"""Stands in for the process dying: DBOS leaves the workflow `PENDING` for recovery."""
@stable_ids_agent.tool
def record_ids(ctx: RunContext[object]) -> str:
stable_ids_seen.append((ctx.run_id, ctx.conversation_id))
return 'ok'
@DBOS.workflow()
async def stable_ids_workflow() -> list[tuple[str, str]]:
first = await stable_ids_agent.run('First.')
# Concurrent runs resolve their IDs before either starts a step.
results = [first, *await asyncio.gather(stable_ids_agent.run('Second.'), stable_ids_agent.run('Third.'))]
if len(stable_ids_seen) == 3:
raise _ProcessCrash
return [(result.run_id, result.conversation_id) for result in results]
async def test_dbos_default_ids_survive_recovery(dbos: DBOS) -> None:
"""Without `run_id=` or `conversation_id=`, a recovered workflow keeps the IDs its steps already saw."""
stable_ids_seen.clear()
workflow_id = f'stable-ids-{uuid.uuid4()}'
with SetWorkflowID(workflow_id), pytest.raises(_ProcessCrash):
await stable_ids_workflow()
seen = sorted(stable_ids_seen)
assert [run_id for run_id, _ in seen] == [f'{workflow_id}:1', f'{workflow_id}:3', f'{workflow_id}:3:1']
assert len({*seen[0], *seen[1], *seen[2]}) == 6
# Recovery re-executes the workflow function: model requests replay from their recorded steps, while
# the function tool, which runs in the workflow, runs again and sees the same IDs.
handle = await asyncio.to_thread(DBOS._execute_workflow_id, workflow_id) # pyright: ignore[reportPrivateUsage]
assert sorted(tuple(ids) for ids in await asyncio.to_thread(handle.get_result)) == seen
assert sorted(stable_ids_seen[3:]) == seen
@@ -12,6 +12,7 @@ from __future__ import annotations
import errno
import pickle
import uuid
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import Any
@@ -129,6 +130,13 @@ class FreshWorkspaces(AbstractCapability[Any]):
return backend
async def test_an_engine_without_default_ids_keeps_fresh_uuid7s() -> None:
"""An engine built on the public base supplies no replay-stable `run_id`, so the defaults stay UUID7s."""
agent = Agent(TestModel(), name='engine_without_default_ids', capabilities=[FakeDurability()])
result = await agent.run('Hello.')
assert [uuid.UUID(result.run_id).version, uuid.UUID(result.conversation_id).version] == [7, 7]
def _run_context() -> RunContext[None]:
return RunContext(deps=None, model=TestModel(), usage=RunUsage())
+27 -8
View File
@@ -17,7 +17,7 @@ import pytest
from inline_snapshot import snapshot
from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import LocalWorkspace
from pydantic_ai.capabilities import LocalWorkspace, WrapperCapability
from pydantic_ai.durable_exec._workspace import WorkspaceCall
from pydantic_ai.models.test import TestModel
from pydantic_ai.workspaces import WorkspaceTimeoutError
@@ -139,20 +139,39 @@ async def test_prefect_flow_retry_replays_workspace_tasks_and_run_ids() -> None:
)
async def test_prefect_run_ids_without_a_workspace_are_fresh_uuid7s() -> None:
async def test_prefect_default_ids_without_a_workspace_survive_a_flow_retry() -> None:
agent = Agent(TestModel(), name='prefect_run_id', capabilities=[PrefectDurability()])
ids: list[str] = []
attempts: list[list[tuple[str, str]]] = []
@flow(retries=1)
async def run() -> None:
ids.append((await agent.run('hello')).run_id)
if len(ids) == 1:
results = [await agent.run('First.'), await agent.run('Second.')]
attempts.append([(result.run_id, result.conversation_id) for result in results])
if len(attempts) == 1:
raise RuntimeError('retry')
await run()
# Without a workspace, a run inside a flow gets the same fresh UUID7 as outside one.
assert [uuid.UUID(run_id).version for run_id in ids] == [7, 7]
assert ids[0] != ids[1]
# Each agent run in the flow gets its own IDs, and the retry gives each the same ones again.
first, second = attempts[0]
assert attempts[1] == attempts[0]
assert len({*first, *second}) == 4
assert first[0].endswith(':0') and second[0].endswith(':1')
# Outside a flow, the agent keeps generating fresh UUID7s.
outside = await agent.run('Outside.')
assert [uuid.UUID(outside.run_id).version, uuid.UUID(outside.conversation_id).version] == [7, 7]
async def test_prefect_default_ids_reach_through_a_wrapped_durability_capability() -> None:
agent = Agent(TestModel(), name='prefect_wrapped_run_id', capabilities=[WrapperCapability(PrefectDurability())])
@flow
async def run() -> tuple[str, str]:
result = await agent.run('Hello.')
return result.run_id, result.conversation_id
run_id, conversation_id = await run()
assert run_id.endswith(':0')
assert uuid.UUID(conversation_id).version == 5
async def test_prefect_repeated_identical_reads_are_not_served_from_cache() -> None: