From 5e5a30566b92b1db4dad07aa319aad8a56d3a1f2 Mon Sep 17 00:00:00 2001
From: Justin Beckwith <jbeckwith@openai.com>
Date: Wed, 7 Oct 2026 10:02:05 -0700
Subject: [PATCH 1/9] fix: batch OpenAI conversation session writes within the
 API limit

---
 .../memory/openai_conversations_session.py    |  18 ++-
 .../test_openai_conversations_session.py      | 140 ++++++++++++++++++
 2 files changed, 154 insertions(+), 4 deletions(-)

diff --git a/src/agents/memory/openai_conversations_session.py b/src/agents/memory/openai_conversations_session.py
index d9fb4f1c18..fe9713fdd2 100644
--- a/src/agents/memory/openai_conversations_session.py
+++ b/src/agents/memory/openai_conversations_session.py
@@ -112,14 +112,24 @@ async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
         return all_items  # type: ignore
 
     async def add_items(self, items: list[TResponseInputItem]) -> None:
+        """Append items in order, in requests of at most 20 items each.
+
+        Writes spanning multiple requests are not atomic. If a request fails or
+        is cancelled, earlier batches remain saved and later batches are not
+        sent. The original exception propagates. Before retrying, callers must
+        reconcile the remote history; retrying the entire list can duplicate
+        items that were already saved, including an unacknowledged request.
+        """
         if not items:
             return
 
         session_id = await self._get_session_id()
-        await self._openai_client.conversations.items.create(
-            conversation_id=session_id,
-            items=items,
-        )
+        # The Conversations items-create endpoint accepts up to 20 items per request.
+        for offset in range(0, len(items), 20):
+            await self._openai_client.conversations.items.create(
+                conversation_id=session_id,
+                items=items[offset : offset + 20],
+            )
 
     async def pop_item(self) -> TResponseInputItem | None:
         session_id = await self._get_session_id()
diff --git a/tests/memory/test_openai_conversations_session.py b/tests/memory/test_openai_conversations_session.py
index 5d5868f7b9..b2a6028e84 100644
--- a/tests/memory/test_openai_conversations_session.py
+++ b/tests/memory/test_openai_conversations_session.py
@@ -3,15 +3,19 @@
 from __future__ import annotations
 
 import asyncio
+import json
 from typing import Any, cast
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx
 import pytest
+from openai import AsyncOpenAI, BadRequestError
 from openai.types.responses.response_output_item import Program, ProgramOutput
 
 from agents import (
     Agent,
     ProgrammaticToolCallingTool,
+    RunConfig,
     Runner,
     TResponseInputItem,
     function_tool,
@@ -565,6 +569,142 @@ async def slow_delete(*args: Any, **kwargs: Any) -> Any:
         mock_openai_client.conversations.create.assert_called_once_with(items=[])
 
 
+class TestOpenAIConversationsSessionBatches:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("count", "expected_sizes"), [(0, []), (20, [20]), (21, [20, 1]), (41, [20, 20, 1])]
+    )
+    async def test_request_sizes_and_order(self, count: int, expected_sizes: list[int]):
+        batches: list[list[dict[str, Any]]] = []
+
+        def capture(request: httpx.Request) -> httpx.Response:
+            assert request.url.path == "/v1/conversations/conv_test/items"
+            batches.append(json.loads(request.content)["items"])
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        items: list[TResponseInputItem] = [
+            {"role": "user", "content": f"message {i}"} for i in range(count)
+        ]
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+        ) as client:
+            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            await session.add_items(items)
+
+        assert [len(batch) for batch in batches] == expected_sizes
+        assert [item for batch in batches for item in batch] == items
+
+    @pytest.mark.asyncio
+    async def test_later_failure_preserves_prefix_and_stops(self):
+        batches: list[list[dict[str, Any]]] = []
+        saved: list[dict[str, Any]] = []
+
+        def capture(request: httpx.Request) -> httpx.Response:
+            batch = json.loads(request.content)["items"]
+            batches.append(batch)
+            if len(batches) == 2:
+                return httpx.Response(400, json={"error": {"message": "synthetic failure"}})
+            saved.extend(batch)
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        items: list[TResponseInputItem] = [
+            {"role": "user", "content": f"message {i}"} for i in range(41)
+        ]
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+        ) as client:
+            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            with pytest.raises(BadRequestError, match="synthetic failure") as caught:
+                await session.add_items(items)
+            assert caught.value.status_code == 400
+
+        assert batches == [items[:20], items[20:40]]
+        assert saved == items[:20]
+
+    @pytest.mark.asyncio
+    async def test_cancellation_preserves_prefix_and_stops(self):
+        batches: list[list[dict[str, Any]]] = []
+        saved: list[dict[str, Any]] = []
+        second_started = asyncio.Event()
+
+        async def capture(request: httpx.Request) -> httpx.Response:
+            batch = json.loads(request.content)["items"]
+            batches.append(batch)
+            if len(batches) == 2:
+                second_started.set()
+                await asyncio.Future()
+            saved.extend(batch)
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        items: list[TResponseInputItem] = [
+            {"role": "user", "content": f"message {i}"} for i in range(41)
+        ]
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+        ) as client:
+            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            write = asyncio.create_task(session.add_items(items))
+            try:
+                await asyncio.wait_for(second_started.wait(), timeout=5)
+            finally:
+                write.cancel()
+                with pytest.raises(asyncio.CancelledError):
+                    await write
+
+        assert batches == [items[:20], items[20:40]]
+        assert saved == items[:20]
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("stream", [False, True])
+    async def test_runner_batches_input_with_lazy_creation(self, stream: bool):
+        requests: list[tuple[str, str]] = []
+        batches: list[list[dict[str, Any]]] = []
+
+        def capture(request: httpx.Request) -> httpx.Response:
+            requests.append((request.method, request.url.path))
+            if request.url.path == "/v1/conversations":
+                assert json.loads(request.content) == {"items": []}
+                return httpx.Response(
+                    200, json={"id": "conv_test", "object": "conversation", "created_at": 0}
+                )
+            assert request.url.path == "/v1/conversations/conv_test/items"
+            if request.method == "POST":
+                batches.append(json.loads(request.content)["items"])
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        items: list[TResponseInputItem] = [
+            {"role": "user", "content": f"message {i}"} for i in range(41)
+        ]
+        model = ScriptedModel()
+        model.enqueue([get_text_message("done")])
+        agent = Agent(name="test", model=model)
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+        ) as client:
+            session = OpenAIConversationsSession(openai_client=client)
+            if stream:
+                result = Runner.run_streamed(
+                    agent, items, session=session, run_config=RunConfig(tracing_disabled=True)
+                )
+                async for _ in result.stream_events():
+                    pass
+                assert result.final_output == "done"
+            else:
+                result_sync = await Runner.run(
+                    agent, items, session=session, run_config=RunConfig(tracing_disabled=True)
+                )
+                assert result_sync.final_output == "done"
+
+        assert requests.count(("POST", "/v1/conversations")) == 1
+        assert [len(batch) for batch in batches] == [20, 20, 1, 1]
+        assert [item for batch in batches[:3] for item in batch] == items
+        assert batches[3][0]["content"][0]["text"] == "done"
+
+
 class TestOpenAIConversationsSessionRunnerIntegration:
     """Test integration with Agent Runner using simple mocking."""
 

