fix(server): handle unwrapped legacy push notification configurations

src/a2a/server/tasks/database_push_notification_config_store.py

@@ -185,6 +185,7 @@ def _from_orm(
                 decrypted_payload = self._fernet.decrypt(payload)
                 return self._parse_config(
                     decrypted_payload.decode('utf-8'),
+                    model_instance.task_id,
                     model_instance.protocol_version,
                 )
             except Exception as e:
@@ -218,7 +219,9 @@ def _from_orm(
                 else payload
             )
             return self._parse_config(
-                payload_str, model_instance.protocol_version
+                payload_str,
+                model_instance.task_id,
+                model_instance.protocol_version,
             )
 
         except Exception as e:
@@ -341,20 +344,27 @@ async def delete_info(
                 )
 
     def _parse_config(
-        self, json_payload: str, protocol_version: str | None = None
+        self,
+        json_payload: str,
+        task_id: str | None = None,
+        protocol_version: str | None = None,
     ) -> TaskPushNotificationConfig:
         """Parses a JSON payload into a TaskPushNotificationConfig proto.
 
-        Uses protocol_version to decide between modern parsing and legacy conversion.
+        Args:
+            json_payload: The JSON payload to parse.
+            task_id: The unique identifier of the task. Only required for legacy
+                (0.3) protocol versions.
+            protocol_version: The protocol version used for serialization.
         """
         if protocol_version == '1.0':
             return Parse(json_payload, TaskPushNotificationConfig())
-
-        legacy_instance = (
-            types_v03.TaskPushNotificationConfig.model_validate_json(
-                json_payload
-            )
+        inner_config = types_v03.PushNotificationConfig.model_validate_json(
+            json_payload
         )
         return conversions.to_core_task_push_notification_config(
-            legacy_instance
+            types_v03.TaskPushNotificationConfig(
+                task_id=task_id or '',
+                push_notification_config=inner_config,
+            )
         )

tests/server/tasks/test_database_push_notification_config_store.py

@@ -220,10 +220,10 @@ async def test_set_and_get_info_multiple_configs(
 
     task_id = 'task-1'
     config1 = TaskPushNotificationConfig(
-        id='config-1', url='http://example.com/1'
+        id='config-1', task_id=task_id, url='http://example.com/1'
     )
     config2 = TaskPushNotificationConfig(
-        id='config-2', url='http://example.com/2'
+        id='config-2', task_id=task_id, url='http://example.com/2'
     )
 
     await db_store_parameterized.set_info(
@@ -738,16 +738,13 @@ async def test_get_0_3_push_notification_config_detailed(
     context_user = ServerCallContext(user=SampleUser(user_name=owner))
 
     # 1. Create a legacy PushNotificationConfig using v0.3 models
-    legacy_config = types_v03.TaskPushNotificationConfig(
-        task_id=task_id,
-        push_notification_config=types_v03.PushNotificationConfig(
-            id=config_id,
-            url='https://example.com/push',
-            token='legacy-token',
-            authentication=types_v03.PushNotificationAuthenticationInfo(
-                schemes=['bearer'],
-                credentials='legacy-creds',
-            ),
+    legacy_config = types_v03.PushNotificationConfig(
+        id=config_id,
+        url='https://example.com/push',
+        token='legacy-token',
+        authentication=types_v03.PushNotificationAuthenticationInfo(
+            schemes=['bearer'],
+            credentials='legacy-creds',
         ),
     )