From a6b0f793822c677df1527d516b0ed15cf506ba93 Mon Sep 17 00:00:00 2001 From: Alex Chang Date: Mon, 5 Oct 2026 14:43:20 +0000 Subject: [PATCH 1/5] feat(agents): run local tools during session creation --- docs/helpers/agent-sessions.md | 24 ++++ .../models/beta/agents/AgentToolHandlers.kt | 32 +++++ .../async/beta/agents/SessionServiceAsync.kt | 18 +++ .../beta/agents/SessionServiceAsyncImpl.kt | 19 +++ .../beta/agents/AgentSessionStream.kt | 29 +++-- .../beta/agents/AgentSessionStreamAsync.kt | 73 +++++++---- .../beta/agents/AgentSessionStreamSupport.kt | 48 ++++++-- .../blocking/beta/agents/SessionService.kt | 19 +++ .../beta/agents/SessionServiceImpl.kt | 11 ++ .../beta/agents/AgentSessionStreamTest.kt | 113 ++++++++++++++++++ 10 files changed, 339 insertions(+), 47 deletions(-) create mode 100644 openai-java-core/src/main/kotlin/com/openai/models/beta/agents/AgentToolHandlers.kt diff --git a/docs/helpers/agent-sessions.md b/docs/helpers/agent-sessions.md index ead402780..5931708f8 100644 --- a/docs/helpers/agent-sessions.md +++ b/docs/helpers/agent-sessions.md @@ -26,6 +26,30 @@ try (StreamResponse stream = client.beta().agents().sessions( } ``` +For the initial prompt, pass the same local handlers to `createStreaming`: + +```java +import com.openai.models.beta.agents.AgentToolHandlers; +import com.openai.models.beta.agents.sessions.SessionCreateParams; +import com.openai.services.beta.agents.AgentTurnResults; + +var handlers = AgentToolHandlers.builder() + .toolHandler("weather", args -> weatherService.lookup(args)) + .build(); +var creation = SessionCreateParams.builder() + .agent(agent) // Configure the corresponding function definition on the agent. + .environmentNone() + .input("Check the weather in Paris") + .build(); + +try (var stream = client.beta().agents().sessions().createStreaming(creation, handlers)) { + System.out.println(AgentTurnResults.getFinalResult(stream).outputText()); +} +``` + +The async service accepts the same overload; use `asyncToolHandler` for callbacks +returning a `CompletionStage`. Handlers run locally while consuming the stream. + A handler receives a deep copy of the JSON object arguments and can return a string, JSON object (`Map`), `List`, `AgentFunctionCallOutputParam`, or `null`. Invalid arguments and handler exceptions diff --git a/openai-java-core/src/main/kotlin/com/openai/models/beta/agents/AgentToolHandlers.kt b/openai-java-core/src/main/kotlin/com/openai/models/beta/agents/AgentToolHandlers.kt new file mode 100644 index 000000000..0c48fe295 --- /dev/null +++ b/openai-java-core/src/main/kotlin/com/openai/models/beta/agents/AgentToolHandlers.kt @@ -0,0 +1,32 @@ +package com.openai.models.beta.agents + +import java.util.concurrent.CompletableFuture +import java.util.concurrent.CompletionStage +import java.util.function.Function + +/** Local function handlers for a beta Agents creation stream. Never sent to the API. */ +class AgentToolHandlers +private constructor( + internal val handlers: Map, CompletionStage<*>>> +) { + companion object { + @JvmStatic fun builder() = Builder() + } + + class Builder { + private val handlers = + linkedMapOf, CompletionStage<*>>>() + + /** Uses the same return values and sanitized errors as session stream tool handlers. */ + fun toolHandler(name: String, handler: Function, Any?>) = apply { + handlers[name] = Function { CompletableFuture.completedFuture(handler.apply(it)) } + } + + fun asyncToolHandler( + name: String, + handler: Function, CompletionStage<*>>, + ) = apply { handlers[name] = handler } + + fun build() = AgentToolHandlers(handlers.toMap()) + } +} diff --git a/openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsync.kt b/openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsync.kt index ffd81c8b8..c0e170583 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsync.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsync.kt @@ -12,6 +12,7 @@ import com.openai.models.beta.agents.AgentSession import com.openai.models.beta.agents.AgentSessionDeleted import com.openai.models.beta.agents.AgentSessionEvent import com.openai.models.beta.agents.AgentSessionStreamParams +import com.openai.models.beta.agents.AgentToolHandlers import com.openai.models.beta.agents.sessions.SessionCreateParams import com.openai.models.beta.agents.sessions.SessionDeleteParams import com.openai.models.beta.agents.sessions.SessionListPageAsync @@ -94,6 +95,23 @@ interface SessionServiceAsync { requestOptions: RequestOptions = RequestOptions.none(), ): AsyncStreamResponse + /** Creates and handles the initial turn using local function callbacks. */ + fun createStreaming( + params: SessionCreateParams, + handlers: AgentToolHandlers, + ): AsyncStreamResponse = + createStreaming(params, handlers, RequestOptions.none()) + + /** @see createStreaming */ + fun createStreaming( + params: SessionCreateParams, + handlers: AgentToolHandlers, + requestOptions: RequestOptions, + ): AsyncStreamResponse = + throw UnsupportedOperationException( + "This service does not implement creation tool handlers" + ) + /** * Retrieves the current state of a managed agent session. See * [managing sessions](https://developers.openai.com/api/docs/guides/agents-api/sessions/manage). diff --git a/openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsyncImpl.kt b/openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsyncImpl.kt index b87985ed8..2c7341d37 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsyncImpl.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsyncImpl.kt @@ -29,6 +29,7 @@ import com.openai.models.beta.agents.AgentSession import com.openai.models.beta.agents.AgentSessionDeleted import com.openai.models.beta.agents.AgentSessionEvent import com.openai.models.beta.agents.AgentSessionStreamParams +import com.openai.models.beta.agents.AgentToolHandlers import com.openai.models.beta.agents.sessions.SessionCreateParams import com.openai.models.beta.agents.sessions.SessionDeleteParams import com.openai.models.beta.agents.sessions.SessionListPageAsync @@ -129,6 +130,24 @@ class SessionServiceAsyncImpl internal constructor(private val clientOptions: Cl .toAsync(clientOptions.streamHandlerExecutor) .let { AgentTurnResults.collecting(it) } + override fun createStreaming( + params: SessionCreateParams, + handlers: AgentToolHandlers, + requestOptions: RequestOptions, + ): AsyncStreamResponse = + AgentTurnResults.collecting( + AgentSessionStreamAsync( + this, + null, + requestOptions, + clientOptions.streamHandlerExecutor, + clientOptions.sleeper, + params, + handlers, + ), + handlers.handlers.keys, + ) + override fun retrieve( params: SessionRetrieveParams, requestOptions: RequestOptions, diff --git a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStream.kt b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStream.kt index da4c9125c..14338d33b 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStream.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStream.kt @@ -4,6 +4,8 @@ import com.openai.core.RequestOptions import com.openai.core.http.StreamResponse import com.openai.models.beta.agents.AgentSessionEvent import com.openai.models.beta.agents.AgentSessionStreamParams +import com.openai.models.beta.agents.AgentToolHandlers +import com.openai.models.beta.agents.sessions.SessionCreateParams import com.openai.models.beta.agents.sessions.events.EventCreateParams import com.openai.services.blocking.beta.agents.SessionService import java.util.Spliterator @@ -15,17 +17,22 @@ import java.util.stream.Stream import java.util.stream.StreamSupport /** - * One turn's original typed events. Requires an idle session and a single input writer. Use - * try-with-resources. Closing releases the connection without cancelling the backend turn. Unknown - * tools remain manual. Registered handlers run sequentially after their event is delivered. + * One turn's original typed events. Follow-up streams require an idle session and a single input + * writer. Use try-with-resources. Closing releases the connection without cancelling the backend + * turn. Unknown tools remain manual. Registered handlers run sequentially after their event is + * delivered. */ internal class AgentSessionStream( private val sessions: SessionService, - params: AgentSessionStreamParams, + params: AgentSessionStreamParams?, private val options: RequestOptions, private val sleeper: com.openai.core.Sleeper, + private val creation: SessionCreateParams? = null, + handlers: AgentToolHandlers? = null, ) : StreamResponse { - private val support = AgentSessionStreamSupport(params) + private val support = + if (creation == null) AgentSessionStreamSupport(checkNotNull(params)) + else AgentSessionStreamSupport(checkNotNull(handlers), creation._headers()) private val closed = AtomicBoolean() private val submissionLock = Any() private val consumed = AtomicBoolean() @@ -35,18 +42,22 @@ internal class AgentSessionStream( private val response: com.openai.core.http.HttpResponseFor> init { - support.checkIdle(sessions.retrieve(support.retrieveParams(), options)) + if (creation == null) + support.checkIdle(sessions.retrieve(support.retrieveParams(), options)) response = - sessions.events().withRawResponse().streamStreaming(support.streamParams(), options) + if (creation == null) + sessions.events().withRawResponse().streamStreaming(support.streamParams(), options) + else sessions.withRawResponse().createStreaming(creation, options) source = try { - response.parse() + AgentTurnResults.uncollected(response.parse()) } catch (error: Throwable) { response.close() throw error } try { - sessions.events().withRawResponse().create(support.inputParams(), options).close() + if (creation == null) + sessions.events().withRawResponse().create(support.inputParams(), options).close() } catch (error: Throwable) { close() throw error diff --git a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamAsync.kt b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamAsync.kt index 616ccc7ed..3f5065826 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamAsync.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamAsync.kt @@ -5,6 +5,8 @@ import com.openai.core.http.AsyncStreamResponse import com.openai.core.http.StreamResponse import com.openai.models.beta.agents.AgentSessionEvent import com.openai.models.beta.agents.AgentSessionStreamParams +import com.openai.models.beta.agents.AgentToolHandlers +import com.openai.models.beta.agents.sessions.SessionCreateParams import com.openai.models.beta.agents.sessions.events.EventCreateParams import com.openai.services.async.beta.agents.SessionServiceAsync import java.util.Optional @@ -15,12 +17,16 @@ import java.util.concurrent.atomic.AtomicBoolean /** Async one-turn stream; uses the SDK executor and never allocates a thread pool. */ internal class AgentSessionStreamAsync( private val sessions: SessionServiceAsync, - params: AgentSessionStreamParams, + params: AgentSessionStreamParams?, private val options: RequestOptions, private val defaultExecutor: Executor, private val sleeper: com.openai.core.Sleeper, + private val creation: SessionCreateParams? = null, + handlers: AgentToolHandlers? = null, ) : AsyncStreamResponse { - private val support = AgentSessionStreamSupport(params) + private val support = + if (creation == null) AgentSessionStreamSupport(checkNotNull(params)) + else AgentSessionStreamSupport(checkNotNull(handlers), creation._headers()) private val subscribed = AtomicBoolean() private val closed = AtomicBoolean() private val submissionLock = Any() @@ -58,33 +64,48 @@ internal class AgentSessionStreamAsync( execute { notifyComplete(null) } return@apply } - await(sessions.retrieve(support.retrieveParams(), options), cancelOnClose = false) { session - -> - support.checkIdle(session) - val opening = - sessions.events().withRawResponse().streamStreaming(support.streamParams(), options) - // Closing during an in-flight open must also close the eventual HTTP response. - opening.whenComplete { response, _ -> - openingResponse = response - if (closed.get()) response?.close() + if (creation != null) open(sessions.withRawResponse().createStreaming(creation, options)) + else + await(sessions.retrieve(support.retrieveParams(), options), cancelOnClose = false) { + session -> + support.checkIdle(session) + open( + sessions + .events() + .withRawResponse() + .streamStreaming(support.streamParams(), options) + ) } - await(opening, cancelOnClose = false) { response -> - openingResponse = response - val stream = - try { - response.parse() - } catch (error: Throwable) { - response.close() - throw error - } - source = stream - if (closed.get()) { - stream.close() - return@await + } + + private fun open( + opening: + CompletableFuture< + com.openai.core.http.HttpResponseFor> + > + ) { + // A response arriving after close must still be released. + opening.whenComplete { response, _ -> + openingResponse = response + if (closed.get()) response?.close() + } + await(opening, cancelOnClose = false) { response -> + openingResponse = response + val stream = + try { + AgentTurnResults.uncollected(response.parse()) + } catch (error: Throwable) { + response.close() + throw error } - iterator = stream.stream().iterator() - create(support.inputParams())?.let { await(it) { pump() } } + source = stream + if (closed.get()) { + stream.close() + return@await } + iterator = stream.stream().iterator() + if (creation != null) pump() + else create(support.inputParams())?.let { await(it) { pump() } } } } diff --git a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt index 88b523429..aad20bae5 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt @@ -2,6 +2,7 @@ package com.openai.services.beta.agents import com.fasterxml.jackson.core.type.TypeReference import com.fasterxml.jackson.databind.JsonNode +import com.openai.core.http.Headers import com.openai.core.jsonMapper import com.openai.errors.BadRequestException import com.openai.models.beta.agents.* @@ -11,12 +12,31 @@ import com.openai.models.beta.agents.sessions.events.EventStreamParams import java.util.UUID import java.util.concurrent.CompletableFuture import java.util.concurrent.CompletionException +import java.util.concurrent.CompletionStage +import java.util.function.Function import java.util.concurrent.ExecutionException import kotlin.jvm.optionals.getOrNull -internal class AgentSessionStreamSupport(val params: AgentSessionStreamParams) { +internal class AgentSessionStreamSupport +private constructor( + private val params: AgentSessionStreamParams?, + private val handlers: + Map< + String, + Function, CompletionStage<*>>, + >, + private val headers: Headers, +) { + constructor(params: AgentSessionStreamParams) : this(params, params.handlers, params.headers) + + constructor( + handlers: AgentToolHandlers, + headers: Headers, + ) : this(null, handlers.handlers, headers.toBuilder().remove("Idempotency-Key").build()) + + private var sessionId = params?.sessionId private val mapper = jsonMapper() - private val inputKey = params.idempotencyKey ?: UUID.randomUUID().toString() + private val inputKey = params?.idempotencyKey ?: UUID.randomUUID().toString() private val eventIds = linkedSetOf() private val handledCalls = hashSetOf>() private var turnId: String? = null @@ -24,22 +44,22 @@ internal class AgentSessionStreamSupport(val params: AgentSessionStreamParams) { fun retrieveParams() = SessionRetrieveParams.builder() - .sessionId(params.sessionId) - .putAllAdditionalHeaders(params.headers) + .sessionId(checkNotNull(sessionId) { "Creation stream has not identified its session" }) + .putAllAdditionalHeaders(headers) .build() fun streamParams() = EventStreamParams.builder() - .sessionId(params.sessionId) - .putAllAdditionalHeaders(params.headers) + .sessionId(checkNotNull(sessionId) { "Creation stream has not identified its session" }) + .putAllAdditionalHeaders(headers) .build() fun inputParams() = EventCreateParams.builder() - .sessionId(params.sessionId) + .sessionId(checkNotNull(sessionId) { "Creation stream has not identified its session" }) .idempotencyKey(inputKey) - .putAllAdditionalHeaders(params.headers) - .addAgentSessionInputMessageEvent(params.input) + .putAllAdditionalHeaders(headers) + .addAgentSessionInputMessageEvent(checkNotNull(params).input) .build() fun checkIdle(session: AgentSession) { @@ -49,6 +69,7 @@ internal class AgentSessionStreamSupport(val params: AgentSessionStreamParams) { } fun accept(event: AgentSessionEvent): Boolean { + event.created().getOrNull()?.let { if (sessionId == null) sessionId = it.session().id() } val id = mapper.valueToTree(event).path("event_id").asText() if (!eventIds.add(id)) return false if (eventIds.size > 1024) @@ -73,7 +94,8 @@ internal class AgentSessionStreamSupport(val params: AgentSessionStreamParams) { val call = event.turnItemAdded().getOrNull()?.item()?.functionCall()?.getOrNull() ?: return CompletableFuture.completedFuture(null) - val handler = params.handlers[call.name()] ?: return CompletableFuture.completedFuture(null) + val handler = handlers[call.name()] ?: return CompletableFuture.completedFuture(null) + checkNotNull(sessionId) { "Creation stream has not identified its session" } if (!handledCalls.add(call.turnId() to call.callId())) return CompletableFuture.completedFuture(null) val output = @@ -122,9 +144,11 @@ internal class AgentSessionStreamSupport(val params: AgentSessionStreamParams) { .error("Tool handler failed.") } EventCreateParams.builder() - .sessionId(params.sessionId) + .sessionId( + checkNotNull(sessionId) { "Creation stream has not identified its session" } + ) .idempotencyKey(UUID.randomUUID().toString()) - .putAllAdditionalHeaders(params.headers) + .putAllAdditionalHeaders(headers) .addEvent(result.build()) .build() } diff --git a/openai-java-core/src/main/kotlin/com/openai/services/blocking/beta/agents/SessionService.kt b/openai-java-core/src/main/kotlin/com/openai/services/blocking/beta/agents/SessionService.kt index 28d7ef725..e9a1d7d0e 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/blocking/beta/agents/SessionService.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/blocking/beta/agents/SessionService.kt @@ -11,6 +11,7 @@ import com.openai.models.beta.agents.AgentSession import com.openai.models.beta.agents.AgentSessionDeleted import com.openai.models.beta.agents.AgentSessionEvent import com.openai.models.beta.agents.AgentSessionStreamParams +import com.openai.models.beta.agents.AgentToolHandlers import com.openai.models.beta.agents.sessions.SessionCreateParams import com.openai.models.beta.agents.sessions.SessionDeleteParams import com.openai.models.beta.agents.sessions.SessionListPage @@ -95,6 +96,24 @@ interface SessionService { requestOptions: RequestOptions = RequestOptions.none(), ): StreamResponse + /** Creates and handles the initial turn using local function callbacks. */ + @MustBeClosed + fun createStreaming( + params: SessionCreateParams, + handlers: AgentToolHandlers, + ): StreamResponse = createStreaming(params, handlers, RequestOptions.none()) + + /** @see createStreaming */ + @MustBeClosed + fun createStreaming( + params: SessionCreateParams, + handlers: AgentToolHandlers, + requestOptions: RequestOptions, + ): StreamResponse = + throw UnsupportedOperationException( + "This service does not implement creation tool handlers" + ) + /** * Retrieves the current state of a managed agent session. See * [managing sessions](https://developers.openai.com/api/docs/guides/agents-api/sessions/manage). diff --git a/openai-java-core/src/main/kotlin/com/openai/services/blocking/beta/agents/SessionServiceImpl.kt b/openai-java-core/src/main/kotlin/com/openai/services/blocking/beta/agents/SessionServiceImpl.kt index 09a7f3412..ada22cf67 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/blocking/beta/agents/SessionServiceImpl.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/blocking/beta/agents/SessionServiceImpl.kt @@ -27,6 +27,7 @@ import com.openai.models.beta.agents.AgentSession import com.openai.models.beta.agents.AgentSessionDeleted import com.openai.models.beta.agents.AgentSessionEvent import com.openai.models.beta.agents.AgentSessionStreamParams +import com.openai.models.beta.agents.AgentToolHandlers import com.openai.models.beta.agents.sessions.SessionCreateParams import com.openai.models.beta.agents.sessions.SessionDeleteParams import com.openai.models.beta.agents.sessions.SessionListPage @@ -113,6 +114,16 @@ class SessionServiceImpl internal constructor(private val clientOptions: ClientO // post /agents/sessions withRawResponse().createStreaming(params, requestOptions).parse() + override fun createStreaming( + params: SessionCreateParams, + handlers: AgentToolHandlers, + requestOptions: RequestOptions, + ): StreamResponse = + AgentTurnResults.collecting( + AgentSessionStream(this, null, requestOptions, clientOptions.sleeper, params, handlers), + handlers.handlers.keys, + ) + override fun retrieve( params: SessionRetrieveParams, requestOptions: RequestOptions, diff --git a/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt b/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt index 9d8482c82..1a882759e 100644 --- a/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt +++ b/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt @@ -8,6 +8,7 @@ import com.openai.core.http.* import com.openai.core.jsonMapper import com.openai.helpers.beta.agents.AgentFunctionTool import com.openai.models.beta.agents.* +import com.openai.models.beta.agents.sessions.SessionCreateParams import java.io.ByteArrayInputStream import java.io.ByteArrayOutputStream import java.time.Duration @@ -72,6 +73,7 @@ internal class AgentSessionStreamTest { var inputResponse: HttpResponse? = null val requests = java.util.Collections.synchronizedList(mutableListOf()) val options = java.util.Collections.synchronizedList(mutableListOf()) + var creationBody: JsonNode? = null val posts = java.util.Collections.synchronizedList(mutableListOf()) var status = "idle" var postError: String? = null @@ -155,6 +157,13 @@ internal class AgentSessionStreamTest { requests.add(request) options.add(requestOptions) return when { + request.pathSegments.last() == "sessions" && request.method == HttpMethod.POST -> { + val bytes = ByteArrayOutputStream() + request.body!!.writeTo(bytes) + creationBody = mapper.readTree(bytes.toByteArray()) + opened = true + response(events.joinToString("") { "data: $it\n\n" }, stream = true) + } request.method == HttpMethod.POST -> { check(opened) { "input submitted before subscription" } val bytes = ByteArrayOutputStream() @@ -236,6 +245,110 @@ internal class AgentSessionStreamTest { .also { clients.add(it) } } + private fun creationParams() = + SessionCreateParams.builder() + .agentId("agent") + .environmentNone() + .input("Question") + .putAdditionalHeader("X-Request-Context", "test") + .putAdditionalHeader("Idempotency-Key", "creation-key") + .build() + + private fun creationEvents() = + listOf( + """{"type":"agent.session.created","event_id":"created-session","session":{"id":"s","status":"in_progress"}}""", + """{"type":"agent.session.turn.created","event_id":"created-root","session_id":"s","turn_id":"root","turn":{"id":"root","session_id":"s","subagent_id":null}}""", + call(args = "{}"), + call(args = "{}", event = "duplicate-call"), + """{"type":"agent.session.requires_action","event_id":"action","session":{"id":"s","status":"requires_action","required_actions":[{"type":"function_call","name":"tool","call_id":"call","turn_id":"root"}]}}""", + """{"type":"agent.session.turn.item.done","event_id":"answer","session_id":"s","turn_id":"root","output_index":0,"item":{"id":"answer","type":"message","role":"assistant","turn_id":"root","phase":"final_answer","status":"completed","content":[{"type":"output_text","text":"Answer","annotations":[]}]}}""", + """{"type":"agent.session.turn.completed","event_id":"completed-root","session_id":"s","turn_id":"root","turn":{"id":"root","session_id":"s","subagent_id":null}}""", + """{"type":"agent.session.idle","event_id":"idle","session":{"id":"s","status":"idle"}}""", + ) + + @ParameterizedTest + @ValueSource(booleans = [false, true]) + fun `creation dispatches initial tools using shared retries and collects final output`( + async: Boolean + ) { + val t = Transport(creationEvents()) + t.postError = + """{"error":{"code":"invalid_request_error","message":"Unknown pending tool call: call"}}""" + t.postErrors = 1 + val calls = AtomicInteger() + val handlers = + AgentToolHandlers.builder() + .asyncToolHandler("tool") { + calls.incrementAndGet() + CompletableFuture.completedFuture("Result") + } + .build() + val client = t.client() + val result = + if (async) { + val stream = + client + .async() + .beta() + .agents() + .sessions() + .createStreaming(creationParams(), handlers) + AgentTurnResults.getFinalResult(stream).get(ASYNC_TIMEOUT_SECONDS, TimeUnit.SECONDS) + } else + client.beta().agents().sessions().createStreaming(creationParams(), handlers).use { + AgentTurnResults.getFinalResult(it) + } + assertThat(result.outputText()).isEqualTo("Answer") + assertThat(result.sessionId()).isEqualTo("s") + assertThat(calls.get()).isEqualTo(1) + assertThat(t.creationBody!!.path("input").asText()).isEqualTo("Question") + assertThat(t.creationBody!!.has("handlers")).isFalse() + assertThat(t.requests).hasSize(3).allSatisfy { + assertThat(it.method).isEqualTo(HttpMethod.POST) + } + assertThat(t.requests.drop(1)).allSatisfy { + assertThat(it.pathSegments).endsWith("s", "events") + assertThat(it.headers.values("X-Request-Context")).containsExactly("test") + assertThat(it.headers.values("Idempotency-Key")).doesNotContain("creation-key") + } + assertThat(t.requests[1].headers.values("Idempotency-Key")) + .isEqualTo(t.requests[2].headers.values("Idempotency-Key")) + assertThat(t.posts).hasSize(2).allSatisfy { + assertThat(it.path("events").first().path("output").asText()).isEqualTo("Result") + } + assertThat(t.streamClosed).isTrue() + } + + @ParameterizedTest + @ValueSource(booleans = [false, true]) + fun `closing creation stream on call delivery does not run pending callback`(async: Boolean) { + val t = Transport(creationEvents()) + val calls = AtomicInteger() + val handlers = + AgentToolHandlers.builder().toolHandler("tool") { calls.incrementAndGet() }.build() + val client = t.client() + if (async) { + val stream = + client + .async() + .beta() + .agents() + .sessions() + .createStreaming(creationParams(), handlers) + stream + .subscribe { if (it.isTurnItemAdded()) stream.close() } + .onCompleteFuture() + .get(ASYNC_TIMEOUT_SECONDS, TimeUnit.SECONDS) + } else + client.beta().agents().sessions().createStreaming(creationParams(), handlers).use { + stream -> + stream.stream().forEach { if (it.isTurnItemAdded()) stream.close() } + } + assertThat(calls.get()).isZero() + assertThat(t.posts).isEmpty() + assertThat(t.streamClosed).isTrue() + } + private fun params() = AgentSessionStreamParams.builder().sessionId("s").input("hello") class LookupItem(val itemId: String) From 3f586ba2a3d57d45056bf3433f153d0d93592c11 Mon Sep 17 00:00:00 2001 From: Alex Chang Date: Mon, 5 Oct 2026 14:44:29 +0000 Subject: [PATCH 2/5] style(agents): format shared dispatcher types --- .../services/beta/agents/AgentSessionStreamSupport.kt | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt index aad20bae5..8db1d7127 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt @@ -13,18 +13,14 @@ import java.util.UUID import java.util.concurrent.CompletableFuture import java.util.concurrent.CompletionException import java.util.concurrent.CompletionStage -import java.util.function.Function import java.util.concurrent.ExecutionException +import java.util.function.Function import kotlin.jvm.optionals.getOrNull internal class AgentSessionStreamSupport private constructor( private val params: AgentSessionStreamParams?, - private val handlers: - Map< - String, - Function, CompletionStage<*>>, - >, + private val handlers: Map, CompletionStage<*>>>, private val headers: Headers, ) { constructor(params: AgentSessionStreamParams) : this(params, params.handlers, params.headers) From 94cfa2268ae72a5647a35bd7290088e16b94ab74 Mon Sep 17 00:00:00 2001 From: Alex Chang Date: Mon, 5 Oct 2026 16:06:08 +0000 Subject: [PATCH 3/5] fix(agents): finish inputless creation streams at idle --- .../beta/agents/AgentSessionStream.kt | 2 +- .../beta/agents/AgentSessionStreamAsync.kt | 2 +- .../beta/agents/AgentSessionStreamSupport.kt | 14 ++++- .../beta/agents/AgentSessionStreamTest.kt | 57 +++++++++++++++++++ 4 files changed, 70 insertions(+), 5 deletions(-) diff --git a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStream.kt b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStream.kt index 14338d33b..a091868d6 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStream.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStream.kt @@ -32,7 +32,7 @@ internal class AgentSessionStream( ) : StreamResponse { private val support = if (creation == null) AgentSessionStreamSupport(checkNotNull(params)) - else AgentSessionStreamSupport(checkNotNull(handlers), creation._headers()) + else AgentSessionStreamSupport(checkNotNull(handlers), creation) private val closed = AtomicBoolean() private val submissionLock = Any() private val consumed = AtomicBoolean() diff --git a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamAsync.kt b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamAsync.kt index 3f5065826..4dff7314a 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamAsync.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamAsync.kt @@ -26,7 +26,7 @@ internal class AgentSessionStreamAsync( ) : AsyncStreamResponse { private val support = if (creation == null) AgentSessionStreamSupport(checkNotNull(params)) - else AgentSessionStreamSupport(checkNotNull(handlers), creation._headers()) + else AgentSessionStreamSupport(checkNotNull(handlers), creation) private val subscribed = AtomicBoolean() private val closed = AtomicBoolean() private val submissionLock = Any() diff --git a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt index 8db1d7127..0e070d954 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt @@ -6,6 +6,7 @@ import com.openai.core.http.Headers import com.openai.core.jsonMapper import com.openai.errors.BadRequestException import com.openai.models.beta.agents.* +import com.openai.models.beta.agents.sessions.SessionCreateParams import com.openai.models.beta.agents.sessions.SessionRetrieveParams import com.openai.models.beta.agents.sessions.events.EventCreateParams import com.openai.models.beta.agents.sessions.events.EventStreamParams @@ -22,13 +23,19 @@ private constructor( private val params: AgentSessionStreamParams?, private val handlers: Map, CompletionStage<*>>>, private val headers: Headers, + private val expectsTurn: Boolean = true, ) { constructor(params: AgentSessionStreamParams) : this(params, params.handlers, params.headers) constructor( handlers: AgentToolHandlers, - headers: Headers, - ) : this(null, handlers.handlers, headers.toBuilder().remove("Idempotency-Key").build()) + creation: SessionCreateParams, + ) : this( + null, + handlers.handlers, + creation._headers().toBuilder().remove("Idempotency-Key").build(), + creation.input().isPresent, + ) private var sessionId = params?.sessionId private val mapper = jsonMapper() @@ -84,7 +91,8 @@ private constructor( return true } - fun terminal(event: AgentSessionEvent) = event.isFailed() || (event.isIdle() && turnEnded) + fun terminal(event: AgentSessionEvent) = + event.isFailed() || (event.isIdle() && (!expectsTurn || turnEnded)) fun result(event: AgentSessionEvent): CompletableFuture { val call = diff --git a/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt b/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt index 1a882759e..1a36048f9 100644 --- a/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt +++ b/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt @@ -349,6 +349,63 @@ internal class AgentSessionStreamTest { assertThat(t.streamClosed).isTrue() } + @ParameterizedTest + @ValueSource(booleans = [false, true]) + fun `inputless self hosted creation ends at idle without fabricating a turn`(async: Boolean) { + val params = + SessionCreateParams.builder() + .agentId("agent") + .selfHostedEnvironment("/workspace") + .build() + val handlers = + AgentToolHandlers.builder() + .toolHandler("tool") { error("No function call was received") } + .build() + for (collect in listOf(false, true)) { + val t = Transport(listOf(creationEvents().first(), creationEvents().last())) + val client = t.client() + val seen = mutableListOf() + val failure = + if (async) { + val stream = + client.async().beta().agents().sessions().createStreaming(params, handlers) + if (collect) + runCatching { + AgentTurnResults.getFinalResult(stream) + .get(ASYNC_TIMEOUT_SECONDS, TimeUnit.SECONDS) + } + .exceptionOrNull() + ?.let(::unwrap) + else { + stream + .subscribe { seen.add(it) } + .onCompleteFuture() + .get(ASYNC_TIMEOUT_SECONDS, TimeUnit.SECONDS) + null + } + } else + client.beta().agents().sessions().createStreaming(params, handlers).use { stream + -> + if (collect) + runCatching { AgentTurnResults.getFinalResult(stream) } + .exceptionOrNull() + else { + stream.stream().forEach { seen.add(it) } + null + } + } + if (collect) { + assertThat(failure).isInstanceOf(AgentTurnResultException::class.java) + assertThat((failure as AgentTurnResultException).reason()) + .isEqualTo(AgentTurnResultException.Reason.INCOMPLETE_STREAM) + } else + assertThat(seen.map { it.isCreated() || it.isIdle() }).containsExactly(true, true) + assertThat(t.requests).hasSize(1) + assertThat(t.posts).isEmpty() + assertThat(t.streamClosed).isTrue() + } + } + private fun params() = AgentSessionStreamParams.builder().sessionId("s").input("hello") class LookupItem(val itemId: String) From f910f8e7490316653103819a9f6fbba83ba0c455 Mon Sep 17 00:00:00 2001 From: Alex Chang Date: Mon, 5 Oct 2026 16:12:25 +0000 Subject: [PATCH 4/5] fix(agents): preserve raw inputs and follow-up diagnostics --- .../beta/agents/AgentSessionStreamSupport.kt | 7 ++--- .../beta/agents/AgentSessionStreamTest.kt | 27 +++++++++++++++++++ 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt index 42cc7b4dd..a39e9028f 100644 --- a/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt +++ b/openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamSupport.kt @@ -34,9 +34,10 @@ private constructor( null, handlers.handlers, creation._headers().toBuilder().remove("Idempotency-Key").build(), - creation.input().isPresent, + !creation._input().isMissing() && !creation._input().isNull(), ) + private val onToolError = params?.onToolError private var sessionId = params?.sessionId private val mapper = jsonMapper() private val inputKey = params?.idempotencyKey ?: UUID.randomUUID().toString() @@ -155,11 +156,11 @@ private constructor( if (failure is AgentToolArgumentException) AgentToolError.Stage.ARGUMENTS else stage try { - params.onToolError?.accept( + onToolError?.accept( AgentToolError( original, call.name(), - params.sessionId, + checkNotNull(sessionId), call.turnId(), call.callId(), failureStage, diff --git a/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt b/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt index 1ff1301af..bbf0c650e 100644 --- a/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt +++ b/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt @@ -406,6 +406,33 @@ internal class AgentSessionStreamTest { } } + @ParameterizedTest + @ValueSource(booleans = [false, true]) + fun `creation forwards unknown input shapes without typed decoding`(async: Boolean) { + val input = mapOf("future_input" to "value") + val params = + creationParams().toBuilder().input(com.openai.core.JsonValue.from(input)).build() + val handlers = AgentToolHandlers.builder().toolHandler("tool") { "Result" }.build() + val t = Transport(creationEvents()) + val client = t.client() + if (async) + client + .async() + .beta() + .agents() + .sessions() + .createStreaming(params, handlers) + .subscribe {} + .onCompleteFuture() + .get(ASYNC_TIMEOUT_SECONDS, TimeUnit.SECONDS) + else + client.beta().agents().sessions().createStreaming(params, handlers).use { + it.stream().forEach {} + } + assertThat(t.creationBody!!.path("input")).isEqualTo(mapper.valueToTree(input)) + assertThat(t.posts).hasSize(1) + } + private fun params() = AgentSessionStreamParams.builder().sessionId("s").input("hello") class LookupItem(val itemId: String) From 41a3ef1c74f409da74a33e362d30ecf9e3c357dc Mon Sep 17 00:00:00 2001 From: Alex Chang Date: Mon, 5 Oct 2026 16:28:13 +0000 Subject: [PATCH 5/5] test(agents): keep stream regression tests consumer-compatible --- .../com/openai/services/beta/agents/AgentSessionStreamTest.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt b/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt index bbf0c650e..681bb571d 100644 --- a/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt +++ b/openai-java-core/src/test/kotlin/com/openai/services/beta/agents/AgentSessionStreamTest.kt @@ -375,7 +375,7 @@ internal class AgentSessionStreamTest { .get(ASYNC_TIMEOUT_SECONDS, TimeUnit.SECONDS) } .exceptionOrNull() - ?.let(::unwrap) + ?.cause else { stream .subscribe { seen.add(it) }