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.
"""
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,
):
"""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.
"""
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}")
self.session_id = session_id
self.underlying_session = underlying_session
self._client = client
self.model = model
self.compaction_mode = compaction_mode
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 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.
"""
await self._run_compaction(args, wrapper=wrapper)
async def _run_compaction(
self,
args: OpenAIResponsesCompactionArgs | None,
*,
wrapper: RunContextWrapper[Any] | None,
read_items: Callable[[], Awaitable[list[TResponseInputItem]]] | None = None,
prepare_items: Callable[[list[TResponseInputItem]], list[TResponseInputItem]] | 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
)
async def _run_compaction_locked(
self,
args: OpenAIResponsesCompactionArgs | None,
*,
wrapper: RunContextWrapper[Any] | None,
read_items: Callable[[], Awaitable[list[TResponseInputItem]]] | None = None,
prepare_items: Callable[[list[TResponseInputItem]], list[TResponseInputItem]] | 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."
)
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
self._deferred_response_id = None
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:
compact_kwargs["input"] = 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 [])
)
# Prepare output before any destructive operation. Keep the rollback snapshot in
# its original storage form so restoring encrypted history does not renew its TTL.
stored_output_items = (
prepare_items(output_items) if prepare_items is not None else output_items
)
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
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]:
return await self.underlying_session.get_items(limit=_ALL_SESSION_ITEMS_LIMIT)
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[[], Awaitable[list[TResponseInputItem]]] | 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):
self._compaction_candidate_items = None
self._session_items = None
self._mutation_generation += 1
raise
if popped:
self._compaction_candidate_items = None
self._session_items = 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):
self._compaction_candidate_items = None
self._session_items = None
self._deferred_response_id = None
self._mutation_generation += 1
raise
self._compaction_candidate_items = []
self._session_items = []
self._deferred_response_id = None
self._mutation_generation += 1
async def _ensure_compaction_candidates(
self,
read_items: Callable[[], Awaitable[list[TResponseInputItem]]] | None = None,
) -> tuple[list[TResponseInputItem], list[TResponseInputItem]]:
"""Lazy-load and cache compaction candidates."""
if read_items is not None:
# The outer view can change through TTL expiration without a mutation. Its
# logical items also differ from the raw items maintained by our append cache.
history = _normalize_compaction_session_items(await read_items())
return select_compaction_candidate_items(history), history
if self._compaction_candidate_items is not None and self._session_items is not None:
return (self._compaction_candidate_items[:], self._session_items[:])
history = _normalize_compaction_session_items(await self.underlying_session.get_items())
candidates = select_compaction_candidate_items(history)
self._compaction_candidate_items = candidates
self._session_items = history
logger.debug(
"candidates: initialized (history=%s, candidates=%s)",
len(history),
len(candidates),
)
return (candidates[:], history[:])