Skip to content

fix(sessions): preserve writes that land while responses.compact is in flight #4680

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
Om-singhaI wants to merge 14 commits into
base: main
Choose a base branch
from
Open

fix(sessions): preserve writes that land while responses.compact is in flight #4680

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
8dd60cf
fix(sessions): preserve writes that land while responses.compact is i…
Om-singhaI Aug 26, 2026
e0566c1
fix(sessions): detect empty snapshot clears and pin the response id a…
Om-singhaI Aug 26, 2026
868becf
fix(sessions): count a compaction replacement as a history rewrite
Om-singhaI Aug 26, 2026
8acf6c3
Merge remote-tracking branch 'origin/main' into fix/compaction-concur…
Om-singhaI Aug 26, 2026
49f34b9
Merge remote-tracking branch 'origin/main' into fix/compaction-concur…
Om-singhaI Aug 27, 2026
833ed99
fix(sessions): preserve turns persisted after a response's recorded b…
Om-singhaI Aug 27, 2026
c755d21
fix(sessions): skip compaction when a recorded boundary has been dropped
Om-singhaI Aug 27, 2026
901a543
fix(sessions): compute response boundaries before the batch append
Om-singhaI Aug 27, 2026
21768f2
fix(sessions): harden compaction edge paths and simplify boundary boo…
Om-singhaI Aug 27, 2026
439dc29
fix(sessions): record response boundaries only for writes the run owns
Om-singhaI Aug 28, 2026
2fa608c
fix(sessions): record no boundary when the request input skips history
Om-singhaI Aug 28, 2026
bb5f52a
fix(sessions): arm the gate for stale tokens and void filter rewrites
Om-singhaI Aug 28, 2026
225ee41
fix(sessions): void nested handoff rewrites and arm the gate before a…
Om-singhaI Aug 28, 2026
7bfbb96
fix(sessions): void ownership on any short request history read
Om-singhaI Aug 28, 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
447 changes: 385 additions & 62 deletions src/agents/memory/openai_responses_compaction_session.py
View file
Open in desktop

Large diffs are not rendered by default.

