fix(sessions): preserve Dapr session created_at across writes

src/agents/extensions/memory/dapr_session.py

@@ -323,11 +323,26 @@ 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()))
+            created_at = now
+            try:
+                existing_meta_response = await self._dapr_client.get_state(
+                    store_name=self._state_store_name,
+                    key=self._metadata_key,
+                    state_metadata=self._get_read_metadata(),
+                )
+                if existing_meta_response.data:
+                    existing_meta = json.loads(existing_meta_response.data.decode("utf-8"))
+                    if isinstance(existing_meta, dict) and existing_meta.get("created_at"):
+                        created_at = str(existing_meta["created_at"])
+            except (json.JSONDecodeError, UnicodeDecodeError, AttributeError):
+                # Corrupt or missing metadata — start fresh with current timestamp.
+                pass
             metadata = {
                 "session_id": self.session_id,
-                "created_at": str(int(time.time())),
-                "updated_at": str(int(time.time())),
+                "created_at": created_at,
+                "updated_at": now,
             }
             await self._dapr_client.save_state(
                 store_name=self._state_store_name,
@@ -358,7 +373,6 @@ async def pop_item(self) -> TResponseInputItem | None:
                 last_item = messages.pop()
                 messages_json = json.dumps(messages, separators=(",", ":"))
                 etag = getattr(response, "etag", None) or None
-                etag = getattr(response, "etag", None) or None
                 try:
                     await self._dapr_client.save_state(
                         store_name=self._state_store_name,

tests/extensions/memory/test_dapr_session.py

@@ -413,6 +413,32 @@ async def test_add_empty_items_list(fake_dapr_client: FakeDaprClient):
         await session.close()
 
 
+async def test_metadata_preserves_created_at(fake_dapr_client: FakeDaprClient):
+    """add_items must preserve created_at across writes; only updated_at advances."""
+    session = await _create_test_session(fake_dapr_client)
+    try:
+        await session.add_items([{"role": "user", "content": "first"}])
+        first_meta_raw = fake_dapr_client._state[session._metadata_key].decode("utf-8")
+        first_meta = json.loads(first_meta_raw)
+        first_created = first_meta["created_at"]
+        first_updated = first_meta["updated_at"]
+
+        # Wait one second so timestamps are guaranteed to differ.
+        import time as _time
+
+        _time.sleep(1)
+
+        await session.add_items([{"role": "user", "content": "second"}])
+        second_meta = json.loads(fake_dapr_client._state[session._metadata_key].decode("utf-8"))
+
+        assert second_meta["created_at"] == first_created, (
+            "created_at must be preserved across add_items calls"
+        )
+        assert int(second_meta["updated_at"]) >= int(first_updated)
+    finally:
+        await session.close()
+
+
 async def test_unicode_content(fake_dapr_client: FakeDaprClient):
     """Test that session correctly stores and retrieves unicode/non-ASCII content."""
     session = await _create_test_session(fake_dapr_client)