fix: OpenAI Prompt output for reasoning models

.agents/skills/code-review/SKILL.md

@@ -0,0 +1,16 @@
+---
+name: code-review
+description: Quick review checklist for python and scala code changes before callings it done. 
+---
+
+# Code Review
+
+Use this skill when reviewing SynapseML changes.
+
+## Steps
+
+1. Inspect diff: `git --no-pager diff --stat && git --no-pager diff`
+2. Run Scala style: `sbt scalastyle test:scalastyle`
+3. Run Python format check: `black --check --extend-exclude 'docs/' .`
+4. Run targeted tests for touched code.
+5. Report only concrete issues with file paths and fixes.

cognitive/src/main/scala/com/microsoft/azure/synapse/ml/services/openai/OpenAIPrompt.scala

@@ -381,29 +381,48 @@ class OpenAIPrompt(override val uid: String) extends Transformer
     (dfWithFilenames, pathColumnNames, filenameColMapping, templateWithFilenameRefs)
   }
 
-  private def validateResponsesApiParams(): Unit = {
-    val currentApiType = if (isSet(apiType)) getApiType else "chat_completions"
+  private def resolvedApiType: String = if (isSet(apiType)) getApiType else "chat_completions"
 
-    if (currentApiType != "responses") {
-      if (isSet(store) && getStore) {
-        throw new IllegalArgumentException(
-          "store parameter requires apiType='responses'. Use .setApiType(\"responses\")")
-      }
+  private def hasPreviousResponseIdConfigured: Boolean = {
+    get(previousResponseId).orElse(getDefault(previousResponseId)).isDefined
+  }
 
-      if (get(previousResponseId).orElse(getDefault(previousResponseId)).isDefined) {
-        throw new IllegalArgumentException(
-          "previousResponseId requires apiType='responses'. Use .setApiType(\"responses\")")
-      }
+  private def validateResponsesApiCompatibility(currentApiType: String): Unit = {
+    if (currentApiType == "responses" && hasAIFoundryModel) {
+      throw new IllegalArgumentException(
+        "apiType='responses' is not supported with AI Foundry chat endpoints. " +
+          "Use .setApiType(\"chat_completions\") or configure an OpenAI endpoint with deploymentName.")
+    }
+  }
+
+  private def validateResponsesOnlyParams(currentApiType: String): Unit = {
+    if (currentApiType != "responses" && isSet(store) && getStore) {
+      throw new IllegalArgumentException(
+        "store parameter requires apiType='responses'. Use .setApiType(\"responses\")")
     }
 
+    if (currentApiType != "responses" && hasPreviousResponseIdConfigured) {
+      throw new IllegalArgumentException(
+        "previousResponseId requires apiType='responses'. Use .setApiType(\"responses\")")
+    }
+  }
+
+  private def validateUsageColSupport(currentApiType: String): Unit = {
     if (isSet(usageCol) && !hasAIFoundryModel &&
-        getApiType != "chat_completions" && getApiType != "responses") {
+        currentApiType != "chat_completions" && currentApiType != "responses") {
       throw new IllegalArgumentException(
         s"usageCol not supported for apiType='$currentApiType'. " +
-        "Use 'chat_completions', 'responses', or AI Foundry chat APIs.")
+          "Use 'chat_completions', 'responses', or AI Foundry chat APIs.")
     }
   }
 
+  private def validateResponsesApiParams(): Unit = {
+    val currentApiType = resolvedApiType
+    validateResponsesApiCompatibility(currentApiType)
+    validateResponsesOnlyParams(currentApiType)
+    validateUsageColSupport(currentApiType)
+  }
+
   override def transform(dataset: Dataset[_]): DataFrame = {
     transferGlobalParamsToParamMap()
     validateResponsesApiParams()
@@ -613,16 +632,21 @@ class OpenAIPrompt(override val uid: String) extends Transformer
     else "unsupported"
   }
 
-  private[openai] def hasAIFoundryModel: Boolean = this.isDefined(model)
+  private def isAIFoundryEndpoint: Boolean = {
+    val host = get(url).orElse(getDefault(url)).flatMap { raw =>
+      Try(new URI(raw).getHost).toOption.orElse(Try(new URL(raw).getHost).toOption)
+    }
+    host.exists(_.toLowerCase.endsWith("services.ai.azure.com"))
+  }
+
+  private[openai] def hasAIFoundryModel: Boolean = this.isDefined(model) && isAIFoundryEndpoint
 
   //deployment name can be set by user, it doesn't have to match with model name
   private def getOpenAIChatService: OpenAIServicesBase with HasTextOutput = {
-
     val completion: OpenAIServicesBase with HasTextOutput =
       if (hasAIFoundryModel) {
         new AIFoundryChatCompletion()
-      }
-      else {
+      } else {
         // Use the apiType parameter to decide between chat_completions and responses
         getApiType match {
           case "responses" => new OpenAIResponses()
@@ -634,6 +658,13 @@ class OpenAIPrompt(override val uid: String) extends Transformer
       .filter(p => !localParamNames.contains(p.param.name) && completion.hasParam(p.param.name))
       .foreach(p => completion.set(completion.getParam(p.param.name), p.value))
 
+    completion match {
+      case resp: OpenAIResponses
+          if this.isDefined(model) && get(deploymentName).orElse(getDefault(deploymentName)).isEmpty =>
+        resp.setDeploymentName(getModel)
+      case _ =>
+    }
+
     completion
   }
 

cognitive/src/main/scala/com/microsoft/azure/synapse/ml/services/openai/OpenAIResponses.scala

@@ -224,19 +224,43 @@ class OpenAIResponses(override val uid: String) extends OpenAIServicesBase(uid)
   }
 
   override private[openai] def getOutputMessageText(outputColName: String): org.apache.spark.sql.Column = {
-    F.element_at(F.element_at(F.col(outputColName).getField("output"), 1)
-      .getField("content"), 1).getField("text")
+    val outputEntries = F.col(outputColName).getField("output")
+    val lastOutputEntry = F.element_at(outputEntries, -1)
+    val textValues = F.transform(lastOutputEntry.getField("content"), part => part.getField("text"))
+    val definedTextValues = F.filter(textValues, text => text.isNotNull)
+    F.element_at(definedTextValues, 1)
+  }
+
+  private def outputEntries(outputRow: Row): Seq[Row] = {
+    Option(outputRow).flatMap(r => Option(r.getAs[Seq[Row]]("output"))).getOrElse(Seq.empty)
+  }
+
+  private def lastOutputEntry(outputRow: Row): Option[Row] = {
+    outputEntries(outputRow).lastOption
+  }
+
+  private def firstDefinedText(contentParts: Seq[Row]): Option[String] = {
+    contentParts.iterator
+      .flatMap(part => Option(part.getAs[String]("text")))
+      .toSeq
+      .headOption
+  }
+
+  private def lastOutputText(outputRow: Row): Option[String] = {
+    lastOutputEntry(outputRow).flatMap { outputEntry =>
+      firstDefinedText(Option(outputEntry.getAs[Seq[Row]]("content")).getOrElse(Seq.empty))
+    }
   }
 
   override private[openai] def isContentFiltered(outputRow: Row): Boolean = {
-    val result = ResponsesModelResponse.makeFromRowConverter(outputRow)
-    val firstOutput = result.output.head
-    Option(firstOutput.content).isEmpty
+    lastOutputText(outputRow).isEmpty
   }
 
   override private[openai] def getFilterReason(outputRow: Row): String = {
-    val result = ResponsesModelResponse.makeFromRowConverter(outputRow)
-    result.output.head.status
+    lastOutputEntry(outputRow).iterator
+      .flatMap(outputEntry => Option(outputEntry.getAs[String]("status")))
+      .find(_.nonEmpty)
+      .getOrElse("content_filtered_or_empty")
   }
 
 }

cognitive/src/test/scala/com/microsoft/azure/synapse/ml/services/openai/OpenAIPromptResponsesSuite.scala

@@ -0,0 +1,72 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in project root for information.
+
+package com.microsoft.azure.synapse.ml.services.openai
+
+import com.microsoft.azure.synapse.ml.Secrets.getAccessToken
+import com.microsoft.azure.synapse.ml.core.test.base.Flaky
+import org.apache.spark.sql.DataFrame
+
+class OpenAIPromptResponsesSuite extends Flaky with OpenAIAPIKey {
+
+  import spark.implicits._
+
+  override def beforeAll(): Unit = {
+    val aadToken = getAccessToken("https://cognitiveservices.azure.com/")
+    println(s"Triggering token creation early ${aadToken.length}")
+    super.beforeAll()
+  }
+
+  lazy val df: DataFrame = Seq(
+    ("apple", "fruits"),
+    ("mercedes", "cars"),
+    ("cake", "dishes")
+  ).toDF("text", "category")
+
+  private def responsesPrompt(outputCol: String, deployment: String): OpenAIPrompt = {
+    new OpenAIPrompt()
+      .setSubscriptionKey(openAIAPIKey)
+      .setDeploymentName(deployment)
+      .setCustomServiceName(openAIServiceName)
+      .setApiType("responses")
+      .setApiVersion("2025-04-01-preview")
+      .setOutputCol(outputCol)
+      .setTemperature(0)
+  }
+
+  private def assertResponsesOutputForDeployment(
+      deployment: String,
+      outputCol: String,
+      expectedToken: String): Unit = {
+    val prompt = responsesPrompt(outputCol, deployment)
+      .setPromptTemplate(s"Return exactly the word $expectedToken for {text}.")
+
+    if (deployment.toLowerCase.contains("gpt-5")) {
+      prompt.setReasoningEffort("low")
+      prompt.setVerbosity("low")
+    }
+
+    val output = prompt.transform(df.limit(1))
+      .select(outputCol)
+      .collect()
+      .head
+      .getString(0)
+
+    assert(output != null)
+    assert(output.toLowerCase.contains(expectedToken.toLowerCase))
+  }
+
+  test("Responses API OpenAIPrompt returns text for gpt-4.1 outputs") {
+    assertResponsesOutputForDeployment(
+      deploymentName4p1,
+      "responses_gpt41_output",
+      "fruit")
+  }
+
+  test("Responses API OpenAIPrompt returns text for gpt-5 outputs") {
+    assertResponsesOutputForDeployment(
+      deploymentName5,
+      "responses_gpt5_output",
+      "fruit")
+  }
+}

cognitive/src/test/scala/com/microsoft/azure/synapse/ml/services/openai/OpenAIPromptSuite.scala

@@ -6,12 +6,15 @@ package com.microsoft.azure.synapse.ml.services.openai
 import com.microsoft.azure.synapse.ml.Secrets.{AIFoundryApiKey, getAccessToken}
 import com.microsoft.azure.synapse.ml.core.test.base.Flaky
 import com.microsoft.azure.synapse.ml.core.test.fuzzing.{TestObject, TransformerFuzzing}
+import org.apache.http.entity.AbstractHttpEntity
+import org.apache.http.util.EntityUtils
 import org.apache.spark.ml.util.MLReadable
 import org.apache.spark.sql.{DataFrame, Row}
 import org.apache.spark.sql.functions.{col, lit}
 import org.apache.spark.sql.types.{ArrayType, StringType, StructType}
 import org.scalactic.Equality
 import com.microsoft.azure.synapse.ml.services.aifoundry.AIFoundryAPIKey
+import spray.json._
 
 import java.nio.charset.StandardCharsets
 import java.nio.file.Files
@@ -144,7 +147,7 @@ class OpenAIPromptSuite extends TransformerFuzzing[OpenAIPrompt] with OpenAIAPIK
     assert(nonNullCount == 3)
   }
 
