This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch openai-agents-durable
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/openai-agents-durable by this
push:
new 7715f05b85d Run a failed OpenAI Agents tool call again on retry
instead of replaying its error
7715f05b85d is described below
commit 7715f05b85d226dcca2d4d68ee05755981bd4914
Author: Kaxil Naik <[email protected]>
AuthorDate: Fri Oct 16 16:10:59 2026 +0100
Run a failed OpenAI Agents tool call again on retry instead of replaying
its error
A function tool's exception that failure_error_function turned into text was
recorded as the tool's result, so a retry replayed the error and never ran
the
tool again. Such a call is no longer recorded, as a call that raised is not.
The guide now splits what a retry replays from what always runs live, names
the journal's storage, and the tests cover unreadable recordings, structured
results, sessions, and a tool-raised ModelBehaviorError.
---
.../common/ai/docs/frameworks/openai_agents.rst | 72 ++++++-----
.../providers/common/ai/durable/openai_agents.py | 133 +++++++++++++--------
.../unit/common/ai/durable/test_openai_agents.py | 116 +++++++++++++++++-
3 files changed, 235 insertions(+), 86 deletions(-)
diff --git a/providers/common/ai/docs/frameworks/openai_agents.rst
b/providers/common/ai/docs/frameworks/openai_agents.rst
index 1fcc1c17c56..a704a4298d9 100644
--- a/providers/common/ai/docs/frameworks/openai_agents.rst
+++ b/providers/common/ai/docs/frameworks/openai_agents.rst
@@ -152,7 +152,9 @@ again from the prompt: every model call is paid for again,
and every tool call h
again. On a task with ``retries``, run the agent with
:func:`~airflow.providers.common.ai.durable.openai_agents.run_durably_sync`, or
:func:`~airflow.providers.common.ai.durable.openai_agents.run_durably` from
async code,
-instead. They take the same arguments as ``Runner.run_sync`` and
``Runner.run``:
+instead. They take the same arguments as ``Runner.run_sync`` and
``Runner.run``, except
+``previous_response_id``, ``conversation_id`` and
``auto_previous_response_id``, which they
+refuse: with them OpenAI keeps the conversation, and a replayed call never
reaches it.
.. exampleinclude::
/../../ai/src/airflow/providers/common/ai/example_dags/example_openai_agents_durable.py
:language: python
@@ -161,40 +163,52 @@ instead. They take the same arguments as
``Runner.run_sync`` and ``Runner.run``:
Each model call and function tool call is recorded in the task instance's
durable
journal as it completes, the journal ``AgentOperator(durable=True)`` uses (see
-:doc:`../durable_execution`). On a retry the agent runs again from the start,
and each
-call that asks what the failed try asked, in the same order, returns what it
returned
-then. From the first call that differs, that call and every call after it run
live. Once
-the run succeeds, what was recorded is deleted, so clearing the task later
starts over.
+:doc:`../durable_execution`). On Airflow 3.3 and later the journal is kept in
the task
+state store; on earlier versions, set ``[common.ai] durable_cache_path`` to
where it is
+kept. Once the run succeeds, what was recorded is deleted, so clearing the
task later
+starts over. Outside a task there is nothing to replay into, and the agent
runs as
+``Runner.run`` runs it.
-A retry runs some work again:
+What a retry replays
+^^^^^^^^^^^^^^^^^^^^
-- A call that was running when the try failed runs again, so tools that change
data
- must still be safe to repeat.
+On a retry the agent runs again from the start, and each call that asks what
the failed
+try asked, in the same order, returns what it returned then: the calls of
every model
+given by name or in ``run_config``, the model and function tools of the
starting agent
+and of the agents in its ``handoffs`` list, and an agent used through
``as_tool``, as one
+call. Recorded tool results pass through Airflow's secret masker before they
are saved,
+and a replayed model call reports no token usage.
+
+From the first call that differs, that call and every call after it run live:
+
+- A retry whose prompt, instructions, model settings or tools differ from the
failed
+ try's runs everything live from the first call that sees the difference.
- When the run failed because of the model's answers (``MaxTurnsExceeded``,
``ModelBehaviorError``, ``ModelRefusalError``, or an output or tool input
guardrail),
replaying those answers would fail it the same way, so the retry asks the
model again
- from its last request. A ``ModelBehaviorError`` that a tool raises, such as
arguments
- that fail validation with ``failure_error_function=None``, counts as the
tool's failure
- instead: the tool call runs again, but the answer that asked for it replays.
-- A retry whose prompt, instructions, model settings or tools differ from the
failed
- try's runs everything live from the first call that sees the difference.
-- These are not recorded and run live on every try: tools other than function
tools
- (hosted tools such as web search, the tools of ``mcp_servers``, shell and
computer
- tools); a function tool whose ``needs_approval`` is a function, since the
SDK checks the
- arguments it approved only through its own invoker; the function tools of an
agent
- reached through ``handoff()``, and its model unless that is given by name or
left to
- the default; and a streamed run. An agent used through ``as_tool`` is
recorded as one
- call.
-- A ``session`` makes every retry run live: the SDK adds the failed try's
prompt to it
+ from its last request.
+
+What always runs live
+^^^^^^^^^^^^^^^^^^^^^
+
+- A call that was running when the try failed, so tools that change data must
still be
+ safe to repeat.
+- A tool call that failed, including one whose exception
``failure_error_function`` turned
+ into text for the model. A ``ModelBehaviorError`` that a tool raises, such
as arguments
+ that fail validation with ``failure_error_function=None``, counts as the
tool's failure:
+ the tool call runs again, but the answer that asked for it replays.
+- Tools other than function tools: hosted tools such as web search, the tools
of
+ ``mcp_servers``, and shell and computer tools.
+- A function tool whose ``needs_approval`` is a function, since the SDK checks
the
+ arguments it approved only through its own invoker.
+- The function tools of an agent reached through ``handoff()``, and its model
too unless
+ the agent's ``model`` is a model name or not set.
+- A streamed run.
+- Every call, when the run has a ``session``: the SDK adds the failed try's
prompt to it
before the first model call, so that call no longer asks what it asked then.
-- A ``SandboxToolset`` without ``attach_to`` is destroyed when the try ends,
so the retry
- starts with a new, empty sandbox that the replayed conversation does not
describe.
-
-Recorded tool results pass through Airflow's secret masker before they are
saved, and a
-replayed model call reports no token usage. ``run_durably`` refuses
-``previous_response_id``, ``conversation_id`` and
``auto_previous_response_id``: with them
-OpenAI keeps the conversation, and a replayed call never reaches it. Outside a
task there
-is nothing to replay into, and it runs the agent as ``Runner.run`` does.
+
+A ``SandboxToolset`` without ``attach_to`` is destroyed when the try ends, so
the retry
+starts with a new, empty sandbox that the replayed conversation does not
describe.
Differences from ``AgentOperator``
----------------------------------
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/durable/openai_agents.py
b/providers/common/ai/src/airflow/providers/common/ai/durable/openai_agents.py
index bc0af085627..1224565c184 100644
---
a/providers/common/ai/src/airflow/providers/common/ai/durable/openai_agents.py
+++
b/providers/common/ai/src/airflow/providers/common/ai/durable/openai_agents.py
@@ -53,6 +53,11 @@ try:
ToolInputGuardrailTripwireTriggered,
)
from agents.items import TResponseOutputItem
+ from agents.tool import (
+ default_tool_error_function,
+ resolve_function_tool_failure_error_function,
+ set_function_tool_failure_error_function,
+ )
except ImportError as e:
raise AirflowOptionalProviderFeatureException(e)
@@ -91,6 +96,10 @@ _MODEL_AT_FAULT = (
# The tool results the SDK hands the model as structured content rather than
as text.
_STRUCTURED_OUTPUTS = (ToolOutputText, ToolOutputImage, ToolOutputFileContent)
+# Set on a tool call's context by the SDK when its default
failure_error_function turned the
+# tool's exception into the text the model sees.
+_DEFAULT_FAILURE_HANDLED = "_function_tool_default_failure_handled"
+
# Arguments that keep the conversation on OpenAI's servers, which a replayed
model call never reaches.
_SERVER_SIDE_STATE = ("previous_response_id", "conversation_id",
"auto_previous_response_id")
@@ -133,9 +142,11 @@ async def run_durably(
other than function tools (hosted tools such as web search, the tools of
``mcp_servers``, shell and computer tools), a function tool whose
``needs_approval`` is
a function, since the SDK checks its arguments only through its own
invoker, the
- function tools of an agent reached through ``handoff()`` and its model
unless that is
- given by name or left to the default, and streamed runs. A replayed model
call reports
- no usage, since it costs nothing.
+ function tools of an agent reached through ``handoff()``, and its model
too unless the
+ agent's ``model`` is a model name or not set, and streamed runs. A tool
call that fails,
+ including one whose exception ``failure_error_function`` turns into text
for the model,
+ is not recorded either, so a retry runs it again. A replayed model call
reports no
+ usage, since it costs nothing.
A ``session`` makes every retry run live: the SDK adds the failed try's
prompt to it
before the first model call, so that call no longer asks what it asked
then.
@@ -150,8 +161,7 @@ async def run_durably(
``auto_previous_response_id`` is passed: a replayed call never reaches
the
conversation OpenAI keeps for them.
"""
- journal = current_journal()
- if journal is None:
+ if (journal := current_journal()) is None:
return await Runner.run(starting_agent, input, run_config=run_config,
**kwargs)
if stateful := [name for name in _SERVER_SIDE_STATE if kwargs.get(name)]:
raise ValueError(
@@ -163,7 +173,7 @@ async def run_durably(
config.model_provider = _DurableModelProvider(config.model_provider,
durable_run)
if isinstance(config.model, Model):
config.model = _DurableModel(config.model, durable_run)
- agent = _durable_agent(starting_agent, durable_run, {})
+ agent = _get_durable_agent(starting_agent, durable_run, {})
try:
result = await Runner.run(agent, input, run_config=config, **kwargs)
except BaseException as error:
@@ -185,32 +195,46 @@ def run_durably_sync(
return run_coroutine_sync(run_durably(starting_agent, input,
run_config=run_config, **kwargs))
-def _durable_agent(agent: Agent[Any], durable_run: DurableRun, seen: dict[int,
Agent[Any]]) -> Agent[Any]:
+def _get_durable_agent(agent: Agent[Any], durable_run: DurableRun, seen:
dict[int, Agent[Any]]) -> Agent[Any]:
"""Return a copy of ``agent`` whose model, function tools and handoff
agents go through the journal."""
if (done := seen.get(id(agent))) is not None:
return done
- durable = agent.clone()
+ durable_agent = agent.clone()
# Recorded before the handoffs are copied, so agents that hand off to each
other copy once.
- seen[id(agent)] = durable
+ seen[id(agent)] = durable_agent
if isinstance(agent.model, Model):
- durable.model = _DurableModel(agent.model, durable_run)
- durable.tools = [_durable_tool(tool, durable_run) for tool in agent.tools]
- durable.handoffs = [
- _durable_agent(target, durable_run, seen) if isinstance(target, Agent)
else target
+ durable_agent.model = _DurableModel(agent.model, durable_run)
+ durable_agent.tools = [_get_durable_tool(tool, durable_run) for tool in
agent.tools]
+ durable_agent.handoffs = [
+ _get_durable_agent(target, durable_run, seen) if isinstance(target,
Agent) else target
for target in agent.handoffs
]
- return durable
+ return durable_agent
-def _durable_tool(tool: Tool, durable_run: DurableRun) -> Tool:
+def _get_durable_tool(tool: Tool, durable_run: DurableRun) -> Tool:
# A tool whose approval depends on its arguments is left as it is: the SDK
checks that the
# arguments it approved are the ones the tool runs with only through its
own invoker.
if not isinstance(tool, FunctionTool) or callable(tool.needs_approval):
return tool
+ # A shallow copy, not dataclasses.replace: the tool's schema is already
prepared.
+ durable_tool = copy.copy(tool)
invoke = tool.on_invoke_tool
+ # Exceptions a custom failure_error_function turned into text, by tool
call.
+ handled: dict[str, Exception] = {}
+ formatter = resolve_function_tool_failure_error_function(tool)
+ bind = getattr(invoke, "__agents_bind_function_tool__", None)
+ if formatter is not None and formatter is not default_tool_error_function
and bind is not None:
+
+ def remember(ctx: Any, error: Exception) -> Any:
+ handled[ctx.tool_call_id] = error
+ return formatter(ctx, error)
+
+ set_function_tool_failure_error_function(durable_tool, remember)
+ invoke = bind(durable_tool)
async def on_invoke_tool(ctx: ToolContext[Any], arguments: str) -> Any:
- if _inside_a_call(durable_run):
+ if _is_inside_a_call(durable_run):
return await invoke(ctx, arguments)
try:
parsed = json.loads(arguments) if arguments else {}
@@ -222,19 +246,26 @@ def _durable_tool(tool: Tool, durable_run: DurableRun) ->
Tool:
mask_secrets(parsed if isinstance(parsed, dict) else {"arguments":
parsed}),
ctx.tool_call_id,
)
- return await durable_run.run(
- f"tool:{tool.name}",
- kind="tool",
- fingerprint=fingerprint,
- body=lambda: invoke(ctx, arguments),
+ step = durable_run.claim(f"tool:{tool.name}", kind="tool",
fingerprint=fingerprint)
+ if step.replayed:
+ return step.payload
+ try:
+ with step.executing():
+ output = await invoke(ctx, arguments)
+ except Exception as error:
+ step.fail(error)
+ raise
+ # A failure the SDK turned into text for the model is not recorded, so
a retry runs the
+ # call again, as it does a call that raised.
+ if (handled_error := handled.pop(ctx.tool_call_id, None)) is not None:
+ step.fail(handled_error)
+ elif not getattr(ctx, _DEFAULT_FAILURE_HANDLED, False):
# Secrets must not reach the journal any more than the model.
- to_record=lambda output: mask_secrets(_as_sent(output)),
- )
+ step.record(mask_secrets(_as_sent(output)))
+ return output
- # A shallow copy, not dataclasses.replace: the tool's schema is already
prepared.
- durable = copy.copy(tool)
- durable.on_invoke_tool = on_invoke_tool
- return durable
+ durable_tool.on_invoke_tool = on_invoke_tool
+ return durable_tool
def _as_sent(output: Any) -> Any:
@@ -257,7 +288,7 @@ def _as_sent(output: Any) -> Any:
return str(output)
-def _inside_a_call(durable_run: DurableRun) -> bool:
+def _is_inside_a_call(durable_run: DurableRun) -> bool:
"""
Whether a recorded call of ``durable_run`` is running, such as an agent
called through ``as_tool``.
@@ -272,24 +303,24 @@ class _DurableModelProvider(ModelProvider):
"""Wrap every model the run resolves by name in a
:class:`_DurableModel`."""
def __init__(self, provider: ModelProvider, durable_run: DurableRun) ->
None:
- self._provider = provider
- self._durable_run = durable_run
- self._models: dict[str | None, _DurableModel] = {}
+ self.provider = provider
+ self.durable_run = durable_run
+ self.models: dict[str | None, _DurableModel] = {}
def get_model(self, model_name: str | None) -> Model:
- if model_name not in self._models:
- self._models[model_name] =
_DurableModel(self._provider.get_model(model_name), self._durable_run)
- return self._models[model_name]
+ if model_name not in self.models:
+ self.models[model_name] =
_DurableModel(self.provider.get_model(model_name), self.durable_run)
+ return self.models[model_name]
class _DurableModel(Model):
"""Replay a model call the previous attempt recorded, or make it and
record the response."""
def __init__(self, model: Model, durable_run: DurableRun) -> None:
- self._model = model
- self._durable_run = durable_run
+ self.wrapped = model
+ self.durable_run = durable_run
# The step of a call that raised, which the SDK's own model retry may
make again.
- self._failed: JournalStep | None = None
+ self.failed_step: JournalStep | None = None
async def get_response(
self,
@@ -302,14 +333,14 @@ class _DurableModel(Model):
tracing: ModelTracing,
**kwargs: Any,
) -> ModelResponse:
- if _inside_a_call(self._durable_run):
- return await self._model.get_response(
+ if _is_inside_a_call(self.durable_run):
+ return await self.wrapped.get_response(
system_instructions, input, model_settings, tools,
output_schema, handoffs, tracing, **kwargs
)
fingerprint = _fingerprint_request(
- self._model, system_instructions, input, model_settings, tools,
output_schema, handoffs, kwargs
+ self.wrapped, system_instructions, input, model_settings, tools,
output_schema, handoffs, kwargs
)
- step = self._retried(fingerprint) or self._durable_run.claim(
+ step = self._retried(fingerprint) or self.durable_run.claim(
"model", kind="model", fingerprint=fingerprint
)
if step.replayed:
@@ -320,17 +351,17 @@ class _DurableModel(Model):
response_id=step.payload.get("response_id"),
)
except (ValidationError, KeyError, TypeError) as error:
- self._durable_run.reject(step, reason=f"the recorded model
response cannot be read: {error}")
+ self.durable_run.reject(step, reason=f"the recorded model
response cannot be read: {error}")
async def call() -> ModelResponse:
- return await self._model.get_response(
+ return await self.wrapped.get_response(
system_instructions, input, model_settings, tools,
output_schema, handoffs, tracing, **kwargs
)
try:
return await step.run(call, to_record=_record_response)
except Exception:
- self._failed = step
+ self.failed_step = step
raise
def _retried(self, fingerprint: str | None) -> JournalStep | None:
@@ -340,10 +371,10 @@ class _DurableModel(Model):
The SDK's model retry asks again with the same request and nothing in
between; it is
the same call, so it takes the same position, and the next try lines
up with this one.
"""
- failed, self._failed = self._failed, None
+ failed, self.failed_step = self.failed_step, None
if failed is None or failed.fingerprint != fingerprint:
return None
- return failed if self._durable_run.position == failed.position + 1
else None
+ return failed if self.durable_run.position == failed.position + 1 else
None
def stream_response(
self,
@@ -357,23 +388,23 @@ class _DurableModel(Model):
**kwargs: Any,
) -> AsyncIterator[TResponseStreamEvent]:
# A streamed response is not recorded: it reaches the caller event by
event.
- return self._model.stream_response(
+ return self.wrapped.stream_response(
system_instructions, input, model_settings, tools, output_schema,
handoffs, tracing, **kwargs
)
def _supports_default_prompt_cache_key(self) -> bool:
# The runner asks the model whether to add a prompt cache key; the
answer is the wrapped model's.
- supports = getattr(self._model, "_supports_default_prompt_cache_key",
None)
+ supports = getattr(self.wrapped, "_supports_default_prompt_cache_key",
None)
return bool(supports()) if callable(supports) else False
def get_retry_advice(self, request: ModelRetryAdviceRequest) ->
ModelRetryAdvice | None:
- return self._model.get_retry_advice(request)
+ return self.wrapped.get_retry_advice(request)
async def _cleanup_on_run_end(self, owner: object) -> None:
- await self._model._cleanup_on_run_end(owner)
+ await self.wrapped._cleanup_on_run_end(owner)
async def close(self) -> None:
- await self._model.close()
+ await self.wrapped.close()
def _record_response(response: ModelResponse) -> dict[str, Any]:
diff --git
a/providers/common/ai/tests/unit/common/ai/durable/test_openai_agents.py
b/providers/common/ai/tests/unit/common/ai/durable/test_openai_agents.py
index f1bffe35775..30baff77db8 100644
--- a/providers/common/ai/tests/unit/common/ai/durable/test_openai_agents.py
+++ b/providers/common/ai/tests/unit/common/ai/durable/test_openai_agents.py
@@ -32,17 +32,19 @@ from agents import (
ModelResponse,
ModelSettings,
RunConfig,
+ SQLiteSession,
ToolOutputText,
Usage,
function_tool,
)
-from agents.exceptions import MaxTurnsExceeded
+from agents.exceptions import MaxTurnsExceeded, ModelBehaviorError
from agents.retry import ModelRetryBackoffSettings, ModelRetrySettings
from openai.types.responses import ResponseFunctionToolCall,
ResponseOutputMessage, ResponseOutputText
from pydantic import BaseModel
+from airflow.providers.common.ai.durable.base import build_step_key
from airflow.providers.common.ai.durable.journal import DurableJournal,
journal_scope
-from airflow.providers.common.ai.durable.openai_agents import _durable_tool,
run_durably_sync
+from airflow.providers.common.ai.durable.openai_agents import
_get_durable_tool, run_durably_sync
from airflow.providers.common.ai.tools import AirflowTool, ToolResult
from airflow.providers.common.ai.tools.openai_agents import AirflowTools
@@ -316,6 +318,10 @@ class TestRunDurably:
pytest.param(datetime.datetime(2026, 10, 14, 9, 30),
id="datetime"),
pytest.param({"rows": 3}, id="dict"),
pytest.param(ToolOutputText(text="3 rows"), id="structured"),
+ pytest.param(
+ [ToolOutputText(text="3 rows"), ToolOutputText(text="2
orders")], id="structured-list"
+ ),
+ pytest.param({"type": "text", "text": "3 rows"}, id="typed-dict"),
],
)
def test_a_tool_result_replays_as_the_model_saw_it(self, result):
@@ -400,12 +406,110 @@ class TestRunDurably:
"""Pay an amount."""
return "paid"
- assert _durable_tool(pay, DurableJournal(MemoryStorage()).start_run())
is pay
+ assert _get_durable_tool(pay,
DurableJournal(MemoryStorage()).start_run()) is pay
- @pytest.mark.parametrize("option", ["previous_response_id",
"conversation_id"])
- def test_refuses_a_conversation_kept_on_openais_servers(self, option):
+ @pytest.mark.parametrize(
+ ("option", "value"),
+ [
+ ("previous_response_id", "resp_1"),
+ ("conversation_id", "conv_1"),
+ ("auto_previous_response_id", True),
+ ],
+ )
+ def test_refuses_a_conversation_kept_on_openais_servers(self, option,
value):
with pytest.raises(ValueError, match=option):
- _attempt(MemoryStorage(), _ScriptedModel(), _CountingTool(),
**{option: "resp_1"})
+ _attempt(MemoryStorage(), _ScriptedModel(), _CountingTool(),
**{option: value})
+
+ def
test_a_recorded_answer_that_cannot_be_read_runs_live_with_every_call_after_it(self):
+ storage, tool = MemoryStorage(), _CountingTool()
+ with pytest.raises(RuntimeError):
+ _attempt(storage, _ScriptedModel(fail_on_answer=True), tool)
+ storage.entries[build_step_key(0, 0)]["payload"]["output"] = [{"type":
"not an output item"}]
+
+ retry_model = _ScriptedModel()
+ _, journal = _attempt(storage, retry_model, tool)
+
+ assert retry_model.calls == 2
+ assert tool.calls == 2
+ assert journal.stats.replayed == {"model": 0}
+
+ @pytest.mark.parametrize(
+ "failure_error_function",
+ [
+ pytest.param({}, id="default-formatter"),
+ pytest.param(
+ {"failure_error_function": lambda ctx, error: f"lookup failed:
{error}"}, id="custom"
+ ),
+ ],
+ )
+ def test_a_tool_call_that_failed_runs_again_on_retry(self,
failure_error_function):
+ """The SDK hands the model the error as text; that text must not
replay as the tool's result."""
+ calls = []
+
+ @function_tool(**failure_error_function)
+ def lookup(key: str) -> str:
+ """Look a key up."""
+ calls.append(key)
+ if len(calls) == 1:
+ raise ConnectionError("warehouse unreachable")
+ return f"{key}=42"
+
+ storage = MemoryStorage()
+
+ def attempt(model: _ScriptedModel) -> DurableJournal:
+ journal = DurableJournal(storage, clean_up_after_run=True)
+ with journal_scope(journal):
+ run_durably_sync(
+ Agent(name="analyst", model=model, tools=[lookup]), "go",
run_config=NO_TRACING
+ )
+ return journal
+
+ with pytest.raises(RuntimeError, match="worker lost"):
+ attempt(_ScriptedModel(fail_on_answer=True))
+ retry_model = _ScriptedModel()
+ attempt(retry_model)
+
+ assert len(calls) == 2
+ assert retry_model.calls == 1
+
+ def
test_a_model_behavior_error_a_tool_raises_reruns_the_tool_and_replays_the_answer_before_it(self):
+ calls = []
+
+ @function_tool(failure_error_function=None)
+ def lookup(key: str) -> str:
+ """Look a key up."""
+ calls.append(key)
+ if len(calls) == 1:
+ raise ModelBehaviorError("lookup was called with a key that
does not exist")
+ return f"{key}=42"
+
+ storage = MemoryStorage()
+
+ def attempt(model: _ScriptedModel) -> None:
+ with journal_scope(DurableJournal(storage,
clean_up_after_run=True)):
+ run_durably_sync(
+ Agent(name="analyst", model=model, tools=[lookup]), "go",
run_config=NO_TRACING
+ )
+
+ with pytest.raises(ModelBehaviorError):
+ attempt(_ScriptedModel())
+ retry_model = _ScriptedModel()
+ attempt(retry_model)
+
+ assert len(calls) == 2
+ # The answer that asked for the lookup replays; only the final answer
is asked for.
+ assert retry_model.calls == 1
+
+ def test_a_run_with_a_session_runs_live_on_retry(self, tmp_path):
+ """The session already holds the failed try's prompt, so the first
request differs."""
+ storage, tool = MemoryStorage(), _CountingTool()
+ session = SQLiteSession("analyst", str(tmp_path / "session.db"))
+ with pytest.raises(RuntimeError):
+ _attempt(storage, _ScriptedModel(fail_on_answer=True), tool,
session=session)
+
+ _, journal = _attempt(storage, _ScriptedModel(), tool, session=session)
+
+ assert journal.stats.replayed == {}
def test_outside_a_task_it_runs_the_agent_as_runner_run_does(self):
model, tool = _ScriptedModel(), _CountingTool()
