Skip to content

fix(sessions): reconcile pending input appends #4906

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
seanxuu wants to merge 14 commits into
base: main
Choose a base branch
from
Open

fix(sessions): reconcile pending input appends #4906

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
b928be7
fix(sessions): reconcile pending input appends
seanxuu Sep 8, 2026
f446681
Merge branch 'main' into fix/issue-4775-pending-input-recovery
seanxuu Sep 8, 2026
1d5fcb5
fix(sessions): reconcile pending input appends
seanxuu Sep 8, 2026
454d221
Merge latest main into pending input fix
seanxuu Sep 8, 2026
78784a3
fix(sessions): preserve pending write ownership metadata
seanxuu Sep 8, 2026
4a9e9f5
fix(run-state): materialize computer output iterables
seanxuu Sep 8, 2026
315b4bb
fix(run): admit input added during session writes
seanxuu Sep 8, 2026
5c7106f
fix(run-state): restore staged local-shell input
seanxuu Sep 8, 2026
c289a5d
fix(run-state): preserve replay provider metadata
seanxuu Sep 8, 2026
b377972
test(run-state): cover provider metadata lost ack
seanxuu Sep 8, 2026
6702e26
Merge latest main into pending input fix
seanxuu Sep 8, 2026
ddf3cbf
Merge branch 'main' into fix/issue-4775-pending-input-recovery
seanxuu Sep 8, 2026
ddd3367
Merge branch 'main' into fix/issue-4775-pending-input-recovery
seanxuu Sep 8, 2026
58de902
Merge branch 'main' into fix/issue-4775-pending-input-recovery
seanxuu Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions integration_tests/packaging/test_run_state_compatibility.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import pytest

