fix: enforce per-task push notification config limit

docs/content/dev/configuration.md

@@ -65,6 +65,15 @@ a2a.request-context.populate-referred-tasks=true
 
 When enabled, task IDs referenced in incoming messages are looked up in the `TaskStore` and made available to the `AgentExecutor` via `RequestContext.getRelatedTasks()`. This is useful for multi-task conversations where the agent needs access to state from related tasks. Enabled by default; set to `false` to avoid extra `TaskStore` lookups when not needed.
 
+### Push Notification Config Store
+
+```properties
+# Maximum push notification configs per task (default: 100)
+a2a.push-notification-config.max-per-task=100
+```
+
+Limits the number of distinct push notification configurations a single task may register. Each config consumes memory and can trigger an outbound HTTP request on every task event. Re-registering an existing config ID (updating it) does not count against the limit. Both `InMemoryPushNotificationConfigStore` and `JpaDatabasePushNotificationConfigStore` enforce this limit, throwing `InvalidParamsError` when exceeded.
+
 ### Tuning Guidelines
 
 - **Streaming Performance**: The executor handles streaming subscriptions. Too few threads can cause timeouts under concurrent load.

extras/push-notification-config-store-database-jpa/src/main/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStore.java

@@ -7,12 +7,15 @@
 import jakarta.annotation.Priority;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.enterprise.inject.Alternative;
+import jakarta.inject.Inject;
 import jakarta.persistence.EntityManager;
 import jakarta.persistence.PersistenceContext;
 import jakarta.transaction.Transactional;
 
 import org.a2aproject.sdk.jsonrpc.common.json.JsonProcessingException;
+import org.a2aproject.sdk.server.config.A2AConfigProvider;
 import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore;
+import org.a2aproject.sdk.spec.InvalidParamsError;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult;
 import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
@@ -34,6 +37,9 @@ public class JpaDatabasePushNotificationConfigStore implements PushNotificationC
     @PersistenceContext(unitName = "a2a-java")
     EntityManager em;
 
+    @Inject
+    A2AConfigProvider config;
+
     @Transactional
     @Override
     public TaskPushNotificationConfig setInfo(TaskPushNotificationConfig notificationConfig) {
@@ -66,6 +72,17 @@ public TaskPushNotificationConfig setInfo(TaskPushNotificationConfig notificatio
                 LOGGER.debug("Updated existing PushNotificationConfig for Task '{}' with ID: {}",
                         taskId, notificationConfig.id());
             } else {
+                // Enforce the per-task limit (configurable via
+                // a2a.push-notification-config.max-per-task); only genuinely new
+                // configs count against it.
+                int maxPerTask = PushNotificationConfigStore.maxPushConfigsPerTask(config);
+                Long existingCount = em.createQuery(
+                        "SELECT COUNT(c) FROM JpaPushNotificationConfig c WHERE c.id.taskId = :taskId",
+                        Long.class).setParameter("taskId", taskId).getSingleResult();
+                if (existingCount >= maxPerTask) {
+                    throw new InvalidParamsError("Too many push notification configs for task " + taskId
+                            + " (max " + maxPerTask + ")");
+                }
                 // Create new entity
                 JpaPushNotificationConfig jpaConfig = JpaPushNotificationConfig.createFromConfig(taskId, notificationConfig, resolvedVersion);
                 em.persist(jpaConfig);

extras/push-notification-config-store-database-jpa/src/test/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStoreIntegrationTest.java

@@ -28,6 +28,7 @@
 import org.a2aproject.sdk.spec.AgentCard;
 import org.a2aproject.sdk.spec.DeleteTaskPushNotificationConfigParams;
 import org.a2aproject.sdk.spec.GetTaskPushNotificationConfigParams;
+import org.a2aproject.sdk.spec.InvalidParamsError;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult;
 import org.a2aproject.sdk.spec.Message;
@@ -581,4 +582,29 @@ private void createSamples(String taskId, int size) {
             }
         }
     }
+
+    @Test
+    @Transactional
+    public void testPushConfigLimitEnforced() {
+        String taskId = "task_limit_" + System.currentTimeMillis();
+        int limit = org.a2aproject.sdk.server.tasks.PushNotificationConfigStore.DEFAULT_MAX_PUSH_CONFIGS_PER_TASK;
+
+        // Register configs up to the limit.
+        for (int i = 0; i < limit; i++) {
+            TaskPushNotificationConfig config = createSamplePushConfig(
+                    "http://limit" + i + ".com/callback", "limit-cfg" + i, "token" + i);
+            pushNotificationConfigStore.setInfo(TaskPushNotificationConfig.builder(config).taskId(taskId).build());
+        }
+
+        // The (limit+1)-th distinct config must be rejected.
+        TaskPushNotificationConfig extra = createSamplePushConfig(
+                "http://extra.com/callback", "limit-extra", "token-extra");
+        assertThrows(InvalidParamsError.class, () ->
+                pushNotificationConfigStore.setInfo(TaskPushNotificationConfig.builder(extra).taskId(taskId).build()));
+
+        // Re-registering an existing config ID at the limit is still allowed.
+        TaskPushNotificationConfig existing = createSamplePushConfig(
+                "http://limit0.com/callback-updated", "limit-cfg0", "token-updated");
+        pushNotificationConfigStore.setInfo(TaskPushNotificationConfig.builder(existing).taskId(taskId).build());
+    }
 }

