fix(sessions): recover fresh streamed handoffs after session append failures - #4835
fix(sessions): recover fresh streamed handoffs after session append failures#4835mittalpk wants to merge 5 commits into
Conversation
…ailures start_streaming()'s generic-loop NextStepHandoff branch awaited the fallible session append (_save_stream_items_without_count) before updating current_agent, run_state._current_agent, streamed_result.current_agent, and run_state._current_step. If that append raised (e.g. a transient session backend error), the run failed with those fields still pointing at the pre-handoff agent, even though the handoff had already fully executed. Resuming from result.to_state() after such a failure then re-invoked the wrong agent with input that already contained its own handoff call/output. This is the same defect PR openai#4725 fixed in the sibling is_resumed_state branch (used when resuming an interrupted run) by moving the state updates ahead of the fallible save. This applies the same reordering to the generic branch, which every fresh streamed run's handoffs go through, not just resumed ones.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 09e928a53c
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| run_state._current_agent = current_agent | ||
| _publish_streamed_result_agent(streamed_result, current_agent) | ||
| if streamed_result._state is not None: | ||
| streamed_result._state._current_step = NextStepRunAgain() |
There was a problem hiding this comment.
When Session.add_items() fails before appending a fresh streamed handoff, this publishes a resumable NextStepRunAgain state, but _save_stream_items_without_count() uses the ordinary save path and does not create a pending_session_write. Resuming therefore invokes the delegate successfully but only persists the delegate's later turn; the completed function call and handoff output are permanently missing from session history. Checkpoint and retry this batch as the resumed-handoff branch does; the added test currently verifies the agent but never verifies the stored handoff pair.
AGENTS.md reference: AGENTS.md:L149-L149
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed. Live-reproduced the data loss flagged: after a failed handoff append and a successful resume, session.get_items() was missing the handoff-1 function_call/function_call_output pair entirely, exactly as described. Threaded resumed_write_state through _save_stream_items into save_result_to_session for the handoff batch (gated on _current_step already being NextStepRunAgain), reusing the same pending_session_write/resume_pending_session_write mechanism the is_resumed_state branch already relies on. Extended the existing test to assert the pair is present in session history after resume, confirmed it fails without the fix and passes with it.
| current_agent = turn_result.next_step.new_agent | ||
| if run_state is not None: | ||
| run_state._current_agent = current_agent | ||
| _publish_streamed_result_agent(streamed_result, current_agent) |
There was a problem hiding this comment.
When the session append raises after the handoff has executed, this publishes the delegate through current_agent and the resumable state, but AgentUpdatedStreamEvent remains after the failing await and is never queued. Stream consumers can therefore observe handoff items followed by an error without the semantic agent transition, even though the result and resumed run identify the delegate; applications that track the active agent from stream events remain stale. Queue the agent-update event with the state transition before attempting the append.
AGENTS.md reference: AGENTS.md:L192-L192
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed. Moved the AgentUpdatedStreamEvent queue call to before the fallible append, alongside the other state-transition updates. Added a test using a session double that yields before failing (a purely synchronous raise never gives stream_events() a scheduling boundary to prove event delivery either way), confirming the event is now observed before the error surfaces. Left the identical pre-existing ordering in the already-merged is_resumed_state branch (#4725) untouched, since that's a separate branch outside this PR's diff.
…ent-update event Two issues from the automated Codex review on this PR, both real: 1. _save_stream_items_without_count() never registered a pending_session_write checkpoint for the handoff batch, unlike the sibling is_resumed_state branch's _save_resumed_stream_items(). A failed append followed by a successful resume invoked the correct (delegate) agent but permanently dropped the handoff's function_call/function_call_output pair from session history, since nothing recorded the batch for the existing resume_pending_session_write() recovery path to replay. Fixed by threading resumed_write_state through _save_stream_items into save_result_to_session, gated on the handoff branch already having set _current_step to NextStepRunAgain. 2. AgentUpdatedStreamEvent was still queued after the fallible session append, so a live stream_events() consumer would see handoff items followed directly by an error with no semantic agent-transition event, even though the result and resumed run both correctly identify the new agent. Moved the event queue call to sit with the other state-transition updates, before the append. Both fixes are scoped to only the generic-loop branch this PR already touches; the already-merged is_resumed_state branch (openai#4725) has the same pre-existing event-ordering gap but is out of scope here. Extended test_fresh_streamed_handoff_preserves_agent_after_session_append_failure with a session-history assertion for issue 1, and added test_fresh_streamed_handoff_publishes_agent_update_before_session_append_failure for issue 2 (using a new session double that yields before failing, since a purely synchronous raise never gives stream_events() a scheduling boundary to prove event delivery either way).
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 75fc64bae5
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| current_agent = turn_result.next_step.new_agent | ||
| if run_state is not None: | ||
| run_state._current_agent = current_agent | ||
| _publish_streamed_result_agent(streamed_result, current_agent) |
There was a problem hiding this comment.
When a fresh streamed turn reaches a handoff while a parallel input guardrail is still running, this block changes the current agent before _save_stream_items_without_count() calls _should_persist_stream_items() and awaits that guardrail. If the guardrail then raises a non-tripwire error, the run fails but current_agent and the resumable state already identify the delegate; resuming that state invokes the delegate without rerunning the initial input guardrails. Keep the guardrail wait ahead of the transition and move only the actual session append behind it.
AGENTS.md reference: AGENTS.md:L149-L150
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed. Live-reproduced: a slow parallel input guardrail that raises a genuine (non-tripwire) exception left current_agent and the resumable state already pointing at the delegate before the guardrail settled. Added an explicit await on input_guardrail_tripwire_triggered_for_stream() as the first statement in the handoff branch, before any state mutation, so the transition only commits once the guardrail has actually resolved. Added a regression test with a real timing window (not just an instant-draining happy path, which passed even without the fix on my first attempt), confirmed it fails pre-fix and passes after.
| streamed_result._event_queue.put_nowait( | ||
| AgentUpdatedStreamEvent(new_agent=current_agent) | ||
| ) |
There was a problem hiding this comment.
When Session.add_items() raises without yielding, queueing the transition here still does not ensure consumers receive it: the run-loop task can record the exception before the consumer requests its next event, and stream_events() checks _stored_exception before reading the queue and later drains the queued update. Fresh evidence is that the added _FailSecondAddItemsSessionWithYield test deliberately inserts a scheduler yield because the ordinary immediate-failure session cannot demonstrate delivery. Mark this session-error path to drain already queued events, or otherwise synchronize delivery before surfacing the exception.
AGENTS.md reference: AGENTS.md:L192-L192
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed. Reproduced that queuing the event before the append isn't sufficient by itself: stream_events() checks the stored exception before draining the queue, so any consumer that isn't instant can lose the already-queued event to a task that raised without being marked for draining -- this reproduces even with the existing yield-based test double under a realistic per-event consumer delay. Wrapped the fallible append in a try/except that calls _mark_error_to_drain_stream_events() before re-raising, the same pattern already used for model-behavior errors. New test uses a per-event consumer delay to force the real race rather than relying on favorable scheduling.
| resumed_write_state=( | ||
| run_state | ||
| if run_state is not None | ||
| and isinstance(run_state._current_step, NextStepRunAgain) | ||
| else None |
There was a problem hiding this comment.
When the session is an OpenAIResponsesCompactionSession, this checkpoint recovers only the raw append after a failure or lost acknowledgement; _pending_session_write does not retain the handoff's response_id, store, or deferred-compaction action. On resume, resume_pending_session_write() settles and clears the batch, then the delegate runs without the original save_result_to_session() call ever reaching _defer_compaction(). A supported custom compaction decision hook that selects the handoff response but not the delegate's later response therefore silently loses the requested compaction. Preserve and replay the post-write compaction metadata as part of recovery.
AGENTS.md reference: AGENTS.md:L147-L150
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed. The checkpoint only carried session_id/items/before/persisted_count, so a batch settled later via a separate resume_pending_session_write() call never got the response_id/store/local-tool-output info needed to replay the same compaction decision save_result_to_session would have applied inline. Extracted that decision into a shared _apply_post_write_compaction() helper, extended the checkpoint with those fields (optional, so an old-shaped serialized RunState still round-trips without a schema bump), and call the helper once from resume_pending_session_write() regardless of whether the checkpoint settles inline or on a later resume, avoiding a double-compaction call. Regression test confirmed the checkpoint now carries response_id and that compaction is applied after a checkpointed resume.
… compaction Three more issues from the automated Codex review on commit 75fc64b, all verified with live reproduction before being fixed: 1. The handoff transition committed current_agent/run_state before a still-in-flight parallel input guardrail had resolved. A non-tripwire exception from that guardrail then left the resumable state pointing at the delegate agent, even though the starting agent's input guardrails never definitively cleared. Fixed by explicitly awaiting input_guardrail_tripwire_triggered_for_stream() as the first statement in the handoff branch, before any state mutation. 2. Queuing AgentUpdatedStreamEvent before the fallible session append doesn't guarantee delivery: stream_events() checks a stored exception before draining the queue, so a real (non-instant) consumer can lose an already-queued event to a task that raised without ever being marked for draining. Fixed by marking the session-persistence exception via _mark_error_to_drain_stream_events() before re-raising, the same pattern already used for model-behavior errors. 3. The pending_session_write checkpoint recovers the raw item append but never carried enough information (response_id, store, whether the batch had local tool outputs) for a later, separate resume to replay the same post-write Responses compaction decision save_result_to_session would have applied inline. Extracted the compaction decision into a shared _apply_post_write_compaction() helper, extended the checkpoint schema with those fields (optional, so an old-shaped serialized RunState still round-trips), and call the helper from resume_pending_session_write() once a checkpoint settles -- whether inline or on a separate resume -- instead of duplicating the call at both sites. Added 3 new regression tests to tests/test_run_impl_resume_paths.py (72 total in the file, up from 69), each confirmed to fail against the pre-fix code and pass after. Full verification stack clean: make format/ lint/typecheck, and the full suite (9372 passed, 33 skipped, 0 failed).
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 92891a6ea9
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| finally: | ||
| run_state._session_write_in_progress = False | ||
|
|
||
| await _apply_post_write_compaction( |
There was a problem hiding this comment.
When _defer_compaction() or run_compaction() raises or is cancelled after the session append settles, this newly added await runs after _pending_session_write has already been cleared. The resulting RunState therefore cannot retry the handoff batch's post-write compaction on resume, so the delegate proceeds while the requested deferred/forced compaction is silently lost. Fresh evidence in this revision is that the compaction metadata is now checkpointed, but the checkpoint is still discarded before the fallible compaction operation; retain a post-write stage until this await succeeds.
AGENTS.md reference: AGENTS.md:L149-L149
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed. Live-reproduced: after a checkpointed append settled successfully, a failing compaction call still left run_state._pending_session_write cleared, so a later retry had nothing to redo compaction with. Moved the _apply_post_write_compaction() call inside the try block, before clearing the checkpoint -- the existing digest-based reconciliation already makes a retry safe against duplicate appends, so this only changes when the checkpoint is released. New test confirms the checkpoint survives a compaction failure and that a subsequent retry applies compaction without re-appending items, and fails without the fix.
…ttles One more issue from the automated Codex review on commit 92891a6, verified with live reproduction before being fixed: resume_pending_session_write() cleared run_state._pending_session_write before calling the newly-added _apply_post_write_compaction(), so if that call raised or was cancelled, the checkpoint was already gone. A later retry would then have nothing to redo the compaction step with, silently and permanently losing the requested deferred/forced Responses compaction even though the append itself had already succeeded. Fixed by moving the compaction call inside the try block, before clearing the checkpoint. The append reconciliation above already makes a retry safe against duplicate appends (it detects an already-committed batch via digest matching and skips re-appending), so this only changes when the checkpoint is released, not the retry logic itself. Added test_fresh_streamed_handoff_retains_checkpoint_when_post_write_compaction_fails to tests/test_run_impl_resume_paths.py (73 total, up from 72), confirmed to fail against the pre-fix code (checkpoint cleared despite the compaction failure) and pass after. Full verification stack clean: make format/lint/typecheck, and the full suite (9373 passed, 33 skipped, 0 failed).
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 06e9e40081
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| # handoff transition, so a tripwire or guardrail exception is surfaced instead | ||
| # of the state (current_agent, run_state, published events) racing ahead of an | ||
| # input guardrail that was still validating the original input. | ||
| await input_guardrail_tripwire_triggered_for_stream(streamed_result) |
There was a problem hiding this comment.
When a parallel input guardrail finishes normally with tripwire_triggered=True while the first streamed turn produces a handoff, this helper returns True rather than raising, but the return value is ignored here. The following lines therefore commit the delegate and NextStepRunAgain; a slow consumer then receives the tripwire, yet to_state() points at the delegate, so resuming skips the starting agent's input guardrails and processes rejected input under the delegate. Fresh evidence beyond the earlier non-tripwire-error report is that the added test covers only a guardrail exception, while the helper's ordinary tripwire return remains unchecked. Branch on the result and raise InputGuardrailTripwireTriggered before publishing the transition.
AGENTS.md reference: AGENTS.md:L192-L192
Useful? React with 👍 / 👎.
sylvesterkaczmarek
left a comment
sylvesterkaczmarek
left a comment
There was a problem hiding this comment.
The pending-write checkpoint stores the response id, store flag and local-output state, but not whether compaction was forced from a deferred response. run_compaction() clears _deferred_response_id before the API call; if that call fails, a later resume_pending_session_write() recomputes force=False, so a custom compaction policy can skip work that was already forced. Could we persist the force/deferred decision (or clear the deferred marker only after success) and add a fail-then-resume regression?
… success One more issue, this time from a human reviewer (sylvesterkaczmarek) on commit 06e9e40, verified with live reproduction before being fixed: OpenAIResponsesCompactionSession.run_compaction() cleared self._deferred_response_id before calling the fallible client.responses.compact() API. If that call raised, the deferred marker was already gone. The checkpoint-recovery code added in the last two commits recomputes force=True purely from whether this marker is still set, so on retry it silently recomputed force=False and could skip compaction that was still owed -- even though the round-3 fix already let the checkpoint itself survive for a retry. Fixed by moving the clear to after compaction actually settles (after the API call and the underlying session replacement both succeed), not before attempting them. The digest-based retry-safety already added for the append doesn't need any changes; this only moves when one session-internal flag gets cleared. Added test_run_compaction_retains_deferred_marker_when_api_call_fails to tests/memory/test_openai_responses_compaction_session.py, confirmed to fail against the pre-fix code (assert None == 'resp-handoff') and pass after. Full verification stack clean: make format/lint/typecheck, and the full suite (9374 passed, 33 skipped, 0 failed).
|
@sylvesterkaczmarek confirmed and fixed in a follow-up commit. Live-reproduced: |
2 participants
Summary
start_streaming()'s generic-loopNextStepHandoffbranch awaited the fallible session append (_save_stream_items_without_count) before updatingcurrent_agent,run_state._current_agent,streamed_result.current_agent, andrun_state._current_step. If that session write raised (e.g. a transient session-backend error), the run failed with those fields still pointing at the pre-handoff agent, even though the handoff had already fully executed. Resuming fromresult.to_state()after such a failure then re-invoked the wrong agent with input that already contained its own handoff call/output.This is the same defect #4725 fixed in the sibling
is_resumed_statebranch (used when resuming an interrupted run), by moving the state updates ahead of the fallible_save_resumed_items()call. This applies the same reordering to the generic branch, which every fresh streamed run's handoffs go through (not just resumed ones), so it's hit far more often than the branch #4725 covered.The fix is a pure reordering — no new state, no new branches: move
current_agent/run_state._current_agent/_publish_streamed_result_agent/streamed_result._state._current_stepabove theawait _save_stream_items_without_count(...)call.Test plan
Added
test_fresh_streamed_handoff_preserves_agent_after_session_append_failureintests/test_run_impl_resume_paths.py, alongside the existing sibling coverage (test_resumed_handoff_session_append_is_recovered_before_next_model). It drives a fresh (non-RunState-input) streamed run through a handoff whose session append fails on that specific call, asserts the failed result'sto_state()._current_agentand.current_agentalready reflect the new agent, then resumes and confirms the retried turn correctly re-invokes the new agent's model.run_loop.py(git stash) withAssertionError: assert 'triage' == 'delegate', and passes after the fix.make format— clean.make lint— clean.make typecheck(mypy + pyright) — 0 errors, 0 warnings.make tests— 9291 passed, 29 skipped.make tests-serial— 77 passed, 4 skipped.tests/test_run_impl_resume_paths.py(68 tests, including all parametrizations of the sibling fix(sessions): recover resumed handoffs after session append failures #4725 test) — all pass, no regressions.Issue number
None — found via direct code audit while comparing this branch against the sibling fix in #4725; no existing issue.
Checks
.agents/skills/code-change-verification/scripts/run.sh(its parallellint+typecheck+testsphase was killed by this sandbox's resource limits when run concurrently; ran the samemake format,make lint,make typecheck,make testssequence sequentially instead, per the script's own source)/reviewbefore submitting this PR (not applicable — not using Codex)