@@ -7,7 +7,7 @@
 import time
 from contextlib import closing
 from pathlib import Path
-from typing import Any, cast
+from typing import Any, ClassVar, cast
 
 from agents.result import RunResult
 from agents.usage import Usage
@@ -72,7 +72,7 @@ def __init__(
                 self._init_structure_tables()
             else:
                 with self._locked_connection() as conn:
-                    self._verify_structure_tables_scope(conn)
+                    self._claim_structure_tables(conn)
         except BaseException:
             try:
                 self.close()
@@ -111,47 +111,61 @@ def _commit_branch_pointer(self, branch_id: str, generation: int) -> bool:
             self._current_branch_id = branch_id
             return True
 
-    def _verify_structure_tables_scope(self, conn: sqlite3.Connection) -> None:
-        """Reject a database file whose structure tables belong to other base tables.
+    # The structure tables that record which base-table pair owns a database file, and the
+    # foreign keys that record it. `branch_reservations` and `session_clear_generations` carry no
+    # foreign keys, but enforcing one pair per file keeps them unambiguous too.
+    _STRUCTURE_TABLE_OWNERS: ClassVar[dict[str, tuple[str, ...]]] = {
+        "message_structure": ("session_id", "message_id"),
+        "turn_usage": ("session_id",),
+    }
+
+    def _claim_structure_tables(self, conn: sqlite3.Connection) -> None:
+        """Require a complete structure-table layout owned by the configured base-table pair.
 
         The structure tables are not named after ``sessions_table``/``messages_table``, so a
-        database file can only hold the structure rows of a single base-table pair. ``CREATE
-        TABLE IF NOT EXISTS`` keeps the first pair's foreign keys, so a second session configured
-        with different base table names would join ``message_structure`` against the wrong
-        messages table and read back rows that belong to the other pair.
-
-        Both owner-bearing structure tables are checked, so a file whose two structure tables were
-        created for different pairs is rejected instead of being accepted by whichever one happens
-        to match.
+        database file can only hold the structure rows of a single pair. ``CREATE TABLE IF NOT
+        EXISTS`` keeps the first pair's foreign keys, so a second session configured with
+        different base table names would join ``message_structure`` against the wrong messages
+        table and read back rows that belong to the other pair.
+
+        An incomplete layout is rejected rather than accepted, so a ``create_tables=False``
+        session cannot open a file before any pair has claimed it and then have another pair
+        claim it underneath.
         """
-        expected = {
-            "message_structure": {
-                "session_id": self.sessions_table,
-                "message_id": self.messages_table,
-            },
-            "turn_usage": {"session_id": self.sessions_table},
-        }
-        for table, expected_owners in expected.items():
-            # SQLite resolves table names case-insensitively but `PRAGMA foreign_key_list` reports
-            # the spelling used at CREATE time, so compare case-folded.
-            owners = {
-                row[3].casefold(): row[2]
-                for row in conn.execute(f"PRAGMA foreign_key_list({table})")
-            }
-            if not owners:
-                continue
+        owned_by = {"session_id": self.sessions_table, "message_id": self.messages_table}
+        for table, columns in self._STRUCTURE_TABLE_OWNERS.items():
+            owners = {row[3]: row[2] for row in conn.execute(f"PRAGMA foreign_key_list({table})")}
+            missing = [column for column in columns if column not in owners]
+            if missing:
+                raise ValueError(
+                    f"The `{table}` table in {self.db_path} does not exist yet or does not record "
+                    "which sessions_table/messages_table pair owns it. Construct an "
+                    "AdvancedSQLiteSession with create_tables=True to create and claim the "
+                    "structure tables before opening the database without them."
+                )
             if any(
-                owners.get(column, "").casefold() != owner.casefold()
-                for column, owner in expected_owners.items()
+                not self._identifiers_equal(conn, owners[column], owned_by[column])
+                for column in columns
             ):
-                found = "/".join(owners.get(column, "?") for column in expected_owners)
-                configured = "/".join(expected_owners.values())
+                found = "/".join(owners[column] for column in columns)
+                configured = "/".join(owned_by[column] for column in columns)
                 raise ValueError(
                     f"The `{table}` table in {self.db_path} already belongs to '{found}', not to "
                     f"the configured '{configured}'. Structure tables are shared per database "
                     "file, so give each sessions_table/messages_table pair its own db_path."
                 )
 
+    @staticmethod
+    def _identifiers_equal(conn: sqlite3.Connection, left: str, right: str) -> bool:
+        """Compare two table names the way SQLite compares identifiers.
+
+        SQLite folds identifiers with ASCII rules only, so `NOCASE` is asked directly rather than
+        reimplemented. Python's ``casefold()`` would equate names SQLite keeps distinct, for
+        example ``ßsessions`` and ``sssessions``.
+        """
+        row = conn.execute("SELECT ? = ? COLLATE NOCASE", (left, right)).fetchone()
+        return bool(row[0])
+
     def _init_structure_tables(self):
         """Add structure and usage tracking tables.
 
@@ -230,7 +244,7 @@ def _init_structure_tables(self):
                 ON turn_usage(session_id, branch_id, user_turn_number)
             """)
 
-            self._verify_structure_tables_scope(conn)
+            self._claim_structure_tables(conn)
 
             conn.commit()
 
 
@@ -33,6 +33,34 @@
 pytestmark = pytest.mark.asyncio
 
 
+def _claim_structure_tables_in_process(
+    db_path: str,
+    sessions_table: str,
+    messages_table: str,
+    ready: Any,
+    start: Any,
+    results: Any,
+) -> None:
+    """Construct a create_tables session in a child process and report the outcome."""
+    ready.set()
+    start.wait(timeout=30)
+    try:
+        session = AdvancedSQLiteSession(
+            session_id="concurrent",
+            db_path=db_path,
+            create_tables=True,
+            sessions_table=sessions_table,
+            messages_table=messages_table,
+        )
+        session.close()
+    except ValueError:
+        results.put(("rejected", sessions_table))
+    except BaseException as exc:  # pragma: no cover - surfaced in the assertion below
+        results.put((f"error:{type(exc).__name__}", sessions_table))
+    else:
+        results.put(("claimed", sessions_table))
+
+
 def _multiprocessing_context() -> Any:
     method = "spawn" if sys.platform == "win32" else "forkserver"
     return multiprocessing.get_context(method)
@@ -3844,3 +3872,117 @@ async def test_structure_tables_accept_equivalent_identifier_casing(tmp_path: Pa
         assert await recased.get_items() == [{"role": "user", "content": "first"}]
     finally:
         recased.close()
+
+
+async def test_structure_tables_reject_distinct_non_ascii_identifiers(tmp_path: Path) -> None:
+    """SQLite folds identifiers with ASCII rules, so these are two different pairs.
+
+    Python's `casefold()` equates `ßsessions` and `sssessions`, which would let the second pair
+    through and restore the cross-table mixing this change prevents.
+    """
+    assert "ßsessions".casefold() == "sssessions".casefold()
+
+    db_path = tmp_path / "advanced_non_ascii_structure.db"
+    first = AdvancedSQLiteSession(
+        session_id="shared",
+        db_path=db_path,
+        create_tables=True,
+        sessions_table="ßsessions",
+        messages_table="ßmessages",
+    )
+    try:
+        await first.add_items([{"role": "user", "content": "first"}])
+
+        with pytest.raises(ValueError, match="ßsessions"):
+            AdvancedSQLiteSession(
+                session_id="shared",
+                db_path=db_path,
+                create_tables=True,
+                sessions_table="sssessions",
+                messages_table="ssmessages",
+            )
+
+        assert await first.get_items() == [{"role": "user", "content": "first"}]
+    finally:
+        first.close()
+
+
+async def test_no_create_session_rejects_a_database_without_an_owner(tmp_path: Path) -> None:
+    """A no-create session must not open a file before a pair has claimed the structure tables.
+
+    Accepting it would let another pair claim the tables afterwards, leaving this session writing
+    and reading structure rows owned by that other pair.
+    """
+    db_path = tmp_path / "advanced_unclaimed_structure.db"
+
+    with pytest.raises(ValueError, match="create_tables=True"):
+        AdvancedSQLiteSession(session_id="shared", db_path=db_path, create_tables=False)
+
+    owner = AdvancedSQLiteSession(session_id="shared", db_path=db_path, create_tables=True)
+    try:
+        await owner.add_items([{"role": "user", "content": "first"}])
+    finally:
+        owner.close()
+
+    # Once a pair owns the layout, the same pair may open it without creating anything.
+    reader = AdvancedSQLiteSession(session_id="shared", db_path=db_path, create_tables=False)
+    try:
+        assert await reader.get_items() == [{"role": "user", "content": "first"}]
+    finally:
+        reader.close()
+
+
+@pytest.mark.review_optional
+async def test_concurrent_structure_table_claims_leave_one_coherent_owner(tmp_path: Path) -> None:
+    """Two processes claiming a fresh file with different pairs must not split the layout."""
+    db_path = tmp_path / "advanced_concurrent_claim.db"
+    pairs = [("a_sessions", "a_messages"), ("b_sessions", "b_messages")]
+
+    # Create the file with WAL already enabled. Switching journal mode needs an exclusive lock,
+    # so two processes opening a brand-new file race there first, in SQLiteSession.__init__,
+    # before either reaches the structure tables. Settling it up front keeps this test about the
+    # claim.
+    with contextlib.closing(sqlite3.connect(db_path)) as setup_conn:
+        setup_conn.execute("PRAGMA journal_mode=WAL")
+
+    context = _multiprocessing_context()
+    start = context.Event()
+    results = context.Queue()
+    ready_events = [context.Event(), context.Event()]
+    processes = [
+        context.Process(
+            target=_claim_structure_tables_in_process,
+            args=(str(db_path), sessions_table, messages_table, ready, start, results),
+        )
+        for (sessions_table, messages_table), ready in zip(pairs, ready_events, strict=False)
+    ]
+
+    try:
+        for process in processes:
+            process.start()
+        for ready in ready_events:
+            assert ready.wait(timeout=30)
+        start.set()
+        for process in processes:
+            process.join(timeout=30)
+            assert process.exitcode == 0
+
+        outcomes = [results.get(timeout=5), results.get(timeout=5)]
+        claimed = [pair for status, pair in outcomes if status == "claimed"]
+        assert len(claimed) == 1, outcomes
+        assert all(status in {"claimed", "rejected"} for status, _ in outcomes), outcomes
+
+        # Every owner-bearing structure table must name the one pair that won.
+        winner = claimed[0]
+        with contextlib.closing(sqlite3.connect(db_path)) as conn:
+            for table in ("message_structure", "turn_usage"):
+                owners = {
+                    row[3]: row[2] for row in conn.execute(f"PRAGMA foreign_key_list({table})")
+                }
+                assert owners["session_id"] == winner, (table, owners)
+    finally:
+        start.set()
+        for process in processes:
+            if process.is_alive():
+                process.terminate()
+            process.join(timeout=5)