fix: Don't generate empty metadata change events in VertexTaskStore

src/a2a/contrib/tasks/vertex_task_store.py

@@ -109,7 +109,12 @@ def _get_status_details_change_event(
     def _get_metadata_change_event(
         self, previous_task: Task, task: Task, event_sequence_number: int
     ) -> vertexai_types.TaskEvent | None:
-        if task.metadata != previous_task.metadata:
+        # We generate metadata change events if the metadata was changed.
+        # We don't generate events if the metadata was changed from
+        # one empty value to another, e.g. {} to None.
+        if task.metadata != previous_task.metadata and (
+            task.metadata or previous_task.metadata
+        ):
             return vertexai_types.TaskEvent(
                 event_data=vertexai_types.TaskEventData(
                     metadata_change=vertexai_types.TaskMetadataChange(

tests/contrib/tasks/test_vertex_task_store.py

@@ -508,6 +508,53 @@ async def test_metadata_field_mapping(
     assert retrieved_none.metadata == {}
 
 
+@pytest.mark.asyncio
+async def test_metadata_empty_transitions(
+    vertex_store: VertexTaskStore,
+) -> None:
+    """Test that updating metadata between {} and None does not generate events."""
+    task_id = 'task-metadata-empty-test'
+
+    # Step 1: Create task with metadata={}
+    task = Task(
+        id=task_id,
+        context_id='session-meta-empty',
+        status=TaskStatus(state=TaskState.submitted),
+        kind='task',
+        metadata={},
+    )
+    await vertex_store.save(task)
+
+    full_name = f'{vertex_store._agent_engine_resource_id}/a2aTasks/{task_id}'
+
+    # Get initial event sequence number
+    stored_task_before = (
+        await vertex_store._client.aio.agent_engines.a2a_tasks.get(full_name)
+    )
+    initial_seq = stored_task_before.next_event_sequence_number
+
+    # Step 2: Update metadata to None
+    updated_task = task.model_copy(deep=True)
+    updated_task.metadata = None
+    await vertex_store.save(updated_task)
+
+    # Step 3: Update back to {}
+    task_back = updated_task.model_copy(deep=True)
+    task_back.metadata = {}
+    await vertex_store.save(task_back)
+
+    # Verify that retrieved task still has {} (due to mapping)
+    retrieved = await vertex_store.get(task_id)
+    assert retrieved is not None
+    assert retrieved.metadata == {}
+
+    # Verify that next_event_sequence_number did NOT increase (no events generated)
+    stored_task_after = (
+        await vertex_store._client.aio.agent_engines.a2a_tasks.get(full_name)
+    )
+    assert stored_task_after.next_event_sequence_number == initial_seq
+
+
 @pytest.mark.asyncio
 async def test_update_task_status_details(
     vertex_store: VertexTaskStore,