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".
…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".
… 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".
…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: |
seratch
left a comment
seratch
left a comment
There was a problem hiding this comment.
The handoff persistence recovery is improved, but the current head still ignores the boolean returned by input_guardrail_tripwire_triggered_for_stream(). A normal tripwire result therefore publishes the delegate and a resumable NextStepRunAgain. Please stop before that transition when the helper returns true, and verify the serialized state after a delayed stream consumer observes the tripwire.
sylvesterkaczmarek
left a comment
sylvesterkaczmarek
left a comment
There was a problem hiding this comment.
Re-reviewed the fail-then-resume compaction case. _deferred_response_id is now cleared only after compaction has settled, and the added regression verifies the marker survives an API failure and forces the retry. My previous blocker is resolved.
3 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)