fix(sessions): support MySQL and MariaDB schema creation

src/agents/extensions/memory/sqlalchemy_session.py

@@ -62,6 +62,14 @@
 
 _T = TypeVar("_T")
 
+# MySQL-family dialects require a bounded VARCHAR for indexed string columns.
+_MYSQL_SESSION_ID_MAX_LENGTH = 190
+_SESSION_ID_TYPE = String().with_variant(
+    String(_MYSQL_SESSION_ID_MAX_LENGTH),
+    "mysql",
+    "mariadb",
+)
+
 
 class SQLAlchemySession(SessionABC):
     """SQLAlchemy implementation of [`Session`][agents.memory.session.Session]."""
@@ -163,7 +171,8 @@ def __init__(
                 'mysql+aiomysql://', or 'sqlite+aiosqlite://').
             create_tables (bool, optional): Whether to automatically create the required
                 tables and indexes. Defaults to False for production use. Set to True for
-                development and testing when migrations aren't used.
+                development and testing when migrations aren't used. Automatically created
+                MySQL and MariaDB schemas store session IDs in VARCHAR(190) columns.
             sessions_table (str, optional): Override the default table name for sessions if needed.
             messages_table (str, optional): Override the default table name for messages if needed.
             session_settings (SessionSettings | None, optional): Session configuration settings
@@ -189,7 +198,7 @@ def __init__(
         self._sessions = Table(
             sessions_table,
             self._metadata,
-            Column("session_id", String, primary_key=True),
+            Column("session_id", _SESSION_ID_TYPE, primary_key=True),
             Column(
                 "created_at",
                 TIMESTAMP(timezone=False),
@@ -211,7 +220,7 @@ def __init__(
             Column("id", Integer, primary_key=True, autoincrement=True),
             Column(
                 "session_id",
-                String,
+                _SESSION_ID_TYPE,
                 ForeignKey(f"{sessions_table}.session_id", ondelete="CASCADE"),
                 nullable=False,
             ),

tests/extensions/memory/test_sqlalchemy_session.py

@@ -17,7 +17,8 @@
     ResponseReasoningItemParam,
     Summary,
 )
-from sqlalchemy import event, insert, select, text, update
+from sqlalchemy import create_mock_engine, event, insert, select, text, update
+from sqlalchemy.dialects import postgresql, sqlite
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 from sqlalchemy.sql import Select
 
@@ -35,6 +36,53 @@
 DB_URL = "sqlite+aiosqlite:///:memory:"
 
 
+@pytest.mark.parametrize("dialect_url", ["mysql://", "mariadb://"])
+async def test_schema_create_all_compiles_for_mysql_family(dialect_url: str):
+    """MySQL-family schema creation includes both tables and the session-time index."""
+    session = SQLAlchemySession.from_url("schema_compile", url=DB_URL)
+    tables = (session._sessions, session._messages)
+    statements: list[str] = []
+
+    def record(statement: Any, *args: Any, **kwargs: Any) -> None:
+        statements.append(str(statement.compile(dialect=engine.dialect)))
+
+    engine = create_mock_engine(dialect_url, record)
+
+    try:
+        session._metadata.create_all(engine)
+        for table in tables:
+            assert table.c.session_id.type.compile(dialect=engine.dialect) == "VARCHAR(190)"
+    finally:
+        await session.engine.dispose()
+
+    assert any("CREATE TABLE agent_sessions" in statement for statement in statements)
+    messages_ddl = next(
+        statement for statement in statements if "CREATE TABLE agent_messages" in statement
+    )
+    assert (
+        "FOREIGN KEY(session_id) REFERENCES agent_sessions (session_id) ON DELETE CASCADE"
+        in messages_ddl
+    )
+    assert any(
+        "CREATE INDEX idx_agent_messages_session_time "
+        "ON agent_messages (session_id, created_at)" in statement
+        for statement in statements
+    )
+
+
+async def test_schema_keeps_unbounded_session_ids_for_sqlite_and_postgresql():
+    """SQLite and PostgreSQL retain the pre-existing unbounded string type."""
+    session = SQLAlchemySession.from_url("schema_compile", url=DB_URL)
+
+    try:
+        for table in (session._sessions, session._messages):
+            session_id_type = table.c.session_id.type
+            assert session_id_type.compile(dialect=postgresql.dialect()) == "VARCHAR"
+            assert session_id_type.compile(dialect=sqlite.dialect()) == "VARCHAR"
+    finally:
+        await session.engine.dispose()
+
+
 def _make_message_item(item_id: str, text_value: str) -> TResponseInputItem:
     content: ResponseOutputTextParam = {
         "type": "output_text",