diff --git a/src/google/adk/integrations/firestore/__init__.py b/src/google/adk/integrations/firestore/__init__.py
index 7c76d28c93d..122121eae2c 100644
--- a/src/google/adk/integrations/firestore/__init__.py
+++ b/src/google/adk/integrations/firestore/__init__.py
@@ -12,6 +12,41 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+"""Firestore integrations for ADK.
+
+This module provides session and memory services backed by Google Cloud
+Firestore. They require the optional ``google-cloud-firestore`` package.
+"""
+
 from __future__ import annotations
 
-"""Firestore integrations for ADK."""
+import typing
+
+if typing.TYPE_CHECKING:
+  from .firestore_memory_service import FirestoreMemoryService
+  from .firestore_session_service import FirestoreSessionService
+
+# Map attribute names to relative module paths.
+_lazy_imports = {
+    "FirestoreMemoryService": ".firestore_memory_service",
+    "FirestoreSessionService": ".firestore_session_service",
+}
+
+__all__ = [
+    "FirestoreMemoryService",
+    "FirestoreSessionService",
+]
+
+
+def __getattr__(name: str) -> typing.Any:
+  if name in _lazy_imports:
+    import importlib
+
+    module_path = _lazy_imports[name]
+    module = importlib.import_module(module_path, __name__)
+    return getattr(module, name)
+  raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def __dir__() -> list[str]:
+  return list(_lazy_imports.keys())
diff --git a/src/google/adk/integrations/firestore/firestore_memory_service.py b/src/google/adk/integrations/firestore/firestore_memory_service.py
index 286aa769b2f..c330c43e62b 100644
--- a/src/google/adk/integrations/firestore/firestore_memory_service.py
+++ b/src/google/adk/integrations/firestore/firestore_memory_service.py
@@ -15,6 +15,9 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Mapping
+from collections.abc import Sequence
+import hashlib
 import logging
 import re
 from typing import Optional
@@ -32,6 +35,7 @@
 if TYPE_CHECKING:
   from google.cloud import firestore
 
+  from ...events.event import Event
   from ...sessions.session import Session
 
 logger = logging.getLogger("google_adk." + __name__)
@@ -40,6 +44,18 @@
 DEFAULT_MEMORIES_COLLECTION = "memories"
 
 
