fix: #3348 AdvancedSQLiteSession.add_items silent partial save

src/agents/agent_output.py

@@ -180,15 +180,25 @@ def _is_subclass_of_base_model_or_dict(t: Any) -> bool:
     return issubclass(t, BaseModel | dict)
 
 
-def _type_to_str(t: type[Any]) -> str:
+def _type_to_str(t: Any) -> str:
     origin = get_origin(t)
     args = get_args(t)
 
     if origin is None:
-        # It's a simple type like `str`, `int`, etc.
-        return t.__name__
+        # Plain type (str, int, MyModel, ...) or a non-type value supplied as a
+        # type argument — e.g. the "ok" inside `Literal["ok"]` is a str instance,
+        # not a class, so `t.__name__` would raise. Fall back to repr() in that
+        # case so nested forms like `list[Literal["ok"]]` still format cleanly.
+        if isinstance(t, type):
+            return t.__name__
+        return repr(t)
     elif args:
         args_str = ", ".join(_type_to_str(arg) for arg in args)
-        return f"{origin.__name__}[{args_str}]"
+        # `typing.Literal`/`typing.Union`/etc. expose `_name` rather than
+        # `__name__` on some Python versions.
+        origin_name = (
+            getattr(origin, "__name__", None) or getattr(origin, "_name", None) or str(origin)
+        )
+        return f"{origin_name}[{args_str}]"
     else:
         return str(t)

src/agents/extensions/memory/advanced_sqlite_session.py

@@ -133,26 +133,18 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
         def _add_items_sync():
             """Synchronous helper to add items and structure metadata together."""
             with self._locked_connection() as conn:
-                # Keep both writes in one critical section so message IDs and metadata stay aligned.
-                self._insert_items(conn, items)
-                conn.commit()
+                # The base message rows and the matching `message_structure` rows have to
+                # land together — advanced reads join through `message_structure`, so a
+                # success that only writes one side leaves the new items invisible. Keep
+                # both writes in a single transaction and roll back on any failure so
+                # callers see the original error and can retry cleanly.
                 try:
+                    self._insert_items(conn, items)
                     self._insert_structure_metadata(conn, items)
                     conn.commit()
-                except Exception as e:
+                except Exception:
                     conn.rollback()
-                    self._logger.error(
-                        f"Failed to add structure metadata for session {self.session_id}: {e}"
-                    )
-                    try:
-                        deleted_count = self._cleanup_orphaned_messages_sync(conn)
-                        if deleted_count:
-                            conn.commit()
-                        else:
-                            conn.rollback()
-                    except Exception as cleanup_error:
-                        conn.rollback()
-                        self._logger.error(f"Failed to cleanup orphaned messages: {cleanup_error}")
+                    raise
 
         await asyncio.to_thread(_add_items_sync)
 

tests/extensions/memory/test_advanced_sqlite_session.py

@@ -1422,3 +1422,53 @@ async def test_output_tokens_details_persisted_when_input_details_missing():
     assert turn_usage["output_tokens_details"] == {"reasoning_tokens": 42}
     assert turn_usage["input_tokens_details"] is None
     session.close()
+
+
+async def test_add_items_rolls_back_when_structure_metadata_fails():
+    """Regression for #3348.
+
+    The structure-metadata write and the base message rows must land together. If
+    the metadata write fails, the base rows should be rolled back instead of being
+    left as orphans that advanced reads can never see, and the original error
+    should propagate to the caller so a retry path can react.
+    """
+
+    class BrokenMetadataSession(AdvancedSQLiteSession):
+        fail_metadata = True
+
+        def _insert_structure_metadata(
+            self,
+            conn: Any,
+            items: list[TResponseInputItem],
+        ) -> None:
+            if self.fail_metadata:
+                raise RuntimeError("metadata write failed")
+            super()._insert_structure_metadata(conn, items)
+
+    session = BrokenMetadataSession(session_id="add_items_rollback", create_tables=True)
+    try:
+        with pytest.raises(RuntimeError, match="metadata write failed"):
+            await session.add_items([{"role": "user", "content": "hello"}])
+
+        # Both tables should be untouched after the rollback.
+        assert await session.get_items() == []
+        with session._locked_connection() as conn:
+            raw_count = conn.execute(
+                f"SELECT COUNT(*) FROM {session.messages_table} WHERE session_id = ?",
+                (session.session_id,),
+            ).fetchone()[0]
+            structure_count = conn.execute(
+                "SELECT COUNT(*) FROM message_structure WHERE session_id = ?",
+                (session.session_id,),
+            ).fetchone()[0]
+        assert raw_count == 0
+        assert structure_count == 0
+
+        # A retry after the transient failure should land cleanly when the
+        # underlying error goes away.
+        session.fail_metadata = False
+        await session.add_items([{"role": "user", "content": "hello"}])
+        items = await session.get_items()
+        assert items == [{"role": "user", "content": "hello"}]
+    finally:
+        session.close()

tests/test_output_tool.py

@@ -1,5 +1,5 @@
 import json
-from typing import Any
+from typing import Any, Literal, cast
 
 import pytest
 from pydantic import BaseModel
@@ -94,6 +94,33 @@ def test_structured_output_generic_dict_rejects_wrapper_shape():
         output_schema.validate_json(json.dumps({"response": {"foo": 1}}))
 
 
+def test_structured_output_literal_name_does_not_crash():
+    # `AgentOutputSchema.name()` used to raise `AttributeError` on `Literal["ok"]`
+    # because the Literal value "ok" is a `str` instance rather than a class, and
+    # the name formatter unconditionally read `__name__`. See issue #3357.
+    schema = AgentOutputSchema(cast(type[Any], Literal["ok"]))
+    assert schema.name() == "Literal['ok']"
+
+    # Multiple Literal members format cleanly.
+    schema_multi = AgentOutputSchema(cast(type[Any], Literal["ok", "done"]))
+    assert schema_multi.name() == "Literal['ok', 'done']"
+
+    # Literal nested inside a generic still works.
+    schema_nested = AgentOutputSchema(
+        cast(type[Any], list[Literal["ok", "done"]]),
+        strict_json_schema=False,
+    )
+    assert schema_nested.name() == "list[Literal['ok', 'done']]"
+
+    # Non-string Literal values use repr() so they keep their original literal form.
+    schema_int = AgentOutputSchema(cast(type[Any], Literal[1, 2]))
+    assert schema_int.name() == "Literal[1, 2]"
+
+    # Plain and other generic types are unchanged by the fix.
+    assert AgentOutputSchema(str).name() == "str"
+    assert AgentOutputSchema(list[int]).name() == "list[int]"
+
+
 def test_bad_json_raises_error(mocker):
     agent = Agent(name="test", output_type=Foo)
     output_schema = get_output_schema(agent)