fix(sessions): the held record owns the conversion policy of its items

A detached re-park cannot see the Session backend, so it folded new items under
the resuming run's own reasoning-id policy. For a Conversations-origin batch
that strips the server id at the one point where nothing can restore it, and
the reattach then drops the reasoning item as unpersistable.

The park now records the conversion policy it actually used (None for a
Conversations backend, the run's policy otherwise) on the held record, and a
fold converts under the record's policy instead of the caller's. The key is
gated and validated with the other held keys under the unreleased 1.18 schema,
refused on ordinary pending writes and on unknown values; an absent key falls
back to the caller's policy.

Pinned at the fold and at the park, plus the validator rejection, each guard
proven red by mutation.

Author: dixso

src/agents/run_internal/session_persistence.py

@@ -1096,9 +1096,13 @@ def defer_interrupted_session_write(
 
     # The normal persistence path forces the reasoning-id policy to ``None`` for a
     # Conversations backend so a server-identified reasoning item stays persistable;
-    # the registration conversion must match or the sanitization later drops it.
+    # the registration conversion must match or the sanitization later drops it. A
+    # standing record owns the policy its items were converted under, so a re-park
+    # folds new items under the same conversion instead of the resuming run's own.
     if isinstance(session, OpenAIConversationsSession):
         reasoning_item_id_policy = None
+    if pending is not None and "reasoning_item_id_policy" in pending:
+        reasoning_item_id_policy = pending["reasoning_item_id_policy"]
     converted_run_items: list[TResponseInputItem] = []
     for run_item in run_items:
         as_input = run_item_to_input_item(run_item, reasoning_item_id_policy)
@@ -1140,6 +1144,7 @@ def defer_interrupted_session_write(
         if (pending is not None and "response_id" in pending)
         else response_id,
         "store": pending["store"] if (pending is not None and "store" in pending) else store,
+        "reasoning_item_id_policy": reasoning_item_id_policy,
     }
     run_state._pending_session_write = record
 
@@ -1155,7 +1160,9 @@ def extend_held_session_write(
     With no Session attached the resolved turn's save is a no-op, so the executed
     tool output exists only in this process; folding it into the held batch lets the
     reattaching resume settle call and output together. Does nothing when no held
-    batch stands.
+    batch stands. The fold converts under the batch's registration policy, not the
+    caller's: a detached run cannot see the original backend, and a server reasoning
+    id stripped here could not be restored at the settle.
     """
     if run_state is None or run_state._pending_session_write is None:
         return

src/agents/run_state.py

@@ -109,6 +109,7 @@
 from .run_context import RunContextWrapper
 from .run_internal.items import (
     NestedHistoryOwnedItemRef,
+    ReasoningItemIdPolicy,
     digest_input_item,
     ensure_nested_history_run_item_occurrence_key,
     nested_history_run_item_occurrence_key,
@@ -180,6 +181,11 @@ class _PendingSessionWrite(TypedDict):
     ``response_id`` records the model response the withheld batch belongs to, and
     ``store`` the store setting that response was produced under, so the settle runs
     the same compaction bookkeeping the ordinary persistence path would have run for
+    ``reasoning_item_id_policy`` records how the batch's items were converted, so a
+    detached re-park folds new items under the same conversion: a Conversations-origin
+    batch preserves server reasoning ids even when the resuming run's own policy would
+    omit them, and an id stripped at registration cannot be restored at the settle.
+
     it instead of appending behind its back.
     """
 
@@ -190,6 +196,7 @@ class _PendingSessionWrite(TypedDict):
     held: NotRequired[bool]
     response_id: NotRequired[str | None]
     store: NotRequired[bool | None]
+    reasoning_item_id_policy: NotRequired[ReasoningItemIdPolicy | None]
 
 
 def _default_run_state_validation_error(
@@ -248,7 +255,8 @@ def _default_run_state_validation_error(
     ),
     "1.18": (
         "Persists the interrupted turn's withheld Session write, including the response it "
-        "belongs to, so an approval resume can settle it under the output-guardrail gate."
+        "belongs to and the conversion policy its items were registered under, so an "
+        "approval resume can settle it under the output-guardrail gate."
     ),
 }
 SUPPORTED_SCHEMA_VERSIONS = frozenset(SCHEMA_VERSION_SUMMARIES)
@@ -4403,7 +4411,11 @@ async def _build_run_state_from_json(
             for part in _HELD_PENDING_SESSION_WRITE_MIN_SCHEMA_VERSION.split(".", maxsplit=1)
         )
         base_keys = {"session_id", "items", "before", "persisted_count"}
-        held_keys = {"held", "response_id", "store"} if held_keys_allowed else set()
+        held_keys = (
+            {"held", "response_id", "store", "reasoning_item_id_policy"}
+            if held_keys_allowed
+            else set()
+        )
         if (
             (schema_major, schema_minor) < (1, 17)
             or not isinstance(state._current_step, NextStepRunAgain | NextStepInterruption)
@@ -4420,12 +4432,20 @@ async def _build_run_state_from_json(
                 and pending_write["store"] is not None
                 and type(pending_write["store"]) is not bool
             )
-            # Both keys describe the withheld response, so they are meaningless on an
+            or (
+                "reasoning_item_id_policy" in pending_write
+                and pending_write["reasoning_item_id_policy"] not in (None, "preserve", "omit")
+            )
+            # These keys describe the withheld batch, so they are meaningless on an
             # ordinary pending write and are refused there rather than restored as
             # state nothing consumes.
             or (
                 not pending_write.get("held")
-                and ("response_id" in pending_write or "store" in pending_write)
+                and (
+                    "response_id" in pending_write
+                    or "store" in pending_write
+                    or "reasoning_item_id_policy" in pending_write
+                )
             )
             or not isinstance(pending_write.get("session_id"), str)
             or not isinstance(pending_write.get("items"), list)

tests/test_deferred_interrupted_session_write.py

@@ -1487,6 +1487,80 @@ class _Session:
     assert state._pending_session_write["response_id"] == "resp_parked"
 
 
+@pytest.mark.asyncio
+async def test_a_detached_re_park_folds_under_the_batch_registration_policy() -> None:
+    # A Conversations-origin batch was converted preserving server reasoning ids. The
+    # detached re-park cannot see the backend, so it must fold under the policy the
+    # record carries rather than the resuming run's own: an id stripped here is
+    # unrecoverable and the reattach would drop the reasoning item as unpersistable.
+    from agents.items import ReasoningItem
+    from agents.run_internal.session_persistence import extend_held_session_write
+
+    agent = _make_deferring_agent()
+    state = object.__new__(RunState)
+    state._pending_session_write = {
+        "session_id": "conv_abc",
+        "items": [
+            {"type": "function_call", "call_id": "call_PARKED", "name": "t", "arguments": "{}"}
+        ],
+        "before": None,
+        "persisted_count": 1,
+        "held": True,
+        "response_id": "resp_parked",
+        "store": None,
+        "reasoning_item_id_policy": None,
+    }
+    state._current_turn_persisted_item_count = 0
+    reasoning = ReasoningItem(
+        agent=agent,
+        raw_item={"id": "rs_SERVER_ID", "type": "reasoning", "summary": [], "content": []},
+    )
+
+    extend_held_session_write(state, run_items=[reasoning], reasoning_item_id_policy="omit")
+
+    items = state._pending_session_write["items"]
+    reasoning_ids = [i.get("id") for i in items if i.get("type") == "reasoning"]
+    assert reasoning_ids == ["rs_SERVER_ID"]
+    assert state._pending_session_write["reasoning_item_id_policy"] is None
+
+
+@pytest.mark.asyncio
+async def test_the_park_records_the_conversion_policy_it_used() -> None:
+    # The record owns how its items were converted. A Conversations park forces the
+    # preserving policy regardless of the run's own setting, and the recorded value is
+    # what a later detached fold must reuse.
+    from agents.items import ToolCallItem
+    from agents.memory.openai_conversations_session import OpenAIConversationsSession
+    from agents.run_internal.session_persistence import defer_interrupted_session_write
+
+    session = object.__new__(OpenAIConversationsSession)
+    session._session_id = "conv_abc"
+    state = object.__new__(RunState)
+    state._pending_session_write = None
+    state._current_turn_persisted_item_count = 0
+    call = ToolCallItem(
+        agent=_make_deferring_agent(),
+        raw_item={
+            "type": "function_call",
+            "call_id": "call_PARKED",
+            "name": "t",
+            "arguments": "{}",
+        },
+    )
+
+    defer_interrupted_session_write(
+        state,
+        session,
+        run_items=[call],
+        reasoning_item_id_policy="omit",
+        response_id="resp_parked",
+        store=None,
+    )
+
+    assert state._pending_session_write is not None
+    assert state._pending_session_write["reasoning_item_id_policy"] is None
+
+
 class _CompactionRecordingSession(SimpleListSession):
     """Record the compaction bookkeeping a compaction-aware backend expects."""
 

tests/test_run_impl_resume_paths.py

@@ -508,6 +508,7 @@ async def test_failed_streamed_result_checkpoint_retains_detached_pending_write(
         "held-with-before",
         "held-under-1-17",
         "held-keys-without-held",
+        "policy-shape",
     ],
 )
 async def test_pending_session_write_rejects_invalid_serialized_checkpoint(invalid: str) -> None:
@@ -526,6 +527,12 @@ async def test_pending_session_write_rejects_invalid_serialized_checkpoint(inval
         # response_id and store describe the withheld response, so they are refused on
         # an ordinary pending write where nothing consumes them.
         payload["pending_session_write"]["response_id"] = "resp_1"
+    elif invalid == "policy-shape":
+        # The conversion-policy key only speaks the two policy literals or None; any
+        # other value would silently change how a fold converts the batch's items.
+        payload["pending_session_write"]["held"] = True
+        payload["pending_session_write"]["before"] = None
+        payload["pending_session_write"]["reasoning_item_id_policy"] = "banana"
     elif invalid == "held-under-1-17":
         # 1.17 defined the pending write as exactly four keys, so the held variant is
         # only readable under the version that introduced it.