From 4b3a79b717a310a5c7d719eb81d0a755399849b0 Mon Sep 17 00:00:00 2001
From: Justin Beckwith <jbeckwith@openai.com>
Date: Wed, 7 Oct 2026 10:19:24 -0700
Subject: [PATCH 2/9] fix: serialize conversation session mutations across
 batches

---
 .../memory/openai_conversations_session.py    | 40 +++++----
 .../test_openai_conversations_session.py      | 81 ++++++++++++++++++-
 2 files changed, 100 insertions(+), 21 deletions(-)

diff --git a/src/agents/memory/openai_conversations_session.py b/src/agents/memory/openai_conversations_session.py
index fe9713fdd2..9e6a6f4310 100644
--- a/src/agents/memory/openai_conversations_session.py
+++ b/src/agents/memory/openai_conversations_session.py
@@ -36,6 +36,7 @@ def __init__(
     ):
         self._session_id: str | None = conversation_id
         self._session_id_lock = asyncio.Lock()
+        self._mutation_lock = asyncio.Lock()
         self.session_settings = (
             coerce_session_settings(session_settings)
             if session_settings is not None
@@ -114,6 +115,9 @@ async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
     async def add_items(self, items: list[TResponseInputItem]) -> None:
         """Append items in order, in requests of at most 20 items each.
 
+        Mutations through this session instance are serialized. Separate instances
+        or external writers require application-level coordination.
+
         Writes spanning multiple requests are not atomic. If a request fails or
         is cancelled, earlier batches remain saved and later batches are not
         sent. The original exception propagates. Before retrying, callers must
@@ -123,27 +127,29 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
         if not items:
             return
 
-        session_id = await self._get_session_id()
-        # The Conversations items-create endpoint accepts up to 20 items per request.
-        for offset in range(0, len(items), 20):
-            await self._openai_client.conversations.items.create(
-                conversation_id=session_id,
-                items=items[offset : offset + 20],
-            )
+        async with self._mutation_lock:
+            session_id = await self._get_session_id()
+            # The Conversations items-create endpoint accepts up to 20 items per request.
+            for offset in range(0, len(items), 20):
+                await self._openai_client.conversations.items.create(
+                    conversation_id=session_id,
+                    items=items[offset : offset + 20],
+                )
 
     async def pop_item(self) -> TResponseInputItem | None:
-        session_id = await self._get_session_id()
-        items = await self.get_items(limit=1)
-        if not items:
-            return None
-        item_id: str = str(items[0]["id"])  # type: ignore [typeddict-item]
-        await self._openai_client.conversations.items.delete(
-            conversation_id=session_id, item_id=item_id
-        )
-        return items[0]
+        async with self._mutation_lock:
+            session_id = await self._get_session_id()
+            items = await self.get_items(limit=1)
+            if not items:
+                return None
+            item_id: str = str(items[0]["id"])  # type: ignore [typeddict-item]
+            await self._openai_client.conversations.items.delete(
+                conversation_id=session_id, item_id=item_id
+            )
+            return items[0]
 
     async def clear_session(self) -> None:
-        async with self._session_id_lock:
+        async with self._mutation_lock, self._session_id_lock:
             if self._session_id is None:
                 return
 
diff --git a/tests/memory/test_openai_conversations_session.py b/tests/memory/test_openai_conversations_session.py
index b2a6028e84..b01d8e5220 100644
--- a/tests/memory/test_openai_conversations_session.py
+++ b/tests/memory/test_openai_conversations_session.py
@@ -570,6 +570,77 @@ async def slow_delete(*args: Any, **kwargs: Any) -> Any:
 
 
 class TestOpenAIConversationsSessionBatches:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("mutation", ["add", "pop", "clear"])
+    async def test_mutations_wait_for_complete_append(self, mock_openai_client, mutation: str):
+        stored: list[TResponseInputItem] = []
+        first_started = asyncio.Event()
+        release_first = asyncio.Event()
+        other_started = asyncio.Event()
+        items: list[TResponseInputItem] = [
+            cast(TResponseInputItem, {"role": "user", "content": f"message {i}", "id": f"msg_{i}"})
+            for i in range(21)
+        ]
+        other: TResponseInputItem = {"role": "user", "content": "other"}
+
+        async def create(*, conversation_id, items):
+            stored.extend(items)
+            if len(stored) == 20:
+                first_started.set()
+                await release_first.wait()
+
+        async def delete_item(*, conversation_id, item_id):
+            assert stored[-1]["id"] == item_id
+            stored.pop()
+
+        async def delete_conversation(*, conversation_id):
+            stored.clear()
+
+        mock_openai_client.conversations.items.create.side_effect = create
+        mock_openai_client.conversations.items.delete.side_effect = delete_item
+        mock_openai_client.conversations.delete.side_effect = delete_conversation
+        session = OpenAIConversationsSession(
+            conversation_id="conv_test", openai_client=mock_openai_client
+        )
+        with patch.object(session, "get_items", side_effect=lambda limit: stored[-limit:]):
+
+            async def mutate():
+                other_started.set()
+                if mutation == "add":
+                    await session.add_items([other])
+                elif mutation == "pop":
+                    return await session.pop_item()
+                else:
+                    await session.clear_session()
+                return None
+
+            append = asyncio.create_task(session.add_items(items))
+            follower = None
+            try:
+                await asyncio.wait_for(first_started.wait(), timeout=5)
+                follower = asyncio.create_task(mutate())
+                await asyncio.wait_for(other_started.wait(), timeout=5)
+                assert not follower.done()
+                assert stored == items[:20]
+                release_first.set()
+                await append
+                result = await follower
+            finally:
+                release_first.set()
+                tasks = [append] + ([follower] if follower is not None else [])
+                for task in tasks:
+                    task.cancel()
+                await asyncio.gather(*tasks, return_exceptions=True)
+
+        if mutation == "add":
+            assert stored == items + [other]
+        elif mutation == "pop":
+            assert result == items[-1]
+            assert stored == items[:-1]
+        else:
+            assert stored == []
+            assert session._session_id is None
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         ("count", "expected_sizes"), [(0, []), (20, [20]), (21, [20, 1]), (41, [20, 20, 1])]
@@ -619,9 +690,10 @@ def capture(request: httpx.Request) -> httpx.Response:
             with pytest.raises(BadRequestError, match="synthetic failure") as caught:
                 await session.add_items(items)
             assert caught.value.status_code == 400
+            await session.add_items(items[40:])
 
-        assert batches == [items[:20], items[20:40]]
-        assert saved == items[:20]
+        assert batches == [items[:20], items[20:40], items[40:]]
+        assert saved == items[:20] + items[40:]
 
     @pytest.mark.asyncio
     async def test_cancellation_preserves_prefix_and_stops(self):
@@ -653,9 +725,10 @@ async def capture(request: httpx.Request) -> httpx.Response:
                 write.cancel()
                 with pytest.raises(asyncio.CancelledError):
                     await write
+            await session.add_items(items[40:])
 
-        assert batches == [items[:20], items[20:40]]
-        assert saved == items[:20]
+        assert batches == [items[:20], items[20:40], items[40:]]
+        assert saved == items[:20] + items[40:]
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("stream", [False, True])

From d53264ca0bd83c967d476b574a0529c490cb326e Mon Sep 17 00:00:00 2001
From: Justin Beckwith <jbeckwith@openai.com>
Date: Wed, 7 Oct 2026 10:32:28 -0700
Subject: [PATCH 3/9] fix: serialize conversation reads with batched writes

---
 .../memory/openai_conversations_session.py    |  9 +-
 .../test_openai_conversations_session.py      | 83 +++++++++++--------
 2 files changed, 54 insertions(+), 38 deletions(-)

diff --git a/src/agents/memory/openai_conversations_session.py b/src/agents/memory/openai_conversations_session.py
index 9e6a6f4310..7cdcc92336 100644
--- a/src/agents/memory/openai_conversations_session.py
+++ b/src/agents/memory/openai_conversations_session.py
@@ -84,6 +84,11 @@ async def _clear_session_id(self) -> None:
         self._session_id = None
 
     async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
+        async with self._mutation_lock:
+            return await self._get_items(limit)
+
+    async def _get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
+        """Read history while the caller holds the instance's mutation lock."""
         session_id = await self._get_session_id()
 
         session_limit = resolve_session_limit(limit, self.session_settings)
@@ -115,7 +120,7 @@ async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
     async def add_items(self, items: list[TResponseInputItem]) -> None:
         """Append items in order, in requests of at most 20 items each.
 
-        Mutations through this session instance are serialized. Separate instances
+        Reads and mutations through this session instance are serialized. Separate instances
         or external writers require application-level coordination.
 
         Writes spanning multiple requests are not atomic. If a request fails or
@@ -139,7 +144,7 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
     async def pop_item(self) -> TResponseInputItem | None:
         async with self._mutation_lock:
             session_id = await self._get_session_id()
-            items = await self.get_items(limit=1)
+            items = await self._get_items(limit=1)
             if not items:
                 return None
             item_id: str = str(items[0]["id"])  # type: ignore [typeddict-item]
diff --git a/tests/memory/test_openai_conversations_session.py b/tests/memory/test_openai_conversations_session.py
index b01d8e5220..108511c6b1 100644
--- a/tests/memory/test_openai_conversations_session.py
+++ b/tests/memory/test_openai_conversations_session.py
@@ -337,10 +337,10 @@ async def test_pop_item_with_items(self, mock_openai_client):
             conversation_id="test_id", openai_client=mock_openai_client
         )
 
-        # Mock get_items to return one item
+        # Mock the already-locked read to return one item
         latest_item = {"id": "item_123", "role": "assistant", "content": "Latest message"}
 
-        with patch.object(session, "get_items", return_value=[latest_item]):
+        with patch.object(session, "_get_items", return_value=[latest_item]):
             popped_item = await session.pop_item()
 
             assert popped_item == latest_item
@@ -355,8 +355,8 @@ async def test_pop_item_empty_session(self, mock_openai_client):
             conversation_id="test_id", openai_client=mock_openai_client
         )
 
-        # Mock get_items to return empty list
-        with patch.object(session, "get_items", return_value=[]):
+        # Mock the already-locked read to return empty list
+        with patch.object(session, "_get_items", return_value=[]):
             popped_item = await session.pop_item()
 
             assert popped_item is None
@@ -571,8 +571,8 @@ async def slow_delete(*args: Any, **kwargs: Any) -> Any:
 
 class TestOpenAIConversationsSessionBatches:
     @pytest.mark.asyncio
-    @pytest.mark.parametrize("mutation", ["add", "pop", "clear"])
-    async def test_mutations_wait_for_complete_append(self, mock_openai_client, mutation: str):
+    @pytest.mark.parametrize("mutation", ["add", "pop", "clear", "read"])
+    async def test_operations_wait_for_complete_append(self, mock_openai_client, mutation: str):
         stored: list[TResponseInputItem] = []
         first_started = asyncio.Event()
         release_first = asyncio.Event()
@@ -602,38 +602,49 @@ async def delete_conversation(*, conversation_id):
         session = OpenAIConversationsSession(
             conversation_id="conv_test", openai_client=mock_openai_client
         )
-        with patch.object(session, "get_items", side_effect=lambda limit: stored[-limit:]):
-
-            async def mutate():
-                other_started.set()
-                if mutation == "add":
-                    await session.add_items([other])
-                elif mutation == "pop":
-                    return await session.pop_item()
-                else:
-                    await session.clear_session()
-                return None
-
-            append = asyncio.create_task(session.add_items(items))
-            follower = None
-            try:
-                await asyncio.wait_for(first_started.wait(), timeout=5)
-                follower = asyncio.create_task(mutate())
-                await asyncio.wait_for(other_started.wait(), timeout=5)
-                assert not follower.done()
-                assert stored == items[:20]
-                release_first.set()
-                await append
-                result = await follower
-            finally:
-                release_first.set()
-                tasks = [append] + ([follower] if follower is not None else [])
-                for task in tasks:
-                    task.cancel()
-                await asyncio.gather(*tasks, return_exceptions=True)
+
+        async def list_items(*, conversation_id, order):
+            snapshot = stored[:] if order == "asc" else list(reversed(stored))
+            for item in snapshot:
+                yield MagicMock(model_dump=MagicMock(return_value=item))
+
+        mock_openai_client.conversations.items.list = MagicMock(side_effect=list_items)
+
+        async def mutate():
+            other_started.set()
+            if mutation == "add":
+                await session.add_items([other])
+            elif mutation == "pop":
+                return await session.pop_item()
+            elif mutation == "read":
+                return await session.get_items()
+            else:
+                await session.clear_session()
+            return None
+
+        append = asyncio.create_task(session.add_items(items))
+        follower = None
+        try:
+            await asyncio.wait_for(first_started.wait(), timeout=5)
+            follower = asyncio.create_task(mutate())
+            await asyncio.wait_for(other_started.wait(), timeout=5)
+            assert not follower.done()
+            assert stored == items[:20]
+            release_first.set()
+            await append
+            result = await asyncio.wait_for(follower, timeout=5)
+        finally:
+            release_first.set()
+            tasks = [append] + ([follower] if follower is not None else [])
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
 
         if mutation == "add":
             assert stored == items + [other]
+        elif mutation == "read":
+            assert result == items
+            assert stored == items
         elif mutation == "pop":
             assert result == items[-1]
             assert stored == items[:-1]
@@ -948,7 +959,7 @@ async def test_invalid_item_id_in_pop_item(self, mock_openai_client):
         # Mock item without ID
         invalid_item = {"role": "assistant", "content": "No ID"}
 
-        with patch.object(session, "get_items", return_value=[invalid_item]):
+        with patch.object(session, "_get_items", return_value=[invalid_item]):
             # This should raise a KeyError because 'id' field is missing
             with pytest.raises(KeyError, match="'id'"):
                 await session.pop_item()

From ea1da4f367ff2788432720923e99a996362a0a4e Mon Sep 17 00:00:00 2001
From: Justin Beckwith <jbeckwith@openai.com>
Date: Wed, 7 Oct 2026 11:13:28 -0700
Subject: [PATCH 4/9] fix: reject ambiguous partial session writes on resume

---
 .../run_internal/session_persistence.py       |  14 +-
 .../test_openai_conversations_session.py      | 202 +++++++++++++++++-
 2 files changed, 213 insertions(+), 3 deletions(-)

diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py
index 26dc6cdc7c..a16c5d31bd 100644
--- a/src/agents/run_internal/session_persistence.py
+++ b/src/agents/run_internal/session_persistence.py
@@ -1496,7 +1496,8 @@ def digests(items: Sequence[TResponseInputItem]) -> list[str]:
             pending["before"] = digests(tail)
             append = True
         else:
-            expected = (before or []) + digests(pending["items"])
+            batch = digests(pending["items"])
+            expected = (before or []) + batch
             observed_generation: int | None = None
             get_with_generation = getattr(session, "_get_items_with_generation", None)
             if wrapper is not None and callable(get_with_generation):
@@ -1509,7 +1510,16 @@ def digests(items: Sequence[TResponseInputItem]) -> list[str]:
             observed = digests(tail)
             committed = observed == expected
             unchanged = observed[-len(before) :] == before if before else not observed
-            if (acknowledged and not committed) or (not acknowledged and committed == unchanged):
+            # Repeated content can make a partial append look unchanged or fully committed.
+            # The saved tail plus any proper batch prefix fits in this read window.
+            # Do not settle an unacknowledged write when that evidence also admits a prefix.
+            partial = not acknowledged and any(
+                observed[-(len(before or []) + count) :] == (before or []) + batch[:count]
+                for count in range(1, len(batch))
+            )
+            if (acknowledged and not committed) or (
+                not acknowledged and (partial or committed == unchanged)
+            ):
                 raise UserError(
                     "Cannot reconcile the pending Session write: history changed or is "
                     "ambiguous. Repair the original Session before resuming; do not rerun "
diff --git a/tests/memory/test_openai_conversations_session.py b/tests/memory/test_openai_conversations_session.py
index 108511c6b1..f6be34bab5 100644
--- a/tests/memory/test_openai_conversations_session.py
+++ b/tests/memory/test_openai_conversations_session.py
@@ -14,18 +14,26 @@
 
 from agents import (
     Agent,
+    GuardrailFunctionOutput,
+    HandoffInputData,
+    MessageOutputItem,
     ProgrammaticToolCallingTool,
     RunConfig,
     Runner,
+    RunState,
     TResponseInputItem,
     function_tool,
+    handoff,
+    output_guardrail,
 )
+from agents.decorators import tool
+from agents.exceptions import UserError
 from agents.memory.openai_conversations_session import (
     OpenAIConversationsSession,
     start_openai_conversations_session,
 )
 from agents.testing import ScriptedModel
-from tests.test_responses import get_text_message
+from tests.test_responses import get_function_tool_call, get_text_message
 
 
 @pytest.fixture
@@ -788,6 +796,198 @@ def capture(request: httpx.Request) -> httpx.Response:
         assert [item for batch in batches[:3] for item in batch] == items
         assert batches[3][0]["content"][0]["text"] == "done"
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("stream", [False, True])
+    @pytest.mark.parametrize(
+        "history_size,failure",
+        [(1, "partial"), (40, "partial"), (1, "before"), (1, "lost-ack")],
+    )
+    async def test_runner_resume_does_not_replay_partial_repeated_history(
+        self, stream: bool, history_size: int, failure: str
+    ):
+        """A repeated boundary must not disguise a partial append as unchanged history."""
+        messages = [
+            get_text_message(f"message {i}").model_copy(update={"id": f"msg_{i}"})
+            for i in range(19)
+        ] + [get_text_message("S").model_copy(update={"id": "msg_19"})]
+        # A long periodic history exercises the bounded saved tail as well as the
+        # original one-item counterexample. Provider IDs differ at every occurrence.
+        stored: list[dict[str, Any]] = [
+            {**messages[i % 20].model_dump(exclude_none=True), "id": f"old_{i}"}
+            for i in range(history_size - 1)
+        ]
+        writes: list[list[dict[str, Any]]] = []
+        effects: list[str] = []
+
+        def capture(request: httpx.Request) -> httpx.Response:
+            if request.method == "GET":
+                data = stored[::-1] if request.url.params.get("order") == "desc" else stored[:]
+                return httpx.Response(200, json={"object": "list", "data": data, "has_more": False})
+            batch = json.loads(request.content)["items"]
+            writes.append(batch)
+            fail = len(writes) == (2 if failure == "before" else 3)
+            if not fail or failure == "lost-ack":
+                for item in batch:
+                    stored.append({**item, "id": f"item_{len(stored)}"})
+            if fail:
+                return httpx.Response(
+                    400,
+                    json={
+                        "error": {"message": "synthetic failure", "type": "invalid_request_error"}
+                    },
+                )
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        @tool(needs_approval=True)
+        async def lookup() -> str:
+            effects.append("lookup")
+            return "found"
+
+        @tool
+        async def finish() -> str:
+            return "done"
+
+        @output_guardrail
+        async def gate(ctx, agent, output):
+            return GuardrailFunctionOutput(output_info=None, tripwire_triggered=False)
+
+        model = ScriptedModel(
+            [
+                messages + [get_function_tool_call("lookup", "{}", call_id="lookup_1")],
+                [get_function_tool_call("finish", "{}", call_id="finish_1")],
+            ]
+        )
+        agent = Agent(
+            name="test",
+            model=model,
+            tools=[lookup, finish],
+            output_guardrails=[gate],
+            tool_use_behavior={"stop_at_tool_names": ["finish"]},
+        )
+        config = RunConfig(tracing_disabled=True)
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            max_retries=0,
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+        ) as client:
+            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+
+            async def run(value):
+                if stream:
+                    result = Runner.run_streamed(agent, value, session=session, run_config=config)
+                    async for _ in result.stream_events():
+                        pass
+                    return result
+                return await Runner.run(agent, value, session=session, run_config=config)
+
+            original = get_text_message("S").model_dump(exclude_none=True)
+            original.pop("id")
+            paused = await run([cast(TResponseInputItem, original)])
+            assert len(stored) == history_size
+            state = paused.to_state()
+            state.approve(state.get_interruptions()[0])
+            if stream:
+                failed = Runner.run_streamed(agent, state, session=session, run_config=config)
+                with pytest.raises(BadRequestError, match="synthetic failure"):
+                    async for _ in failed.stream_events():
+                        pass
+                state = failed.to_state()
+            else:
+                with pytest.raises(BadRequestError, match="synthetic failure"):
+                    await run(state)
+            assert effects == ["lookup"]
+            assert len(model.calls) == 1
+            state = await RunState.from_json(agent, json.loads(json.dumps(state.to_json())))
+            if failure == "partial":
+                assert len(stored) == history_size + 20
+                snapshot = stored[:]
+                with pytest.raises(UserError, match="Cannot reconcile the pending Session write"):
+                    await run(state)
+                assert stored == snapshot
+                assert [len(batch) for batch in writes] == [1, 20, 2]
+                assert len(model.calls) == 1
+            else:
+                result = await run(state)
+                assert result.final_output == "done"
+                texts = [item["content"][0]["text"] for item in stored if item["type"] == "message"]
+                assert texts == ["S"] + [f"message {i}" for i in range(19)] + ["S"]
+                assert len(model.calls) == 2
+            assert effects == ["lookup"]
+
+    @pytest.mark.asyncio
+    async def test_runner_resume_rejects_partial_history_that_looks_fully_committed(self):
+        """A bounded periodic tail cannot prove every item in a failed append was saved."""
+        messages = [
+            get_text_message("ABC"[i % 3]).model_copy(update={"id": f"msg_{i}"}) for i in range(30)
+        ]
+        stored: list[dict[str, Any]] = [
+            {**item.model_dump(exclude_none=True), "id": f"old_{i}"}
+            for i, item in enumerate(messages[:29])
+        ]
+        writes: list[list[dict[str, Any]]] = []
+
+        def capture(request: httpx.Request) -> httpx.Response:
+            if request.method == "GET":
+                data = stored[::-1] if request.url.params.get("order") == "desc" else stored[:]
+                return httpx.Response(200, json={"object": "list", "data": data, "has_more": False})
+            batch = json.loads(request.content)["items"]
+            writes.append(batch)
+            if len(writes) == 3:
+                return httpx.Response(
+                    400,
+                    json={
+                        "error": {"message": "synthetic failure", "type": "invalid_request_error"}
+                    },
+                )
+            for item in batch:
+                stored.append({**item, "id": f"item_{len(stored)}"})
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        def retain_messages(data: HandoffInputData) -> HandoffInputData:
+            return data.clone(
+                new_items=tuple(
+                    item for item in data.new_items if isinstance(item, MessageOutputItem)
+                )
+            )
+
+        model = ScriptedModel(
+            [
+                messages[:23]
+                + [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff_1")],
+                [get_text_message("done")],
+            ]
+        )
+        delegate = Agent(name="delegate", model=model)
+        agent = Agent(
+            name="test", model=model, handoffs=[handoff(delegate, input_filter=retain_messages)]
+        )
+        config = RunConfig(tracing_disabled=True)
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            max_retries=0,
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+        ) as client:
+            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            original = messages[29].model_dump(exclude_none=True)
+            original.pop("id")
+            failed = Runner.run_streamed(
+                agent, [cast(TResponseInputItem, original)], session=session, run_config=config
+            )
+            with pytest.raises(BadRequestError, match="synthetic failure"):
+                async for _ in failed.stream_events():
+                    pass
+            assert len(stored) == 50
+            assert [len(batch) for batch in writes] == [1, 20, 3]
+            snapshot = stored[:]
+            state = await RunState.from_json(agent, failed.to_state().to_json())
+            with pytest.raises(UserError, match="Cannot reconcile the pending Session write"):
+                await Runner.run(agent, state, session=session, run_config=config)
+            assert stored == snapshot
+            assert len(writes) == 3
+            assert len(model.calls) == 1
+            # The missing final three messages remain pending rather than being discarded.
+            assert len(state.to_json()["pending_session_write"]["items"]) == 23
+
 
 class TestOpenAIConversationsSessionRunnerIntegration:
     """Test integration with Agent Runner using simple mocking."""

From a674687b440ae68f49dabc9fd6339c41b4fe5633 Mon Sep 17 00:00:00 2001
From: Justin Beckwith <jbeckwith@openai.com>
Date: Wed, 7 Oct 2026 11:32:26 -0700
Subject: [PATCH 5/9] fix: preserve atomic session recovery and use complete
 history counts

---
 .../memory/openai_conversations_session.py    |  6 +-
 .../run_internal/session_persistence.py       | 32 ++++++--
 .../test_openai_conversations_session.py      | 77 ++++++++++++++-----
 3 files changed, 87 insertions(+), 28 deletions(-)

diff --git a/src/agents/memory/openai_conversations_session.py b/src/agents/memory/openai_conversations_session.py
index 7cdcc92336..5caa7d1cb7 100644
--- a/src/agents/memory/openai_conversations_session.py
+++ b/src/agents/memory/openai_conversations_session.py
@@ -11,6 +11,8 @@
 from .session import SessionABC, _await_mutation
 from .session_settings import SessionSettings, coerce_session_settings, resolve_session_limit
 
+_MAX_ITEMS_PER_REQUEST = 20
+
 
 async def start_openai_conversations_session(openai_client: AsyncOpenAI | None = None) -> str:
     _maybe_openai_client = openai_client
@@ -135,10 +137,10 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
         async with self._mutation_lock:
             session_id = await self._get_session_id()
             # The Conversations items-create endpoint accepts up to 20 items per request.
-            for offset in range(0, len(items), 20):
+            for offset in range(0, len(items), _MAX_ITEMS_PER_REQUEST):
                 await self._openai_client.conversations.items.create(
                     conversation_id=session_id,
-                    items=items[offset : offset + 20],
+                    items=items[offset : offset + _MAX_ITEMS_PER_REQUEST],
                 )
 
     async def pop_item(self) -> TResponseInputItem | None:
diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py
index a16c5d31bd..bd982d41f4 100644
--- a/src/agents/run_internal/session_persistence.py
+++ b/src/agents/run_internal/session_persistence.py
@@ -39,7 +39,10 @@
     SessionSettings,
     is_openai_responses_compaction_aware_session,
 )
-from ..memory.openai_conversations_session import OpenAIConversationsSession
+from ..memory.openai_conversations_session import (
+    _MAX_ITEMS_PER_REQUEST,
+    OpenAIConversationsSession,
+)
 from ..memory.session import _call_session_method, _get_session_wrapper
 from ..models.fake_id import FAKE_RESPONSES_ID
 from ..run_context import RunContextWrapper
@@ -1510,13 +1513,26 @@ def digests(items: Sequence[TResponseInputItem]) -> list[str]:
             observed = digests(tail)
             committed = observed == expected
             unchanged = observed[-len(before) :] == before if before else not observed
-            # Repeated content can make a partial append look unchanged or fully committed.
-            # The saved tail plus any proper batch prefix fits in this read window.
-            # Do not settle an unacknowledged write when that evidence also admits a prefix.
-            partial = not acknowledged and any(
-                observed[-(len(before or []) + count) :] == (before or []) + batch[:count]
-                for count in range(1, len(batch))
-            )
+            # Only multi-request Conversations writes admit partially committed chunks.
+            # Other backends and single-request writes retain their atomic recovery contract.
+            partial = False
+            if (
+                isinstance(session, OpenAIConversationsSession)
+                and not acknowledged
+                and len(batch) > _MAX_ITEMS_PER_REQUEST
+            ):
+                before_count = len(before or [])
+                complete_before = before_count < len(batch) + 1
+                # A short snapshot captured all prior history: count must agree with
+                # zero progress as well as each candidate chunk prefix. A full-size
+                # snapshot may omit older items, so repeated content remains ambiguous.
+                if complete_before:
+                    unchanged = unchanged and len(observed) == before_count
+                partial = any(
+                    (not complete_before or len(observed) == before_count + count)
+                    and observed[-(before_count + count) :] == (before or []) + batch[:count]
+                    for count in range(_MAX_ITEMS_PER_REQUEST, len(batch), _MAX_ITEMS_PER_REQUEST)
+                )
             if (acknowledged and not committed) or (
                 not acknowledged and (partial or committed == unchanged)
             ):
diff --git a/tests/memory/test_openai_conversations_session.py b/tests/memory/test_openai_conversations_session.py
index f6be34bab5..278c2803cf 100644
--- a/tests/memory/test_openai_conversations_session.py
+++ b/tests/memory/test_openai_conversations_session.py
@@ -34,6 +34,7 @@
 )
 from agents.testing import ScriptedModel
 from tests.test_responses import get_function_tool_call, get_text_message
+from tests.utils.simple_session import IdStrippingSession
 
 
 @pytest.fixture
@@ -915,14 +916,20 @@ async def run(value):
             assert effects == ["lookup"]
 
     @pytest.mark.asyncio
-    async def test_runner_resume_rejects_partial_history_that_looks_fully_committed(self):
+    @pytest.mark.parametrize(
+        "atomic,output_count,history_size",
+        [(False, 23, 30), (True, 23, 30), (False, 2, 30), (False, 23, 1), (False, 21, 1)],
+    )
+    async def test_runner_resume_handles_periodic_history_for_session_atomicity(
+        self, atomic: bool, output_count: int, history_size: int
+    ):
         """A bounded periodic tail cannot prove every item in a failed append was saved."""
         messages = [
             get_text_message("ABC"[i % 3]).model_copy(update={"id": f"msg_{i}"}) for i in range(30)
         ]
         stored: list[dict[str, Any]] = [
             {**item.model_dump(exclude_none=True), "id": f"old_{i}"}
-            for i, item in enumerate(messages[:29])
+            for i, item in enumerate(messages[: history_size - 1])
         ]
         writes: list[list[dict[str, Any]]] = []
 
@@ -932,17 +939,26 @@ def capture(request: httpx.Request) -> httpx.Response:
                 return httpx.Response(200, json={"object": "list", "data": data, "has_more": False})
             batch = json.loads(request.content)["items"]
             writes.append(batch)
-            if len(writes) == 3:
+            fail = len(writes) == (2 if output_count <= 20 else 3)
+            if not fail or output_count <= 20 or history_size == 1:
+                for item in batch:
+                    stored.append({**item, "id": f"item_{len(stored)}"})
+            if fail:
                 return httpx.Response(
                     400,
                     json={
                         "error": {"message": "synthetic failure", "type": "invalid_request_error"}
                     },
                 )
-            for item in batch:
-                stored.append({**item, "id": f"item_{len(stored)}"})
             return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
 
+        class LostAckSession(IdStrippingSession):
+            async def add_items(self, items: list[TResponseInputItem]) -> None:
+                writes.append(cast(list[dict[str, Any]], items))
+                await super().add_items(items)
+                if len(writes) == 2:
+                    raise RuntimeError("synthetic failure")
+
         def retain_messages(data: HandoffInputData) -> HandoffInputData:
             return data.clone(
                 new_items=tuple(
@@ -952,7 +968,7 @@ def retain_messages(data: HandoffInputData) -> HandoffInputData:
 
         model = ScriptedModel(
             [
-                messages[:23]
+                messages[:output_count]
                 + [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff_1")],
                 [get_text_message("done")],
             ]
@@ -967,26 +983,51 @@ def retain_messages(data: HandoffInputData) -> HandoffInputData:
             max_retries=0,
             http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
         ) as client:
-            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            session = (
+                LostAckSession(history=cast(list[TResponseInputItem], stored))
+                if atomic
+                else OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            )
             original = messages[29].model_dump(exclude_none=True)
             original.pop("id")
             failed = Runner.run_streamed(
                 agent, [cast(TResponseInputItem, original)], session=session, run_config=config
             )
-            with pytest.raises(BadRequestError, match="synthetic failure"):
+            with pytest.raises(
+                RuntimeError if atomic else BadRequestError, match="synthetic failure"
+            ):
                 async for _ in failed.stream_events():
                     pass
-            assert len(stored) == 50
-            assert [len(batch) for batch in writes] == [1, 20, 3]
-            snapshot = stored[:]
+            snapshot = await session.get_items()
+            assert len(snapshot) == (
+                history_size + output_count
+                if atomic or output_count <= 20 or history_size == 1
+                else history_size + 20
+            )
+            assert [len(batch) for batch in writes] == (
+                [1, output_count] if atomic or output_count <= 20 else [1, 20, output_count - 20]
+            )
             state = await RunState.from_json(agent, failed.to_state().to_json())
-            with pytest.raises(UserError, match="Cannot reconcile the pending Session write"):
-                await Runner.run(agent, state, session=session, run_config=config)
-            assert stored == snapshot
-            assert len(writes) == 3
-            assert len(model.calls) == 1
-            # The missing final three messages remain pending rather than being discarded.
-            assert len(state.to_json()["pending_session_write"]["items"]) == 23
+            if atomic or output_count <= 20 or history_size == 1:
+                resumed = await Runner.run(agent, state, session=session, run_config=config)
+                assert resumed.final_output == "done"
+                history = await session.get_items()
+                assert history[:-1] == snapshot
+                assert len(history) == history_size + output_count + 1
+                assert [len(batch) for batch in writes] == (
+                    [1, output_count, 1]
+                    if atomic or output_count <= 20
+                    else [1, 20, output_count - 20, 1]
+                )
+                assert len(model.calls) == 2
+            else:
+                with pytest.raises(UserError, match="Cannot reconcile the pending Session write"):
+                    await Runner.run(agent, state, session=session, run_config=config)
+                assert await session.get_items() == snapshot
+                assert len(writes) == 3
+                assert len(model.calls) == 1
+                # Missing messages remain pending rather than being discarded.
+                assert len(state.to_json()["pending_session_write"]["items"]) == 23
 
 
 class TestOpenAIConversationsSessionRunnerIntegration:

From 972d573d3e4a36474d0a06e856e5d700ea8547eb Mon Sep 17 00:00:00 2001
From: Justin Beckwith <jbeckwith@openai.com>
Date: Wed, 7 Oct 2026 12:10:45 -0700
Subject: [PATCH 6/9] fix: settle cancelled conversation batches before
 releasing ownership

---
 .../memory/openai_conversations_session.py    | 11 ++-
 .../test_openai_conversations_session.py      | 87 +++++++++++++++++--
 2 files changed, 87 insertions(+), 11 deletions(-)

diff --git a/src/agents/memory/openai_conversations_session.py b/src/agents/memory/openai_conversations_session.py
index 5caa7d1cb7..a8183b073f 100644
--- a/src/agents/memory/openai_conversations_session.py
+++ b/src/agents/memory/openai_conversations_session.py
@@ -127,7 +127,10 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
 
         Writes spanning multiple requests are not atomic. If a request fails or
         is cancelled, earlier batches remain saved and later batches are not
-        sent. The original exception propagates. Before retrying, callers must
+        sent. For multi-request writes, cancellation waits for the current request
+        to complete or fail before releasing the instance lock and propagating
+        cancellation. Single-request writes retain their existing cancellation behavior.
+        The original exception propagates. Before retrying, callers must
         reconcile the remote history; retrying the entire list can duplicate
         items that were already saved, including an unacknowledged request.
         """
@@ -138,10 +141,14 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
             session_id = await self._get_session_id()
             # The Conversations items-create endpoint accepts up to 20 items per request.
             for offset in range(0, len(items), _MAX_ITEMS_PER_REQUEST):
-                await self._openai_client.conversations.items.create(
+                request = self._openai_client.conversations.items.create(
                     conversation_id=session_id,
                     items=items[offset : offset + _MAX_ITEMS_PER_REQUEST],
                 )
+                if len(items) > _MAX_ITEMS_PER_REQUEST:
+                    await _await_mutation(request)
+                else:
+                    await request
 
     async def pop_item(self) -> TResponseInputItem | None:
         async with self._mutation_lock:
diff --git a/tests/memory/test_openai_conversations_session.py b/tests/memory/test_openai_conversations_session.py
index 278c2803cf..13e32603b6 100644
--- a/tests/memory/test_openai_conversations_session.py
+++ b/tests/memory/test_openai_conversations_session.py
@@ -716,39 +716,108 @@ def capture(request: httpx.Request) -> httpx.Response:
         assert saved == items[:20] + items[40:]
 
     @pytest.mark.asyncio
-    async def test_cancellation_preserves_prefix_and_stops(self):
+    @pytest.mark.parametrize("request_fails", [False, True])
+    async def test_cancelled_batch_settles_before_queued_writer(self, request_fails: bool):
         batches: list[list[dict[str, Any]]] = []
         saved: list[dict[str, Any]] = []
         second_started = asyncio.Event()
+        release_request = asyncio.Event()
+        follower_started = asyncio.Event()
+        remote_tasks: list[asyncio.Task[None]] = []
+
+        async def remote_commit(batch: list[dict[str, Any]]) -> None:
+            await release_request.wait()
+            if not request_fails:
+                saved.extend(batch)
 
         async def capture(request: httpx.Request) -> httpx.Response:
             batch = json.loads(request.content)["items"]
             batches.append(batch)
             if len(batches) == 2:
+                # An accepted server mutation survives cancellation of the HTTP caller.
+                remote = asyncio.create_task(remote_commit(batch))
+                remote_tasks.append(remote)
                 second_started.set()
-                await asyncio.Future()
-            saved.extend(batch)
+                await asyncio.shield(remote)
+                if request_fails:
+                    return httpx.Response(400, json={"error": {"message": "synthetic failure"}})
+            else:
+                saved.extend(batch)
             return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
 
         items: list[TResponseInputItem] = [
             {"role": "user", "content": f"message {i}"} for i in range(41)
         ]
+        survivor: list[TResponseInputItem] = [{"role": "user", "content": "queued writer"}]
         async with AsyncOpenAI(
             api_key="test-placeholder",
+            max_retries=0,
             http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
         ) as client:
             session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+
+            async def append_survivor() -> None:
+                follower_started.set()
+                await session.add_items(survivor)
+
             write = asyncio.create_task(session.add_items(items))
+            follower = None
             try:
                 await asyncio.wait_for(second_started.wait(), timeout=5)
+                follower = asyncio.create_task(append_survivor())
+                await asyncio.wait_for(follower_started.wait(), timeout=5)
+                write.cancel("original cancellation")
+                await asyncio.sleep(0)
+                write.cancel("repeated cancellation")
+                await asyncio.sleep(0)
+                assert not write.done()
+                assert not follower.done()
+                assert saved == items[:20]
+                assert batches == [items[:20], items[20:40]]
+                release_request.set()
+                with pytest.raises(asyncio.CancelledError) as cancelled:
+                    await asyncio.wait_for(write, timeout=5)
+                assert cancelled.value.args == ("original cancellation",)
+                await asyncio.wait_for(follower, timeout=5)
             finally:
-                write.cancel()
-                with pytest.raises(asyncio.CancelledError):
-                    await write
-            await session.add_items(items[40:])
+                release_request.set()
+                await asyncio.gather(
+                    write,
+                    *([follower] if follower is not None else []),
+                    *remote_tasks,
+                    return_exceptions=True,
+                )
 
-        assert batches == [items[:20], items[20:40], items[40:]]
-        assert saved == items[:20] + items[40:]
+        assert batches == [items[:20], items[20:40], survivor]
+        assert saved == (items[:20] if request_fails else items[:40]) + survivor
+
+    @pytest.mark.asyncio
+    async def test_single_request_cancellation_does_not_wait_for_response(self):
+        started = asyncio.Event()
+        release_request = asyncio.Event()
+
+        async def capture(request: httpx.Request) -> httpx.Response:
+            started.set()
+            await release_request.wait()
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+        ) as client:
+            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            write = asyncio.create_task(
+                session.add_items([{"role": "user", "content": "message"}] * 20)
+            )
+            try:
+                await asyncio.wait_for(started.wait(), timeout=5)
+                write.cancel()
+                done, _ = await asyncio.wait({write}, timeout=5)
+                assert write in done
+                assert write.cancelled()
+            finally:
+                release_request.set()
+                await asyncio.gather(write, return_exceptions=True)
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("stream", [False, True])

From d44fcf1366ff777349a8d2f1e6942c7e8c96f899 Mon Sep 17 00:00:00 2001
From: Justin Beckwith <jbeckwith@openai.com>
Date: Wed, 7 Oct 2026 12:20:02 -0700
Subject: [PATCH 7/9] test: inspect cancellation before Python 3.10 task
 propagation

---
 .../memory/test_openai_conversations_session.py | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/tests/memory/test_openai_conversations_session.py b/tests/memory/test_openai_conversations_session.py
index 13e32603b6..e11eebd8e2 100644
--- a/tests/memory/test_openai_conversations_session.py
+++ b/tests/memory/test_openai_conversations_session.py
@@ -760,7 +760,17 @@ async def append_survivor() -> None:
                 follower_started.set()
                 await session.add_items(survivor)
 
-            write = asyncio.create_task(session.add_items(items))
+            cancellations: list[asyncio.CancelledError] = []
+
+            async def append_cancelled() -> None:
+                try:
+                    await session.add_items(items)
+                except asyncio.CancelledError as exc:
+                    # Python 3.10 can drop the message when a task exposes cancellation.
+                    cancellations.append(exc)
+                    raise
+
+            write = asyncio.create_task(append_cancelled())
             follower = None
             try:
                 await asyncio.wait_for(second_started.wait(), timeout=5)
@@ -775,9 +785,10 @@ async def append_survivor() -> None:
                 assert saved == items[:20]
                 assert batches == [items[:20], items[20:40]]
                 release_request.set()
-                with pytest.raises(asyncio.CancelledError) as cancelled:
+                with pytest.raises(asyncio.CancelledError):
                     await asyncio.wait_for(write, timeout=5)
-                assert cancelled.value.args == ("original cancellation",)
+                assert len(cancellations) == 1
+                assert cancellations[0].args == ("original cancellation",)
                 await asyncio.wait_for(follower, timeout=5)
             finally:
                 release_request.set()

From 8ecddf7a3e9c80d2851ba960b9f13853cc3ef842 Mon Sep 17 00:00:00 2001
From: Justin Beckwith <jbeckwith@openai.com>
Date: Wed, 7 Oct 2026 12:54:38 -0700
Subject: [PATCH 8/9] fix: disable automatic retries for chunked conversation
 writes

---
 .../memory/openai_conversations_session.py    | 13 +++-
 .../test_openai_conversations_session.py      | 73 ++++++++++++++++++-
 2 files changed, 83 insertions(+), 3 deletions(-)

diff --git a/src/agents/memory/openai_conversations_session.py b/src/agents/memory/openai_conversations_session.py
index a8183b073f..6e3d6b1b58 100644
--- a/src/agents/memory/openai_conversations_session.py
+++ b/src/agents/memory/openai_conversations_session.py
@@ -130,6 +130,9 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
         sent. For multi-request writes, cancellation waits for the current request
         to complete or fail before releasing the instance lock and propagating
         cancellation. Single-request writes retain their existing cancellation behavior.
+        Automatic client retries are disabled for multi-request writes, even if
+        the supplied client enables retries, to avoid duplicating a saved chunk
+        after a lost response. Single-request writes retain the client's retry policy.
         The original exception propagates. Before retrying, callers must
         reconcile the remote history; retrying the entire list can duplicate
         items that were already saved, including an unacknowledged request.
@@ -139,13 +142,19 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
 
         async with self._mutation_lock:
             session_id = await self._get_session_id()
+            multiple_requests = len(items) > _MAX_ITEMS_PER_REQUEST
+            client = (
+                self._openai_client.with_options(max_retries=0)
+                if multiple_requests
+                else self._openai_client
+            )
             # The Conversations items-create endpoint accepts up to 20 items per request.
             for offset in range(0, len(items), _MAX_ITEMS_PER_REQUEST):
-                request = self._openai_client.conversations.items.create(
+                request = client.conversations.items.create(
                     conversation_id=session_id,
                     items=items[offset : offset + _MAX_ITEMS_PER_REQUEST],
                 )
-                if len(items) > _MAX_ITEMS_PER_REQUEST:
+                if multiple_requests:
                     await _await_mutation(request)
                 else:
                     await request
diff --git a/tests/memory/test_openai_conversations_session.py b/tests/memory/test_openai_conversations_session.py
index e11eebd8e2..8ddce719ba 100644
--- a/tests/memory/test_openai_conversations_session.py
+++ b/tests/memory/test_openai_conversations_session.py
@@ -9,7 +9,7 @@
 
 import httpx
 import pytest
-from openai import AsyncOpenAI, BadRequestError
+from openai import APIConnectionError, AsyncOpenAI, BadRequestError
 from openai.types.responses.response_output_item import Program, ProgramOutput
 
 from agents import (
@@ -41,6 +41,7 @@
 def mock_openai_client():
     """Create a mock OpenAI client for testing."""
     client = AsyncMock()
+    client.with_options = MagicMock(return_value=client)
 
     # Mock conversations.create
     client.conversations.create.return_value = MagicMock(id="test_conversation_id")
@@ -715,6 +716,76 @@ def capture(request: httpx.Request) -> httpx.Response:
         assert batches == [items[:20], items[20:40], items[40:]]
         assert saved == items[:20] + items[40:]
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("configured_retries", [None, 4])
+    @pytest.mark.parametrize("lost_response_batch", [1, 2])
+    async def test_chunk_lost_response_is_not_retried(
+        self, configured_retries: int | None, lost_response_batch: int
+    ):
+        saved: list[dict[str, Any]] = []
+        batches: list[list[dict[str, Any]]] = []
+
+        def capture(request: httpx.Request) -> httpx.Response:
+            batch = json.loads(request.content)["items"]
+            batches.append(batch)
+            saved.extend(batch)
+            if len(batches) == lost_response_batch:
+                raise httpx.ReadError("synthetic lost response", request=request)
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        items: list[TResponseInputItem] = [
+            {"role": "user", "content": f"message {i}"} for i in range(41)
+        ]
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+            **({"max_retries": configured_retries} if configured_retries is not None else {}),
+        ) as client:
+            original_retries = client.max_retries
+            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            with pytest.raises(APIConnectionError):
+                await session.add_items(items)
+            assert client.max_retries == original_retries
+            assert not client.is_closed()
+
+        assert len(batches) == lost_response_batch
+        assert saved == items[: 20 * lost_response_batch]
+
+    @pytest.mark.asyncio
+    async def test_chunk_retry_override_preserves_single_request_retries(self):
+        calls = 0
+        saved: list[dict[str, Any]] = []
+
+        def capture(request: httpx.Request) -> httpx.Response:
+            nonlocal calls
+            calls += 1
+            if calls == 3:
+                return httpx.Response(
+                    429,
+                    headers={"retry-after-ms": "1"},
+                    json={"error": {"message": "synthetic rate limit"}},
+                )
+            saved.extend(json.loads(request.content)["items"])
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        items: list[TResponseInputItem] = [
+            {"role": "user", "content": f"message {i}"} for i in range(21)
+        ]
+        single: list[TResponseInputItem] = [{"role": "user", "content": "single request"}]
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            max_retries=1,
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+        ) as client:
+            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            await session.add_items(items)
+            await session.add_items(single)
+            assert client.max_retries == 1
+            assert not client.is_closed()
+
+        assert calls == 4
+        assert saved == items + single
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize("request_fails", [False, True])
     async def test_cancelled_batch_settles_before_queued_writer(self, request_fails: bool):

From d166df74ed2a6420d9413128a7b6a4bce13c6a94 Mon Sep 17 00:00:00 2001
From: Justin Beckwith <jbeckwith@openai.com>
Date: Wed, 7 Oct 2026 13:05:37 -0700
Subject: [PATCH 9/9] fix: deliver cancellation before sending subsequent
 chunks

---
 .../memory/openai_conversations_session.py    |  3 ++
 .../test_openai_conversations_session.py      | 29 +++++++++++++++++++
 2 files changed, 32 insertions(+)

diff --git a/src/agents/memory/openai_conversations_session.py b/src/agents/memory/openai_conversations_session.py
index 6e3d6b1b58..d85bdadccd 100644
--- a/src/agents/memory/openai_conversations_session.py
+++ b/src/agents/memory/openai_conversations_session.py
@@ -150,6 +150,9 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
             )
             # The Conversations items-create endpoint accepts up to 20 items per request.
             for offset in range(0, len(items), _MAX_ITEMS_PER_REQUEST):
