From caf4591dfb2a63202ce7ce96294c6f503e9f76f5 Mon Sep 17 00:00:00 2001 From: ayaangazali Date: Mon, 24 Aug 2026 21:53:28 -0700 Subject: [PATCH] fix(sessions): recover failed resumed Session writes on a renewed interruption A resumed turn that resolves one approval but leaves another pending advances to NextStepInterruption. save_resumed_turn_items() and the RunState pending write validator both gated recovery on NextStepRunAgain, so that transition recorded no pending write. A failed append was dropped and the completed tool output never reached the durable Session, while RunState kept it. Widen both gates to accept NextStepInterruption so the existing pending write mechanism covers the transition. --- .../run_internal/session_persistence.py | 3 +- src/agents/run_state.py | 4 +- tests/test_run_impl_resume_paths.py | 65 +++++++++++++++++++ 3 files changed, 69 insertions(+), 3 deletions(-) diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py index 809f1648c3..bfe500b544 100644 --- a/src/agents/run_internal/session_persistence.py +++ b/src/agents/run_internal/session_persistence.py @@ -739,7 +739,8 @@ async def save_resumed_turn_items( wrapper=wrapper, resumed_write_state=( run_state - if run_state is not None and isinstance(run_state._current_step, NextStepRunAgain) + if run_state is not None + and isinstance(run_state._current_step, NextStepRunAgain | NextStepInterruption) else None ), ) diff --git a/src/agents/run_state.py b/src/agents/run_state.py index b196bbaf85..228dd574d8 100644 --- a/src/agents/run_state.py +++ b/src/agents/run_state.py @@ -4358,11 +4358,11 @@ async def _build_run_state_from_json( ) pending_write = state_json.get("pending_session_write") if pending_write is not None: - from .run_internal.run_steps import NextStepRunAgain + from .run_internal.run_steps import NextStepInterruption, NextStepRunAgain if ( (schema_major, schema_minor) < (1, 17) - or not isinstance(state._current_step, NextStepRunAgain) + or not isinstance(state._current_step, NextStepRunAgain | NextStepInterruption) or not isinstance(pending_write, dict) or set(pending_write) != {"session_id", "items", "before", "persisted_count"} or not isinstance(pending_write.get("session_id"), str) diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py index 9a4d88f061..6d1f82babf 100644 --- a/tests/test_run_impl_resume_paths.py +++ b/tests/test_run_impl_resume_paths.py @@ -210,6 +210,71 @@ async def test_resumed_session_append_survives_repeated_failure_and_late_input() assert effects == [7] +async def _partially_approved_session_state(streamed: bool): + """Pause on two approval-gated calls in one response and approve only the first.""" + effects: list[int] = [] + + @tool(needs_approval=True) + async def charge(amount: int) -> str: + effects.append(amount) + return "receipt-7" + + @tool(needs_approval=True) + async def notify() -> str: + raise AssertionError("the unresolved approval must not execute") + + model = ScriptedModel( + [ + [ + get_function_tool_call("charge", '{"amount":7}', call_id="charge-1"), + get_function_tool_call("notify", "{}", call_id="notify-1"), + ], + [get_text_message("done")], + ] + ) + agent = Agent(name="payment", model=model, tools=[charge, notify]) + session = _FailingResumeSession() + paused = await _run_session_resume(agent, "charge 7 and notify", session, streamed) + state = paused.to_state() + charge_approval = next( + item for item in state.get_interruptions() if item.raw_item.call_id == "charge-1" + ) + state.approve(charge_approval) + return agent, model, session, state, effects + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +@pytest.mark.parametrize("round_trip", [False, True], ids=["live", "json"]) +async def test_renewed_interruption_recovers_failed_resumed_session_append( + streamed: bool, round_trip: bool +) -> None: + agent, model, session, state, effects = await _partially_approved_session_state(streamed) + session.failure = "before" + with pytest.raises(RuntimeError) as error: + await _run_session_resume(agent, state, session, streamed) + assert error.value is session.error + assert effects == [7] + assert _charge_pair(await session.get_items()) == ["function_call"] + + if round_trip: + state = await RunState.from_json(agent, state.to_json()) + + pending = await _run_session_resume(agent, state, session, streamed) + pending_state = pending.to_state() + remaining = pending_state.get_interruptions() + assert [item.raw_item.call_id for item in remaining] == ["notify-1"] + assert len(model.calls) == 1 + + pending_state.reject(remaining[0], rejection_message="declined") + result = await _run_session_resume(agent, pending_state, session, streamed) + assert result.final_output == "done" + assert effects == [7] + expected_pair = ["function_call", "function_call_output"] + assert _charge_pair(await session.get_items()) == expected_pair + assert _charge_pair(result.to_input_list()) == expected_pair + + @pytest.mark.asyncio @pytest.mark.parametrize("streamed", [False, True]) @pytest.mark.parametrize("round_trip", [False, True], ids=["live", "json"])