fix(server): start event bus processor for manual wiring

docs/content/dev/server.md

@@ -247,4 +247,51 @@ See [Backward Compatibility](compatibility) for multi-version modules, version r
 - **Quarkus** — Reference implementations are Quarkus-based (JSON-RPC, gRPC, REST)
 - **Jakarta EE** — [a2a-jakarta](https://github.com/wildfly-extras/a2a-jakarta) works with any Jakarta EE Web Profile runtime
 
+### Manual Wiring
+
+For example, when wiring the server components as Spring beans, expose the event processor as a bean and pass it to the request handler. The handler starts the processor when it is constructed, including when the processor was created manually:
+
+```java
+@Configuration
+class A2AServerConfiguration {
+
+    @Bean
+    MainEventBus mainEventBus() {
+        return new MainEventBus();
+    }
+
+    @Bean
+    MainEventBusProcessor mainEventBusProcessor(MainEventBus mainEventBus, TaskStore taskStore,
+                                                PushNotificationSender pushSender, QueueManager queueManager) {
+        return new MainEventBusProcessor(mainEventBus, taskStore, pushSender, queueManager);
+    }
+
+    @Bean(destroyMethod = "shutdown")
+    ExecutorService a2aEventConsumerExecutor() {
+        return Executors.newCachedThreadPool();
+    }
+
+    @Bean
+    RequestHandler requestHandler(AgentExecutor agentExecutor, TaskStore taskStore,
+                                  QueueManager queueManager, PushNotificationConfigStore pushConfigStore,
+                                  MainEventBusProcessor mainEventBusProcessor,
+                                  TaskAuthorizationProvider authorizationProvider,
+                                  @Qualifier("a2aInternalExecutor") Executor executor,
+                                  @Qualifier("a2aEventConsumerExecutor") Executor eventConsumerExecutor) {
+        return DefaultRequestHandler.builder()
+                .agentExecutor(agentExecutor)
+                .taskStore(taskStore)
+                .queueManager(queueManager)
+                .pushConfigStore(pushConfigStore)
+                .mainEventBusProcessor(mainEventBusProcessor)
+                .authorizationProvider(authorizationProvider)
+                .executor(executor)
+                .eventConsumerExecutor(eventConsumerExecutor)
+                .build();
+    }
+}
+```
+
+Register the remaining dependencies, including an `a2aInternalExecutor` bean and a `TaskAuthorizationProvider`, as Spring beans. Keep the cached event consumer executor separate from the agent executor: blocking event polling can exhaust a bounded pool under load.
+
 See [CONTRIBUTING_INTEGRATIONS.md](https://github.com/a2aproject/a2a-java/blob/main/CONTRIBUTING_INTEGRATIONS.md) to submit your own integration.

server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java

@@ -156,25 +156,25 @@ int pushNotificationChainCount() {
         return pushNotificationChains.size();
     }
 
+    /**
+     * Starts the background processor thread if it is not already running.
+     */
     @SuppressWarnings("NullAway.Init")
     @PostConstruct
-    void start() {
+    public synchronized void start() {
+        if (processorThread != null && processorThread.isAlive()) {
+            LOGGER.debug("MainEventBusProcessor already started");
+            return;
+        }
+        running = true;
         processorThread = new Thread(this, "MainEventBusProcessor");
         processorThread.setDaemon(true); // Allow JVM to exit even if this thread is running
         processorThread.start();
         LOGGER.info("MainEventBusProcessor started");
     }
 
-    /**
-     * No-op method to force CDI proxy resolution and ensure @PostConstruct has been called.
-     * Called by MainEventBusProcessorInitializer during application startup.
-     */
-    public void ensureStarted() {
-        // Method intentionally empty - just forces proxy resolution
-    }
-
     @PreDestroy
-    void stop() {
+    synchronized void stop() {
         LOGGER.info("MainEventBusProcessor stopping...");
         running = false;
         if (processorThread != null) {

server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java

@@ -27,14 +27,12 @@ public class MainEventBusProcessorInitializer {
 
     /**
      * Observes ApplicationScoped initialization to force eager creation of MainEventBusProcessor.
-     * The injection of MainEventBusProcessor in this bean triggers its creation, and calling
-     * ensureStarted() forces the CDI proxy to be resolved, which ensures @PostConstruct has been
+     * Calling start() forces the CDI proxy to be resolved, which ensures @PostConstruct has been
      * called and the background thread is running.
      */
     void onStart(@Observes @Initialized(ApplicationScoped.class) Object event) {
         if (processor != null) {
-            // Force proxy resolution to ensure @PostConstruct has been called
-            processor.ensureStarted();
+            processor.start();
             LOGGER.info("MainEventBusProcessor initialized and started");
         } else {
             LOGGER.error("MainEventBusProcessor is null - initialization failed!");

server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java

@@ -308,6 +308,10 @@ protected DefaultRequestHandler() {
         this.eventConsumerExecutor = null;
     }
 
+    /**
+     * Constructor used by CDI to create the request handler. Its signature may change between
+     * releases; application code should use {@link #builder()} to configure and create a handler.
+     */
     @Inject
     public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore,
                                  QueueManager queueManager, PushNotificationConfigStore pushConfigStore,
@@ -326,6 +330,7 @@ public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore,
         //  I am unsure about the correct scope.
         //  Also reworked to make a Supplier since otherwise the builder gets polluted with wrong tasks
         this.requestContextBuilder = () -> new SimpleRequestContextBuilder(taskStore, false, null);
+        this.mainEventBusProcessor.start();
     }
 
     @SuppressWarnings("NullAway.Init")

server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java

@@ -186,6 +186,51 @@ void testInitConfigReadsBlockingTimeouts() {
         assertEquals(7, handler.reconciliationTimeoutSeconds);
     }
 
+    @Test
+    void testBuilderStartsManuallyConstructedMainEventBusProcessor() throws Exception {
+        InMemoryTaskStore manualTaskStore = new InMemoryTaskStore();
+        PushNotificationConfigStore manualPushConfigStore = new InMemoryPushNotificationConfigStore();
+        MainEventBus manualMainEventBus = new MainEventBus();
+        InMemoryQueueManager manualQueueManager = new InMemoryQueueManager(manualTaskStore, manualMainEventBus);
+        MainEventBusProcessor manualProcessor = new MainEventBusProcessor(
+            manualMainEventBus, manualTaskStore, NOOP_PUSHNOTIFICATION_SENDER, manualQueueManager);
+
+        try {
+            AgentExecutor manualAgentExecutor = new AgentExecutor() {
+                @Override
+                public void execute(RequestContext context, AgentEmitter agentEmitter) {
+                    agentEmitter.sendMessage("manual lifecycle response");
+                }
+
+                @Override
+                public void cancel(RequestContext context, AgentEmitter agentEmitter) {
+                    throw new AssertionError("Cancel should not be invoked");
+                }
+            };
+            DefaultRequestHandler manualRequestHandler = DefaultRequestHandler.builder()
+                    .agentExecutor(manualAgentExecutor)
+                    .taskStore(manualTaskStore)
+                    .queueManager(manualQueueManager)
+                    .pushConfigStore(manualPushConfigStore)
+                    .mainEventBusProcessor(manualProcessor)
+                    .executor(internalExecutor)
+                    .eventConsumerExecutor(internalExecutor)
+                    .authorizationRequired(false)
+                    .build();
+            manualRequestHandler.agentCompletionTimeoutSeconds = 5;
+            manualRequestHandler.consumptionCompletionTimeoutSeconds = 2;
+            manualRequestHandler.reconciliationTimeoutSeconds = 1;
+
+            EventKind eventKind = manualRequestHandler.onMessageSend(
+                MessageSendParams.builder().message(MESSAGE).build(), NULL_CONTEXT);
+
+            assertInstanceOf(Message.class, eventKind);
+            assertEquals(Message.Role.ROLE_AGENT, ((Message) eventKind).role());
+        } finally {
+            EventQueueUtil.stop(manualProcessor);
+        }
+    }
+
     /**
      * Test 1: Non-streaming AUTH_REQUIRED returns immediately while agent continues.
      * Verifies: