feat: add TaskStreamLifecycleHook for stream lifecycle management

docs/content/dev/examples.md

@@ -144,6 +144,45 @@ The client expects an OpenTelemetry collector on port 5317. The easiest way is t
 
 For more information, see the [OpenTelemetry extras module](extras/opentelemetry).
 
+## Stream Lifecycle Hook
+
+This example demonstrates the `TaskStreamLifecycleHook` — a CDI-discoverable hook that observes stream lifecycle events and can close all active streams on demand.
+
+The server implements a hook that closes all subscriber streams when 3 clients connect to the same task. The client creates 3 subscribers sequentially, sending messages while the first two are active. When the third subscriber connects, the hook fires and all streams close gracefully.
+
+### Start the Server
+
+```bash
+cd examples/stream-lifecycle/server
+mvn quarkus:dev
+```
+
+### Run the Client
+
+```bash
+cd examples/stream-lifecycle/client
+mvn exec:java
+```
+
+The client logs each event received by each subscriber, showing events flowing to subscribers 1 and 2 before the hook triggers.
+
+#### Transport Protocol Selection
+
+```bash
+# JSON-RPC (default)
+mvn exec:java
+
+# gRPC
+mvn exec:java -Dquarkus.agentcard.protocol=GRPC
+
+# HTTP+JSON/REST
+mvn exec:java -Dquarkus.agentcard.protocol=HTTP+JSON
+```
+
+Select the same protocol on both server and client.
+
+For implementation details, see the [Stream Lifecycle Hook section](server#6-stream-lifecycle-hook-optional) in the Server Guide.
+
 ## More Examples
 
 - [a2a-samples repository](https://github.com/a2aproject/a2a-samples/tree/main/samples/java/agents) — Additional agent examples in Java and other languages

docs/content/dev/server.md

@@ -158,6 +158,71 @@ See [Configuration](configuration) for all config properties and tuning.
 
 See [Task Authorization](authorization) for per-user access control.
 
+## 6. Stream Lifecycle Hook (Optional)
+
+The `TaskStreamLifecycleHook` lets you observe and control streaming connections for a task. You are notified when clients subscribe, unsubscribe, or when events are distributed, and you can close all active streams on demand via the `StreamCloseHandle`.
+
+### Implementing a Hook
+
+Create a CDI bean that implements `TaskStreamLifecycleHook` and overrides the default no-op:
+
+```java
+@ApplicationScoped
+@Alternative
+@Priority(1)
+public class MyStreamHook implements TaskStreamLifecycleHook {
+
+    @Override
+    public void onSubscribe(String taskId, StreamCloseHandle handle) {
+        // Called when a client subscribes to a task's event stream
+    }
+
+    @Override
+    public void onUnsubscribe(String taskId, StreamCloseHandle handle) {
+        // Called when a client disconnects
+    }
+
+    @Override
+    public void onEvent(String taskId, Event event, StreamCloseHandle handle) {
+        // Called after an event is persisted and distributed to all subscribers
+    }
+}
+```
+
+### StreamCloseHandle
+
+Each callback receives a `StreamCloseHandle` with two methods:
+
+- **`closeStreams()`** — Gracefully closes all active subscriber streams for the task. The agent executor continues running and the MainQueue stays alive (for non-finalized tasks), so new clients can resubscribe.
+- **`getActiveSubscriberCount()`** — Returns the number of currently connected subscribers.
+
+### Example: Close Streams at a Subscriber Threshold
+
+```java
+@ApplicationScoped
+@Alternative
+@Priority(1)
+public class CloseStreamsHook implements TaskStreamLifecycleHook {
+
+    private static final int MAX_SUBSCRIBERS = 3;
+
+    @Override
+    public void onSubscribe(String taskId, StreamCloseHandle handle) {
+        if (handle.getActiveSubscriberCount() >= MAX_SUBSCRIBERS) {
+            handle.closeStreams();
+        }
+    }
+
+    @Override
+    public void onUnsubscribe(String taskId, StreamCloseHandle handle) { }
+
+    @Override
+    public void onEvent(String taskId, Event event, StreamCloseHandle handle) { }
+}
+```
+
+See the [`examples/stream-lifecycle`](https://github.com/a2aproject/a2a-java/tree/main/examples/stream-lifecycle) directory for a complete working example with server, client, and integration tests for all three transports.
+
 ## Backward Compatibility with v0.3
 
 See [Backward Compatibility](compatibility) for multi-version modules, version routing, and v0.3 client support.

examples/stream-lifecycle/README.md

@@ -0,0 +1,135 @@
+# Stream Lifecycle Hook Example
+
+This example demonstrates `TaskStreamLifecycleHook`, a CDI-discoverable hook that lets you observe and control streaming connections for a task. The server closes all active streams when 3 subscribers connect to the same task.
+
+For example, in order to save resources associated with streaming connections, you might want to close streams:
+
+* If no events are received for a Task in a given timeframe. This can help if you have a lot of Tasks taking a very long time to complete
+* Stopping rogue client opening too many subscriptions to the same Task
+
+## Prerequisites
+
+- Java 17 or higher
+- Maven
+
+## What It Does
+
+**Server** — An agent sends 20 progress messages (one every 500ms). A `CloseStreamsHook` monitors subscriber count and calls `StreamCloseHandle.closeStreams()` when 3 subscribers are connected, gracefully closing all streams.
+
+**Client** — Connects 3 subscribers sequentially:
+1. Subscriber 1 sends a message (creates the task, starts streaming)
+2. Subscriber 2 subscribes to the same task after 1.5 seconds
+3. Both subscribers receive progress messages for 2 seconds
+4. Subscriber 3 subscribes — the hook fires and closes all streams
+5. All 3 subscribers see their streams end gracefully
+
+## Run the Example
+
+### 1. Build the SDK
+
+From the repository root:
+
+```bash
+mvn clean install -DskipTests
+```
+
+### 2. Start the Server
+
+```bash
+cd examples/stream-lifecycle/server
+mvn quarkus:dev
+```
+
+### 3. Run the Client
+
+In a separate terminal:
+
+```bash
+cd examples/stream-lifecycle/client
+mvn exec:java
+```
+
+### Expected Output (Client)
+
+The exact event interleaving depends on timing, but you should see something like:
+
+```
+Resolved agent card: Stream Lifecycle Demo Agent
+[Sub-1] Sending message to create task...
+[Sub-1] StatusUpdate — TASK_STATE_WORKING
+
+=== Task created: <task-id> ===
+
+[Sub-1] ArtifactEvent — Progress update 1/20
+[Sub-1] ArtifactEvent — Progress update 2/20
+[Sub-1] ArtifactEvent — Progress update 3/20
+[Sub-2] Subscribing to task <task-id>...
+
+=== Subscribers 1 and 2 are active — receiving events... ===
+
+[Sub-2] TaskEvent — state: TASK_STATE_WORKING, id: <task-id>
+[Sub-1] ArtifactEvent — Progress update 4/20
+[Sub-2] ArtifactEvent — Progress update 4/20
+[Sub-1] ArtifactEvent — Progress update 5/20
+[Sub-2] ArtifactEvent — Progress update 5/20
+...
+[Sub-3] Subscribing to task <task-id> (will trigger stream close)...
+[Sub-1] Stream closed.
+[Sub-2] Stream closed.
+[Sub-3] Stream closed.
+
+=== All streams closed. ===
+```
+
+### Expected Output (Server)
+
+```
+[HOOK] Subscriber added for task <id>. Active subscribers: 1
+[AGENT] Starting execution for task <id>
+[AGENT] Sending: Progress update 1/20
+[HOOK] Event distributed for task <id>: Message (subscribers: 1)
+...
+[HOOK] Subscriber added for task <id>. Active subscribers: 2
+...
+[HOOK] Subscriber added for task <id>. Active subscribers: 3
+[HOOK] Subscriber count reached 3 for task <id> — closing all streams
+[HOOK] Subscriber removed for task <id>. Active subscribers: 2
+[HOOK] Subscriber removed for task <id>. Active subscribers: 1
+[HOOK] Subscriber removed for task <id>. Active subscribers: 0
+```
+
+## Transport Protocol Selection
+
+Set `quarkus.agentcard.protocol` on both server and client (must match). Available values:
+
+| Value | Transport |
+|-------|-----------|
+| `JSONRPC` | JSON-RPC 2.0 (default) |
+| `GRPC` | gRPC |
+| `HTTP+JSON` | HTTP+JSON/REST |
+
+```bash
+# Server — gRPC example
+mvn quarkus:dev -Dquarkus.agentcard.protocol=GRPC
+
+# Client — must use the same value
+mvn exec:java -Dquarkus.agentcard.protocol=GRPC
+```
+
+## Key Files
+
+| File | Description |
+|------|-------------|
+| `server/.../CloseStreamsHook.java` | `TaskStreamLifecycleHook` implementation — closes streams at 3 subscribers |
+| `server/.../AgentExecutorProducer.java` | Agent that sends 20 progress messages over 10 seconds |
+| `server/.../AgentCardProducer.java` | Agent card with streaming enabled |
+| `client/.../StreamLifecycleClient.java` | Client that creates 3 subscribers and logs events |
+
+## Integration Tests
+
+The server module includes `@QuarkusTest` integration tests that verify the hook behavior across all three transports:
+
+```bash
+cd examples/stream-lifecycle/server
+mvn test
+```

examples/stream-lifecycle/client/pom.xml

@@ -0,0 +1,58 @@
+<?xml version="1.0"?>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.a2aproject.sdk</groupId>
+        <artifactId>a2a-java-sdk-examples-stream-lifecycle-parent</artifactId>
+        <version>1.1.1.Final-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>a2a-java-sdk-examples-stream-lifecycle-client</artifactId>
+
+    <name>Java SDK A2A Examples - Stream Lifecycle Client</name>
+    <description>Client demonstrating TaskStreamLifecycleHook with multiple subscribers</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.a2aproject.sdk</groupId>
+            <artifactId>a2a-java-sdk-client</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.a2aproject.sdk</groupId>
+            <artifactId>a2a-java-sdk-jsonrpc-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.a2aproject.sdk</groupId>
+            <artifactId>a2a-java-sdk-client-transport-grpc</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-netty</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.a2aproject.sdk</groupId>
+            <artifactId>a2a-java-sdk-client-transport-rest</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>exec-maven-plugin</artifactId>
+                <version>3.6.3</version>
+                <configuration>
+                    <mainClass>org.a2aproject.sdk.examples.streamlifecycle.client.StreamLifecycleClient</mainClass>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>