From e016c25169675083884b684103345829cb9eeee7 Mon Sep 17 00:00:00 2001
From: Aditya Singh <adisin650@gmail.com>
Date: Wed, 5 Aug 2026 06:12:09 -0700
Subject: [PATCH 1/9] fix(memory): preserve DaprSession created_at across
 writes

DaprSession.add_items rewrites the whole metadata document on every call
and sets created_at to the current clock, so a session's persisted
created_at always equals updated_at and its age is unrecoverable after
the second turn.

Read the stored created_at first and keep it when it is present, matching
RedisSession, which uses hsetnx, and MongoDBSession, which uses
setOnInsert.
---
 src/agents/extensions/memory/dapr_session.py | 26 +++++++++++++++++---
 tests/extensions/memory/test_dapr_session.py | 24 ++++++++++++++++++
 2 files changed, 47 insertions(+), 3 deletions(-)

diff --git a/src/agents/extensions/memory/dapr_session.py b/src/agents/extensions/memory/dapr_session.py
index 20b300ce3c..2c1c245993 100644
--- a/src/agents/extensions/memory/dapr_session.py
+++ b/src/agents/extensions/memory/dapr_session.py
@@ -184,6 +184,25 @@ def _get_metadata(self) -> dict[str, str]:
             metadata["ttlInSeconds"] = str(self._ttl)
         return metadata
 
+    async def _read_created_at(self) -> str | None:
+        """Return the stored creation timestamp, or None when it is missing or unreadable."""
+        response = await self._dapr_client.get_state(
+            store_name=self._state_store_name,
+            key=self._metadata_key,
+            state_metadata=self._get_read_metadata(),
+        )
+        data = response.data
+        if not data:
+            return None
+        try:
+            stored = json.loads(data.decode("utf-8"))
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            return None
+        if not isinstance(stored, dict):
+            return None
+        created_at = stored.get("created_at")
+        return created_at if isinstance(created_at, str) and created_at else None
+
     async def _serialize_item(self, item: TResponseInputItem) -> str:
         """Serialize an item to JSON string. Can be overridden by subclasses."""
         return json.dumps(item, separators=(",", ":"))
@@ -362,11 +381,12 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
                         continue
                     raise
 
