items = createResponse.input().get().asResponse();
+ boolean needsReparse = false;
+ for (ResponseInputItem item : items) {
+ if (!item.isEasyInputMessage() && !item.isMessage()
+ && !item.isFunctionCall() && !item.isFunctionCallOutput()
+ && !item.isComputerCall() && !item.isComputerCallOutput()
+ && !item.isItemReference() && !item.isCodeInterpreterCall()
+ && item._json().isPresent()) {
+ needsReparse = true;
+ break;
+ }
+ }
+
+ if (needsReparse && node.has("input") && node.get("input").isArray()) {
+ ArrayNode inputArray = (ArrayNode) node.get("input");
+ boolean modified = false;
+ for (int i = 0; i < inputArray.size(); i++) {
+ JsonNode element = inputArray.get(i);
+ if (element.isObject() && !element.has("type")
+ && element.has("role") && element.has("content")) {
+ ((ObjectNode) element).put("type", "message");
+ modified = true;
+ }
+ }
+ if (modified) {
+ LOGGER.debug("Injected 'type: message' into input items missing type discriminator");
+ createResponse = p.getCodec().treeToValue(node, ResponseCreateParams.Body.class);
+ }
+ }
+ }
+ return createResponse;
+ }
+ }
+}
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponseContext.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponseContext.java
new file mode 100644
index 000000000000..f08ba31b8568
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponseContext.java
@@ -0,0 +1,355 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.microsoft.agentserver.api.implementation.IdGenerator;
+import com.microsoft.agentserver.api.implementation.ItemConversion;
+import com.openai.models.responses.ResponseCreateParams;
+import com.openai.models.responses.ResponseInputItem;
+import com.openai.models.responses.ResponseItem;
+import com.openai.models.responses.ResponseOutputItem;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Enhanced implementation of {@link ResponseContext} that resolves input items
+ * and conversation history from the request, using lazy-cached async resolution.
+ * Inline items are converted via {@link ItemConversion}; item references are
+ * resolved via {@link ResponsesProvider#getItemsAsync}.
+ *
+ * Accepts either {@link ResponseCreateParams} or {@link ResponseCreateParams.Body}
+ * as input — both expose the same API surface ({@code input()},
+ * {@code previousResponseId()}, {@code conversation()}, etc.).
+ */
+final class AgentServerResponseContext implements ResponseContext {
+
+ /**
+ * Default maximum number of history items to fetch.
+ */
+ static final int DEFAULT_FETCH_HISTORY_COUNT = 100;
+
+ private final String responseId;
+ private final ResponsesProvider provider;
+ private final ResponseCreateParams.Body request;
+ private final JsonNode rawBody;
+ private final int historyLimit;
+ private final IsolationContext isolation;
+ private final Map clientHeaders;
+ private final Map queryParameters;
+ private final String requestId;
+ private final String sessionId;
+ // Lazy-cached async results (AtomicReference for thread-safe single-init)
+ private final AtomicReference>> inputItemsRef =
+ new AtomicReference<>();
+ private final AtomicReference>> historyItemIdsRef =
+ new AtomicReference<>();
+ private final AtomicReference>> historyRef =
+ new AtomicReference<>();
+ private volatile boolean shutdownRequested;
+ private volatile boolean cancelled;
+
+ /**
+ * Initializes a new instance of {@code AgentServerResponseContext}.
+ *
+ * Use {@link ResponseContext#builder()} to construct instances via the fluent builder API.
+ *
+ * @param responseId the unique response identifier.
+ * @param provider the responses provider for resolving item references and history.
+ * @param request the create-response request body containing input items.
+ * @param rawBody the full raw JSON request body, or {@code null} if not available.
+ * @param historyLimit maximum number of history items to fetch, or {@code null} for the default (100).
+ * @param isolation the isolation context, or {@code null} for {@link IsolationContext#EMPTY}.
+ * @param clientHeaders the forwarded client headers, or {@code null} for empty.
+ * @param queryParameters the query parameters, or {@code null} for empty.
+ * @param requestId the request ID for correlation, or {@code null}.
+ * @param sessionId the resolved session ID per {@code SessionIdResolver}, or {@code null}.
+ */
+ AgentServerResponseContext(
+ String responseId,
+ ResponsesProvider provider,
+ ResponseCreateParams.Body request,
+ JsonNode rawBody,
+ Integer historyLimit,
+ IsolationContext isolation,
+ Map clientHeaders,
+ Map queryParameters,
+ String requestId,
+ String sessionId) {
+ this.responseId = Objects.requireNonNull(responseId, "responseId must not be null");
+ this.provider = Objects.requireNonNull(provider, "provider must not be null");
+ this.request = Objects.requireNonNull(request, "request must not be null");
+ this.rawBody = rawBody;
+ this.historyLimit = historyLimit != null ? historyLimit : DEFAULT_FETCH_HISTORY_COUNT;
+ this.isolation = isolation != null ? isolation : IsolationContext.EMPTY;
+ this.clientHeaders = clientHeaders != null ? clientHeaders : Collections.emptyMap();
+ this.queryParameters = queryParameters != null ? queryParameters : Collections.emptyMap();
+ this.requestId = requestId;
+ this.sessionId = sessionId;
+ }
+
+ /**
+ * Backward-compatible constructor without platform headers.
+ */
+ AgentServerResponseContext(
+ String responseId,
+ ResponsesProvider provider,
+ ResponseCreateParams.Body request,
+ JsonNode rawBody,
+ Integer historyLimit) {
+ this(responseId, provider, request, rawBody, historyLimit, null, null, null, null, null);
+ }
+
+ /**
+ * Converts a {@link ResponseItem} back to a {@link ResponseOutputItem} for input resolution.
+ *
+ * Note: {@code FunctionCall} items cannot be converted because
+ * {@link ResponseItem#asFunctionCall()} returns {@code ResponseFunctionToolCallItem}
+ * (input variant) while {@link ResponseOutputItem#ofFunctionCall} requires
+ * {@code ResponseFunctionToolCall} (output variant).
+ */
+ private static ResponseOutputItem toOutputItem(ResponseItem item) {
+ if (item.isResponseOutputMessage()) {
+ return ResponseOutputItem.ofMessage(item.asResponseOutputMessage());
+ }
+ if (item.isFileSearchCall()) {
+ return ResponseOutputItem.ofFileSearchCall(item.asFileSearchCall());
+ }
+ if (item.isWebSearchCall()) {
+ return ResponseOutputItem.ofWebSearchCall(item.asWebSearchCall());
+ }
+ if (item.isComputerCall()) {
+ return ResponseOutputItem.ofComputerCall(item.asComputerCall());
+ }
+ if (item.isReasoning()) {
+ return ResponseOutputItem.ofReasoning(item.asReasoning());
+ }
+ if (item.isCodeInterpreterCall()) {
+ return ResponseOutputItem.ofCodeInterpreterCall(item.asCodeInterpreterCall());
+ }
+ if (item.isShellCall()) {
+ return ResponseOutputItem.ofShellCall(item.asShellCall());
+ }
+ if (item.isShellCallOutput()) {
+ return ResponseOutputItem.ofShellCallOutput(item.asShellCallOutput());
+ }
+ if (item.isApplyPatchCall()) {
+ return ResponseOutputItem.ofApplyPatchCall(item.asApplyPatchCall());
+ }
+ if (item.isApplyPatchCallOutput()) {
+ return ResponseOutputItem.ofApplyPatchCallOutput(item.asApplyPatchCallOutput());
+ }
+ return null;
+ }
+
+ /**
+ * Lazily initializes and caches a {@link CompletableFuture} in the given
+ * {@link AtomicReference}, ensuring at most one computation.
+ */
+ private static CompletableFuture getOrInit(
+ AtomicReference> ref,
+ java.util.function.Supplier> supplier) {
+ CompletableFuture existing = ref.get();
+ if (existing != null) {
+ return existing;
+ }
+ CompletableFuture created = supplier.get();
+ if (ref.compareAndSet(null, created)) {
+ return created;
+ }
+ return ref.get();
+ }
+
+ private static boolean isNullOrEmpty(String s) {
+ return s == null || s.isEmpty();
+ }
+
+ @Override
+ public String getResponseId() {
+ return responseId;
+ }
+
+ @Override
+ public boolean isShutdownRequested() {
+ return shutdownRequested;
+ }
+
+ /**
+ * Sets whether the server is shutting down.
+ * This is called by the hosting infrastructure, not by handlers.
+ */
+ public void setShutdownRequested(boolean shutdownRequested) {
+ this.shutdownRequested = shutdownRequested;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return cancelled;
+ }
+
+ @Override
+ public void cancel() {
+ this.cancelled = true;
+ }
+
+ @Override
+ public JsonNode getRawBody() {
+ return rawBody;
+ }
+
+ @Override
+ public IsolationContext getIsolation() {
+ return isolation;
+ }
+
+ @Override
+ public Map getClientHeaders() {
+ return clientHeaders;
+ }
+
+ @Override
+ public Map getQueryParameters() {
+ return queryParameters;
+ }
+
+ @Override
+ public String getRequestId() {
+ return requestId;
+ }
+
+ /**
+ * Package-private accessor for the original request body (needed by API helpers).
+ */
+ ResponseCreateParams.Body getRequestBody() {
+ return request;
+ }
+
+ @Override
+ public String getSessionId() {
+ return sessionId;
+ }
+
+ // ── Private resolution methods ──────────────────────────────
+
+ @Override
+ public CompletableFuture> getInputItemsAsync() {
+ return getOrInit(inputItemsRef, this::resolveInputItemsAsync);
+ }
+
+ @Override
+ public CompletableFuture> getHistoryAsync() {
+ return getOrInit(historyRef, this::resolveHistoryAsync);
+ }
+
+ /**
+ * Gets the cached history item IDs. Used by the orchestrator to pass IDs
+ * to the provider without duplicating storage.
+ *
+ * @return a future containing the resolved history item IDs.
+ */
+ public CompletableFuture> getHistoryItemIdsAsync() {
+ return getOrInit(historyItemIdsRef, this::resolveHistoryItemIdsAsync);
+ }
+
+ private CompletableFuture> resolveInputItemsAsync() {
+ Optional inputOpt = request.input();
+ if (inputOpt.isEmpty()) {
+ return CompletableFuture.completedFuture(Collections.emptyList());
+ }
+
+ ResponseCreateParams.Input input = inputOpt.get();
+
+ // If the input is a plain text string, there are no structured items to resolve
+ if (input.isText()) {
+ return CompletableFuture.completedFuture(Collections.emptyList());
+ }
+
+ List items = input.asResponse();
+ if (items.isEmpty()) {
+ return CompletableFuture.completedFuture(Collections.emptyList());
+ }
+
+ IdGenerator idGen = new IdGenerator(IdGenerator.extractPartitionKey(responseId));
+
+ // Separate inline items from item references
+ List results = new ArrayList<>(items.size());
+ List referenceIds = new ArrayList<>();
+ List referencePositions = new ArrayList<>();
+
+ for (ResponseInputItem item : items) {
+ if (item.isItemReference()) {
+ referenceIds.add(item.asItemReference().id());
+ referencePositions.add(results.size());
+ results.add(null); // placeholder — index matches results.size()
+ } else {
+ ResponseOutputItem output = ItemConversion.toOutputItem(item, idGen);
+ if (output != null) {
+ results.add(output);
+ }
+ // Non-convertible, non-reference items are silently skipped
+ }
+ }
+
+ // If no references, return the inline results immediately
+ if (referenceIds.isEmpty()) {
+ return CompletableFuture.completedFuture(Collections.unmodifiableList(results));
+ }
+
+ // Batch-resolve references via the provider
+ return provider.getItemsAsync(referenceIds)
+ .thenApply(resolved -> {
+ for (int i = 0; i < referencePositions.size(); i++) {
+ int pos = referencePositions.get(i);
+ if (i < resolved.size() && resolved.get(i) != null) {
+ ResponseOutputItem converted = toOutputItem(resolved.get(i));
+ if (converted != null) {
+ results.set(pos, converted);
+ }
+ }
+ }
+ // Remove unresolved placeholders (nulls remaining from failed references)
+ return results.stream()
+ .filter(Objects::nonNull)
+ .toList();
+ });
+ }
+
+ // ── Utility ─────────────────────────────────────────────────
+
+ private CompletableFuture> resolveHistoryItemIdsAsync() {
+ String previousResponseId = request.previousResponseId().orElse(null);
+ String conversationId = request.conversation()
+ .flatMap(conv -> conv.isId() ? Optional.of(conv.asId()) : Optional.empty())
+ .orElse(null);
+
+ if (isNullOrEmpty(previousResponseId) && isNullOrEmpty(conversationId)) {
+ return CompletableFuture.completedFuture(Collections.emptyList());
+ }
+
+ return provider.getHistoryItemIdsAsync(previousResponseId, conversationId, historyLimit);
+ }
+
+ private CompletableFuture> resolveHistoryAsync() {
+ return getHistoryItemIdsAsync()
+ .thenCompose(ids -> {
+ if (ids.isEmpty()) {
+ return CompletableFuture.completedFuture(Collections.emptyList());
+ }
+ return provider.getItemsAsync(ids)
+ .thenApply(items -> items.stream()
+ .filter(Objects::nonNull)
+ .toList());
+ });
+ }
+}
+
+
+
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponseEventStream.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponseEventStream.java
new file mode 100644
index 000000000000..064c299b67dc
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponseEventStream.java
@@ -0,0 +1,1217 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import com.microsoft.agentserver.api.implementation.IdGenerator;
+import com.openai.core.JsonNull;
+import com.openai.core.http.StreamResponse;
+import com.openai.models.chat.completions.ChatCompletionChunk;
+import com.openai.models.responses.Response;
+import com.openai.models.responses.ResponseCodeInterpreterToolCall;
+import com.openai.models.responses.ResponseCompletedEvent;
+import com.openai.models.responses.ResponseContentPartAddedEvent;
+import com.openai.models.responses.ResponseContentPartDoneEvent;
+import com.openai.models.responses.ResponseCreatedEvent;
+import com.openai.models.responses.ResponseCustomToolCall;
+import com.openai.models.responses.ResponseError;
+import com.openai.models.responses.ResponseFailedEvent;
+import com.openai.models.responses.ResponseFileSearchToolCall;
+import com.openai.models.responses.ResponseFunctionCallArgumentsDeltaEvent;
+import com.openai.models.responses.ResponseFunctionCallArgumentsDoneEvent;
+import com.openai.models.responses.ResponseFunctionToolCall;
+import com.openai.models.responses.ResponseFunctionWebSearch;
+import com.openai.models.responses.ResponseInProgressEvent;
+import com.openai.models.responses.ResponseIncompleteEvent;
+import com.openai.models.responses.ResponseOutputItem;
+import com.openai.models.responses.ResponseOutputItemAddedEvent;
+import com.openai.models.responses.ResponseOutputItemDoneEvent;
+import com.openai.models.responses.ResponseOutputMessage;
+import com.openai.models.responses.ResponseOutputText;
+import com.openai.models.responses.ResponseQueuedEvent;
+import com.openai.models.responses.ResponseReasoningItem;
+import com.openai.models.responses.ResponseStatus;
+import com.openai.models.responses.ResponseStreamEvent;
+import com.openai.models.responses.ResponseTextDeltaEvent;
+import com.openai.models.responses.ResponseTextDoneEvent;
+import com.openai.models.responses.ResponseUsage;
+import com.openai.models.responses.ToolChoiceOptions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Flow;
+import java.util.concurrent.SubmissionPublisher;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+
+/**
+ * Fluent builder for constructing a streaming response event sequence.
+ *
+ * Manages a global sequence counter, output index counter, and an accumulated
+ * {@link Response} snapshot. Every {@code emit*()} method appends a {@link ResponseEvent}
+ * to the internal list and returns {@code this} for chaining.
+ *
+ * Nested output-item scopes use the {@link Consumer} callback pattern so that
+ * builder lifecycles are lexically scoped and automatically finalized.
+ *
+ * Call {@link #subscribe(Consumer, Consumer, Runnable)} to receive events
+ * in real-time as they are produced — including text deltas from deeply nested builders.
+ * Events are delivered via a {@link SubmissionPublisher} from the JDK's
+ * {@link Flow} reactive-streams API.
+ *
+ *
Threading model
+ *
+ * This class is designed for a single-writer pattern: all {@code emit*()}
+ * and {@code addOutput*()} methods must be called from a single thread (or with external
+ * serialization). The {@link #subscribe(Consumer, Consumer, Runnable)} callbacks are
+ * invoked by the {@link SubmissionPublisher}'s executor. Internal synchronization ensures
+ * safe hand-off between the writer and subscriber.
+ *
+ *
Example usage (synchronous)
+ * {@code
+ * ResponseEventStream stream = ResponseEventStream.create(request, context)
+ * .emitCreated()
+ * .emitInProgress()
+ * .addOutputMessage(msg -> msg
+ * .emitAdded()
+ * .addTextPart(text -> text
+ * .emitAdded()
+ * .emitDelta("Hello!")
+ * .emitDone("Hello!"))
+ * .emitDone())
+ * .emitCompleted();
+ *
+ * List events = stream.getEvents();
+ * }
+ *
+ * Example usage (reactive streaming)
+ * {@code
+ * ResponseEventStream stream = ResponseEventStream.create(request, context);
+ * stream.subscribe(
+ * event -> System.out.println(event.eventName()),
+ * failure -> failure.printStackTrace(),
+ * () -> System.out.println("Done"));
+ *
+ * // Build on another thread — subscribers see events immediately.
+ * // awaitSubscription() blocks until subscribe() has been called,
+ * // preventing events from being emitted before a subscriber is ready.
+ * executor.submit(() -> {
+ * stream.awaitSubscription();
+ * stream.emitCreated()
+ * .emitInProgress()
+ * .addOutputMessage(msg -> msg
+ * .emitAdded()
+ * .addTextPart(text -> text
+ * .emitAdded()
+ * .emitDelta("Hello!")
+ * .emitDone("Hello!"))
+ * .emitDone())
+ * .emitCompleted();
+ * });
+ * }
+ */
+final class AgentServerResponseEventStream implements ResponseEventStream {
+
+ private final IdGenerator idGenerator;
+ private final String responseId;
+ // Accumulated events and output items.
+ private final CopyOnWriteArrayList events = new CopyOnWriteArrayList<>();
+ // Concurrent map because trackCompletedOutputItem() may be called from nested
+ // builders or forwardUpstream() while a subscriber is consuming events on a
+ // different thread (e.g., via the SubmissionPublisher executor).
+ private final Map outputItems = new java.util.concurrent.ConcurrentSkipListMap<>();
+ // JDK reactive publisher for push-based event delivery.
+ // Synchronous executor (Runnable::run) ensures events are delivered inline
+ // on the writer thread, matching the original Multi emitter semantics.
+ private final SubmissionPublisher publisher =
+ new SubmissionPublisher<>(Runnable::run, Flow.defaultBufferSize());
+ private final CountDownLatch subscriptionLatch = new CountDownLatch(1);
+ private final AtomicLong sequenceNumber = new AtomicLong();
+ private final AtomicLong outputIndex = new AtomicLong();
+ private final AtomicBoolean terminated = new AtomicBoolean();
+ // Mutable response builder – rebuilt into an immutable snapshot per event.
+ // The OpenAI SDK builder is mutable (setters return `this`), but we reassign
+ // defensively so the code also works if the builder were ever replaced with
+ // an immutable variant.
+ // THREADING: all reads/writes of responseBuilder MUST hold builderLock.
+ private Response.Builder responseBuilder;
+ private final Object builderLock = new Object();
+
+ // ── Construction ──────────────────────────────────────────
+
+ private AgentServerResponseEventStream(AgentServerCreateResponse request, ResponseContext context) {
+ Objects.requireNonNull(context, "context must not be null");
+ Objects.requireNonNull(request, "request must not be null");
+
+ this.idGenerator = new IdGenerator(IdGenerator.extractPartitionKey(context.getResponseId()));
+ this.responseId = context.getResponseId();
+
+ String conversationId = request.responseCreateParams().conversation()
+ .flatMap(conv -> {
+ if (conv.isId()) {
+ return Optional.of(conv.asId());
+ }
+ if (conv.isResponseConversationParam()) {
+ return Optional.of(conv.asResponseConversationParam().id());
+ }
+ return Optional.empty();
+ })
+ .orElse(null);
+
+ this.responseBuilder = Response.builder()
+ .id(context.getResponseId())
+ .createdAt(System.currentTimeMillis() / 1000.0)
+ .status(ResponseStatus.QUEUED)
+ .error(Optional.empty())
+ .incompleteDetails(Optional.empty())
+ .instructions(Optional.empty())
+ .metadata(Optional.empty())
+ .parallelToolCalls(false)
+ .temperature(Optional.empty())
+ .toolChoice(ToolChoiceOptions.AUTO)
+ .tools(List.of())
+ .topP(Optional.empty())
+ .output(List.of())
+ .previousResponseId(request.responseCreateParams().previousResponseId().orElse(null))
+ .background(request.responseCreateParams().background().orElse(null))
+ .putAdditionalProperty("agent_id", new JsonNull());
+
+ if (request.responseCreateParams().model().isPresent()) {
+ request.responseCreateParams().model().ifPresent(this.responseBuilder::model);
+ } else {
+ String modelName = FoundryEnvironment.AGENT_NAME != null
+ ? FoundryEnvironment.AGENT_NAME : "unknown-agent";
+ this.responseBuilder.model(modelName);
+ }
+
+ if (conversationId != null) {
+ this.responseBuilder = this.responseBuilder.conversation(
+ Response.Conversation.builder().id(conversationId).build());
+ }
+ }
+
+ /**
+ * Creates a new event stream from a request and context.
+ */
+ public static ResponseEventStream create(ResponseContext context, AgentServerCreateResponse request) {
+ return new AgentServerResponseEventStream(request, context);
+ }
+
+ // ── Results ───────────────────────────────────────────────
+
+ /**
+ * Resolves the SSE event name for a given {@link ResponseStreamEvent}.
+ * Returns {@code null} for unrecognized or lifecycle events.
+ */
+ private static String resolveEventName(ResponseStreamEvent event) {
+ if (event.isOutputItemAdded()) return "response.output_item.added";
+ if (event.isOutputItemDone()) return "response.output_item.done";
+ if (event.isOutputTextDelta()) return "response.output_text.delta";
+ if (event.isOutputTextDone()) return "response.output_text.done";
+ if (event.isContentPartAdded()) return "response.content_part.added";
+ if (event.isContentPartDone()) return "response.content_part.done";
+ if (event.isFunctionCallArgumentsDelta()) return "response.function_call_arguments.delta";
+ if (event.isFunctionCallArgumentsDone()) return "response.function_call_arguments.done";
+ // Add more as needed for reasoning, file search, etc.
+ return null;
+ }
+
+ /**
+ * Creates a default zero-usage object with all required fields populated.
+ * The OpenAI SDK requires {@code inputTokensDetails} and {@code outputTokensDetails}
+ * even when usage is not tracked.
+ */
+ private static ResponseUsage defaultUsage() {
+ return ResponseUsage.builder()
+ .inputTokens(0)
+ .outputTokens(0)
+ .totalTokens(0)
+ .inputTokensDetails(ResponseUsage.InputTokensDetails.builder()
+ .cachedTokens(0)
+ .build())
+ .outputTokensDetails(ResponseUsage.OutputTokensDetails.builder()
+ .reasoningTokens(0)
+ .build())
+ .build();
+ }
+
+ // ── Reactive Stream ───────────────────────────────────────
+
+ /**
+ * Returns an unmodifiable view of all events emitted so far.
+ */
+ @Override
+ public List getEvents() {
+ return Collections.unmodifiableList(events);
+ }
+
+ /**
+ * Returns a snapshot of the current {@link Response} being constructed.
+ */
+ @Override
+ public Response getResponse() {
+ synchronized (builderLock) {
+ return responseBuilder.build();
+ }
+ }
+
+ /**
+ * Stamps the resolved {@code agent_session_id} onto every snapshot
+ * produced by this stream from this point forward. No-op for null/empty input.
+ * Package-private — invoked by {@link AgentServerResponsesApi} once the
+ * session ID is resolved for the current request.
+ */
+ void setAgentSessionId(String sessionId) {
+ if (sessionId == null || sessionId.isEmpty()) {
+ return;
+ }
+ synchronized (builderLock) {
+ responseBuilder = responseBuilder.putAdditionalProperty(
+ "agent_session_id", com.openai.core.JsonValue.from(sessionId));
+ }
+ }
+
+ // ── Response Lifecycle Events ─────────────────────────────
+
+ /**
+ * Subscribes to this event stream with the given callbacks.
+ *
+ * Every event produced by any builder — including deeply nested text deltas —
+ * is pushed to the {@code onEvent} callback immediately. If events have already
+ * been emitted before subscription, they are replayed first.
+ *
+ * Terminal events ({@code emitCompleted}, {@code emitFailed}, {@code emitIncomplete})
+ * automatically invoke {@code onComplete}.
+ *
+ * This method is synchronized on the same monitor as {@link #addEvent} to
+ * guarantee that no events are lost between the replay loop and the
+ * {@link SubmissionPublisher} subscription.
+ *
+ * @param onEvent called for each {@link ResponseEvent} as it is produced
+ * @param onFailure called if the stream encounters an error
+ * @param onComplete called when the stream terminates normally
+ */
+ @Override
+ public synchronized void subscribe(Consumer onEvent, Consumer onFailure, Runnable onComplete) {
+ // Replay events already emitted before subscription.
+ for (ResponseEvent event : events) {
+ onEvent.accept(event);
+ }
+
+ // If the stream already terminated, signal completion immediately.
+ if (terminated.get()) {
+ subscriptionLatch.countDown();
+ onComplete.run();
+ return;
+ }
+
+ // Wire up a Flow.Subscriber that delegates to the provided callbacks.
+ // Because addEvent() is also synchronized on `this`, no submit() can
+ // happen between the replay above and the registration below.
+ publisher.subscribe(new Flow.Subscriber<>() {
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(ResponseEvent event) {
+ onEvent.accept(event);
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ onFailure.accept(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ onComplete.run();
+ }
+ });
+
+ // Signal any threads waiting in awaitSubscription().
+ subscriptionLatch.countDown();
+ }
+
+ /**
+ * Blocks the calling thread until {@link #subscribe} has been called,
+ * ensuring that no events are emitted before a subscriber is ready.
+ * Times out after 30 seconds to avoid deadlocks.
+ *
+ * @throws InterruptedException if the thread is interrupted while waiting
+ */
+ @Override
+ public void awaitSubscription() throws InterruptedException {
+ if (!subscriptionLatch.await(30, TimeUnit.SECONDS)) {
+ throw new IllegalStateException(
+ "Timed out waiting for a subscriber to be registered. " +
+ "Ensure subscribe() is called within 30 seconds of stream creation.");
+ }
+ }
+
+ /**
+ * Emits a {@code response.queued} event. Sets status to {@link ResponseStatus#QUEUED}.
+ */
+ @Override
+ public ResponseEventStream emitQueued() {
+ synchronized (builderLock) {
+ responseBuilder = responseBuilder.status(ResponseStatus.QUEUED);
+ addEvent(new ResponseEvent("response.queued",
+ ResponseStreamEvent.ofQueued(new ResponseQueuedEvent.Builder()
+ .sequenceNumber(nextSequenceNumber())
+ .response(snapshot())
+ .build())));
+ }
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.created} event with status {@link ResponseStatus#IN_PROGRESS}.
+ */
+ @Override
+ public ResponseEventStream emitCreated() {
+ return emitCreated(ResponseStatus.IN_PROGRESS);
+ }
+
+ /**
+ * Emits a {@code response.created} event with the given status.
+ */
+ @Override
+ public ResponseEventStream emitCreated(ResponseStatus status) {
+ synchronized (builderLock) {
+ responseBuilder = responseBuilder.status(status);
+ }
+ addEvent(new ResponseEvent("response.created",
+ ResponseStreamEvent.ofCreated(new ResponseCreatedEvent.Builder()
+ .sequenceNumber(nextSequenceNumber())
+ .response(snapshot())
+ .build())));
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.in_progress} event. Sets status to {@link ResponseStatus#IN_PROGRESS}.
+ */
+ @Override
+ public ResponseEventStream emitInProgress() {
+ synchronized (builderLock) {
+ responseBuilder = responseBuilder.status(ResponseStatus.IN_PROGRESS);
+ }
+ addEvent(new ResponseEvent("response.in_progress",
+ ResponseStreamEvent.ofInProgress(new ResponseInProgressEvent.Builder()
+ .sequenceNumber(nextSequenceNumber())
+ .response(snapshot())
+ .build())));
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.completed} event with no usage data.
+ */
+ @Override
+ public ResponseEventStream emitCompleted() {
+ return emitCompleted(null);
+ }
+
+ /**
+ * Emits a {@code response.completed} event with optional usage data.
+ *
+ * @throws IllegalStateException if a terminal event has already been emitted
+ */
+ @Override
+ public ResponseEventStream emitCompleted(ResponseUsage usage) {
+ guardTerminal();
+ // Default to zero-usage if none provided — clients may require the field.
+ ResponseUsage effectiveUsage = usage != null ? usage : defaultUsage();
+ synchronized (builderLock) {
+ responseBuilder = responseBuilder
+ .status(ResponseStatus.COMPLETED)
+ .completedAt(System.currentTimeMillis() / 1000.0)
+ .output(collectOutputItems())
+ .usage(effectiveUsage);
+ }
+ addEvent(new ResponseEvent("response.completed",
+ ResponseStreamEvent.ofCompleted(new ResponseCompletedEvent.Builder()
+ .sequenceNumber(nextSequenceNumber())
+ .response(snapshot())
+ .build())));
+ completeMulti();
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.failed} event with a server error.
+ */
+ @Override
+ public ResponseEventStream emitFailed() {
+ return emitFailed(ResponseError.Code.SERVER_ERROR, "An internal server error occurred.", null);
+ }
+
+ /**
+ * Emits a {@code response.failed} event with the given error.
+ */
+ @Override
+ public ResponseEventStream emitFailed(ResponseError.Code code, String message) {
+ return emitFailed(code, message, null);
+ }
+
+ /**
+ * Emits a {@code response.failed} event with the given error and optional usage data.
+ *
+ * @throws IllegalStateException if a terminal event has already been emitted
+ */
+ @Override
+ public ResponseEventStream emitFailed(ResponseError.Code code, String message, ResponseUsage usage) {
+ guardTerminal();
+ ResponseUsage effectiveUsage = usage != null ? usage : defaultUsage();
+ synchronized (builderLock) {
+ responseBuilder = responseBuilder
+ .status(ResponseStatus.FAILED)
+ .completedAt(System.currentTimeMillis() / 1000.0)
+ .error(ResponseError.builder().code(code).message(message).build())
+ .output(collectOutputItems())
+ .usage(effectiveUsage);
+ }
+ addEvent(new ResponseEvent("response.failed",
+ ResponseStreamEvent.ofFailed(new ResponseFailedEvent.Builder()
+ .sequenceNumber(nextSequenceNumber())
+ .response(snapshot())
+ .build())));
+ completeMulti();
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.incomplete} event with no details.
+ */
+ @Override
+ public ResponseEventStream emitIncomplete() {
+ return emitIncomplete(null, null);
+ }
+
+ // ── Output Item Scope Factories (Consumer pattern) ────────
+
+ /**
+ * Emits a {@code response.incomplete} event with the given reason.
+ */
+ @Override
+ public ResponseEventStream emitIncomplete(Response.IncompleteDetails.Reason reason) {
+ return emitIncomplete(reason, null);
+ }
+
+ /**
+ * Emits a {@code response.incomplete} event with optional reason and usage.
+ *
+ * @throws IllegalStateException if a terminal event has already been emitted
+ */
+ @Override
+ public ResponseEventStream emitIncomplete(Response.IncompleteDetails.Reason reason, ResponseUsage usage) {
+ guardTerminal();
+ ResponseUsage effectiveUsage = usage != null ? usage : defaultUsage();
+ synchronized (builderLock) {
+ responseBuilder = responseBuilder
+ .status(ResponseStatus.INCOMPLETE)
+ .completedAt(System.currentTimeMillis() / 1000.0)
+ .output(collectOutputItems())
+ .usage(effectiveUsage);
+ }
+ if (reason != null) {
+ synchronized (builderLock) {
+ responseBuilder = responseBuilder.incompleteDetails(
+ Response.IncompleteDetails.builder().reason(reason).build());
+ }
+ }
+ addEvent(new ResponseEvent("response.incomplete",
+ ResponseStreamEvent.ofIncomplete(new ResponseIncompleteEvent.Builder()
+ .sequenceNumber(nextSequenceNumber())
+ .response(snapshot())
+ .build())));
+ completeMulti();
+ return this;
+ }
+
+ /**
+ * Adds a message output item. The consumer configures the message builder
+ * (emit added/done, add text parts, etc.) within a lexically scoped block.
+ */
+ @Override
+ public ResponseEventStream addOutputMessage(Consumer config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateMessageItemId();
+ OutputMessageBuilder builder = new OutputMessageBuilder(this, idx, itemId);
+ config.accept(builder);
+ return this;
+ }
+
+ /**
+ * Adds a function call output item. The consumer configures the function call builder
+ * (emit added/done, emit argument deltas, etc.) within a lexically scoped block.
+ */
+ @Override
+ public ResponseEventStream addOutputFunctionCall(Consumer config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateFunctionCallItemId();
+ config.accept(new OutputFunctionCallBuilder(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Adds a reasoning output item.
+ */
+ @Override
+ public ResponseEventStream addOutputReasoningItem(Consumer> config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateReasoningItemId();
+ config.accept(new OutputItemBuilder<>(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Adds a file search call output item.
+ */
+ @Override
+ public ResponseEventStream addOutputFileSearchCall(Consumer> config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateFileSearchCallItemId();
+ config.accept(new OutputItemBuilder<>(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Adds a web search call output item.
+ */
+ @Override
+ public ResponseEventStream addOutputWebSearchCall(Consumer> config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateWebSearchCallItemId();
+ config.accept(new OutputItemBuilder<>(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Adds a code interpreter call output item.
+ */
+ @Override
+ public ResponseEventStream addOutputCodeInterpreterCall(
+ Consumer> config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateCodeInterpreterCallItemId();
+ config.accept(new OutputItemBuilder<>(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Adds an image generation call output item.
+ */
+ @Override
+ public ResponseEventStream addOutputImageGenCall(
+ Consumer> config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateImageGenCallItemId();
+ config.accept(new OutputItemBuilder<>(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Adds an MCP tool call output item.
+ */
+ @Override
+ public ResponseEventStream addOutputMcpCall(Consumer> config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateMcpCallItemId();
+ config.accept(new OutputItemBuilder<>(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Adds an MCP list-tools output item.
+ */
+ @Override
+ public ResponseEventStream addOutputMcpListTools(
+ Consumer> config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateMcpListToolsItemId();
+ config.accept(new OutputItemBuilder<>(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Adds a custom tool call output item.
+ */
+ @Override
+ public ResponseEventStream addOutputCustomToolCall(Consumer> config) {
+ long idx = outputIndex.getAndIncrement();
+ String itemId = idGenerator.generateCustomToolCallItemId();
+ config.accept(new OutputItemBuilder<>(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Adds a generic output item for types with no dedicated factory.
+ */
+ @Override
+ public ResponseEventStream addOutputItem(String itemId, Consumer> config) {
+ Objects.requireNonNull(itemId, "itemId must not be null");
+ long idx = outputIndex.getAndIncrement();
+ config.accept(new OutputItemBuilder<>(this, idx, itemId));
+ return this;
+ }
+
+ /**
+ * Appends a pre-built event directly. For advanced / interop scenarios.
+ */
+ @Override
+ public ResponseEventStream emit(ResponseEvent event) {
+ addEvent(event);
+ return this;
+ }
+
+ // ── Internal helpers ──────────────────────────────────────
+
+ /**
+ * Forwards an upstream OpenAI streaming response through this stream.
+ * All content events (output items, text deltas, function call arguments,
+ * reasoning, etc.) are forwarded directly. Upstream lifecycle events
+ * (created, in_progress, completed, queued) are skipped since this stream
+ * owns its own lifecycle.
+ *
+ * Throws {@link RuntimeException} if the upstream reports a failure event.
+ *
+ *
Example usage
+ * {@code
+ * stream.emitCreated().emitInProgress();
+ * try (var upstream = client.responses().createStreaming(params)) {
+ * stream.forwardUpstream(upstream.stream());
+ * }
+ * stream.emitCompleted();
+ * }
+ *
+ * @param upstreamEvents the stream of events from the upstream OpenAI client
+ * @return this stream for chaining
+ * @throws RuntimeException if the upstream reports a failure event
+ */
+ @Override
+ public ResponseEventStream forwardUpstream(java.util.stream.Stream upstreamEvents) {
+ upstreamEvents.forEach(event -> {
+ // Skip upstream lifecycle events — we own the response envelope.
+ if (event.isCreated() || event.isInProgress() || event.isCompleted() || event.isQueued()) {
+ return;
+ }
+
+ // Detect upstream failure.
+ if (event.isFailed()) {
+ throw new RuntimeException("Upstream request failed");
+ }
+
+ // Forward all content events (output_item.added, output_item.done,
+ // text deltas, function_call_arguments, content_part events, etc.)
+ // by determining the event name and re-emitting through our stream.
+ String eventName = resolveEventName(event);
+ if (eventName != null) {
+ // Track output items for the terminal completed event.
+ if (event.isOutputItemDone()) {
+ ResponseOutputItem item = event.asOutputItemDone().item();
+ long idx = outputIndex.getAndIncrement();
+ trackCompletedOutputItem(item, idx);
+ }
+ addEvent(new ResponseEvent(eventName, event));
+ }
+ });
+ return this;
+ }
+
+ public long nextSequenceNumber() {
+ return sequenceNumber.getAndIncrement();
+ }
+
+ public synchronized void addEvent(ResponseEvent event) {
+ events.add(event);
+ publisher.submit(event);
+ }
+
+ public void trackCompletedOutputItem(ResponseOutputItem item, long idx) {
+ outputItems.put(idx, item);
+ }
+
+ /**
+ * Returns a copy of the accumulated output items in index order.
+ */
+ private List collectOutputItems() {
+ return new ArrayList<>(outputItems.values());
+ }
+
+ private Response snapshot() {
+ synchronized (builderLock) {
+ return responseBuilder.build();
+ }
+ }
+
+ /**
+ * Atomically guards against calling a terminal method more than once.
+ *
+ * @throws IllegalStateException if a terminal event has already been emitted
+ */
+ private void guardTerminal() {
+ if (!terminated.compareAndSet(false, true)) {
+ throw new IllegalStateException(
+ "Stream has already been terminated. " +
+ "emitCompleted/emitFailed/emitIncomplete may only be called once.");
+ }
+ }
+
+ private void completeMulti() {
+ publisher.close();
+ }
+
+ // ══════════════════════════════════════════════════════════
+ // Scoped builders
+ // ══════════════════════════════════════════════════════════
+
+ /**
+ * Fluent builder for a generic output item. Returned by the various
+ * {@code addOutput*()} factory methods.
+ */
+ public static class OutputItemBuilder implements ResponseEventStream.OutputItemBuilder {
+ protected final ResponseEventStream stream;
+ protected final long outputIdx;
+ protected final String itemId;
+
+ OutputItemBuilder(ResponseEventStream stream, long outputIdx, String itemId) {
+ this.stream = stream;
+ this.outputIdx = outputIdx;
+ this.itemId = itemId;
+ }
+
+ static ResponseOutputItem toOutputItem(Object item) {
+ if (item instanceof ResponseOutputItem roi) return roi;
+ if (item instanceof ResponseOutputMessage msg) return ResponseOutputItem.ofMessage(msg);
+ if (item instanceof ResponseFunctionToolCall ftc) return ResponseOutputItem.ofFunctionCall(ftc);
+ if (item instanceof ResponseReasoningItem ri) return ResponseOutputItem.ofReasoning(ri);
+ if (item instanceof ResponseFileSearchToolCall fsc) return ResponseOutputItem.ofFileSearchCall(fsc);
+ if (item instanceof ResponseFunctionWebSearch ws) return ResponseOutputItem.ofWebSearchCall(ws);
+ if (item instanceof ResponseCodeInterpreterToolCall ci) return ResponseOutputItem.ofCodeInterpreterCall(ci);
+ if (item instanceof ResponseOutputItem.ImageGenerationCall ig)
+ return ResponseOutputItem.ofImageGenerationCall(ig);
+ if (item instanceof ResponseOutputItem.McpCall mc) return ResponseOutputItem.ofMcpCall(mc);
+ if (item instanceof ResponseOutputItem.McpListTools mlt) return ResponseOutputItem.ofMcpListTools(mlt);
+ if (item instanceof ResponseCustomToolCall ct) return ResponseOutputItem.ofCustomToolCall(ct);
+ throw new IllegalArgumentException("Unsupported output item type: " + item.getClass());
+ }
+
+ /**
+ * Emits a {@code response.output_item.added} event for this item.
+ */
+ public OutputItemBuilder emitAdded(T item) {
+ ResponseOutputItem outputItem = toOutputItem(item);
+ stream.addEvent(new ResponseEvent("response.output_item.added",
+ ResponseStreamEvent.ofOutputItemAdded(new ResponseOutputItemAddedEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .outputIndex(outputIdx)
+ .item(outputItem)
+ .build())));
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.output_item.done} event for this item
+ * and tracks it in the parent stream's output list.
+ */
+ public OutputItemBuilder emitDone(T item) {
+ ResponseOutputItem outputItem = toOutputItem(item);
+ stream.trackCompletedOutputItem(outputItem, outputIdx);
+ stream.addEvent(new ResponseEvent("response.output_item.done",
+ ResponseStreamEvent.ofOutputItemDone(new ResponseOutputItemDoneEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .outputIndex(outputIdx)
+ .item(outputItem)
+ .build())));
+ return this;
+ }
+
+ /**
+ * Returns the auto-generated item ID for this output item.
+ */
+ public String getItemId() {
+ return itemId;
+ }
+ }
+
+ /**
+ * Fluent builder for a function call output item. Provides convenience methods
+ * for the common function call lifecycle: added, argument deltas, argument done,
+ * and item done — without requiring the caller to manage sequence numbers or
+ * construct raw event objects.
+ *
+ * Example usage
+ * {@code
+ * stream.addOutputFunctionCall(func -> func
+ * .emitAdded("get_weather", "call_123")
+ * .emitArgumentsDelta("{\"location\":")
+ * .emitArgumentsDelta("\"Seattle\"}")
+ * .emitArgumentsDone("get_weather", "{\"location\":\"Seattle\"}")
+ * .emitDone());
+ * }
+ */
+ public static class OutputFunctionCallBuilder implements ResponseEventStream.OutputFunctionCallBuilder {
+ private final AgentServerResponseEventStream stream;
+ private final long outputIdx;
+ private final String itemId;
+ private final StringBuilder accumulatedArguments = new StringBuilder();
+ private String name;
+ private String callId;
+
+ OutputFunctionCallBuilder(AgentServerResponseEventStream stream, long outputIdx, String itemId) {
+ this.stream = stream;
+ this.outputIdx = outputIdx;
+ this.itemId = itemId;
+ }
+
+ /**
+ * Emits a {@code response.output_item.added} event with an in-progress function call.
+ *
+ * @param name the function name (e.g. "get_weather")
+ * @param callId the call ID for this function invocation
+ */
+ public OutputFunctionCallBuilder emitAdded(String name, String callId) {
+ this.name = name;
+ this.callId = callId;
+ ResponseFunctionToolCall call = ResponseFunctionToolCall.builder()
+ .id(itemId)
+ .callId(callId)
+ .name(name)
+ .arguments("")
+ .status(ResponseFunctionToolCall.Status.IN_PROGRESS)
+ .build();
+ stream.addEvent(new ResponseEvent("response.output_item.added",
+ ResponseStreamEvent.ofOutputItemAdded(new ResponseOutputItemAddedEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .outputIndex(outputIdx)
+ .item(ResponseOutputItem.ofFunctionCall(call))
+ .build())));
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.function_call_arguments.delta} event and accumulates
+ * the arguments text.
+ *
+ * @param delta the argument text chunk
+ */
+ public OutputFunctionCallBuilder emitArgumentsDelta(String delta) {
+ accumulatedArguments.append(delta);
+ stream.addEvent(new ResponseEvent("response.function_call_arguments.delta",
+ ResponseStreamEvent.ofFunctionCallArgumentsDelta(
+ ResponseFunctionCallArgumentsDeltaEvent.builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .itemId(itemId)
+ .outputIndex(outputIdx)
+ .delta(delta)
+ .build())));
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.function_call_arguments.done} event with the final
+ * function name and complete arguments string.
+ *
+ * @param name the function name
+ * @param arguments the complete arguments JSON string
+ */
+ public OutputFunctionCallBuilder emitArgumentsDone(String name, String arguments) {
+ accumulatedArguments.setLength(0);
+ accumulatedArguments.append(arguments);
+ stream.addEvent(new ResponseEvent("response.function_call_arguments.done",
+ ResponseStreamEvent.ofFunctionCallArgumentsDone(
+ ResponseFunctionCallArgumentsDoneEvent.builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .itemId(itemId)
+ .outputIndex(outputIdx)
+ .name(name)
+ .arguments(arguments)
+ .build())));
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.output_item.done} event with the completed function call
+ * and tracks it in the parent stream's output list. Uses the accumulated arguments
+ * and the name/callId from {@link #emitAdded(String, String)}.
+ */
+ public OutputFunctionCallBuilder emitDone() {
+ ResponseFunctionToolCall completedCall = ResponseFunctionToolCall.builder()
+ .id(itemId)
+ .callId(callId)
+ .name(name)
+ .arguments(accumulatedArguments.toString())
+ .status(ResponseFunctionToolCall.Status.COMPLETED)
+ .build();
+ ResponseOutputItem outputItem = ResponseOutputItem.ofFunctionCall(completedCall);
+ stream.trackCompletedOutputItem(outputItem, outputIdx);
+ stream.addEvent(new ResponseEvent("response.output_item.done",
+ ResponseStreamEvent.ofOutputItemDone(new ResponseOutputItemDoneEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .outputIndex(outputIdx)
+ .item(outputItem)
+ .build())));
+ return this;
+ }
+
+ /**
+ * Returns the auto-generated item ID for this function call.
+ */
+ public String getItemId() {
+ return itemId;
+ }
+ }
+
+ /**
+ * Fluent builder for a message output item. Provides convenience methods
+ * for the common message lifecycle and lexically scoped text-part builders.
+ */
+ public static class OutputMessageBuilder implements ResponseEventStream.OutputMessageBuilder {
+ private final AgentServerResponseEventStream stream;
+ private final long outputIdx;
+ private final String itemId;
+ private final String responseId;
+ private final List contentParts = new ArrayList<>();
+
+ OutputMessageBuilder(AgentServerResponseEventStream stream, long outputIdx, String itemId) {
+ this.stream = stream;
+ this.outputIdx = outputIdx;
+ this.itemId = itemId;
+ this.responseId = stream.responseId;
+ }
+
+ /**
+ * Emits a {@code response.output_item.added} event with an in-progress message.
+ */
+ public OutputMessageBuilder emitAdded() {
+ ResponseOutputItem outputItem = ResponseOutputItem.ofMessage(
+ buildMessage(ResponseOutputMessage.Status.IN_PROGRESS));
+ stream.addEvent(new ResponseEvent("response.output_item.added",
+ ResponseStreamEvent.ofOutputItemAdded(new ResponseOutputItemAddedEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .outputIndex(outputIdx)
+ .item(outputItem)
+ .build())));
+ return this;
+ }
+
+ /**
+ * Adds a text content part. The consumer configures delta emissions within
+ * a lexically scoped block.
+ *
+ * When the consumer returns, the following events are automatically emitted
+ * if the consumer did not already emit them:
+ *
+ * - {@code response.output_text.done} — if not already emitted by
+ * {@link TextPartBuilder#emitDone(String)}
+ * - {@code response.content_part.done} — always emitted
+ *
+ */
+ public OutputMessageBuilder addTextPart(Consumer config) {
+ long contentIndex = contentParts.size();
+ TextPartBuilder textBuilder = new TextPartBuilder(stream, itemId, outputIdx, contentIndex);
+ config.accept(textBuilder);
+
+ String finalTextValue = textBuilder.accumulatedText.toString();
+
+ // Auto-emit output_text.done if the consumer didn't call emitDone
+ if (!textBuilder.doneEmitted) {
+ textBuilder.emitDone(finalTextValue);
+ }
+
+ // Always emit content_part.done
+ ResponseOutputText finalText = ResponseOutputText.builder()
+ .text(finalTextValue)
+ .annotations(List.of())
+ .logprobs(List.of())
+ .build();
+ contentParts.add(ResponseOutputMessage.Content.ofOutputText(finalText));
+
+ stream.addEvent(new ResponseEvent("response.content_part.done",
+ ResponseStreamEvent.ofContentPartDone(new ResponseContentPartDoneEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .itemId(itemId)
+ .outputIndex(outputIdx)
+ .contentIndex(contentIndex)
+ .part(finalText)
+ .build())));
+ return this;
+ }
+
+ /**
+ * Convenience method that emits a complete text message in one call.
+ * Equivalent to:
+ * {@code
+ * msg.emitAdded()
+ * .addTextPart(text -> text.emitAdded().emitDelta(content).emitDone(content))
+ * .emitDone();
+ * }
+ *
+ * @param content The full text content to emit as a single delta and done event.
+ * @return this builder for chaining.
+ */
+ public OutputMessageBuilder outputItemMessage(String content) {
+ return emitAdded()
+ .addTextPart(text -> text
+ .emitAdded()
+ .emitDelta(content)
+ .emitDone(content))
+ .emitDone();
+ }
+
+ /**
+ * Emits a {@code response.output_item.done} event with the completed message
+ * and tracks it in the parent stream's output list.
+ */
+ public OutputMessageBuilder emitDone() {
+ ResponseOutputMessage msg = buildMessage(ResponseOutputMessage.Status.COMPLETED);
+ ResponseOutputItem outputItem = ResponseOutputItem.ofMessage(msg);
+ stream.trackCompletedOutputItem(outputItem, outputIdx);
+ stream.addEvent(new ResponseEvent("response.output_item.done",
+ ResponseStreamEvent.ofOutputItemDone(new ResponseOutputItemDoneEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .outputIndex(outputIdx)
+ .item(outputItem)
+ .build())));
+ return this;
+ }
+
+ /**
+ * Returns the auto-generated item ID for this message.
+ */
+ public String getItemId() {
+ return itemId;
+ }
+
+ @Override
+ public OutputMessageBuilder streamChatCompletion(StreamResponse streamResponse) {
+ return emitAdded()
+ .addTextPart(text -> text.emitAdded().streamDeltas(streamResponse))
+ .emitDone();
+ }
+
+ private ResponseOutputMessage buildMessage(ResponseOutputMessage.Status status) {
+ // Build the created_by object matching Foundry platform expectations:
+ // {"agent": {"type": "agent_id", "name": "", "version": ""}, "response_id": "..."}
+ Map agentObj = new LinkedHashMap<>();
+ agentObj.put("type", "agent_id");
+ agentObj.put("name", "");
+ agentObj.put("version", "");
+ Map createdBy = new LinkedHashMap<>();
+ createdBy.put("agent", agentObj);
+ createdBy.put("response_id", responseId);
+
+ return ResponseOutputMessage.builder()
+ .id(itemId)
+ .content(new ArrayList<>(contentParts))
+ .status(status)
+ .putAdditionalProperty("created_by", com.openai.core.JsonValue.from(createdBy))
+ .build();
+ }
+ }
+
+ /**
+ * Fluent builder for a text content part within a message.
+ * Tracks accumulated delta text for the final {@code content_part.done} event
+ * (emitted automatically by the parent {@link OutputMessageBuilder}).
+ */
+ public static class TextPartBuilder implements ResponseEventStream.TextPartBuilder {
+ final StringBuilder accumulatedText = new StringBuilder();
+ private final AgentServerResponseEventStream stream;
+ private final String itemId;
+ private final long outputIdx;
+ private final long contentIndex;
+ boolean doneEmitted;
+
+ TextPartBuilder(AgentServerResponseEventStream stream, String itemId, long outputIdx, long contentIndex) {
+ this.stream = stream;
+ this.itemId = itemId;
+ this.outputIdx = outputIdx;
+ this.contentIndex = contentIndex;
+ }
+
+ /**
+ * Emits a {@code response.content_part.added} event with an empty text part.
+ */
+ public TextPartBuilder emitAdded() {
+ ResponseOutputText empty = ResponseOutputText.builder()
+ .text("")
+ .annotations(List.of())
+ .build();
+ stream.addEvent(new ResponseEvent("response.content_part.added",
+ ResponseStreamEvent.ofContentPartAdded(new ResponseContentPartAddedEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .itemId(itemId)
+ .outputIndex(outputIdx)
+ .contentIndex(contentIndex)
+ .part(empty)
+ .build())));
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.output_text.delta} event and accumulates the text.
+ */
+ public TextPartBuilder emitDelta(String delta) {
+ accumulatedText.append(delta);
+ stream.addEvent(new ResponseEvent("response.output_text.delta",
+ ResponseStreamEvent.ofOutputTextDelta(new ResponseTextDeltaEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .itemId(itemId)
+ .outputIndex(outputIdx)
+ .contentIndex(contentIndex)
+ .delta(delta)
+ .logprobs(List.of())
+ .build())));
+ return this;
+ }
+
+ /**
+ * Emits a {@code response.output_text.done} event with the final text.
+ * Also updates the accumulated text to the provided value.
+ */
+ public TextPartBuilder emitDone(String text) {
+ accumulatedText.setLength(0);
+ accumulatedText.append(text);
+ doneEmitted = true;
+ stream.addEvent(new ResponseEvent("response.output_text.done",
+ ResponseStreamEvent.ofOutputTextDone(new ResponseTextDoneEvent.Builder()
+ .sequenceNumber(stream.nextSequenceNumber())
+ .itemId(itemId)
+ .outputIndex(outputIdx)
+ .contentIndex(contentIndex)
+ .text(text)
+ .logprobs(List.of())
+ .build())));
+ return this;
+ }
+
+ @Override
+ public TextPartBuilder streamDeltas(StreamResponse streamResponse) {
+ try (streamResponse) {
+ streamResponse.stream().forEach(chunk ->
+ chunk.choices().forEach(choice ->
+ choice.delta().content().ifPresent(delta -> {
+ if (!delta.isEmpty()) {
+ emitDelta(delta);
+ }
+ })
+ )
+ );
+ }
+ return this;
+ }
+ }
+}
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponseItemList.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponseItemList.java
new file mode 100644
index 000000000000..fea83b682fb9
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponseItemList.java
@@ -0,0 +1,17 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import com.fasterxml.jackson.annotation.JsonUnwrapped;
+import com.openai.models.responses.inputitems.ResponseItemList;
+
+/**
+ * Wraps a {@link ResponseItemList} together with an optional {@link AgentReference}
+ * for the agent server protocol's list-input-items response.
+ *
+ * @param agent the agent reference, or {@code null} if not applicable
+ * @param responseItemList the paginated list of response items
+ */
+public record AgentServerResponseItemList(AgentReference agent, @JsonUnwrapped ResponseItemList responseItemList) {
+}
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponsesApi.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponsesApi.java
new file mode 100644
index 000000000000..cdc9a995be74
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerResponsesApi.java
@@ -0,0 +1,981 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import com.microsoft.agentserver.api.implementation.FoundryStorageProvider;
+import com.microsoft.agentserver.api.implementation.IdGenerator;
+import com.microsoft.agentserver.api.implementation.InMemoryResponseProvider;
+import com.microsoft.agentserver.api.implementation.ItemConversion;
+import com.openai.models.responses.Response;
+import com.openai.models.responses.ResponseCreateParams;
+import com.openai.models.responses.ResponseItem;
+import com.openai.models.responses.ResponseOutputItem;
+import com.openai.models.responses.ResponseOutputMessage;
+import com.openai.models.responses.ResponseStatus;
+import com.openai.models.responses.ToolChoiceOptions;
+import com.openai.models.responses.inputitems.ResponseItemList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * Default implementation of {@link ResponsesApi} for the agent server protocol.
+ *
+ * Orchestrates request handling by delegating to a {@link ResponseHandler} for response
+ * generation, and a {@link ResponsesProvider} for state persistence. Supports both
+ * synchronous and streaming response creation.
+ *
+ * In hosted environments ({@link FoundryEnvironment#IS_HOSTED}), uses
+ * {@link FoundryStorageProvider} for persistence; otherwise falls back to
+ * {@link InMemoryResponseProvider}.
+ *
+ * Threading model: This class is a synchronous façade over the
+ * asynchronous {@link ResponsesProvider}. Methods such as {@link #getResponse},
+ * {@link #deleteResponse}, and {@link #listInputItems} block the calling thread
+ * via {@link java.util.concurrent.CompletableFuture#join()} until the provider
+ * operation completes. Callers should be aware that these methods will block and
+ * should not be invoked from async pipelines or reactive threads without offloading.
+ */
+class AgentServerResponsesApi implements ResponsesApi, AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(AgentServerResponsesApi.class);
+
+ private final ResponseHandler responseHandler;
+ private final ResponsesProvider provider;
+
+ /**
+ * Executor for background ({@code background=true}) response processing. Daemon
+ * threads so the JVM is not kept alive by in-flight background work.
+ */
+ private final ExecutorService backgroundExecutor = Executors.newCachedThreadPool(r -> {
+ Thread t = new Thread(r, "agentserver-background");
+ t.setDaemon(true);
+ return t;
+ });
+
+ /**
+ * Process-scoped registry of in-flight background executions.
+ * Used to coordinate cancellation with the
+ * background finalisation so cancellation always wins.
+ */
+ private final ExecutionTracker executionTracker = new ExecutionTracker();
+
+ /**
+ * In-memory buffer of SSE events for replay via {@code GET /responses/{id}?stream=true}.
+ *
+ */
+ private final EventReplayStore eventReplayStore = new EventReplayStore();
+
+ public AgentServerResponsesApi(ResponseHandler responseHandler, ResponsesProvider provider) {
+ this.responseHandler = responseHandler;
+ this.provider = provider;
+ }
+
+ /**
+ * Tracks active background executions by response ID so that cancel/finalise
+ * can coordinate.
+ */
+ static final class ExecutionTracker {
+ private final Map executions = new ConcurrentHashMap<>();
+
+ Execution register(String responseId, boolean background, boolean store) {
+ Execution execution = new Execution(background, store);
+ executions.put(responseId, execution);
+ return execution;
+ }
+
+ Execution get(String responseId) {
+ return executions.get(responseId);
+ }
+
+ void evict(String responseId) {
+ executions.remove(responseId);
+ }
+ }
+
+ /**
+ * Coordination state for a single in-flight execution. The mutable flags
+ * ({@code cancelRequested}, {@code finalized}) are guarded by the monitor of
+ * the {@code Execution} instance itself.
+ */
+ static final class Execution {
+ final boolean background;
+ final boolean store;
+ boolean cancelRequested;
+ boolean finalized;
+
+ Execution(boolean background, boolean store) {
+ this.background = background;
+ this.store = store;
+ }
+ }
+
+ private static String extractResponseIdFromMetadata(AgentServerCreateResponse createResponse) {
+ try {
+ return createResponse.responseCreateParams().metadata()
+ .map(meta -> {
+ com.openai.core.JsonValue val = meta._additionalProperties().get("response_id");
+ if (val == null) return null;
+ @SuppressWarnings("unchecked")
+ java.util.Optional str = val.asString();
+ return str.orElse(null);
+ })
+ .orElse(null);
+ } catch (Exception e) {
+ LOGGER.debug("Could not extract response_id from metadata", e);
+ return null;
+ }
+ }
+
+ private static String extractResponseItemId(ResponseItem item) {
+ return ItemConversion.extractItemId(item);
+ }
+
+ /**
+ * Resolves the response ID for a new create request, in priority order:
+ *
+ * - {@code x-agent-response-id} HTTP header.
+ * - {@code metadata.response_id} field on the request body (legacy override).
+ * - Freshly generated via {@link IdGenerator#generateResponseId()}.
+ *
+ */
+ private static String resolveResponseId(AgentServerCreateResponse createResponse, RequestMetadata metadata) {
+ if (metadata != null) {
+ String override = metadata.getResponseIdOverride();
+ if (override != null && !override.isEmpty()) {
+ return override;
+ }
+ }
+ String fromBody = extractResponseIdFromMetadata(createResponse);
+ if (fromBody != null) {
+ return fromBody;
+ }
+ return new IdGenerator(null).generateResponseId();
+ }
+
+ /**
+ * Auto-stamps the resolved {@code agent_session_id} onto a
+ * {@link Response} by adding it to the additional-properties map. Returns the
+ * original instance unchanged when {@code sessionId} is {@code null}/empty.
+ */
+ private static Response stampSessionId(Response resp, String sessionId) {
+ if (resp == null || sessionId == null || sessionId.isEmpty()) {
+ return resp;
+ }
+ return resp.toBuilder()
+ .putAdditionalProperty("agent_session_id", com.openai.core.JsonValue.from(sessionId))
+ .build();
+ }
+
+ /**
+ * Combines (response-ID resolution) and (session-ID stamping) and
+ * normalizes all child IDs (output message ids, conversation id) to share
+ * the resolved response's partition key. The Foundry storage backend routes
+ * an envelope to a single shard by partition key, so a mix of partitions in
+ * a single envelope causes the storage create to fail with HTTP 500.
+ */
+ private static Response normalizeIdsAndStamp(Response handlerResp, String responseId, String sessionId, AgentReference agentRef) {
+ Response.Builder rebuilt = handlerResp.toBuilder();
+ if (!responseId.equals(handlerResp.id())) {
+ rebuilt.id(responseId);
+ }
+
+ // Re-partition any output message item IDs whose partition doesn't already
+ // match the resolved response. The Foundry storage backend routes envelopes
+ // by partition key; mixed partitions in a single envelope are rejected.
+ // (Streaming already produces aligned IDs; this primarily fixes the sync
+ // path where ResponseBuilder generates a fresh partition.)
+ String partitionKey;
+ try {
+ partitionKey = IdGenerator.extractPartitionKey(responseId);
+ } catch (RuntimeException e) {
+ partitionKey = null;
+ }
+
+ if (partitionKey != null) {
+ IdGenerator idGen = new IdGenerator(partitionKey);
+ List normalizedOutput = new ArrayList<>();
+ for (ResponseOutputItem item : handlerResp.output()) {
+ if (item.isMessage()) {
+ ResponseOutputMessage msg = item.asMessage();
+ if (!sharesPartition(msg.id(), partitionKey)) {
+ normalizedOutput.add(ResponseOutputItem.ofMessage(
+ msg.toBuilder().id(idGen.generateMessageItemId()).build()));
+ continue;
+ }
+ }
+ normalizedOutput.add(item);
+ }
+ rebuilt.output(normalizedOutput);
+ }
+
+ if (sessionId != null && !sessionId.isEmpty()) {
+ rebuilt.putAdditionalProperty(
+ "agent_session_id", com.openai.core.JsonValue.from(sessionId));
+ }
+
+ // echo the request's agent_reference onto the response so the
+ // platform storage backend can correlate the persisted response with the
+ // originating agent. Stored as an additional property because the openai
+ // Response model has no first-class agent_reference field.
+ if (agentRef != null) {
+ com.fasterxml.jackson.databind.node.ObjectNode refNode = com.microsoft.agentserver.api.serialization.ObjectMapperFactory
+ .getObjectMapper().createObjectNode();
+ if (agentRef.type() != null) {
+ refNode.put("type", agentRef.type().toString().toLowerCase(java.util.Locale.ROOT));
+ }
+ if (agentRef.name() != null) {
+ refNode.put("name", agentRef.name());
+ }
+ if (agentRef.version() != null) {
+ refNode.put("version", agentRef.version());
+ }
+ if (agentRef.label() != null) {
+ refNode.put("label", agentRef.label());
+ }
+ rebuilt.putAdditionalProperty("agent_reference", com.openai.core.JsonValue.from(refNode));
+ }
+
+ return rebuilt.build();
+ }
+
+ private static boolean sharesPartition(String id, String partitionKey) {
+ if (id == null) {
+ return false;
+ }
+ try {
+ return partitionKey.equals(IdGenerator.extractPartitionKey(id));
+ } catch (RuntimeException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Validates a response-ID path parameter before any lookup. Malformed
+ * IDs produce HTTP 400 ({@code "Malformed identifier."}) rather than 404.
+ *
+ * @param responseId the path parameter to validate
+ * @throws ApiException 400 if the ID is not a well-formed response ID
+ */
+ private static void validateResponseId(String responseId) throws ApiException {
+ if (!IdGenerator.isValidResponseId(responseId)) {
+ // param="responseId{}" so clients can echo the offending value.
+ String param = "responseId{" + (responseId == null ? "" : responseId) + "}";
+ throw new ApiException(400,
+ ApiError.invalidRequest("Malformed identifier.", ApiError.CODE_INVALID_PARAMETERS, param));
+ }
+ }
+
+ @Override
+ public CreateResponse createResponse(AgentServerCreateResponse createResponse) throws ApiException {
+ return createResponse(createResponse, RequestMetadata.EMPTY);
+ }
+
+ @Override
+ public CreateResponse createResponse(AgentServerCreateResponse createResponse, RequestMetadata metadata) throws ApiException {
+ LOGGER.debug("createResponse called");
+
+ ResponseCreateParams.Body params = createResponse.responseCreateParams();
+ boolean background = params.background().orElse(false);
+ boolean store = params.store().orElse(true);
+
+ // background=true requires store=true.
+ if (background && !store) {
+ throw new ApiException(400, ApiError.invalidRequest("The 'background' parameter requires 'store' to be true.", ApiError.CODE_UNSUPPORTED_PARAMETER, "background"));
+ }
+
+ String responseId = resolveResponseId(createResponse, metadata);
+ // resolve the session ID once for this request; stamped on every
+ // Response object that leaves the API.
+ String sessionId = SessionIdResolver.resolve(createResponse, FoundryEnvironment.SESSION_ID);
+
+ ResponseContext context = buildContext(responseId, sessionId, createResponse, metadata);
+
+ // : emit the spec-required invoke_agent {model} span around handler
+ // execution. The span scopes everything the handler does so per-tool /
+ // per-LLM child spans (when emitted) attach to it.
+ io.opentelemetry.api.trace.Span span = Observability.startInvokeAgentSpan(
+ extractModelName(params),
+ responseId,
+ extractConversationId(params),
+ createResponse.agent() != null ? createResponse.agent().name() : null,
+ createResponse.agent() != null ? createResponse.agent().version() : null,
+ false);
+ try (io.opentelemetry.context.Scope ignored = span.makeCurrent()) {
+ if (background) {
+ CreateResponse bg = createBackgroundResponse(responseId, sessionId, context, createResponse);
+ return bg;
+ }
+
+ CreateResponse response;
+ try {
+ response = responseHandler.createResponse(context, createResponse);
+ } catch (RuntimeException e) {
+ String code = e.getClass().getSimpleName();
+ String msg = e.getMessage();
+ if (e.getCause() instanceof ApiException ae) {
+ code = ae.getError() != null ? ae.getError().code() : code;
+ msg = ae.getMessage();
+ }
+ Observability.recordInvokeAgentError(span, code, msg, e);
+ throw e;
+ }
+
+ // ensure the resolved response ID (which may have come from the
+ // x-agent-response-id header or `metadata.response_id`) is the one that
+ // both the client sees and storage uses — even when handlers build their
+ // own ID internally via ResponseBuilder. Also stamp the session ID,
+ // and re-partition child IDs (msg_, conv_) to match the resolved response's
+ // partition key (required by Foundry storage so the envelope routes to a
+ // single shard).
+ Response handlerResp = response.response();
+ if (handlerResp != null) {
+ handlerResp = normalizeIdsAndStamp(handlerResp, responseId, sessionId, createResponse.agent());
+ response = new CreateResponse(response.agent(), handlerResp);
+ }
+
+ persistResponse(context, response);
+
+ return response;
+ } finally {
+ span.end();
+ }
+ }
+
+ /**
+ * Handles {@code background=true}, non-streaming creation (matrix C3 / Rules
+ * ). Persists an initial {@code in_progress} snapshot so the
+ * response is immediately retrievable via GET, returns that snapshot to the
+ * caller right away, and processes the handler asynchronously, persisting the
+ * terminal result when done. Cancellation always wins.
+ */
+ private CreateResponse createBackgroundResponse(
+ String responseId, String sessionId, ResponseContext context, AgentServerCreateResponse createResponse) {
+
+ // Phase 1: persist the in_progress snapshot (with input items) so GET works
+ // while processing runs in the background. stamp the resolved session ID.
+ Response initial = stampSessionId(
+ buildInitialResponse(responseId, createResponse, ResponseStatus.IN_PROGRESS),
+ sessionId);
+ persistStreamingResponse(context, initial);
+
+ Execution execution = executionTracker.register(responseId, true, true);
+ final String finalResponseId = responseId;
+ final String finalSessionId = sessionId;
+
+ backgroundExecutor.submit(() -> {
+ try {
+ CreateResponse result = responseHandler.createResponse(context, createResponse);
+ // Re-stamp the terminal response with the response ID we already handed
+ // to the client (Phase 1), the background flag, and the session ID, so
+ // GET on that ID observes a coherent completed background response.
+ Response terminal = result.response();
+ if (terminal != null) {
+ Response.Builder rebuilt = terminal.toBuilder().background(true);
+ if (!finalResponseId.equals(terminal.id())) {
+ rebuilt.id(finalResponseId);
+ }
+ rebuilt.putAdditionalProperty(
+ "agent_session_id", com.openai.core.JsonValue.from(finalSessionId));
+ terminal = rebuilt.build();
+ }
+ final Response finalTerminal = terminal;
+ synchronized (execution) {
+ if (!execution.cancelRequested && !execution.finalized && finalTerminal != null) {
+ persistStreamingResponse(context, finalTerminal);
+ execution.finalized = true;
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.error("Background response {} processing failed", finalResponseId, e);
+ synchronized (execution) {
+ if (!execution.cancelRequested && !execution.finalized) {
+ Response failed = stampSessionId(
+ buildInitialResponse(finalResponseId, createResponse, ResponseStatus.FAILED),
+ finalSessionId);
+ persistStreamingResponse(context, failed);
+ execution.finalized = true;
+ }
+ }
+ } finally {
+ executionTracker.evict(finalResponseId);
+ }
+ });
+
+ // return immediately with the in_progress snapshot.
+ return new CreateResponse(null, initial);
+ }
+
+ /**
+ * Builds a minimal {@link Response} snapshot for a background response in the
+ * given lifecycle {@code status} (used for the initial {@code in_progress}
+ * persistence and for failure fallback).
+ */
+ private Response buildInitialResponse(
+ String responseId, AgentServerCreateResponse createResponse, ResponseStatus status) {
+
+ ResponseCreateParams.Body params = createResponse.responseCreateParams();
+
+ String conversationId = params.conversation()
+ .flatMap(conv -> {
+ if (conv.isId()) {
+ return Optional.of(conv.asId());
+ }
+ if (conv.isResponseConversationParam()) {
+ return Optional.of(conv.asResponseConversationParam().id());
+ }
+ return Optional.empty();
+ })
+ .orElse(null);
+
+ Response.Builder builder = Response.builder()
+ .id(responseId)
+ .createdAt(System.currentTimeMillis() / 1000.0)
+ .status(status)
+ .background(true)
+ .output(List.of())
+ .error(Optional.empty())
+ .incompleteDetails(Optional.empty())
+ .instructions(Optional.empty())
+ .metadata(Optional.empty())
+ .parallelToolCalls(false)
+ .temperature(Optional.empty())
+ .toolChoice(ToolChoiceOptions.AUTO)
+ .tools(List.of())
+ .topP(Optional.empty())
+ .previousResponseId(params.previousResponseId().orElse(null));
+
+ if (params.model().isPresent()) {
+ params.model().ifPresent(builder::model);
+ } else {
+ builder.model(FoundryEnvironment.AGENT_NAME != null ? FoundryEnvironment.AGENT_NAME : "unknown-agent");
+ }
+
+ if (conversationId != null) {
+ builder.conversation(Response.Conversation.builder().id(conversationId).build());
+ }
+
+ return builder.build();
+ }
+
+ @Override
+ public ResponseEventStream createStreamingResponse(AgentServerCreateResponse createResponse) throws ApiException {
+ return createStreamingResponse(createResponse, RequestMetadata.EMPTY);
+ }
+
+ @Override
+ public ResponseEventStream createStreamingResponse(AgentServerCreateResponse createResponse, RequestMetadata metadata) throws ApiException {
+ LOGGER.debug("createStreamingResponse called");
+
+ ResponseCreateParams.Body params = createResponse.responseCreateParams();
+ boolean background = params.background().orElse(false);
+ boolean store = params.store().orElse(true);
+
+ // background=true requires store=true (matrix C8).
+ if (background && !store) {
+ throw new ApiException(400, ApiError.invalidRequest("The 'background' parameter requires 'store' to be true.", ApiError.CODE_UNSUPPORTED_PARAMETER, "background"));
+ }
+
+ String responseId = resolveResponseId(createResponse, metadata);
+ // resolve the session ID once for this request.
+ String sessionId = SessionIdResolver.resolve(createResponse, FoundryEnvironment.SESSION_ID);
+
+ ResponseContext context = buildContext(responseId, sessionId, createResponse, metadata);
+
+ // : emit the spec-required invoke_agent {model} span. For streaming the
+ // span lives until the stream terminates (completion or failure), so the
+ // handler's tool/LLM child spans (when emitted) attach to it.
+ io.opentelemetry.api.trace.Span invokeSpan = Observability.startInvokeAgentSpan(
+ extractModelName(params),
+ responseId,
+ extractConversationId(params),
+ createResponse.agent() != null ? createResponse.agent().name() : null,
+ createResponse.agent() != null ? createResponse.agent().version() : null,
+ true);
+ ResponseEventStream stream;
+ try (io.opentelemetry.context.Scope ignored = invokeSpan.makeCurrent()) {
+ stream = responseHandler.createAsync(context, createResponse);
+ } catch (RuntimeException e) {
+ Observability.recordInvokeAgentError(invokeSpan,
+ e.getClass().getSimpleName(), e.getMessage(), e);
+ invokeSpan.end();
+ throw e;
+ }
+
+ // stamp the resolved session ID on every snapshot the stream produces.
+ if (stream instanceof AgentServerResponseEventStream impl) {
+ impl.setAgentSessionId(sessionId);
+ }
+
+ // C4 (background+stream): persist an initial in_progress snapshot so GET
+ // works immediately while the response streams in the background.
+ if (background) {
+ Response initial = stampSessionId(
+ buildInitialResponse(responseId, createResponse, ResponseStatus.IN_PROGRESS),
+ sessionId);
+ persistStreamingResponse(context, initial);
+ }
+
+ // For stored responses, init the replay buffer so a mid-stream replay
+ // request sees `hasBuffer()==true` (avoiding a false "not stream=true"
+ // rejection) and gets whatever has been emitted so far.
+ if (store) {
+ eventReplayStore.initBuffer(responseId);
+ // Capture any events the handler emitted synchronously before we wired
+ // up the subscriber below.
+ for (ResponseEvent existing : stream.getEvents()) {
+ eventReplayStore.append(responseId, existing);
+ }
+ }
+
+ // Register the in-flight execution so cancel/disconnect can coordinate with
+ // the terminal persistence. Background streams are tagged as background so
+ // signalClientDisconnected is a no-op for them; explicit cancel
+ // remains supported. Non-background streams enable.
+ final Execution execution = executionTracker.register(responseId, background, store);
+
+ final String finalResponseId = responseId;
+ final boolean storeEnabled = store;
+ final boolean isBackground = background;
+ // Track the count of events we already appended above so we don't
+ // double-buffer those when the subscribe callback fires.
+ final int alreadyBuffered = store ? stream.getEvents().size() : 0;
+ final java.util.concurrent.atomic.AtomicInteger seenCount =
+ new java.util.concurrent.atomic.AtomicInteger(0);
+ stream.subscribe(
+ event -> {
+ // Live mid-stream buffering: append each new event
+ // so a concurrent replay request observes it. Skip the prefix the
+ // synchronous handler already emitted before subscribe attached.
+ if (storeEnabled && seenCount.getAndIncrement() >= alreadyBuffered) {
+ eventReplayStore.append(finalResponseId, event);
+ }
+ },
+ failure -> {
+ LOGGER.error("Stream failed for response {}", finalResponseId, failure);
+ Observability.recordInvokeAgentError(invokeSpan,
+ failure.getClass().getSimpleName(), failure.getMessage(), failure);
+ invokeSpan.end();
+ },
+ () -> {
+ LOGGER.debug("Stream completed for response {}", finalResponseId);
+ try {
+ synchronized (execution) {
+ if (execution.cancelRequested || execution.finalized) {
+ // Cancellation won the race; do not overwrite
+ // the cancelled snapshot with the natural terminal state.
+ return;
+ }
+ Response resp = stream.getResponse();
+ if (resp != null) {
+ if (isBackground) {
+ // C4: re-stamp background=true so GET on the terminal
+ // observes a coherent background response (mirroring C3).
+ resp = resp.toBuilder().background(true).build();
+ }
+ // Match the sync path: normalize child IDs to the resolved
+ // response's partition, stamp agent_reference and
+ // agent_session_id. Without these the storage POST
+ // fails (envelope partition mismatch, missing agent_reference).
+ resp = normalizeIdsAndStamp(resp, finalResponseId, sessionId, createResponse.agent());
+ persistStreamingResponse(context, resp);
+ }
+ execution.finalized = true;
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed to persist streaming response", e);
+ } finally {
+ executionTracker.evict(finalResponseId);
+ invokeSpan.end();
+ }
+ }
+ );
+
+ return stream;
+ }
+
+ private static String extractModelName(ResponseCreateParams.Body params) {
+ try {
+ if (params.model().isPresent()) {
+ var model = params.model().get();
+ if (model._json().isPresent() && model._json().get().asString().isPresent()) {
+ return model._json().get().asString().get().toString();
+ }
+ }
+ } catch (Exception ignored) {
+ // fall through
+ }
+ return null;
+ }
+
+ private static String extractConversationId(ResponseCreateParams.Body params) {
+ try {
+ return params.conversation()
+ .flatMap(c -> c.isId() ? java.util.Optional.of(c.asId()) : java.util.Optional.empty())
+ .orElse(null);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ @Override
+ public com.openai.models.responses.Response getResponse(String responseId, List include) throws ApiException {
+ return getResponse(responseId, include, RequestMetadata.EMPTY);
+ }
+
+ @Override
+ public com.openai.models.responses.Response getResponse(String responseId, List include, RequestMetadata metadata) throws ApiException {
+ LOGGER.debug("getResponse called for responseId={}", responseId);
+ validateResponseId(responseId);
+ com.openai.models.responses.Response resp = provider.getResponseAsync(responseId, metadata.getIsolation())
+ .join()
+ .orElse(null);
+
+ if (resp == null) {
+ throw new ApiException(404, ApiError.invalidRequest(responseId + " not found"));
+ }
+
+ return resp;
+ }
+
+ @Override
+ public ResponseStreamReplay replayResponseStream(String responseId, Integer startingAfter) throws ApiException {
+ LOGGER.debug("replayResponseStream called for responseId={}, startingAfter={}", responseId, startingAfter);
+ validateResponseId(responseId);
+
+ // Not found / store=false → 404.
+ com.openai.models.responses.Response resp = provider.getResponseAsync(responseId)
+ .join()
+ .orElse(null);
+ if (resp == null) {
+ throw new ApiException(404, ApiError.invalidRequest(responseId + " not found"));
+ }
+
+ // SSE replay requires background=true (checked before stream).
+ if (!resp.background().orElse(false)) {
+ throw new ApiException(400, ApiError.invalidRequest(
+ "This response cannot be streamed because it was not created with background=true.",
+ ApiError.CODE_INVALID_REQUEST, "stream"));
+ }
+
+ // SSE replay requires stream=true at creation. The presence of a replay
+ // buffer indicates the response was created streaming.
+ if (!eventReplayStore.hasBuffer(responseId)) {
+ throw new ApiException(400, ApiError.invalidRequest(
+ "This response cannot be streamed because it was not created with stream=true.",
+ ApiError.CODE_INVALID_REQUEST, "stream"));
+ }
+
+ List events = eventReplayStore.replay(responseId, startingAfter)
+ .orElseGet(List::of);
+ return new ResponseStreamReplay(events);
+ }
+
+ @Override
+ public com.openai.models.responses.Response cancelResponse(String responseId) throws ApiException {
+ LOGGER.debug("cancelResponse called for responseId={}", responseId);
+ validateResponseId(responseId);
+
+ com.openai.models.responses.Response resp = provider.getResponseAsync(responseId)
+ .join()
+ .orElse(null);
+
+ // (not found) → 404. Non-background in-flight responses are also not findable.
+ if (resp == null) {
+ throw new ApiException(404, ApiError.invalidRequest(responseId + " not found"));
+ }
+
+ // the background check happens first, regardless of status.
+ boolean background = resp.background().orElse(false);
+ if (!background) {
+ throw new ApiException(400, ApiError.invalidRequest("Cannot cancel a synchronous response."));
+ }
+
+ com.openai.models.responses.ResponseStatus status = resp.status().orElse(null);
+
+ // cancelling an already-cancelled response is idempotent.
+ if (com.openai.models.responses.ResponseStatus.CANCELLED.equals(status)) {
+ return resp;
+ }
+
+ // terminal states cannot be cancelled.
+ if (com.openai.models.responses.ResponseStatus.COMPLETED.equals(status)) {
+ throw new ApiException(400, ApiError.invalidRequest("Cannot cancel a completed response."));
+ }
+ if (com.openai.models.responses.ResponseStatus.FAILED.equals(status)) {
+ throw new ApiException(400, ApiError.invalidRequest("Cannot cancel a failed response."));
+ }
+ if (com.openai.models.responses.ResponseStatus.INCOMPLETE.equals(status)) {
+ throw new ApiException(400, ApiError.invalidRequest("Cannot cancel a response in terminal state."));
+ }
+
+ // queued or in_progress → wind down to cancelled with output cleared.
+ com.openai.models.responses.Response cancelled = resp.toBuilder()
+ .status(com.openai.models.responses.ResponseStatus.CANCELLED)
+ .output(Collections.emptyList())
+ .build();
+
+ // Coordinate with any in-flight background execution so the background
+ // finaliser does not overwrite the cancelled state.
+ Execution execution = executionTracker.get(responseId);
+ if (execution != null) {
+ synchronized (execution) {
+ execution.cancelRequested = true;
+ provider.saveResponseAsync(responseId, cancelled, null, null).join();
+ }
+ } else {
+ provider.saveResponseAsync(responseId, cancelled, null, null).join();
+ }
+ LOGGER.debug("Response {} cancelled (output cleared)", responseId);
+ return cancelled;
+ }
+
+ @Override
+ public void signalClientDisconnected(String responseId) {
+ // only non-background responses are affected by client disconnect.
+ // Background responses outlive the connection — no-op there.
+ Execution execution = executionTracker.get(responseId);
+ if (execution == null || execution.background) {
+ return;
+ }
+ synchronized (execution) {
+ if (execution.cancelRequested || execution.finalized) {
+ return;
+ }
+ execution.cancelRequested = true;
+ // Persist the cancelled snapshot only when store=true; per, store=false
+ // disconnects produce no retrievable response (GET → 404).
+ if (execution.store) {
+ try {
+ com.openai.models.responses.Response existing =
+ provider.getResponseAsync(responseId).join().orElse(null);
+ com.openai.models.responses.Response.Builder builder = existing != null
+ ? existing.toBuilder()
+ : com.openai.models.responses.Response.builder()
+ .id(responseId)
+ .createdAt(System.currentTimeMillis() / 1000.0)
+ .model("")
+ .parallelToolCalls(false)
+ .tools(java.util.List.of())
+ .error(java.util.Optional.empty())
+ .incompleteDetails(java.util.Optional.empty())
+ .instructions(java.util.Optional.empty())
+ .metadata(java.util.Optional.empty())
+ .temperature(java.util.Optional.empty())
+ .topP(java.util.Optional.empty())
+ .toolChoice(com.openai.models.responses.ToolChoiceOptions.AUTO);
+ com.openai.models.responses.Response cancelled = builder
+ .status(com.openai.models.responses.ResponseStatus.CANCELLED)
+ .output(java.util.Collections.emptyList())
+ .build();
+ provider.saveResponseAsync(responseId, cancelled, null, null).join();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to persist cancelled-on-disconnect snapshot for {}", responseId, e);
+ }
+ }
+ LOGGER.debug("Response {} cancelled on client disconnect", responseId);
+ }
+ }
+
+ // ── Private helpers ─────────────────────────────────────────
+
+ private ResponseContext buildContext(String responseId, String sessionId, AgentServerCreateResponse createResponse, RequestMetadata metadata) {
+ return ResponseContext.builder()
+ .responseId(responseId)
+ .provider(provider)
+ .request(createResponse.responseCreateParams())
+ .isolation(metadata.getIsolation())
+ .clientHeaders(metadata.getClientHeaders())
+ .queryParameters(metadata.getQueryParameters())
+ .requestId(metadata.getRequestId())
+ .sessionId(sessionId)
+ .build();
+ }
+
+ @Override
+ public void deleteResponse(String responseId) throws ApiException {
+ LOGGER.debug("deleteResponse called for responseId={}", responseId);
+ validateResponseId(responseId);
+ provider.deleteResponseAsync(responseId).join();
+ // Best-effort removal of any buffered SSE events so replay is no longer possible.
+ eventReplayStore.delete(responseId);
+ }
+
+ @Override
+ public AgentServerResponseItemList listInputItems(String responseId, Integer limit, String order, String after, String before, List include) throws ApiException {
+ LOGGER.debug("listInputItems called for responseId={}", responseId);
+ validateResponseId(responseId);
+
+ // Verify the response exists
+ com.openai.models.responses.Response resp = provider.getResponseAsync(responseId)
+ .join()
+ .orElse(null);
+
+ if (resp == null) {
+ throw new ApiException(404, ApiError.invalidRequest(responseId + " not found"));
+ }
+
+ // Retrieve the stored input items (not output items)
+ List allItems = provider.getInputItemsForResponseAsync(responseId)
+ .join();
+
+ int effectiveLimit = (limit != null) ? limit : 20;
+ boolean descending = "desc".equalsIgnoreCase(order);
+
+
+ // Apply cursor-based pagination. `after` slices off everything up to and including
+ // that item; `before` slices off everything from that item onward; both combine
+ // as an exclusive range. Items always operate on the underlying stored order
+ // (i.e. cursor semantics are positional and order-independent, per ).
+ int fromIdx = 0;
+ int toIdx = allItems.size();
+ if (after != null && !after.isEmpty()) {
+ for (int i = 0; i < allItems.size(); i++) {
+ if (after.equals(extractResponseItemId(allItems.get(i)))) {
+ fromIdx = Math.max(fromIdx, i + 1);
+ break;
+ }
+ }
+ }
+ if (before != null && !before.isEmpty()) {
+ for (int i = 0; i < allItems.size(); i++) {
+ if (before.equals(extractResponseItemId(allItems.get(i)))) {
+ toIdx = Math.min(toIdx, i);
+ break;
+ }
+ }
+ }
+ List windowed = fromIdx < toIdx
+ ? allItems.subList(fromIdx, toIdx)
+ : List.of();
+
+ // Apply ordering
+ List ordered;
+ if (descending) {
+ ordered = new ArrayList<>(windowed);
+ Collections.reverse(ordered);
+ } else {
+ ordered = windowed;
+ }
+
+ // Apply limit
+ List page = ordered.size() > effectiveLimit
+ ? ordered.subList(0, effectiveLimit)
+ : ordered;
+
+ boolean hasMore = ordered.size() > effectiveLimit;
+ String firstId = page.isEmpty() ? null : extractResponseItemId(page.get(0));
+ String lastId = page.isEmpty() ? null : extractResponseItemId(page.get(page.size() - 1));
+
+ ResponseItemList itemList = ResponseItemList.builder()
+ .data(page)
+ .firstId(firstId != null ? firstId : "")
+ .lastId(lastId != null ? lastId : "")
+ .hasMore(hasMore)
+ .build();
+
+ return new AgentServerResponseItemList(null, itemList);
+ }
+
+ private void persistResponse(ResponseContext context, CreateResponse response) {
+ if (response.response() != null) {
+ com.openai.models.responses.Response resp = response.response();
+ persistStreamingResponse(context, resp);
+ }
+ }
+
+ private void persistStreamingResponse(ResponseContext context, com.openai.models.responses.Response resp) {
+ String previousResponseId = resp.previousResponseId().orElse(null);
+ String conversationId = resp.conversation()
+ .map(com.openai.models.responses.Response.Conversation::id)
+ .orElse(null);
+
+ // Build role-preserving input_items directly from the raw request body.
+ // The context.getInputItemsAsync() path returns ResponseOutputItems which
+ // are hardcoded to role=assistant — wrong for stored conversation history.
+ List inputItems = Collections.emptyList();
+ try {
+ ResponseCreateParams.Body body = context instanceof AgentServerResponseContext c
+ ? c.getRequestBody() : null;
+ if (body != null && body.input().isPresent()) {
+ ResponseCreateParams.Input in = body.input().get();
+ if (in.isResponse()) {
+ IdGenerator idGen = new IdGenerator(IdGenerator.extractPartitionKey(resp.id()));
+ List items = new ArrayList<>();
+ for (com.openai.models.responses.ResponseInputItem raw : in.asResponse()) {
+ ResponseItem converted = ItemConversion.toResponseItem(raw, idGen);
+ if (converted != null) {
+ items.add(converted);
+ }
+ }
+ inputItems = items;
+ } else if (in.isText() && !in.asText().isEmpty()) {
+ // Plain-text input — synthesize a single user message.
+ IdGenerator idGen = new IdGenerator(IdGenerator.extractPartitionKey(resp.id()));
+ inputItems = List.of(ItemConversion.toResponseItem(
+ com.openai.models.responses.ResponseInputItem.ofEasyInputMessage(
+ com.openai.models.responses.EasyInputMessage.builder()
+ .role(com.openai.models.responses.EasyInputMessage.Role.USER)
+ .content(com.openai.models.responses.EasyInputMessage.Content.ofTextInput(in.asText()))
+ .build()),
+ idGen));
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.warn("Failed to gather input items for response {}", resp.id(), e);
+ }
+
+ // Get history item IDs if there's a previous response
+ List historyItemIds = Collections.emptyList();
+ if (previousResponseId != null) {
+ try {
+ historyItemIds = provider.getHistoryItemIdsAsync(previousResponseId, conversationId, 100).join();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to get history item IDs for response {}", resp.id(), e);
+ }
+ }
+
+ // Single envelope call: POST /responses with {response, input_items, history_item_ids}
+ try {
+ // Pass isolation context so storage API receives platform isolation headers
+ IsolationContext isolation = context.getIsolation();
+ if (provider instanceof FoundryStorageProvider foundryProvider) {
+ foundryProvider.createResponseAsync(resp.id(), resp, inputItems, historyItemIds, isolation).join();
+ } else {
+ provider.createResponseAsync(resp.id(), resp, inputItems, historyItemIds).join();
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed to persist response {} to Foundry storage", resp.id(), e);
+ }
+ }
+
+ /**
+ * Shuts down the background executor, waiting up to 5 seconds for in-flight
+ * tasks to complete. Should be called when the hosting container is stopping
+ * (e.g. via a JVM shutdown hook or framework lifecycle callback).
+ */
+ @Override
+ public void close() {
+ backgroundExecutor.shutdown();
+ try {
+ if (!backgroundExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
+ LOGGER.warn("Background executor did not terminate within 5s; forcing shutdown");
+ backgroundExecutor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ LOGGER.warn("Interrupted while awaiting background executor shutdown");
+ backgroundExecutor.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+}
+
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerVersion.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerVersion.java
new file mode 100644
index 000000000000..180f954c9752
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/AgentServerVersion.java
@@ -0,0 +1,45 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+/**
+ * Provides the {@code x-platform-server} header value identifying this library.
+ *
+ * The header value is computed once at class initialization from the library version
+ * and the current JVM version, e.g.:
+ * {@code azure-ai-agentserver-java/1.0.0 (java/21)}
+ */
+public final class AgentServerVersion {
+
+ private static final AgentServerVersion INSTANCE = new AgentServerVersion();
+
+ private final String composedHeader;
+
+ private AgentServerVersion() {
+ String javaVersion = System.getProperty("java.version", "unknown");
+ composedHeader = "azure-ai-agentserver-java/" + getLibraryVersion() + " (java/" + javaVersion + ")";
+ }
+
+ /**
+ * Returns the singleton instance.
+ */
+ public static AgentServerVersion getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * Returns the {@code x-platform-server} header value for this library.
+ */
+ public String getHeaderValue() {
+ return composedHeader;
+ }
+
+ private static String getLibraryVersion() {
+ Package pkg = AgentServerVersion.class.getPackage();
+ if (pkg != null && pkg.getImplementationVersion() != null) {
+ return pkg.getImplementationVersion();
+ }
+ return "1.0.0-SNAPSHOT";
+ }
+}
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/ApiError.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/ApiError.java
new file mode 100644
index 000000000000..91cebc0e7ac6
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/ApiError.java
@@ -0,0 +1,97 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Structured HTTP error body as defined by the API spec.
+ * Serialised inside an outer envelope: {@code { "error": ApiError }}.
+ *
+ * Required fields: {@link #message}, {@link #type}, {@link #code} (which may be
+ * {@code null} but is always present). Optional fields ({@link #param},
+ * {@link #details}, {@link #additionalInfo}) are omitted when {@code null}.
+ *
+ * @param message human-readable description (always present)
+ * @param type error category, e.g. {@code "invalid_request_error"}, {@code "server_error"} (always present)
+ * @param code sub-code such as {@code "invalid_parameters"}, {@code "unsupported_parameter"};
+ * always serialised, may be {@code null}
+ * @param param the offending request parameter, JSON-path or path-param name; optional
+ * @param details nested error objects for multi-error validation; optional
+ * @param additionalInfo arbitrary supplemental key/value context; optional
+ */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public record ApiError(
+ @JsonProperty("message") String message,
+ @JsonProperty("type") String type,
+ // `code` is always serialised (may be null) per the envelope spec.
+ @JsonInclude(JsonInclude.Include.ALWAYS) @JsonProperty("code") String code,
+ @JsonProperty("param") String param,
+ @JsonProperty("details") List details,
+ @JsonProperty("additionalInfo") Map additionalInfo) {
+
+ /**
+ * Standard error type for client-side / 4xx errors.
+ */
+ public static final String TYPE_INVALID_REQUEST = "invalid_request_error";
+ /**
+ * Standard error type for unhandled server errors (500).
+ */
+ public static final String TYPE_SERVER_ERROR = "server_error";
+
+ /**
+ * Generic 4xx sub-code used when no more specific code applies.
+ */
+ public static final String CODE_INVALID_REQUEST = "invalid_request_error";
+ /**
+ * Malformed path identifier.
+ */
+ public static final String CODE_INVALID_PARAMETERS = "invalid_parameters";
+ /**
+ * Parameter present but unsupported in this combination.
+ */
+ public static final String CODE_UNSUPPORTED_PARAMETER = "unsupported_parameter";
+ /**
+ * Required parameter missing from the request.
+ */
+ public static final String CODE_MISSING_REQUIRED_PARAMETER = "missing_required_parameter";
+ /**
+ * Unknown / unrecognised parameter on the request.
+ */
+ public static final String CODE_UNKNOWN_PARAMETER = "unknown_parameter";
+ /**
+ * A generic field-level validation failure inside a {@code details[]} entry.
+ */
+ public static final String CODE_INVALID_VALUE = "invalid_value";
+ /**
+ * Unhandled server-side failure.
+ */
+ public static final String CODE_SERVER_ERROR = "server_error";
+
+ /**
+ * Convenience for the common 4xx case with no offending parameter.
+ */
+ public static ApiError invalidRequest(String message) {
+ return new ApiError(message, TYPE_INVALID_REQUEST, CODE_INVALID_REQUEST, null, null, null);
+ }
+
+ /**
+ * Convenience for a 4xx case with a specific sub-code and offending parameter.
+ */
+ public static ApiError invalidRequest(String message, String code, String param) {
+ return new ApiError(message, TYPE_INVALID_REQUEST, code, param, null, null);
+ }
+
+ /**
+ * Convenience for a 500.
+ */
+ public static ApiError serverError(String message) {
+ return new ApiError(message, TYPE_SERVER_ERROR, CODE_SERVER_ERROR, null, null, null);
+ }
+}
+
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/ApiException.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/ApiException.java
new file mode 100644
index 000000000000..58dc85a29e6f
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/ApiException.java
@@ -0,0 +1,50 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import java.util.Objects;
+
+/**
+ * Exception thrown when an API operation fails.
+ *
+ * Carries an HTTP status code and a structured {@link ApiError} body that the
+ * hosting framework adapter serialises as the standard error envelope
+ * ({@code { "error": { "message", "type", "code", "param"?, "details"?, "additionalInfo"? } }})
+ * per the API spec.
+ */
+public class ApiException extends Exception {
+
+ private final int statusCode;
+ private final ApiError error;
+
+ /**
+ * Creates an {@code ApiException} with the given status code and structured error body.
+ *
+ * @param statusCode the HTTP status code representing the error
+ * @param error the structured error body (required)
+ */
+ public ApiException(int statusCode, ApiError error) {
+ super("API error: HTTP " + statusCode + " — " + Objects.requireNonNull(error, "error").message());
+ this.statusCode = statusCode;
+ this.error = error;
+ }
+
+ /**
+ * Returns the HTTP status code associated with this exception.
+ *
+ * @return the status code
+ */
+ public int getStatusCode() {
+ return statusCode;
+ }
+
+ /**
+ * Returns the structured error body, never {@code null}.
+ *
+ * @return the structured {@link ApiError}
+ */
+ public ApiError getError() {
+ return error;
+ }
+}
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/CreateResponse.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/CreateResponse.java
new file mode 100644
index 000000000000..b1e8873b8cb1
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/CreateResponse.java
@@ -0,0 +1,20 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import com.fasterxml.jackson.annotation.JsonUnwrapped;
+import com.openai.models.responses.Response;
+
+/**
+ * Wraps an OpenAI {@link Response} together with an optional {@link AgentReference}
+ * for the agent server protocol's create-response endpoint.
+ *
+ * The {@code response} field is serialized unwrapped (its properties merge into
+ * the top-level JSON object), while the {@code agent} field is a nested object.
+ *
+ * @param agent the agent that produced this response, or {@code null}
+ * @param response the OpenAI Response object
+ */
+public record CreateResponse(AgentReference agent, @JsonUnwrapped Response response) {
+}
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/EventReplayStore.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/EventReplayStore.java
new file mode 100644
index 000000000000..80c467685d4b
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/EventReplayStore.java
@@ -0,0 +1,164 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.agentserver.api.serialization.ObjectMapperFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * Process-scoped, in-memory buffer of serialized SSE events keyed by response ID,
+ * used to support SSE replay via {@code GET /responses/{id}?stream=true}.
+ *
+ *
+ * Each event is retained for a minimum of {@code ttl} from when it was buffered
+ * (default 10 minutes). The per-event TTL means individual events may be
+ * evicted independently — JSON GET is unaffected by replay-buffer eviction.
+ *
+ * Events may be appended {@linkplain #append(String, ResponseEvent) one at a time}
+ * (live, while the response is still streaming — supports mid-stream replay) or in
+ * a single batch via {@linkplain #store(String, List)}. {@linkplain #initBuffer(String)}
+ * registers an empty buffer so {@linkplain #hasBuffer(String)} returns {@code true}
+ * even before the first event arrives — required so a replay request that races
+ * the first emission is not mis-classified as "not created with stream=true".
+ */
+final class EventReplayStore {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EventReplayStore.class);
+
+ /**
+ * Default per-event replay retention.
+ */
+ static final Duration DEFAULT_TTL = Duration.ofMinutes(10);
+
+ private final Duration ttl;
+ private final Map> buffers = new ConcurrentHashMap<>();
+
+ EventReplayStore() {
+ this(DEFAULT_TTL);
+ }
+
+ EventReplayStore(Duration ttl) {
+ this.ttl = ttl;
+ }
+
+ /**
+ * A single buffered event: its sequence number, name, JSON payload, and expiry.
+ */
+ record StoredEvent(long sequenceNumber, String eventName, String data, long expiresAtMillis) {
+ }
+
+ /**
+ * Registers an empty replay buffer for a response. Used by the streaming
+ * create path so {@link #hasBuffer(String)} reflects "was created with
+ * stream=true" before any event has been appended.
+ */
+ void initBuffer(String responseId) {
+ buffers.computeIfAbsent(responseId, k -> new CopyOnWriteArrayList<>());
+ }
+
+ /**
+ * Appends a single event to the replay buffer (used for live mid-stream
+ * buffering of background streaming responses). Idempotent on the buffer
+ * existence — if no buffer was {@linkplain #initBuffer(String) initialised}
+ * first, one is created on demand.
+ */
+ void append(String responseId, ResponseEvent event) {
+ if (event == null) {
+ return;
+ }
+ StoredEvent stored = toStoredEvent(event, buffers.getOrDefault(responseId, List.of()).size());
+ if (stored == null) {
+ return;
+ }
+ buffers.computeIfAbsent(responseId, k -> new CopyOnWriteArrayList<>()).add(stored);
+ }
+
+ /**
+ * Replaces the buffer for {@code responseId} with the supplied events.
+ * Convenience for the "buffer at terminal" pattern.
+ */
+ void store(String responseId, List events) {
+ if (events == null || events.isEmpty()) {
+ buffers.put(responseId, new CopyOnWriteArrayList<>());
+ return;
+ }
+ List stored = new CopyOnWriteArrayList<>();
+ long index = 0;
+ for (ResponseEvent event : events) {
+ StoredEvent s = toStoredEvent(event, index);
+ if (s != null) {
+ stored.add(s);
+ }
+ index++;
+ }
+ buffers.put(responseId, stored);
+ }
+
+ private StoredEvent toStoredEvent(ResponseEvent event, long fallbackIndex) {
+ ObjectMapper mapper = ObjectMapperFactory.getObjectMapper();
+ try {
+ String data = mapper.writeValueAsString(event.streamEvent());
+ long seq = fallbackIndex;
+ JsonNode node = mapper.readTree(data);
+ if (node.hasNonNull("sequence_number")) {
+ seq = node.get("sequence_number").asLong(fallbackIndex);
+ }
+ long expiresAt = System.currentTimeMillis() + ttl.toMillis();
+ return new StoredEvent(seq, event.eventName(), data, expiresAt);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to serialize replay event {}", fallbackIndex, e);
+ return null;
+ }
+ }
+
+ /**
+ * Returns {@code true} if a replay buffer exists for the response — i.e. it was
+ * created with {@code stream=true} (even if all its events have since expired).
+ */
+ boolean hasBuffer(String responseId) {
+ return buffers.containsKey(responseId);
+ }
+
+ /**
+ * Returns the non-expired events for a response with {@code sequence_number}
+ * greater than {@code startingAfter} (when provided), in order.
+ */
+ Optional> replay(String responseId, Integer startingAfter) {
+ List all = buffers.get(responseId);
+ if (all == null) {
+ return Optional.empty();
+ }
+ long now = System.currentTimeMillis();
+ long after = startingAfter != null ? startingAfter : Long.MIN_VALUE;
+ List live = new ArrayList<>();
+ for (StoredEvent e : all) {
+ if (e.expiresAtMillis() <= now) {
+ continue; // per-event TTL eviction
+ }
+ if (e.sequenceNumber() <= after) {
+ continue; // starting_after cursor
+ }
+ live.add(new ResponseStreamReplay.ReplayEvent(e.sequenceNumber(), e.eventName(), e.data()));
+ }
+ return Optional.of(live);
+ }
+
+ /**
+ * Removes any buffered events for a response (best-effort, on delete).
+ */
+ void delete(String responseId) {
+ buffers.remove(responseId);
+ }
+}
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/FoundryEnvironment.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/FoundryEnvironment.java
new file mode 100644
index 000000000000..70a3b15c4711
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/FoundryEnvironment.java
@@ -0,0 +1,167 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import com.azure.core.credential.TokenCredential;
+import com.azure.identity.DefaultAzureCredentialBuilder;
+import com.azure.identity.ManagedIdentityCredentialBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URI;
+
+/**
+ * Provides strongly-typed access to Foundry platform environment variables
+ * injected by the Azure AI Foundry hosting infrastructure.
+ *
+ * All values are read once at class load and cached for the lifetime of the process.
+ * This class is a static utility and cannot be instantiated.
+ *
+ * Equivalent to the C# {@code FoundryEnvironment}.
+ */
+public final class FoundryEnvironment {
+
+ /**
+ * The agent name. Sourced from {@code FOUNDRY_AGENT_NAME}.
+ */
+ public static final String AGENT_NAME = System.getenv("FOUNDRY_AGENT_NAME");
+ /**
+ * The agent version. Sourced from {@code FOUNDRY_AGENT_VERSION}.
+ */
+ public static final String AGENT_VERSION = System.getenv("FOUNDRY_AGENT_VERSION");
+ /**
+ * The Foundry project endpoint. Sourced from {@code FOUNDRY_PROJECT_ENDPOINT},
+ * with fallback to {@code AZURE_AI_PROJECT_ENDPOINT}.
+ */
+ public static final String PROJECT_ENDPOINT = resolveProjectEndpoint();
+ /**
+ * The full ARM ID of the Foundry project. Sourced from {@code FOUNDRY_PROJECT_ARM_ID}.
+ */
+ public static final String PROJECT_ARM_ID = System.getenv("FOUNDRY_PROJECT_ARM_ID");
+ /**
+ * The session ID. Sourced from {@code FOUNDRY_AGENT_SESSION_ID}.
+ */
+ public static final String SESSION_ID = System.getenv("FOUNDRY_AGENT_SESSION_ID");
+ /**
+ * The model deployment name. Sourced from {@code MODEL_DEPLOYMENT_NAME}.
+ * Default: {@code "gpt-4.1-mini"}.
+ */
+ public static final String MODEL_DEPLOYMENT_NAME = resolveWithDefault("MODEL_DEPLOYMENT_NAME", "gpt-4.1-mini");
+ /**
+ * The Azure managed identity client ID. Sourced from {@code AZURE_CLIENT_ID}.
+ *
+ * When set, prefer {@code ManagedIdentityCredential} with this client ID.
+ * When absent, {@code DefaultAzureCredential} should be used.
+ */
+ public static final String AZURE_CLIENT_ID = System.getenv("AZURE_CLIENT_ID");
+ /**
+ * The Azure OpenAI endpoint derived from the project endpoint (scheme + host),
+ * with fallback to {@code AZURE_OPENAI_ENDPOINT} or {@code AZURE_ENDPOINT}.
+ *
+ * For example, if {@link #PROJECT_ENDPOINT} is
+ * {@code "https://account.services.ai.azure.com/api/projects/proj"},
+ * this resolves to {@code "https://account.services.ai.azure.com"}.
+ */
+ public static final String OPENAI_ENDPOINT = resolveOpenAiEndpoint();
+ /**
+ * The HTTP listen port. Sourced from {@code PORT}. Default: 8088.
+ */
+ public static final int PORT = resolvePort();
+ /**
+ * The OTLP exporter endpoint. Sourced from {@code OTEL_EXPORTER_OTLP_ENDPOINT}.
+ */
+ public static final String OTLP_ENDPOINT = System.getenv("OTEL_EXPORTER_OTLP_ENDPOINT");
+ /**
+ * The Application Insights connection string. Sourced from {@code APPLICATIONINSIGHTS_CONNECTION_STRING}.
+ *
+ * Security note: This value is a credential. Avoid logging or exposing it in diagnostics.
+ */
+ public static final String APP_INSIGHTS_CONNECTION_STRING = System.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING");
+ /**
+ * Indicates whether the process is running in a Foundry hosted environment.
+ * Returns {@code true} when the {@code FOUNDRY_HOSTING_ENVIRONMENT} environment variable
+ * is set to a non-empty value.
+ */
+ public static final boolean IS_HOSTED = isNonEmpty(System.getenv("FOUNDRY_HOSTING_ENVIRONMENT"));
+ private static final Logger LOGGER = LoggerFactory.getLogger(FoundryEnvironment.class);
+
+ private FoundryEnvironment() {
+ // Static utility class
+ }
+
+ /**
+ * Resolves the appropriate {@link TokenCredential} for the current environment.
+ *
+ * When {@link #AZURE_CLIENT_ID} is set (typical in hosted environments with managed identity),
+ * returns a {@code ManagedIdentityCredential} scoped to that client ID.
+ * Otherwise, returns a {@code DefaultAzureCredential} for local development and
+ * environments without an explicit managed identity.
+ *
+ * @return a {@link TokenCredential} suitable for authenticating with Azure services
+ */
+ public static TokenCredential resolveCredential() {
+ if (isNonEmpty(AZURE_CLIENT_ID)) {
+ LOGGER.debug("Using ManagedIdentityCredential with client ID: {}", AZURE_CLIENT_ID);
+ return new ManagedIdentityCredentialBuilder()
+ .clientId(AZURE_CLIENT_ID)
+ .build();
+ }
+ LOGGER.debug("Using DefaultAzureCredential");
+ return new DefaultAzureCredentialBuilder().build();
+ }
+
+ private static String resolveProjectEndpoint() {
+ String endpoint = System.getenv("FOUNDRY_PROJECT_ENDPOINT");
+ if (isNonEmpty(endpoint)) {
+ return endpoint;
+ }
+ return System.getenv("AZURE_AI_PROJECT_ENDPOINT");
+ }
+
+ private static String resolveOpenAiEndpoint() {
+ // Derive from project endpoint (scheme + host)
+ if (isNonEmpty(PROJECT_ENDPOINT)) {
+ try {
+ URI uri = URI.create(PROJECT_ENDPOINT);
+ return uri.getScheme() + "://" + uri.getHost();
+ } catch (IllegalArgumentException ignored) {
+ // fall through to explicit env vars
+ }
+ }
+ // Fallback to explicit env vars
+ String explicit = System.getenv("AZURE_OPENAI_ENDPOINT");
+ if (isNonEmpty(explicit)) {
+ return explicit;
+ }
+ return System.getenv("AZURE_ENDPOINT");
+ }
+
+ private static String resolveWithDefault(String envVar, String defaultValue) {
+ String value = System.getenv(envVar);
+ return isNonEmpty(value) ? value : defaultValue;
+ }
+
+ private static int resolvePort() {
+ String portEnv = System.getenv("PORT");
+ if (portEnv == null || portEnv.isEmpty()) {
+ return 8088;
+ }
+ try {
+ int port = Integer.parseInt(portEnv);
+ if (port < 1 || port > 65535) {
+ throw new IllegalStateException(
+ "The PORT environment variable value '" + portEnv + "' is not a valid port number (1–65535).");
+ }
+ return port;
+ } catch (NumberFormatException e) {
+ throw new IllegalStateException(
+ "The PORT environment variable value '" + portEnv + "' is not a valid port number (1–65535).", e);
+ }
+ }
+
+ private static boolean isNonEmpty(String value) {
+ return value != null && !value.isEmpty();
+ }
+}
+
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/HealthApi.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/HealthApi.java
new file mode 100644
index 000000000000..beaf47206b67
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/HealthApi.java
@@ -0,0 +1,41 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+/**
+ * Interface for Kubernetes-style health probe endpoints.
+ *
+ * Implementations indicate whether the application is ready to serve traffic
+ * (readiness) and whether the process is alive (liveness).
+ *
+ * The default implementation always returns {@code true} for both probes.
+ * Override to add custom health checks (e.g., database connectivity, downstream
+ * service availability).
+ */
+public interface HealthApi {
+
+ /**
+ * Returns whether the application is ready to serve requests.
+ *
+ * A {@code false} return value causes the orchestrator to stop routing traffic
+ * to this instance until it becomes ready again.
+ *
+ * @return {@code true} if ready, {@code false} otherwise
+ */
+ default boolean isReady() {
+ return true;
+ }
+
+ /**
+ * Returns whether the application process is alive and responsive.
+ *
+ * A {@code false} return value causes the orchestrator to restart the container.
+ *
+ * @return {@code true} if alive, {@code false} otherwise
+ */
+ default boolean isAlive() {
+ return true;
+ }
+}
+
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/IsolationContext.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/IsolationContext.java
new file mode 100644
index 000000000000..a20d9e93d407
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/IsolationContext.java
@@ -0,0 +1,131 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Carries the platform-injected isolation keys for a single request.
+ * The Foundry platform sets {@code x-agent-user-isolation-key} and
+ * {@code x-agent-chat-isolation-key} headers on every protocol request.
+ * These opaque partition keys let handlers scope user-private and
+ * conversation-shared state without inspecting user identity.
+ *
+ * User isolation key — the partition for data that belongs to the
+ * individual who initiated the request (e.g., OAuth tokens, personal memory,
+ * per-user preferences, cache entries). Stable for a given user across sessions.
+ *
+ * Chat isolation key — the partition for conversation-scoped state
+ * (e.g., conversation history, turn state, shared files). In a 1:1 user↔agent
+ * chat this equals the user isolation key. It differs only in shared-surface
+ * scenarios (e.g., a Teams group chat) where it represents the common partition
+ * all participants write to.
+ *
+ * Both keys are opaque, platform-generated, and scoped to the agent — data
+ * cannot leak between agents. Neither key is guaranteed to be present when
+ * running locally (outside the platform); handlers should handle {@code null}
+ * values gracefully (e.g., fall back to a default partition).
+ */
+public record IsolationContext(String userIsolationKey, String chatIsolationKey) {
+
+ /**
+ * An empty {@link IsolationContext} with both keys {@code null}.
+ * Used when the platform headers are absent (e.g., local development).
+ */
+ public static final IsolationContext EMPTY = new IsolationContext(null, null);
+
+ /**
+ * Creates a new {@link IsolationContext} with the given keys.
+ *
+ * @param userIsolationKey the value of the {@code x-agent-user-isolation-key} header,
+ * or {@code null} if the header was absent.
+ * @param chatIsolationKey the value of the {@code x-agent-chat-isolation-key} header,
+ * or {@code null} if the header was absent.
+ */
+ public IsolationContext {
+ }
+
+ /**
+ * Extracts isolation keys from a map of HTTP headers (case-insensitive lookup).
+ * Returns {@link #EMPTY} when neither platform header is present.
+ *
+ * @param headers a map of header name → value (case-insensitive keys recommended).
+ * @return an {@link IsolationContext} with the extracted keys.
+ */
+ public static IsolationContext fromHeaders(Map headers) {
+ if (headers == null || headers.isEmpty()) {
+ return EMPTY;
+ }
+ String userKey = normalizeHeaderValue(headers.get(PlatformHeaders.USER_ISOLATION_KEY));
+ String chatKey = normalizeHeaderValue(headers.get(PlatformHeaders.CHAT_ISOLATION_KEY));
+
+ if (userKey == null && chatKey == null) {
+ return EMPTY;
+ }
+ return new IsolationContext(userKey, chatKey);
+ }
+
+ /**
+ * Extracts client headers (those prefixed with {@code x-client-}) from a map of
+ * HTTP request headers. The prefix is preserved in the returned keys.
+ *
+ * @param headers a map of all request headers.
+ * @return an unmodifiable map of client headers (may be empty, never null).
+ */
+ public static Map extractClientHeaders(Map headers) {
+ if (headers == null || headers.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ Map clientHeaders = new LinkedHashMap<>();
+ for (Map.Entry entry : headers.entrySet()) {
+ String key = entry.getKey();
+ if (key != null && key.toLowerCase().startsWith(PlatformHeaders.CLIENT_HEADER_PREFIX)) {
+ clientHeaders.put(key.toLowerCase(), entry.getValue());
+ }
+ }
+ return Collections.unmodifiableMap(clientHeaders);
+ }
+
+ /**
+ * Gets the user isolation key — the partition under which
+ * user-private state should be stored.
+ *
+ * @return an opaque string from {@code x-agent-user-isolation-key},
+ * or {@code null} when running outside the platform.
+ */
+ @Override
+ public String userIsolationKey() {
+ return userIsolationKey;
+ }
+
+ /**
+ * Gets the chat isolation key — the partition under which
+ * conversation / shared state should be stored.
+ *
+ * @return an opaque string from {@code x-agent-chat-isolation-key},
+ * or {@code null} when running outside the platform.
+ * In a 1:1 user↔agent chat this equals {@link #userIsolationKey ()}.
+ */
+ @Override
+ public String chatIsolationKey() {
+ return chatIsolationKey;
+ }
+
+ /**
+ * Returns {@code true} if both isolation keys are absent.
+ */
+ public boolean isEmpty() {
+ return userIsolationKey == null && chatIsolationKey == null;
+ }
+
+ private static String normalizeHeaderValue(String value) {
+ if (value == null || value.isBlank()) {
+ return null;
+ }
+ return value.trim();
+ }
+}
+
diff --git a/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/Observability.java b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/Observability.java
new file mode 100644
index 000000000000..a8aaa5b0e6da
--- /dev/null
+++ b/sdk/agentserver/azure-agentserver-api/src/main/java/com/microsoft/agentserver/api/Observability.java
@@ -0,0 +1,404 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.agentserver.api;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapGetter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+/**
+ * Provides OpenTelemetry instrumentation for the Azure AI Foundry Agent Server server.
+ *
+ * This class uses the OpenTelemetry API which is a no-op by default. It becomes active when:
+ *
+ * - The OpenTelemetry Java Agent is attached via {@code javaagent:} JVM flag
+ * - The OpenTelemetry SDK is programmatically configured (e.g., via {@code opentelemetry-sdk-extension-autoconfigure})
+ * - The Application Insights Java Agent is attached
+ *
+ *
+ * The instrumentation scope is {@code "Azure.AI.AgentServer.Responses"}, matching the.NET convention.
+ *
+ * Environment variables (used by OTel SDK/Agent when present):
+ *
+ * - {@code OTEL_EXPORTER_OTLP_ENDPOINT} — OTLP exporter endpoint
+ * - {@code OTEL_SERVICE_NAME} — service name (defaults to agent name)
+ * - {@code APPLICATIONINSIGHTS_CONNECTION_STRING} — enables Application Insights export
+ *
+ */
+public final class Observability {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(Observability.class);
+
+ /**
+ * Instrumentation scope name for the protocol, matching.NET convention.
+ */
+ public static final String SCOPE_RESPONSES = "Azure.AI.AgentServer.Responses";
+
+ /**
+ * Instrumentation scope name for the core hosting layer.
+ */
+ public static final String SCOPE_CORE = "Azure.AI.AgentServer.Core";
+
+ private static volatile Tracer responsesTracer;
+ private static volatile Tracer coreTracer;
+
+ private Observability() {
+ // Static utility class
+ }
+
+ /**
+ * Gets the tracer for the protocol scope.
+ * Uses {@link GlobalOpenTelemetry} which is auto-configured by the OTel Java Agent
+ * or SDK autoconfigure module.
+ *
+ * @return the protocol tracer.
+ */
+ public static Tracer getResponsesTracer() {
+ if (responsesTracer == null) {
+ responsesTracer = GlobalOpenTelemetry.getTracer(SCOPE_RESPONSES);
+ }
+ return responsesTracer;
+ }
+
+ /**
+ * Gets the tracer for the core hosting scope.
+ *
+ * @return the Core hosting tracer.
+ */
+ public static Tracer getCoreTracer() {
+ if (coreTracer == null) {
+ coreTracer = GlobalOpenTelemetry.getTracer(SCOPE_CORE);
+ }
+ return coreTracer;
+ }
+
+ /**
+ * Starts a new SERVER span for an incoming HTTP request.
+ * Extracts trace context from request headers (W3C traceparent) using
+ * the global propagator.
+ *
+ * @param spanName the span name (e.g., "POST /responses").
+ * @param headers the request headers for context propagation.
+ * @return the started span (caller must close/end it).
+ */
+ public static Span startServerSpan(String spanName, Map headers) {
+ OpenTelemetry otel = GlobalOpenTelemetry.get();
+ Context extractedContext = otel.getPropagators()
+ .getTextMapPropagator()
+ .extract(Context.current(), headers, MapTextMapGetter.INSTANCE);
+
+ return getCoreTracer()
+ .spanBuilder(spanName)
+ .setSpanKind(SpanKind.SERVER)
+ .setParent(extractedContext)
+ .startSpan();
+ }
+
+ /**
+ * Records standard HTTP attributes on a span.
+ *
+ * @param span the span to set attributes on.
+ * @param method the HTTP method (GET, POST, etc.).
+ * @param path the request path.
+ * @param statusCode the HTTP response status code.
+ */
+ public static void setHttpAttributes(Span span, String method, String path, int statusCode) {
+ span.setAttribute("http.request.method", method);
+ span.setAttribute("url.path", path);
+ span.setAttribute("http.response.status_code", statusCode);
+
+ if (statusCode >= 500) {
+ span.setStatus(StatusCode.ERROR, "HTTP " + statusCode);
+ }
+ }
+
+ /**
+ * Starts the spec-required {@code invoke_agent {model}} span.
+ *
+ * Sets all required GenAI semantic-convention tags and namespaced
+ * {@code azure.ai.agentserver.responses.*} tags so the platform's
+ * trajectory UI can render the response in context.
+ *
+ * @param model resolved model name (may be {@code null} or empty)
+ * @param responseId the response id (e.g. {@code caresp_…}); never null/empty
+ * @param conversationId conversation id, or {@code null} if absent
+ * @param agentName agent name, or {@code null}
+ * @param agentVersion agent version, or {@code null}
+ * @param streaming whether this is a streaming invocation
+ * @return the started span; caller is responsible for ending it
+ */
+ public static Span startInvokeAgentSpan(
+ String model,
+ String responseId,
+ String conversationId,
+ String agentName,
+ String agentVersion,
+ boolean streaming) {
+
+ String spanName = (model != null && !model.isEmpty())
+ ? "invoke_agent " + model
+ : "invoke_agent";
+
+ Span span = getResponsesTracer()
+ .spanBuilder(spanName)
+ .setSpanKind(SpanKind.SERVER)
+ .startSpan();
+
+ // GenAI semantic-convention identity tags
+ span.setAttribute("service.name", "azure.ai.agentserver");
+ span.setAttribute("gen_ai.provider.name", "AzureAI Hosted Agents");
+ span.setAttribute("gen_ai.operation.name", "invoke_agent");
+ span.setAttribute("gen_ai.response.id", responseId);
+ if (model != null && !model.isEmpty()) {
+ span.setAttribute("gen_ai.request.model", model);
+ }
+ if (conversationId != null && !conversationId.isEmpty()) {
+ span.setAttribute("gen_ai.conversation.id", conversationId);
+ }
+
+ String agentId;
+ if (agentName != null && !agentName.isEmpty()) {
+ agentId = agentName + ":" + (agentVersion != null ? agentVersion : "");
+ span.setAttribute("gen_ai.agent.name", agentName);
+ if (agentVersion != null && !agentVersion.isEmpty()) {
+ span.setAttribute("gen_ai.agent.version", agentVersion);
+ }
+ } else {
+ agentId = "";
+ }
+ span.setAttribute("gen_ai.agent.id", agentId);
+
+ // Namespaced tags for dashboards / correlation
+ span.setAttribute("azure.ai.agentserver.responses.response_id", responseId);
+ span.setAttribute("azure.ai.agentserver.responses.conversation_id",
+ conversationId != null ? conversationId : "");
+ span.setAttribute("azure.ai.agentserver.responses.streaming", streaming);
+
+ String projectArmId = FoundryEnvironment.PROJECT_ARM_ID;
+ if (projectArmId != null && !projectArmId.isEmpty()) {
+ span.setAttribute("microsoft.foundry.project.id", projectArmId);
+ }
+
+ return span;
+ }
+
+ /**
+ * Records an error on an {@code invoke_agent} span per
+ * the spec (error tags). Sets the status to ERROR
+ * and adds the recommended namespaced + OTel exception tags.
+ */
+ public static void recordInvokeAgentError(Span span, String code, String message, Throwable t) {
+ if (span == null) {
+ return;
+ }
+ if (code != null) {
+ span.setAttribute("azure.ai.agentserver.responses.error.code", code);
+ span.setAttribute("error.type", code);
+ }
+ if (message != null) {
+ span.setAttribute("azure.ai.agentserver.responses.error.message", message);
+ span.setAttribute("otel.status_description", message);
+ }
+ if (t != null) {
+ span.recordException(t);
+ }
+ span.setStatus(StatusCode.ERROR, message != null ? message : "");
+ maybeLogRbacGuidance(t);
+ }
+
+ /**
+ * Pattern matching a missing data-action message of the form:
+ * "The principal `` lacks the required data action `/`..."
+ * emitted by Azure Cognitive Services / OpenAI on 401 PermissionDenied.
+ */
+ private static final java.util.regex.Pattern PRINCIPAL_LACKS_ACTION = java.util.regex.Pattern.compile(
+ "principal\\s+`?([0-9a-fA-F-]{36})`?\\s+lacks the required data action\\s+`?([^`\\s\"]+)`?",
+ java.util.regex.Pattern.CASE_INSENSITIVE);
+
+ /**
+ * Backstop: catches the generic "Principal does not have access to API/Operation."
+ */
+ private static final java.util.regex.Pattern GENERIC_PRINCIPAL_DENIED = java.util.regex.Pattern.compile(
+ "Principal does not have access to API/Operation",
+ java.util.regex.Pattern.CASE_INSENSITIVE);
+
+ private static final java.util.Set RBAC_GUIDANCE_PRINCIPALS =
+ java.util.Collections.synchronizedSet(new java.util.HashSet<>());
+
+ /**
+ * Detects Azure RBAC-related authentication failures from an exception
+ * chain and prints a one-time human-readable remediation message that
+ * includes the exact {@code az role assignment create} command to run.
+ */
+ public static void maybeLogRbacGuidance(Throwable t) {
+ if (t == null) {
+ return;
+ }
+
+ Throwable cur = t;
+ for (int depth = 0; cur != null && depth < 8; depth++, cur = cur.getCause()) {
+ String msg = cur.getMessage();
+ if (msg == null) {
+ continue;
+ }
+ java.util.regex.Matcher m = PRINCIPAL_LACKS_ACTION.matcher(msg);
+ if (m.find()) {
+ String principalId = m.group(1);
+ String action = m.group(2);
+ if (RBAC_GUIDANCE_PRINCIPALS.add("data-action:" + principalId + ":" + action)) {
+ logDataActionGuidance(principalId, action);
+ }
+ return;
+ }
+ if (GENERIC_PRINCIPAL_DENIED.matcher(msg).find()) {
+ if (RBAC_GUIDANCE_PRINCIPALS.add("generic:" + System.identityHashCode(cur))) {
+ logGenericPrincipalDeniedGuidance();
+ }
+ return;
+ }
+ }
+ }
+
+ private static void logDataActionGuidance(String principalId, String action) {
+ String role = guessRoleForAction(action);
+ StringBuilder sb = new StringBuilder();
+ sb.append("\n");
+ sb.append("==============================================================================\n");
+ sb.append(" AZURE RBAC: The agent's managed identity is missing a required role.\n");
+ sb.append("==============================================================================\n");
+ sb.append(" Principal id : ").append(principalId).append("\n");
+ sb.append(" Missing action : ").append(action).append("\n");
+ if (role != null) {
+ sb.append(" Suggested role : \"").append(role).append("\"\n");
+ }
+ sb.append("\n");
+ sb.append(" Run the following to grant access (replace with the resource id of\n");
+ sb.append(" the Cognitive Services / Foundry account the request was made against):\n");
+ sb.append("\n");
+ sb.append(" az role assignment create \\\n");
+ sb.append(" --assignee ").append(principalId).append(" \\\n");
+ sb.append(" --role \"").append(role != null ? role : "Cognitive Services OpenAI User").append("\" \\\n");
+ sb.append(" --scope \n");
+ sb.append("\n");
+ sb.append(" Allow ~30s for the assignment to propagate, then retry.\n");
+ sb.append("==============================================================================");
+ LOGGER.error(sb.toString());
+ }
+
+ private static void logGenericPrincipalDeniedGuidance() {
+ LOGGER.error("""
+ ==============================================================================
+ AZURE RBAC: The agent's managed identity is missing a required role.
+ ==============================================================================
+ The platform returned: "Principal does not have access to API/Operation."
+ The principal id is not in the message, but it is the agent's instance
+ identity (visible via `az agentservers show` or in the agent metadata).
+
+ For Azure OpenAI calls against a Foundry project, the agent typically needs:
+ - "Cognitive Services OpenAI User" on the Cognitive Services account, AND
+ - "Azure AI User" on the project (scope ending in /projects/)
+
+ Use:
+ az role assignment create --assignee \\
+ --role "Azure AI User" --scope
+ ==============================================================================""");
+ }
+
+ private static String guessRoleForAction(String action) {
+ if (action == null) {
+ return null;
+ }
+ String lower = action.toLowerCase(java.util.Locale.ROOT);
+ if (lower.contains("openai") && lower.contains("chat")) {
+ return "Cognitive Services OpenAI User";
+ }
+ if (lower.contains("openai")) {
+ return "Cognitive Services OpenAI Contributor";
+ }
+ if (lower.contains("cognitiveservices")) {
+ return "Cognitive Services User";
+ }
+ return null;
+ }
+
+ /**
+ * Starts the spec-required {@code tools/call {tool_name}} span
+ * for outbound calls to the Foundry
+ * Toolboxes (MCP) proxy. Use {@link #recordToolboxResult} to set the
+ * success/failure tags before ending the span.
+ *
+ * @param toolboxName the toolbox name used for this invocation
+ * @param fullToolName fully-qualified tool name, e.g. {@code everything.echo}
+ * @param serverLabel server-label prefix, e.g. {@code everything}
+ * @return the started span; caller is responsible for ending it
+ */
+ public static Span startToolboxCallSpan(String toolboxName, String fullToolName, String serverLabel) {
+ Span span = getResponsesTracer()
+ .spanBuilder("tools/call " + fullToolName)
+ .setSpanKind(SpanKind.CLIENT)
+ .startSpan();
+ if (toolboxName != null) {
+ span.setAttribute("azure.ai.agentserver.toolbox.name", toolboxName);
+ }
+ if (fullToolName != null) {
+ span.setAttribute("azure.ai.agentserver.toolbox.tool_name", fullToolName);
+ }
+ if (serverLabel != null) {
+ span.setAttribute("azure.ai.agentserver.toolbox.server_label", serverLabel);
+ }
+ return span;
+ }
+
+ /**
+ * Records the success/failure outcome and latency on a {@link
+ * #startToolboxCallSpan} span.
+ */
+ public static void recordToolboxResult(Span span, boolean success, long latencyMs,
+ String errorCode, String errorMessage, Throwable t) {
+ if (span == null) {
+ return;
+ }
+ span.setAttribute("azure.ai.agentserver.toolbox.success", success);
+ span.setAttribute("azure.ai.agentserver.toolbox.latency_ms", latencyMs);
+ if (!success) {
+ if (errorCode != null) {
+ span.setAttribute("azure.ai.agentserver.toolbox.error.code", errorCode);
+ }
+ if (errorMessage != null) {
+ span.setAttribute("azure.ai.agentserver.toolbox.error.message", errorMessage);
+ }
+ if (t != null) {
+ span.recordException(t);
+ }
+ span.setStatus(StatusCode.ERROR, errorMessage != null ? errorMessage : "");
+ }
+ }
+
+ /**
+ * TextMapGetter for extracting trace context from a {@code Map}.
+ */
+ private enum MapTextMapGetter implements TextMapGetter