fix: harden event consumer and queue

[ez-lbz]

What changed

1. A plain Message event no longer terminates the stream

Problem: In EventConsumer.consumeAll(), every Message event was marked isFinalEvent = true, so the first message emitted by an agent closed the stream immediately. Per the A2A protocol (§3.1.6) the stream MUST terminate only when the task reaches a terminal state (completed, failed, canceled, rejected). An intermediate Message sent before the agent finishes (or a message-only response with follow-up events) incorrectly cut off the rest of the stream — a client could miss the task's terminal status update. Note the queue machinery already treated Message as non-final (MainQueue.isFinalEvent only considers Task/TaskStatusUpdateEvent), so the consumer was inconsistent with the queue.

Fix (server-common/src/main/java/org/a2aproject/sdk/server/events/EventConsumer.java):

• Removed the event instanceof Message → isFinalEvent = true branch. The stream now terminates only on a final TaskStatusUpdateEvent, a terminal Task, a QueueClosedEvent, an A2AError, or via the existing agent-completed grace period when no final event arrives.
• Documented the intent next to the final-event determination.

Fix (server-common/src/test/java/org/a2aproject/sdk/server/events/EventConsumerTest.java):

• Updated testConsumeMessageEvents: two messages are now both delivered (the stream stays open after the first message) and terminates when the queue is closed.

Behavior change: message-only streams no longer close on the first Message; they close when the task reaches a terminal state, or ~1.5s after the agent completes (existing agent-completed grace period) when no terminal event is emitted. Clients that stream messages with a task lifecycle see no change in event delivery order — only the stream-close trigger changes.

2. Buffer-flush delay is configurable instead of hardcoded

Problem: EventConsumer blocked its polling thread with a hardcoded Thread.sleep(150) (BUFFER_FLUSH_DELAY_MS) before completing a stream. The value was not tunable for transports where the flush needs differ.

Fix (EventConsumer.java):

• The flush delay is now read from the a2a.eventconsumer.bufferFlushDelayMs system property (default 150 ms, clamped to >= 0; invalid values fall back to the default). Semantics are preserved: the delay still runs only once per stream, after the final event is sent, to let the SSE write callback fire before tube.complete().
• Added a package-private bufferFlushDelayMs() accessor and tests (testBufferFlushDelayMsDefaultsTo150, testBufferFlushDelayMsReadsConfiguredValue, testBufferFlushDelayMsRejectsInvalidValues).

Behavior change: none unless the system property is set.

3. ChildQueue is bounded by the parent queue size

Problem: ChildQueue built its LinkedBlockingDeque without a capacity, so queue.offer() always returned true and a slow subscriber could grow the deque without limit (the "queue is full" → immediate-close path was dead code).

Fix (server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java):

• ChildQueue now constructs its deque with the parent's queueSize (new LinkedBlockingDeque<>(parent.getQueueSize())), restoring the intended overflow → immediate-close behavior for slow consumers.

Fix (server-common/src/test/java/org/a2aproject/sdk/server/events/EventQueueTest.java):

• testChildQueueIsBoundedByParentQueueSize verifies the child deque's remainingCapacity equals the configured queue size.

4. Semaphore permit is released when MainEventBus.submit fails

Problem: MainQueue.enqueueItem acquires a semaphore permit, then calls mainEventBus.submit(...). If submit throws (e.g. interrupted), the permit was never released — the normal release only happens in MainEventBusProcessor.processEvent's finally, which never runs for an unsubmitted event. Repeated failures would leak all permits and permanently block event processing for the task.

Fix (EventQueue.java):

• Wrapped mainEventBus.submit(...) in try/catch: on a RuntimeException the acquired permit is released before rethrowing. The success path is unchanged (release still happens once in MainEventBusProcessor).

Fix (EventQueueTest.java):

• testSemaphorePermitReleasedWhenSubmitFails mocks MainEventBus.submit to throw and asserts mainQueue.size() returns to 0 (no leaked permit).

Testing

• mvn -pl server-common test — 452 tests run, 0 failures, 0 errors, 0 skipped (BUILD SUCCESS), including 5 new/updated regression tests.
• mvn -pl transport/jsonrpc,transport/grpc,transport/rest test — 124 tests run, 0 failures, 1 skipped (BUILD SUCCESS); the streaming transports that consume the real EventConsumer are unaffected.