fix: make get_info_for_dispatch abstract to prevent silent notification drops

src/a2a/server/tasks/push_notification_config_store.py

@@ -1,14 +1,9 @@
-import logging
-
 from abc import ABC, abstractmethod
 
 from a2a.server.context import ServerCallContext
 from a2a.types.a2a_pb2 import TaskPushNotificationConfig
 
 
-logger = logging.getLogger(__name__)
-
-
 class PushNotificationConfigStore(ABC):
     """Interface for storing and retrieving push notification configurations for tasks."""
 
@@ -34,39 +29,26 @@ async def get_info(
         context).
         """
 
+    @abstractmethod
     async def get_info_for_dispatch(
         self,
         task_id: str,
     ) -> list[TaskPushNotificationConfig]:
         """Retrieves all push notification configurations for a task, across all owners.
 
         This is the internal read path used by the push-notification
-        dispatch loop. Implementations SHOULD override this method to
-        return every configuration registered for task_id regardless of
-        which user registered it. Authorization already happened at
-        registration time and the dispatch path fires every registered
-        webhook for the task.
-
-        The default implementation falls back to calling get_info with
-        a synthetic empty ServerCallContext. This preserves 1.0
-        behavior for subclasses that have not implemented the override
-        but is INCORRECT for any deployment with multiple owners: the
-        empty context resolves to the empty-string owner partition and
-        returns no configs (silently dropping every notification). A
-        warning is logged on every call to flag the misconfiguration.
-        Custom subclasses MUST override this method to deliver
-        notifications correctly in multi-owner deployments.
+        dispatch loop. Implementations MUST return every configuration
+        registered for task_id regardless of which user registered it.
+        Authorization already happened at registration time and the
+        dispatch path fires every registered webhook for the task.
+
+        The previous non-abstract default fell back to ``get_info`` with a
+        synthetic empty ``ServerCallContext``, which resolves to the
+        empty-string owner partition and silently dropped every
+        notification in any deployment with multiple owners. Making this
+        method abstract forces every store implementation to provide the
+        cross-owner read path explicitly instead of failing silently.
         """
-        logger.warning(
-            '%s does not override '
-            'PushNotificationConfigStore.get_info_for_dispatch; falling back '
-            'to a context-less get_info call which silently drops '
-            'notifications in any deployment with multiple owners. Override '
-            'get_info_for_dispatch to return all configs for task_id across '
-            'every owner.',
-            type(self).__name__,
-        )
-        return await self.get_info(task_id, ServerCallContext())
 
     @abstractmethod
     async def delete_info(

tests/server/tasks/test_inmemory_push_notifications.py

@@ -572,3 +572,43 @@ async def test_cross_user_dispatch_alice_registers_bob_triggers(
 
 if __name__ == '__main__':
     unittest.main()
+
+
+class TestPushNotificationConfigStoreContract(unittest.TestCase):
+    """The dispatch read path must be implemented explicitly."""
+
+    def test_get_info_for_dispatch_is_abstract(self):
+        """A store that forgets get_info_for_dispatch cannot be instantiated.
+
+        Previously the base class silently fell back to an owner-scoped
+        get_info call that drops every notification in multi-owner
+        deployments; now the method is abstract so the failure is loud.
+        """
+        from a2a.server.tasks.push_notification_config_store import (
+            PushNotificationConfigStore,
+        )
+
+        class IncompleteStore(PushNotificationConfigStore):
+            async def set_info(self, task_id, notification_config, context):
+                pass
+
+            async def get_info(self, task_id, context):
+                return []
+
+            async def delete_info(self, task_id, context, config_id=None):
+                pass
+
+        with self.assertRaises(TypeError):
+            IncompleteStore()
+
+    def test_builtin_stores_implement_dispatch_read_path(self):
+        """Both shipped stores provide the cross-owner dispatch read path."""
+        from a2a.server.tasks.inmemory_push_notification_config_store import (
+            InMemoryPushNotificationConfigStore,
+        )
+
+        store = InMemoryPushNotificationConfigStore()
+        self.assertTrue(
+            hasattr(store, 'get_info_for_dispatch')
+            and callable(store.get_info_for_dispatch)
+        )