fix: correct header overrides and streaming lifecycle behavior

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- Repeated default header settings now replace earlier values case-insensitively, while per-request headers retain precedence.
+- Realtime handshake and connection failures now retain the underlying exception through `RealtimeException.getCause()` for diagnostics.
+- Resumable SSE subscriptions honor valid `Retry-After` and `retry-after-ms` response headers after HTTP retries are exhausted, while cancellation and total deadlines still interrupt reconnect waits.
+- Closing an active event subscription records cancellation before releasing its reader, preventing a race that could incorrectly report successful completion.
+
 ### Validation
 
 - Manual Maven Central release workflow with source/CI checks, verified artifact digests, signing, retained deployment recovery, public package verification and matching GitHub Release notes. Publishing uses repository Actions secrets and proceeds after preflight without a separate environment approval.

docs/identities.md

@@ -1,6 +1,6 @@
 # Identity and configuration
 
-The Identity resource currently exposes eleven operations, its Config child exposes four, and its Memory Store child exposes three. All use the client's shared PAT/SA authentication, request options and retry policy. See [Memory Store mounts](sessions.md#memory-store-mounts) for list/mount/detach. Identity SAT issuance is a separate planned addition.
+The Identity resource exposes eleven operations, its Config child exposes four, and its Memory Store child exposes three. These operations use the client's shared PAT/SA authentication, request options and retry policy. See [Memory Store mounts](sessions.md#memory-store-mounts) for list/mount/detach. Identity SAT issuance is also available through `client.identities().serviceAccountTokens()`, using a separate per-call SA Key and disabling automatic retries; see [Forward SAT management](usage-sat-drive.md).
 
 ## Resource methods
 
@@ -18,6 +18,7 @@ The Identity resource currently exposes eleven operations, its Config child expo
 | | `retrieve(identityId, templateId)` | Stored `IdentityConfig` |
 | | `upsert(identityId, templateId, params)` | Updated `IdentityConfig` |
 | | `getEffective(identityId, templateId)` | Compiled `EffectiveConfig` |
+| `client.identities().serviceAccountTokens()` | `create(identityId, saKey, params)` | `ServiceAccountToken` |
 
 Each method also accepts `RequestOptions` as its final argument. `client.async()` exposes the same methods with `CompletableFuture` results. `withResponse()` returns HTTP status, headers and request ID in addition to the typed data.
 

docs/realtime.md

@@ -42,6 +42,7 @@ Creation requires an application-generated stable `Idempotency-Key`; retry the s
 - Sends are serialized. Successful completion means the transport wrote the message, not that the server processed it. Cancelling a queued send removes it; cancelling or timing out a started send closes the connection because delivery is uncertain. Input is never replayed automatically.
 - Async continuations can execute on SDK/transport threads. Use `thenAcceptAsync`/an application executor for blocking work; do not synchronously wait for another event inside a completion callback.
 - A normal peer close ends `receive` with `null`. Buffered events drain before a terminal error. Abnormal closes, replacement and fatal server error events terminate the connection; `RealtimeException` preserves conversation/request IDs, close code/reason and the triggering error event through explicit getters. Nonfatal `error` and `work.failed` remain events for the application to handle.
+- Transport failures wrapped in `RealtimeException` retain their underlying exception through `getCause()`, while the top-level message remains generic.
 - Closing the parent client stops its connections. Always close a connection or use try-with-resources, including after asynchronous use. Closing a non-owning `client.async()` view does not close its parent.
 
 There is no automatic reconnect or input replay. A caller can explicitly reconnect with the conversation ID and query persisted History to decide how to recover. WS `sequence`/`event_id` are not History or SSE cursors.

docs/session-streaming.md

