fix(engine): keep notification streams open across task result states

workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java

@@ -180,7 +180,7 @@ private static String extractAgentUrl(AgentCard agentCard) {
     return "?";
   }
 
-  private static void onEvent(
+  static void onEvent(
       String agentName,
       ClientEvent event,
       List<ClientEvent> events,
@@ -210,7 +210,10 @@ private static void onEvent(
             e);
       }
     }
-    if (isTerminal(event)) {
+    // Notification-T is a channel-scoped subscription, not the lifetime of a
+    // Task-T task. Result statuses on its stream must not release the stream
+    // latch while heartbeats and later notifications may still arrive.
+    if (lastActivityNanos == null && isTerminal(event)) {
       log.info("[A2ARuntime] Terminal event for '{}': {}", agentName, describeTerminalEvent(event));
       done.countDown();
     }
@@ -581,13 +584,13 @@ private void awaitCompletion(
    *
    * <p>Subscription tasks stay {@code WORKING} for their whole lifetime and deliver events
    * (heartbeats included) without ever completing, so a finality-based wait would tear down a
-   * healthy subscription once {@code sendTimeoutSeconds} elapsed. The stream instead ends through
-   * the error/terminal-event latch: a server-side close, a transport failure, or {@link
+   * healthy subscription once {@code sendTimeoutSeconds} elapsed. Task status is not a subscription
+   * termination signal. The stream instead ends through a transport failure or {@link
    * dev.openan.workflow.engine.client.NotificationSubscription#close()}. The stream is only reaped
    * when no transport activity or decoded event has arrived for {@code sendTimeoutSeconds}.
    *
    * @param agentName target agent display name, for diagnostics only
-   * @param done latch released on terminal event or stream error
+   * @param done latch released on stream error or explicit local close
    * @param lastActivityNanos monotonic time of the most recent stream activity
    */
   void awaitNotificationActivity(

workflow-engine/src/test/java/dev/openan/workflow/engine/client/NotificationStreamTimeoutTest.java

@@ -19,12 +19,24 @@
 
 package dev.openan.workflow.engine.client;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import org.a2aproject.sdk.client.ClientEvent;
+import org.a2aproject.sdk.client.TaskEvent;
+import org.a2aproject.sdk.client.TaskUpdateEvent;
+import org.a2aproject.sdk.spec.Task;
+import org.a2aproject.sdk.spec.TaskState;
+import org.a2aproject.sdk.spec.TaskStatus;
+import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
 import org.junit.jupiter.api.Test;
 
 /**
@@ -123,7 +135,7 @@ void alreadyIdleStreamTimesOutWithoutAnotherFullWait() {
   }
 
   @Test
-  void terminalEventEndsNotificationStreamImmediately() throws Exception {
+  void explicitCloseEndsNotificationStreamImmediately() throws Exception {
     DefaultA2AJavaClientRuntime runtime = runtimeWithOneSecondIdleBudget();
     CountDownLatch done = new CountDownLatch(1);
     done.countDown();
@@ -133,6 +145,104 @@ void terminalEventEndsNotificationStreamImmediately() throws Exception {
 
     assertTrue(
         TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started) < 500,
-        "terminal stream should return immediately");
+        "explicitly closed stream should return immediately");
+  }
+
+  @Test
+  void taskResultStatesDoNotEndAnActiveNotificationStream() {
+    CountDownLatch notificationDone = new CountDownLatch(1);
+    AtomicLong lastActivityNanos = new AtomicLong(System.nanoTime() - TimeUnit.SECONDS.toNanos(2));
+    AtomicLong eventCount = new AtomicLong();
+    AtomicLong deliveredCount = new AtomicLong();
+    AtomicReference<ClientEvent> lastEvent = new AtomicReference<>();
+    List<ClientEvent> retainedEvents = new ArrayList<>();
+    for (TaskState state :
+        List.of(
+            TaskState.TASK_STATE_COMPLETED,
+            TaskState.TASK_STATE_FAILED,
+            TaskState.TASK_STATE_CANCELED,
+            TaskState.TASK_STATE_REJECTED,
+            TaskState.TASK_STATE_INPUT_REQUIRED,
+            TaskState.TASK_STATE_AUTH_REQUIRED)) {
+      TaskStatus status = new TaskStatus(state);
+      Task task =
+          Task.builder()
+              .id("task-1")
+              .contextId("notification-context")
+              .status(status)
+              .artifacts(List.of())
+              .history(List.of())
+              .metadata(Map.of())
+              .build();
+      DefaultA2AJavaClientRuntime.onEvent(
+          "agent",
+          new TaskEvent(task),
+          retainedEvents,
+          lastEvent,
+          ignored -> deliveredCount.incrementAndGet(),
+          notificationDone,
+          lastActivityNanos,
+          eventCount);
+      DefaultA2AJavaClientRuntime.onEvent(
+          "agent",
+          new TaskUpdateEvent(
+              task, new TaskStatusUpdateEvent("task-1", status, "notification-context", Map.of())),
+          retainedEvents,
+          lastEvent,
+          ignored -> deliveredCount.incrementAndGet(),
+          notificationDone,
+          lastActivityNanos,
+          eventCount);
+      assertEquals(1, notificationDone.getCount(), "task status must not close Notification-T");
+    }
+    TaskStatus heartbeat = new TaskStatus(TaskState.TASK_STATE_WORKING);
+    Task heartbeatTask =
+        Task.builder()
+            .id("subscription-task")
+            .contextId("notification-context")
+            .status(heartbeat)
+            .artifacts(List.of())
+            .history(List.of())
+            .metadata(Map.of())
+            .build();
+    DefaultA2AJavaClientRuntime.onEvent(
+        "agent",
+        new TaskUpdateEvent(
+            heartbeatTask,
+            new TaskStatusUpdateEvent(
+                "subscription-task", heartbeat, "notification-context", Map.of())),
+        retainedEvents,
+        lastEvent,
+        ignored -> deliveredCount.incrementAndGet(),
+        notificationDone,
+        lastActivityNanos,
+        eventCount);
+    assertEquals(
+        1, notificationDone.getCount(), "heartbeat must still reach an active subscription");
+    assertEquals(13, eventCount.get());
+    assertEquals(13, deliveredCount.get());
+    assertTrue(retainedEvents.isEmpty(), "long-lived streams must not retain every event");
+    assertTrue(lastActivityNanos.get() > System.nanoTime() - TimeUnit.SECONDS.toNanos(1));
+
+    CountDownLatch taskDone = new CountDownLatch(1);
+    Task completedTask =
+        Task.builder()
+            .id("task-2")
+            .contextId("task-context")
+            .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED))
+            .artifacts(List.of())
+            .history(List.of())
+            .metadata(Map.of())
+            .build();
+    DefaultA2AJavaClientRuntime.onEvent(
+        "agent",
+        new TaskEvent(completedTask),
+        new ArrayList<>(),
+        new AtomicReference<>(),
+        null,
+        taskDone,
+        null,
+        null);
+    assertEquals(0, taskDone.getCount(), "ordinary Task-T must retain terminal handling");
   }
 }