diff --git a/docs/helpers/agent-sessions.md b/docs/helpers/agent-sessions.md index 6f8664225..716bac118 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..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 @@ -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) 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..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 @@ -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) 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 026dfca75..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 @@ -2,21 +2,45 @@ 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.* +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 import java.util.UUID import java.util.concurrent.CompletableFuture import java.util.concurrent.CompletionException +import java.util.concurrent.CompletionStage import java.util.concurrent.ExecutionException +import java.util.function.Function import kotlin.jvm.optionals.getOrNull -internal class AgentSessionStreamSupport(val params: AgentSessionStreamParams) { +internal class AgentSessionStreamSupport +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, + creation: SessionCreateParams, + ) : this( + null, + handlers.handlers, + creation._headers().toBuilder().remove("Idempotency-Key").build(), + !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() + private val inputKey = params?.idempotencyKey ?: UUID.randomUUID().toString() private val eventIds = linkedSetOf() private val handledCalls = hashSetOf>() private var turnId: String? = null @@ -24,22 +48,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 +73,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) @@ -67,13 +92,15 @@ internal class AgentSessionStreamSupport(val params: AgentSessionStreamParams) { 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 = 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) var stage = AgentToolError.Stage.ARGUMENTS @@ -129,11 +156,11 @@ internal class AgentSessionStreamSupport(val params: AgentSessionStreamParams) { 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, @@ -148,9 +175,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 846ff97ff..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 @@ -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,194 @@ 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() + } + + @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() + ?.cause + 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() + } + } + + @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)