-            # Update metadata
+            # Update metadata, preserving created_at across subsequent writes.
+            now = str(int(time.time()))
             metadata = {
                 "session_id": self.session_id,
-                "created_at": str(int(time.time())),
-                "updated_at": str(int(time.time())),
+                "created_at": await self._read_created_at() or now,
+                "updated_at": now,
             }
             await self._dapr_client.save_state(
                 store_name=self._state_store_name,
diff --git a/tests/extensions/memory/test_dapr_session.py b/tests/extensions/memory/test_dapr_session.py
index 702cb7388c..3dd57d03bf 100644
--- a/tests/extensions/memory/test_dapr_session.py
+++ b/tests/extensions/memory/test_dapr_session.py
@@ -1327,3 +1327,27 @@ def entry_signalling_resolve(*args: Any, **kwargs: Any) -> Any:
                 task.cancel()
                 with suppress(asyncio.CancelledError, RuntimeError):
                     await task
+
+
+async def test_add_items_preserves_created_at_metadata(
+    fake_dapr_client: FakeDaprClient, monkeypatch: pytest.MonkeyPatch
+):
+    """`created_at` must be set once and not overwritten by subsequent add_items calls."""
+    import agents.extensions.memory.dapr_session as dapr_session_module
+
+    session = await _create_test_session(fake_dapr_client)
+
+    try:
+        monkeypatch.setattr(dapr_session_module.time, "time", lambda: 1000.0)
+        await session.add_items([{"role": "user", "content": "first"}])
+        first = json.loads(fake_dapr_client._state[session._metadata_key].decode("utf-8"))
+        assert first["created_at"] == "1000"
+        assert first["updated_at"] == "1000"
+
+        monkeypatch.setattr(dapr_session_module.time, "time", lambda: 2000.0)
+        await session.add_items([{"role": "user", "content": "second"}])
+        second = json.loads(fake_dapr_client._state[session._metadata_key].decode("utf-8"))
+        assert second["created_at"] == "1000"
+        assert second["updated_at"] == "2000"
+    finally:
+        await session.close()

From 2c7eb2534370ed71bce5b89a573cc9a483a647ba Mon Sep 17 00:00:00 2001
From: Aditya Singh <adisin650@gmail.com>
Date: Wed, 5 Aug 2026 07:02:21 -0700
Subject: [PATCH 2/9] Make the DaprSession metadata write conditional on its
 etag

The metadata save was unconditional while the messages save used etags, so
two processes appending to the same new session could both read no metadata,
each pick their own now, and let the later save overwrite created_at. Carry
the metadata etag through the read and save with first_write concurrency,
retrying through the existing conflict handler.

Also switch the created_at tests to a string monkeypatch target, which clears
four mypy attr-defined errors on the module's re-exported time attribute.
---
 src/agents/extensions/memory/dapr_session.py | 60 +++++++++++++-------
 tests/extensions/memory/test_dapr_session.py | 55 ++++++++++++++++--
 2 files changed, 91 insertions(+), 24 deletions(-)

diff --git a/src/agents/extensions/memory/dapr_session.py b/src/agents/extensions/memory/dapr_session.py
index 2c1c245993..6d3f6c7c97 100644
--- a/src/agents/extensions/memory/dapr_session.py
+++ b/src/agents/extensions/memory/dapr_session.py
@@ -184,24 +184,29 @@ def _get_metadata(self) -> dict[str, str]:
             metadata["ttlInSeconds"] = str(self._ttl)
         return metadata
 
-    async def _read_created_at(self) -> str | None:
-        """Return the stored creation timestamp, or None when it is missing or unreadable."""
+    async def _read_created_at(self) -> tuple[str | None, str | None]:
+        """Return the stored creation timestamp and the metadata etag backing it.
+
+        The timestamp is None when it is missing or unreadable. The etag is returned so the
+        caller can write the metadata conditionally against the same revision it read.
+        """
         response = await self._dapr_client.get_state(
             store_name=self._state_store_name,
             key=self._metadata_key,
             state_metadata=self._get_read_metadata(),
         )
+        etag = response.etag
         data = response.data
         if not data:
-            return None
+            return None, etag
         try:
             stored = json.loads(data.decode("utf-8"))
         except (json.JSONDecodeError, UnicodeDecodeError):
-            return None
+            return None, etag
         if not isinstance(stored, dict):
-            return None
+            return None, etag
         created_at = stored.get("created_at")
-        return created_at if isinstance(created_at, str) and created_at else None
+        return (created_at if isinstance(created_at, str) and created_at else None), etag
 
     async def _serialize_item(self, item: TResponseInputItem) -> str:
         """Serialize an item to JSON string. Can be overridden by subclasses."""
@@ -381,20 +386,35 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
                         continue
                     raise
 
-            # Update metadata, preserving created_at across subsequent writes.
-            now = str(int(time.time()))
-            metadata = {
-                "session_id": self.session_id,
-                "created_at": await self._read_created_at() or now,
-                "updated_at": now,
-            }
-            await self._dapr_client.save_state(
-                store_name=self._state_store_name,
-                key=self._metadata_key,
-                value=json.dumps(metadata),
-                state_metadata=self._get_metadata(),
-                options=self._get_state_options(),
-            )
+            # Update metadata, preserving created_at across subsequent writes. This mirrors the
+            # etag-guarded loop used for the messages key above. A plain write would let two
+            # processes appending to the same new session each read no metadata, pick their own
+            # now, and have the later save overwrite created_at with the later timestamp.
+            attempt = 0
+            while True:
+                attempt += 1
+                stored_created_at, metadata_etag = await self._read_created_at()
+                now = str(int(time.time()))
+                metadata = {
+                    "session_id": self.session_id,
+                    "created_at": stored_created_at or now,
+                    "updated_at": now,
+                }
+                try:
+                    await self._dapr_client.save_state(
+                        store_name=self._state_store_name,
+                        key=self._metadata_key,
+                        value=json.dumps(metadata),
+                        etag=metadata_etag,
+                        state_metadata=self._get_metadata(),
+                        options=self._get_state_options(concurrency=Concurrency.first_write),
+                    )
+                    break
+                except Exception as error:
+                    should_retry = await self._handle_concurrency_conflict(error, attempt)
+                    if should_retry:
+                        continue
+                    raise
 
     async def pop_item(self) -> TResponseInputItem | None:
         """Remove and return the most recent item from the session.
diff --git a/tests/extensions/memory/test_dapr_session.py b/tests/extensions/memory/test_dapr_session.py
index 3dd57d03bf..67d5d44418 100644
--- a/tests/extensions/memory/test_dapr_session.py
+++ b/tests/extensions/memory/test_dapr_session.py
@@ -1333,21 +1333,68 @@ async def test_add_items_preserves_created_at_metadata(
     fake_dapr_client: FakeDaprClient, monkeypatch: pytest.MonkeyPatch
 ):
     """`created_at` must be set once and not overwritten by subsequent add_items calls."""
-    import agents.extensions.memory.dapr_session as dapr_session_module
-
     session = await _create_test_session(fake_dapr_client)
 
     try:
-        monkeypatch.setattr(dapr_session_module.time, "time", lambda: 1000.0)
+        monkeypatch.setattr("agents.extensions.memory.dapr_session.time.time", lambda: 1000.0)
         await session.add_items([{"role": "user", "content": "first"}])
         first = json.loads(fake_dapr_client._state[session._metadata_key].decode("utf-8"))
         assert first["created_at"] == "1000"
         assert first["updated_at"] == "1000"
 
-        monkeypatch.setattr(dapr_session_module.time, "time", lambda: 2000.0)
+        monkeypatch.setattr("agents.extensions.memory.dapr_session.time.time", lambda: 2000.0)
         await session.add_items([{"role": "user", "content": "second"}])
         second = json.loads(fake_dapr_client._state[session._metadata_key].decode("utf-8"))
         assert second["created_at"] == "1000"
         assert second["updated_at"] == "2000"
     finally:
         await session.close()
+
+
+async def test_concurrent_first_add_items_does_not_regress_created_at(
+    fake_dapr_client: FakeDaprClient, monkeypatch: pytest.MonkeyPatch
+):
+    """A racing first write must not overwrite `created_at` with its own later timestamp.
+
+    Two processes appending to the same new session can both read the metadata key before
+    either save is visible, so both pick their own `now`. The metadata save is conditional on
+    the etag that was read, so the loser retries and adopts the winner's `created_at`.
+    """
+    session_id = "shared_session_created_at_race"
+    winner = await _create_test_session(fake_dapr_client, session_id=session_id)
+    loser = await _create_test_session(fake_dapr_client, session_id=session_id)
+
+    try:
+        # Capture the state the loser observes before the winner's metadata save lands: the
+        # metadata key does not exist yet, so there is no created_at and no etag.
+        stale_read = (None, None)
+        real_read = loser._read_created_at
+        calls = 0
+
+        async def read_stale_once() -> tuple[str | None, str | None]:
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                return stale_read
+            return await real_read()
+
+        monkeypatch.setattr("agents.extensions.memory.dapr_session.time.time", lambda: 1000.0)
+        await winner.add_items([{"role": "user", "content": "winner"}])
+        assert (
+            json.loads(fake_dapr_client._state[winner._metadata_key].decode("utf-8"))["created_at"]
+            == "1000"
+        )
+
+        monkeypatch.setattr(loser, "_read_created_at", read_stale_once)
+        monkeypatch.setattr("agents.extensions.memory.dapr_session.time.time", lambda: 2000.0)
+        await loser.add_items([{"role": "user", "content": "loser"}])
+
+        final = json.loads(fake_dapr_client._state[loser._metadata_key].decode("utf-8"))
+        # The stale save is rejected, so the retry reads the winner's value and keeps it.
+        assert final["created_at"] == "1000"
+        assert final["updated_at"] == "2000"
+        # Two reads means the conditional save actually rejected the stale one and retried.
+        assert calls == 2
+    finally:
+        await winner.close()
+        await loser.close()

From d4f4d3ba55837bac7513ac64bebcfa4e71d9ef90 Mon Sep 17 00:00:00 2001
From: Aditya Singh <adisin650@gmail.com>
Date: Wed, 5 Aug 2026 15:51:04 -0700
Subject: [PATCH 3/9] Do not fail an append when only the metadata write gives
 up

The messages key is committed before the metadata key, so exhausting the
metadata retry budget raised from add_items() for a batch that was already
stored. A caller acting on that error by retrying would append the same items
a second time.

Treat the post-commit metadata refresh as best effort and log a warning when
it gives up, since it is derived bookkeeping rather than conversation state.
---
 src/agents/extensions/memory/dapr_session.py | 27 ++++++++++----
 tests/extensions/memory/test_dapr_session.py | 39 ++++++++++++++++++++
 2 files changed, 58 insertions(+), 8 deletions(-)

diff --git a/src/agents/extensions/memory/dapr_session.py b/src/agents/extensions/memory/dapr_session.py
index 6d3f6c7c97..c942b29349 100644
--- a/src/agents/extensions/memory/dapr_session.py
+++ b/src/agents/extensions/memory/dapr_session.py
@@ -390,17 +390,22 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
             # etag-guarded loop used for the messages key above. A plain write would let two
             # processes appending to the same new session each read no metadata, pick their own
             # now, and have the later save overwrite created_at with the later timestamp.
+            #
+            # The messages key is already committed once we reach here, so raising would tell
+            # the caller the append failed while their items are in the session, and the
+            # natural response of retrying add_items() would store the batch twice. Metadata
+            # is derived bookkeeping, so give up on it with a warning instead.
             attempt = 0
             while True:
                 attempt += 1
-                stored_created_at, metadata_etag = await self._read_created_at()
-                now = str(int(time.time()))
-                metadata = {
-                    "session_id": self.session_id,
-                    "created_at": stored_created_at or now,
-                    "updated_at": now,
-                }
                 try:
+                    stored_created_at, metadata_etag = await self._read_created_at()
+                    now = str(int(time.time()))
+                    metadata = {
+                        "session_id": self.session_id,
+                        "created_at": stored_created_at or now,
+                        "updated_at": now,
+                    }
                     await self._dapr_client.save_state(
                         store_name=self._state_store_name,
                         key=self._metadata_key,
@@ -414,7 +419,13 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
                     should_retry = await self._handle_concurrency_conflict(error, attempt)
                     if should_retry:
                         continue
-                    raise
+                    logger.warning(
+                        "DaprSession stored the new items for session %s but could not update "
+                        "its metadata: %s",
+                        self.session_id,
+                        error,
+                    )
+                    break
 
     async def pop_item(self) -> TResponseInputItem | None:
         """Remove and return the most recent item from the session.
diff --git a/tests/extensions/memory/test_dapr_session.py b/tests/extensions/memory/test_dapr_session.py
index 67d5d44418..f1aa3e053e 100644
--- a/tests/extensions/memory/test_dapr_session.py
+++ b/tests/extensions/memory/test_dapr_session.py
@@ -1398,3 +1398,42 @@ async def read_stale_once() -> tuple[str | None, str | None]:
     finally:
         await winner.close()
         await loser.close()
+
+
+async def test_add_items_survives_metadata_write_giving_up(
+    fake_dapr_client: FakeDaprClient, monkeypatch: pytest.MonkeyPatch, caplog: Any
+):
+    """A metadata write that exhausts its retries must not fail an append that already landed.
+
+    The messages key is saved before the metadata key, so raising here would report failure for
+    items that are already in the session, and the natural retry of `add_items` would store the
+    same batch a second time.
+    """
+    import logging
+
+    session = await _create_test_session(fake_dapr_client, "metadata_gives_up")
+
+    try:
+        real_save = fake_dapr_client.save_state
+
+        async def fail_only_metadata_saves(
+            store_name: str,
+            key: str,
+            value: str | bytes,
+            **kwargs: Any,
+        ) -> None:
+            if key == session._metadata_key:
+                raise RuntimeError("etag mismatch")
+            await real_save(store_name, key, value, **kwargs)
+
+        monkeypatch.setattr(fake_dapr_client, "save_state", fail_only_metadata_saves)
+        monkeypatch.setattr(session, "_calculate_retry_delay", lambda attempt: 0.0)
+
+        with caplog.at_level(logging.WARNING):
+            await session.add_items([{"role": "user", "content": "kept"}])
+
+        items = await session.get_items()
+        assert [item["content"] for item in items] == ["kept"]
+        assert any("could not update" in record.getMessage() for record in caplog.records)
+    finally:
+        await session.close()

From d829b2f0cab1027d6e8c769142f2cd40be15a973 Mon Sep 17 00:00:00 2001
From: Aditya Singh <adisin650@gmail.com>
Date: Wed, 5 Aug 2026 16:04:18 -0700
Subject: [PATCH 4/9] Redact the metadata warning through the shared logging
 helper

The warning put the caller-supplied session id and the provider exception
straight on the LogRecord, so a sidecar error carrying tenant or backend
detail was logged even with the SDK data flags enabled.

Route it through log_model_and_tool_action_warning with a fixed message and
the session id supplied as diagnostic context instead.
---
 src/agents/extensions/memory/dapr_session.py | 15 ++++++++++-----
 tests/extensions/memory/test_dapr_session.py | 11 ++++++++++-
 2 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/src/agents/extensions/memory/dapr_session.py b/src/agents/extensions/memory/dapr_session.py
index c942b29349..8ab551a427 100644
--- a/src/agents/extensions/memory/dapr_session.py
+++ b/src/agents/extensions/memory/dapr_session.py
@@ -43,7 +43,11 @@
     )
 
 from ...items import TResponseInputItem
