This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 36d7515fe7 [Fix][Core] Fix SeaTunnel CLI reasoning replay for tool
calls (#10902)
36d7515fe7 is described below
commit 36d7515fe7be40bf300bd01a6ef6558398613906
Author: yzeng1618 <[email protected]>
AuthorDate: Sat May 23 09:34:02 2026 +0800
[Fix][Core] Fix SeaTunnel CLI reasoning replay for tool calls (#10902)
Co-authored-by: zengyi <[email protected]>
---
seatunnel-cli/README.md | 10 +
seatunnel-cli/README.zh-CN.md | 10 +
seatunnel-cli/env.example.sh | 1 +
seatunnel-cli/seatunnel_cli/cli.py | 4 +-
seatunnel-cli/seatunnel_cli/llm_provider.py | 263 +++++++++++++++++++++++-
seatunnel-cli/seatunnel_cli/skills.py | 4 +-
seatunnel-cli/tests/test_llm_provider_openai.py | 258 +++++++++++++++++++++++
7 files changed, 537 insertions(+), 13 deletions(-)
diff --git a/seatunnel-cli/README.md b/seatunnel-cli/README.md
index f2e5055417..f268e3ff12 100644
--- a/seatunnel-cli/README.md
+++ b/seatunnel-cli/README.md
@@ -101,6 +101,9 @@ export
ANTHROPIC_SMALL_FAST_MODEL='us.anthropic.claude-haiku-4-5-20251001-v1:0'
# export AWS_SECRET_ACCESS_KEY=...
```
+The Bedrock provider preserves Claude `reasoningContent` blocks in streamed
+Converse responses when Bedrock returns them.
+
#### Option B: Anthropic API
```bash
@@ -112,6 +115,9 @@ export ANTHROPIC_MODEL=claude-sonnet-4-20250514
export ANTHROPIC_SMALL_FAST_MODEL=claude-haiku-4-5-20251001
```
+The Anthropic provider preserves Claude thinking blocks (`thinking`,
`signature`, and
+`redacted_thinking`) in assistant history when the API returns them.
+
#### Option C: OpenAI API
```bash
@@ -124,6 +130,9 @@ export OPENAI_SMALL_FAST_MODEL=gpt-4o-mini
# Custom base URL for compatible APIs (Azure OpenAI, local models, etc.)
# export OPENAI_BASE_URL=https://your-endpoint.openai.azure.com/
+
+# Keep enabled for compatible reasoning models that require reasoning_content
replay
+# export OPENAI_ECHO_REASONING_CONTENT=true
```
### SEATUNNEL_HOME
@@ -173,6 +182,7 @@ When the engine is running, the CLI operates in **cluster
mode** with live conne
| `ANTHROPIC_API_KEY` | Anthropic | -- | Anthropic API key |
| `OPENAI_API_KEY` | OpenAI | -- | OpenAI API key |
| `OPENAI_BASE_URL` | No | -- | Custom endpoint for OpenAI-compatible APIs |
+| `OPENAI_ECHO_REASONING_CONTENT` | No | `true` | Preserve and replay
`reasoning_content` for OpenAI-compatible reasoning models such as DeepSeek or
GLM thinking mode |
| `ANTHROPIC_MODEL` | No | Provider default | Override primary model ID |
| `ANTHROPIC_SMALL_FAST_MODEL` | No | Provider default | Override fast model
ID |
| `OPENAI_MODEL` | No | `gpt-4o` | Primary model for OpenAI provider |
diff --git a/seatunnel-cli/README.zh-CN.md b/seatunnel-cli/README.zh-CN.md
index ef3cb39880..f59ea644ff 100644
--- a/seatunnel-cli/README.zh-CN.md
+++ b/seatunnel-cli/README.zh-CN.md
@@ -101,6 +101,9 @@ export
ANTHROPIC_SMALL_FAST_MODEL='us.anthropic.claude-haiku-4-5-20251001-v1:0'
# export AWS_SECRET_ACCESS_KEY=...
```
+当 Bedrock 流式 Converse 响应返回 Claude `reasoningContent` blocks 时，
+Bedrock provider 会在 assistant 历史中保留并回传。
+
#### 方案 B：Anthropic API
```bash
@@ -112,6 +115,9 @@ export ANTHROPIC_MODEL=claude-sonnet-4-20250514
export ANTHROPIC_SMALL_FAST_MODEL=claude-haiku-4-5-20251001
```
+当 API 返回 Claude thinking blocks 时，Anthropic provider 会在 assistant 历史中保留并回传
+`thinking`、`signature` 和 `redacted_thinking`。
+
#### 方案 C：OpenAI API
```bash
@@ -124,6 +130,9 @@ export OPENAI_SMALL_FAST_MODEL=gpt-4o-mini
# 兼容 API 的自定义端点（Azure OpenAI、本地模型等）
# export OPENAI_BASE_URL=https://your-endpoint.openai.azure.com/
+
+# 对需要 reasoning_content 回放的兼容推理模型保持开启
+# export OPENAI_ECHO_REASONING_CONTENT=true
```
### SEATUNNEL_HOME
@@ -173,6 +182,7 @@ export SEATUNNEL_API_BASE=http://localhost:5801 # 默认值
| `ANTHROPIC_API_KEY` | Anthropic 必需 | -- | Anthropic API 密钥 |
| `OPENAI_API_KEY` | OpenAI 必需 | -- | OpenAI API 密钥 |
| `OPENAI_BASE_URL` | 否 | -- | OpenAI 兼容 API 的自定义端点 |
+| `OPENAI_ECHO_REASONING_CONTENT` | 否 | `true` | 为 DeepSeek、GLM 思考模式等 OpenAI
兼容推理模型保留并回传 `reasoning_content` |
| `ANTHROPIC_MODEL` | 否 | 提供商默认值 | 覆盖主模型 ID |
| `ANTHROPIC_SMALL_FAST_MODEL` | 否 | 提供商默认值 | 覆盖快速模型 ID |
| `OPENAI_MODEL` | 否 | `gpt-4o` | OpenAI 提供商的主模型 |
diff --git a/seatunnel-cli/env.example.sh b/seatunnel-cli/env.example.sh
index 797b8d8b94..180f2d9161 100644
--- a/seatunnel-cli/env.example.sh
+++ b/seatunnel-cli/env.example.sh
@@ -37,6 +37,7 @@
# export OPENAI_MODEL=gpt-4o # optional override
# export OPENAI_SMALL_FAST_MODEL=gpt-4o-mini # optional override
# export OPENAI_BASE_URL= # optional: for Azure,
DeepSeek, local models, etc.
+# export OPENAI_ECHO_REASONING_CONTENT=true # optional: keep true for
reasoning models that require replay
# ─── Option C: AWS Bedrock (AI_PROVIDER=bedrock) ───
# export AWS_REGION=us-east-1
diff --git a/seatunnel-cli/seatunnel_cli/cli.py
b/seatunnel-cli/seatunnel_cli/cli.py
index 709d52ef04..e357d99783 100644
--- a/seatunnel-cli/seatunnel_cli/cli.py
+++ b/seatunnel-cli/seatunnel_cli/cli.py
@@ -39,7 +39,7 @@ from prompt_toolkit.completion import WordCompleter
from prompt_toolkit.history import FileHistory
from . import __version__, get_data_dir
-from .llm_provider import create_provider
+from .llm_provider import create_provider, format_llm_error
from .agents import Orchestrator
@@ -928,7 +928,7 @@ class SeaTunnelCLI:
result = self.orchestrator.process_user_input(user_input)
except Exception as e:
self._stop_live()
- self.console.print(f"\n[error]Error: {e}[/error]")
+ self.console.print(f"\n[error]Error:
{format_llm_error(e)}[/error]")
import traceback
self.console.print(f"[dim]{traceback.format_exc()}[/dim]")
return None
diff --git a/seatunnel-cli/seatunnel_cli/llm_provider.py
b/seatunnel-cli/seatunnel_cli/llm_provider.py
index d5663bbfd2..ef91498c0b 100644
--- a/seatunnel-cli/seatunnel_cli/llm_provider.py
+++ b/seatunnel-cli/seatunnel_cli/llm_provider.py
@@ -36,6 +36,40 @@ from typing import Generator
logger = logging.getLogger(__name__)
+
+def _env_bool(name: str, default: bool) -> bool:
+ """Read a boolean environment flag."""
+ value = os.environ.get(name)
+ if value is None:
+ return default
+ return value.strip().lower() not in {"0", "false", "no", "off"}
+
+
+def format_llm_error(error: Exception) -> str:
+ """Return an actionable error message for common provider protocol
issues."""
+ message = str(error)
+ lower = message.lower()
+ if "reasoning_content" in lower and "passed back" in lower:
+ return (
+ f"{message}\n\n"
+ "Hint: This looks like an OpenAI-compatible reasoning model "
+ "requiring reasoning_content replay after a tool call. SeaTunnel
CLI "
+ "preserves and sends reasoning_content by default for the OpenAI "
+ "provider. If this session was created before the fix, start a
fresh "
+ "session with /new. Also make sure OPENAI_ECHO_REASONING_CONTENT
is "
+ "not set to false."
+ )
+ if "thinking" in lower and "signature" in lower:
+ return (
+ f"{message}\n\n"
+ "Hint: This looks like a Claude thinking blocks replay issue. "
+ "SeaTunnel CLI preserves Anthropic thinking/signature blocks and "
+ "Bedrock reasoningContent blocks in assistant history. If this "
+ "session was created before the fix, start a fresh session with
/new."
+ )
+ return message
+
+
# ─── Common internal message format ───
# We reuse the Bedrock Converse API message shape as our internal format:
#
@@ -49,6 +83,11 @@ logger = logging.getLogger(__name__)
# "stopReason": "end_turn" | "tool_use",
# }
#
+# Provider-owned reasoning state is preserved as opaque replay data:
+# {"reasoningContent": "..."} # OpenAI-compatible
+# {"reasoningContent": {"reasoningText": {...}}} # Bedrock Converse
+# {"anthropicThinking": {"thinking": "...", ...}} # Anthropic Messages
+#
# Tools follow the Bedrock Converse toolSpec shape:
# {"toolSpec": {"name": "...", "description": "...", "inputSchema": {"json":
{...}}}}
@@ -130,17 +169,112 @@ class LLMProvider(abc.ABC):
"""Reconstruct a full internal-format response from collected stream
events."""
content: list[dict] = []
current_text = ""
+ current_reasoning = ""
+ current_anthropic_thinking: dict | None = None
+ current_bedrock_reasoning: dict | None = None
current_tool: dict | None = None
stop_reason = "end_turn"
+ def flush_text() -> None:
+ nonlocal current_text
+ if current_text:
+ content.append({"text": current_text})
+ current_text = ""
+
+ def flush_reasoning() -> None:
+ nonlocal current_reasoning
+ if current_reasoning:
+ content.append({"reasoningContent": current_reasoning})
+ current_reasoning = ""
+
+ def flush_anthropic_thinking() -> None:
+ nonlocal current_anthropic_thinking
+ if current_anthropic_thinking:
+ block = {
+ "thinking": current_anthropic_thinking.get("thinking", ""),
+ }
+ signature = current_anthropic_thinking.get("signature", "")
+ if signature:
+ block["signature"] = signature
+ if block["thinking"] or signature:
+ content.append({"anthropicThinking": block})
+ current_anthropic_thinking = None
+
+ def flush_bedrock_reasoning() -> None:
+ nonlocal current_bedrock_reasoning
+ if current_bedrock_reasoning:
+ reasoning_text = current_bedrock_reasoning.get("reasoningText")
+ if isinstance(reasoning_text, dict) and not
reasoning_text.get("signature"):
+ reasoning_text.pop("signature", None)
+ content.append({"reasoningContent": current_bedrock_reasoning})
+ current_bedrock_reasoning = None
+
+ def flush_model_state() -> None:
+ flush_reasoning()
+ flush_anthropic_thinking()
+ flush_bedrock_reasoning()
+
for event in events:
etype = event.get("type", "")
if etype == "text_delta":
+ flush_model_state()
current_text += event["text"]
+ elif etype == "reasoning_delta":
+ flush_text()
+ flush_anthropic_thinking()
+ flush_bedrock_reasoning()
+ current_reasoning += event["text"]
+ elif etype == "thinking_start":
+ flush_text()
+ flush_reasoning()
+ flush_bedrock_reasoning()
+ flush_anthropic_thinking()
+ current_anthropic_thinking = {
+ "thinking": event.get("thinking", ""),
+ "signature": event.get("signature", ""),
+ }
+ elif etype == "thinking_delta":
+ flush_text()
+ flush_reasoning()
+ flush_bedrock_reasoning()
+ if current_anthropic_thinking is None:
+ current_anthropic_thinking = {"thinking": "", "signature":
""}
+ current_anthropic_thinking["thinking"] += event.get("text", "")
+ elif etype == "signature_delta":
+ if current_anthropic_thinking is None:
+ current_anthropic_thinking = {"thinking": "", "signature":
""}
+ current_anthropic_thinking["signature"] +=
event.get("signature", "")
+ elif etype == "thinking_stop":
+ flush_anthropic_thinking()
+ elif etype == "redacted_thinking":
+ flush_text()
+ flush_model_state()
+ content.append({
+ "anthropicRedactedThinking": {"data": event.get("data",
"")}
+ })
+ elif etype == "bedrock_reasoning_delta":
+ flush_text()
+ flush_reasoning()
+ flush_anthropic_thinking()
+ redacted = event.get("redacted_content")
+ if redacted is not None:
+ flush_bedrock_reasoning()
+ content.append({"reasoningContent": {"redactedContent":
redacted}})
+ else:
+ if (
+ current_bedrock_reasoning is None
+ or "reasoningText" not in current_bedrock_reasoning
+ ):
+ current_bedrock_reasoning = {
+ "reasoningText": {"text": "", "signature": ""}
+ }
+ reasoning_text = current_bedrock_reasoning["reasoningText"]
+ reasoning_text["text"] += event.get("text", "")
+ if event.get("signature"):
+ reasoning_text["signature"] += event["signature"]
elif etype == "tool_start":
- if current_text:
- content.append({"text": current_text})
- current_text = ""
+ flush_text()
+ flush_model_state()
current_tool = {
"toolUseId": event["tool_use_id"],
"name": event["name"],
@@ -166,8 +300,8 @@ class LLMProvider(abc.ABC):
elif etype == "message_stop":
stop_reason = event.get("stop_reason", "end_turn")
- if current_text:
- content.append({"text": current_text})
+ flush_text()
+ flush_model_state()
return {
"output": {"message": {"role": "assistant", "content": content}},
@@ -263,7 +397,21 @@ class BedrockProvider(LLMProvider):
yield {"type": "tool_start", "tool_use_id":
current_tool_id, "name": tu.get("name", "")}
elif "contentBlockDelta" in event:
delta = event["contentBlockDelta"].get("delta", {})
- if "text" in delta:
+ if "reasoningContent" in delta:
+ rc = delta["reasoningContent"]
+ reasoning_text = rc.get("reasoningText", {})
+ if reasoning_text:
+ yield {
+ "type": "bedrock_reasoning_delta",
+ "text": reasoning_text.get("text", ""),
+ "signature": reasoning_text.get("signature", ""),
+ }
+ if "redactedContent" in rc:
+ yield {
+ "type": "bedrock_reasoning_delta",
+ "redacted_content": rc["redactedContent"],
+ }
+ elif "text" in delta:
yield {"type": "text_delta", "text": delta["text"]}
elif "toolUse" in delta:
yield {"type": "tool_input_delta", "tool_use_id":
current_tool_id or "", "delta": delta["toolUse"].get("input", "")}
@@ -357,14 +505,29 @@ class AnthropicProvider(LLMProvider):
kwargs["tools"] = self._to_anthropic_tools(tools)
current_tool_id = None
+ block_types: dict[int, str] = {}
with self._client.messages.stream(**kwargs) as stream:
for event in stream:
etype = getattr(event, "type", "")
if etype == "content_block_start":
+ index = getattr(event, "index", 0)
block = getattr(event, "content_block", None)
- if block and getattr(block, "type", "") == "tool_use":
+ block_type = getattr(block, "type", "") if block else ""
+ block_types[index] = block_type
+ if block_type == "tool_use":
current_tool_id = getattr(block, "id", "")
yield {"type": "tool_start", "tool_use_id":
current_tool_id, "name": getattr(block, "name", "")}
+ elif block_type == "thinking":
+ yield {
+ "type": "thinking_start",
+ "thinking": getattr(block, "thinking", ""),
+ "signature": getattr(block, "signature", ""),
+ }
+ elif block_type == "redacted_thinking":
+ yield {
+ "type": "redacted_thinking",
+ "data": getattr(block, "data", ""),
+ }
elif etype == "content_block_delta":
delta = getattr(event, "delta", None)
if delta:
@@ -373,10 +536,18 @@ class AnthropicProvider(LLMProvider):
yield {"type": "text_delta", "text":
getattr(delta, "text", "")}
elif dt == "input_json_delta":
yield {"type": "tool_input_delta", "tool_use_id":
current_tool_id or "", "delta": getattr(delta, "partial_json", "")}
+ elif dt == "thinking_delta":
+ yield {"type": "thinking_delta", "text":
getattr(delta, "thinking", "")}
+ elif dt == "signature_delta":
+ yield {"type": "signature_delta", "signature":
getattr(delta, "signature", "")}
elif etype == "content_block_stop":
- if current_tool_id:
+ index = getattr(event, "index", 0)
+ block_type = block_types.pop(index, "")
+ if block_type == "tool_use" and current_tool_id:
yield {"type": "tool_stop", "tool_use_id":
current_tool_id}
current_tool_id = None
+ elif block_type == "thinking":
+ yield {"type": "thinking_stop"}
elif etype == "message_stop":
msg = getattr(event, "message", None)
sr = getattr(msg, "stop_reason", "end_turn") if msg else
"end_turn"
@@ -393,6 +564,19 @@ class AnthropicProvider(LLMProvider):
for block in content:
if "text" in block:
anthropic_content.append({"type": "text", "text":
block["text"]})
+ elif "anthropicThinking" in block:
+ thinking = dict(block["anthropicThinking"])
+ anthropic_content.append({
+ "type": "thinking",
+ "thinking": thinking.get("thinking", ""),
+ "signature": thinking.get("signature", ""),
+ })
+ elif "anthropicRedactedThinking" in block:
+ redacted = dict(block["anthropicRedactedThinking"])
+ anthropic_content.append({
+ "type": "redacted_thinking",
+ "data": redacted.get("data", ""),
+ })
elif "toolUse" in block:
tu = block["toolUse"]
anthropic_content.append({
@@ -431,7 +615,20 @@ class AnthropicProvider(LLMProvider):
"""Convert Anthropic API response to internal format."""
content = []
for block in response.content:
- if block.type == "text":
+ if block.type == "thinking":
+ content.append({
+ "anthropicThinking": {
+ "thinking": getattr(block, "thinking", ""),
+ "signature": getattr(block, "signature", ""),
+ }
+ })
+ elif block.type == "redacted_thinking":
+ content.append({
+ "anthropicRedactedThinking": {
+ "data": getattr(block, "data", ""),
+ }
+ })
+ elif block.type == "text":
content.append({"text": block.text})
elif block.type == "tool_use":
content.append({
@@ -478,6 +675,7 @@ class OpenAIProvider(LLMProvider):
self._client = openai.OpenAI(**client_kwargs)
self._model_id = os.environ.get("OPENAI_MODEL", "gpt-4o")
self._fast_model_id = os.environ.get("OPENAI_SMALL_FAST_MODEL",
"gpt-4o-mini")
+ self._echo_reasoning_content =
_env_bool("OPENAI_ECHO_REASONING_CONTENT", True)
@property
def provider_name(self) -> str:
@@ -543,6 +741,10 @@ class OpenAIProvider(LLMProvider):
choice = chunk.choices[0]
delta = choice.delta
+ reasoning_content = self._get_openai_field(delta,
"reasoning_content")
+ if reasoning_content:
+ yield {"type": "reasoning_delta", "text": reasoning_content}
+
if delta and delta.content:
yield {"type": "text_delta", "text": delta.content}
@@ -564,6 +766,7 @@ class OpenAIProvider(LLMProvider):
def _to_openai_messages(self, messages: list[dict], system: str = "") ->
list[dict]:
"""Convert internal message format to OpenAI API format."""
result = []
+ echo_reasoning = getattr(self, "_echo_reasoning_content", True)
if system:
result.append({"role": "system", "content": system})
@@ -591,10 +794,13 @@ class OpenAIProvider(LLMProvider):
if has_tool_use:
text_parts = []
+ reasoning_parts = []
tool_calls = []
for block in content:
if "text" in block:
text_parts.append(block["text"])
+ elif "reasoningContent" in block and
isinstance(block["reasoningContent"], str):
+ reasoning_parts.append(block["reasoningContent"])
elif "toolUse" in block:
tu = block["toolUse"]
tool_calls.append({
@@ -608,6 +814,8 @@ class OpenAIProvider(LLMProvider):
msg_dict = {"role": "assistant"}
if text_parts:
msg_dict["content"] = "\n".join(text_parts)
+ if echo_reasoning and reasoning_parts:
+ msg_dict["reasoning_content"] = "\n".join(reasoning_parts)
if tool_calls:
msg_dict["tool_calls"] = tool_calls
result.append(msg_dict)
@@ -615,8 +823,21 @@ class OpenAIProvider(LLMProvider):
# Regular text message
text_parts = [block["text"] for block in content if "text" in
block]
+ reasoning_parts = [
+ block["reasoningContent"]
+ for block in content
+ if "reasoningContent" in block and
isinstance(block["reasoningContent"], str)
+ ]
if text_parts:
- result.append({"role": role, "content": "\n".join(text_parts)})
+ msg_dict = {"role": role, "content": "\n".join(text_parts)}
+ if echo_reasoning and role == "assistant" and reasoning_parts:
+ msg_dict["reasoning_content"] = "\n".join(reasoning_parts)
+ result.append(msg_dict)
+ elif echo_reasoning and role == "assistant" and reasoning_parts:
+ result.append({
+ "role": role,
+ "reasoning_content": "\n".join(reasoning_parts),
+ })
return result
@@ -643,6 +864,10 @@ class OpenAIProvider(LLMProvider):
message = choice.message
content = []
+ reasoning_content = OpenAIProvider._get_openai_field(message,
"reasoning_content")
+ if reasoning_content:
+ content.append({"reasoningContent": reasoning_content})
+
if message.content:
content.append({"text": message.content})
@@ -665,6 +890,24 @@ class OpenAIProvider(LLMProvider):
"stopReason": stop_reason,
}
+ @staticmethod
+ def _get_openai_field(obj, field_name: str):
+ """Read SDK fields, including provider-specific extra fields."""
+ if obj is None:
+ return None
+ if isinstance(obj, dict):
+ return obj.get(field_name)
+
+ value = getattr(obj, field_name, None)
+ if value is not None:
+ return value
+
+ model_extra = getattr(obj, "model_extra", None)
+ if isinstance(model_extra, dict):
+ return model_extra.get(field_name)
+
+ return None
+
# ─── Config file ───
diff --git a/seatunnel-cli/seatunnel_cli/skills.py
b/seatunnel-cli/seatunnel_cli/skills.py
index 72d1e1940e..60fcdf9d58 100644
--- a/seatunnel-cli/seatunnel_cli/skills.py
+++ b/seatunnel-cli/seatunnel_cli/skills.py
@@ -188,7 +188,9 @@ def _try_parse_json_plan(plan_text: str) -> StructuredPlan
| None:
src = p.get("source", {})
sink = p.get("sink", {})
tables = p.get("tables", [])
- if isinstance(tables, str):
+ if tables is None:
+ tables = []
+ elif isinstance(tables, str):
tables = [tables]
pipelines.append(PipelineSlot(
pipeline_id=p.get("id", f"pipeline_{len(pipelines) + 1}"),
diff --git a/seatunnel-cli/tests/test_llm_provider_openai.py
b/seatunnel-cli/tests/test_llm_provider_openai.py
new file mode 100644
index 0000000000..f44c995b82
--- /dev/null
+++ b/seatunnel-cli/tests/test_llm_provider_openai.py
@@ -0,0 +1,258 @@
+#
+# 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.
+#
+
+import unittest
+from types import SimpleNamespace
+
+from seatunnel_cli.llm_provider import (
+ LLMProvider,
+ OpenAIProvider,
+ format_llm_error,
+)
+
+
+class _FakeOpenAICompletions:
+ def __init__(self, stream):
+ self.stream = stream
+ self.kwargs = None
+
+ def create(self, **kwargs):
+ self.kwargs = kwargs
+ return self.stream
+
+
+class _FakeOpenAIClient:
+ def __init__(self, stream):
+ self.completions = _FakeOpenAICompletions(stream)
+ self.chat = SimpleNamespace(completions=self.completions)
+
+
+def _chunk(delta, finish_reason=None):
+ return SimpleNamespace(
+ choices=[
+ SimpleNamespace(
+ delta=delta,
+ finish_reason=finish_reason,
+ )
+ ]
+ )
+
+
+class OpenAIProviderReasoningContentTest(unittest.TestCase):
+ def test_openai_stream_preserves_reasoning_content_delta(self):
+ provider = OpenAIProvider.__new__(OpenAIProvider)
+ provider._model_id = "reasoning-model"
+ provider._client = _FakeOpenAIClient(
+ [
+ _chunk(
+ SimpleNamespace(
+ content=None,
+ reasoning_content="inspect connector metadata",
+ tool_calls=None,
+ )
+ ),
+ _chunk(SimpleNamespace(content="PLAN: use Jdbc",
tool_calls=None)),
+ _chunk(
+ SimpleNamespace(content=None, tool_calls=None),
+ finish_reason="stop",
+ ),
+ ]
+ )
+
+ events = list(
+ provider.chat_stream(
+ messages=[
+ {
+ "role": "user",
+ "content": [{"text": "sync oracle to iceberg"}],
+ }
+ ]
+ )
+ )
+ response = LLMProvider.collect_stream(events)
+
+ self.assertEqual(
+ response["output"]["message"]["content"],
+ [
+ {"reasoningContent": "inspect connector metadata"},
+ {"text": "PLAN: use Jdbc"},
+ ],
+ )
+
+ def test_openai_messages_send_reasoning_content_back_to_api(self):
+ provider = OpenAIProvider.__new__(OpenAIProvider)
+
+ messages = [
+ {
+ "role": "assistant",
+ "content": [
+ {"reasoningContent": "need source and sink connector
metadata"},
+ {"text": "PLAN: inspect connectors"},
+ {
+ "toolUse": {
+ "toolUseId": "tool-1",
+ "name": "get_connector_info",
+ "input": {"name": "Jdbc", "type": "source"},
+ }
+ },
+ ],
+ }
+ ]
+
+ self.assertEqual(
+ provider._to_openai_messages(messages),
+ [
+ {
+ "role": "assistant",
+ "content": "PLAN: inspect connectors",
+ "reasoning_content": "need source and sink connector
metadata",
+ "tool_calls": [
+ {
+ "id": "tool-1",
+ "type": "function",
+ "function": {
+ "name": "get_connector_info",
+ "arguments": '{"name": "Jdbc", "type":
"source"}',
+ },
+ }
+ ],
+ }
+ ],
+ )
+
+ def test_openai_messages_skip_empty_reasoning_content_for_tool_calls(self):
+ provider = OpenAIProvider.__new__(OpenAIProvider)
+
+ messages = [
+ {
+ "role": "assistant",
+ "content": [
+ {
+ "toolUse": {
+ "toolUseId": "tool-1",
+ "name": "get_connector_info",
+ "input": {"name": "Doris", "connector_type":
"sink"},
+ }
+ },
+ ],
+ }
+ ]
+
+ self.assertEqual(
+ provider._to_openai_messages(messages),
+ [
+ {
+ "role": "assistant",
+ "tool_calls": [
+ {
+ "id": "tool-1",
+ "type": "function",
+ "function": {
+ "name": "get_connector_info",
+ "arguments": '{"name": "Doris",
"connector_type": "sink"}',
+ },
+ }
+ ],
+ }
+ ],
+ )
+
+ def test_openai_messages_can_disable_reasoning_content_echo(self):
+ provider = OpenAIProvider.__new__(OpenAIProvider)
+ provider._echo_reasoning_content = False
+
+ messages = [
+ {
+ "role": "assistant",
+ "content": [
+ {"reasoningContent": "provider-specific hidden state"},
+ {"text": "done"},
+ ],
+ }
+ ]
+
+ self.assertEqual(
+ provider._to_openai_messages(messages),
+ [{"role": "assistant", "content": "done"}],
+ )
+
+ def test_openai_messages_ignore_bedrock_reasoning_content_blocks(self):
+ provider = OpenAIProvider.__new__(OpenAIProvider)
+
+ messages = [
+ {
+ "role": "assistant",
+ "content": [
+ {
+ "reasoningContent": {
+ "reasoningText": {
+ "text": "provider-specific hidden state",
+ "signature": "sig-1",
+ }
+ }
+ },
+ {"text": "done"},
+ ],
+ }
+ ]
+
+ self.assertEqual(
+ provider._to_openai_messages(messages),
+ [{"role": "assistant", "content": "done"}],
+ )
+
+ def
test_openai_response_preserves_reasoning_content_from_extra_fields(self):
+ message = SimpleNamespace(
+ content="done",
+ model_extra={"reasoning_content": "checked config shape"},
+ tool_calls=None,
+ )
+ response = SimpleNamespace(
+ choices=[SimpleNamespace(message=message, finish_reason="stop")]
+ )
+
+ self.assertEqual(
+ OpenAIProvider._from_openai_response(response),
+ {
+ "output": {
+ "message": {
+ "role": "assistant",
+ "content": [
+ {"reasoningContent": "checked config shape"},
+ {"text": "done"},
+ ],
+ }
+ },
+ "stopReason": "end_turn",
+ },
+ )
+
+ def test_reasoning_content_error_gets_actionable_hint(self):
+ err = Exception(
+ "Error code: 400 - Param Incorrect: "
+ "The reasoning_content in the thinking mode must be passed back to
the API."
+ )
+
+ message = format_llm_error(err)
+
+ self.assertIn("reasoning_content", message)
+ self.assertIn("OpenAI-compatible reasoning model", message)
+ self.assertIn("OPENAI_ECHO_REASONING_CONTENT", message)
+
+
+if __name__ == "__main__":
+ unittest.main()
