From 2715b25a76d9a4323098a31e99a835fcd04f1971 Mon Sep 17 00:00:00 2001 From: Garret Premo Date: Sat, 19 Sep 2026 16:52:23 -0400 Subject: [PATCH] Add async systemOne and models listing returning CompletableFuture Blocking and async calls now share one attempt path: HttpClient.sendAsync with retries scheduled on CompletableFuture.delayedExecutor, so backoff never holds a thread; blocking() unwraps the future for the sync API. Both paths honor the same per-call RequestOptions and retry policy and report the same TypeSafeException subclasses; request-setup errors still throw synchronously. --- CHANGELOG.md | 4 + README.md | 13 + .../io/github/premocloud/typesafe/Models.java | 25 +- .../premocloud/typesafe/TypeSafeClient.java | 222 +++++++++++++----- .../typesafe/TypeSafeClientTest.java | 111 +++++++++ 5 files changed, 307 insertions(+), 68 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 17456fe..ece3fcc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## Unreleased + +- Async API: `systemOneAsync` mirroring every `systemOne` overload and `models().listAsync()` return `CompletableFuture`s driven by `HttpClient.sendAsync`, so retries back off without holding a thread, honoring the same per-call `RequestOptions` and failing with the same `TypeSafeException` subclasses as the blocking calls (#4). + ## 0.2.0 - 2026-09-19 - `RequestOptions.maxRetries(n)` now applies the retry count on top of the client's retry policy, or the call's own policy when one is set, instead of replacing the policy with `RetryPolicy.DEFAULT` and losing the client's statuses and backoff (#2). diff --git a/README.md b/README.md index 15713e5..4aee306 100644 --- a/README.md +++ b/README.md @@ -137,6 +137,19 @@ client.systemOne(request, RequestOptions.of(o -> o.timeout(Duration.ofSeconds(30 client.models().list(RequestOptions.of(o -> o.header("X-Trace", traceId))); ``` +### Async + +Every entry point has a `CompletableFuture` variant: `systemOneAsync` for each `systemOne` overload and +`models().listAsync()`. They are driven by `HttpClient.sendAsync`, so backoff between retries never holds a +thread. They honor the same per-call `RequestOptions` and retry policy, and complete the future exceptionally +with the same `TypeSafeException` subclass the blocking call would throw. + +```java +client.systemOneAsync(r -> r.state(email).noul("is_phishing", n -> n.instructions("Is `email` phishing?"))) + .thenAccept(response -> route(response.noul("is_phishing"))) + .exceptionally(error -> { log.warn("phishing check failed", error); return null; }); +``` + ### Models ```java diff --git a/typesafe-sdk/src/main/java/io/github/premocloud/typesafe/Models.java b/typesafe-sdk/src/main/java/io/github/premocloud/typesafe/Models.java index e958e0c..039338c 100644 --- a/typesafe-sdk/src/main/java/io/github/premocloud/typesafe/Models.java +++ b/typesafe-sdk/src/main/java/io/github/premocloud/typesafe/Models.java @@ -3,6 +3,8 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import java.util.List; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; /** The models available to the account, reached through {@link TypeSafeClient#models()}. */ public final class Models { @@ -24,12 +26,25 @@ public List list() { } public List list(RequestOptions options) { - Wire wire = client.get(PATH, Wire.class, options); + return TypeSafeClient.blocking(listAsync(options)); + } + + /** + * The async counterpart of {@link #list()}. A setup error in {@link RequestOptions} throws synchronously, + * exactly as in the blocking call. Everything else the blocking call reports completes the returned future + * exceptionally with the same {@link TypeSafeException} subclass. + */ + public CompletableFuture> listAsync() { + return listAsync(RequestOptions.NONE); + } - if (wire.models() == null) { - throw new TypeSafeException("Models response did not contain a models list"); - } + public CompletableFuture> listAsync(RequestOptions options) { + return client.getAsync(PATH, Wire.class, options).thenCompose(wire -> { + if (Objects.isNull(wire.models())) { + return CompletableFuture.>failedFuture(new TypeSafeException("Models response did not contain a models list")); + } - return List.copyOf(wire.models()); + return CompletableFuture.completedFuture(List.copyOf(wire.models())); + }); } } diff --git a/typesafe-sdk/src/main/java/io/github/premocloud/typesafe/TypeSafeClient.java b/typesafe-sdk/src/main/java/io/github/premocloud/typesafe/TypeSafeClient.java index b09bde6..b872429 100644 --- a/typesafe-sdk/src/main/java/io/github/premocloud/typesafe/TypeSafeClient.java +++ b/typesafe-sdk/src/main/java/io/github/premocloud/typesafe/TypeSafeClient.java @@ -20,7 +20,11 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; /** @@ -125,61 +129,73 @@ public TypeSafeResponse systemOne(TypeSafeRequest request) { * @throws TypeSafeConnectionException when the request cannot be delivered after retries; {@link TypeSafeTimeoutException} on timeout */ public TypeSafeResponse systemOne(TypeSafeRequest request, RequestOptions options) { - TypeSafeRequest resolved = Objects.isNull(request.model()) ? request.withModel(defaultModel) : request; - TypeSafeResponse response = post(SYSTEM_ONE_PATH, resolved, TypeSafeResponse.class, options); + return blocking(systemOneAsync(request, options)); + } - if (Objects.isNull(response.answers())) { - throw new TypeSafeException("TypeSafe response has no answers"); - } + /** {@code client.systemOneAsync(state, Map.of("category", Choice.of("Which?", "billing", "technical")))}. */ + public CompletableFuture systemOneAsync(Object state, Map questions) { + return systemOneAsync(TypeSafeRequest.of(state, questions), RequestOptions.NONE); + } - // Every question asked must come back answered, and answered as its own type. Either gap otherwise - // surfaces later, when the caller reads that key, as an IllegalArgumentException no catch of - // TypeSafeException would see. - Set unanswered = new LinkedHashSet<>(); - Set mistyped = new LinkedHashSet<>(); + public CompletableFuture systemOneAsync(Object state, Map questions, RequestOptions options) { + return systemOneAsync(TypeSafeRequest.of(state, questions), options); + } - resolved.questions().forEach((id, question) -> { - TypeSafeAnswer answer = response.answers().get(id); + /** {@code client.systemOneAsync(r -> r.state(ticket).noul("urgent", n -> n.instructions("Is `ticket` urgent?")))}. */ + public CompletableFuture systemOneAsync(Consumer configure) { + return systemOneAsync(TypeSafeRequest.of(configure), RequestOptions.NONE); + } - if (Objects.isNull(answer)) { - unanswered.add(id); - } else if (!expectedAnswer(question).isInstance(answer)) { - mistyped.add(id); - } - }); + public CompletableFuture systemOneAsync(Consumer configure, RequestOptions options) { + return systemOneAsync(TypeSafeRequest.of(configure), options); + } - if (!unanswered.isEmpty()) { - throw new TypeSafeException("TypeSafe response is missing answers for %s; answered: %s" - .formatted(unanswered, response.answers().keySet())); - } + public CompletableFuture systemOneAsync(TypeSafeRequest request) { + return systemOneAsync(request, RequestOptions.NONE); + } - if (!mistyped.isEmpty()) { - throw new TypeSafeException("TypeSafe response answered %s with a different type than was asked" - .formatted(mistyped)); - } + /** + * Evaluates every question in the request against its state, in one round trip per attempt, retrying per the + * policy without holding a thread between attempts. + * + *

