fix(sessions): guard Dapr metadata writes with ETag + first-write

The metadata read-modify-write was unconditional, so two writers (or one
writer with a stale eventual read) could clobber an existing created_at.
Apply the same first-write concurrency + retry-on-conflict pattern used
for the messages key so the field is preserved across concurrent writes.

Author: adityasingh2400

src/agents/extensions/memory/dapr_session.py

@@ -329,33 +329,48 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
                     raise
 
             # Update metadata, preserving created_at across subsequent writes.
+            # Use first-write concurrency with the read ETag so a concurrent write
+            # that already established `created_at` can't be clobbered by a stale
+            # read that saw no metadata.
             now = str(int(time.time()))
-            created_at = now
-            try:
+            meta_attempt = 0
+            while True:
+                meta_attempt += 1
                 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(),
                 )
+                created_at = now
                 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": created_at,
-                "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(),
-            )
+                    try:
+                        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 metadata — start fresh with current timestamp.
+                        pass
+                metadata = {
+                    "session_id": self.session_id,
+                    "created_at": created_at,
+                    "updated_at": now,
+                }
+                meta_etag = getattr(existing_meta_response, "etag", None) or None
+                try:
+                    await self._dapr_client.save_state(
+                        store_name=self._state_store_name,
+                        key=self._metadata_key,
+                        value=json.dumps(metadata),
+                        etag=meta_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, meta_attempt)
+                    if should_retry:
+                        continue
+                    raise
 
     async def pop_item(self) -> TResponseInputItem | None:
         """Remove and return the most recent item from the session.