+                if offset:
+                    # Deliver queued cancellation before scheduling another mutation.
+                    await asyncio.sleep(0)
                 request = client.conversations.items.create(
                     conversation_id=session_id,
                     items=items[offset : offset + _MAX_ITEMS_PER_REQUEST],
diff --git a/tests/memory/test_openai_conversations_session.py b/tests/memory/test_openai_conversations_session.py
index 8ddce719ba..a966930116 100644
--- a/tests/memory/test_openai_conversations_session.py
+++ b/tests/memory/test_openai_conversations_session.py
@@ -873,6 +873,35 @@ async def append_cancelled() -> None:
         assert batches == [items[:20], items[20:40], survivor]
         assert saved == (items[:20] if request_fails else items[:40]) + survivor
 
+    @pytest.mark.asyncio
+    async def test_cancellation_between_chunks_stops_next_request(self):
+        batches: list[list[dict[str, Any]]] = []
+
+        def capture(request: httpx.Request) -> httpx.Response:
+            batches.append(json.loads(request.content)["items"])
+            if len(batches) == 1:
+                request_task = asyncio.current_task()
+                assert request_task is not None
+                # Queue cancellation after the completed request wakes its caller.
+                request_task.add_done_callback(
+                    lambda _: asyncio.get_running_loop().call_soon(write.cancel)
+                )
+            return httpx.Response(200, json={"object": "list", "data": [], "has_more": False})
+
+        items: list[TResponseInputItem] = [
+            {"role": "user", "content": f"message {i}"} for i in range(41)
+        ]
+        async with AsyncOpenAI(
+            api_key="test-placeholder",
+            http_client=httpx.AsyncClient(transport=httpx.MockTransport(capture)),
+        ) as client:
+            session = OpenAIConversationsSession(conversation_id="conv_test", openai_client=client)
+            write = asyncio.create_task(session.add_items(items))
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(write, timeout=5)
+
+        assert batches == [items[:20]]
+
     @pytest.mark.asyncio
     async def test_single_request_cancellation_does_not_wait_for_response(self):
         started = asyncio.Event()