fix(workflow-operator): degrade instead of raising on malformed chat responses

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/HuggingFaceCodegenBase.scala

@@ -260,6 +260,37 @@ object HuggingFaceCodegenBase {
        |        summary = "; ".join(errors) if errors else "no providers available"
        |        return last_resp, summary
        |
+       |    def _chat_message_content(self, body):
+       |        '''Return the assistant text from a chat-completions response, or None
+       |        when the body is not that shape. Providers differ and malformed 200s
+       |        happen, so every level is type-checked rather than indexed: parsing
+       |        runs once per row, and an exception here aborts the whole run. Callers
+       |        fall back to their native shape, or to json.dumps(body), on None.
+       |        '''
+       |        if not isinstance(body, dict):
+       |            return None
+       |        choices = body.get("choices")
+       |        if not isinstance(choices, list) or not choices:
+       |            return None
+       |        first = choices[0]
+       |        if not isinstance(first, dict):
+       |            return None
+       |        message = first.get("message")
+       |        if not isinstance(message, dict):
+       |            return None
+       |        content = message.get("content")
+       |        if isinstance(content, str):
+       |            return content
+       |        if isinstance(content, list):
+       |            # Some OpenAI-compatible providers return content as a list of
+       |            # parts ({"type": "text", "text": ...}); join the text of those.
+       |            parts = [
+       |                part["text"] for part in content
+       |                if isinstance(part, dict) and isinstance(part.get("text"), str)
+       |            ]
+       |            return "".join(parts) if parts else None
+       |        return None
+       |
        |    def _chat_content_for_task(self, pipeline_payload, prompt_value):
        |        '''Reformulate a structured task (question-answering,
        |        table-question-answering, zero-shot-classification,

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/QaRankingCodegen.scala

@@ -76,21 +76,24 @@ object QaRankingCodegen extends TaskCodegen {
 
   override def parsePython(ctx: CodegenContext): String =
     """            if task == "question-answering":
+      |                # Third-party chat providers answer via choices[0].message;
+      |                # hf-inference returns the native {"answer": ...} shape.
+      |                content = self._chat_message_content(body)
+      |                if content is not None:
+      |                    return content
       |                if isinstance(body, dict):
-      |                    # Third-party chat providers answer via choices[0].message;
-      |                    # hf-inference returns the native {"answer": ...} shape.
-      |                    if "choices" in body:
-      |                        return body["choices"][0]["message"]["content"]
       |                    return body.get("answer", json.dumps(body))
       |                return json.dumps(body)
       |            elif task == "table-question-answering":
+      |                content = self._chat_message_content(body)
+      |                if content is not None:
+      |                    return content
       |                if isinstance(body, dict):
-      |                    if "choices" in body:
-      |                        return body["choices"][0]["message"]["content"]
       |                    return body.get("answer", json.dumps(body))
       |                return json.dumps(body)
       |            elif task in ("zero-shot-classification", "sentence-similarity", "text-ranking"):
-      |                if isinstance(body, dict) and "choices" in body:
-      |                    return body["choices"][0]["message"]["content"]
+      |                content = self._chat_message_content(body)
+      |                if content is not None:
+      |                    return content
       |                return json.dumps(body)""".stripMargin
 }

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/TextGenCodegen.scala

@@ -28,7 +28,9 @@ package org.apache.texera.amber.operator.huggingFace.codegen
   * (Cerebras, Groq, Sambanova, Together, …) accepts.
   *
   * The parse step pulls `body["choices"][0]["message"]["content"]` out of
-  * the response.
+  * the response, degrading to the raw JSON body when a provider returns a
+  * shape that does not carry it — parsing runs per row, so raising here
+  * would abort the whole run over one malformed response.
   */
 object TextGenCodegen extends TaskCodegen {
 
@@ -50,5 +52,8 @@ object TextGenCodegen extends TaskCodegen {
 
   override def parsePython(ctx: CodegenContext): String =
     """            if task == "text-generation":
-      |                return body["choices"][0]["message"]["content"]""".stripMargin
+      |                content = self._chat_message_content(body)
+      |                if content is not None:
+      |                    return content
+      |                return json.dumps(body)""".stripMargin
 }

common/workflow-operator/src/test/resources/python/hf_parse_probe.py

@@ -0,0 +1,76 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""Run the generated Hugging Face response parser against real response shapes.
+
+Reads {"source": <generated operator Python>, "task": <task tag>,
+"bodies": [<parsed response>, ...]} on stdin and writes
+{"results": [{"value": <cell text>} | {"raised": "<ExcType>"}, ...]} on stdout.
+
+The generated module cannot be imported as-is (it depends on the pytexera
+runtime), so the parser and its helpers are lifted out of the source and
+executed on their own. That keeps the assertion about the emitted code itself
+rather than a transcription of it.
+"""
+import io
+import json
+import re
+import sys
+
+# Exactly what _parse_response can reach: itself, the chat-content helper, and
+# the data-URL helper used by the image-to-image branch. Anything else in the
+# generated class belongs to the request loop, not to parsing.
+WANTED = ("_parse_response", "_chat_message_content", "_url_to_data_url")
+
+
+def lift(source, name):
+    """Return the text of a 4-space-indented method, or '' when absent."""
+    out, started = [], False
+    for line in source.split("\n"):
+        stripped = line.strip()
+        if stripped.startswith("def " + name + "("):
+            started = True
+        elif started and (line.startswith("    def ") or (line and not line.startswith(" "))):
+            break
+        if started:
+            out.append(line)
+    return "\n".join(out)
+
+
+def main():
+    request = json.load(sys.stdin)
+    methods = [m for m in (lift(request["source"], n) for n in WANTED) if m.strip()]
+    module = "import json\nclass Parser:\n    TASK = %r\n%s\n" % (
+        request["task"],
+        "\n".join(methods),
+    )
+    namespace = {}
+    exec(compile(module, "<generated>", "exec"), namespace)  # noqa: S102 - the point of the probe
+    parser = namespace["Parser"]()
+
+    results = []
+    for body in request["bodies"]:
+        try:
+            results.append({"value": parser._parse_response(body)})
+        except Exception as exc:  # noqa: BLE001 - reporting the type is the assertion
+            results.append({"raised": type(exc).__name__})
+    json.dump({"results": results}, sys.stdout)
+
+
+if __name__ == "__main__":
+    main()

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceInferenceOpDescSpec.scala

@@ -111,7 +111,9 @@ class HuggingFaceInferenceOpDescSpec extends AnyFlatSpec with Matchers {
     code should include("self.MAX_NEW_TOKENS")
     code should include("self.TEMPERATURE")
     // Parse — text-gen pulls choices[0].message.content out of the response.
-    code should include("""body["choices"][0]["message"]["content"]""")
+    code should include(
+      """content = self._chat_message_content(body)"""
+    )
   }
 
   it should "send the provider-specific model id on provider-scoped chat routes" in {
@@ -228,7 +230,9 @@ class HuggingFaceInferenceOpDescSpec extends AnyFlatSpec with Matchers {
       safeTemp = 0.0
     )
     TextGenCodegen.payloadPython(ctx) should include("self.MODEL_ID")
-    TextGenCodegen.parsePython(ctx) should include("""body["choices"][0]["message"]["content"]""")
+    TextGenCodegen.parsePython(ctx) should include(
+      """content = self._chat_message_content(body)"""
+    )
   }
 
   "image task family" should
@@ -568,7 +572,9 @@ class HuggingFaceInferenceOpDescSpec extends AnyFlatSpec with Matchers {
     code should include("Context column")
     code should include("""payload = {"inputs": {"question": prompt_value, "context": ctx_val}}""")
     code should include("""body.get("answer", json.dumps(body))""")
-    code should include("""body["choices"][0]["message"]["content"]""")
+    code should include(
+      """content = self._chat_message_content(body)"""
+    )
   }
 
   it should "route table-question-answering with a precomputed table payload" in {
@@ -577,7 +583,9 @@ class HuggingFaceInferenceOpDescSpec extends AnyFlatSpec with Matchers {
     code should include("table_dict = {}")
     code should include("""payload = {"inputs": {"query": prompt_value, "table": table_dict}}""")
     code should include("""body.get("answer", json.dumps(body))""")
-    code should include("""body["choices"][0]["message"]["content"]""")
+    code should include(
+      """content = self._chat_message_content(body)"""
+    )
   }
 
   it should "route zero-shot-classification with candidate labels" in {

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/codegen/HuggingFaceCodegenBaseSpec.scala

@@ -144,4 +144,19 @@ class HuggingFaceCodegenBaseSpec extends AnyFlatSpec with Matchers {
     out should not include "MARKER_TASK_zXyq42"
     out should not include "MARKER_SYSTEM_zXyq42"
   }
+
+  // Review feedback on #8617 (@Copilot): a truthy `choices` is not enough — the
+  // value may not be a list, its first item may not be a dict, and `message` or
+  // `content` may be null or a list of parts. Every level is type-checked in one
+  // shared helper so all eight chat extractions degrade identically.
+  it should "emit a type-checked chat-content helper" in {
+    val out = HuggingFaceCodegenBase.render(makeCtx(), StubCodegen)
+    out should include("def _chat_message_content(self, body):")
+    val helper = out.split("def _chat_message_content")(1).split("    def ")(0)
+    helper should include("isinstance(choices, list)")
+    helper should include("isinstance(first, dict)")
+    helper should include("isinstance(message, dict)")
+    helper should include("isinstance(content, str)")
+    helper should include("isinstance(content, list)")
+  }
 }