@@ -47,7 +47,7 @@ try (EventStream<SessionEvent> stream = client.sessions().events().resumableStre
 }
 ```
 
-`resumableStream` is lazy and reconnects after EOF, transport failures and retryable HTTP errors. Reconnect waits use jittered exponential backoff from 500 ms to a 10-second cap. This continues until closed, the total deadline expires, a terminal event arrives, or a nonretryable error occurs. `maxRetries` governs each HTTP handshake, not the lifetime of this explicitly persistent subscription. A connection that yields events after five seconds resets its backoff.
+`resumableStream` is lazy and reconnects after EOF, transport failures and retryable HTTP errors. For HTTP errors, reconnect waits honor valid `retry-after-ms` or `Retry-After` (seconds or HTTP date), including after the HTTP retry budget is exhausted. Server delays must be positive and at most 2³¹−1 milliseconds, following the ordinary HTTP retry policy. Otherwise, reconnect waits use jittered exponential backoff from 500 ms to a 10-second cap. Cancellation and the total deadline interrupt either kind of wait. This continues until closed, the total deadline expires, a terminal event arrives, or a nonretryable error occurs. `maxRetries` governs each HTTP handshake, not the lifetime of this explicitly persistent subscription. A connection that yields events after five seconds resets its backoff.
 
 - Reconnect retains the path, filters and request options, but replaces `Last-Event-ID` with the last fully **yielded** frame's ID. Calling `hasNext()` alone does not advance the public checkpoint. An incomplete or invalid frame is not acknowledged. An empty SSE ID clears the cursor. The SSE frame ID is used, not an assumed JSON payload ID.
 - It calls the shared Credential again for each connection. `ServiceAccountCredential` reuses or refreshes SAT according to expiry; it does not send an SA Key to business endpoints. There is no mid-connection token refresh or forced reconnect just because a token expires.

docs/usage-sat-drive.md

@@ -1,6 +1,6 @@
 # Usage、Forward SAT 管理与 Drive
 
-本阶段实现 10 个 HTTP 操作：Usage 2 个、Forward SAT 管理 3 个、Drive 5 个。均提供同步、异步和 `withResponse()`，共用请求超时、取消、错误与响应元信息。当前仅完成离线验证，未在 CN/Global 执行这些 Java 示例。
+本节覆盖 10 个 HTTP 操作：Usage 2 个、Forward SAT 管理 3 个、Drive 5 个。均提供同步、异步和 `withResponse()`，共用请求超时、取消、错误与响应元信息。默认测试使用本地模拟服务；CN/Global 的示例与功能验收按[线上测试指南](live-testing.md)显式运行。
 
 ## Usage
 
@@ -119,4 +119,4 @@ client.templates().update(templateId, patch);
 
 该配置会更新所选 Template 的工具列表，按需要保留其他工具；SDK 不暴露 `/forward-mcp/...` 内部工具路由。不要把 Drive 文件管理与 Agent 内部工具调用混为一类接口。
 
-下一步：实现 Session / Thread SSE，再实现 Realtime 与 History，随后统一做真实环境验收。
+SDK 同时提供 [Session / Thread SSE](session-streaming.md) 和 [Realtime 与 History](realtime.md)。对应文档说明订阅、连接、取消和历史分页的使用方式；真实环境验收见[线上测试指南](live-testing.md)。

src/main/java/com/qoder/cloudagents/ForwardClient.java

@@ -219,6 +219,7 @@ public Builder maxRetries(int value) {
 
     public Builder defaultHeader(String name, String value) {
       RequestOptions.builder().header(name, value);
+      headers.keySet().removeIf(key -> key.equalsIgnoreCase(name));
       headers.put(name, value);
       return this;
     }

src/main/java/com/qoder/cloudagents/core/internal/RetryPolicy.java

@@ -23,16 +23,21 @@ public static boolean shouldRetry(
   }
 
   public static long delayMillis(Headers headers, int attempt, Clock clock, double random) {
-    Double delay = number(headers.get("retry-after-ms"));
+    Long delay = retryAfterMillis(headers.get("retry-after-ms"), headers.get("retry-after"), clock);
+    if (delay != null) return delay;
+    return (long) (Math.min(500 * Math.pow(2, Math.min(attempt, 5)), 8000) * (1 - random * .25));
+  }
+
+  public static Long retryAfterMillis(String retryAfterMs, String retryAfter, Clock clock) {
+    Double delay = number(retryAfterMs);
     if (delay == null || delay == 0) {
-      String after = headers.get("retry-after");
-      Double seconds = number(after);
+      Double seconds = number(retryAfter);
       if (seconds != null) delay = seconds * 1000;
-      else if (after != null) {
+      else if (retryAfter != null) {
         try {
           delay =
               (double)
-                  (ZonedDateTime.parse(after, DateTimeFormatter.RFC_1123_DATE_TIME)
+                  (ZonedDateTime.parse(retryAfter, DateTimeFormatter.RFC_1123_DATE_TIME)
                           .toInstant()
                           .toEpochMilli()
                       - clock.millis());
@@ -43,7 +48,7 @@ else if (after != null) {
     }
     if (delay != null && Double.isFinite(delay) && delay > 0 && delay <= MAX_DELAY_MS)
       return Math.max(1, (long) Math.ceil(delay));
-    return (long) (Math.min(500 * Math.pow(2, Math.min(attempt, 5)), 8000) * (1 - random * .25));
+    return null;
   }
 
   private static Double number(String value) {

src/main/java/com/qoder/cloudagents/errors/RealtimeException.java

@@ -17,7 +17,18 @@ public RealtimeException(
       Integer closeCode,
       String closeReason,
       JsonValue event) {
-    super(message);
+    this(message, conversationId, requestId, closeCode, closeReason, event, null);
+  }
+
+  public RealtimeException(
+      String message,
+      String conversationId,
+      String requestId,
+      Integer closeCode,
+      String closeReason,
+      JsonValue event,
+      Throwable cause) {
+    super(message, cause);
     this.conversationId = conversationId;
     this.requestId = requestId;
     this.closeCode = closeCode;

src/main/java/com/qoder/cloudagents/forward/realtime/RealtimeConnection.java

@@ -435,7 +435,14 @@ private RuntimeException asRuntime(Throwable e) {
     Throwable cause = Futures.unwrap(e);
     return cause instanceof RuntimeException
         ? (RuntimeException) cause
-        : problem("Realtime connection failed", null);
+        : new RealtimeException(
+            "Realtime connection failed",
+            conversationId(),
+            requestId(),
+            closeCode,
+            closeReason,
+            null,
+            cause);
   }
 
   private RealtimeException problem(String message, JsonValue event) {

src/main/java/com/qoder/cloudagents/streaming/EventStream.java

@@ -5,6 +5,7 @@
 import com.qoder.cloudagents.errors.*;
 import java.io.IOException;
 import java.nio.charset.CharacterCodingException;
+import java.time.Clock;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -167,7 +168,7 @@ private Item<T> readNext() {
             finish();
             return null;
           }
-          reconnect();
+          reconnect(null);
           continue;
         }
         if (frame.type.equals("ping")) continue;
@@ -198,10 +199,22 @@ private boolean recover(RuntimeException error) {
       fail(error);
       throw error;
     }
-    reconnect();
+    Long delay = null;
+    if (error instanceof ApiStatusException) {
+      ApiStatusException status = (ApiStatusException) error;
+      delay =
+          RetryPolicy.retryAfterMillis(
+              header(status, "retry-after-ms"), header(status, "retry-after"), Clock.systemUTC());
+    }
+    reconnect(delay);
     return true;
   }
 
+  private static String header(ApiStatusException error, String name) {
+    List<String> values = error.headers().get(name);
+    return values == null || values.isEmpty() ? null : values.get(values.size() - 1);
+  }
+
   private static boolean retryable(RuntimeException error) {
     if (error instanceof ApiStatusException) {
       ApiStatusException status = (ApiStatusException) error;
@@ -214,9 +227,12 @@ private static boolean retryable(RuntimeException error) {
     return error instanceof ApiConnectionException;
   }
 
-  private synchronized void reconnect() {
+  private synchronized void reconnect(Long retryAfterMillis) {
     releaseConnection();
-    long delay = ThreadLocalRandom.current().nextLong(Math.max(1, backoff / 2), backoff + 1);
+    long delay =
+        retryAfterMillis != null
+            ? retryAfterMillis
+            : ThreadLocalRandom.current().nextLong(Math.max(1, backoff / 2), backoff + 1);
     backoff = Math.min(backoff * 2, 10000);
     long until = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(delay);
     try {

src/main/java/com/qoder/cloudagents/streaming/EventSubscription.java

@@ -22,7 +22,8 @@ public CompletableFuture<Void> completion() {
   /** In-flight user callbacks may finish; no further callback is scheduled. */
   @Override
   public void close() {
-    close.run();
+    // Record cancellation before waking the reader, which could otherwise report success.
     completion.cancel(false);
+    close.run();
   }
 }

src/test/java/com/qoder/cloudagents/ForwardClientTest.java

@@ -68,6 +68,36 @@ void modelContractPreservesPrecisionAndRequestMetadata() throws Exception {
     assertEquals("0", request.getHeader("X-Qoder-Retry-Count"));
   }
 
+  @Test
+  void defaultHeadersUseLastValueRegardlessOfCaseAndAllowRequestOverrides() throws Exception {
+    try (ForwardClient configured =
+        builder()
+            .defaultHeader("X-Test-Header", "initial")
+            .defaultHeader("x-test-header", "middle")
+            .defaultHeader("X-Test-Header", "final")
+            .build()) {
+      server.enqueue(json(MODELS));
+      configured.models().list();
+      RecordedRequest request = server.takeRequest(2, TimeUnit.SECONDS);
+      assertNotNull(request);
+      assertEquals(
+          java.util.Collections.singletonList("final"),
+          request.getHeaders().values("x-test-header"));
+
+      server.enqueue(json(MODELS));
+      configured
+          .async()
+          .models()
+          .list(RequestOptions.builder().header("X-TEST-HEADER", "request").build())
+          .get(3, TimeUnit.SECONDS);
+      RecordedRequest overridden = server.takeRequest(2, TimeUnit.SECONDS);
+      assertNotNull(overridden);
+      assertEquals(
+          java.util.Collections.singletonList("request"),
+          overridden.getHeaders().values("x-test-header"));
+    }
+  }
+
   @Test
   void asyncTemplateQueriesAndOpaquePathIdsAreEncoded() throws Exception {
     server.enqueue(

src/test/java/com/qoder/cloudagents/RealtimeConnectionTest.java

@@ -6,6 +6,7 @@
 import com.qoder.cloudagents.core.*;
 import com.qoder.cloudagents.errors.*;
 import com.qoder.cloudagents.forward.realtime.*;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.*;
@@ -417,10 +418,29 @@ void abruptDisconnectIsNotReconnected() throws Exception {
       RealtimeException e =
           assertThrows(RealtimeException.class, () -> c.receive(Duration.ofSeconds(3)));
       assertEquals(1006, e.closeCode());
+      assertEquals("conv_test", e.conversationId());
+      assertEquals("req-ws", e.requestId());
+      assertInstanceOf(IOException.class, e.getCause());
+      assertEquals("Realtime connection failed", e.getMessage());
       assertEquals(1, server.getRequestCount());
     }
   }
 
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  void handshakeDisconnectPreservesTransportCause(boolean async) {
+    server.enqueue(new MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AT_START));
+    Throwable error =
+        async
+            ? failed(client.async().realtime().connect(params().build()))
+            : assertThrows(RealtimeException.class, this::connect);
+    RealtimeException failure = assertInstanceOf(RealtimeException.class, error);
+    assertInstanceOf(IOException.class, failure.getCause());
+    assertEquals("conv_test", failure.conversationId());
+    assertEquals(1006, failure.closeCode());
+    assertEquals("Realtime connection failed", failure.getMessage());
+  }
+
   @Test
   void normalCloseReturnsEndOfStream() throws Exception {
     upgrade(true, false);

src/test/java/com/qoder/cloudagents/SessionStreamsTest.java

@@ -14,6 +14,7 @@
 import okhttp3.mockwebserver.*;
 import org.junit.jupiter.api.*;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 @Timeout(15)
@@ -378,6 +379,93 @@ void explicitResumeRetriesTransientStatus(int code) throws Exception {
     assertEquals(2, server.getRequestCount());
   }
 
+  @ParameterizedTest
+  @CsvSource({
+    "429, Retry-After, 1, 0",
+    "503, retry-after-ms, 800, 0",
+    "429, retry-after-ms, 800, 1"
+  })
+  void reconnectHonorsServerDelayAfterHttpRetryBudget(
+      int status, String header, String value, int retries) throws Exception {
+    if (retries > 0)
+      server.enqueue(
+          new MockResponse()
+              .setResponseCode(status)
+              .setHeader("retry-after-ms", "1")
+              .setBody("{}"));
+    server.enqueue(
+        new MockResponse().setResponseCode(status).setHeader(header, value).setBody("{}"));
+    server.enqueue(sse(event("end", "session.deleted")));
+    try (EventStream<SessionEvent> stream =
+        client
+            .sessions()
+            .events()
+            .resumableStream(
+                "s",
+                SessionEventStreamParams.builder().build(),
+                RequestOptions.builder()
+                    .maxRetries(retries)
+                    .totalTimeout(Duration.ofSeconds(5))
+                    .build())) {
+      EventSubscription subscription = stream.subscribe(e -> {});
+      for (int i = 0; i <= retries; i++) take();
+      assertNull(server.takeRequest(600, TimeUnit.MILLISECONDS));
+      take();
+      subscription.completion().get(2, TimeUnit.SECONDS);
+      assertEquals(retries + 2, server.getRequestCount());
+    }
+  }
+
+  @Test
+  void totalDeadlineInterruptsServerRetryDelay() {
+    server.enqueue(
+        new MockResponse().setResponseCode(429).setHeader("Retry-After", "30").setBody("{}"));
+    server.enqueue(sse(event("end", "session.deleted")));
+    try (EventStream<SessionEvent> stream =
+        client
+            .sessions()
+            .events()
+            .resumableStream("s", SessionEventStreamParams.builder().build(), deadline(1000))) {
+      assertThrows(ApiTimeoutException.class, () -> stream.iterator().hasNext());
+      assertEquals(1, server.getRequestCount());
+    }
+  }
+
+  @Test
+  void subscriptionCloseInterruptsServerRetryDelayAndReportsCancellation() throws Exception {
+    server.enqueue(
+        new MockResponse().setResponseCode(429).setHeader("Retry-After", "30").setBody("{}"));
+    server.enqueue(sse(event("end", "session.deleted")));
+    ExecutorService reader = Executors.newSingleThreadExecutor();
+    CompletableFuture<Void> finished = new CompletableFuture<>();
+    try (EventStream<SessionEvent> stream =
+        client
+            .sessions()
+            .events()
+            .resumableStream("s", SessionEventStreamParams.builder().build(), deadline(5000))) {
+      EventSubscription subscription =
+          stream.subscribe(
+              e -> fail("No event expected during the server retry delay"),
+              task ->
+                  reader.execute(
+                      () -> {
+                        try {
+                          task.run();
+                        } finally {
+                          finished.complete(null);
+                        }
+                      }));
+      take();
+      assertNull(server.takeRequest(600, TimeUnit.MILLISECONDS));
+      subscription.close();
+      assertThrows(CancellationException.class, () -> subscription.completion().get());
+      finished.get(1, TimeUnit.SECONDS);
+      assertEquals(1, server.getRequestCount());
+    } finally {
+      reader.shutdownNow();
+    }
+  }
+
   @Test
   void retryHintFalseStopsResume() {
     server.enqueue(