OpenAIResponsesCompactionSession drops session items written while responses.compact is in flight #4679

Please read this first

• Have you read the docs? Yes.
• Have you searched for related issues? Yes. OpenAIResponsesCompactionSession can lose session history if add_items fails after clear #3116, OpenAIConversationsSession can split history during concurrent lazy initialization #3820, and SQLAlchemySession.pop_item returns the same item to concurrent callers #4205 are the closest matches: they cover replacement rollback and concurrent access in other paths. None of them covers a write landing while responses.compact is in flight.

Describe the bug

OpenAIResponsesCompactionSession.run_compaction snapshots the session history (_ensure_compaction_candidates, line 204 in openai_responses_compaction_session.py on main at e773b154), awaits the responses.compact API call (line 238), and then, inside _mutation_lock (lines 248 to 255), clears the underlying session and rewrites it purely from that snapshot. The snapshot is taken before the API call and the call takes seconds on real conversations, so any wrapper write that lands while the request is in flight is destroyed by the replacement:

• An item added through add_items during the call is silently deleted. No error, no log.
• A clear_session issued during the call is undone: the wiped history comes back as the compacted summary of the snapshot.

previous_items fetched inside the locked block is used only as rollback state for a failed replacement and is never compared against the snapshot. Since the runner triggers run_compaction automatically after every turn save (src/agents/run_internal/session_persistence.py), two overlapping Runner.run calls on the same session can hit this window in normal operation, and the _mutation_lock comment in openai_responses_compaction_session.py already treats concurrent write safety as an invariant of this class.

Debug information

• Agents SDK version: v0.22.0 (also reproduces on main at e773b154)
• Python version: 3.14.3
• Operating system: macOS 26.5.2
• Model and model provider: OpenAI Responses API. The script below mocks the compact call; the deletion happens entirely in local SDK code.
• Does the issue reproduce with the latest Agents SDK release? Yes
• Does the issue occur consistently or intermittently? The script below is deterministic. In real runs it depends on a write landing during the compact request, so it shows up intermittently under concurrency.

items before replacement: 13
items after replacement: 1
concurrent item survived: False

Repro steps

The script gates the mocked compact call on an asyncio.Event so the concurrent write lands while the request is in flight. The mock only supplies the latency window; the replacement that deletes the item is ordinary SDK code writing to a local SQLiteSession.

import asyncio
from typing import Any
from unittest.mock import AsyncMock, MagicMock

from agents.memory import OpenAIResponsesCompactionSession, SQLiteSession


async def main() -> None:
    underlying = SQLiteSession("bug-repro")
    await underlying.add_items(
        [{"type": "message", "role": "assistant", "content": f"msg{i}"} for i in range(12)]
    )

    compact_entered = asyncio.Event()
    release_compact = asyncio.Event()
    compact_response = MagicMock()
    compact_response.output = [{"type": "compaction", "summary": "compacted"}]

    async def gated_compact(**kwargs: Any) -> MagicMock:
        # Stands in for the multi second responses.compact network call.
        compact_entered.set()
        await release_compact.wait()
        return compact_response

    client = MagicMock()
    client.responses.compact = AsyncMock(side_effect=gated_compact)

    session = OpenAIResponsesCompactionSession(
        session_id="bug-repro",
        underlying_session=underlying,
        client=client,
        compaction_mode="input",
    )

    task = asyncio.create_task(session.run_compaction({"force": True}))
    await compact_entered.wait()

    # A concurrent Runner.run saving its turn while compaction is in flight.
    await session.add_items(
        [{"type": "message", "role": "user", "content": "written while compact was in flight"}]
    )
    print("items before replacement:", len(await underlying.get_items()))

    release_compact.set()
    await task

    items = await underlying.get_items()
    print("items after replacement:", len(items))
    print(
        "concurrent item survived:",
        any(
            isinstance(item, dict)
            and item.get("content") == "written while compact was in flight"
            for item in items
        ),
    )


asyncio.run(main())

Replacing the concurrent add_items with await session.clear_session() shows the second half of the problem: the session ends up containing the compacted summary of the history the caller just wiped.

Expected behavior

Items added while the compact request is in flight survive the replacement and appear after the compacted output, and a clear_session issued during the request leaves the session empty instead of resurrecting the summarized history.

I have a fix with regression tests and will open a PR shortly.