feat: add approval override argument parity for RunState resumes

Author: seratch

src/agents/memory/__init__.py

@@ -1,11 +1,19 @@
 from .openai_conversations_session import OpenAIConversationsSession
 from .openai_responses_compaction_session import OpenAIResponsesCompactionSession
 from .session import (
+    SERVER_MANAGED_CONVERSATION_SESSION_ATTR,
     OpenAIResponsesCompactionArgs,
     OpenAIResponsesCompactionAwareSession,
+    ServerManagedConversationSession,
     Session,
     SessionABC,
+    SessionHistoryMutation,
+    SessionHistoryRewriteArgs,
+    SessionHistoryRewriteAwareSession,
+    apply_session_history_mutations,
     is_openai_responses_compaction_aware_session,
+    is_server_managed_conversation_session,
+    is_session_history_rewrite_aware_session,
 )
 from .session_settings import SessionSettings
 from .sqlite_session import SQLiteSession
@@ -21,5 +29,13 @@
     "OpenAIResponsesCompactionSession",
     "OpenAIResponsesCompactionArgs",
     "OpenAIResponsesCompactionAwareSession",
+    "SERVER_MANAGED_CONVERSATION_SESSION_ATTR",
+    "SessionHistoryMutation",
+    "SessionHistoryRewriteArgs",
+    "SessionHistoryRewriteAwareSession",
+    "ServerManagedConversationSession",
+    "apply_session_history_mutations",
+    "is_server_managed_conversation_session",
     "is_openai_responses_compaction_aware_session",
+    "is_session_history_rewrite_aware_session",
 ]

src/agents/memory/openai_conversations_session.py