from agents import Agent, RunState
from agents.run_state import SUPPORTED_SCHEMA_VERSIONS
from agents.run_state import CURRENT_SCHEMA_VERSION, SUPPORTED_SCHEMA_VERSIONS
from integration_tests._contract_state import (
_deserialize_common_sandbox_session_state,
_redaction_observables,
Expand All @@ -21,7 +21,7 @@


def test_installed_distribution_supports_the_historical_fixture_corpus() -> None:
assert frozenset(SOURCES["versions"]) == SUPPORTED_SCHEMA_VERSIONS
assert frozenset(SOURCES["versions"]) | {CURRENT_SCHEMA_VERSION} == SUPPORTED_SCHEMA_VERSIONS


@pytest.mark.parametrize(
Expand Down
18 changes: 15 additions & 3 deletions src/agents/result.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,12 +119,25 @@ def _populate_state_from_result(
) -> RunState[Any]:
"""Populate a RunState with common fields from a RunResult."""
state._current_agent = result.last_agent
source_state = getattr(result, "_state", None)
pending_input_write = (
source_state._pending_session_write
if isinstance(source_state, RunState)
and source_state._pending_session_write is not None
and "pending_input" in source_state._pending_session_write
else None
)
model_input_items = getattr(result, "_model_input_items", None)
if isinstance(model_input_items, list):
if pending_input_write is not None:
assert isinstance(source_state, RunState)
state._generated_items = list(source_state._generated_items)
state._session_items = list(source_state._session_items)
elif isinstance(model_input_items, list):
state._generated_items = list(model_input_items)
else:
state._generated_items = result.new_items
state._session_items = list(result.new_items)
if pending_input_write is None:
state._session_items = list(result.new_items)
snapshot_refs = _state_snapshot_owned_item_refs(result, state._original_input)
live_refs = rebase_nested_history_owned_item_refs(
state._original_input,
Expand All @@ -144,7 +157,6 @@ def _populate_state_from_result(
state._conversation_id = conversation_id
state._previous_response_id = previous_response_id
state._auto_previous_response_id = auto_previous_response_id
source_state = getattr(result, "_state", None)
if isinstance(source_state, RunState):
state._generated_prompt_cache_key = source_state._generated_prompt_cache_key
state._pending_input = copy.deepcopy(source_state._pending_input)
Expand Down
74 changes: 42 additions & 32 deletions src/agents/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1447,40 +1447,50 @@ def _mark_response_hooks_started() -> None:
if run_state._current_step is None:
run_state._current_step = NextStepRunAgain()

pending_input = run_state.pending_input
if pending_input:
pending_guardrails = current_agent.input_guardrails + (
run_config.input_guardrails or []
)
try:
await run_input_guardrails(
current_agent,
pending_guardrails,
pending_input,
context_wrapper,
input_guardrail_results,
if run_state._pending_input:
while True:
pending_input = run_state.pending_input
if not pending_input:
break
pending_guardrails = current_agent.input_guardrails + (
run_config.input_guardrails or []
)
finally:
run_state._input_guardrail_results = list(input_guardrail_results)
try:
await run_input_guardrails(
current_agent,
pending_guardrails,
pending_input,
context_wrapper,
input_guardrail_results,
)
finally:
run_state._input_guardrail_results = list(
input_guardrail_results
)

admission_items = await admit_pending_input(
run_state=run_state,
agent=current_agent,
session=session,
server_conversation_tracker=server_conversation_tracker,
store=store_setting,
wrapper=context_wrapper,
)
generated_items.extend(admission_items)
session_items.extend(admission_items)
if pending_server_items is not None:
pending_server_items.extend(admission_items)
pending_input_admission_items = [
item for item in admission_items if isinstance(item, InputItem)
]
if not run_state._pending_input:
run_state._generated_items = list(generated_items)
run_state._session_items = list(session_items)
admission_items = await admit_pending_input(
run_state=run_state,
agent=current_agent,
session=session,
server_conversation_tracker=server_conversation_tracker,
store=store_setting,
wrapper=context_wrapper,
)
generated_items.extend(admission_items)
session_items.extend(admission_items)
if pending_server_items is not None:
pending_server_items.extend(admission_items)
pending_input_admission_items.extend(
item for item in admission_items if isinstance(item, InputItem)
)
if not run_state._pending_input:
run_state._generated_items = list(generated_items)
run_state._session_items = list(session_items)
if (
server_conversation_tracker is not None
or not run_state._pending_input
):
break
all_tools = await get_all_tools(execution_agent, context_wrapper)
all_tools = await initialize_computer_tools(
tools=all_tools, context_wrapper=context_wrapper
Expand Down
103 changes: 54 additions & 49 deletions src/agents/run_internal/run_loop.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1496,58 +1496,63 @@ async def _save_max_turns_items(
if run_state is not None and run_state._pending_input:
if run_state._current_step is None:
run_state._current_step = NextStepRunAgain()
pending_input = run_state.pending_input
pending_guardrails = current_agent.input_guardrails + (
run_config.input_guardrails or []
)
previous_result_count = len(streamed_result.input_guardrail_results)
try:
await run_input_guardrails_with_queue(
current_agent,
pending_guardrails,
pending_input,
context_wrapper,
streamed_result,
current_span,
while True:
pending_input = run_state.pending_input
if not pending_input:
break
pending_guardrails = current_agent.input_guardrails + (
run_config.input_guardrails or []
)
finally:
run_state._input_guardrail_results = list(
streamed_result.input_guardrail_results
previous_result_count = len(streamed_result.input_guardrail_results)
try:
await run_input_guardrails_with_queue(
current_agent,
pending_guardrails,
pending_input,
context_wrapper,
streamed_result,
current_span,
)
finally:
run_state._input_guardrail_results = list(
streamed_result.input_guardrail_results
)
tripping_result = next(
(
result
for result in streamed_result.input_guardrail_results[
previous_result_count:
]
if result.output.tripwire_triggered
),
None,
)
tripping_result = next(
(
result
for result in streamed_result.input_guardrail_results[
previous_result_count:
]
if result.output.tripwire_triggered
),
None,
)
if tripping_result is not None:
raise InputGuardrailTripwireTriggered(tripping_result)
if tripping_result is not None:
raise InputGuardrailTripwireTriggered(tripping_result)

store_setting = current_agent.model_settings.resolve(
run_config.model_settings
).store
admission_items = await admit_pending_input(
run_state=run_state,
agent=current_agent,
session=session,
server_conversation_tracker=server_conversation_tracker,
store=store_setting,
wrapper=context_wrapper,
)
streamed_result._model_input_items.extend(admission_items)
streamed_result.new_items.extend(admission_items)
if pending_server_items is not None:
pending_server_items.extend(admission_items)
pending_input_admission_items = [
item for item in admission_items if isinstance(item, InputItem)
]
if not run_state._pending_input:
run_state._generated_items = list(streamed_result._model_input_items)
run_state._session_items = list(streamed_result.new_items)
store_setting = current_agent.model_settings.resolve(
run_config.model_settings
).store
admission_items = await admit_pending_input(
run_state=run_state,
agent=current_agent,
session=session,
server_conversation_tracker=server_conversation_tracker,
store=store_setting,
wrapper=context_wrapper,
)
streamed_result._model_input_items.extend(admission_items)
streamed_result.new_items.extend(admission_items)
if pending_server_items is not None:
pending_server_items.extend(admission_items)
pending_input_admission_items.extend(
item for item in admission_items if isinstance(item, InputItem)
)
if not run_state._pending_input:
run_state._generated_items = list(streamed_result._model_input_items)
run_state._session_items = list(streamed_result.new_items)
if server_conversation_tracker is not None or not run_state._pending_input:
break

all_tools = await get_all_tools(execution_agent, context_wrapper)
all_tools = await initialize_computer_tools(
Expand Down
Loading