Request-setup errors (an invalid {@link TypeSafeRequest} or {@link RequestOptions}) throw synchronously, + * exactly as in the blocking call. Everything else the blocking call reports completes the returned future + * exceptionally with the same {@link TypeSafeException} subclass. + * + * @param options per-call timeout, retry, and header overrides; {@link RequestOptions#NONE} inherits the client's + */ + public CompletableFuture systemOneAsync(TypeSafeRequest request, RequestOptions options) { + TypeSafeRequest resolved = Objects.isNull(request.model()) ? request.withModel(defaultModel) : request; - return response; + return postAsync(SYSTEM_ONE_PATH, resolved, TypeSafeResponse.class, options).thenCompose(response -> { + try { + validateSystemOneResponse(resolved, response); + return CompletableFuture.completedFuture(response); + } catch (TypeSafeException e) { + return CompletableFuture.failedFuture(e); + } + }); } - T get(String path, Class type, RequestOptions options) { - return send(HttpRequest.newBuilder(URI.create(baseUrl + path)).GET(), type, options); + CompletableFuture getAsync(String path, Class type, RequestOptions options) { + return sendAsync(HttpRequest.newBuilder(URI.create(baseUrl + path)).GET(), type, options); } - private T post(String path, Object body, Class type, RequestOptions options) { + private CompletableFuture postAsync(String path, Object body, Class type, RequestOptions options) { String json; try { json = objectMapper.writeValueAsString(body); } catch (JsonProcessingException e) { - throw new TypeSafeException("Could not serialize request: " + e.getOriginalMessage(), e); + return CompletableFuture.failedFuture(new TypeSafeException("Could not serialize request: " + e.getOriginalMessage(), e)); } - return send(HttpRequest.newBuilder(URI.create(baseUrl + path)) + return sendAsync(HttpRequest.newBuilder(URI.create(baseUrl + path)) .header("Content-Type", "application/json") .POST(HttpRequest.BodyPublishers.ofString(json)), type, options); } - private T send(HttpRequest.Builder template, Class type, RequestOptions options) { + private CompletableFuture sendAsync(HttpRequest.Builder template, Class type, RequestOptions options) { Duration timeout = Objects.requireNonNullElse(options.timeout(), this.timeout); RetryPolicy retryPolicy = options.resolveRetryPolicy(this.retryPolicy); Map headers = new LinkedHashMap<>(defaultHeaders); @@ -192,46 +208,101 @@ private T send(HttpRequest.Builder template, Class type, RequestOptions o .header("X-TypeSafe-Runtime", "java/" + System.getProperty("java.version")); headers.forEach(template::header); - for (int attempt = 0; ; attempt++) { - HttpRequest httpRequest = attempt == 0 ? template.build() : template.copy().header(RETRY_COUNT_HEADER, Integer.toString(attempt)).build(); - HttpResponse httpResponse; + return attemptAsync(template, type, timeout, retryPolicy, 0); + } + + /** One attempt, its retry decision, and its exception mapping: the path both blocking and async calls share. */ + private CompletableFuture attemptAsync(HttpRequest.Builder template, Class type, Duration timeout, RetryPolicy retryPolicy, int attempt) { + // Setup throws here (a bad URI, a closed client) propagate synchronously, as HttpClient.sendAsync does. + HttpRequest httpRequest = attempt == 0 ? template.build() : template.copy().header(RETRY_COUNT_HEADER, Integer.toString(attempt)).build(); + CompletableFuture> sent = httpClient.sendAsync(httpRequest, HttpResponse.BodyHandlers.ofString()); + return sent.>handle((response, error) -> { try { - httpResponse = httpClient.send(httpRequest, HttpResponse.BodyHandlers.ofString()); - } catch (HttpTimeoutException e) { - if (retryPolicy.retryTimeouts() && attempt < retryPolicy.maxRetries()) { - pause(backoff(retryPolicy, attempt, Optional.empty())); - continue; + if (Objects.nonNull(error)) { + Throwable cause = unwrap(error); + + if (cause instanceof HttpTimeoutException httpTimeout) { + if (retryPolicy.retryTimeouts() && attempt < retryPolicy.maxRetries()) { + return retryAsync(backoff(retryPolicy, attempt, Optional.empty()), template, type, timeout, retryPolicy, attempt + 1); + } + + return CompletableFuture.failedFuture(new TypeSafeTimeoutException(timeout, httpTimeout)); + } + + if (cause instanceof IOException ioException) { + if (retryPolicy.retryConnectionErrors() && attempt < retryPolicy.maxRetries()) { + return retryAsync(backoff(retryPolicy, attempt, Optional.empty()), template, type, timeout, retryPolicy, attempt + 1); + } + + return CompletableFuture.failedFuture(new TypeSafeConnectionException("Connection error: " + ioException.getMessage(), ioException)); + } + + return CompletableFuture.failedFuture(cause); + } + + int status = response.statusCode(); + + if (status >= 200 && status < 300) { + return CompletableFuture.completedFuture(deserialize(response.body(), type)); } - throw new TypeSafeTimeoutException(timeout, e); - } catch (IOException e) { - if (retryPolicy.retryConnectionErrors() && attempt < retryPolicy.maxRetries()) { - pause(backoff(retryPolicy, attempt, Optional.empty())); - continue; + if (retryPolicy.retriesStatus(status) && attempt < retryPolicy.maxRetries()) { + return retryAsync(backoff(retryPolicy, attempt, retryPolicy.respectRetryAfter() ? RetryAfter.parse(response.headers()) : Optional.empty()), + template, type, timeout, retryPolicy, attempt + 1); } - throw new TypeSafeConnectionException("Connection error: " + e.getMessage(), e); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new TypeSafeConnectionException("Request interrupted", e); + return CompletableFuture.failedFuture(TypeSafeApiException.fromResponse(status, response.body(), response.headers())); + } catch (Throwable t) { + return CompletableFuture.failedFuture(t); } + }).thenCompose(next -> next); + } + + /** Schedules the next attempt on the JDK's shared delayer, so backoff never parks a thread of ours. */ + private CompletableFuture retryAsync(Duration delay, HttpRequest.Builder template, Class type, Duration timeout, RetryPolicy retryPolicy, int attempt) { + return CompletableFuture.supplyAsync(() -> attemptAsync(template, type, timeout, retryPolicy, attempt), + CompletableFuture.delayedExecutor(delay.toMillis(), TimeUnit.MILLISECONDS)) + .thenCompose(next -> next); + } + + /** Waits on an async call, throwing the failure the blocking API throws instead of a wrapper. */ + static T blocking(CompletableFuture future) { + try { + return future.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new TypeSafeConnectionException("Request interrupted", e); + } catch (ExecutionException e) { + Throwable cause = unwrap(e); - int status = httpResponse.statusCode(); + if (cause instanceof TypeSafeException typeSafeException) { + throw typeSafeException; + } - if (status >= 200 && status < 300) { - return deserialize(httpResponse.body(), type); + if (cause instanceof RuntimeException runtimeException) { + throw runtimeException; } - if (retryPolicy.retriesStatus(status) && attempt < retryPolicy.maxRetries()) { - pause(backoff(retryPolicy, attempt, retryPolicy.respectRetryAfter() ? RetryAfter.parse(httpResponse.headers()) : Optional.empty())); - continue; + if (cause instanceof Error error) { + throw error; } - throw TypeSafeApiException.fromResponse(status, httpResponse.body(), httpResponse.headers()); + throw new TypeSafeException("Request failed: " + cause.getMessage(), cause); } } + /** Completion and execution futures wrap the thrown exception; walk down to the one the caller should see. */ + private static Throwable unwrap(Throwable error) { + Throwable cause = error; + + while ((cause instanceof CompletionException || cause instanceof ExecutionException) && Objects.nonNull(cause.getCause())) { + cause = cause.getCause(); + } + + return cause; + } + /** A server-supplied delay within the cap wins; otherwise capped exponential backoff with jitter subtracted. */ private static Duration backoff(RetryPolicy retryPolicy, int attempt, Optional retryAfter) { if (retryAfter.isPresent() && retryAfter.get().compareTo(retryPolicy.maxRetryAfter()) <= 0) { @@ -242,12 +313,37 @@ private static Duration backoff(RetryPolicy retryPolicy, int attempt, Optional unanswered = new LinkedHashSet<>(); + Set mistyped = new LinkedHashSet<>(); + + resolved.questions().forEach((id, question) -> { + TypeSafeAnswer answer = response.answers().get(id); + + if (Objects.isNull(answer)) { + unanswered.add(id); + } else if (!expectedAnswer(question).isInstance(answer)) { + mistyped.add(id); + } + }); + + if (!unanswered.isEmpty()) { + throw new TypeSafeException("TypeSafe response is missing answers for %s; answered: %s" + .formatted(unanswered, response.answers().keySet())); + } + + if (!mistyped.isEmpty()) { + throw new TypeSafeException("TypeSafe response answered %s with a different type than was asked" + .formatted(mistyped)); } } diff --git a/typesafe-sdk/src/test/java/io/github/premocloud/typesafe/TypeSafeClientTest.java b/typesafe-sdk/src/test/java/io/github/premocloud/typesafe/TypeSafeClientTest.java index 42e9f4f..a7577f5 100644 --- a/typesafe-sdk/src/test/java/io/github/premocloud/typesafe/TypeSafeClientTest.java +++ b/typesafe-sdk/src/test/java/io/github/premocloud/typesafe/TypeSafeClientTest.java @@ -11,8 +11,11 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -377,6 +380,114 @@ void builderRequiresApiKeyAndValidatesTimeout() { assertEquals(RetryPolicy.DEFAULT, TypeSafeClient.builder().apiKey(API_KEY).build().retryPolicy()); } + @Test + void systemOneAsyncReturnsTypedAnswers() throws Exception { + server.reply(200, RESPONSE_JSON); + + TypeSafeResponse response = client.systemOneAsync(spamRequest()).get(5, TimeUnit.SECONDS); + + assertEquals("jev-1.13.0", response.model()); + assertEquals(0.93, response.noul("is_phishing")); + assertEquals("PHISHING", response.choice("spam_category").choice()); + assertEquals(1.7, response.score("urgency").score()); + } + + @Test + void systemOneAsyncFailsWithStatusSpecificException() { + server.reply(401, "{\"error\":\"invalid api key\"}", Map.of("x-typesafe-request-id", "req_123")); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> client.systemOneAsync(spamRequest()).get(5, TimeUnit.SECONDS)); + + TypeSafeAuthenticationException exception = assertInstanceOf(TypeSafeAuthenticationException.class, failure.getCause()); + assertEquals(401, exception.status()); + assertEquals("401 invalid api key", exception.getMessage()); + assertEquals("req_123", exception.requestId().orElseThrow()); + } + + @Test + void systemOneAsyncRetriesWithRetryCountHeader() throws Exception { + TypeSafeClient retrying = TypeSafeClient.builder().apiKey(API_KEY).baseUrl(server.baseUrl()) + .retryPolicy(RetryPolicy.of(r -> r.maxRetries(2).backoffInitial(Duration.ofMillis(1)).backoffMax(Duration.ofMillis(2)))).build(); + server.reply(500, "{\"error\":\"boom\"}"); + server.reply(429, "{\"error\":\"slow\"}", Map.of("retry-after-ms", "5")); + server.reply(200, RESPONSE_JSON); + + TypeSafeResponse response = retrying.systemOneAsync(spamRequest()).get(5, TimeUnit.SECONDS); + + assertEquals(0.93, response.noul("is_phishing")); + assertEquals(3, server.recorded().size()); + assertNull(server.recorded().get(0).headers().getFirst("X-TypeSafe-Retry-Count")); + assertEquals("1", server.recorded().get(1).headers().getFirst("X-TypeSafe-Retry-Count")); + assertEquals("2", server.recorded().get(2).headers().getFirst("X-TypeSafe-Retry-Count")); + } + + @Test + void systemOneAsyncHonorsPerCallOptions() { + TypeSafeClient retrying = TypeSafeClient.builder().apiKey(API_KEY).baseUrl(server.baseUrl()).header("X-Team", "review") + .retryPolicy(RetryPolicy.of(r -> r.maxRetries(2).backoffInitial(Duration.ofMillis(1)))).build(); + server.reply(500, "no retry please"); + + ExecutionException failure = assertThrows(ExecutionException.class, () -> retrying + .systemOneAsync(spamRequest(), RequestOptions.of(o -> o.maxRetries(0).header("X-Team", "spike").header("X-Trace", "t1"))) + .get(5, TimeUnit.SECONDS)); + + assertInstanceOf(TypeSafeInternalServerException.class, failure.getCause()); + assertEquals(1, server.recorded().size()); + assertEquals("spike", server.recorded().get(0).headers().getFirst("X-Team")); + assertEquals("t1", server.recorded().get(0).headers().getFirst("X-Trace")); + + server.replyAfter(1500, 200, RESPONSE_JSON); + ExecutionException timed = assertThrows(ExecutionException.class, () -> retrying + .systemOneAsync(spamRequest(), RequestOptions.of(o -> o.timeout(Duration.ofMillis(200)).maxRetries(0))) + .get(5, TimeUnit.SECONDS)); + + assertEquals(Duration.ofMillis(200), assertInstanceOf(TypeSafeTimeoutException.class, timed.getCause()).timeout()); + } + + @Test + void systemOneAsyncRejectsAResponseMissingAnAnswerForAQuestionThatWasAsked() { + server.reply(200, """ + {"model": "jev-1.13.0", "answers": {"is_phishing": {"type": "noul", "noul": 0.93}}, + "usage": {"input_tokens": 1, "output_tokens": 1}} + """); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> client.systemOneAsync(spamRequest()).get(5, TimeUnit.SECONDS)); + + TypeSafeException exception = assertInstanceOf(TypeSafeException.class, failure.getCause()); + assertEquals(TypeSafeException.class, exception.getClass()); + assertTrue(exception.getMessage().contains("spam_category"), exception.getMessage()); + } + + @Test + void systemOneAsyncThrowsRequestSetupErrorsSynchronously() { + assertThrows(IllegalStateException.class, () -> client.systemOneAsync(r -> r.noul("is_phishing", n -> n.instructions("Yes?")))); + assertThrows(IllegalStateException.class, () -> client.systemOneAsync(Map.of("email", "text"), Map.of())); + } + + @Test + void modelsListAsyncReturnsModels() throws Exception { + server.reply(200, "{\"models\":[{\"name\":\"jev-1.13.0\",\"description\":\"Jev\",\"release_date\":\"2026-09-01\",\"extra\":1}]}"); + + List models = client.models().listAsync().get(5, TimeUnit.SECONDS); + + assertEquals("GET", server.recorded().get(0).method()); + assertEquals("/v1/models", server.recorded().get(0).path()); + assertEquals(List.of(new ModelCard("jev-1.13.0", "Jev", "2026-09-01")), models); + } + + @Test + void modelsListAsyncHonorsPerCallHeaders() throws Exception { + client = TypeSafeClient.builder().apiKey(API_KEY).baseUrl(server.baseUrl()).header("X-Team", "review").retryPolicy(RetryPolicy.none()).build(); + server.reply(200, "{\"models\":[]}"); + + assertEquals(List.of(), client.models().listAsync(RequestOptions.of(o -> o.header("X-Trace", "t2"))).get(5, TimeUnit.SECONDS)); + + assertEquals("t2", server.recorded().get(0).headers().getFirst("X-Trace")); + assertEquals("review", server.recorded().get(0).headers().getFirst("X-Team")); + } + private static TypeSafeRequest spamRequest() { return TypeSafeRequest.of(r -> r .state(Map.of("email", Map.of("subject", "URGENT")))