@@ -21,6 +21,7 @@ async def start_openai_conversations_session(openai_client: AsyncOpenAI | None =
 
 
 class OpenAIConversationsSession(SessionABC):
+    _server_managed_conversation_session = True
     session_settings: SessionSettings | None = None
 
     def __init__(

src/agents/memory/openai_responses_compaction_session.py

@@ -11,6 +11,9 @@
     OpenAIResponsesCompactionArgs,
     OpenAIResponsesCompactionAwareSession,
     SessionABC,
+    SessionHistoryRewriteArgs,
+    apply_session_history_mutations,
+    is_session_history_rewrite_aware_session,
 )
 
 if TYPE_CHECKING:
@@ -129,48 +132,87 @@ def __init__(
         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
+        self._last_store: bool | None = None
+        self._has_pending_local_history_rewrite = False
+        self._local_history_rewrite_response_id: str | None = None
+        self._has_unacknowledged_local_session_adds = False
 
     @property
     def client(self) -> AsyncOpenAI:
         if self._client is None:
             self._client = get_default_openai_client() or AsyncOpenAI()
         return self._client
 
-    def _resolve_compaction_mode_for_response(
+    def _resolve_compaction_mode(
         self,
         *,
+        requested_mode: OpenAIResponsesCompactionMode,
         response_id: str | None,
         store: bool | None,
-        requested_mode: OpenAIResponsesCompactionMode | None,
+        turn_has_local_adds_without_new_response_id: bool,
     ) -> _ResolvedCompactionMode:
-        mode = requested_mode or self.compaction_mode
+        resolved_mode = _resolve_compaction_mode(
+            requested_mode,
+            response_id=response_id,
+            store=store,
+        )
+
+        if turn_has_local_adds_without_new_response_id and resolved_mode == "previous_response_id":
+            self._has_unacknowledged_local_session_adds = False
+            self._mark_local_history_rewrite()
+            logger.debug(
+                "compact: forcing input mode after local session delta without new response id"
+            )
+            return "input"
+
+        if not self._has_pending_local_history_rewrite:
+            return resolved_mode
+
         if (
-            mode == "auto"
-            and store is None
+            self._local_history_rewrite_response_id is not None
             and response_id is not None
-            and response_id == self._last_unstored_response_id
+            and response_id != self._local_history_rewrite_response_id
         ):
+            self._has_pending_local_history_rewrite = False
+            self._local_history_rewrite_response_id = None
+            return resolved_mode
+
+        if resolved_mode == "previous_response_id":
+            if self._local_history_rewrite_response_id is None and response_id is not None:
+                self._local_history_rewrite_response_id = response_id
+            logger.debug("compact: forcing input mode after local history rewrite")
             return "input"
-        return _resolve_compaction_mode(mode, response_id=response_id, store=store)
+
+        return resolved_mode
 
     async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None) -> None:
         """Run compaction using responses.compact API."""
+        previous_response_id = self._response_id
         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
+            store: bool | None = args["store"]
+            self._last_store = store
         else:
-            store = None
-        resolved_mode = self._resolve_compaction_mode_for_response(
+            store = self._last_store
+        turn_has_local_adds_without_new_response_id = (
+            self._has_unacknowledged_local_session_adds
+            and (args is None or args.get("response_id") in {None, previous_response_id})
+        )
+        if (
+            args
+            and args.get("response_id") is not None
+            and args["response_id"] != previous_response_id
+        ):
+            self._has_unacknowledged_local_session_adds = False
+        resolved_mode = self._resolve_compaction_mode(
             response_id=self._response_id,
             store=store,
-            requested_mode=requested_mode,
+            requested_mode=requested_mode or self.compaction_mode,
+            turn_has_local_adds_without_new_response_id=(
+                turn_has_local_adds_without_new_response_id
+            ),
         )
 
         if resolved_mode == "previous_response_id" and not self._response_id:
@@ -198,6 +240,15 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
             )
             return
 
+        unresolved_function_calls = _find_unresolved_function_calls_without_results(session_items)
+        if unresolved_function_calls:
+            logger.debug(
+                "compact: blocked unresolved function calls for %s: %s",
+                self._response_id,
+                unresolved_function_calls,
+            )
+            return
+
         self._deferred_response_id = None
         logger.debug(
             f"compact: start for {self._response_id} using {self.model} (mode={resolved_mode})"
@@ -239,14 +290,37 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
     async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
         return await self.underlying_session.get_items(limit)
 
+    async def apply_history_mutations(self, args: SessionHistoryRewriteArgs) -> None:
+        """Rewrite persisted history and keep compaction caches aligned with the new transcript."""
+        mutations = list(args.get("mutations", []))
+        if not mutations:
+            return
+
+        if is_session_history_rewrite_aware_session(self.underlying_session):
+            await self.underlying_session.apply_history_mutations({"mutations": mutations})
+            await self._refresh_caches_from_underlying_session()
+            self._mark_local_history_rewrite()
+            return
+
+        rewritten_items = apply_session_history_mutations(
+            await self.underlying_session.get_items(),
+            mutations,
+        )
+        await self.underlying_session.clear_session()
+        if rewritten_items:
+            await self.underlying_session.add_items(rewritten_items)
+        self._session_items = rewritten_items
+        self._compaction_candidate_items = select_compaction_candidate_items(rewritten_items)
+        self._mark_local_history_rewrite()
+
     async def _defer_compaction(self, response_id: str, store: bool | None = None) -> None:
         if self._deferred_response_id is not None:
             return
         compaction_candidate_items, session_items = await self._ensure_compaction_candidates()
-        resolved_mode = self._resolve_compaction_mode_for_response(
+        resolved_mode = _resolve_compaction_mode(
+            self.compaction_mode,
             response_id=response_id,
-            store=store,
-            requested_mode=None,
+            store=store if store is not None else self._last_store,
         )
         should_compact = self.should_trigger_compaction(
             {
@@ -266,7 +340,10 @@ def _clear_deferred_compaction(self) -> None:
         self._deferred_response_id = None
 
     async def add_items(self, items: list[TResponseInputItem]) -> None:
+        if not items:
+            return
         await self.underlying_session.add_items(items)
+        self._has_unacknowledged_local_session_adds = True
         if self._compaction_candidate_items is not None:
             new_candidates = select_compaction_candidate_items(items)
             if new_candidates:
@@ -286,6 +363,15 @@ async def clear_session(self) -> None:
         self._compaction_candidate_items = []
         self._session_items = []
         self._deferred_response_id = None
+        self._has_pending_local_history_rewrite = False
+        self._local_history_rewrite_response_id = None
+        self._has_unacknowledged_local_session_adds = False
+        self._last_store = None
+
+    async def _refresh_caches_from_underlying_session(self) -> None:
+        history = await self.underlying_session.get_items()
+        self._session_items = history
+        self._compaction_candidate_items = select_compaction_candidate_items(history)
 
     async def _ensure_compaction_candidates(
         self,
@@ -304,10 +390,37 @@ async def _ensure_compaction_candidates(
         )
         return (candidates[:], history[:])
 
+    def _mark_local_history_rewrite(self) -> None:
+        self._has_pending_local_history_rewrite = True
+        self._local_history_rewrite_response_id = self._response_id
+
 
 _ResolvedCompactionMode = Literal["previous_response_id", "input"]
 
 
+def _find_unresolved_function_calls_without_results(items: list[TResponseInputItem]) -> list[str]:
+    """Return function-call ids that do not yet have matching outputs."""
+    function_calls: dict[str, TResponseInputItem] = {}
+    resolved_call_ids: set[str] = set()
+
+    for item in items:
+        if isinstance(item, dict):
+            item_type = item.get("type")
+            call_id = item.get("call_id")
+        else:
+            item_type = getattr(item, "type", None)
+            call_id = getattr(item, "call_id", None)
+
+        if not isinstance(call_id, str):
+            continue
+        if item_type == "function_call":
+            function_calls[call_id] = item
+        elif item_type == "function_call_output":
+            resolved_call_ids.add(call_id)
+
+    return [call_id for call_id in function_calls if call_id not in resolved_call_ids]
+
+
 def _resolve_compaction_mode(
     requested_mode: OpenAIResponsesCompactionMode,
     *,

src/agents/memory/session.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import copy
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, Literal, Protocol, runtime_checkable
 
@@ -9,6 +10,8 @@
     from ..items import TResponseInputItem
     from .session_settings import SessionSettings
 
+SERVER_MANAGED_CONVERSATION_SESSION_ATTR = "_server_managed_conversation_session"
+
 
 @runtime_checkable
 class Session(Protocol):
@@ -104,6 +107,105 @@ async def clear_session(self) -> None:
         ...
 
 
+@runtime_checkable
+class ServerManagedConversationSession(Session, Protocol):
+    """Protocol for sessions whose canonical history is managed by a remote service."""
+
+    _server_managed_conversation_session: Literal[True]
+
+
+def is_server_managed_conversation_session(
+    session: Session | None,
+) -> TypeGuard[ServerManagedConversationSession]:
+    """Check whether the session advertises server-managed history semantics."""
+    if session is None:
+        return False
+    try:
+        marker = getattr(session, SERVER_MANAGED_CONVERSATION_SESSION_ATTR, False)
+    except Exception:
+        return False
+    return marker is True
+
+
+class ReplaceFunctionCallSessionHistoryMutation(TypedDict):
+    """Replace the canonical persisted function call for a tool call."""
+
+    type: Literal["replace_function_call"]
+    call_id: str
+    replacement: TResponseInputItem
+
+
+SessionHistoryMutation = ReplaceFunctionCallSessionHistoryMutation
+
+
+class SessionHistoryRewriteArgs(TypedDict):
+    """Arguments for persisted-history rewrites."""
+
+    mutations: list[SessionHistoryMutation]
+
+
+@runtime_checkable
+class SessionHistoryRewriteAwareSession(Session, Protocol):
+    """Protocol for sessions that can rewrite previously persisted history."""
+
+    async def apply_history_mutations(self, args: SessionHistoryRewriteArgs) -> None:
+        """Apply structured history mutations to the persisted session history."""
+        ...
+
+
+def is_session_history_rewrite_aware_session(
+    session: Session | None,
+) -> TypeGuard[SessionHistoryRewriteAwareSession]:
+    """Check whether a session supports persisted-history rewrites."""
+    if session is None:
+        return False
+    try:
+        apply_history_mutations = getattr(session, "apply_history_mutations", None)
+    except Exception:
+        return False
+    return callable(apply_history_mutations)
+
+
+def apply_session_history_mutations(
+    items: list[TResponseInputItem],
+    mutations: list[SessionHistoryMutation],
+) -> list[TResponseInputItem]:
+    """Apply structured history mutations to a list of persisted session items."""
+    next_items = [copy.deepcopy(item) for item in items]
+    for mutation in mutations:
+        if mutation["type"] == "replace_function_call":
+            next_items = _apply_replace_function_call_mutation(next_items, mutation)
+    return next_items
+
+
+def _apply_replace_function_call_mutation(
+    items: list[TResponseInputItem],
+    mutation: ReplaceFunctionCallSessionHistoryMutation,
+) -> list[TResponseInputItem]:
+    """Replace the first matching function call and drop later duplicates for the same call id."""
+    replacement = copy.deepcopy(mutation["replacement"])
+    next_items: list[TResponseInputItem] = []
+    kept_replacement = False
+
+    for item in items:
+        if _is_matching_function_call(item, mutation["call_id"]):
+            if not kept_replacement:
+                next_items.append(replacement)
+                kept_replacement = True
+            continue
+        next_items.append(item)
+
+    return next_items
+
+
+def _is_matching_function_call(item: TResponseInputItem, call_id: str) -> bool:
+    if isinstance(item, dict):
+        return item.get("type") == "function_call" and item.get("call_id") == call_id
+    item_type = getattr(item, "type", None)
+    item_call_id = getattr(item, "call_id", None)
+    return item_type == "function_call" and item_call_id == call_id
+
+
 class OpenAIResponsesCompactionArgs(TypedDict, total=False):
     """Arguments for the run_compaction method."""