Skip to content

fix: defer session save until after output guardrails #3998

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

Closed
hsusul wants to merge 3 commits into from
+483 −16
Closed

fix: defer session save until after output guardrails #3998

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
aa7c6b0
fix: defer session save until after output guardrails
hsusul Jul 28, 2026
564e67d
fix: persist resumed finals after output guardrails
hsusul Jul 28, 2026
449c6a7
fix: keep executed tool items when a final turn is rejected
hsusul Aug 3, 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
98 changes: 83 additions & 15 deletions src/agents/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -446,6 +446,12 @@ def run_streamed(
)


# Only the model's own output for the final turn waits for the output-guardrail verdict.
# This is an allow-list on purpose: a future item type that records something the run
# already did defaults to "persist" rather than silently inheriting "discard".
_DEFERRED_FINAL_OUTPUT_ITEM_TYPES = frozenset({"message_output_item", "reasoning_item"})


class AgentRunner:
"""
WARNING: this class is experimental and not part of the public API
Expand Down Expand Up @@ -919,15 +925,27 @@ def _finalize_result(result: RunResult) -> RunResult:
session_items=session_items,
)

# A resumed turn that ends the run carries the same obligation as a
# fresh one: side-effect records persist now, model output waits for
# the guardrail verdict below.
resumed_items_now = (
[
item
for item in turn_session_items
if item.type not in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES
]
if isinstance(turn_result.next_step, NextStepFinalOutput)
else turn_session_items
)
if (
session_persistence_enabled
and turn_session_items
and resumed_items_now
and run_state is not None
):
run_state._current_turn_persisted_item_count = (
await save_resumed_turn_items(
session=session,
items=turn_session_items,
items=resumed_items_now,
persisted_count=(
run_state._current_turn_persisted_item_count
),
Expand Down Expand Up @@ -1414,16 +1432,35 @@ def _finalize_result(result: RunResult) -> RunResult:
)
):
items_to_save_turn.append(item)
if items_to_save_turn:
# Defer only the model's own output until after output guardrails
# succeed, so a tripwire does not leave rejected assistant output in
# the session. Tool calls and their outputs record side effects that
# already happened, so they persist now even on a final-output turn
# (tool_use_behavior="stop_on_first_tool" ends the run on such a turn).
if isinstance(turn_result.next_step, NextStepFinalOutput):
items_now_turn = [
item
for item in items_to_save_turn
if item.type not in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES
Comment on lines +1440 to +1444

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve item order when final output passes guardrails

When a successful non-streamed final turn contains a message or reasoning item before a tool call, this split persists the tool call and output first and appends the earlier model item only after guardrails pass. The resulting session order no longer matches the model response, so a subsequent run replays reordered history; defer the decision rather than the individual items so the complete original batch can be appended in order on success, while saving only executed side-effect records on rejection.

AGENTS.md reference: AGENTS.md:L128-L128

Useful? React with 👍 / 👎.

]
deferred_items_turn = [
item
for item in items_to_save_turn
if item.type in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES
]
else:
items_now_turn = items_to_save_turn
deferred_items_turn = []
if items_now_turn:
logger.debug(
"Persisting turn items (types=%s)",
[item.type for item in items_to_save_turn],
[item.type for item in items_now_turn],
)
if is_resumed_state and run_state is not None:
saved_count = await save_result_to_session(
session,
[],
items_to_save_turn,
items_now_turn,
None,
response_id=turn_result.model_response.response_id,
reasoning_item_id_policy=(
Expand All @@ -1436,7 +1473,7 @@ def _finalize_result(result: RunResult) -> RunResult:
await save_result_to_session(
session,
[],
items_to_save_turn,
items_now_turn,
run_state,
response_id=turn_result.model_response.response_id,
store=store_setting,
Expand Down Expand Up @@ -1484,15 +1521,46 @@ def _finalize_result(result: RunResult) -> RunResult:
result._current_turn_persisted_item_count = (
run_state._current_turn_persisted_item_count
)
await save_turn_items_if_needed(
session=session,
run_state=run_state,
session_persistence_enabled=session_persistence_enabled,
input_guardrail_results=input_guardrail_results,
items=session_items_for_turn(turn_result),
response_id=turn_result.model_response.response_id,
store=store_setting,
)
if (
session_persistence_enabled
and deferred_items_turn
and not input_guardrails_triggered(input_guardrail_results)
):
# When earlier items from this turn were already persisted
# (resume / soft-cancel), save_turn_items_if_needed would
# no-op. Use the resumed-turn path so accepted final items
# are still appended after output guardrails succeed.
if (
run_state is not None
and run_state._current_turn_persisted_item_count > 0
):
run_state._current_turn_persisted_item_count = (
await save_resumed_turn_items(
session=session,
items=deferred_items_turn,
persisted_count=(
run_state._current_turn_persisted_item_count
),
response_id=turn_result.model_response.response_id,
reasoning_item_id_policy=(
run_state._reasoning_item_id_policy
),
store=store_setting,
)
)
result._current_turn_persisted_item_count = (
run_state._current_turn_persisted_item_count
)
else:
await save_turn_items_if_needed(
session=session,
run_state=run_state,
session_persistence_enabled=session_persistence_enabled,
input_guardrail_results=input_guardrail_results,
items=deferred_items_turn,
response_id=turn_result.model_response.response_id,
store=store_setting,
)
result._original_input = copy_input_items(original_input)
return _finalize_result(result)
elif isinstance(turn_result.next_step, NextStepInterruption):
Expand Down
18 changes: 17 additions & 1 deletion src/agents/run_internal/run_loop.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -414,6 +414,12 @@ async def _run_output_guardrails_for_stream(
raise


# Only the model's own output for the final turn waits for the output-guardrail verdict.
# This is an allow-list on purpose: a future item type that records something the run
# already did defaults to "persist" rather than silently inheriting "discard".
_DEFERRED_FINAL_OUTPUT_ITEM_TYPES = frozenset({"message_output_item", "reasoning_item"})


async def _finalize_streamed_final_output(
*,
streamed_result: RunResultStreaming,
Expand All @@ -426,6 +432,15 @@ async def _finalize_streamed_final_output(
response_id: str | None,
store_setting: bool | None,
) -> None:
# Tool calls and their outputs record side effects that already happened, so they persist
# before the guardrails can raise. Only the model's own output waits for a clean verdict.
side_effect_items = [
item for item in items if item.type not in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES
]
deferred_items = [item for item in items if item.type in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES]
if side_effect_items:
await save_items(side_effect_items, response_id, store_setting)

output_guardrail_results = await _run_output_guardrails_for_stream(
agent=agent,
run_config=run_config,
Expand All @@ -437,7 +452,8 @@ async def _finalize_streamed_final_output(
streamed_result.final_output = output
streamed_result.is_complete = True

await save_items(items, response_id, store_setting)
if deferred_items:
await save_items(deferred_items, response_id, store_setting)

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve deferred items in mixed streamed final turns

When a streamed final turn contains both a message or reasoning item and an executed tool call, the first save_items(side_effect_items, ...) call increments the backing RunState._current_turn_persisted_item_count; this second call then passes only the shorter deferred subset to _save_stream_items_with_count, so save_result_to_session treats that entire subset as already persisted and saves nothing. Thus, when output guardrails pass, the session silently loses the accepted assistant message or reasoning item; save the original ordered batch on success or otherwise avoid applying the full-turn count to filtered subsets.

AGENTS.md reference: AGENTS.md:L118-L118

Useful? React with 👍 / 👎.

@LHMQ878 LHMQ878 Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Both of these are real, and I can confirm them from the other side: I made exactly this mistake on my own branch an hour ago and had to back it out. No stake in this PR — @hsusul's call — but the evidence may save a round.

I filed #4148 for the pre-existing streamed half of this (a committed tool's session record is dropped when a streamed output guardrail trips). My first commit there split the save unconditionally, the same shape as here. Codex's P1 is right, and the mechanism is exactly as described.

Reproduction — mixed final turn (message emitted before the tool call, tool_use_behavior="stop_on_first_tool"), printing what lands in the session:

                      unconditional split      decision deferred
RUN    trip=False  -> user message fc fco       user message fc fco
RUN    trip=True   -> user message fc fco       user message fc fco
STREAM trip=False  -> user         fc fco  <--  user message fc fco
STREAM trip=True   -> user         fc fco       user         fc fco

The STREAM trip=False row is the P1: the accepted assistant message is silently lost. save_result_to_session computes

already_persisted = run_state._current_turn_persisted_item_count if run_state else 0
if already_persisted >= len(new_items):
    new_run_items = []

The first save advanced that count to the full turn length, so the shorter deferred subset trips already_persisted >= len(new_items) and nothing is written. It fails open — no exception, no log.

The P2 ordering claim holds for the same input: the model emitted message then function_call, and a two-phase save persists them in the reverse order, so a later run replays reordered history.

What worked. Defer the decision rather than the items — keep the success path as one unsplit save, and write a subset only on the tripwire path, which by definition is discarding the deliverable output anyway:

    try:
        output_guardrail_results = await _run_output_guardrails_for_stream(...)
    except Exception:
        committed = [i for i in items if i.type in _COMMITTED_ITEM_TYPES]
        if committed:
            await save_items(committed, response_id, store_setting)
        raise
    ...
    await save_items(items, response_id, store_setting)   # unchanged: full ordered batch

That fixes both claims at once: the passing path never sees a subset, so neither the count arithmetic nor the ordering can be perturbed by it.

Two notes if you go this route:

  • Allow-list the committed types, don't deny-list them. A future item type that records a side effect then defaults to being persisted rather than silently inheriting "discard on tripwire".
  • The regression test has to cover the passing mixed turn, not just the tripwire. My tripwire tests all passed while the P1 bug was live — it only shows up in trip=False. I ended up parametrizing over run/streamed × pass/trip; the pass rows are what caught it.

Full suite on my branch with the deferred-decision shape: 69 failed / 5482 passed vs 69 failed / 5474 passed at base, identical failing test-ID sets (compared as JUnit XML, since the raw count is flaky on Windows).


streamed_result._event_queue.put_nowait(QueueCompleteSentinel())

Expand Down
Loading