From bc1f1706c2969d70148f8365e58794257809b569 Mon Sep 17 00:00:00 2001 From: "yuechen.wyh" <300878504+moonyue-w@users.noreply.github.com> Date: Thu, 22 Oct 2026 17:39:16 +0800 Subject: [PATCH 1/2] feat(java): close Forward field gaps and enable client extensions --- CHANGELOG.md | 10 + README.md | 1 + docs/channels.md | 2 + docs/client-configuration.md | 70 +- docs/forward-api-coverage.md | 2 + docs/memory-stores.md | 2 +- docs/schedules-batches.md | 2 + docs/sessions.md | 2 + examples/README.md | 2 + .../examples/ClientExtensionsExample.java | 53 ++ .../ClientExtensionsConsumerTest.java | 104 +++ .../qoder/cloudagents/AsyncForwardClient.java | 31 +- .../com/qoder/cloudagents/ForwardClient.java | 120 +++- .../qoder/cloudagents/core/ApiRequest.java | 76 +++ .../core/http/HttpInterceptor.java | 18 + .../cloudagents/core/http/HttpRequest.java | 104 +++ .../cloudagents/core/http/HttpResponse.java | 50 ++ .../cloudagents/core/http/HttpTransport.java | 15 + .../core/internal/ClientEngine.java | 39 +- .../core/internal/FileDownloads.java | 12 +- .../core/internal/HttpExtensions.java | 383 +++++++++++ .../cloudagents/forward/ChannelPairing.java | 8 + .../cloudagents/forward/ImageSource.java | 8 + .../forward/MemoryStoreCreateMount.java | 8 + .../qoder/cloudagents/forward/Schedule.java | 4 + .../cloudagents/forward/ScheduleRun.java | 4 + .../cloudagents/ClientExtensionsTest.java | 631 ++++++++++++++++++ .../cloudagents/ForwardFieldParityTest.java | 102 +++ 28 files changed, 1834 insertions(+), 29 deletions(-) create mode 100644 examples/src/main/java/com/qoder/cloudagents/examples/ClientExtensionsExample.java create mode 100644 examples/src/test/java/com/qoder/cloudagents/examples/ClientExtensionsConsumerTest.java create mode 100644 src/main/java/com/qoder/cloudagents/core/ApiRequest.java create mode 100644 src/main/java/com/qoder/cloudagents/core/http/HttpInterceptor.java create mode 100644 src/main/java/com/qoder/cloudagents/core/http/HttpRequest.java create mode 100644 src/main/java/com/qoder/cloudagents/core/http/HttpResponse.java create mode 100644 src/main/java/com/qoder/cloudagents/core/http/HttpTransport.java create mode 100644 src/main/java/com/qoder/cloudagents/core/internal/HttpExtensions.java create mode 100644 src/test/java/com/qoder/cloudagents/ClientExtensionsTest.java create mode 100644 src/test/java/com/qoder/cloudagents/ForwardFieldParityTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 33674e0..f097712 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## [Unreleased] +### Added + +- Public Java 8 HTTP transport and per-attempt interceptor contracts for custom networking and tracing, without exposing OkHttp types. Cancellation reaches downstream transport futures; SSE bodies remain streaming. Storage downloads use the transport without API credentials or interceptors. WebSockets and SA Key exchange retain their existing transports. +- Synchronous/asynchronous `request(ApiRequest, RequestOptions)` for additional JSON endpoints, reusing authentication, retries, cancellation and response metadata with relative path segments. +- Synchronous/asynchronous `withOptions(...)` to derive independently closeable clients while retaining resolved authentication and unspecified settings. Caller-supplied credentials and transports remain caller-owned. + +### Fixed + +- Add typed image `file_id`, Memory Store creation-mount `access`, Schedule/Run `initial_events_version` and Pairing `created_at` / `updated_at` fields. Missing versions remain unknown rather than becoming zero. + ## [0.1.0] - 2026-10-21 First stable release of the Java SDK for the Forward API, using `com.qoder.sdk:qoder-cloud-agents-sdk:0.1.0`. diff --git a/README.md b/README.md index 1798a99..7a3406b 100644 --- a/README.md +++ b/README.md @@ -227,6 +227,7 @@ Microphone capture and audio playback belong to the application. Connections do - [Runnable examples](examples/README.md): shared PAT/SA setup, resource lifecycles, Session SSE and Realtime. - [API coverage](docs/forward-api-coverage.md): supported operations and guides for each resource. +- [Client extensions (unreleased)](docs/client-configuration.md#client-extensions-unreleased): custom HTTP transport, interceptors, generic JSON requests and configuration copies. - [Files and downloads](docs/environments-files.md) and [Drive, Usage and SAT management](docs/usage-sat-drive.md). - [Live testing](docs/live-testing.md): opt-in CN/Global verification and coverage boundaries. diff --git a/docs/channels.md b/docs/channels.md index ffdcc99..df04521 100644 --- a/docs/channels.md +++ b/docs/channels.md @@ -102,6 +102,8 @@ DeletedChannelPairing removed = client.channelPairings().delete(pairing.id()); Use the returned Pairing ID for revocation and Schedule sinks. Current API supports completion by pending Pairing ID as well as code; code-only older SDK types are not copied. An ID-based completion can be retried against the same binding without depending on code expiry. Used/expired codes cannot be treated as fresh pairing requests. Permission, ownership and concurrent-change errors remain server decisions. +The development SDK also exposes nullable `createdAt()` and `updatedAt()` strings on `ChannelPairing`, preserving timestamps returned by the service. These getters are not included in the published `0.1.0` JAR. + `delete` removes the Channel connection; it is distinct from `channelPairings().delete`, which revokes one binding. Already absent/unpaired bindings can return `deleted=true`. SDK keeps the declared deletion response and accepts an empty 204 without inventing a JSON body; use `withResponse()` for metadata in that case. ## Lists and examples diff --git a/docs/client-configuration.md b/docs/client-configuration.md index f263c38..1c8a011 100644 --- a/docs/client-configuration.md +++ b/docs/client-configuration.md @@ -87,7 +87,75 @@ com.qoder.cloudagents.core.ApiResponse r - Models provide `requestId()`, `field(name)` for unknown fields and `rawJson()`. Unknown fields and numeric precision are preserved; invalid typed field access raises `ResponseValidationException`. Request metadata is separate from the JSON body. - Template list parameters support `limit`, `status`, `afterId` or `beforeId`. Use `page.data()` for one page, `getNextPage()` / `getNextPageAsync()` for manual pagination, or `for (Template template : page)` for lazy synchronous iteration. Server order is retained; filters and request options follow subsequent pages. Check `hasNextPage()` before fetching; an exhausted page or repeated cursor raises `PaginationException`. -## Template configuration +## Client extensions (unreleased) + +The following client extensions are on the development branch and are not included in the published `0.1.0` JAR. Build and install the local SDK to try them. + +### Derived clients + +```java +try (ForwardClient slow = client.withOptions(options -> options + .timeout(java.time.Duration.ofSeconds(60)) + .maxRetries(1) + .defaultHeader("X-Trace-Source", "background-job"))) { + slow.models().list(); +} +``` + +`withOptions` retains the resolved BaseURL, authentication, headers, interceptors and custom transport unless changed. It does not reread environment credentials. Header setters replace matching names case-insensitively; interceptors append in registration order. Use `clearDefaultHeaders()` or `clearInterceptors()` to replace those collections. Per-request options still take precedence. `client.async().withOptions(...)` creates an independently owned async client. + +Copies have separate default connection pools, retry timers and in-flight request lifecycles. Closing either client does not close the other. Custom transports and credentials are shared by reference and remain caller-owned; keep them open until all clients and streams finish. Changing `.pat(...)`, `.accessToken(...)` or `.credential(...)` on a copy replaces its inherited credential source. For a static SAT use `.authMode(AuthMode.SA).accessToken(token)`. + +### Additional JSON endpoints + +```java +import com.qoder.cloudagents.core.ApiRequest; +import com.qoder.cloudagents.core.ApiResponse; +import com.qoder.cloudagents.core.JsonValue; + +ApiResponse response = client.request( + ApiRequest.builder("GET", "models").build(), + RequestOptions.builder().totalTimeout(java.time.Duration.ofSeconds(30)).build()); +System.out.println(response.requestId()); +``` + +Pass unencoded path **segments**, for example `ApiRequest.builder("POST", "sessions", sessionId, "new-action")`; the SDK encodes each segment beneath its configured BaseURL. Absolute URLs, slash-separated paths, traversal and embedded queries/fragments are rejected. Use `RequestOptions.query(...)` and `header(...)` for query parameters and headers, and `.body(JsonValue)` for a JSON body. GET/HEAD cannot have a body. HEAD and 204/205 responses return JSON null. + +The return type includes JSON, status and response headers; `client.async().request(...)` returns a cancellable `CompletableFuture>`. HTTP errors retain the existing exception mapping. This entry point uses business PAT/SAT credentials and the shared retry rules; it does not add resource-specific validation. Prefer typed resource methods where available, particularly SA Key token management and credential rotation. Use the existing resource APIs for multipart upload, downloads and SSE. + +### HTTP interceptors and custom transport + +```java +try (ForwardClient traced = client.withOptions(options -> options.interceptor((request, chain) -> { + long started = System.nanoTime(); + return chain.proceed(request.withHeader("X-Trace-Source", "example")) + .whenComplete((response, error) -> { + long elapsedMs = (System.nanoTime() - started) / 1_000_000; + System.out.println("http_elapsed_ms=" + elapsedMs); + }); +}))) { + traced.models().list(); +} +``` + +`HttpInterceptor` runs after authentication on **every attempt**, including SSE connection/reconnection and explicit SA Key token-management requests. It may inspect metadata, add ordinary headers or return a response, but may call `chain.proceed(...)` only once. It cannot change routing, method, body, timeout, Authorization or idempotency. Do not log headers, bodies or full URLs, or read a streaming body just to log it. Interceptor callbacks must return promptly. Their completion future measures response-header arrival, not SSE/download completion. + +Implement `com.qoder.cloudagents.core.http.HttpTransport` and pass it to `.transport(...)` when an application needs its own HTTP library, proxy, TLS settings or connection pool. The SDK passes an immutable `HttpRequest` with URI, headers, connect/read/write timeouts and an optional streaming `Body`. Use `Body.contentType()`, `contentLength()` and `writeTo(OutputStream)` when sending uploads; do not buffer all uploads or close the destination from `writeTo`. + +The transport must return a `CompletableFuture` promptly, perform one attempt without automatic redirects/retries/authentication, honor timeouts, and abort its I/O when its future is cancelled. Return an unconsumed `InputStream` in `HttpResponse`; the SDK closes it on completion, cancellation or failure. Transport implementations must release responses that lose a race with cancellation. Read timeout is zero for an accepted SSE stream; SDK header and optional total deadlines remain active. Cancelling an SDK request also cancels downstream transport futures even when an interceptor has wrapped them in `thenApply`/`whenComplete`. + +| Operation | Custom transport | API interceptors | +| --- | --- | --- | +| JSON, multipart, authenticated downloads | Yes | Yes | +| Session/Thread SSE, including resumption | Yes | Yes, per connection | +| File signed-URL storage download | Yes | No; no API headers/credentials copied | +| Realtime create/History HTTP | Yes | Yes | +| Realtime WebSocket | No | No | +| `ServiceAccountCredential` platform exchange | No | No | + +The default HTTP implementation remains OkHttp. Injected transports and interceptors do not change the SDK's retry safety rules or grant additional API permissions. + +## Template configuration example ```java import com.qoder.cloudagents.forward.TemplateCreateParams; diff --git a/docs/forward-api-coverage.md b/docs/forward-api-coverage.md index 7d5d851..b5029ac 100644 --- a/docs/forward-api-coverage.md +++ b/docs/forward-api-coverage.md @@ -10,6 +10,8 @@ HTTP BaseURL 为 CN `https://api.qoder.com.cn/api/v1/forward` 或 Global `https: 这里只盘点协议入口；参数、响应、分页、认证和重试行为由对应资源指南与契约测试说明。 +开发分支已按 Go `a996a88`、Python `2f8c625`、TypeScript `f4e3a4f` 的 main 补齐 ImageSource `file_id`、创建 Memory Store 时的 mount `access`、Schedule/Run `initial_events_version` 和 ChannelPairing 时间戳。接口操作集合仍为 123 项；客户端 HTTP 扩展、通用 JSON 请求和配置复制见[客户端配置](client-configuration.md)，这些新增能力尚不在已发布的 `0.1.0` 中。 + ## 使用指南 - [客户端认证、超时、重试与取消](client-configuration.md) diff --git a/docs/memory-stores.md b/docs/memory-stores.md index f4f3a66..e435f82 100644 --- a/docs/memory-stores.md +++ b/docs/memory-stores.md @@ -42,7 +42,7 @@ MemoryStoreCreateMount mount = MemoryStoreCreateMount.builder() // Supply .mount(mount) on MemoryStoreCreateParams.builder(). ``` -Both IDs are required when supplying `mount`. This is a read-only explicit mount, matching the separate Identity mount API. Store `bindingInfo().identityTemplateCount()` reports active mounts; unknown binding fields remain available through `field(...)`. +Both IDs are required when supplying `mount`. The development SDK also supports `.access("read_only")` or `.access("read_write")`; omitting it leaves the default to the service. This typed setter is not included in the published `0.1.0` JAR. Store `bindingInfo().identityTemplateCount()` reports active mounts; unknown binding fields remain available through `field(...)`. Store metadata is a string map with at most 15 user keys. `created_by` is reserved and added by the service. Memory metadata allows 16 keys and does not reserve `created_by`. Key/value size rules and other business constraints are enforced by the service. The SDK does not insert reserved metadata. diff --git a/docs/schedules-batches.md b/docs/schedules-batches.md index 2a0496a..906019e 100644 --- a/docs/schedules-batches.md +++ b/docs/schedules-batches.md @@ -24,6 +24,8 @@ Schedule schedule = client.schedules().create( Schedule initial events use string `content` as documented by this endpoint; ordinary Session events have their own content-block model. `ScheduleTriggerPolicyParams` exposes request fields (`type`, `expression`, `timezone`), while `ScheduleTriggerPolicy` includes computed response fields such as `upcomingRunsAt` and `lastRunAt`. Cron/timezone, once-time and interval rules are validated by the server. +The development SDK exposes `initialEventsVersion()` on both `Schedule` and `ScheduleRun`. It returns a nullable `Long`: an absent version remains unknown, distinct from an explicit zero. This getter is not included in the published `0.1.0` JAR. + - `schedules().list/create/retrieve/update/archive/archiveMany/pause/unpause/run` cover the nine Schedule operations. - `scheduleRuns().list/retrieve` cover Run queries. Listing requires `ScheduleRunListParams.identityId`; retrieval accepts optional `ScheduleRunRetrieveParams.identityId`. - An omitted trigger policy or explicit `null` creates a manual Schedule. Updating `triggerPolicy(null)` switches to manual. Updating `sinks(null)` or an empty list clears delivery targets; omitting a field retains it. Execution fields merge. Metadata values of JSON `null` remove keys. diff --git a/docs/sessions.md b/docs/sessions.md index 880a008..0c0db1f 100644 --- a/docs/sessions.md +++ b/docs/sessions.md @@ -66,6 +66,8 @@ System.out.println(accepted.requestId()); `EventContent.text(...)` creates the content-block array required for a user message. Use `EventContent.blocks(...)` with typed `ContentBlock` and `ImageSource` for text/image inputs. `EventContent.ofString(...)` represents string content for event types such as tool results; do not use it for a `user.message`. +The development SDK adds `ImageSource.builder().type("file").fileId(fileId)` and `fileId()` for images referenced by file ID. These typed methods are not included in the published `0.1.0` JAR; the service validates the image source and file permissions. + `SessionEventParam` also exposes tool confirmation/result fields, custom tool result IDs, interrupt reason and outcome definition. `OutcomeRubric` is an object with a type and text/file ID, matching the documented outcome contract. Event-specific validation and permissions remain server-side. A successful send acknowledges accepted input; it does not mean the model finished. Reading history returns persisted events, not a live subscription: diff --git a/examples/README.md b/examples/README.md index 21abe93..996b50e 100644 --- a/examples/README.md +++ b/examples/README.md @@ -17,6 +17,8 @@ For SDK development or an unpublished version, install the local SDK first: Verification uses a local fake service and asserts the SDK was loaded from a JAR. It does not use production credentials. +`ClientExtensionsExample` demonstrates a read-only generic JSON request, a per-attempt tracing interceptor and an independently closeable client copy. It also accepts an application-owned `HttpTransport`. These extension APIs are unreleased; install the development SDK first, then run with `-Dexec.mainClass=com.qoder.cloudagents.examples.ClientExtensionsExample`. The packaged-consumer test verifies an adapter built with Java 8 `HttpURLConnection`, without importing SDK internals. + To run against your environment, configure credentials in your shell and choose one mode: | Mode | Environment | diff --git a/examples/src/main/java/com/qoder/cloudagents/examples/ClientExtensionsExample.java b/examples/src/main/java/com/qoder/cloudagents/examples/ClientExtensionsExample.java new file mode 100644 index 0000000..67dd0ae --- /dev/null +++ b/examples/src/main/java/com/qoder/cloudagents/examples/ClientExtensionsExample.java @@ -0,0 +1,53 @@ +package com.qoder.cloudagents.examples; + +import com.qoder.cloudagents.ForwardClient; +import com.qoder.cloudagents.core.ApiRequest; +import com.qoder.cloudagents.core.ApiResponse; +import com.qoder.cloudagents.core.JsonValue; +import com.qoder.cloudagents.core.RequestOptions; +import com.qoder.cloudagents.core.http.HttpTransport; +import java.io.PrintStream; +import java.time.Duration; + +/** Read-only model listing through a copied client, HTTP interceptor and generic JSON request. */ +public final class ClientExtensionsExample { + private ClientExtensionsExample() {} + + public static void main(String[] args) throws Exception { + try (ExampleClient setup = ExampleClient.fromEnvironment()) { + run(setup.client(), null, System.out); + } + } + + /** Supply a caller-owned transport to use the application's HTTP implementation. */ + public static void run(ForwardClient client, HttpTransport transport, PrintStream output) { + try (ForwardClient derived = + client.withOptions( + options -> { + options + .timeout(Duration.ofSeconds(30)) + .interceptor( + (request, chain) -> { + long started = System.nanoTime(); + return chain + .proceed(request.withHeader("X-Trace-Source", "sdk-example")) + .whenComplete( + (response, error) -> + output.println( + "http_elapsed_ms=" + + (System.nanoTime() - started) / 1_000_000)); + }); + if (transport != null) options.transport(transport); + })) { + ApiResponse response = + derived.request( + ApiRequest.builder("GET", "models").build(), + RequestOptions.builder().totalTimeout(Duration.ofSeconds(30)).build()); + output.println( + "models=" + + response.data().get("data").asArray().size() + + " request_id=" + + response.requestId()); + } + } +} diff --git a/examples/src/test/java/com/qoder/cloudagents/examples/ClientExtensionsConsumerTest.java b/examples/src/test/java/com/qoder/cloudagents/examples/ClientExtensionsConsumerTest.java new file mode 100644 index 0000000..b4bd95c --- /dev/null +++ b/examples/src/test/java/com/qoder/cloudagents/examples/ClientExtensionsConsumerTest.java @@ -0,0 +1,104 @@ +package com.qoder.cloudagents.examples; + +import static org.junit.jupiter.api.Assertions.*; + +import com.qoder.cloudagents.ForwardClient; +import com.qoder.cloudagents.core.http.*; +import java.io.*; +import java.net.HttpURLConnection; +import java.nio.charset.StandardCharsets; +import java.util.*; +import java.util.concurrent.*; +import okhttp3.mockwebserver.*; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(15) +class ClientExtensionsConsumerTest { + @Test + void consumerCanImplementTransportWithoutInternalSdkOrOkHttpTypes() throws Exception { + assertTrue( + ForwardClient.class + .getProtectionDomain() + .getCodeSource() + .getLocation() + .getPath() + .endsWith(".jar")); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try (MockWebServer server = new MockWebServer()) { + server.start(); + server.enqueue( + new MockResponse().setHeader("X-Request-ID", "req_consumer").setBody("{\"data\":[]}")); + server.enqueue(new MockResponse().setBody("{\"data\":[]}")); + // This read-only adapter demonstrates public types. Production adapters also handle uploads. + HttpTransport transport = + request -> { + CompletableFuture future = new CompletableFuture<>(); + executor.execute( + () -> { + try { + HttpURLConnection connection = + (HttpURLConnection) request.uri().toURL().openConnection(); + future.whenComplete( + (value, error) -> { + if (future.isCancelled()) connection.disconnect(); + }); + connection.setInstanceFollowRedirects(false); + connection.setConnectTimeout((int) request.connectTimeout().toMillis()); + connection.setReadTimeout((int) request.readTimeout().toMillis()); + connection.setRequestMethod(request.method()); + request + .headers() + .forEach( + (name, values) -> + values.forEach( + value -> connection.addRequestProperty(name, value))); + if (future.isCancelled()) { + connection.disconnect(); + return; + } + int status = connection.getResponseCode(); + Map> headers = + new LinkedHashMap<>(connection.getHeaderFields()); + headers.remove(null); + InputStream stream = + new FilterInputStream(connection.getInputStream()) { + @Override + public void close() throws IOException { + try { + super.close(); + } finally { + connection.disconnect(); + } + } + }; + HttpResponse response = new HttpResponse(status, headers, stream); + if (!future.complete(response)) response.close(); + } catch (Exception error) { + future.completeExceptionally(error); + } + }); + return future; + }; + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ForwardClient client = + ForwardClient.builder() + .pat("consumer-secret") + .baseUrl(server.url("/api/v1/forward").toString()) + .build(); + PrintStream output = new PrintStream(bytes, true, "UTF-8")) { + ClientExtensionsExample.run(client, transport, output); + client.models().list(); // Closing the derived client did not close the parent. + } + String text = new String(bytes.toByteArray(), StandardCharsets.UTF_8); + assertTrue(text.contains("models=0 request_id=req_consumer")); + assertFalse(text.contains("consumer-secret")); + RecordedRequest first = server.takeRequest(); + assertEquals("Bearer consumer-secret", first.getHeader("Authorization")); + assertEquals("sdk-example", first.getHeader("X-Trace-Source")); + assertNull(server.takeRequest().getHeader("X-Trace-Source")); + } finally { + executor.shutdownNow(); + } + } +} diff --git a/src/main/java/com/qoder/cloudagents/AsyncForwardClient.java b/src/main/java/com/qoder/cloudagents/AsyncForwardClient.java index be2ea8d..e3df025 100644 --- a/src/main/java/com/qoder/cloudagents/AsyncForwardClient.java +++ b/src/main/java/com/qoder/cloudagents/AsyncForwardClient.java @@ -1,6 +1,8 @@ package com.qoder.cloudagents; +import com.qoder.cloudagents.core.*; import com.qoder.cloudagents.core.internal.ClientEngine; +import com.qoder.cloudagents.errors.ConfigurationException; import com.qoder.cloudagents.forward.AsyncBatches; import com.qoder.cloudagents.forward.AsyncChannelPairings; import com.qoder.cloudagents.forward.AsyncChannels; @@ -19,6 +21,9 @@ import com.qoder.cloudagents.forward.AsyncUsage; import com.qoder.cloudagents.forward.AsyncVaults; import com.qoder.cloudagents.forward.realtime.AsyncRealtime; +import java.util.Collections; +import java.util.concurrent.CompletableFuture; +import java.util.function.Consumer; /** Native asynchronous Forward client. Build with ForwardClient.builder().buildAsync(). */ public final class AsyncForwardClient implements AutoCloseable { @@ -27,6 +32,7 @@ public final class AsyncForwardClient implements AutoCloseable { private final AsyncDrives drives; private final AsyncRealtime realtime; private final ClientEngine engine; + private final ForwardClient.Builder configuration; private final boolean ownsEngine; private final AsyncModels models; private final AsyncFiles files; @@ -43,7 +49,8 @@ public final class AsyncForwardClient implements AutoCloseable { private final AsyncSessions sessions; private final AsyncMemoryStores memoryStores; - AsyncForwardClient(ClientEngine engine, boolean ownsEngine) { + AsyncForwardClient(ClientEngine engine, boolean ownsEngine, ForwardClient.Builder configuration) { + this.configuration = configuration; this.engine = engine; realtime = new AsyncRealtime(engine); drives = new AsyncDrives(engine); @@ -138,6 +145,28 @@ public AsyncChannelPairings channelPairings() { return channelPairings; } + /** Creates an independently owned async client; the original client is unchanged. */ + public AsyncForwardClient withOptions(Consumer configure) { + if (configure == null) + throw new ConfigurationException("Configuration callback must not be null"); + ForwardClient.Builder copy = configuration.copy(); + configure.accept(copy); + return copy.buildAsync(); + } + + public CompletableFuture> request(ApiRequest request) { + return request(request, RequestOptions.DEFAULT); + } + + /** The returned future directly owns cancellation of the underlying operation. */ + public CompletableFuture> request( + ApiRequest request, RequestOptions options) { + if (request == null || options == null) + throw new ConfigurationException("Request and options must not be null"); + return engine.request( + request.method(), request.path(), Collections.emptyMap(), request.body(), true, options); + } + @Override public void close() { if (ownsEngine) engine.close(); diff --git a/src/main/java/com/qoder/cloudagents/ForwardClient.java b/src/main/java/com/qoder/cloudagents/ForwardClient.java index 9fc499f..f434ee9 100644 --- a/src/main/java/com/qoder/cloudagents/ForwardClient.java +++ b/src/main/java/com/qoder/cloudagents/ForwardClient.java @@ -2,9 +2,15 @@ import com.qoder.cloudagents.auth.AuthMode; import com.qoder.cloudagents.auth.Credential; +import com.qoder.cloudagents.core.ApiRequest; +import com.qoder.cloudagents.core.ApiResponse; +import com.qoder.cloudagents.core.JsonValue; import com.qoder.cloudagents.core.RequestOptions; +import com.qoder.cloudagents.core.http.HttpInterceptor; +import com.qoder.cloudagents.core.http.HttpTransport; import com.qoder.cloudagents.core.internal.Authentication; import com.qoder.cloudagents.core.internal.ClientEngine; +import com.qoder.cloudagents.core.internal.Futures; import com.qoder.cloudagents.errors.ConfigurationException; import com.qoder.cloudagents.forward.Batches; import com.qoder.cloudagents.forward.ChannelPairings; @@ -25,8 +31,11 @@ import com.qoder.cloudagents.forward.Vaults; import com.qoder.cloudagents.forward.realtime.Realtime; import java.time.Duration; +import java.util.ArrayList; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; +import java.util.function.Consumer; import okhttp3.HttpUrl; /** @@ -38,6 +47,7 @@ public final class ForwardClient implements AutoCloseable { private final Drives drives; private final Realtime realtime; private final ClientEngine engine; + private final Builder configuration; private final Models models; private final Files files; private final ChannelPairings channelPairings; @@ -54,10 +64,11 @@ public final class ForwardClient implements AutoCloseable { private final MemoryStores memoryStores; private final AsyncForwardClient async; - private ForwardClient(ClientEngine engine) { + private ForwardClient(ClientEngine engine, Builder configuration) { + this.configuration = configuration; this.engine = engine; realtime = new Realtime(engine); - async = new AsyncForwardClient(engine, false); + async = new AsyncForwardClient(engine, false, configuration); models = new Models(async.models()); drives = new Drives(async.drives()); serviceAccountTokens = new ServiceAccountTokens(async.serviceAccountTokens()); @@ -161,6 +172,22 @@ public ChannelPairings channelPairings() { return channelPairings; } + /** Creates an independently closeable client, retaining unspecified resolved options. */ + public ForwardClient withOptions(Consumer configure) { + Builder copy = configuration.copy(); + Builder.required(configure).accept(copy); + return copy.build(); + } + + /** Calls an additional JSON API using the client's authentication, retries and error handling. */ + public ApiResponse request(ApiRequest request) { + return request(request, RequestOptions.DEFAULT); + } + + public ApiResponse request(ApiRequest request, RequestOptions options) { + return Futures.await(async.request(request, options)); + } + @Override public void close() { engine.close(); @@ -170,6 +197,10 @@ public static final class Builder { private String baseUrl, pat, accessToken; private AuthMode authMode; private Credential credential; + private Credential resolvedCredential; + private HttpTransport transport; + private final List interceptors = new ArrayList<>(); + private boolean derived, modeChanged; private Duration timeout = Duration.ofMinutes(10); private int maxRetries = 2; private final Map headers = new LinkedHashMap<>(); @@ -181,20 +212,35 @@ public Builder baseUrl(String value) { public Builder authMode(AuthMode value) { authMode = required(value); + modeChanged = true; return this; } public Builder pat(String value) { + if (derived) { + credential = null; + accessToken = null; + if (!modeChanged) authMode = AuthMode.PAT; + } pat = Authentication.validateToken(value); return this; } public Builder accessToken(String value) { + if (derived) { + credential = null; + pat = null; + } accessToken = Authentication.validateToken(value); return this; } public Builder credential(Credential value) { + if (derived) { + pat = null; + accessToken = null; + if (!modeChanged) authMode = null; + } credential = required(value); return this; } @@ -224,7 +270,45 @@ public Builder defaultHeader(String name, String value) { return this; } - private ClientEngine engine() { + /** Caller-owned transport, also used for isolated storage downloads; not WebSockets. */ + public Builder transport(HttpTransport value) { + transport = required(value); + return this; + } + + /** Appends an API interceptor; storage downloads do not invoke it. */ + public Builder interceptor(HttpInterceptor value) { + interceptors.add(required(value)); + return this; + } + + public Builder clearInterceptors() { + interceptors.clear(); + return this; + } + + public Builder clearDefaultHeaders() { + headers.clear(); + return this; + } + + Builder copy() { + Builder copy = new Builder(); + copy.baseUrl = baseUrl; + copy.pat = pat; + copy.accessToken = accessToken; + copy.authMode = authMode; + copy.credential = credential; + copy.timeout = timeout; + copy.maxRetries = maxRetries; + copy.transport = transport; + copy.headers.putAll(headers); + copy.interceptors.addAll(interceptors); + copy.derived = derived; + return copy; + } + + private Builder resolve() { String value = baseUrl; if (value == null) value = System.getenv("QODER_FORWARD_BASE_URL"); if (value == null || value.isEmpty()) value = "https://api.qoder.com/api/v1/forward"; @@ -242,16 +326,40 @@ private ClientEngine engine() { Credential resolved = Authentication.resolve( authMode, pat, accessToken, credential, url.host(), System::getenv); - return new ClientEngine(url, resolved, timeout, maxRetries, headers); + Builder snapshot = copy(); + snapshot.baseUrl = url.toString(); + snapshot.authMode = + authMode != null + ? authMode + : credential instanceof com.qoder.cloudagents.auth.ServiceAccountCredential + ? AuthMode.SA + : AuthMode.PAT; + snapshot.resolvedCredential = resolved; + if (pat == null && accessToken == null) snapshot.credential = resolved; + snapshot.derived = true; + return snapshot; + } + + private ClientEngine engine() { + return new ClientEngine( + HttpUrl.get(baseUrl), + resolvedCredential, + timeout, + maxRetries, + headers, + transport, + interceptors); } public ForwardClient build() { - return new ForwardClient(engine()); + Builder snapshot = resolve(); + return new ForwardClient(snapshot.engine(), snapshot); } /** Builds an independently owned async client with the same configuration options. */ public AsyncForwardClient buildAsync() { - return new AsyncForwardClient(engine(), true); + Builder snapshot = resolve(); + return new AsyncForwardClient(snapshot.engine(), true, snapshot); } } } diff --git a/src/main/java/com/qoder/cloudagents/core/ApiRequest.java b/src/main/java/com/qoder/cloudagents/core/ApiRequest.java new file mode 100644 index 0000000..9426234 --- /dev/null +++ b/src/main/java/com/qoder/cloudagents/core/ApiRequest.java @@ -0,0 +1,76 @@ +package com.qoder.cloudagents.core; + +import com.qoder.cloudagents.errors.ConfigurationException; +import java.util.Arrays; +import java.util.Locale; + +/** A JSON request relative to the client's API base URL. Paths are unencoded segments. */ +public final class ApiRequest { + private final String method; + private final String[] path; + private final JsonValue body; + + private ApiRequest(Builder builder) { + method = builder.method; + path = builder.path.clone(); + body = builder.body; + } + + /** For example, builder("POST", "sessions", sessionId, "new-action"). */ + public static Builder builder(String method, String... path) { + return new Builder(method, path); + } + + public String method() { + return method; + } + + public String[] path() { + return path.clone(); + } + + public JsonValue body() { + return body; + } + + public static final class Builder { + private final String method; + private final String[] path; + private JsonValue body; + + private Builder(String method, String[] path) { + if (method == null + || !Arrays.asList("GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS") + .contains(method.toUpperCase(Locale.ROOT))) + throw new ConfigurationException("Unsupported HTTP method"); + this.method = method.toUpperCase(Locale.ROOT); + if (path == null || path.length == 0) + throw new ConfigurationException("Supply relative API path segments"); + this.path = path.clone(); + for (String part : this.path) + if (part == null + || part.isEmpty() + || part.equals(".") + || part.equals("..") + || part.indexOf(':') >= 0 + || part.indexOf('/') >= 0 + || part.indexOf('\\') >= 0 + || part.indexOf('?') >= 0 + || part.indexOf('#') >= 0) + throw new ConfigurationException("Supply unencoded relative path segments, not a URL"); + } + + public Builder body(JsonValue value) { + if (method.equals("GET") || method.equals("HEAD")) + throw new ConfigurationException("GET and HEAD requests cannot have a body"); + if (value == null || value.isMissing()) + throw new ConfigurationException("Use JsonValue.of(null) for an explicit JSON null"); + body = value; + return this; + } + + public ApiRequest build() { + return new ApiRequest(this); + } + } +} diff --git a/src/main/java/com/qoder/cloudagents/core/http/HttpInterceptor.java b/src/main/java/com/qoder/cloudagents/core/http/HttpInterceptor.java new file mode 100644 index 0000000..9117143 --- /dev/null +++ b/src/main/java/com/qoder/cloudagents/core/http/HttpInterceptor.java @@ -0,0 +1,18 @@ +package com.qoder.cloudagents.core.http; + +import java.util.concurrent.CompletableFuture; + +/** + * Wraps each authenticated API attempt, including SSE handshakes and SAT management. Does not + * intercept storage downloads, WebSockets or SA Key exchange. Return promptly; never log secrets or + * consume a streaming body for logging. Interceptors run in registration order. + */ +@FunctionalInterface +public interface HttpInterceptor { + CompletableFuture intercept(HttpRequest request, Chain chain); + + interface Chain { + /** Proceed at most once, optionally adding headers with HttpRequest.withHeader. */ + CompletableFuture proceed(HttpRequest request); + } +} diff --git a/src/main/java/com/qoder/cloudagents/core/http/HttpRequest.java b/src/main/java/com/qoder/cloudagents/core/http/HttpRequest.java new file mode 100644 index 0000000..8482cde --- /dev/null +++ b/src/main/java/com/qoder/cloudagents/core/http/HttpRequest.java @@ -0,0 +1,104 @@ +package com.qoder.cloudagents.core.http; + +import com.qoder.cloudagents.core.ApiResponse; +import com.qoder.cloudagents.core.RequestOptions; +import com.qoder.cloudagents.errors.ConfigurationException; +import java.io.IOException; +import java.io.OutputStream; +import java.net.URI; +import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** Immutable single-attempt request. Headers and URL may contain secrets; toString is redacted. */ +public final class HttpRequest { + private final String method; + private final URI uri; + private final Map> headers; + private final Body body; + private final Duration timeout; + private final boolean eventStream; + + public HttpRequest( + String method, + URI uri, + Map> headers, + Body body, + Duration timeout, + boolean eventStream) { + if (method == null || uri == null || headers == null) + throw new ConfigurationException("HTTP request metadata must not be null"); + RequestOptions.positiveMillis(timeout); + this.method = method; + this.uri = uri; + this.headers = new ApiResponse(null, 0, headers).headers(); + this.body = body; + this.timeout = timeout; + this.eventStream = eventStream; + } + + public String method() { + return method; + } + + public URI uri() { + return uri; + } + + public Map> headers() { + return headers; + } + + public String header(String name) { + List values = headers.get(name.toLowerCase(java.util.Locale.ROOT)); + return values == null || values.isEmpty() ? null : values.get(0); + } + + public Body body() { + return body; + } + + public Duration connectTimeout() { + return Duration.ofSeconds(5); + } + + /** Zero disables read-idle timeout for SSE; header and total deadlines remain SDK-owned. */ + public Duration readTimeout() { + return eventStream ? Duration.ZERO : timeout; + } + + public Duration writeTimeout() { + return timeout; + } + + /** Adds or replaces an observability header without modifying routing or authentication. */ + public HttpRequest withHeader(String name, String value) { + RequestOptions.builder().header(name, value); + if (name.equalsIgnoreCase("Host") + || name.equalsIgnoreCase("Idempotency-Key") + || name.equalsIgnoreCase("Content-Length") + || name.equalsIgnoreCase("Transfer-Encoding")) + throw new ConfigurationException("This header is controlled by the request or transport"); + Map> copy = new LinkedHashMap<>(headers); + copy.put(name.toLowerCase(java.util.Locale.ROOT), Collections.singletonList(value)); + return new HttpRequest(method, uri, copy, body, timeout, eventStream); + } + + @Override + public String toString() { + return "HttpRequest([redacted])"; + } + + /** Stream a body without buffering uploads; do not close the supplied destination. */ + public interface Body { + String contentType(); + + long contentLength() throws IOException; + + boolean replayable(); + + void writeTo(OutputStream destination) throws IOException; + } +} diff --git a/src/main/java/com/qoder/cloudagents/core/http/HttpResponse.java b/src/main/java/com/qoder/cloudagents/core/http/HttpResponse.java new file mode 100644 index 0000000..ea49848 --- /dev/null +++ b/src/main/java/com/qoder/cloudagents/core/http/HttpResponse.java @@ -0,0 +1,50 @@ +package com.qoder.cloudagents.core.http; + +import com.qoder.cloudagents.core.ApiResponse; +import com.qoder.cloudagents.errors.ConfigurationException; +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +/** A response whose streaming body is transferred to the SDK. Closing is idempotent. */ +public final class HttpResponse implements Closeable { + private final ApiResponse metadata; + private final InputStream body; + private final AtomicBoolean closed = new AtomicBoolean(); + + public HttpResponse(int statusCode, Map> headers, InputStream body) { + if (statusCode < 100 || statusCode > 599 || headers == null || body == null) + throw new ConfigurationException("Invalid HTTP response"); + metadata = new ApiResponse<>(null, statusCode, headers); + this.body = body; + } + + public int statusCode() { + return metadata.statusCode(); + } + + public Map> headers() { + return metadata.headers(); + } + + public String header(String name) { + return metadata.header(name); + } + + public InputStream body() { + return body; + } + + @Override + public void close() throws IOException { + if (closed.compareAndSet(false, true)) body.close(); + } + + @Override + public String toString() { + return "HttpResponse(" + statusCode() + ", [redacted])"; + } +} diff --git a/src/main/java/com/qoder/cloudagents/core/http/HttpTransport.java b/src/main/java/com/qoder/cloudagents/core/http/HttpTransport.java new file mode 100644 index 0000000..ad82d66 --- /dev/null +++ b/src/main/java/com/qoder/cloudagents/core/http/HttpTransport.java @@ -0,0 +1,15 @@ +package com.qoder.cloudagents.core.http; + +import java.util.concurrent.CompletableFuture; + +/** + * Caller-owned HTTP transport for API requests and isolated storage downloads. Implementations must + * be thread-safe, return promptly, honor request timeouts, and abort I/O when the returned future + * is cancelled. Perform exactly one attempt without redirects, authentication or retries. The SDK + * owns response bodies, but never closes this transport. WebSockets and SA Key exchange use their + * own transports. + */ +@FunctionalInterface +public interface HttpTransport { + CompletableFuture execute(HttpRequest request); +} diff --git a/src/main/java/com/qoder/cloudagents/core/internal/ClientEngine.java b/src/main/java/com/qoder/cloudagents/core/internal/ClientEngine.java index 6f4ae04..d919458 100644 --- a/src/main/java/com/qoder/cloudagents/core/internal/ClientEngine.java +++ b/src/main/java/com/qoder/cloudagents/core/internal/ClientEngine.java @@ -5,6 +5,8 @@ import com.qoder.cloudagents.core.FileDownload; import com.qoder.cloudagents.core.JsonValue; import com.qoder.cloudagents.core.RequestOptions; +import com.qoder.cloudagents.core.http.HttpInterceptor; +import com.qoder.cloudagents.core.http.HttpTransport; import com.qoder.cloudagents.errors.*; import java.io.IOException; import java.io.InputStream; @@ -36,12 +38,14 @@ public final class ClientEngine implements AutoCloseable { private final int maxRetries; private final Map defaultHeaders; private final OkHttpClient transport = Transports.create(); + private final HttpTransport customTransport; + private final List interceptors; private final ScheduledThreadPoolExecutor timer = new ScheduledThreadPoolExecutor(1, Transports.threads("qca-timer")); private final Set> active = ConcurrentHashMap.newKeySet(); private final Set> streams = ConcurrentHashMap.newKeySet(); private volatile boolean closed; - private final FileDownloads downloads = new FileDownloads(this, transport, timer); + private final FileDownloads downloads = new FileDownloads(this, timer); public ClientEngine( HttpUrl baseUrl, @@ -49,6 +53,19 @@ public ClientEngine( Duration timeout, int maxRetries, Map defaultHeaders) { + this(baseUrl, credential, timeout, maxRetries, defaultHeaders, null, Collections.emptyList()); + } + + public ClientEngine( + HttpUrl baseUrl, + Credential credential, + Duration timeout, + int maxRetries, + Map defaultHeaders, + HttpTransport customTransport, + List interceptors) { + this.customTransport = customTransport; + this.interceptors = Collections.unmodifiableList(new ArrayList<>(interceptors)); this.baseUrl = baseUrl; this.credential = credential; timeoutMillis = RequestOptions.positiveMillis(timeout); @@ -242,11 +259,12 @@ private CompletableFuture> launch(Operation operation) { } private static JsonValue decodeJson(Response response) throws IOException { + if (response.code() == 204 + || response.code() == 205 + || response.request().method().equals("HEAD")) return JsonValue.of(null); String text = response.body() == null ? "" : response.body().string(); try { - return response.code() == 204 || response.code() == 205 - ? JsonValue.of(null) - : JsonValue.parse(text); + return JsonValue.parse(text); } catch (ResponseValidationException error) { if (response.isSuccessful()) throw error; return JsonValue.of(text); @@ -272,6 +290,11 @@ public void close() { Transports.close(transport); } + Call storageCall(Request request, long timeout) { + return HttpExtensions.create( + transport, customTransport, Collections.emptyList(), request, timeout, false, true); + } + private final class Operation { final CompletableFuture> result = new CompletableFuture<>(); final String method; @@ -417,12 +440,8 @@ void send(String token, int number) { Request request = builder.build(); AtomicBoolean headersFinished = new AtomicBoolean(); final Call attemptCall = - transport - .newBuilder() - .readTimeout(sse ? 0 : timeout, TimeUnit.MILLISECONDS) - .writeTimeout(timeout, TimeUnit.MILLISECONDS) - .build() - .newCall(request); + HttpExtensions.create( + transport, customTransport, interceptors, request, timeout, sse, replayable); synchronized (this) { if (result.isDone()) { attemptCall.cancel(); diff --git a/src/main/java/com/qoder/cloudagents/core/internal/FileDownloads.java b/src/main/java/com/qoder/cloudagents/core/internal/FileDownloads.java index a1d66d0..690f59e 100644 --- a/src/main/java/com/qoder/cloudagents/core/internal/FileDownloads.java +++ b/src/main/java/com/qoder/cloudagents/core/internal/FileDownloads.java @@ -12,14 +12,12 @@ /** Two-hop download lifecycle; the storage hop never resolves credentials or copies API headers. */ public final class FileDownloads implements AutoCloseable { private final ClientEngine engine; - private final OkHttpClient transport; private final ScheduledThreadPoolExecutor timer; private final Set active = ConcurrentHashMap.newKeySet(); private volatile boolean closed; - FileDownloads(ClientEngine engine, OkHttpClient transport, ScheduledThreadPoolExecutor timer) { + FileDownloads(ClientEngine engine, ScheduledThreadPoolExecutor timer) { this.engine = engine; - this.transport = transport; this.timer = timer; } @@ -108,13 +106,7 @@ synchronized void send(HttpUrl url, int attempt, int redirects) { try { Request request = new Request.Builder().url(url).get().build(); AtomicBoolean headersFinished = new AtomicBoolean(); - Call sending = - transport - .newBuilder() - .readTimeout(timeout, TimeUnit.MILLISECONDS) - .writeTimeout(timeout, TimeUnit.MILLISECONDS) - .build() - .newCall(request); + Call sending = engine.storageCall(request, timeout); call = sending; headerWait = timer.schedule( diff --git a/src/main/java/com/qoder/cloudagents/core/internal/HttpExtensions.java b/src/main/java/com/qoder/cloudagents/core/internal/HttpExtensions.java new file mode 100644 index 0000000..d021f29 --- /dev/null +++ b/src/main/java/com/qoder/cloudagents/core/internal/HttpExtensions.java @@ -0,0 +1,383 @@ +package com.qoder.cloudagents.core.internal; + +import com.qoder.cloudagents.core.http.HttpInterceptor; +import com.qoder.cloudagents.core.http.HttpRequest; +import com.qoder.cloudagents.core.http.HttpResponse; +import com.qoder.cloudagents.core.http.HttpTransport; +import java.io.*; +import java.time.Duration; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; +import okhttp3.*; +import okio.*; + +/** Bridges public transport contracts to the existing SDK retry and streaming lifecycle. */ +final class HttpExtensions implements Call { + private final OkHttpClient client; + private final HttpTransport transport; + private final List interceptors; + private final Request request; + private final Call metadata; + private final long timeout; + private final boolean sse, replayable; + private final List> pending = new ArrayList<>(); + private final Set responses = Collections.newSetFromMap(new IdentityHashMap<>()); + private boolean executed, cancelled; + + static Call create( + OkHttpClient client, + HttpTransport transport, + List interceptors, + Request request, + long timeout, + boolean sse, + boolean replayable) { + OkHttpClient configured = + client + .newBuilder() + .readTimeout(sse ? 0 : timeout, TimeUnit.MILLISECONDS) + .writeTimeout(timeout, TimeUnit.MILLISECONDS) + .build(); + if (transport == null && interceptors.isEmpty()) return configured.newCall(request); + return new HttpExtensions( + configured, transport, interceptors, request, timeout, sse, replayable); + } + + private HttpExtensions( + OkHttpClient client, + HttpTransport transport, + List interceptors, + Request request, + long timeout, + boolean sse, + boolean replayable) { + this.client = client; + this.transport = transport; + this.interceptors = interceptors; + this.request = request; + this.metadata = client.newCall(request); + this.timeout = timeout; + this.sse = sse; + this.replayable = replayable; + } + + @Override + public Request request() { + return request; + } + + @Override + public synchronized boolean isExecuted() { + return executed; + } + + @Override + public synchronized boolean isCanceled() { + return cancelled; + } + + @Override + public Timeout timeout() { + return Timeout.NONE; + } + + @Override + public void addEventListener(okhttp3.EventListener listener) { + metadata.addEventListener(listener); + } + + @Override + public T tag(Class type) { + return metadata.tag(type); + } + + @Override + public T tag(kotlin.reflect.KClass type) { + return metadata.tag(type); + } + + @Override + public T tag(Class type, kotlin.jvm.functions.Function0 factory) { + return metadata.tag(type, factory); + } + + @Override + public T tag( + kotlin.reflect.KClass type, kotlin.jvm.functions.Function0 factory) { + return metadata.tag(type, factory); + } + + @Override + public Call clone() { + return new HttpExtensions(client, transport, interceptors, request, timeout, sse, replayable); + } + + @Override + public Response execute() { + throw new UnsupportedOperationException("SDK operations use asynchronous transport calls"); + } + + @Override + public void enqueue(Callback callback) { + synchronized (this) { + if (executed) throw new IllegalStateException("Already executed"); + executed = true; + } + try { + client + .dispatcher() + .executorService() + .execute( + () -> { + try { + HttpRequest original = adaptRequest(); + CompletableFuture result = invoke(0, original, original); + result.whenComplete( + (value, error) -> { + if (error != null) { + cancel(); + callback.onFailure(this, ioFailure(error)); + return; + } + if (isCanceled()) { + close(value); + callback.onFailure(this, new IOException("HTTP call cancelled")); + return; + } + Response response = null; + try { + response = adaptResponse(value); + callback.onResponse(this, response); + } catch (IOException | RuntimeException failure) { + if (response != null) response.close(); + else close(value); + cancel(); + callback.onFailure(this, ioFailure(failure)); + } + }); + } catch (RuntimeException failure) { + cancel(); + callback.onFailure(this, ioFailure(failure)); + } + }); + } catch (RejectedExecutionException failure) { + cancel(); + callback.onFailure(this, ioFailure(failure)); + } + } + + private CompletableFuture invoke( + int index, HttpRequest next, HttpRequest original) { + if (isCanceled()) return failed(new CancellationException()); + validate(next, original); + CompletableFuture result; + if (index == interceptors.size()) + result = transport == null ? send(next) : transport.execute(next); + else { + AtomicBoolean proceeded = new AtomicBoolean(); + result = + interceptors + .get(index) + .intercept( + next, + modified -> { + if (!proceeded.compareAndSet(false, true)) + throw new IllegalStateException( + "An interceptor may proceed only once per attempt"); + return invoke(index + 1, modified, original); + }); + } + if (result == null) throw new IllegalStateException("HTTP extension returned no future"); + track(result); + CompletableFuture checked = + result.thenApply( + response -> { + if (response == null) + throw new IllegalStateException("HTTP extension returned no response"); + return response; + }); + track(checked); + return checked; + } + + private synchronized void track(CompletableFuture future) { + pending.add(future); + future.whenComplete( + (response, error) -> { + if (response != null) { + boolean release; + synchronized (HttpExtensions.this) { + release = cancelled; + if (!release) responses.add(response); + } + if (release) close(response); + } + }); + if (cancelled) future.cancel(true); + } + + private void validate(HttpRequest value, HttpRequest original) { + if (value == null + || !value.uri().equals(original.uri()) + || !value.method().equals(original.method()) + || value.body() != original.body() + || !value.readTimeout().equals(original.readTimeout()) + || !value.writeTimeout().equals(original.writeTimeout())) + throw new IllegalArgumentException( + "Interceptors may change headers, not routing, body or timeouts"); + for (String header : + Arrays.asList( + "Authorization", "Host", "Idempotency-Key", "Content-Length", "Transfer-Encoding")) + if (!Objects.equals( + value.headers().get(header.toLowerCase(Locale.ROOT)), + original.headers().get(header.toLowerCase(Locale.ROOT)))) + throw new IllegalArgumentException("Interceptor changed a protected header"); + } + + private HttpRequest adaptRequest() { + RequestBody source = request.body(); + HttpRequest.Body body = + source == null + ? null + : new HttpRequest.Body() { + public String contentType() { + return source.contentType() == null ? null : source.contentType().toString(); + } + + public long contentLength() throws IOException { + return source.contentLength(); + } + + public boolean replayable() { + return replayable; + } + + public void writeTo(OutputStream destination) throws IOException { + BufferedSink sink = Okio.buffer(Okio.sink(destination)); + source.writeTo(sink); + sink.flush(); + } + }; + return new HttpRequest( + request.method(), + request.url().uri(), + request.headers().toMultimap(), + body, + Duration.ofMillis(timeout), + sse); + } + + private CompletableFuture send(HttpRequest value) { + Headers.Builder headers = new Headers.Builder(); + value.headers().forEach((name, values) -> values.forEach(v -> headers.add(name, v))); + Call call = client.newCall(request.newBuilder().headers(headers.build()).build()); + CompletableFuture future = new CompletableFuture<>(); + future.whenComplete( + (response, error) -> { + if (future.isCancelled()) call.cancel(); + }); + call.enqueue( + new Callback() { + public void onFailure(Call ignored, IOException error) { + future.completeExceptionally(error); + } + + public void onResponse(Call ignored, Response response) { + InputStream stream = + response.body() == null + ? new ByteArrayInputStream(new byte[0]) + : response.body().byteStream(); + HttpResponse converted = + new HttpResponse(response.code(), response.headers().toMultimap(), stream); + if (!future.complete(converted)) { + close(converted); + response.close(); + } + } + }); + return future; + } + + private Response adaptResponse(HttpResponse value) { + Headers.Builder headers = new Headers.Builder(); + value.headers().forEach((name, values) -> values.forEach(v -> headers.add(name, v))); + InputStream stream = + new FilterInputStream(value.body()) { + @Override + public void close() throws IOException { + try { + value.close(); + } finally { + // Also release downstream responses replaced by an interceptor, including retries. + HttpExtensions.this.cancel(); + } + } + }; + ResponseBody body = + new ResponseBody() { + private final BufferedSource source = Okio.buffer(Okio.source(stream)); + + public MediaType contentType() { + String type = value.header("Content-Type"); + return type == null ? null : MediaType.parse(type); + } + + public long contentLength() { + try { + return Long.parseLong(value.header("Content-Length")); + } catch (NumberFormatException ignored) { + return -1; + } + } + + public BufferedSource source() { + return source; + } + }; + return new Response.Builder() + .request(request) + .protocol(Protocol.HTTP_1_1) + .code(value.statusCode()) + .message("") + .headers(headers.build()) + .body(body) + .build(); + } + + @Override + public void cancel() { + List> futures; + List bodies; + synchronized (this) { + if (cancelled) return; + cancelled = true; + futures = new ArrayList<>(pending); + bodies = new ArrayList<>(responses); + responses.clear(); + } + futures.forEach(future -> future.cancel(true)); + bodies.forEach(HttpExtensions::close); + } + + private static IOException ioFailure(Throwable error) { + while ((error instanceof CompletionException || error instanceof ExecutionException) + && error.getCause() != null) error = error.getCause(); + return error instanceof IOException + ? (IOException) error + : new IOException("HTTP extension failed", error); + } + + private static CompletableFuture failed(Throwable error) { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(error); + return future; + } + + private static void close(HttpResponse response) { + try { + response.close(); + } catch (IOException ignored) { + } + } +} diff --git a/src/main/java/com/qoder/cloudagents/forward/ChannelPairing.java b/src/main/java/com/qoder/cloudagents/forward/ChannelPairing.java index 5bbf2de..bcfe02f 100644 --- a/src/main/java/com/qoder/cloudagents/forward/ChannelPairing.java +++ b/src/main/java/com/qoder/cloudagents/forward/ChannelPairing.java @@ -9,6 +9,14 @@ public ChannelPairing(JsonValue value, String requestId) { super(value, requestId); } + public String createdAt() { + return optional("created_at", JsonValue::asString); + } + + public String updatedAt() { + return optional("updated_at", JsonValue::asString); + } + public String id() { return optional("id", JsonValue::asString); } diff --git a/src/main/java/com/qoder/cloudagents/forward/ImageSource.java b/src/main/java/com/qoder/cloudagents/forward/ImageSource.java index 1a58e52..eb44600 100644 --- a/src/main/java/com/qoder/cloudagents/forward/ImageSource.java +++ b/src/main/java/com/qoder/cloudagents/forward/ImageSource.java @@ -9,6 +9,10 @@ private ImageSource(JsonValue value) { super(value); } + public String fileId() { + return optional("file_id", JsonValue::asString); + } + public static ImageSource fromJson(JsonValue value) { return new ImageSource(value); } @@ -52,6 +56,10 @@ public Builder url(String value) { return put("url", value); } + public Builder fileId(String value) { + return put("file_id", value); + } + public ImageSource build() { require("type"); return new ImageSource(json()); diff --git a/src/main/java/com/qoder/cloudagents/forward/MemoryStoreCreateMount.java b/src/main/java/com/qoder/cloudagents/forward/MemoryStoreCreateMount.java index 803e20e..7bdfde4 100644 --- a/src/main/java/com/qoder/cloudagents/forward/MemoryStoreCreateMount.java +++ b/src/main/java/com/qoder/cloudagents/forward/MemoryStoreCreateMount.java @@ -8,6 +8,10 @@ private MemoryStoreCreateMount(JsonValue value) { super(value); } + public String access() { + return optional("access", JsonValue::asString); + } + public static MemoryStoreCreateMount fromJson(JsonValue value) { return new MemoryStoreCreateMount(value); } @@ -35,6 +39,10 @@ public Builder templateId(String value) { return put("template_id", value); } + public Builder access(String value) { + return put("access", value); + } + public MemoryStoreCreateMount build() { require("identity_id"); require("template_id"); diff --git a/src/main/java/com/qoder/cloudagents/forward/Schedule.java b/src/main/java/com/qoder/cloudagents/forward/Schedule.java index ca25695..a02d0ad 100644 --- a/src/main/java/com/qoder/cloudagents/forward/Schedule.java +++ b/src/main/java/com/qoder/cloudagents/forward/Schedule.java @@ -9,6 +9,10 @@ public Schedule(JsonValue value, String requestId) { super(value, requestId); } + public Long initialEventsVersion() { + return optional("initial_events_version", JsonValue::asLong); + } + public String id() { return optional("id", JsonValue::asString); } diff --git a/src/main/java/com/qoder/cloudagents/forward/ScheduleRun.java b/src/main/java/com/qoder/cloudagents/forward/ScheduleRun.java index 9bed123..240515f 100644 --- a/src/main/java/com/qoder/cloudagents/forward/ScheduleRun.java +++ b/src/main/java/com/qoder/cloudagents/forward/ScheduleRun.java @@ -9,6 +9,10 @@ public ScheduleRun(JsonValue value, String requestId) { super(value, requestId); } + public Long initialEventsVersion() { + return optional("initial_events_version", JsonValue::asLong); + } + public String id() { return optional("id", JsonValue::asString); } diff --git a/src/test/java/com/qoder/cloudagents/ClientExtensionsTest.java b/src/test/java/com/qoder/cloudagents/ClientExtensionsTest.java new file mode 100644 index 0000000..8ca7555 --- /dev/null +++ b/src/test/java/com/qoder/cloudagents/ClientExtensionsTest.java @@ -0,0 +1,631 @@ +package com.qoder.cloudagents; + +import static org.junit.jupiter.api.Assertions.*; + +import com.qoder.cloudagents.auth.AuthMode; +import com.qoder.cloudagents.core.*; +import com.qoder.cloudagents.core.http.*; +import com.qoder.cloudagents.errors.*; +import com.qoder.cloudagents.forward.*; +import com.qoder.cloudagents.streaming.EventStream; +import java.io.*; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import okhttp3.mockwebserver.*; +import org.junit.jupiter.api.*; + +@Timeout(15) +class ClientExtensionsTest { + private MockWebServer server; + + @BeforeEach + void start() throws Exception { + server = new MockWebServer(); + server.start(); + } + + @AfterEach + void stop() throws Exception { + server.close(); + } + + ForwardClient.Builder builder() { + return ForwardClient.builder() + .baseUrl(server.url("/api/v1/forward").toString()) + .pat("test-pat") + .maxRetries(0); + } + + static HttpResponse response(int code, String body) { + return new HttpResponse( + code, + Collections.singletonMap("content-type", Collections.singletonList("application/json")), + new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8))); + } + + static CompletableFuture json(String body) { + return CompletableFuture.completedFuture(response(200, body)); + } + + static ApiRequest get() { + return ApiRequest.builder("GET", "models").build(); + } + + @Test + void genericJsonUsesAuthOptionsMetadataAndAsyncErrorMapping() throws Exception { + server.enqueue( + new MockResponse() + .setBody("{\"value\":9007199254740993}") + .setHeader("X-Request-ID", "req_generic")); + server.enqueue( + new MockResponse() + .setResponseCode(403) + .setBody("{\"error\":{\"code\":\"denied\",\"message\":\"No\"}}")); + try (ForwardClient client = builder().build()) { + ApiResponse reply = + client.request( + ApiRequest.builder("POST", "new-resource", "id +") + .body(JsonValue.parse("{\"enabled\":false,\"limit\":0,\"clear\":null}")) + .build(), + RequestOptions.builder() + .idempotencyKey("generic-key") + .query("filter", "a+b") + .build()); + assertEquals(9007199254740993L, reply.data().get("value").asLong()); + assertEquals("req_generic", reply.requestId()); + RecordedRequest wire = server.takeRequest(); + assertEquals("Bearer test-pat", wire.getHeader("Authorization")); + assertEquals("generic-key", wire.getHeader("Idempotency-Key")); + assertEquals("/api/v1/forward/new-resource/id%20+", wire.getRequestUrl().encodedPath()); + assertEquals("a+b", wire.getRequestUrl().queryParameter("filter")); + assertTrue(JsonValue.parse(wire.getBody().readUtf8()).get("clear").isNull()); + ExecutionException error = + assertThrows( + ExecutionException.class, + () -> client.async().request(get()).get(3, TimeUnit.SECONDS)); + assertEquals(403, ((ApiStatusException) error.getCause()).statusCode()); + } + } + + @Test + void genericPathsCannotEscapeTheBaseAndHeadAllowsAnEmptyBody() { + for (String path : + Arrays.asList("https://other.test", "//other.test", "..", ".", "a/b", "a?b", "a#b", "a\\b")) + assertThrows(ConfigurationException.class, () -> ApiRequest.builder("GET", path)); + assertThrows( + ConfigurationException.class, + () -> ApiRequest.builder("GET", "x").body(JsonValue.of(null))); + server.enqueue(new MockResponse().setResponseCode(200)); + try (ForwardClient client = builder().build()) { + assertTrue(client.request(ApiRequest.builder("HEAD", "models").build()).data().isNull()); + } + } + + @Test + void genericRequestsPreserveSafeRetryRules() { + AtomicInteger calls = new AtomicInteger(); + HttpTransport failing = + request -> { + calls.incrementAndGet(); + return CompletableFuture.completedFuture(response(503, "{}")); + }; + try (ForwardClient client = builder().transport(failing).maxRetries(1).build()) { + assertThrows( + ApiStatusException.class, + () -> client.request(ApiRequest.builder("POST", "actions").build())); + assertEquals(1, calls.get()); + assertThrows( + ApiStatusException.class, + () -> + client.request( + ApiRequest.builder("POST", "actions").build(), + RequestOptions.builder().idempotencyKey("safe-to-retry").build())); + assertEquals(3, calls.get()); + assertThrows(ApiStatusException.class, () -> client.request(get())); + assertEquals(5, calls.get()); + } + } + + @Test + void interceptorsWrapEachDefaultTransportAttemptInOrder() throws Exception { + server.enqueue( + new MockResponse().setResponseCode(503).setBody("{}").setHeader("Retry-After-Ms", "1")); + server.enqueue(new MockResponse().setBody("{\"data\":[]}")); + List seen = new CopyOnWriteArrayList<>(); + HttpInterceptor first = + (request, chain) -> { + seen.add("first:" + request.header("X-Qoder-Retry-Count")); + return chain.proceed(request.withHeader("X-Trace", "trace-1")); + }; + HttpInterceptor second = + (request, chain) -> { + seen.add("second:" + request.header("X-Trace")); + return chain.proceed(request); + }; + try (ForwardClient client = + builder().maxRetries(1).interceptor(first).interceptor(second).build()) { + client.models().list(); + assertEquals(Arrays.asList("first:0", "second:trace-1", "first:1", "second:trace-1"), seen); + assertEquals("Bearer test-pat", server.takeRequest().getHeader("Authorization")); + assertEquals("trace-1", server.takeRequest().getHeader("X-Trace")); + } + } + + @Test + void interceptorCannotRewriteRoutingOrReplayOneAttempt() { + AtomicInteger calls = new AtomicInteger(); + HttpTransport transport = + request -> { + calls.incrementAndGet(); + return json("{}"); + }; + try (ForwardClient client = + builder() + .transport(transport) + .interceptor( + (request, chain) -> + chain.proceed( + new HttpRequest( + request.method(), + java.net.URI.create("https://other.test"), + request.headers(), + request.body(), + request.writeTimeout(), + false))) + .build()) { + assertThrows(ApiConnectionException.class, () -> client.request(get())); + assertEquals(0, calls.get()); + } + try (ForwardClient client = + builder() + .transport(transport) + .interceptor( + (request, chain) -> { + chain.proceed(request); + return chain.proceed(request); + }) + .build()) { + assertThrows(ApiConnectionException.class, () -> client.request(get())); + assertEquals(1, calls.get()); + } + } + + @Test + void customTransportGetsStreamingBodyAndResolvedCredentialOnEachAttempt() throws Exception { + AtomicInteger tokens = new AtomicInteger(); + List seen = new CopyOnWriteArrayList<>(); + HttpTransport transport = + request -> { + seen.add(request); + return json("{\"id\":\"tmpl_example\"}"); + }; + try (ForwardClient client = + ForwardClient.builder() + .credential( + () -> CompletableFuture.completedFuture("token-" + tokens.incrementAndGet())) + .transport(transport) + .build()) { + client + .templates() + .create( + TemplateCreateParams.builder() + .name("template") + .modelId("auto") + .environmentId("env_example") + .build()); + client.async().templates().retrieve("tmpl_example").get(3, TimeUnit.SECONDS); + assertEquals("Bearer token-1", seen.get(0).header("Authorization")); + assertEquals("Bearer token-2", seen.get(1).header("Authorization")); + assertEquals(Duration.ofSeconds(5), seen.get(0).connectTimeout()); + assertTrue(seen.get(0).body().replayable()); + ByteArrayOutputStream body = new ByteArrayOutputStream(); + seen.get(0).body().writeTo(body); + assertEquals("template", JsonValue.parse(body.toString("UTF-8")).get("name").asString()); + assertFalse(seen.get(0).toString().contains("token-1")); + } + } + + @Test + void interceptorFailureClosesTheReceivedBodyBeforeRetry() { + assertResponseCleanupBeforeRetry(false); + } + + @Test + void interceptorReplacementClosesBothBodiesBeforeRetry() { + assertResponseCleanupBeforeRetry(true); + } + + private void assertResponseCleanupBeforeRetry(boolean replace) { + AtomicInteger receivedCloses = new AtomicInteger(); + AtomicInteger replacementCloses = new AtomicInteger(); + AtomicInteger attempts = new AtomicInteger(); + AtomicBoolean first = new AtomicBoolean(true); + AtomicBoolean cleanedBeforeRetry = new AtomicBoolean(); + HttpResponse received = closingResponse(200, receivedCloses); + HttpResponse replacement = closingResponse(503, replacementCloses); + try (ForwardClient client = + builder() + .maxRetries(1) + .transport( + request -> { + if (attempts.incrementAndGet() == 1) + return CompletableFuture.completedFuture(received); + cleanedBeforeRetry.set( + receivedCloses.get() == 1 && (!replace || replacementCloses.get() == 1)); + return json("{\"data\":[]}"); + }) + .interceptor( + (request, chain) -> + chain + .proceed(request) + .thenApply( + response -> { + if (first.getAndSet(false)) { + if (replace) return replacement; + throw new IllegalStateException( + "Interceptor failed after receiving headers"); + } + return response; + })) + .build()) { + client.request(get()); + assertEquals(2, attempts.get()); + assertTrue(cleanedBeforeRetry.get()); + assertEquals(1, receivedCloses.get()); + assertEquals(replace ? 1 : 0, replacementCloses.get()); + } + } + + private static HttpResponse closingResponse(int status, AtomicInteger closes) { + return new HttpResponse( + status, + Collections.emptyMap(), + new ByteArrayInputStream("{}".getBytes(StandardCharsets.UTF_8)) { + @Override + public void close() { + closes.incrementAndGet(); + } + }); + } + + @Test + void cancellationPropagatesThroughDerivedInterceptorFutures() throws Exception { + CountDownLatch entered = new CountDownLatch(1); + CompletableFuture network = new CompletableFuture<>(); + try (ForwardClient client = + builder() + .transport( + request -> { + entered.countDown(); + return network; + }) + .interceptor((request, chain) -> chain.proceed(request).thenApply(value -> value)) + .build()) { + CompletableFuture pending = client.async().request(get()); + assertTrue(entered.await(2, TimeUnit.SECONDS)); + pending.cancel(true); + assertTrue(network.isCancelled()); + } + } + + @Test + void headerTimeoutCancelsCustomIo() throws Exception { + CompletableFuture network = new CompletableFuture<>(); + try (ForwardClient client = + builder().transport(request -> network).timeout(Duration.ofMillis(100)).build()) { + assertThrows(ApiTimeoutException.class, () -> client.request(get())); + assertTrue(network.isCancelled()); + } + } + + @Test + void lateCustomResponseAfterCancellationIsClosed() throws Exception { + CountDownLatch entered = new CountDownLatch(1); + AtomicBoolean closed = new AtomicBoolean(); + CompletableFuture network = + new CompletableFuture() { + @Override + public boolean cancel(boolean interrupt) { + return false; + } + }; + try (ForwardClient client = + builder() + .transport( + request -> { + entered.countDown(); + return network; + }) + .build()) { + CompletableFuture pending = client.async().request(get()); + assertTrue(entered.await(2, TimeUnit.SECONDS)); + pending.cancel(true); + network.complete( + new HttpResponse( + 200, + Collections.emptyMap(), + new ByteArrayInputStream(new byte[0]) { + @Override + public void close() { + closed.set(true); + } + })); + assertTrue(closed.get()); + } + } + + @Test + void customTransportPreservesSseAndClosesItsBody() throws Exception { + AtomicBoolean closed = new AtomicBoolean(); + AtomicReference seen = new AtomicReference<>(); + byte[] bytes = + "id: checkpoint\ndata: {\"id\":\"evt_1\",\"type\":\"assistant.message\",\"text\":\"hello\"}\n\n" + .getBytes(StandardCharsets.UTF_8); + try (ForwardClient client = + builder() + .transport( + request -> { + seen.set(request); + return CompletableFuture.completedFuture( + new HttpResponse( + 200, + Collections.singletonMap( + "content-type", Collections.singletonList("text/event-stream")), + new ByteArrayInputStream(bytes) { + @Override + public void close() { + closed.set(true); + } + })); + }) + .build()) { + try (EventStream events = + client.sessions().events().streamEvents("sess_example")) { + assertEquals("hello", events.iterator().next().text()); + assertEquals(Duration.ZERO, seen.get().readTimeout()); + assertEquals("Bearer test-pat", seen.get().header("Authorization")); + } + assertTrue(closed.get()); + } + } + + @Test + void storageDownloadUsesCustomTransportWithoutApiHeadersOrInterceptors() throws Exception { + List requests = new CopyOnWriteArrayList<>(); + AtomicInteger intercepted = new AtomicInteger(); + try (ForwardClient client = + builder() + .defaultHeader("X-Private", "private") + .transport( + request -> { + requests.add(request); + return request.uri().getHost().equals("storage.test") + ? json("file-bytes") + : json("{\"url\":\"https://storage.test/blob?sig=grant\"}"); + }) + .interceptor( + (request, chain) -> { + intercepted.incrementAndGet(); + return chain.proceed(request.withHeader("X-Trace", "trace")); + }) + .build()) { + try (FileDownload download = client.files().download("file_example")) { + assertEquals('f', download.body().read()); + } + assertEquals(2, requests.size()); + assertEquals(1, intercepted.get()); + assertNull(requests.get(1).header("Authorization")); + assertNull(requests.get(1).header("X-Private")); + assertNull(requests.get(1).header("X-Trace")); + } + } + + @Test + void clientCopiesHaveIndependentDefaultsAuthAndLifetimes() throws Exception { + List seen = new CopyOnWriteArrayList<>(); + HttpTransport transport = + request -> { + seen.add(request); + return json("{}"); + }; + ForwardClient original = + builder().defaultHeader("X-Default", "original").transport(transport).build(); + try (ForwardClient copy = + original.withOptions( + options -> + options.timeout(Duration.ofSeconds(2)).defaultHeader("X-Default", "copy")); + AsyncForwardClient other = + original + .async() + .withOptions(options -> options.authMode(AuthMode.SA).accessToken("test-sat"))) { + original.request(get()); + copy.request(get()); + original.close(); + copy.request(get()); + other.request(get()).get(3, TimeUnit.SECONDS); + assertEquals("original", seen.get(0).header("X-Default")); + assertEquals(Duration.ofMinutes(10), seen.get(0).writeTimeout()); + assertEquals("copy", seen.get(1).header("X-Default")); + assertEquals(Duration.ofSeconds(2), seen.get(1).writeTimeout()); + assertEquals("Bearer test-pat", seen.get(2).header("Authorization")); + assertEquals("Bearer test-sat", seen.get(3).header("Authorization")); + } finally { + original.close(); + } + } + + @Test + void closingCopyDoesNotAffectParentOrSharedCredential() { + AtomicInteger lookups = new AtomicInteger(); + try (ForwardClient client = + ForwardClient.builder() + .credential( + () -> { + lookups.incrementAndGet(); + return CompletableFuture.completedFuture("custom-pat"); + }) + .transport(request -> json("{}")) + .build()) { + try (ForwardClient copy = client.withOptions(options -> options.maxRetries(1))) { + copy.request(get()); + } + client.request(get()); + assertEquals(2, lookups.get()); + try (ForwardClient replaced = client.withOptions(options -> options.pat("replacement"))) { + replaced.request(get()); + } + assertEquals(2, lookups.get()); + } + } + + @Test + void copyRetainsExplicitPatSemanticsAndAllowsCollectionReplacement() { + AtomicInteger intercepted = new AtomicInteger(); + List requests = new CopyOnWriteArrayList<>(); + try (ForwardClient parent = + builder() + .defaultHeader("X-Original", "value") + .transport( + request -> { + requests.add(request); + return json("{}"); + }) + .interceptor( + (request, chain) -> { + intercepted.incrementAndGet(); + return chain.proceed(request); + }) + .build()) { + assertThrows( + ConfigurationException.class, + () -> parent.withOptions(options -> options.authMode(AuthMode.SA))); + try (ForwardClient copy = + parent.withOptions(options -> options.clearInterceptors().clearDefaultHeaders())) { + copy.request(get()); + assertEquals(0, intercepted.get()); + assertNull(requests.get(0).header("X-Original")); + } + parent.request(get()); + assertEquals(1, intercepted.get()); + assertEquals("value", requests.get(1).header("X-Original")); + } + } + + @Test + void sseResumptionUsesTransportAndInterceptorsForEachConnection() { + List requests = new CopyOnWriteArrayList<>(); + AtomicInteger intercepted = new AtomicInteger(); + try (ForwardClient client = + builder() + .transport( + request -> { + requests.add(request); + String id = requests.size() == 1 ? "first" : "second"; + return CompletableFuture.completedFuture( + new HttpResponse( + 200, + Collections.singletonMap( + "content-type", Collections.singletonList("text/event-stream")), + new ByteArrayInputStream( + ("id: " + + id + + "\ndata: {\"id\":\"evt_" + + id + + "\",\"type\":\"assistant.message\"}\n\n") + .getBytes(StandardCharsets.UTF_8)))); + }) + .interceptor( + (request, chain) -> { + intercepted.incrementAndGet(); + return chain.proceed(request); + }) + .build(); + EventStream stream = + client.sessions().events().resumableStream("sess_example")) { + Iterator events = stream.iterator(); + assertEquals("evt_first", events.next().id()); + assertEquals("evt_second", events.next().id()); + assertEquals("first", requests.get(1).header("Last-Event-ID")); + assertEquals(2, intercepted.get()); + } + } + + @Test + void satManagementRetainsItsPerCallKeyAndZeroRetries() { + AtomicInteger calls = new AtomicInteger(); + AtomicInteger businessTokens = new AtomicInteger(); + try (ForwardClient client = + ForwardClient.builder() + .credential( + () -> { + businessTokens.incrementAndGet(); + return CompletableFuture.completedFuture("business-sat"); + }) + .maxRetries(5) + .transport( + request -> { + calls.incrementAndGet(); + assertEquals("Bearer test-sa-key", request.header("Authorization")); + return CompletableFuture.completedFuture(response(503, "{}")); + }) + .build()) { + assertThrows( + ApiStatusException.class, + () -> + client + .serviceAccountTokens() + .create( + "test-sa-key", + ServiceAccountTokenCreateParams.builder().build(), + RequestOptions.builder().maxRetries(8).build())); + assertEquals(1, calls.get()); + assertEquals(0, businessTokens.get()); + } + } + + @Test + void defaultTransportStillSuppressesHiddenRetriesWhenIntercepted() throws Exception { + server.enqueue( + new MockResponse().setResponseCode(503).setHeader("Retry-After", "0").setBody("{}")); + server.enqueue(new MockResponse().setBody("{}")); + try (ForwardClient client = + builder().interceptor((request, chain) -> chain.proceed(request)).build()) { + assertThrows(ApiStatusException.class, () -> client.request(get())); + assertEquals(1, server.getRequestCount()); + } + } + + @Test + void closeCancelsPendingCustomIoButDoesNotCloseTransportObject() throws Exception { + CountDownLatch entered = new CountDownLatch(1); + CompletableFuture network = new CompletableFuture<>(); + class OwnedTransport implements HttpTransport, AutoCloseable { + boolean closed; + + public CompletableFuture execute(HttpRequest request) { + entered.countDown(); + return network; + } + + public void close() { + closed = true; + } + } + OwnedTransport transport = new OwnedTransport(); + ForwardClient client = builder().transport(transport).build(); + try { + CompletableFuture result = client.async().request(get()); + assertTrue(entered.await(2, TimeUnit.SECONDS)); + client.close(); + assertTrue(network.isCancelled()); + assertTrue(result.isCompletedExceptionally()); + assertFalse(transport.closed); + } finally { + client.close(); + transport.close(); + } + } +} diff --git a/src/test/java/com/qoder/cloudagents/ForwardFieldParityTest.java b/src/test/java/com/qoder/cloudagents/ForwardFieldParityTest.java new file mode 100644 index 0000000..d5cab71 --- /dev/null +++ b/src/test/java/com/qoder/cloudagents/ForwardFieldParityTest.java @@ -0,0 +1,102 @@ +package com.qoder.cloudagents; + +import static org.junit.jupiter.api.Assertions.*; + +import com.qoder.cloudagents.core.*; +import com.qoder.cloudagents.forward.*; +import java.util.Collections; +import java.util.concurrent.TimeUnit; +import okhttp3.mockwebserver.*; +import org.junit.jupiter.api.Test; + +class ForwardFieldParityTest { + @Test + void imageFileAndCreationMountAccessReachTheWire() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.start(); + try (ForwardClient client = + ForwardClient.builder() + .pat("test-pat") + .baseUrl(server.url("/api/v1/forward").toString()) + .build()) { + server.enqueue(new MockResponse().setBody("{\"data\":[]}")); + ImageSource image = ImageSource.builder().type("file").fileId("file_image").build(); + client + .sessions() + .events() + .send( + "sess_example", + SessionEventSendParams.builder() + .events( + Collections.singletonList( + SessionEventParam.builder() + .type("user.message") + .content( + EventContent.blocks( + Collections.singletonList( + ContentBlock.builder() + .type("image") + .source(image) + .build()))) + .build())) + .build()); + JsonValue event = + JsonValue.parse(server.takeRequest().getBody().readUtf8()) + .get("events") + .asArray() + .get(0); + assertEquals( + "file_image", + event.get("content").asArray().get(0).get("source").get("file_id").asString()); + assertEquals( + "file_image", + ImageSource.fromJson(event.get("content").asArray().get(0).get("source")).fileId()); + + server.enqueue(new MockResponse().setBody("{\"id\":\"memstore_example\"}")); + client + .async() + .memoryStores() + .create( + MemoryStoreCreateParams.builder() + .name("memory") + .mount( + MemoryStoreCreateMount.builder() + .identityId("idn_example") + .templateId("tmpl_example") + .access("read_write") + .build()) + .build(), + RequestOptions.builder().idempotencyKey("create-memory").build()) + .get(3, TimeUnit.SECONDS); + RecordedRequest request = server.takeRequest(); + assertEquals("create-memory", request.getHeader("Idempotency-Key")); + assertEquals( + "read_write", + JsonValue.parse(request.getBody().readUtf8()).get("mount").get("access").asString()); + } + } + } + + @Test + void missingVersionIsUnknownAndPairingTimestampsRemainOptional() { + JsonValue json = + JsonValue.parse( + "{\"initial_events_version\":0,\"created_at\":\"2026-10-22T00:00:00Z\",\"updated_at\":null}"); + assertEquals(Long.valueOf(0), new Schedule(json, null).initialEventsVersion()); + assertEquals(Long.valueOf(0), new ScheduleRun(json, null).initialEventsVersion()); + assertNull(new Schedule(JsonValue.parse("{}"), null).initialEventsVersion()); + assertNull(new ScheduleRun(JsonValue.parse("{}"), null).initialEventsVersion()); + ChannelPairing pairing = new ChannelPairing(json, null); + assertEquals("2026-10-22T00:00:00Z", pairing.createdAt()); + assertNull(pairing.updatedAt()); + assertTrue(ImageSource.builder().type("file").build().field("file_id").isMissing()); + assertTrue(ImageSource.builder().type("file").fileId(null).build().field("file_id").isNull()); + assertTrue( + MemoryStoreCreateMount.builder() + .identityId("i") + .templateId("t") + .build() + .field("access") + .isMissing()); + } +} From 5bbdf48bbf1681dba082c659bd471a3cec78069a Mon Sep 17 00:00:00 2001 From: "yuechen.wyh" <300878504+moonyue-w@users.noreply.github.com> Date: Thu, 22 Oct 2026 18:07:06 +0800 Subject: [PATCH 2/2] test(java): await asynchronous transport cancellation signals --- .../cloudagents/ClientExtensionsTest.java | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/src/test/java/com/qoder/cloudagents/ClientExtensionsTest.java b/src/test/java/com/qoder/cloudagents/ClientExtensionsTest.java index 8ca7555..1f505b2 100644 --- a/src/test/java/com/qoder/cloudagents/ClientExtensionsTest.java +++ b/src/test/java/com/qoder/cloudagents/ClientExtensionsTest.java @@ -294,7 +294,12 @@ public void close() { @Test void cancellationPropagatesThroughDerivedInterceptorFutures() throws Exception { CountDownLatch entered = new CountDownLatch(1); + CountDownLatch cancelled = new CountDownLatch(1); CompletableFuture network = new CompletableFuture<>(); + network.whenComplete( + (response, error) -> { + if (network.isCancelled()) cancelled.countDown(); + }); try (ForwardClient client = builder() .transport( @@ -307,7 +312,7 @@ void cancellationPropagatesThroughDerivedInterceptorFutures() throws Exception { CompletableFuture pending = client.async().request(get()); assertTrue(entered.await(2, TimeUnit.SECONDS)); pending.cancel(true); - assertTrue(network.isCancelled()); + assertTrue(cancelled.await(2, TimeUnit.SECONDS)); } } @@ -324,7 +329,7 @@ void headerTimeoutCancelsCustomIo() throws Exception { @Test void lateCustomResponseAfterCancellationIsClosed() throws Exception { CountDownLatch entered = new CountDownLatch(1); - AtomicBoolean closed = new AtomicBoolean(); + CountDownLatch closed = new CountDownLatch(1); CompletableFuture network = new CompletableFuture() { @Override @@ -350,10 +355,10 @@ public boolean cancel(boolean interrupt) { new ByteArrayInputStream(new byte[0]) { @Override public void close() { - closed.set(true); + closed.countDown(); } })); - assertTrue(closed.get()); + assertTrue(closed.await(2, TimeUnit.SECONDS)); } } @@ -601,7 +606,12 @@ void defaultTransportStillSuppressesHiddenRetriesWhenIntercepted() throws Except @Test void closeCancelsPendingCustomIoButDoesNotCloseTransportObject() throws Exception { CountDownLatch entered = new CountDownLatch(1); + CountDownLatch cancelled = new CountDownLatch(1); CompletableFuture network = new CompletableFuture<>(); + network.whenComplete( + (response, error) -> { + if (network.isCancelled()) cancelled.countDown(); + }); class OwnedTransport implements HttpTransport, AutoCloseable { boolean closed; @@ -620,7 +630,7 @@ public void close() { CompletableFuture result = client.async().request(get()); assertTrue(entered.await(2, TimeUnit.SECONDS)); client.close(); - assertTrue(network.isCancelled()); + assertTrue(cancelled.await(2, TimeUnit.SECONDS)); assertTrue(result.isCompletedExceptionally()); assertFalse(transport.closed); } finally {