SQLAlchemySession can fail and drop writes on concurrent first access

[Ratnaditya-J]

Hi team, I think there is a concurrency bug in SQLAlchemySession when multiple tasks write to the same fresh session at the same time.

I can reproduce two related failure modes:

If create_tables=True, concurrent calls into _ensure_tables() can fail with:
OperationalError: table agent_sessions already exists
Even if the tables are created ahead of time, concurrent first writes can still fail with:
IntegrityError: UNIQUE constraint failed: agent_sessions.session_id
In both cases, some writes are lost and the final session history ends up with fewer items than were submitted.

This looks like two separate race conditions:

_ensure_tables() uses _create_tables without synchronization
add_items() does a check-then-insert for the parent session row
Relevant spots:

src/agents/extensions/memory/sqlalchemy_session.py:88
src/agents/extensions/memory/sqlalchemy_session.py:183
src/agents/extensions/memory/sqlalchemy_session.py:260
Minimal repro

import asyncio
import tempfile
from pathlib import Path

from agents.extensions.memory.sqlalchemy_session import SQLAlchemySession


async def main():
    db_path = Path(tempfile.gettempdir()) / "agents_sqlalchemy_race.db"
    if db_path.exists():
        db_path.unlink()

    session = SQLAlchemySession.from_url(
        session_id="race",
        url=f"sqlite+aiosqlite:///{db_path}",
        create_tables=True,
    )

    async def worker(i: int):
        await session.add_items([{"role": "user", "content": f"msg-{i}"}])

    results = await asyncio.gather(
        *(worker(i) for i in range(25)),
        return_exceptions=True,
    )

    items = await session.get_items()
    print("stored items:", len(items))
    print("exceptions:", [type(r).__name__ for r in results if isinstance(r, Exception)])


asyncio.run(main())

I reproduced this with Python 3.12 using the editable package plus SQLAlchemy, aiosqlite, and greenlet.

Typical result:

OperationalError: table agent_sessions already exists
IntegrityError: UNIQUE constraint failed: agent_sessions.session_id
stored item count is well below the number of submitted writes
Expected behavior
Concurrent writes to the same logical session should not fail or silently drop messages.

Actual behavior
Concurrent first writes can fail during schema setup and/or parent session creation, leaving the session history incomplete.

Notes
The SQLite session backend appears to serialize access explicitly, so this may just need the same kind of protection here plus a conflict-safe insert/upsert for the session row.