server-common/src/main/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStore.java

@@ -10,6 +10,8 @@
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 
+import org.a2aproject.sdk.server.config.A2AConfigProvider;
+import org.a2aproject.sdk.spec.InvalidParamsError;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult;
 import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
@@ -24,9 +26,20 @@
 @ApplicationScoped
 public class InMemoryPushNotificationConfigStore implements PushNotificationConfigStore {
 
+    /**
+     * Default maximum number of push notification configs allowed per task.
+     * Prevents a single task from accumulating an unbounded list of configs
+     * (each config consumes memory and can trigger outbound HTTP requests).
+     * Overridable via {@code a2a.push-notification-config.max-per-task}.
+     */
+    public static final int MAX_PUSH_CONFIGS_PER_TASK = PushNotificationConfigStore.DEFAULT_MAX_PUSH_CONFIGS_PER_TASK;
+
     private final Map<String, List<TaskPushNotificationConfig>> pushNotificationInfos = Collections.synchronizedMap(new HashMap<>());
     private final Map<String, String> protocolVersions = Collections.synchronizedMap(new HashMap<>());
 
+    @Inject
+    @Nullable A2AConfigProvider configProvider;
+
     @Inject
     public InMemoryPushNotificationConfigStore() {
     }
@@ -41,14 +54,18 @@ public TaskPushNotificationConfig setInfo(TaskPushNotificationConfig notificatio
         }
         notificationConfig = builder.build();
 
-        Iterator<TaskPushNotificationConfig> notificationConfigIterator = notificationConfigList.iterator();
-        while (notificationConfigIterator.hasNext()) {
-            TaskPushNotificationConfig config = notificationConfigIterator.next();
-            if (config.id() != null  && config.id().equals(notificationConfig.id())) {
-                notificationConfigIterator.remove();
-                break;
-            }
+        // Enforce the per-task limit and remove any existing config with the same
+        // ID in a single pass. Re-registering/updating an already-registered config
+        // ID is allowed; only genuinely new configs count against the limit.
+        String configId = notificationConfig.id();
+        boolean isExistingConfig = notificationConfigList.removeIf(
+                existing -> existing.id() != null && existing.id().equals(configId));
+        int maxPerTask = PushNotificationConfigStore.maxPushConfigsPerTask(configProvider);
+        if (!isExistingConfig && notificationConfigList.size() >= maxPerTask) {
+            throw new InvalidParamsError("Too many push notification configs for task " + taskId
+                    + " (max " + maxPerTask + ")");
         }
+
         notificationConfigList.add(notificationConfig);
         pushNotificationInfos.put(taskId, notificationConfigList);
         return notificationConfig;

server-common/src/main/java/org/a2aproject/sdk/server/tasks/PushNotificationConfigStore.java

@@ -1,5 +1,6 @@
 package org.a2aproject.sdk.server.tasks;
 
+import org.a2aproject.sdk.server.config.A2AConfigProvider;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult;
 import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
@@ -21,6 +22,13 @@
  *   <li>Used for retrieval and deletion of specific configurations</li>
  * </ul>
  *
+ * <h2>Per-task Limit</h2>
+ * Implementations MUST reject registering more than
+ * {@code a2a.push-notification-config.max-per-task} (default 100) distinct configs for
+ * a single task, throwing {@code InvalidParamsError}. Re-registering an already-registered
+ * config ID does not count against the limit. See
+ * {@link #maxPushConfigsPerTask(A2AConfigProvider)}.
+ *
  * <h2>Pagination Support</h2>
  * {@link #getInfo(ListTaskPushNotificationConfigsParams)} supports pagination for tasks
  * with many push notification configurations:
@@ -70,6 +78,13 @@
  */
 public interface PushNotificationConfigStore {
 
+    /**
+     * Default maximum number of push notification configs allowed per task,
+     * used when {@code a2a.push-notification-config.max-per-task} is not
+     * configured. See {@code META-INF/a2a-defaults.properties}.
+     */
+    int DEFAULT_MAX_PUSH_CONFIGS_PER_TASK = 100;
+
     /**
      * Sets or updates the push notification configuration for a task.
      * <p>
@@ -105,6 +120,33 @@ static String resolveProtocolVersion(@Nullable String protocolVersion) {
         return protocolVersion != null ? protocolVersion : org.a2aproject.sdk.spec.AgentInterface.CURRENT_PROTOCOL_VERSION;
     }
 
+    /**
+     * Maximum number of push notification configs allowed per task.
+     *
+     * <p>Reads the {@code a2a.push-notification-config.max-per-task} configuration value;
+     * a {@code null} provider, or a missing, non-numeric, or non-positive value, falls
+     * back to 100. See {@code META-INF/a2a-defaults.properties}.</p>
+     *
+     * @param config the configuration provider, or {@code null} when the store is used
+     *               without CDI injection (e.g., in tests)
+     * @return the per-task limit (always positive)
+     */
+    static int maxPushConfigsPerTask(@Nullable A2AConfigProvider config) {
+        if (config == null) {
+            return DEFAULT_MAX_PUSH_CONFIGS_PER_TASK;
+        }
+        return config.getOptionalValue("a2a.push-notification-config.max-per-task")
+                .map(value -> {
+                    try {
+                        int parsed = Integer.parseInt(value.trim());
+                        return parsed > 0 ? parsed : DEFAULT_MAX_PUSH_CONFIGS_PER_TASK;
+                    } catch (NumberFormatException e) {
+                        return DEFAULT_MAX_PUSH_CONFIGS_PER_TASK;
+                    }
+                })
+                .orElse(DEFAULT_MAX_PUSH_CONFIGS_PER_TASK);
+    }
+
     /**
      * Retrieves push notification configurations for a task with pagination support.
      * <p>

server-common/src/main/resources/META-INF/a2a-defaults.properties

@@ -32,3 +32,9 @@ a2a.executor.queue-capacity=100
 # When true, referenced task IDs in messages are resolved from the TaskStore
 # and made available via RequestContext.getRelatedTasks()
 a2a.request-context.populate-referred-tasks=true
+
+# PushNotificationConfigStore - per-task configuration limit
+# Maximum number of push notification configs a single task may register.
+# Prevents a task from accumulating an unbounded list of configs (each config
+# consumes memory and can trigger outbound HTTP requests).
+a2a.push-notification-config.max-per-task=100

server-common/src/test/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStoreTest.java

@@ -5,6 +5,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.never;
@@ -17,6 +18,7 @@
 import org.a2aproject.sdk.client.http.A2AHttpResponse;
 import org.a2aproject.sdk.common.A2AHeaders;
 import org.a2aproject.sdk.spec.AgentInterface;
+import org.a2aproject.sdk.spec.InvalidParamsError;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult;
 import org.a2aproject.sdk.spec.Task;
@@ -659,4 +661,55 @@ public void testPaginationFullIteration() {
         assertEquals(3, pageCount, "Should have exactly 3 pages (3+3+1)");
     }
 
+    @Test
+    public void testSetInfoAtLimitExactlyAllowed() {
+        String taskId = "task_limit_exact";
+        for (int i = 0; i < InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK; i++) {
+            configStore.setInfo(createSamplePushConfig(taskId,
+                    "http://url" + i + ".com/callback", "cfg" + i, null));
+        }
+
+        ListTaskPushNotificationConfigsResult result = configStore.getInfo(new ListTaskPushNotificationConfigsParams(taskId));
+        assertEquals(InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK, result.configs().size());
+    }
+
+    @Test
+    public void testSetInfoRejectsExceedingPerTaskLimit() {
+        String taskId = "task_limit_exceed";
+        for (int i = 0; i < InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK; i++) {
+            configStore.setInfo(createSamplePushConfig(taskId,
+                    "http://url" + i + ".com/callback", "cfg" + i, null));
+        }
+
+        // The (MAX+1)-th distinct config for the same task must be rejected
+        TaskPushNotificationConfig overflow = createSamplePushConfig(taskId,
+                "http://url-overflow.com/callback", "cfg-overflow", null);
+        assertThrows(InvalidParamsError.class, () -> configStore.setInfo(overflow));
+
+        // The store is unchanged
+        ListTaskPushNotificationConfigsResult result = configStore.getInfo(new ListTaskPushNotificationConfigsParams(taskId));
+        assertEquals(InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK, result.configs().size());
+        assertTrue(result.configs().stream().noneMatch(c -> "cfg-overflow".equals(c.id())));
+    }
+
+    @Test
+    public void testSetInfoUpdateExistingConfigAtLimitAllowed() {
+        String taskId = "task_limit_update";
+        for (int i = 0; i < InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK; i++) {
+            configStore.setInfo(createSamplePushConfig(taskId,
+                    "http://url" + i + ".com/callback", "cfg" + i, null));
+        }
+
+        // Updating an existing config at the limit must still be allowed
+        TaskPushNotificationConfig updated = createSamplePushConfig(taskId,
+                "http://url-updated.com/callback", "cfg0", "new-token");
+        TaskPushNotificationConfig result = configStore.setInfo(updated);
+
+        assertEquals("cfg0", result.id());
+        ListTaskPushNotificationConfigsResult configs = configStore.getInfo(new ListTaskPushNotificationConfigsParams(taskId));
+        assertEquals(InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK, configs.configs().size());
+        assertEquals("http://url-updated.com/callback",
+                configs.configs().stream().filter(c -> "cfg0".equals(c.id())).findFirst().orElseThrow().url());
+    }
+
 }