Firestore session and memory services: fix nested state merge, package exports and memory duplicates

Start in src/google/adk/integrations/firestore/firestore_session_service.py, firestore_memory_service.py, and init.py, then review tests/unittests/sessions/test_session_service.py and the Firestore unit tests. Verify the state replacement, event timestamps, user-state lookup, package exports, idempotent memory writes, and FieldFilter queries against the shared session contract and relevant unit tests.

🔴 Required Information

Is your feature request related to a specific problem?

Yes. Python has shipped FirestoreSessionService and FirestoreMemoryService since v1.31.0
(#5088), but they don't behave like the other session and memory services, and there is no
Python documentation for them. Testing them against a real Firestore database on 2.9.1 and
main (f33d4923), I reproduced these problems:

1. Nested user: / app: state is merged instead of replaced. Updating user:profile
   from {"name": "Alice", "role": "admin"} to {"name": "Alice"} stores
   {"name": "Alice", "role": "admin"}, so the removed key comes back on the next
   get_session. The in-memory, SQLite and database session services replace the value.
   Cause: user_states / app_states are written with transaction.set(..., merge=True) in
   both create_session and append_event, and Firestore deep-merges nested maps.
   Session-scoped state is not affected (it is stored as JSON).
2. The services can't be imported from their package.
   from google.adk.integrations.firestore import FirestoreSessionService raises ImportError
   because integrations/firestore/__init__.py exports nothing. It is the only package under
   integrations/ like this.
3. GetSessionConfig(after_timestamp=...) returns the wrong events. Events are stored with
   "timestamp": firestore.SERVER_TIMESTAMP, so the filter and event ordering use the write
   time instead of event.timestamp.
4. get_user_state is not supported. It raises NotImplementedError, while the in-memory,
   database, SQLite and Redis session services implement it.
5. add_session_to_memory stores duplicates. Each call writes a new document per event, so
   adding the same session again (for example after every turn) stores every memory again.
   Search hides this by de-duplicating results, but storage keeps growing.
   add_events_to_memory is also not implemented; InMemoryMemoryService supports it.
6. Positional-filter warnings. list_sessions and get_session with after_timestamp
   call .where("field", op, value), so google-cloud-firestore emits
   UserWarning: Detected filter using positional arguments on every call.

Describe the Solution You'd Like

Make the Firestore services behave like the other backends, with no change to their public
API other than implementing two existing base-class methods:

1. Write user: / app: state back whole so a new value replaces the old one.
2. Export FirestoreSessionService and FirestoreMemoryService from
   google.adk.integrations.firestore, loaded lazily (as integrations/model_armor does) so
   importing the package still does not require google-cloud-firestore.
3. Store each event's own timestamp so after_timestamp and event ordering use event time.
4. Implement get_user_state by reading the user_states document the service already writes.
5. Give each memory entry a stable document ID derived from app, user, session and event IDs,
   so re-adding a session overwrites instead of duplicating, and implement
   add_events_to_memory.
6. Pass query filters as where(filter=FieldFilter(...)), as FirestoreMemoryService already
   does.

Impact on your work

I build agents for Google Cloud customers, and Firestore is the natural serverless session
store for agents on Cloud Run. Today:

• A value removed from nested user or app state (for example a role or a permission flag)
  silently stays in Firestore and comes back on the next turn.
• Memory storage grows with every turn when sessions are added to memory after each turn.
• Moving an agent between session services changes its behavior.
• Python users have no documentation showing how to use these services: the adk.dev
  Firestore page covers Java only.

Willingness to contribute

Yes. I have the fix ready with unit tests, one commit per item above, and can open the PR
once this is triaged. I will also open a PR in google/adk-docs adding Python usage to the
Firestore page.

🟡 Recommended Information

Describe Alternatives You've Considered

• DatabaseSessionService with Cloud SQL or AlloyDB: works correctly, but needs a database
  instance to run and manage, which Firestore avoids.
• VertexAiSessionService: works, but ties sessions to Agent Engine.
• Workarounds on the current Firestore services: import from the full module path, avoid dict
  values in user: / app: state, and add each session to memory only once. These avoid the
  symptoms but are easy to miss, and nothing in the docs mentions them.

Proposed API / Implementation

No new public API. The changes stay inside src/google/adk/integrations/firestore/ and its
unit tests:

# 1. create_session / append_event: write the full, already-updated dict
transaction.set(user_ref, current_user)  # was: transaction.set(user_ref, current_user, merge=True)

# 3. append_event: store the event's own time
"timestamp": datetime.fromtimestamp(event.timestamp, tz=timezone.utc)  # was: firestore.SERVER_TIMESTAMP

# 5. memory: stable document ID per event
doc_id = sha256("\x00".join((app_name, user_id, session_id or "", event.id)))

After the change, a local run of the shared session contract suite
(tests/unittests/sessions/test_session_service.py) against a real Firestore database goes
from 18 passed / 13 failed to 27 passed / 4 failed. The 4 remaining failures are covered below.

Additional Context

Minimal reproduction for item 1 (pip install google-adk==2.9.1 google-cloud-firestore, a
Firestore Native database):

import asyncio, time
from google.adk.events.event import Event
from google.adk.events.event_actions import EventActions
from google.adk.integrations.firestore.firestore_session_service import FirestoreSessionService
from google.cloud import firestore

async def main():
  client = firestore.AsyncClient(project="PROJECT", database="DATABASE")
  svc = FirestoreSessionService(client=client)
  s = await svc.create_session(app_name="demo", user_id="alice")
  for i, value in enumerate(({"name": "Alice", "role": "admin"}, {"name": "Alice"})):
    await svc.append_event(s, Event(
        author="user", invocation_id=f"i{i}", timestamp=time.time(),
        actions=EventActions(state_delta={"user:profile": value})))
  print("same Session object right after update :", s.state["user:profile"])
  fresh = await svc.get_session(app_name="demo", user_id="alice", session_id=s.id)
  print("after reload (get_session)             :", fresh.state["user:profile"])

asyncio.run(main())

same Session object right after update : {'name': 'Alice'}
after reload (get_session)             : {'name': 'Alice', 'role': 'admin'}

Not proposed here, open questions:

• The 4 remaining contract failures: three come from last_update_time using Firestore's
  server updateTime instead of the appended event's timestamp, which looks intentional after
  #5632 / #5642. The fourth is list_sessions(user_id=None), which needs a single-field
  collection-group index on sessions.appName; I would document that.
• Should the CLI accept --session_service_uri firestore://... and
  --memory_service_uri firestore://...? Today an unregistered session scheme falls back to
  DatabaseSessionService and fails with ValueError: Invalid database URL format. I have a
  working services.py registration and can propose a built-in scheme if that is wanted.
• Registering Firestore in tests/unittests/sessions/_conformance.py needs a stateful Firestore
  fake. I can do that as a follow-up.