-    test("Basic Usage Responses API") {
+  test("Basic Usage Responses API") {
     val nonNullCount = prompt
       .setPromptTemplate("give me a comma separated list of 5 {category}, starting with {text} ")
       .setApiType("responses")
@@ -576,6 +579,36 @@ class OpenAIPromptSuite extends TransformerFuzzing[OpenAIPrompt] with OpenAIAPIK
     assert(p.getPreviousResponseIdCol == "prev_id_column")
   }
 
+  test("responses payload maps model to model field and nests verbosity/reasoning for gpt-5 usage") {
+    val p = new OpenAIPrompt()
+      .setApiType("responses")
+      .setMessagesCol("messages")
+      .setModel("gpt-5-mini")
+      .setVerbosity("low")
+      .setReasoningEffort("low")
+      .setStore(false)
+
+    val requestRow = Seq(
+      Seq(OpenAIMessage("user", "Describe what you see in this image."))
+    ).toDF("messages").collect().head
+
+    val prepareEntity = classOf[OpenAIPrompt].getDeclaredMethod("prepareEntity")
+    prepareEntity.setAccessible(true)
+    val buildEntity = prepareEntity.invoke(p).asInstanceOf[Row => Option[AbstractHttpEntity]]
+
+    val payloadJson = EntityUtils.toString(buildEntity(requestRow).get).parseJson.asJsObject
+
+    assert(payloadJson.fields.contains("input"))
+    assert(!payloadJson.fields.contains("messages"))
+    assert(payloadJson.fields.get("model").contains(JsString("gpt-5-mini")))
+
+    val text = payloadJson.fields("text").asJsObject
+    assert(text.fields.get("verbosity").contains(JsString("low")))
+
+    val reasoning = payloadJson.fields("reasoning").asJsObject
+    assert(reasoning.fields.get("effort").contains(JsString("low")))
+  }
+
   test("store=true with Responses API returns response in outputCol and id in auto-generated responseIdCol") {
     val storePrompt = new OpenAIPrompt()
       .setSubscriptionKey(openAIAPIKey)