+def _memory_doc_id(
+    *, app_name: str, user_id: str, session_id: Optional[str], event_id: str
+) -> str:
+  """Returns a stable memory document ID for an event.
+
+  Hashed because app names and user IDs may contain characters that are not
+  allowed in document IDs, such as "/".
+  """
+  key = "\x00".join((app_name, user_id, session_id or "", event_id))
+  return hashlib.sha256(key.encode("utf-8")).hexdigest()
+
+
 class FirestoreMemoryService(BaseMemoryService):  # type: ignore[misc]
   """Memory service that uses Google Cloud Firestore as the backend.
 
@@ -82,10 +98,48 @@ def __init__(
   @override
   async def add_session_to_memory(self, session: Session) -> None:
     """Extracts keywords from session events and stores them in the memories collection."""
+    await self._write_memories(
+        app_name=session.app_name,
+        user_id=session.user_id,
+        session_id=session.id,
+        events=session.events,
+    )
+
+  @override
+  async def add_events_to_memory(
+      self,
+      *,
+      app_name: str,
+      user_id: str,
+      events: Sequence[Event],
+      session_id: Optional[str] = None,
+      custom_metadata: Optional[Mapping[str, object]] = None,
+  ) -> None:
+    """Adds events, such as the latest turn, to the memories collection.
+
+    Re-adding an event with the same session ID overwrites its entry.
+    """
+    _ = custom_metadata
+    await self._write_memories(
+        app_name=app_name,
+        user_id=user_id,
+        session_id=session_id,
+        events=events,
+    )
+
+  async def _write_memories(
+      self,
+      *,
+      app_name: str,
+      user_id: str,
+      session_id: Optional[str],
+      events: Sequence[Event],
+  ) -> None:
+    """Writes one memory document per event that has text keywords."""
     batch = self.client.batch()
     count = 0
 
-    for event in session.events:
+    for event in events:
       if not event.content or not event.content.parts:
         continue
 
@@ -97,12 +151,20 @@ async def add_session_to_memory(self, session: Session) -> None:
       if not keywords:
         continue
 
-      doc_ref = self.client.collection(self.memories_collection).document()
+      doc_ref = self.client.collection(self.memories_collection).document(
+          _memory_doc_id(
+              app_name=app_name,
+              user_id=user_id,
+              session_id=session_id,
+              event_id=event.id,
+          )
+      )
       batch.set(
           doc_ref,
           {
-              "appName": session.app_name,
-              "userId": session.user_id,
+              "appName": app_name,
+              "userId": user_id,
+              "sessionId": session_id,
               "keywords": list(keywords),
               "author": event.author,
               "content": event.content.model_dump(
diff --git a/src/google/adk/integrations/firestore/firestore_session_service.py b/src/google/adk/integrations/firestore/firestore_session_service.py
index 404fbdc064f..a4e49e4726f 100644
--- a/src/google/adk/integrations/firestore/firestore_session_service.py
+++ b/src/google/adk/integrations/firestore/firestore_session_service.py
@@ -42,6 +42,7 @@
 
 try:
   from google.cloud import firestore
+  from google.cloud.firestore_v1.base_query import FieldFilter
 except ImportError as e:
   raise ImportError(
       "FirestoreSessionService requires google-cloud-firestore. "
@@ -245,14 +246,15 @@ async def _create_txn(
           (user_snap.to_dict() or {}) if user_snap.exists else {}
       )
 
-      # 2. Writes
+      # 2. Writes. Documents are written whole: a merge write deep-merges
+      # nested maps and would keep keys that the new value dropped.
       if app_state_delta:
         current_app.update(app_state_delta)
-        transaction.set(app_ref, current_app, merge=True)
+        transaction.set(app_ref, current_app)
 
       if user_state_delta:
         current_user.update(user_state_delta)
-        transaction.set(user_ref, current_user, merge=True)
+        transaction.set(user_ref, current_user)
 
       transaction.set(session_ref, session_data)
       return current_app, current_user
@@ -324,7 +326,7 @@ async def get_session(
           after_dt = datetime.fromtimestamp(
               config.after_timestamp, tz=timezone.utc
           )
-          query = query.where("timestamp", ">=", after_dt)
+          query = query.where(filter=FieldFilter("timestamp", ">=", after_dt))
         if config.num_recent_events is not None:
           query = query.limit_to_last(config.num_recent_events)
 
@@ -368,12 +370,12 @@ async def list_sessions(
     """Lists sessions from Firestore."""
     if user_id:
       query = self._get_sessions_ref(app_name, user_id).where(
-          "appName", "==", app_name
+          filter=FieldFilter("appName", "==", app_name)
       )
       docs = await query.get()
     else:
       query = self.client.collection_group(self.sessions_collection).where(
-          "appName", "==", app_name
+          filter=FieldFilter("appName", "==", app_name)
       )
       docs = await query.get()
 
@@ -441,6 +443,21 @@ def _iter_sessions_data() -> Iterator[dict[str, Any]]:
     sessions.sort(key=lambda s: (s.last_update_time, s.user_id, s.id))
     return ListSessionsResponse(sessions=sessions)
 
+  async def get_user_state(
+      self, *, app_name: str, user_id: str
+  ) -> dict[str, Any]:
+    """Gets the user-scoped state from Firestore."""
+    user_ref = (
+        self.client.collection(self.user_state_collection)
+        .document(app_name)
+        .collection("users")
+        .document(user_id)
+    )
+    user_doc = await user_ref.get()
+    if not user_doc.exists:
+      return {}
+    return user_doc.to_dict() or {}
+
   async def delete_session(
       self, *, app_name: str, user_id: str, session_id: str
   ) -> None:
@@ -544,16 +561,16 @@ async def _append_txn(transaction: firestore.AsyncTransaction) -> int:
             else None
         )
 
-        # 2. Writes
+        # 2. Writes. Documents are written whole, as in create_session.
         if app_updates and app_snap is not None:
           current_app = (app_snap.to_dict() or {}) if app_snap.exists else {}
           current_app.update(app_updates)
-          transaction.set(app_ref, current_app, merge=True)
+          transaction.set(app_ref, current_app)
 
         if user_updates and user_snap is not None:
-          current_user = user_snap.to_dict() if user_snap.exists else {}
+          current_user = (user_snap.to_dict() or {}) if user_snap.exists else {}
           current_user.update(user_updates)
-          transaction.set(user_ref, current_user, merge=True)
+          transaction.set(user_ref, current_user)
 
         new_revision = current_revision + 1
 
@@ -588,7 +605,10 @@ async def _append_txn(transaction: firestore.AsyncTransaction) -> int:
             event_ref,
             {
                 "event_data": event_data,
-                "timestamp": firestore.SERVER_TIMESTAMP,
+                # Event time, not write time: after_timestamp filters on it.
+                "timestamp": datetime.fromtimestamp(
+                    event.timestamp, tz=timezone.utc
+                ),
                 "appName": session.app_name,
                 "userId": session.user_id,
             },
diff --git a/tests/unittests/integrations/firestore/test_firestore_memory_service.py b/tests/unittests/integrations/firestore/test_firestore_memory_service.py
index afa7f75cacf..acebc1816fc 100644
--- a/tests/unittests/integrations/firestore/test_firestore_memory_service.py
+++ b/tests/unittests/integrations/firestore/test_firestore_memory_service.py
@@ -39,6 +39,13 @@ def mock_firestore_client():
   return client
 
 
+def test_memory_service_is_importable_from_package():
+  """FirestoreMemoryService can be imported from the integration package."""
+  from google.adk.integrations import firestore as firestore_integration
+
+  assert firestore_integration.FirestoreMemoryService is FirestoreMemoryService
+
+
 def test_extract_keywords(mock_firestore_client):
   service = FirestoreMemoryService(client=mock_firestore_client)
   text = "The quick brown fox jumps over the lazy dog."
@@ -291,6 +298,152 @@ async def test_add_session_to_memory(mock_firestore_client):
   assert data["timestamp"] == 1234567890.0
 
 
+def _text_event(text, event_id):
+  return Event(
+      id=event_id,
+      invocation_id="test_inv",
+      author="user",
+      content=types.Content(parts=[types.Part.from_text(text=text)]),
+  )
+
+
+def _written_doc_ids(client):
+  memories = client.collection.return_value
+  return [call.args[0] for call in memories.document.call_args_list]
+
+
+@pytest.mark.asyncio
+async def test_add_session_to_memory_twice_overwrites_instead_of_duplicating(
+    mock_firestore_client,
+):
+  """Re-adding a session writes to the same memory documents as before."""
+  from google.adk.sessions.session import Session
+
+  service = FirestoreMemoryService(client=mock_firestore_client)
+  mock_firestore_client.batch.return_value.commit = mock.AsyncMock()
+  session = Session(
+      id="s1",
+      app_name="test_app",
+      user_id="test_user",
+      events=[
+          _text_event("quick brown fox", "e1"),
+          _text_event("lazy dog", "e2"),
+      ],
+  )
+
+  await service.add_session_to_memory(session)
+  first_ids = _written_doc_ids(mock_firestore_client)
+  await service.add_session_to_memory(session)
+  all_ids = _written_doc_ids(mock_firestore_client)
+
+  assert len(set(first_ids)) == 2
+  assert all_ids == first_ids + first_ids
+
+
+@pytest.mark.asyncio
+async def test_add_session_to_memory_records_the_session_id(
+    mock_firestore_client,
+):
+  """Each memory document stores the ID of the session it came from."""
+  from google.adk.sessions.session import Session
+
+  service = FirestoreMemoryService(client=mock_firestore_client)
+  batch = mock_firestore_client.batch.return_value
+  batch.commit = mock.AsyncMock()
+  session = Session(
+      id="s1",
+      app_name="test_app",
+      user_id="test_user",
+      events=[_text_event("quick brown fox", "e1")],
+  )
+
+  await service.add_session_to_memory(session)
+
+  assert batch.set.call_args.args[1]["sessionId"] == "s1"
+
+
+@pytest.mark.asyncio
+async def test_same_session_id_for_different_users_does_not_collide(
+    mock_firestore_client,
+):
+  """Users sharing a session ID get separate memory documents."""
+  from google.adk.sessions.session import Session
+
+  service = FirestoreMemoryService(client=mock_firestore_client)
+  mock_firestore_client.batch.return_value.commit = mock.AsyncMock()
+  event = _text_event("quick brown fox", "e1")
+
+  for user_id in ("alice", "bob"):
+    await service.add_session_to_memory(
+        Session(id="s1", app_name="test_app", user_id=user_id, events=[event])
+    )
+
+  alice_id, bob_id = _written_doc_ids(mock_firestore_client)
+  assert alice_id != bob_id
+
+
+@pytest.mark.asyncio
+async def test_add_events_to_memory_writes_the_given_events(
+    mock_firestore_client,
+):
+  """add_events_to_memory stores each given event under the app and user."""
+  service = FirestoreMemoryService(client=mock_firestore_client)
+  batch = mock_firestore_client.batch.return_value
+  batch.commit = mock.AsyncMock()
+
+  await service.add_events_to_memory(
+      app_name="test_app",
+      user_id="test_user",
+      session_id="s1",
+      events=[_text_event("quick brown fox", "e1")],
+  )
+
+  data = batch.set.call_args.args[1]
+  assert data["appName"] == "test_app"
+  assert data["userId"] == "test_user"
+  assert data["sessionId"] == "s1"
+  assert "quick" in data["keywords"]
+
+
+@pytest.mark.asyncio
+async def test_add_events_to_memory_without_session_id(mock_firestore_client):
+  """Events added without a session ID are stored with no session ID."""
+  service = FirestoreMemoryService(client=mock_firestore_client)
+  batch = mock_firestore_client.batch.return_value
+  batch.commit = mock.AsyncMock()
+
+  await service.add_events_to_memory(
+      app_name="test_app",
+      user_id="test_user",
+      events=[_text_event("quick brown fox", "e1")],
+  )
+
+  assert batch.set.call_args.args[1]["sessionId"] is None
+  batch.commit.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_add_events_then_session_does_not_duplicate(
+    mock_firestore_client,
+):
+  """An event added as a delta and later with its session keeps one entry."""
+  from google.adk.sessions.session import Session
+
+  service = FirestoreMemoryService(client=mock_firestore_client)
+  mock_firestore_client.batch.return_value.commit = mock.AsyncMock()
+  event = _text_event("quick brown fox", "e1")
+
+  await service.add_events_to_memory(
+      app_name="test_app", user_id="test_user", session_id="s1", events=[event]
+  )
+  await service.add_session_to_memory(
+      Session(id="s1", app_name="test_app", user_id="test_user", events=[event])
+  )
+
+  delta_id, session_id = _written_doc_ids(mock_firestore_client)
+  assert delta_id == session_id
+
+
 @pytest.mark.asyncio
 async def test_add_session_to_memory_no_events(mock_firestore_client):
   service = FirestoreMemoryService(client=mock_firestore_client)
diff --git a/tests/unittests/integrations/firestore/test_firestore_session_service.py b/tests/unittests/integrations/firestore/test_firestore_session_service.py
index 7b13139288d..ec90d3c3809 100644
--- a/tests/unittests/integrations/firestore/test_firestore_session_service.py
+++ b/tests/unittests/integrations/firestore/test_firestore_session_service.py
@@ -35,6 +35,20 @@
 import pytest
 
 
+def _where_filter(where_mock):
+  """Returns (field, op, value) of the FieldFilter passed as `filter=`."""
+  field_filter = where_mock.call_args.kwargs["filter"]
+  return field_filter.field_path, field_filter.op_string, field_filter.value
+
+
+def _stored_snapshot(data):
+  """Returns a snapshot of an existing document holding `data`."""
+  snapshot = mock.MagicMock()
+  snapshot.exists = True
+  snapshot.to_dict.return_value = data
+  return snapshot
+
+
 @pytest.fixture
 def mock_firestore_client():
   client = mock.MagicMock()
@@ -82,6 +96,15 @@ def mock_firestore_client():
   return client
 
 
+def test_session_service_is_importable_from_package():
+  """FirestoreSessionService can be imported from the integration package."""
+  from google.adk.integrations import firestore as firestore_integration
+
+  assert (
+      firestore_integration.FirestoreSessionService is FirestoreSessionService
+  )
+
+
 @pytest.mark.asyncio
 async def test_create_session(mock_firestore_client):
 
@@ -274,6 +297,37 @@ async def test_append_event(mock_firestore_client):
   assert session.last_update_time == event.timestamp
 
 
+@pytest.mark.asyncio
+async def test_append_event_stores_the_event_timestamp(mock_firestore_client):
+  """The stored event time is Event.timestamp, not the Firestore write time."""
+  service = FirestoreSessionService(client=mock_firestore_client)
+  session = Session(id="test_session", app_name="test_app", user_id="test_user")
+  event = Event(
+      invocation_id="test_inv", author="user", timestamp=1700000000.123456
+  )
+
+  root_coll = mock_firestore_client.collection.return_value
+  user_ref = (
+      root_coll.document.return_value.collection.return_value.document.return_value
+  )
+  session_doc_ref = user_ref.collection.return_value.document.return_value
+  session_doc_ref.get = mock.AsyncMock(
+      return_value=_stored_snapshot({"revision": 0})
+  )
+  event_ref = session_doc_ref.collection.return_value.document.return_value
+
+  with mock.patch("google.cloud.firestore.async_transactional", lambda x: x):
+    await service.append_event(session, event)
+
+  transaction = mock_firestore_client.transaction.return_value
+  writes = {
+      call.args[0]: call.args[1] for call in transaction.set.call_args_list
+  }
+  assert writes[event_ref]["timestamp"] == datetime.fromtimestamp(
+      1700000000.123456, tz=timezone.utc
+  )
+
+
 @pytest.mark.asyncio
 async def test_append_event_session_not_found(mock_firestore_client):
   service = FirestoreSessionService(client=mock_firestore_client)
@@ -493,6 +547,84 @@ async def test_create_session_keeps_app_and_user_state_native(
   assert isinstance(persisted_state["session_key"], str)
 
 
+@pytest.mark.asyncio
+async def test_append_event_replaces_nested_app_and_user_state(
+    mock_firestore_client,
+):
+  """A dict-valued app/user delta replaces the stored dict, dropping old keys."""
+  service = FirestoreSessionService(client=mock_firestore_client)
+  session = Session(id="test_session", app_name="test_app", user_id="test_user")
+  old_value = {"name": "alice", "role": "admin"}
+  new_value = {"name": "bob"}
+
+  root_coll = mock_firestore_client.collection.return_value
+  app_ref = root_coll.document.return_value
+  users_coll = app_ref.collection.return_value
+  user_ref = users_coll.document.return_value
+  session_doc_ref = user_ref.collection.return_value.document.return_value
+  session_doc_ref.get = mock.AsyncMock(
+      return_value=_stored_snapshot({"revision": 0})
+  )
+  app_ref.get = mock.AsyncMock(
+      return_value=_stored_snapshot({"cfg": old_value})
+  )
+  user_ref.get = mock.AsyncMock(
+      return_value=_stored_snapshot({"cfg": old_value})
+  )
+
+  with mock.patch("google.cloud.firestore.async_transactional", lambda x: x):
+    await service.append_event(
+        session,
+        Event(
+            invocation_id="test_inv",
+            author="user",
+            actions=EventActions(
+                state_delta={"app:cfg": new_value, "user:cfg": new_value}
+            ),
+        ),
+    )
+
+  transaction = mock_firestore_client.transaction.return_value
+  writes = {call.args[0]: call for call in transaction.set.call_args_list}
+  for ref in (app_ref, user_ref):
+    assert writes[ref].args[1] == {"cfg": new_value}
+    assert not writes[ref].kwargs.get("merge")
+
+
+@pytest.mark.asyncio
+async def test_create_session_replaces_nested_app_and_user_state(
+    mock_firestore_client,
+):
+  """Initial app and user state replace stored dict values instead of merging."""
+  service = FirestoreSessionService(client=mock_firestore_client)
+  old_value = {"name": "alice", "role": "admin"}
+
+  root_coll = mock_firestore_client.collection.return_value
+  app_ref = root_coll.document.return_value
+  user_ref = app_ref.collection.return_value.document.return_value
+  app_ref.get = mock.AsyncMock(
+      return_value=_stored_snapshot({"cfg": old_value})
+  )
+  user_ref.get = mock.AsyncMock(
+      return_value=_stored_snapshot({"cfg": old_value})
+  )
+
+  with mock.patch("google.cloud.firestore.async_transactional", lambda x: x):
+    session = await service.create_session(
+        app_name="test_app",
+        user_id="test_user",
+        state={"app:cfg": {"name": "bob"}, "user:cfg": {"name": "bob"}},
+    )
+
+  transaction = mock_firestore_client.transaction.return_value
+  writes = {call.args[0]: call for call in transaction.set.call_args_list}
+  for ref in (app_ref, user_ref):
+    assert writes[ref].args[1] == {"cfg": {"name": "bob"}}
+    assert not writes[ref].kwargs.get("merge")
+  assert session.state["app:cfg"] == {"name": "bob"}
+  assert session.state["user:cfg"] == {"name": "bob"}
+
+
 @pytest.mark.asyncio
 async def test_append_event_with_temp_state(mock_firestore_client):
   service = FirestoreSessionService(client=mock_firestore_client)
@@ -760,9 +892,9 @@ async def mock_get_all(refs):
   assert session.last_update_time == 1234567890.0
 
   mock_firestore_client.collection_group.assert_called_once_with("sessions")
-  mock_firestore_client.collection_group.return_value.where.assert_called_once_with(
-      "appName", "==", app_name
-  )
+  where = mock_firestore_client.collection_group.return_value.where
+  where.assert_called_once()
+  assert _where_filter(where) == ("appName", "==", app_name)
 
 
 @pytest.mark.asyncio
@@ -824,10 +956,40 @@ async def mock_get_all(refs):
   assert response.sessions[0].app_name == app_name
 
   mock_firestore_client.collection_group.assert_called_once_with("sessions")
-  mock_firestore_client.collection_group.return_value.where.assert_called_once_with(
-      "appName", "==", app_name
+  where = mock_firestore_client.collection_group.return_value.where
+  where.assert_called_once()
+  assert _where_filter(where) == ("appName", "==", app_name)
+
+
+@pytest.mark.asyncio
+async def test_get_user_state_returns_stored_user_state(mock_firestore_client):
+  """get_user_state returns the user_states document for the app and user."""
+  service = FirestoreSessionService(client=mock_firestore_client)
+  root_coll = mock_firestore_client.collection.return_value
+  users_coll = root_coll.document.return_value.collection.return_value
+  users_coll.document.return_value.get = mock.AsyncMock(
+      return_value=_stored_snapshot({"theme": "dark"})
   )
 
+  state = await service.get_user_state(app_name="test_app", user_id="alice")
+
+  assert state == {"theme": "dark"}
+  mock_firestore_client.collection.assert_called_with("user_states")
+  root_coll.document.assert_called_with("test_app")
+  users_coll.document.assert_called_with("alice")
+
+
+@pytest.mark.asyncio
+async def test_get_user_state_is_empty_when_nothing_is_stored(
+    mock_firestore_client,
+):
+  """get_user_state returns an empty dict for a user without stored state."""
+  service = FirestoreSessionService(client=mock_firestore_client)
+
+  state = await service.get_user_state(app_name="test_app", user_id="alice")
+
+  assert state == {}
+
 
 @pytest.mark.asyncio
 async def test_create_session_already_exists(mock_firestore_client):
@@ -965,7 +1127,7 @@ async def test_get_session_after_timestamp_cursor_is_utc_aware(
 ):
   """The after_timestamp cursor must be an aware UTC datetime.
 
-  Events are written with an aware UTC server timestamp, so a naive local
+  Events are written with an aware UTC timestamp, so a naive local
   cursor is compared against them shifted by the host's UTC offset: it
   replays events west of UTC and silently drops them east of it.
   """
@@ -999,7 +1161,7 @@ async def test_get_session_after_timestamp_cursor_is_utc_aware(
     )
 
   events_collection_ref.where.assert_called_once()
-  field, operator, cursor = events_collection_ref.where.call_args.args
+  field, operator, cursor = _where_filter(events_collection_ref.where)
   assert (field, operator) == ("timestamp", ">=")
   assert cursor.utcoffset() == timedelta(0), f"cursor is not UTC: {cursor!r}"
   assert _wire_epoch(cursor) == after_timestamp