fix(sessions): raise AlreadyExistsError on concurrent create_session races

[hungubqn0310]

Link to Issue or Description of Change

• Related: DatabaseSessionService: session row persists in storage without its user_state row ("User state missing" on later append_event) #6823

Problem:
DatabaseSessionService.create_session()'s up-front existence check
(has_user_provided_id and await sql_session.get(...)) is not atomic with
the INSERT that follows it. Two concurrent callers using the same
caller-provided (app_name, user_id, session_id) can both pass the check
and then race the same insert. The loser gets a raw, unhandled
IntegrityError / UniqueViolationError instead of the documented
AlreadyExistsError.

Solution:
Wrap the sql_session.flush() around the session insert in
try/except IntegrityError, raising AlreadyExistsError — the same
pattern _get_or_create_state already uses (via a SAVEPOINT) for the
analogous app_state/user_state race.

This does not address the separate orphaned user_state row issue also
reported in #6823. _rollback_on_exception_session rolls back the whole
transaction on any exception (including the AlreadyExistsError raised
here), so this particular race shouldn't be able to leave a row behind on
its own — it only replaces a confusing raw IntegrityError with a clean,
documented error. The orphan issue needs more data to root-cause.

Testing Plan

Unit Tests:

• I have added or updated unit tests for my change.
• All unit tests pass locally.

Added test_create_session_concurrent_same_id_raises_already_exists_error
in tests/unittests/sessions/test_session_service.py: two concurrent
create_session() calls for the same caller-provided session_id (after
pre-warming app_state/user_state with an unrelated session, to isolate
this race from the separate, already-guarded _get_or_create_state race).
Asserts exactly one caller succeeds and the other raises AlreadyExistsError,
and that exactly one session row survives.

$ pytest tests/unittests/sessions/test_session_service.py -q
399 passed, 2 xfailed in 11.01s

$ pytest ./tests/unittests -q
12619 passed, 88 skipped, 33 xfailed, 1 xpassed, 2 failed in 292.16s

The 2 failures (test_import_loading.py::test_entry_point_loads_only_allowlisted_packages[agent|runner])
are pre-existing and unrelated (an httpx2 eager-import allowlist check) —
reproduced identically on main without this change.

Manual End-to-End (E2E) Tests:

Additionally reproduced and verified the race outside the unit test, against
two real backends, before writing the fix into the codebase:

• sqlite+aiosqlite — 5/5 trials without the patch raised a raw
  sqlite3.IntegrityError: UNIQUE constraint failed: sessions.app_name, sessions.user_id, sessions.id.
• Postgres 17 + asyncpg (disposable local container) — 4/5 trials without
  the patch raised sqlalchemy.dialects.postgresql.asyncpg.IntegrityError
  wrapping asyncpg.exceptions.UniqueViolationError: duplicate key value violates unique constraint "sessions_pkey" (the 5th trial's slower
  coroutine legitimately hit the existing early-exists check before reaching
  the insert).

With the patch applied, 5/5 trials on both backends raised a clean
AlreadyExistsError instead, with exactly one session surviving each time.

Checklist

• I have read the CONTRIBUTING.md document.
• I have performed a self-review of my own code.
• I have commented my code, particularly in hard-to-understand areas.
• I have added tests that prove my fix is effective.
• New and existing unit tests pass locally with my changes.
• I have manually tested my changes end-to-end (see above).
• Any dependent changes have been merged and published in downstream modules. (N/A)

Additional context

Pinned versions in the project where this was first observed (from uv.lock):
google-adk==2.3.0, asyncpg==0.31.0, sqlalchemy==2.0.44.

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[hungubqn0310]

recheck