-from ...logger import log_model_and_tool_action_error, logger
+from ...logger import (
+    log_model_and_tool_action_error,
+    log_model_and_tool_action_warning,
+    logger,
+)
 from ...memory.session import SessionABC
 from ...memory.session_settings import (
     SessionSettings,
@@ -419,11 +423,12 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
                     should_retry = await self._handle_concurrency_conflict(error, attempt)
                     if should_retry:
                         continue
-                    logger.warning(
-                        "DaprSession stored the new items for session %s but could not update "
-                        "its metadata: %s",
-                        self.session_id,
+                    log_model_and_tool_action_warning(
+                        logger,
+                        "DaprSession stored the new items but could not update the session "
+                        "metadata",
                         error,
+                        diagnostic_extra=lambda: {"session_id": self.session_id},
                     )
                     break
 
diff --git a/tests/extensions/memory/test_dapr_session.py b/tests/extensions/memory/test_dapr_session.py
index f1aa3e053e..b53697b3b8 100644
--- a/tests/extensions/memory/test_dapr_session.py
+++ b/tests/extensions/memory/test_dapr_session.py
@@ -1434,6 +1434,15 @@ async def fail_only_metadata_saves(
 
         items = await session.get_items()
         assert [item["content"] for item in items] == ["kept"]
-        assert any("could not update" in record.getMessage() for record in caplog.records)
+
+        warnings = [
+            record for record in caplog.records if "could not update" in record.getMessage()
+        ]
+        assert warnings
+        # Data logging is off by default, so neither the caller-supplied session id nor the
+        # provider error text may reach the record.
+        for record in warnings:
+            assert "metadata_gives_up" not in record.getMessage()
+            assert "etag mismatch" not in record.getMessage()
     finally:
         await session.close()

From a80e04006d7f02d45ec1fccb2e2bbe14df6e12dc Mon Sep 17 00:00:00 2001
From: Aditya Singh <adisin650@gmail.com>
Date: Wed, 5 Aug 2026 16:53:14 -0700
Subject: [PATCH 5/9] Read the item content with .get in the metadata failure
 test

TResponseInputItem is a union of TypedDicts, most of which have no content
key, so subscripting it fails mypy. Match the .get style the rest of this file
uses.
---
 tests/extensions/memory/test_dapr_session.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/extensions/memory/test_dapr_session.py b/tests/extensions/memory/test_dapr_session.py
index b53697b3b8..a950796580 100644
--- a/tests/extensions/memory/test_dapr_session.py
+++ b/tests/extensions/memory/test_dapr_session.py
@@ -1433,7 +1433,7 @@ async def fail_only_metadata_saves(
             await session.add_items([{"role": "user", "content": "kept"}])
 
         items = await session.get_items()
-        assert [item["content"] for item in items] == ["kept"]
+        assert [item.get("content") for item in items] == ["kept"]
 
         warnings = [
             record for record in caplog.records if "could not update" in record.getMessage()

From 7c30e9af4cfd089756d1a3f1797a4366bb09ac56 Mon Sep 17 00:00:00 2001
From: Aditya Singh <adisin650@gmail.com>
Date: Wed, 5 Aug 2026 17:14:18 -0700
Subject: [PATCH 6/9] Guard the metadata write only when a real etag was read

Dapr treats a write with no etag as last-write-wins even when first-write
concurrency is requested, so asking for it on the create claimed a race
guarantee the store does not provide. Request it only once metadata exists.

The missing-metadata race test relied on the fake rejecting an etag-less write,
which real Dapr accepts. Replace it with a stale non-null etag retry, which is
the guarantee that actually holds and the case that stops a later append from
resetting an established created_at.
---
 src/agents/extensions/memory/dapr_session.py | 21 +++++--
 tests/extensions/memory/test_dapr_session.py | 59 +++++++++++---------
 2 files changed, 49 insertions(+), 31 deletions(-)

diff --git a/src/agents/extensions/memory/dapr_session.py b/src/agents/extensions/memory/dapr_session.py
index 8ab551a427..9549304c6c 100644
--- a/src/agents/extensions/memory/dapr_session.py
+++ b/src/agents/extensions/memory/dapr_session.py
@@ -390,10 +390,17 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
                         continue
                     raise
 
-            # Update metadata, preserving created_at across subsequent writes. This mirrors the
-            # etag-guarded loop used for the messages key above. A plain write would let two
-            # processes appending to the same new session each read no metadata, pick their own
-            # now, and have the later save overwrite created_at with the later timestamp.
+            # Update metadata, preserving created_at across subsequent writes. A plain write
+            # would let a later append overwrite created_at with its own now, so the save is
+            # guarded by the etag that backed the value that was read.
+            #
+            # The guard only applies once metadata exists. Dapr documents a write without an
+            # etag as last-write-wins even when first-write concurrency is requested, so
+            # asking for it on the create is not a race guarantee and is left off rather than
+            # implying one. Two writers creating the key concurrently can therefore both
+            # succeed, and the loser's created_at wins by a fraction of a second. Every write
+            # after that is etag guarded, which is the case that actually matters, since it is
+            # what stops a later append from resetting an established created_at.
             #
             # The messages key is already committed once we reach here, so raising would tell
             # the caller the append failed while their items are in the session, and the
@@ -416,7 +423,11 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
                         value=json.dumps(metadata),
                         etag=metadata_etag,
                         state_metadata=self._get_metadata(),
-                        options=self._get_state_options(concurrency=Concurrency.first_write),
+                        options=self._get_state_options(
+                            concurrency=(
+                                Concurrency.first_write if metadata_etag is not None else None
+                            )
+                        ),
                     )
                     break
                 except Exception as error:
diff --git a/tests/extensions/memory/test_dapr_session.py b/tests/extensions/memory/test_dapr_session.py
index a950796580..d3c33c1a02 100644
--- a/tests/extensions/memory/test_dapr_session.py
+++ b/tests/extensions/memory/test_dapr_session.py
@@ -1351,53 +1351,60 @@ async def test_add_items_preserves_created_at_metadata(
         await session.close()
 
 
-async def test_concurrent_first_add_items_does_not_regress_created_at(
+async def test_stale_metadata_etag_retries_and_keeps_created_at(
     fake_dapr_client: FakeDaprClient, monkeypatch: pytest.MonkeyPatch
 ):
-    """A racing first write must not overwrite `created_at` with its own later timestamp.
+    """A metadata save guarded by a stale etag must retry and keep the stored `created_at`.
 
-    Two processes appending to the same new session can both read the metadata key before
-    either save is visible, so both pick their own `now`. The metadata save is conditional on
-    the etag that was read, so the loser retries and adopts the winner's `created_at`.
+    Once the metadata key exists, every later append reads a real etag and saves against it.
+    A writer holding an etag that another append has already superseded is rejected, so it
+    re-reads and adopts the stored `created_at` rather than replacing it with its own `now`.
+
+    Dapr treats a write with no etag as last-write-wins even when first-write concurrency is
+    requested, so the create is deliberately not covered here. This is the guarantee the
+    store actually provides.
     """
-    session_id = "shared_session_created_at_race"
-    winner = await _create_test_session(fake_dapr_client, session_id=session_id)
-    loser = await _create_test_session(fake_dapr_client, session_id=session_id)
+    session_id = "shared_session_created_at_stale_etag"
+    first = await _create_test_session(fake_dapr_client, session_id=session_id)
+    second = await _create_test_session(fake_dapr_client, session_id=session_id)
 
     try:
-        # Capture the state the loser observes before the winner's metadata save lands: the
-        # metadata key does not exist yet, so there is no created_at and no etag.
-        stale_read = (None, None)
-        real_read = loser._read_created_at
+        monkeypatch.setattr("agents.extensions.memory.dapr_session.time.time", lambda: 1000.0)
+        await first.add_items([{"role": "user", "content": "first"}])
+        created = json.loads(fake_dapr_client._state[first._metadata_key].decode("utf-8"))
+        assert created["created_at"] == "1000"
+        stale_etag = fake_dapr_client._etags[first._metadata_key]
+
+        # A second append supersedes that etag, so the value captured above is now stale but
+        # still non-null, which is the situation the guard is actually for.
+        monkeypatch.setattr("agents.extensions.memory.dapr_session.time.time", lambda: 1500.0)
+        await first.add_items([{"role": "user", "content": "second"}])
+        assert fake_dapr_client._etags[first._metadata_key] != stale_etag
+
+        real_read = second._read_created_at
         calls = 0
 
         async def read_stale_once() -> tuple[str | None, str | None]:
             nonlocal calls
             calls += 1
             if calls == 1:
-                return stale_read
+                # Pretend this writer read the metadata before the second append landed.
+                return "1000", stale_etag
             return await real_read()
 
-        monkeypatch.setattr("agents.extensions.memory.dapr_session.time.time", lambda: 1000.0)
-        await winner.add_items([{"role": "user", "content": "winner"}])
-        assert (
-            json.loads(fake_dapr_client._state[winner._metadata_key].decode("utf-8"))["created_at"]
-            == "1000"
-        )
-
-        monkeypatch.setattr(loser, "_read_created_at", read_stale_once)
+        monkeypatch.setattr(second, "_read_created_at", read_stale_once)
         monkeypatch.setattr("agents.extensions.memory.dapr_session.time.time", lambda: 2000.0)
-        await loser.add_items([{"role": "user", "content": "loser"}])
+        await second.add_items([{"role": "user", "content": "third"}])
 
-        final = json.loads(fake_dapr_client._state[loser._metadata_key].decode("utf-8"))
-        # The stale save is rejected, so the retry reads the winner's value and keeps it.
+        final = json.loads(fake_dapr_client._state[second._metadata_key].decode("utf-8"))
+        # The stale save is rejected, so the retry reads the stored value and keeps it.
         assert final["created_at"] == "1000"
         assert final["updated_at"] == "2000"
         # Two reads means the conditional save actually rejected the stale one and retried.
         assert calls == 2
     finally:
-        await winner.close()
-        await loser.close()
+        await first.close()
+        await second.close()
 
 
 async def test_add_items_survives_metadata_write_giving_up(

From b62d510c744352a876f6f2fb840fac86dd1935ba Mon Sep 17 00:00:00 2001
From: Aditya Singh <adisin650@gmail.com>
Date: Wed, 5 Aug 2026 20:52:37 -0700
Subject: [PATCH 7/9] Normalize the missing metadata etag to None

The Dapr SDK reports a missing state etag as an empty string, so the create
path still satisfied 'metadata_etag is not None' and asked for first_write,
which is the guarantee the previous commit said it would stop claiming.

Normalize response.etag to None, make the fake report the empty string the way
the SDK does, and assert that creation sends no etag and leaves concurrency
unspecified while the following update is etag guarded. Also drop the timing
bound and the every-later-write claim from the comment.
---
 src/agents/extensions/memory/dapr_session.py | 15 +++---
 tests/extensions/memory/test_dapr_session.py | 51 +++++++++++++++++++-
 2 files changed, 57 insertions(+), 9 deletions(-)

diff --git a/src/agents/extensions/memory/dapr_session.py b/src/agents/extensions/memory/dapr_session.py
index 9549304c6c..f9462fd465 100644
--- a/src/agents/extensions/memory/dapr_session.py
+++ b/src/agents/extensions/memory/dapr_session.py
@@ -192,14 +192,16 @@ async def _read_created_at(self) -> tuple[str | None, str | None]:
         """Return the stored creation timestamp and the metadata etag backing it.
 
         The timestamp is None when it is missing or unreadable. The etag is returned so the
-        caller can write the metadata conditionally against the same revision it read.
+        caller can write the metadata conditionally against the same revision it read. The
+        Dapr SDK reports a missing etag as an empty string, so it is normalized to None and
+        callers can test for a real etag rather than for a particular empty representation.
         """
         response = await self._dapr_client.get_state(
             store_name=self._state_store_name,
             key=self._metadata_key,
             state_metadata=self._get_read_metadata(),
         )
-        etag = response.etag
+        etag = response.etag or None
         data = response.data
         if not data:
             return None, etag
@@ -394,13 +396,10 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
             # would let a later append overwrite created_at with its own now, so the save is
             # guarded by the etag that backed the value that was read.
             #
-            # The guard only applies once metadata exists. Dapr documents a write without an
-            # etag as last-write-wins even when first-write concurrency is requested, so
+            # The guard only applies once a real etag was read. Dapr documents a write without
+            # an etag as last-write-wins even when first-write concurrency is requested, so
             # asking for it on the create is not a race guarantee and is left off rather than
-            # implying one. Two writers creating the key concurrently can therefore both
-            # succeed, and the loser's created_at wins by a fraction of a second. Every write
-            # after that is etag guarded, which is the case that actually matters, since it is
-            # what stops a later append from resetting an established created_at.
+            # implying one. Concurrent etag-less creation is therefore last-write-wins.
             #
             # The messages key is already committed once we reach here, so raising would tell
             # the caller the append failed while their items are in the session, and the
diff --git a/tests/extensions/memory/test_dapr_session.py b/tests/extensions/memory/test_dapr_session.py
index d3c33c1a02..ae8b2c816f 100644
--- a/tests/extensions/memory/test_dapr_session.py
+++ b/tests/extensions/memory/test_dapr_session.py
@@ -40,7 +40,8 @@ async def get_state(
         """Get state from in-memory store."""
         response = Mock()
         response.data = self._state.get(key, b"")
-        response.etag = self._etags.get(key)
+        # The Dapr SDK reports a missing etag as an empty string rather than None.
+        response.etag = self._etags.get(key, "")
         return response
 
     async def save_state(
@@ -1351,6 +1352,54 @@ async def test_add_items_preserves_created_at_metadata(
         await session.close()
 
 
+async def test_metadata_creation_does_not_request_first_write(
+    fake_dapr_client: FakeDaprClient, monkeypatch: pytest.MonkeyPatch
+):
+    """Creating the metadata key must not ask for first-write concurrency.
+
+    Dapr treats a write with no etag as last-write-wins even when first-write is requested,
+    so asking for it on the create would imply a guarantee the store does not provide. The
+    SDK reports a missing etag as an empty string, so this also pins that the empty value is
+    not mistaken for a real one.
+    """
+    session = await _create_test_session(fake_dapr_client, "metadata_create_concurrency")
+
+    try:
+        real_save = fake_dapr_client.save_state
+        seen: list[tuple[str | None, Any]] = []
+
+        async def record_metadata_saves(
+            store_name: str,
+            key: str,
+            value: str | bytes,
+            **kwargs: Any,
+        ) -> None:
+            if key == session._metadata_key:
+                seen.append(
+                    (kwargs.get("etag"), getattr(kwargs.get("options"), "concurrency", None))
+                )
+            await real_save(store_name, key, value, **kwargs)
+
+        monkeypatch.setattr(fake_dapr_client, "save_state", record_metadata_saves)
+
+        await session.add_items([{"role": "user", "content": "first"}])
+        assert len(seen) == 1
+        create_etag, create_concurrency = seen[0]
+        # No real etag existed, so no etag is sent and concurrency is left unspecified
+        # rather than first-write, which Dapr would ignore here anyway.
+        assert create_etag is None
+        assert getattr(create_concurrency, "name", None) == "unspecified"
+
+        await session.add_items([{"role": "user", "content": "second"}])
+        assert len(seen) == 2
+        update_etag, update_concurrency = seen[1]
+        # Metadata now exists, so the update is guarded by the etag that backed it.
+        assert update_etag is not None
+        assert getattr(update_concurrency, "name", None) == "first_write"
+    finally:
+        await session.close()
+
+
 async def test_stale_metadata_etag_retries_and_keeps_created_at(
     fake_dapr_client: FakeDaprClient, monkeypatch: pytest.MonkeyPatch
 ):

From 16d0e241b04b6fe992d784033caaebed9681cb5d Mon Sep 17 00:00:00 2001
From: Aditya Singh <adisin650@gmail.com>
Date: Thu, 6 Aug 2026 18:15:46 -0700
Subject: [PATCH 8/9] Document the stale-read window under eventual consistency

---
 src/agents/extensions/memory/dapr_session.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/src/agents/extensions/memory/dapr_session.py b/src/agents/extensions/memory/dapr_session.py
index f9462fd465..7dd3d02a40 100644
--- a/src/agents/extensions/memory/dapr_session.py
+++ b/src/agents/extensions/memory/dapr_session.py
@@ -93,7 +93,12 @@ def __init__(
                 the underlying state store implementation. Defaults to None.
             consistency (ConsistencyLevel, optional): Consistency level for state operations.
                 Use DAPR_CONSISTENCY_EVENTUAL or DAPR_CONSISTENCY_STRONG constants.
-                Defaults to DAPR_CONSISTENCY_EVENTUAL.
+                Defaults to DAPR_CONSISTENCY_EVENTUAL. Reads under the eventual level may
+                not observe a write this session just made, so an append that closely
+                follows another one can read back a stale conversation or a stale
+                `created_at`. Use DAPR_CONSISTENCY_STRONG when a session is appended to from
+                more than one place, or in quick succession, and the stored history and
+                timestamps have to be exact.
             session_settings (SessionSettings | None): Session configuration settings including
                 default limit for retrieving items. If None, uses default SessionSettings().
         """

From 8d0e13636331bb05bbaf322547851b8857c0dd57 Mon Sep 17 00:00:00 2001
From: Aditya Singh <adisin650@gmail.com>
Date: Fri, 7 Aug 2026 05:06:07 -0700
Subject: [PATCH 9/9] Drop the consistency read-guarantee claim and cover the
 warning policies

The async client builds GetStateRequest with state_metadata passed as
request metadata and never sets the consistency field, so the strong
level does not deliver the read guarantee the docstring described.

Expand the metadata give-up warning coverage to inspect the whole
LogRecord under model-redacted, tool-redacted and fully diagnostic
modes, and assert in each that the committed items are still there.
---
 src/agents/extensions/memory/dapr_session.py |  7 +-
 tests/extensions/memory/test_dapr_session.py | 86 ++++++++++++++++++++
 2 files changed, 87 insertions(+), 6 deletions(-)

diff --git a/src/agents/extensions/memory/dapr_session.py b/src/agents/extensions/memory/dapr_session.py
index 7dd3d02a40..f9462fd465 100644
--- a/src/agents/extensions/memory/dapr_session.py
+++ b/src/agents/extensions/memory/dapr_session.py
@@ -93,12 +93,7 @@ def __init__(
                 the underlying state store implementation. Defaults to None.
             consistency (ConsistencyLevel, optional): Consistency level for state operations.
                 Use DAPR_CONSISTENCY_EVENTUAL or DAPR_CONSISTENCY_STRONG constants.
-                Defaults to DAPR_CONSISTENCY_EVENTUAL. Reads under the eventual level may
-                not observe a write this session just made, so an append that closely
-                follows another one can read back a stale conversation or a stale
-                `created_at`. Use DAPR_CONSISTENCY_STRONG when a session is appended to from
-                more than one place, or in quick succession, and the stored history and
-                timestamps have to be exact.
+                Defaults to DAPR_CONSISTENCY_EVENTUAL.
             session_settings (SessionSettings | None): Session configuration settings including
                 default limit for retrieving items. If None, uses default SessionSettings().
         """
diff --git a/tests/extensions/memory/test_dapr_session.py b/tests/extensions/memory/test_dapr_session.py
index ae8b2c816f..5458d39128 100644
--- a/tests/extensions/memory/test_dapr_session.py
+++ b/tests/extensions/memory/test_dapr_session.py
@@ -1502,3 +1502,89 @@ async def fail_only_metadata_saves(
             assert "etag mismatch" not in record.getMessage()
     finally:
         await session.close()
+
+
+@pytest.mark.parametrize(
+    ("dont_log_model_data", "dont_log_tool_data", "redacted"),
+    [
+        (True, False, True),
+        (False, True, True),
+        (False, False, False),
+    ],
+    ids=["model-redacted", "tool-redacted", "fully-diagnostic"],
+)
+async def test_metadata_write_warning_respects_data_policies(
+    fake_dapr_client: FakeDaprClient,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: Any,
+    dont_log_model_data: bool,
+    dont_log_tool_data: bool,
+    redacted: bool,
+):
+    """The give-up warning must obey both data policies, and never lose the committed items.
+
+    `log_model_and_tool_action_warning` redacts when either policy is on, so only the mode with
+    both off may carry the session id or the provider error. This inspects the whole LogRecord
+    rather than just the rendered message, because the session id travels in `extra` and the
+    exception travels in `exc_info`, neither of which shows up in `getMessage()`.
+    """
+    import logging
+
+    import agents._debug as _debug
+
+    session_id = f"metadata_policy_{dont_log_model_data}_{dont_log_tool_data}"
+    session = await _create_test_session(fake_dapr_client, session_id)
+
+    try:
+        monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", dont_log_model_data)
+        monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", dont_log_tool_data)
+
+        real_save = fake_dapr_client.save_state
+        error_text = "etag mismatch"
+
+        async def fail_only_metadata_saves(
+            store_name: str,
+            key: str,
+            value: str | bytes,
+            **kwargs: Any,
+        ) -> None:
+            if key == session._metadata_key:
+                raise RuntimeError(error_text)
+            await real_save(store_name, key, value, **kwargs)
+
+        monkeypatch.setattr(fake_dapr_client, "save_state", fail_only_metadata_saves)
+        monkeypatch.setattr(session, "_calculate_retry_delay", lambda attempt: 0.0)
+
+        with caplog.at_level(logging.WARNING):
+            await session.add_items([{"role": "user", "content": "kept"}])
+
+        # The append landed regardless of how the failure was logged.
+        items = await session.get_items()
+        assert [item.get("content") for item in items] == ["kept"]
+
+        records = [record for record in caplog.records if "could not update" in record.getMessage()]
+        assert len(records) == 1
+        record = records[0]
+        rendered = logging.Formatter().format(record)
+
+        if redacted:
+            # Redacted form is the bare message with no exception and no diagnostic context.
+            assert record.msg == "%s"
+            assert record.exc_info is None
+            assert record.exc_text is None
+            assert not hasattr(record, "openai_agents_diagnostic_context")
+            assert session_id not in rendered
+            assert error_text not in rendered
+            assert all(
+                session_id not in str(value) and error_text not in str(value)
+                for value in record.__dict__.values()
+            )
+        else:
+            # Diagnostic form carries the exception and the session id, by design.
+            assert record.msg == "%s: %s"
+            assert record.exc_info is not None
+            assert isinstance(record.exc_info[1], RuntimeError)
+            assert error_text in rendered
+            assert record.openai_agents_diagnostic_context == {"session_id": session_id}
+    finally:
+        await session.close()