38 changes: 38 additions & 0 deletions src/agents/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,7 @@
from .run_internal.session_persistence import (
_session_get_items,
admit_pending_input,
capture_session_ownership_token,
commit_server_pending_input,
persist_session_items_for_guardrail_trip,
prepare_input_with_session,
Expand Down Expand Up @@ -162,6 +163,9 @@
from .tracing.span_data import AgentSpanData, TaskSpanData
from .util import _error_tracing

if TYPE_CHECKING:
from .memory.openai_responses_compaction_session import _SessionOwnershipToken

DEFAULT_AGENT_RUNNER: AgentRunner = None # type: ignore
# the value is set at the end of the module

Expand Down Expand Up @@ -607,6 +611,18 @@ async def _run_impl(
# Track the most recent input batch we persisted so conversation-lock retries can rewind
# exactly those items (and not the full history).
last_saved_input_snapshot_for_rewind: list[TResponseInputItem] | None = None
# Ownership of the session history this run builds its request input
# from; captured right before the history read and threaded through
# every persist so compaction boundaries record only when no other
# writer interleaved. Resumed runs never read the session for their
# request input, so they carry no token and record no boundaries.
# The token is voided wherever the request provably stops covering
# the stored history: a resolved limit truncating the history read,
# a session input callback rebuilding the input, a
# call_model_input_filter rewriting a request, or a handoff input
# filter rewriting the accumulated history. Such runs record no
# boundaries either.
session_ownership_token: _SessionOwnershipToken | None = None

if is_resumed_state and run_state is not None:
(
Expand Down Expand Up @@ -671,6 +687,9 @@ async def _run_impl(
original_input_for_state = raw_input
session_input_items_for_persistence = []
else:
session_ownership_token = await capture_session_ownership_token(
session, wrapper=context_wrapper
)
(
prepared_input,
session_input_items_for_persistence,
Expand All @@ -681,6 +700,7 @@ async def _run_impl(
run_config.session_settings,
reasoning_item_id_policy=resolved_reasoning_item_id_policy,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
original_input_for_state = prepared_input

Expand Down Expand Up @@ -952,6 +972,7 @@ def _mark_response_hooks_started() -> None:
run_state,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
session_input_items_for_persistence = []
except BaseException:
Expand Down Expand Up @@ -1009,6 +1030,7 @@ def _mark_response_hooks_started() -> None:
run_state,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
)
raise
Expand Down Expand Up @@ -1060,6 +1082,7 @@ def _mark_response_hooks_started() -> None:
run_state,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
session_input_items_for_persistence = []
if run_state is not None and run_state._current_step is not None:
Expand Down Expand Up @@ -1320,6 +1343,7 @@ def _mark_response_hooks_started() -> None:
response_id=turn_result.model_response.response_id,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
except BaseException as persistence_error:
raise _safe_redacted_persistence_error(
Expand Down Expand Up @@ -1352,6 +1376,7 @@ def _mark_response_hooks_started() -> None:
response_id=turn_result.model_response.response_id,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
raise

Expand All @@ -1371,6 +1396,7 @@ def _mark_response_hooks_started() -> None:
response_id=turn_result.model_response.response_id,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
current_step = getattr(run_state, "_current_step", None)
approvals_from_state = approvals_from_step(current_step)
Expand Down Expand Up @@ -1444,6 +1470,7 @@ def _mark_response_hooks_started() -> None:
server_conversation_tracker=server_conversation_tracker,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
generated_items.extend(admission_items)
session_items.extend(admission_items)
Expand Down Expand Up @@ -1523,6 +1550,7 @@ async def _save_max_turns_handler_output(
reasoning_item_id_policy=resolved_reasoning_item_id_policy,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
)
if not items:
Expand Down Expand Up @@ -1656,6 +1684,7 @@ async def _save_max_turns_handler_output(
run_state,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
)
raise
Expand Down Expand Up @@ -1685,6 +1714,7 @@ async def _save_max_turns_handler_output(
on_response_accepted=_commit_pending_server_response,
on_response_hooks_started=_mark_response_hooks_started,
run_state=run_state,
ownership_token=session_ownership_token,

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 28, 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 Invalidate ownership after sandbox context rewriting

When a SandboxAgent capability uses the public Capability.process_context() hook to filter stored history, sandbox_runtime.prepare_agent() replaces original_input after the token was captured, but this unchanged token is still forwarded here. The following persist can therefore record a boundary covering the full session even though the model request omitted part of it, and previous_response_id compaction can silently delete that omitted prefix. Invalidate the token whenever sandbox preparation rewrites the input, as is already done for the other input-filtering hooks; the streamed path has the same stale-token forwarding.

AGENTS.md reference: AGENTS.md:L147-L149

Useful? React with 👍 / 👎.

)
)

Expand Down Expand Up @@ -1717,6 +1747,7 @@ async def _save_max_turns_handler_output(
run_state,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
)
raise
Expand Down Expand Up @@ -1760,6 +1791,7 @@ async def _save_max_turns_handler_output(
on_response_accepted=_commit_pending_server_response,
on_response_hooks_started=_mark_response_hooks_started,
run_state=run_state,
ownership_token=session_ownership_token,
)
finally:
if current_turn_span is not None:
Expand Down Expand Up @@ -1869,6 +1901,7 @@ async def _save_max_turns_handler_output(
response_id=turn_result.model_response.response_id,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)

# After the first resumed turn, treat subsequent turns as fresh
Expand Down Expand Up @@ -1947,6 +1980,7 @@ async def _save_max_turns_handler_output(
response_id=turn_result.model_response.response_id,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
except BaseException as persistence_error:
raise _safe_redacted_persistence_error(
Expand Down Expand Up @@ -1979,6 +2013,7 @@ async def _save_max_turns_handler_output(
response_id=turn_result.model_response.response_id,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
raise

Expand All @@ -1998,6 +2033,7 @@ async def _save_max_turns_handler_output(
response_id=turn_result.model_response.response_id,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)

# Ensure starting_input is not None and not RunState
Expand Down Expand Up @@ -2056,6 +2092,7 @@ async def _save_max_turns_handler_output(
response_id=turn_result.model_response.response_id,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
append_model_response_if_new(
model_responses, turn_result.model_response
Expand Down Expand Up @@ -2118,6 +2155,7 @@ async def _save_max_turns_handler_output(
response_id=turn_result.model_response.response_id,
store=store_setting,
wrapper=context_wrapper,
ownership_token=session_ownership_token,
)
continue
else:
Expand Down
12 changes: 11 additions & 1 deletion src/agents/run_internal/agent_runner_helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from __future__ import annotations

from collections.abc import Mapping
from typing import Any, cast
from typing import TYPE_CHECKING, Any, cast

from openai.types.responses.response_usage import OutputTokensDetails

Expand Down Expand Up @@ -45,6 +45,9 @@
from .tool_use_tracker import AgentToolUseTracker, serialize_tool_use_tracker
from .turn_preparation import get_model

if TYPE_CHECKING:
from ..memory.openai_responses_compaction_session import _SessionOwnershipToken

__all__ = [
"apply_resumed_conversation_settings",
"append_model_response_if_new",
Expand Down Expand Up @@ -535,6 +538,7 @@ async def save_turn_items_if_needed(
response_id: str | None,
store: bool | None = None,
wrapper: RunContextWrapper[Any] | None = None,
ownership_token: _SessionOwnershipToken | None = None,
) -> None:
"""Persist turn items when persistence is enabled and guardrails allow it."""
if not session_persistence_enabled:
Expand All @@ -551,6 +555,7 @@ async def save_turn_items_if_needed(
response_id=response_id,
store=store,
wrapper=wrapper,
ownership_token=ownership_token,
)


Expand All @@ -565,13 +570,17 @@ async def save_final_turn_items_after_guardrails(
reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
store: bool | None = None,
wrapper: RunContextWrapper[Any] | None = None,
ownership_token: _SessionOwnershipToken | None = None,
) -> int:
"""Persist deferred final-turn items without skipping a partially persisted resumed turn."""
if not session_persistence_enabled or not items:
return 0
if input_guardrails_triggered(input_guardrail_results):
return 0
if run_state is not None and run_state._current_turn_persisted_item_count > 0:
# The reconciling path may append behind an earlier partial persist, so
# no count read there can prove what this response covers; it carries
# no ownership token and records no boundary.
run_state._current_turn_persisted_item_count = await save_resumed_turn_items(
session=session,
items=items,
Expand All @@ -591,6 +600,7 @@ async def save_final_turn_items_after_guardrails(
reasoning_item_id_policy=reasoning_item_id_policy,
store=store,
wrapper=wrapper,
ownership_token=ownership_token,
)


Expand Down
Loading