class OpenAIResponsesCompactionSession(SessionABC, OpenAIResponsesCompactionAwareSession):
"""Session decorator that triggers responses.compact when stored history grows.
Works with OpenAI Responses API models only. Wraps any Session (except
OpenAIConversationsSession) and automatically calls the OpenAI responses.compact
API after each turn when the decision hook returns True.
Automatic compaction matches stored history against the latest successful model
exchange, preserving order and repeated occurrences. Native SQLite, async SQLite, and
SQLAlchemy (SQLite/PostgreSQL/MySQL) stores can atomically replace a bounded,
model-visible suffix while retaining older
history. Partial suffixes start at a user message to preserve preceding model item
groups. Other stores require complete coverage. Partial snapshots use input mode;
explicit previous_response_id mode requires complete coverage. The decision hook
must approve the selected snapshot and mode before compaction.
Item matching respects the wrapped store's declared ID-matching policy and the
current run's reasoning-ID policy. Explicit manual
``run_compaction()`` calls still compact the stored history and should be used
only when that history may be sent.
"""
def __init__(
self,
session_id: str,
underlying_session: Session,
*,
client: AsyncOpenAI | None = None,
model: str = "gpt-4.1",
compaction_mode: OpenAIResponsesCompactionMode = "auto",
should_trigger_compaction: Callable[[dict[str, Any]], bool] | None = None,
max_rollback_items: int | None = None,
):
"""Initialize the compaction session.
Args:
session_id: Identifier for this session.
underlying_session: Session store that holds the compacted history. Cannot be
OpenAIConversationsSession.
client: OpenAI client for responses.compact API calls. Defaults to
get_default_openai_client() or new AsyncOpenAI().
model: Model to use for responses.compact. Defaults to "gpt-4.1". Must be an
OpenAI model name (gpt-*, o*, or ft:gpt-*).
compaction_mode: Controls how the compaction request provides conversation
history. "auto" (default) uses input when the last response was not
stored or no response_id is available.
should_trigger_compaction: Custom decision hook. Defaults to triggering when
10+ compaction candidates exist.
max_rollback_items: Optional positive item-count budget for complete history
snapshots used by manual compaction and legacy whole-history replacement.
Oversized history raises ValueError before the compaction API call or
replacement. Reads request at most this budget plus one overflow item.
None (default) preserves the existing unlimited rollback policy. This is
not a byte limit or a limit on stored history or model input. Native
automatic suffix replacement does not need a full-history rollback
snapshot and is unaffected. SessionSettings.limit remains a retrieval
default, independent of this budget.
"""
if isinstance(underlying_session, OpenAIConversationsSession):
raise ValueError(
"OpenAIResponsesCompactionSession cannot wrap OpenAIConversationsSession "
"because it manages its own history on the server."
)
if not is_openai_model_name(model):
raise ValueError(f"Unsupported model for OpenAI responses compaction: {model}")
if max_rollback_items is not None and max_rollback_items <= 0:
raise ValueError("max_rollback_items must be positive or None")
self.session_id = session_id
self.underlying_session = underlying_session
self._client = client
self.model = model
self.compaction_mode = compaction_mode
self.max_rollback_items = max_rollback_items
self.should_trigger_compaction = (
should_trigger_compaction
if should_trigger_compaction is not None
else default_should_trigger_compaction
)
# cache for incremental candidate tracking
self._compaction_candidate_items: list[TResponseInputItem] | None = None
self._session_items: list[TResponseInputItem] | None = None
self._response_id: str | None = None
self._deferred_response_id: str | None = None
self._last_unstored_response_id: str | None = None
# Serialize wrapper mutations against compaction snapshot/replace/restore so a
# cancellation rollback cannot rewrite past a newer concurrent write.
self._mutation_lock = asyncio.Lock()
# Runner persistence can carry this wrapper-local generation across the
# append-to-compaction gap. A later wrapper mutation revokes that one
# pending automatic replacement without inferring ownership from history.
self._mutation_generation = 0
@property
def _ignore_ids_for_matching(self) -> bool:
"""Preserve the wrapped store's declared item-matching policy."""
return bool(getattr(self.underlying_session, "_ignore_ids_for_matching", False))
@property
def client(self) -> AsyncOpenAI:
if self._client is None:
default_client = get_default_openai_client()
self._client = default_client if default_client is not None else AsyncOpenAI()
return self._client
def _resolve_compaction_mode_for_response(
self,
*,
response_id: str | None,
store: bool | None,
requested_mode: OpenAIResponsesCompactionMode | None,
) -> _ResolvedCompactionMode:
mode = requested_mode or self.compaction_mode
if (
mode == "auto"
and store is None
and response_id is not None
and response_id == self._last_unstored_response_id
):
return "input"
return _resolve_compaction_mode(mode, response_id=response_id, store=store)
async def run_compaction(
self,
args: OpenAIResponsesCompactionArgs | None = None,
*,
wrapper: RunContextWrapper[Any] | None = None,
) -> None:
"""Run compaction using responses.compact API.
When a run context is provided, the billed compaction request contributes to
that run's usage totals.
Manual calls replace the complete stored history, even when the underlying
retrieval default exposes fewer items to the compaction request. A finite
max_rollback_items budget checks the complete rollback history before loading
candidates, including forced calls. Overflow leaves history unchanged.
"""
await self._run_compaction(args, wrapper=wrapper)
async def _run_compaction(
self,
args: OpenAIResponsesCompactionArgs | None,
*,
wrapper: RunContextWrapper[Any] | None,
read_items: Callable[[int | None], Awaitable[tuple[list[TResponseInputItem], bool]]]
| None = None,
prepare_items: Callable[[list[TResponseInputItem]], list[TResponseInputItem]] | None = None,
read_snapshot: Callable[[int], Awaitable[_CompactionSnapshot | None]] | None = None,
) -> None:
# Keep one wrapper mutation boundary from the snapshot through replacement.
# A concurrent add, pop, or clear waits here and then runs against the
# compacted state instead of being overwritten by a stale replacement.
async with self._mutation_lock:
has_expected_generation = wrapper is not None and hasattr(
wrapper, "_session_compaction_generation"
)
expected_generation = (
getattr(wrapper, "_session_compaction_generation", None)
if has_expected_generation
else None
)
if has_expected_generation and (
not isinstance(expected_generation, int)
or expected_generation != self._mutation_generation
):
logger.warning(
"Skipped compaction because Session history changed after this "
"run appended its items."
)
return
await self._run_compaction_locked(
args,
wrapper=wrapper,
read_items=read_items,
prepare_items=prepare_items,
read_snapshot=read_snapshot,
)
async def _run_compaction_locked(
self,
args: OpenAIResponsesCompactionArgs | None,
*,
wrapper: RunContextWrapper[Any] | None,
read_items: Callable[[int | None], Awaitable[tuple[list[TResponseInputItem], bool]]]
| None = None,
prepare_items: Callable[[list[TResponseInputItem]], list[TResponseInputItem]] | None = None,
read_snapshot: Callable[[int], Awaitable[_CompactionSnapshot | None]] | None = None,
) -> None:
if args and args.get("response_id"):
self._response_id = args["response_id"]
requested_mode = args.get("compaction_mode") if args else None
if args and "store" in args:
store = args["store"]
if store is False and self._response_id:
self._last_unstored_response_id = self._response_id
elif store is True and self._response_id == self._last_unstored_response_id:
self._last_unstored_response_id = None
else:
store = None
resolved_mode = self._resolve_compaction_mode_for_response(
response_id=self._response_id,
store=store,
requested_mode=requested_mode,
)
if resolved_mode == "previous_response_id" and not self._response_id:
raise ValueError(
"OpenAIResponsesCompactionSession.run_compaction requires a response_id "
"when using previous_response_id compaction."
)
is_automatic = wrapper is not None and getattr(
wrapper, "_session_compaction_is_automatic", False
)
if not is_automatic and self.max_rollback_items is not None:
# Check before even a default-unlimited candidate read. Time-filtered
# stores can change during the request despite the wrapper mutation lock.
await self._get_all_underlying_session_items()
compaction_candidate_items, session_items, _ = await self._ensure_compaction_candidates(
read_items
)
force = args.get("force", False) if args else False
should_compact = force or self.should_trigger_compaction(
{
"response_id": self._response_id,
"compaction_mode": resolved_mode,
"compaction_candidate_items": compaction_candidate_items,
"session_items": session_items,
}
)
if not should_compact:
logger.debug(
"skip: decision hook declined compaction for %s (mode=%s)",
self._response_id,
resolved_mode,
)
return
snapshot: _CompactionSnapshot | None = None
suffix_start = 0
if is_automatic:
model_exchange: tuple[tuple[str, ...], ReasoningItemIdPolicy | None] = getattr(
wrapper, "_session_compaction_model_exchange", ((), None)
)
model_items, reasoning_item_id_policy = model_exchange
approved_session_items = session_items
approved_mode = resolved_mode
if read_snapshot is None:
read_snapshot = getattr(self.underlying_session, "_get_compaction_snapshot", None)
if read_snapshot is not None:
snapshot = await read_snapshot(len(model_items) + 1)
if snapshot is not None:
session_items = _normalize_compaction_session_items(snapshot.items)
complete = snapshot.complete
else:
# Backends without atomic suffix replacement still require complete
# coverage, established by a finite read after the hook approves.
_, session_items, complete = await self._ensure_compaction_candidates(
read_items, limit=len(model_items) + 1
)
# Consume model occurrences in order, allowing additional model-only items
# between stored items while rejecting reordered or missing occurrences.
# Replay may omit an old reasoning ID, but new caller input and filters can
# retain it. Select only the representation present in the actual exchange.
remaining = reversed(model_items)
matched_items: list[TResponseInputItem] = []
for item in reversed(session_items):
digest = digest_input_item(
item, ignore_ids_for_matching=self._ignore_ids_for_matching
)
normalized_item = apply_reasoning_item_id_policy([item], reasoning_item_id_policy)[
0
]
normalized_digest = digest_input_item(
normalized_item, ignore_ids_for_matching=self._ignore_ids_for_matching
)
matched_digest = next(
(
candidate
for candidate in remaining
if candidate in (digest, normalized_digest)
),
None,
)
if matched_digest is None:
break
matched_items.append(item if matched_digest == digest else normalized_item)
suffix_start = len(session_items) - len(matched_items)
if not matched_items or (snapshot is None and (not complete or suffix_start > 0)):
logger.warning(
"Skipped automatic compaction because complete stored history could not "
"be matched to the latest model exchange. Session history was retained."
)
return
if snapshot is not None and (not complete or suffix_start > 0):
# A response chain may summarize earlier items that will be retained.
# Only input mode can scope the summary to this exact suffix.
if (requested_mode or self.compaction_mode) == "previous_response_id":
return
resolved_mode = "input"
session_items = list(reversed(matched_items))
if snapshot is not None and (not complete or suffix_start > 0):
# Start at a caller message so retained reasoning and its following
# model-emitted item stay together, including outside this window.
boundary = next(
(index for index, item in enumerate(session_items) if _is_user_message(item)),
None,
)
if boundary is None:
return
suffix_start += boundary
session_items = session_items[boundary:]
# Keep tool/program groups intact. Pruning a broken group here would
# make replacement remove stored items absent from the compact input.
if (
drop_orphan_function_calls(
session_items, output_pruning_indexes=set(range(len(session_items)))
)
!= session_items
):
return
compaction_candidate_items = select_compaction_candidate_items(session_items)
# A bounded reload may include history omitted from the hook's initial
# cached/default-limited view. Require approval of the actual snapshot.
if (
session_items != approved_session_items or resolved_mode != approved_mode
) and not self.should_trigger_compaction(
{
"response_id": self._response_id,
"compaction_mode": resolved_mode,
"compaction_candidate_items": compaction_candidate_items,
"session_items": session_items,
}
):
return
if is_automatic and snapshot is None and self.max_rollback_items is not None:
await self._get_all_underlying_session_items()
logger.debug(
"compact: start for %s using %s (mode=%s)",
self._response_id,
self.model,
resolved_mode,
)
compact_kwargs: dict[str, Any] = {"model": self.model}
if resolved_mode == "previous_response_id":
compact_kwargs["previous_response_id"] = self._response_id
else:
# Manual compaction may reuse cached raw output from the preceding compaction.
compact_kwargs["input"] = normalize_input_items_for_api(session_items)
compacted = await self.client.responses.compact(**compact_kwargs)
compacted_usage = getattr(compacted, "usage", None)
if wrapper is not None and compacted_usage is not None:
wrapper.usage.add(_response_usage_to_usage(compacted_usage))
output_items = _strip_orphaned_assistant_ids(
_normalize_compaction_output_items(compacted.output or [])
)
if snapshot is not None:
async def replace_snapshot(snapshot: _CompactionSnapshot) -> bool:
replaced = await snapshot.replace_suffix(suffix_start, output_items)
if replaced:
# Publish the backend's successful outcome before caller cancellation
# is propagated. Runner can settle its checkpoint without inferring
# success from history that merely differs from the original append.
self._deferred_response_id = None
if wrapper is not None:
wrapper._session_compaction_completed = True # type: ignore[attr-defined]
return replaced
try:
replaced = await _await_mutation(replace_snapshot(snapshot))
if not replaced:
logger.warning(
"Skipped compaction replacement because the stored suffix changed."
)
return
finally:
# A transaction can commit before cancellation or acknowledgement failure.
self._mutation_generation += 1
self._compaction_candidate_items = None
self._session_items = None
else:
# Manual calls and legacy complete snapshots retain whole-history
# replacement with the existing rollback semantics.
stored_output_items = (
prepare_items(output_items) if prepare_items is not None else output_items
)
# Refresh after the API request so rollback cannot revive items that
# expired while awaiting compaction. This read still honors the budget.
previous_items = await self._get_all_underlying_session_items()
try:
await self._replace_underlying_session_items(
output_items=stored_output_items, previous_items=previous_items
)
except (Exception, asyncio.CancelledError):
self._mutation_generation += 1
raise
self._mutation_generation += 1
self._compaction_candidate_items = (
None if read_items is not None else select_compaction_candidate_items(output_items)
)
self._session_items = None if read_items is not None else output_items
# Clear the deferred marker only now that compaction has actually settled. Clearing it
# before the fallible API call/replacement above would let a failed forced compaction
# silently lose its "this must be forced" signal: a later retry recomputes `force` from
# this marker, so an early clear makes the retry decline work that was still owed.
self._deferred_response_id = None
logger.debug(
"compact: done for %s (mode=%s, output=%s, candidates=%s)",
self._response_id,
resolved_mode,
len(output_items),
len(self._compaction_candidate_items or []),
)
async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
return await self.underlying_session.get_items(limit)
async def _get_items_with_generation(
self, read_items: Callable[[], Awaitable[list[TResponseInputItem]]]
) -> tuple[list[TResponseInputItem], int]:
"""Read one Runner snapshot with its exact wrapper generation."""
async with self._mutation_lock:
# Read through the outer Session so its decryption and filtering still apply.
items = await read_items()
return items, self._mutation_generation
async def _get_all_underlying_session_items(self) -> list[TResponseInputItem]:
limit = (
_ALL_SESSION_ITEMS_LIMIT
if self.max_rollback_items is None
else self.max_rollback_items + 1
)
items = await self.underlying_session.get_items(limit=limit)
if self.max_rollback_items is not None and len(items) > self.max_rollback_items:
raise ValueError("Compaction history exceeds max_rollback_items; history was retained")
return items
async def _replace_underlying_session_items(
self,
*,
output_items: list[TResponseInputItem],
previous_items: list[TResponseInputItem],
) -> None:
# Treat clear → add as one replacement transaction. Exception and CancelledError
# both restore previous history, and restore settlement is always drained so a
# cancel during restore cannot leave an empty session.
cleared = False
try:
await self.underlying_session.clear_session()
cleared = True
if output_items:
await self.underlying_session.add_items(output_items)
except Exception as error:
await self._recover_from_failed_replacement(
previous_items=previous_items,
error=error,
cleared=cleared,
)
raise
except asyncio.CancelledError as error:
await self._recover_from_failed_replacement(
previous_items=previous_items,
error=error,
cleared=cleared,
)
raise
async def _recover_from_failed_replacement(
self,
*,
previous_items: list[TResponseInputItem],
error: BaseException,
cleared: bool,
) -> None:
if not cleared:
restore = self._restore_underlying_session_items_after_failed_clear(
previous_items, error
)
else:
restore = self._restore_underlying_session_items(previous_items, error)
await self._await_restore_despite_cancellation(restore)
async def _await_restore_despite_cancellation(self, restore: Awaitable[None]) -> None:
"""Await restore even when the current task keeps receiving cancellation.
``asyncio.shield`` alone is not enough: a second ``task.cancel()`` makes
``await asyncio.shield(restore)`` raise immediately while restore is still
running. Keep re-awaiting the shielded task until it settles, then
re-raise ``CancelledError`` so callers still observe cancellation.
"""
restore_task = asyncio.ensure_future(restore)
try:
await asyncio.shield(restore_task)
except asyncio.CancelledError:
while not restore_task.done():
try:
await asyncio.shield(restore_task)
except asyncio.CancelledError:
continue
# Retrieve the restore outcome so a failed restore does not warn about an
# unretrieved task exception after we re-raise cancellation.
_ = restore_task.exception() if not restore_task.cancelled() else None
raise
async def _restore_underlying_session_items_after_failed_clear(
self,
previous_items: list[TResponseInputItem],
clear_error: BaseException,
) -> None:
try:
current_items = await self._get_all_underlying_session_items()
except Exception as inspection_error:
log_model_and_tool_action_warning(
logger,
"Failed to inspect session history after compaction replacement clear failed.",
inspection_error,
)
return
if current_items == previous_items:
return
await self._restore_underlying_session_items(
previous_items, clear_error, clear_existing_items=False
)
async def _restore_underlying_session_items(
self,
previous_items: list[TResponseInputItem],
replacement_error: BaseException,
*,
clear_existing_items: bool = True,
) -> None:
try:
if clear_existing_items:
await self.underlying_session.clear_session()
if previous_items:
await self.underlying_session.add_items(list(previous_items))
except Exception as restore_error:
log_model_and_tool_action_warning(
logger,
"Failed to restore session history after compaction replacement failed.",
restore_error,
)
return
log_model_and_tool_action_warning(
logger,
"Restored previous session history after compaction replacement failed",
replacement_error,
)
async def _defer_compaction(
self,
response_id: str,
store: bool | None = None,
*,
read_items: Callable[[int | None], Awaitable[tuple[list[TResponseInputItem], bool]]]
| None = None,
) -> None:
async with self._mutation_lock:
if self._deferred_response_id is not None:
return
compaction_candidate_items, session_items, _ = await self._ensure_compaction_candidates(
read_items
)
resolved_mode = self._resolve_compaction_mode_for_response(
response_id=response_id,
store=store,
requested_mode=None,
)
should_compact = self.should_trigger_compaction(
{
"response_id": response_id,
"compaction_mode": resolved_mode,
"compaction_candidate_items": compaction_candidate_items,
"session_items": session_items,
}
)
if should_compact:
self._deferred_response_id = response_id
def _get_deferred_compaction_response_id(self) -> str | None:
return self._deferred_response_id
def _clear_deferred_compaction(self) -> None:
self._deferred_response_id = None
async def add_items(self, items: list[TResponseInputItem]) -> None:
async with self._mutation_lock:
await self._add_items_locked(items)
async def _add_items_with_generation(
self,
write_items: Callable[[], Awaitable[None]],
*,
expected_generation: int | None,
) -> int | None:
"""Append one Runner batch and retain ownership only when its read stayed current."""
# The outer Session applies its transformations before our public add_items acquires
# the mutation lock. Any intervening mutation revokes ownership, including a write
# that completes while the outer Session is awaiting acknowledgement after our append.
await write_items()
if expected_generation is not None and self._mutation_generation == expected_generation + 1:
return self._mutation_generation
return None
async def _add_items_locked(self, items: list[TResponseInputItem]) -> None:
try:
await self.underlying_session.add_items(items)
except (Exception, asyncio.CancelledError):
# The backend may have committed before acknowledgement failed. Re-read its
# authoritative history before compaction instead of retaining a stale cache.
self._compaction_candidate_items = None
self._session_items = None
self._mutation_generation += 1
raise
self._mutation_generation += 1
if self._compaction_candidate_items is not None:
new_items = _normalize_compaction_session_items(items)
new_candidates = select_compaction_candidate_items(new_items)
if new_candidates:
self._compaction_candidate_items.extend(new_candidates)
if self._session_items is not None:
self._session_items.extend(_normalize_compaction_session_items(items))
async def pop_item(self) -> TResponseInputItem | None:
async with self._mutation_lock:
try:
popped = await self.underlying_session.pop_item()
except (Exception, asyncio.CancelledError):
# The deletion may have committed before acknowledgement failed.
self._compaction_candidate_items = None
self._session_items = None
self._response_id = None
self._deferred_response_id = None
self._last_unstored_response_id = None
self._mutation_generation += 1
raise
if popped:
self._compaction_candidate_items = None
self._session_items = None
self._response_id = None
self._deferred_response_id = None
self._last_unstored_response_id = None
self._mutation_generation += 1
return popped
async def clear_session(self) -> None:
async with self._mutation_lock:
try:
await self.underlying_session.clear_session()
except (Exception, asyncio.CancelledError):
# The deletion may have committed before acknowledgement failed.
self._compaction_candidate_items = None
self._session_items = None
self._response_id = None
self._deferred_response_id = None
self._last_unstored_response_id = None
self._mutation_generation += 1
raise
self._compaction_candidate_items = []
self._session_items = []
self._response_id = None
self._deferred_response_id = None
self._last_unstored_response_id = None
self._mutation_generation += 1
async def _ensure_compaction_candidates(
self,
read_items: Callable[[int | None], Awaitable[tuple[list[TResponseInputItem], bool]]]
| None = None,
*,
limit: int | None = None,
) -> tuple[list[TResponseInputItem], list[TResponseInputItem], bool]:
"""Load policy-visible candidates, or a bounded automatic coverage snapshot."""
if read_items is None:
# Resolve the wrapped policy before consulting the cache: visibility can
# change without a wrapper mutation, for example when encrypted items expire.
read_items = getattr(self.underlying_session, "_read_compaction_items", None)
cache_snapshot = read_items is None and limit is None
if (
cache_snapshot
and self._compaction_candidate_items is not None
and self._session_items is not None
):
return (self._compaction_candidate_items[:], self._session_items[:], False)
if read_items is not None:
items, complete = await read_items(limit)
else:
items = await self.underlying_session.get_items(limit=limit)
complete = limit is not None and len(items) < limit
history = _normalize_compaction_session_items(items)
candidates = select_compaction_candidate_items(history)
if not cache_snapshot:
# Explicit coverage limits and outer logical views bypass partial caches.
return candidates, history, complete
self._compaction_candidate_items = candidates
self._session_items = history
logger.debug(
"candidates: initialized (history=%s, candidates=%s)",
len(history),
len(candidates),
)
return (candidates[:], history[:], False)