mirror of
https://wget.la/https://github.com/leookun/cursor-byok
synced 2026-08-20 21:17:47 +08:00
上玄
5d04b5b08b
trimReplayDanglingAssistantToolCalls only collected tool results that immediately followed the assistant tool-call message. Models such as gpt-5.3-codex-spark may emit the function_call item before the explanation text within one response, so history replay order becomes assistant[tool_call] -> assistant[text] -> tool[result]. The call was misjudged as dangling and stripped while the tool result survived, producing a function_call_output without a matching function_call that the Responses API rejects with 400. - widen the response collection window to skip interleaved plain assistant text messages, and drop orphan tool results in the same pass - synthesize a placeholder function_call (or drop the output when the tool name is unknown) in normalizeOpenAIResponsesInput so conversations already persisted with corrupted history can resume
2277 lines
71 KiB
Go
2277 lines
71 KiB
Go
// openai.go 实现 OpenAI 兼容流式适配器。
|
||
package modeladapter
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"cursor/gen/agentv1"
|
||
runtimecore "cursor/internal/backend/agent/core"
|
||
"cursor/internal/modelchannel"
|
||
"cursor/internal/netproxy"
|
||
)
|
||
|
||
// OpenAIAdapter 实现 OpenAI 兼容流式请求。
|
||
type OpenAIAdapter struct {
|
||
// client 负责发送 HTTP 请求。
|
||
client *http.Client
|
||
}
|
||
|
||
type openAIRequestBody struct {
|
||
Model string `json:"model"`
|
||
Tools []json.RawMessage `json:"tools,omitempty"`
|
||
Messages []map[string]any `json:"messages"`
|
||
Stream bool `json:"stream"`
|
||
MaxTokens int `json:"max_tokens,omitempty"`
|
||
StreamOptions map[string]any `json:"stream_options"`
|
||
ReasoningEffort string `json:"reasoning_effort,omitempty"`
|
||
PromptCacheKey string `json:"prompt_cache_key,omitempty"`
|
||
}
|
||
|
||
type openAIResponsesRequestBody struct {
|
||
Model string `json:"model"`
|
||
Instructions string `json:"instructions,omitempty"`
|
||
Input []map[string]any `json:"input"`
|
||
Tools []map[string]any `json:"tools,omitempty"`
|
||
Stream bool `json:"stream"`
|
||
MaxOutputTokens int `json:"max_output_tokens,omitempty"`
|
||
Reasoning *openAIResponsesReasoning `json:"reasoning,omitempty"`
|
||
Include []string `json:"include,omitempty"`
|
||
PromptCacheKey string `json:"prompt_cache_key,omitempty"`
|
||
Store bool `json:"store"`
|
||
}
|
||
|
||
type openAIResponsesReasoning struct {
|
||
Effort string `json:"effort,omitempty"`
|
||
Summary string `json:"summary,omitempty"`
|
||
}
|
||
|
||
type openAIToolAccumulator struct {
|
||
CallID string
|
||
Name string
|
||
Args strings.Builder
|
||
LastEmittedPath string
|
||
LastStreamContent string
|
||
LastCreatePlanSnapshot string
|
||
ProviderItemID string
|
||
ProviderCallID string
|
||
ProviderStatus string
|
||
}
|
||
|
||
type openAIImageGenerationAccumulator struct {
|
||
CallID string
|
||
ImageData string
|
||
OutputFormat string
|
||
ProviderItemID string
|
||
ProviderStatus string
|
||
StartedEmitted bool
|
||
}
|
||
|
||
const (
|
||
openAIThinkOpenTag = "<think>"
|
||
openAIThinkCloseTag = "</think>"
|
||
openAIStreamMaxTokenSize = 64 * 1024 * 1024
|
||
)
|
||
|
||
type openAIContentPartKind string
|
||
|
||
const (
|
||
openAIContentPartText openAIContentPartKind = "text"
|
||
openAIContentPartReasoning openAIContentPartKind = "reasoning"
|
||
openAIContentPartThinkingCompleted openAIContentPartKind = "thinking_completed"
|
||
)
|
||
|
||
type openAIContentPart struct {
|
||
Kind openAIContentPartKind
|
||
Text string
|
||
}
|
||
|
||
// openAIThinkTagParser 负责把某些 OpenAI 兼容 provider 放进 content 的 <think> 标签拆回 reasoning 流。
|
||
type openAIThinkTagParser struct {
|
||
carry string
|
||
inThink bool
|
||
}
|
||
|
||
func (parser *openAIThinkTagParser) Consume(text string) []openAIContentPart {
|
||
if parser == nil || text == "" {
|
||
return nil
|
||
}
|
||
input := parser.carry + text
|
||
parser.carry = ""
|
||
parts := make([]openAIContentPart, 0, 4)
|
||
for input != "" {
|
||
if parser.inThink {
|
||
closeIndex := strings.Index(input, openAIThinkCloseTag)
|
||
if closeIndex >= 0 {
|
||
if closeIndex > 0 {
|
||
parts = append(parts, openAIContentPart{
|
||
Kind: openAIContentPartReasoning,
|
||
Text: input[:closeIndex],
|
||
})
|
||
}
|
||
parts = append(parts, openAIContentPart{Kind: openAIContentPartThinkingCompleted})
|
||
parser.inThink = false
|
||
input = input[closeIndex+len(openAIThinkCloseTag):]
|
||
continue
|
||
}
|
||
carryLen := trailingTagPrefixLength(input, openAIThinkCloseTag)
|
||
if emitText := input[:len(input)-carryLen]; emitText != "" {
|
||
parts = append(parts, openAIContentPart{
|
||
Kind: openAIContentPartReasoning,
|
||
Text: emitText,
|
||
})
|
||
}
|
||
parser.carry = input[len(input)-carryLen:]
|
||
break
|
||
}
|
||
|
||
openIndex := strings.Index(input, openAIThinkOpenTag)
|
||
if openIndex >= 0 {
|
||
if openIndex > 0 {
|
||
parts = append(parts, openAIContentPart{
|
||
Kind: openAIContentPartText,
|
||
Text: input[:openIndex],
|
||
})
|
||
}
|
||
parser.inThink = true
|
||
input = input[openIndex+len(openAIThinkOpenTag):]
|
||
continue
|
||
}
|
||
carryLen := trailingTagPrefixLength(input, openAIThinkOpenTag)
|
||
if emitText := input[:len(input)-carryLen]; emitText != "" {
|
||
parts = append(parts, openAIContentPart{
|
||
Kind: openAIContentPartText,
|
||
Text: emitText,
|
||
})
|
||
}
|
||
parser.carry = input[len(input)-carryLen:]
|
||
break
|
||
}
|
||
return parts
|
||
}
|
||
|
||
func (parser *openAIThinkTagParser) Flush() []openAIContentPart {
|
||
if parser == nil || parser.carry == "" {
|
||
return nil
|
||
}
|
||
kind := openAIContentPartText
|
||
if parser.inThink {
|
||
kind = openAIContentPartReasoning
|
||
}
|
||
text := parser.carry
|
||
parser.carry = ""
|
||
return []openAIContentPart{{
|
||
Kind: kind,
|
||
Text: text,
|
||
}}
|
||
}
|
||
|
||
// NewOpenAIAdapter 创建一个 OpenAI 兼容适配器。
|
||
func NewOpenAIAdapter() *OpenAIAdapter {
|
||
return &OpenAIAdapter{
|
||
client: netproxy.NewHTTPClient(0),
|
||
}
|
||
}
|
||
|
||
func openAIModelSupportsPromptCacheKey(modelID string) bool {
|
||
return strings.Contains(strings.ToLower(strings.TrimSpace(modelID)), "gpt")
|
||
}
|
||
|
||
func openAIPromptCacheKey(req StreamRequest, modelID string) string {
|
||
if !openAIModelSupportsPromptCacheKey(modelID) {
|
||
return ""
|
||
}
|
||
conversationID := strings.TrimSpace(req.ConversationID)
|
||
if conversationID == "" {
|
||
return ""
|
||
}
|
||
return "cursor:" + conversationID
|
||
}
|
||
|
||
func applyOpenAIPromptCacheKeyOverride(body map[string]any, req StreamRequest, modelID string) {
|
||
if len(body) == 0 {
|
||
return
|
||
}
|
||
if !openAIModelSupportsPromptCacheKey(modelID) {
|
||
delete(body, "prompt_cache_key")
|
||
return
|
||
}
|
||
if _, ok := body["prompt_cache_key"]; ok {
|
||
return
|
||
}
|
||
if key := openAIPromptCacheKey(req, modelID); key != "" {
|
||
body["prompt_cache_key"] = key
|
||
}
|
||
}
|
||
|
||
func shouldExposeOpenAIResponsesImageGeneration(req StreamRequest, tools []map[string]any) bool {
|
||
if !openAIResponsesToolNamePresent(tools, "GenerateImage") {
|
||
return false
|
||
}
|
||
return openAITextLooksLikeImageGenerationRequest(openAILatestUserRequestText(req))
|
||
}
|
||
|
||
func ensureOpenAIResponsesImageGenerationTool(tools []map[string]any) []map[string]any {
|
||
for _, tool := range tools {
|
||
if strings.TrimSpace(fmt.Sprint(tool["type"])) == "image_generation" {
|
||
return tools
|
||
}
|
||
}
|
||
return append(tools, map[string]any{"type": "image_generation"})
|
||
}
|
||
|
||
func openAIResponsesToolNamePresent(tools []map[string]any, name string) bool {
|
||
for _, tool := range tools {
|
||
if strings.TrimSpace(fmt.Sprint(tool["name"])) == name {
|
||
return true
|
||
}
|
||
if functionShape, ok := tool["function"].(map[string]any); ok {
|
||
if strings.TrimSpace(fmt.Sprint(functionShape["name"])) == name {
|
||
return true
|
||
}
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func openAILatestUserRequestText(req StreamRequest) string {
|
||
for i := len(req.Messages) - 1; i >= 0; i-- {
|
||
message := req.Messages[i]
|
||
if strings.TrimSpace(strings.ToLower(message.Role)) != "user" {
|
||
continue
|
||
}
|
||
text := message.Content
|
||
if strings.TrimSpace(text) == "" && len(message.ContentParts) > 0 {
|
||
text = collapseTextContentParts(message.ContentParts)
|
||
}
|
||
if tagged := textBetweenOpenAITag(text, "current_user_request"); tagged != "" {
|
||
return tagged
|
||
}
|
||
if tagged := textBetweenOpenAITag(text, "user_query"); tagged != "" {
|
||
return tagged
|
||
}
|
||
if strings.TrimSpace(text) != "" {
|
||
return strings.TrimSpace(text)
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func textBetweenOpenAITag(text string, tag string) string {
|
||
openTag := "<" + tag + ">"
|
||
closeTag := "</" + tag + ">"
|
||
start := strings.LastIndex(text, openTag)
|
||
if start < 0 {
|
||
return ""
|
||
}
|
||
start += len(openTag)
|
||
end := strings.Index(text[start:], closeTag)
|
||
if end < 0 {
|
||
return ""
|
||
}
|
||
return strings.TrimSpace(text[start : start+end])
|
||
}
|
||
|
||
func openAITextLooksLikeImageGenerationRequest(text string) bool {
|
||
trimmed := strings.TrimSpace(strings.ToLower(text))
|
||
if trimmed == "" {
|
||
return false
|
||
}
|
||
imageTerms := []string{
|
||
"图片", "图像", "照片", "相片", "人像", "头像", "插画", "海报", "壁纸", "封面", "摄影", "真实摄影",
|
||
"image", "picture", "photo", "portrait", "illustration", "poster", "wallpaper", "cover", "photorealistic",
|
||
}
|
||
for _, term := range imageTerms {
|
||
if strings.Contains(trimmed, term) {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func OpenAIEndpointURL(baseURL string, endpoint string) string {
|
||
base := strings.TrimRight(strings.TrimSpace(baseURL), "/")
|
||
normalizedEndpoint := strings.TrimSpace(endpoint)
|
||
if normalizedEndpoint == "" {
|
||
normalizedEndpoint = modelchannel.OpenAIEndpointResponses
|
||
}
|
||
if !strings.HasPrefix(normalizedEndpoint, "/") {
|
||
normalizedEndpoint = "/" + normalizedEndpoint
|
||
}
|
||
// 规则0：自定义路径模式
|
||
// - baseURL 已含 endpoint 后缀（/chat/completions 或 /responses）→ 直接用 base
|
||
// - 否则追加 /chat/completions（默认协议形态，覆盖 Z.AI /v4 等场景）
|
||
if normalizedEndpoint == modelchannel.OpenAIEndpointCustom {
|
||
if OpenAIEndpointFromBaseURL(base) != "" {
|
||
return base
|
||
}
|
||
return base + "/chat/completions"
|
||
}
|
||
// 规则1：baseURL 已含 endpoint 后缀 → 直接用 base
|
||
if OpenAIEndpointFromBaseURL(base) != "" {
|
||
return base
|
||
}
|
||
// 规则2：baseURL 以 /vN 结尾时，剥离 endpoint 的版本前缀（/v1/、/v2/ 等）
|
||
// 这样 base=.../v4 + endpoint=/v1/chat/completions → .../v4/chat/completions
|
||
if _, ok := trailingVersionSegment(base); ok {
|
||
if rest, stripped := stripEndpointVersionPrefix(normalizedEndpoint); stripped {
|
||
return base + rest
|
||
}
|
||
}
|
||
// 规则3：兜底原样拼接
|
||
return base + normalizedEndpoint
|
||
}
|
||
|
||
// trailingVersionSegment 检测 URL 末尾是否以 /vN 形式结尾（N 为数字），
|
||
// 返回版本段（如 "v4"）和是否匹配。用于通用版本段去重。
|
||
func trailingVersionSegment(base string) (string, bool) {
|
||
idx := strings.LastIndex(base, "/")
|
||
if idx < 0 {
|
||
return "", false
|
||
}
|
||
seg := base[idx+1:]
|
||
if len(seg) < 2 || seg[0] != 'v' {
|
||
return "", false
|
||
}
|
||
for i := 1; i < len(seg); i++ {
|
||
if seg[i] < '0' || seg[i] > '9' {
|
||
return "", false
|
||
}
|
||
}
|
||
return seg, true
|
||
}
|
||
|
||
// stripEndpointVersionPrefix 剥离 endpoint 路径开头的版本段前缀（/vN/），
|
||
// 返回剩余路径和是否成功剥离。
|
||
// /v1/chat/completions → ("/chat/completions", true)
|
||
// /chat/completions → ("", false)
|
||
func stripEndpointVersionPrefix(endpoint string) (string, bool) {
|
||
if len(endpoint) < 4 || endpoint[0] != '/' || endpoint[1] != 'v' {
|
||
return "", false
|
||
}
|
||
i := 2
|
||
for i < len(endpoint) && endpoint[i] >= '0' && endpoint[i] <= '9' {
|
||
i++
|
||
}
|
||
if i == 2 || i >= len(endpoint) || endpoint[i] != '/' {
|
||
return "", false
|
||
}
|
||
return endpoint[i:], true
|
||
}
|
||
|
||
func ResolveOpenAIEndpoint(baseURL string, endpoint string) string {
|
||
if endpointFromURL := OpenAIEndpointFromBaseURL(baseURL); endpointFromURL != "" {
|
||
return endpointFromURL
|
||
}
|
||
return modelchannel.NormalizeOpenAIEndpoint("openai", endpoint)
|
||
}
|
||
|
||
func OpenAIEndpointFromBaseURL(baseURL string) string {
|
||
base := strings.TrimRight(strings.TrimSpace(strings.ToLower(baseURL)), "/")
|
||
switch {
|
||
case strings.HasSuffix(base, "/responses"):
|
||
return modelchannel.OpenAIEndpointResponses
|
||
case strings.HasSuffix(base, "/chat/completions"):
|
||
return modelchannel.OpenAIEndpointChatCompletions
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func ProviderURLHasEndpoint(baseURL string, endpoints ...string) bool {
|
||
base := strings.TrimRight(strings.TrimSpace(strings.ToLower(baseURL)), "/")
|
||
if base == "" {
|
||
return false
|
||
}
|
||
for _, endpoint := range endpoints {
|
||
normalizedEndpoint := strings.TrimRight(strings.TrimSpace(strings.ToLower(endpoint)), "/")
|
||
if normalizedEndpoint == "" {
|
||
continue
|
||
}
|
||
if !strings.HasPrefix(normalizedEndpoint, "/") {
|
||
normalizedEndpoint = "/" + normalizedEndpoint
|
||
}
|
||
if strings.HasSuffix(base, normalizedEndpoint) {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// Stream 发送 OpenAI 兼容流式请求，并解析统一模型事件。
|
||
func (adapter *OpenAIAdapter) Stream(ctx context.Context, req StreamRequest, sink func(ModelEvent) error) error {
|
||
baseURL := strings.TrimRight(strings.TrimSpace(req.BaseURL), "/")
|
||
if baseURL == "" {
|
||
return fmt.Errorf("openai base url is empty")
|
||
}
|
||
apiKey := strings.TrimSpace(req.APIKey)
|
||
if apiKey == "" {
|
||
return fmt.Errorf("openai api key is empty")
|
||
}
|
||
modelID := strings.TrimSpace(req.ProviderModelID)
|
||
if modelID == "" {
|
||
modelID = strings.TrimSpace(req.ModelID)
|
||
}
|
||
if modelID == "" {
|
||
return fmt.Errorf("openai model id is empty")
|
||
}
|
||
|
||
endpoint := ResolveOpenAIEndpoint(baseURL, req.OpenAIEndpoint)
|
||
if endpoint == "" {
|
||
return fmt.Errorf("openai endpoint is unsupported: %s", strings.TrimSpace(req.OpenAIEndpoint))
|
||
}
|
||
req.OpenAIEndpoint = endpoint
|
||
if req.RequestKnobs != nil {
|
||
req.RequestKnobs["openai_endpoint"] = endpoint
|
||
if modelchannel.OpenAIEndpointShape(endpoint) == "responses" {
|
||
req.RequestKnobs["max_output_tokens"] = req.MaxTokens
|
||
}
|
||
}
|
||
if modelchannel.OpenAIEndpointShape(endpoint) == "responses" {
|
||
return adapter.streamResponses(ctx, req, baseURL, apiKey, modelID, sink)
|
||
}
|
||
return adapter.streamChatCompletions(ctx, req, baseURL, apiKey, modelID, sink)
|
||
}
|
||
|
||
func (adapter *OpenAIAdapter) streamChatCompletions(ctx context.Context, req StreamRequest, baseURL string, apiKey string, modelID string, sink func(ModelEvent) error) error {
|
||
startedAt := time.Now().UTC()
|
||
finishedAt := time.Time{}
|
||
overrideBody := cloneRequestBodyOverride(req.RequestBodyOverride)
|
||
var body any = overrideBody
|
||
if len(overrideBody) == 0 {
|
||
normalizedMessages, err := normalizeOpenAIProviderMessages(req.Messages, strings.TrimSpace(req.ReasoningEffort) != "")
|
||
if err != nil {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
requestBody := openAIRequestBody{
|
||
Model: modelID,
|
||
Messages: normalizedMessages,
|
||
Stream: true,
|
||
StreamOptions: map[string]any{"include_usage": true},
|
||
}
|
||
if shouldSendOpenAIMaxOutputTokens(modelID) {
|
||
requestBody.MaxTokens = req.MaxTokens
|
||
}
|
||
if key := openAIPromptCacheKey(req, modelID); key != "" {
|
||
requestBody.PromptCacheKey = key
|
||
}
|
||
if len(req.Tools) > 0 {
|
||
requestBody.Tools = req.Tools
|
||
}
|
||
if strings.TrimSpace(req.ReasoningEffort) != "" {
|
||
requestBody.ReasoningEffort = req.ReasoningEffort
|
||
}
|
||
body = requestBody
|
||
} else {
|
||
applyOpenAIPromptCacheKeyOverride(overrideBody, req, modelID)
|
||
}
|
||
bodyMap, err := requestBodyToMap(body)
|
||
if err != nil {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
applyOpenAIThinkingDisable(bodyMap, req, baseURL, modelID, req.OpenAIEndpoint)
|
||
if err := ApplyOpenAIExtraParams(bodyMap, req.OpenAIExtraParamsEnabled, req.OpenAIExtraParamsJSON); err != nil {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
body = bodyMap
|
||
requestURL := OpenAIEndpointURL(baseURL, req.OpenAIEndpoint)
|
||
recordLLMRequestArtifact(req, "openai", modelID, "POST", requestURL, body)
|
||
|
||
payload, err := json.Marshal(body)
|
||
if err != nil {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
|
||
streamCtx, streamIdle := newProviderStreamIdleWatchdog(ctx, req.ProviderStreamIdleTimeout)
|
||
defer streamIdle.Stop()
|
||
|
||
buildHTTPRequest := func(requestContext context.Context) (*http.Request, error) {
|
||
httpReq, err := http.NewRequestWithContext(requestContext, http.MethodPost, requestURL, bytes.NewReader(payload))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
httpReq.Header.Set("Authorization", "Bearer "+apiKey)
|
||
httpReq.Header.Set("Content-Type", "application/json")
|
||
httpReq.Header.Set("User-Agent", ClaudeCodeUserAgent)
|
||
if err := ApplyCustomHeaders(httpReq, req.CustomHeadersEnabled, req.CustomHeadersJSON); err != nil {
|
||
return nil, err
|
||
}
|
||
return httpReq, nil
|
||
}
|
||
|
||
resp, err := doProviderRequestWithRetry(streamCtx, adapter.client, "openai", req.RequestID, req.ModelCallID, buildHTTPRequest)
|
||
if err != nil {
|
||
if idleErr := streamIdle.Err(); idleErr != nil {
|
||
err = idleErr
|
||
}
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
streamIdle.AttachBody(resp.Body)
|
||
defer resp.Body.Close()
|
||
|
||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||
err = buildHTTPStatusError("openai adapter", resp)
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
|
||
type openAIToolCallDelta struct {
|
||
Index int `json:"index"`
|
||
ID string `json:"id"`
|
||
Function struct {
|
||
Name string `json:"name"`
|
||
Arguments string `json:"arguments"`
|
||
} `json:"function"`
|
||
}
|
||
type openAIChunk struct {
|
||
Type string `json:"type"`
|
||
RequestID string `json:"request_id"`
|
||
Error *struct {
|
||
Message string `json:"message"`
|
||
Type string `json:"type"`
|
||
Code string `json:"code"`
|
||
} `json:"error,omitempty"`
|
||
Choices []struct {
|
||
Delta struct {
|
||
Content string `json:"content"`
|
||
ReasoningContent string `json:"reasoning_content"`
|
||
ToolCalls []openAIToolCallDelta `json:"tool_calls"`
|
||
} `json:"delta"`
|
||
FinishReason *string `json:"finish_reason"`
|
||
} `json:"choices"`
|
||
Model string `json:"model"`
|
||
Usage *struct {
|
||
PromptTokens int64 `json:"prompt_tokens"`
|
||
CompletionTokens int64 `json:"completion_tokens"`
|
||
PromptTokensDetails *struct {
|
||
CachedTokens int64 `json:"cached_tokens"`
|
||
} `json:"prompt_tokens_details,omitempty"`
|
||
} `json:"usage,omitempty"`
|
||
}
|
||
|
||
tools := make(map[int]*openAIToolAccumulator)
|
||
currentModel := modelID
|
||
inputTokens := int64(0)
|
||
outputTokens := int64(0)
|
||
cacheReadTokens := int64(0)
|
||
cacheWriteTokens := int64(0)
|
||
usagePresent := false
|
||
cacheReadPresent := false
|
||
cacheWritePresent := false
|
||
firstEventAt := time.Time{}
|
||
finishReason := ""
|
||
turnFinishedPending := false
|
||
thinkingStarted := time.Time{}
|
||
thinkingActive := false
|
||
thinkParser := &openAIThinkTagParser{}
|
||
flushThinkingCompleted := func() error {
|
||
if !thinkingActive {
|
||
return nil
|
||
}
|
||
duration := int32(time.Since(thinkingStarted).Milliseconds())
|
||
if duration < 0 {
|
||
duration = 0
|
||
}
|
||
if err := sink(ModelEvent{
|
||
Kind: ModelEventKindThinkingCompleted,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
ThinkingDurationMS: duration,
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
thinkingActive = false
|
||
thinkingStarted = time.Time{}
|
||
return nil
|
||
}
|
||
flushTurnFinished := func() error {
|
||
if !turnFinishedPending {
|
||
return nil
|
||
}
|
||
turnFinishedPending = false
|
||
return sink(ModelEvent{
|
||
Kind: ModelEventKindTurnFinished,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
InputTokens: inputTokens,
|
||
OutputTokens: outputTokens,
|
||
CacheReadTokens: cacheReadTokens,
|
||
CacheWriteTokens: cacheWriteTokens,
|
||
UsagePresent: usagePresent,
|
||
CacheReadPresent: cacheReadPresent,
|
||
CacheWritePresent: cacheWritePresent,
|
||
FinishReason: finishReason,
|
||
})
|
||
}
|
||
emitTextDelta := func(text string) error {
|
||
if text == "" {
|
||
return nil
|
||
}
|
||
streamIdle.MarkEffectiveContent()
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return err
|
||
}
|
||
return sink(ModelEvent{
|
||
Kind: ModelEventKindTextDelta,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
Text: text,
|
||
})
|
||
}
|
||
emitThinkingDelta := func(reasoning string) error {
|
||
if reasoning == "" {
|
||
return nil
|
||
}
|
||
streamIdle.MarkEffectiveContent()
|
||
if !thinkingActive {
|
||
thinkingStarted = time.Now()
|
||
thinkingActive = true
|
||
}
|
||
return sink(ModelEvent{
|
||
Kind: ModelEventKindThinkingDelta,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
Text: reasoning,
|
||
ThinkingStyle: agentv1.ThinkingStyle_THINKING_STYLE_DEFAULT,
|
||
})
|
||
}
|
||
emitTaggedContentParts := func(parts []openAIContentPart) error {
|
||
for _, part := range parts {
|
||
switch part.Kind {
|
||
case openAIContentPartText:
|
||
if err := emitTextDelta(part.Text); err != nil {
|
||
return err
|
||
}
|
||
case openAIContentPartReasoning:
|
||
if err := emitThinkingDelta(part.Text); err != nil {
|
||
return err
|
||
}
|
||
case openAIContentPartThinkingCompleted:
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
flushTaggedContentTail := func() error {
|
||
return emitTaggedContentParts(thinkParser.Flush())
|
||
}
|
||
fail := func(streamErr error) error {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", currentModel, startedAt, firstEventAt, finishedAt, finishReason, inputTokens, outputTokens, cacheReadTokens, cacheWriteTokens, streamErr))
|
||
return streamErr
|
||
}
|
||
errorFromChunk := func(chunk openAIChunk) error {
|
||
finishReason = "error"
|
||
if chunk.Error != nil {
|
||
parts := make([]string, 0, 4)
|
||
if value := strings.TrimSpace(chunk.Error.Type); value != "" {
|
||
parts = append(parts, "type="+value)
|
||
}
|
||
if value := strings.TrimSpace(chunk.Error.Code); value != "" {
|
||
parts = append(parts, "code="+value)
|
||
}
|
||
if value := strings.TrimSpace(chunk.RequestID); value != "" {
|
||
parts = append(parts, "request_id="+value)
|
||
}
|
||
if message := strings.TrimSpace(chunk.Error.Message); message != "" {
|
||
if len(parts) > 0 {
|
||
return fmt.Errorf("openai chat stream error %s: %s", strings.Join(parts, " "), message)
|
||
}
|
||
return fmt.Errorf("openai chat stream error: %s", message)
|
||
}
|
||
if len(parts) > 0 {
|
||
return fmt.Errorf("openai chat stream error %s", strings.Join(parts, " "))
|
||
}
|
||
}
|
||
return fmt.Errorf("openai chat stream error")
|
||
}
|
||
applyUsage := func(usage *struct {
|
||
PromptTokens int64 `json:"prompt_tokens"`
|
||
CompletionTokens int64 `json:"completion_tokens"`
|
||
PromptTokensDetails *struct {
|
||
CachedTokens int64 `json:"cached_tokens"`
|
||
} `json:"prompt_tokens_details,omitempty"`
|
||
}) {
|
||
if usage == nil {
|
||
return
|
||
}
|
||
usagePresent = true
|
||
promptTokens := usage.PromptTokens
|
||
cachedTokens := int64(0)
|
||
if usage.PromptTokensDetails != nil {
|
||
cacheReadPresent = true
|
||
cachedTokens = usage.PromptTokensDetails.CachedTokens
|
||
}
|
||
if promptTokens < 0 {
|
||
promptTokens = 0
|
||
}
|
||
if cachedTokens < 0 {
|
||
cachedTokens = 0
|
||
}
|
||
if cachedTokens > promptTokens {
|
||
cachedTokens = promptTokens
|
||
}
|
||
inputTokens = promptTokens - cachedTokens
|
||
outputTokens = maxInt64(usage.CompletionTokens, 0)
|
||
cacheReadTokens = cachedTokens
|
||
cacheWriteTokens = 0
|
||
cacheWritePresent = true
|
||
}
|
||
scanner := bufio.NewScanner(resp.Body)
|
||
scanner.Buffer(make([]byte, 0, 64*1024), openAIStreamMaxTokenSize)
|
||
for scanner.Scan() {
|
||
rawLine := scanner.Text()
|
||
_, _ = appendLLMResponseArtifact(req, redactOpenAIStreamArtifactLine(rawLine)+"\n")
|
||
line := strings.TrimSpace(rawLine)
|
||
if line == "" || !strings.HasPrefix(line, "data:") {
|
||
continue
|
||
}
|
||
if firstEventAt.IsZero() {
|
||
firstEventAt = time.Now().UTC()
|
||
}
|
||
payloadLine := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
|
||
if payloadLine == "[DONE]" {
|
||
if err := flushTaggedContentTail(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushTurnFinished(); err != nil {
|
||
return fail(err)
|
||
}
|
||
break
|
||
}
|
||
|
||
var chunk openAIChunk
|
||
if err := json.Unmarshal([]byte(payloadLine), &chunk); err != nil {
|
||
return fail(err)
|
||
}
|
||
if strings.TrimSpace(chunk.Type) == "error" || chunk.Error != nil {
|
||
return fail(errorFromChunk(chunk))
|
||
}
|
||
if len(chunk.Choices) == 0 {
|
||
if strings.TrimSpace(chunk.Model) != "" {
|
||
currentModel = strings.TrimSpace(chunk.Model)
|
||
}
|
||
applyUsage(chunk.Usage)
|
||
if err := flushTaggedContentTail(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushTurnFinished(); err != nil {
|
||
return fail(err)
|
||
}
|
||
continue
|
||
}
|
||
choice := chunk.Choices[0]
|
||
if strings.TrimSpace(chunk.Model) != "" {
|
||
currentModel = strings.TrimSpace(chunk.Model)
|
||
}
|
||
applyUsage(chunk.Usage)
|
||
|
||
if text := choice.Delta.Content; text != "" {
|
||
if err := emitTaggedContentParts(thinkParser.Consume(text)); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
if reasoning := choice.Delta.ReasoningContent; reasoning != "" {
|
||
if err := emitThinkingDelta(reasoning); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
|
||
if len(choice.Delta.ToolCalls) > 0 && choice.Delta.Content == "" && choice.Delta.ReasoningContent == "" {
|
||
if err := flushTaggedContentTail(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
for _, item := range choice.Delta.ToolCalls {
|
||
streamIdle.MarkEffectiveContent()
|
||
accumulator, ok := tools[item.Index]
|
||
if !ok {
|
||
accumulator = &openAIToolAccumulator{}
|
||
tools[item.Index] = accumulator
|
||
}
|
||
if strings.TrimSpace(item.ID) != "" {
|
||
accumulator.CallID = namespaceToolCallID(req.ModelCallID, item.ID)
|
||
}
|
||
if strings.TrimSpace(item.Function.Name) != "" {
|
||
accumulator.Name = strings.TrimSpace(item.Function.Name)
|
||
}
|
||
argsTextDelta := ""
|
||
if item.Function.Arguments != "" {
|
||
_, _ = accumulator.Args.WriteString(item.Function.Arguments)
|
||
argsTextDelta = item.Function.Arguments
|
||
}
|
||
if argsTextDelta != "" || (strings.TrimSpace(accumulator.Name) == "CreatePlan" && accumulator.Args.Len() > 0) {
|
||
if err := emitOpenAIToolProgress(sink, currentModel, accumulator, argsTextDelta); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
}
|
||
|
||
if choice.FinishReason != nil && strings.TrimSpace(*choice.FinishReason) != "" {
|
||
if err := flushTaggedContentTail(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return fail(err)
|
||
}
|
||
for _, accumulator := range tools {
|
||
if err := sink(ModelEvent{
|
||
Kind: ModelEventKindToolLikeCompleted,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
ToolInvocation: &runtimecore.ToolInvocation{
|
||
CallID: strings.TrimSpace(accumulator.CallID),
|
||
ToolName: strings.TrimSpace(accumulator.Name),
|
||
ArgsJSON: []byte(accumulator.Args.String()),
|
||
},
|
||
}); err != nil {
|
||
return fail(err)
|
||
}
|
||
streamIdle.MarkEffectiveContent()
|
||
}
|
||
tools = make(map[int]*openAIToolAccumulator)
|
||
finishReason = strings.TrimSpace(*choice.FinishReason)
|
||
turnFinishedPending = true
|
||
}
|
||
}
|
||
for _, accumulator := range tools {
|
||
if err := sink(ModelEvent{
|
||
Kind: ModelEventKindToolLikeCompleted,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
ToolInvocation: &runtimecore.ToolInvocation{
|
||
CallID: strings.TrimSpace(accumulator.CallID),
|
||
ToolName: strings.TrimSpace(accumulator.Name),
|
||
ArgsJSON: []byte(accumulator.Args.String()),
|
||
},
|
||
}); err != nil {
|
||
return fail(err)
|
||
}
|
||
streamIdle.MarkEffectiveContent()
|
||
}
|
||
if err := scanner.Err(); err != nil {
|
||
if idleErr := streamIdle.Err(); idleErr != nil {
|
||
return fail(idleErr)
|
||
}
|
||
return fail(err)
|
||
}
|
||
if err := flushTaggedContentTail(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushTurnFinished(); err != nil {
|
||
return fail(err)
|
||
}
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", currentModel, startedAt, firstEventAt, finishedAt, finishReason, inputTokens, outputTokens, cacheReadTokens, cacheWriteTokens, nil))
|
||
return nil
|
||
}
|
||
|
||
func (adapter *OpenAIAdapter) streamResponses(ctx context.Context, req StreamRequest, baseURL string, apiKey string, modelID string, sink func(ModelEvent) error) error {
|
||
startedAt := time.Now().UTC()
|
||
finishedAt := time.Time{}
|
||
overrideBody := cloneRequestBodyOverride(req.RequestBodyOverride)
|
||
var body any = overrideBody
|
||
if len(overrideBody) == 0 {
|
||
instructions, input, err := normalizeOpenAIResponsesInput(req.Messages)
|
||
if err != nil {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
requestBody := openAIResponsesRequestBody{
|
||
Model: modelID,
|
||
Instructions: instructions,
|
||
Input: input,
|
||
Stream: true,
|
||
Store: false,
|
||
}
|
||
if shouldSendOpenAIMaxOutputTokens(modelID) {
|
||
requestBody.MaxOutputTokens = req.MaxTokens
|
||
}
|
||
if key := openAIPromptCacheKey(req, modelID); key != "" {
|
||
requestBody.PromptCacheKey = key
|
||
}
|
||
if len(req.Tools) > 0 {
|
||
tools, err := normalizeOpenAIResponsesTools(req.Tools)
|
||
if err != nil {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
if shouldExposeOpenAIResponsesImageGeneration(req, tools) {
|
||
tools = ensureOpenAIResponsesImageGenerationTool(tools)
|
||
if req.RequestKnobs != nil {
|
||
req.RequestKnobs["openai_responses_image_generation_tool"] = "auto"
|
||
}
|
||
}
|
||
requestBody.Tools = tools
|
||
}
|
||
if effort := strings.TrimSpace(req.ReasoningEffort); effort != "" {
|
||
requestBody.Reasoning = &openAIResponsesReasoning{Effort: effort, Summary: "auto"}
|
||
requestBody.Include = []string{"reasoning.encrypted_content"}
|
||
}
|
||
body = requestBody
|
||
} else {
|
||
applyOpenAIPromptCacheKeyOverride(overrideBody, req, modelID)
|
||
}
|
||
bodyMap, err := requestBodyToMap(body)
|
||
if err != nil {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
applyOpenAIThinkingDisable(bodyMap, req, baseURL, modelID, req.OpenAIEndpoint)
|
||
if err := ApplyOpenAIExtraParams(bodyMap, req.OpenAIExtraParamsEnabled, req.OpenAIExtraParamsJSON); err != nil {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
body = bodyMap
|
||
|
||
requestURL := OpenAIEndpointURL(baseURL, req.OpenAIEndpoint)
|
||
recordLLMRequestArtifact(req, "openai", modelID, "POST", requestURL, body)
|
||
|
||
payload, err := json.Marshal(body)
|
||
if err != nil {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
|
||
streamCtx, streamIdle := newProviderStreamIdleWatchdog(ctx, req.ProviderStreamIdleTimeout)
|
||
defer streamIdle.Stop()
|
||
|
||
buildHTTPRequest := func(requestContext context.Context) (*http.Request, error) {
|
||
httpReq, err := http.NewRequestWithContext(requestContext, http.MethodPost, requestURL, bytes.NewReader(payload))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
httpReq.Header.Set("Authorization", "Bearer "+apiKey)
|
||
httpReq.Header.Set("Content-Type", "application/json")
|
||
httpReq.Header.Set("User-Agent", ClaudeCodeUserAgent)
|
||
if err := ApplyCustomHeaders(httpReq, req.CustomHeadersEnabled, req.CustomHeadersJSON); err != nil {
|
||
return nil, err
|
||
}
|
||
return httpReq, nil
|
||
}
|
||
|
||
resp, err := doProviderRequestWithRetry(streamCtx, adapter.client, "openai", req.RequestID, req.ModelCallID, buildHTTPRequest)
|
||
if err != nil {
|
||
if idleErr := streamIdle.Err(); idleErr != nil {
|
||
err = idleErr
|
||
}
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
streamIdle.AttachBody(resp.Body)
|
||
defer resp.Body.Close()
|
||
|
||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||
err = buildHTTPStatusError("openai adapter", resp)
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", modelID, startedAt, time.Time{}, finishedAt, "", 0, 0, 0, 0, err))
|
||
return err
|
||
}
|
||
|
||
type openAIResponsesUsage struct {
|
||
InputTokens int64 `json:"input_tokens"`
|
||
OutputTokens int64 `json:"output_tokens"`
|
||
InputTokensDetails *struct {
|
||
CachedTokens int64 `json:"cached_tokens"`
|
||
} `json:"input_tokens_details,omitempty"`
|
||
}
|
||
type openAIResponsesOutputContent struct {
|
||
Type string `json:"type"`
|
||
Text string `json:"text"`
|
||
}
|
||
type openAIResponsesOutputItem struct {
|
||
ID string `json:"id"`
|
||
Type string `json:"type"`
|
||
Status string `json:"status"`
|
||
CallID string `json:"call_id"`
|
||
Name string `json:"name"`
|
||
Arguments string `json:"arguments"`
|
||
EncryptedContent string `json:"encrypted_content"`
|
||
Summary json.RawMessage `json:"summary,omitempty"`
|
||
Content []openAIResponsesOutputContent `json:"content,omitempty"`
|
||
}
|
||
type openAIResponsesResponse struct {
|
||
ID string `json:"id"`
|
||
Model string `json:"model"`
|
||
Status string `json:"status"`
|
||
Output []openAIResponsesOutputItem `json:"output,omitempty"`
|
||
OutputText string `json:"output_text,omitempty"`
|
||
Usage *openAIResponsesUsage `json:"usage,omitempty"`
|
||
IncompleteDetails *struct {
|
||
Reason string `json:"reason"`
|
||
} `json:"incomplete_details,omitempty"`
|
||
Error *struct {
|
||
Message string `json:"message"`
|
||
Type string `json:"type"`
|
||
Code string `json:"code"`
|
||
} `json:"error,omitempty"`
|
||
}
|
||
type openAIResponsesStreamEvent struct {
|
||
Type string `json:"type"`
|
||
RequestID string `json:"request_id"`
|
||
Delta string `json:"delta"`
|
||
Arguments string `json:"arguments"`
|
||
PartialImageB64 string `json:"partial_image_b64"`
|
||
OutputFormat string `json:"output_format"`
|
||
OutputIndex int `json:"output_index"`
|
||
ItemID string `json:"item_id"`
|
||
Item *openAIResponsesOutputItem `json:"item,omitempty"`
|
||
Response *openAIResponsesResponse `json:"response,omitempty"`
|
||
Error *struct {
|
||
Message string `json:"message"`
|
||
Type string `json:"type"`
|
||
Code string `json:"code"`
|
||
} `json:"error,omitempty"`
|
||
}
|
||
|
||
tools := make(map[string]*openAIToolAccumulator)
|
||
completedTools := make(map[string]struct{})
|
||
imageGenerations := make(map[string]*openAIImageGenerationAccumulator)
|
||
completedImageGenerations := make(map[string]struct{})
|
||
currentModel := modelID
|
||
inputTokens := int64(0)
|
||
outputTokens := int64(0)
|
||
cacheReadTokens := int64(0)
|
||
cacheWriteTokens := int64(0)
|
||
usagePresent := false
|
||
cacheReadPresent := false
|
||
cacheWritePresent := false
|
||
firstEventAt := time.Time{}
|
||
finishReason := ""
|
||
turnFinishedPending := false
|
||
emittedToolInvocation := false
|
||
emittedText := false
|
||
thinkingStarted := time.Time{}
|
||
thinkingActive := false
|
||
emittedReasoningSignature := ""
|
||
thinkParser := &openAIThinkTagParser{}
|
||
toolKey := func(itemID string, outputIndex int) string {
|
||
if strings.TrimSpace(itemID) != "" {
|
||
return strings.TrimSpace(itemID)
|
||
}
|
||
return fmt.Sprintf("output:%d", outputIndex)
|
||
}
|
||
effectiveFinishReason := func() string {
|
||
reason := strings.TrimSpace(finishReason)
|
||
if emittedToolInvocation && (reason == "" || reason == "completed") {
|
||
return "tool_calls"
|
||
}
|
||
return reason
|
||
}
|
||
flushThinkingCompleted := func() error {
|
||
if !thinkingActive {
|
||
return nil
|
||
}
|
||
duration := int32(time.Since(thinkingStarted).Milliseconds())
|
||
if duration < 0 {
|
||
duration = 0
|
||
}
|
||
if err := sink(ModelEvent{
|
||
Kind: ModelEventKindThinkingCompleted,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
ThinkingDurationMS: duration,
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
thinkingActive = false
|
||
thinkingStarted = time.Time{}
|
||
return nil
|
||
}
|
||
flushTurnFinished := func() error {
|
||
if !turnFinishedPending {
|
||
return nil
|
||
}
|
||
turnFinishedPending = false
|
||
return sink(ModelEvent{
|
||
Kind: ModelEventKindTurnFinished,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
InputTokens: inputTokens,
|
||
OutputTokens: outputTokens,
|
||
CacheReadTokens: cacheReadTokens,
|
||
CacheWriteTokens: cacheWriteTokens,
|
||
UsagePresent: usagePresent,
|
||
CacheReadPresent: cacheReadPresent,
|
||
CacheWritePresent: cacheWritePresent,
|
||
FinishReason: effectiveFinishReason(),
|
||
})
|
||
}
|
||
emitTextDelta := func(text string) error {
|
||
if text == "" {
|
||
return nil
|
||
}
|
||
streamIdle.MarkEffectiveContent()
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return err
|
||
}
|
||
emittedText = true
|
||
return sink(ModelEvent{
|
||
Kind: ModelEventKindTextDelta,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
Text: text,
|
||
})
|
||
}
|
||
emitThinkingDelta := func(reasoning string) error {
|
||
if reasoning == "" {
|
||
return nil
|
||
}
|
||
streamIdle.MarkEffectiveContent()
|
||
if !thinkingActive {
|
||
thinkingStarted = time.Now()
|
||
thinkingActive = true
|
||
}
|
||
return sink(ModelEvent{
|
||
Kind: ModelEventKindThinkingDelta,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
Text: reasoning,
|
||
ThinkingStyle: agentv1.ThinkingStyle_THINKING_STYLE_DEFAULT,
|
||
})
|
||
}
|
||
emitTaggedContentParts := func(parts []openAIContentPart) error {
|
||
for _, part := range parts {
|
||
switch part.Kind {
|
||
case openAIContentPartText:
|
||
if err := emitTextDelta(part.Text); err != nil {
|
||
return err
|
||
}
|
||
case openAIContentPartReasoning:
|
||
if err := emitThinkingDelta(part.Text); err != nil {
|
||
return err
|
||
}
|
||
case openAIContentPartThinkingCompleted:
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
flushTaggedContentTail := func() error {
|
||
return emitTaggedContentParts(thinkParser.Flush())
|
||
}
|
||
fail := func(streamErr error) error {
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", currentModel, startedAt, firstEventAt, finishedAt, finishReason, inputTokens, outputTokens, cacheReadTokens, cacheWriteTokens, streamErr))
|
||
return streamErr
|
||
}
|
||
applyUsage := func(usage *openAIResponsesUsage) {
|
||
if usage == nil {
|
||
return
|
||
}
|
||
usagePresent = true
|
||
promptTokens := maxInt64(usage.InputTokens, 0)
|
||
cachedTokens := int64(0)
|
||
if usage.InputTokensDetails != nil {
|
||
cacheReadPresent = true
|
||
cachedTokens = maxInt64(usage.InputTokensDetails.CachedTokens, 0)
|
||
}
|
||
if cachedTokens > promptTokens {
|
||
cachedTokens = promptTokens
|
||
}
|
||
inputTokens = promptTokens - cachedTokens
|
||
outputTokens = maxInt64(usage.OutputTokens, 0)
|
||
cacheReadTokens = cachedTokens
|
||
cacheWriteTokens = 0
|
||
cacheWritePresent = true
|
||
}
|
||
completeTool := func(key string, accumulator *openAIToolAccumulator) error {
|
||
if accumulator == nil {
|
||
return nil
|
||
}
|
||
completionKey := firstNonEmptyString(key, accumulator.CallID)
|
||
if strings.TrimSpace(completionKey) == "" {
|
||
completionKey = accumulator.Name + ":" + accumulator.Args.String()
|
||
}
|
||
if _, ok := completedTools[completionKey]; ok {
|
||
return nil
|
||
}
|
||
if strings.TrimSpace(accumulator.CallID) != "" {
|
||
if _, ok := completedTools[strings.TrimSpace(accumulator.CallID)]; ok {
|
||
return nil
|
||
}
|
||
}
|
||
completedTools[completionKey] = struct{}{}
|
||
if strings.TrimSpace(accumulator.CallID) != "" {
|
||
completedTools[strings.TrimSpace(accumulator.CallID)] = struct{}{}
|
||
}
|
||
emittedToolInvocation = true
|
||
if err := sink(ModelEvent{
|
||
Kind: ModelEventKindToolLikeCompleted,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
ToolInvocation: &runtimecore.ToolInvocation{
|
||
CallID: strings.TrimSpace(accumulator.CallID),
|
||
ToolName: strings.TrimSpace(accumulator.Name),
|
||
ArgsJSON: []byte(accumulator.Args.String()),
|
||
ProviderItemID: strings.TrimSpace(accumulator.ProviderItemID),
|
||
ProviderCallID: strings.TrimSpace(accumulator.ProviderCallID),
|
||
ProviderStatus: strings.TrimSpace(accumulator.ProviderStatus),
|
||
},
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
streamIdle.MarkEffectiveContent()
|
||
return nil
|
||
}
|
||
rememberImageGenerationItem := func(item openAIResponsesOutputItem, outputIndex int) *openAIImageGenerationAccumulator {
|
||
key := toolKey(item.ID, outputIndex)
|
||
accumulator, ok := imageGenerations[key]
|
||
if !ok {
|
||
accumulator = &openAIImageGenerationAccumulator{}
|
||
imageGenerations[key] = accumulator
|
||
}
|
||
if itemID := strings.TrimSpace(item.ID); itemID != "" {
|
||
accumulator.ProviderItemID = itemID
|
||
accumulator.CallID = namespaceToolCallID(req.ModelCallID, itemID)
|
||
}
|
||
if status := strings.TrimSpace(item.Status); status != "" {
|
||
accumulator.ProviderStatus = status
|
||
}
|
||
if strings.TrimSpace(accumulator.CallID) == "" {
|
||
accumulator.CallID = namespaceToolCallID(req.ModelCallID, key)
|
||
}
|
||
return accumulator
|
||
}
|
||
emitImageGenerationStarted := func(accumulator *openAIImageGenerationAccumulator) error {
|
||
if accumulator == nil || accumulator.StartedEmitted {
|
||
return nil
|
||
}
|
||
callID := strings.TrimSpace(accumulator.CallID)
|
||
if callID == "" {
|
||
return nil
|
||
}
|
||
accumulator.StartedEmitted = true
|
||
return sink(ModelEvent{
|
||
Kind: ModelEventKindPartialToolCall,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
ToolCallID: callID,
|
||
ToolCall: &agentv1.ToolCall{
|
||
Tool: &agentv1.ToolCall_GenerateImageToolCall{
|
||
GenerateImageToolCall: &agentv1.GenerateImageToolCall{
|
||
Args: &agentv1.GenerateImageArgs{},
|
||
},
|
||
},
|
||
},
|
||
})
|
||
}
|
||
completeImageGeneration := func(key string, accumulator *openAIImageGenerationAccumulator) error {
|
||
if accumulator == nil || strings.TrimSpace(accumulator.ImageData) == "" {
|
||
return nil
|
||
}
|
||
completionKey := firstNonEmptyString(key, accumulator.CallID, accumulator.ProviderItemID)
|
||
if strings.TrimSpace(completionKey) == "" {
|
||
completionKey = accumulator.ImageData
|
||
}
|
||
if _, ok := completedImageGenerations[completionKey]; ok {
|
||
return nil
|
||
}
|
||
if strings.TrimSpace(accumulator.CallID) != "" {
|
||
if _, ok := completedImageGenerations[strings.TrimSpace(accumulator.CallID)]; ok {
|
||
return nil
|
||
}
|
||
}
|
||
completedImageGenerations[completionKey] = struct{}{}
|
||
if strings.TrimSpace(accumulator.CallID) != "" {
|
||
completedImageGenerations[strings.TrimSpace(accumulator.CallID)] = struct{}{}
|
||
}
|
||
argsPayload := map[string]string{"image_data": strings.TrimSpace(accumulator.ImageData)}
|
||
argsJSON, err := json.Marshal(argsPayload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
emittedToolInvocation = true
|
||
if err := sink(ModelEvent{
|
||
Kind: ModelEventKindToolLikeCompleted,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
ToolInvocation: &runtimecore.ToolInvocation{
|
||
CallID: strings.TrimSpace(accumulator.CallID),
|
||
ToolName: "GenerateImage",
|
||
ArgsJSON: argsJSON,
|
||
ProviderItemID: strings.TrimSpace(accumulator.ProviderItemID),
|
||
ProviderStatus: strings.TrimSpace(accumulator.ProviderStatus),
|
||
},
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
streamIdle.MarkEffectiveContent()
|
||
return nil
|
||
}
|
||
emitReasoningSignature := func(signature string, providerItemID string, providerStatus string, providerSummary json.RawMessage) error {
|
||
trimmedSignature := strings.TrimSpace(signature)
|
||
if trimmedSignature == "" || trimmedSignature == emittedReasoningSignature {
|
||
return nil
|
||
}
|
||
duration := int32(0)
|
||
if thinkingActive {
|
||
duration = int32(time.Since(thinkingStarted).Milliseconds())
|
||
if duration < 0 {
|
||
duration = 0
|
||
}
|
||
thinkingActive = false
|
||
thinkingStarted = time.Time{}
|
||
}
|
||
emittedReasoningSignature = trimmedSignature
|
||
return sink(ModelEvent{
|
||
Kind: ModelEventKindThinkingCompleted,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: currentModel,
|
||
ThinkingDurationMS: duration,
|
||
ThinkingSignature: trimmedSignature,
|
||
ThinkingSignatureSource: ReasoningSignatureSourceOpenAIResponses,
|
||
ProviderItemID: strings.TrimSpace(providerItemID),
|
||
ProviderStatus: strings.TrimSpace(providerStatus),
|
||
ProviderSummary: cloneRawJSON(providerSummary),
|
||
})
|
||
}
|
||
applyFunctionCallItem := func(item openAIResponsesOutputItem, outputIndex int, complete bool) error {
|
||
if strings.TrimSpace(item.Type) != "function_call" {
|
||
return nil
|
||
}
|
||
streamIdle.MarkEffectiveContent()
|
||
key := toolKey(firstNonEmptyString(item.ID, item.CallID), outputIndex)
|
||
accumulator, ok := tools[key]
|
||
if !ok {
|
||
accumulator = &openAIToolAccumulator{}
|
||
tools[key] = accumulator
|
||
}
|
||
if strings.TrimSpace(item.ID) != "" {
|
||
accumulator.ProviderItemID = strings.TrimSpace(item.ID)
|
||
}
|
||
if strings.TrimSpace(item.Status) != "" {
|
||
accumulator.ProviderStatus = strings.TrimSpace(item.Status)
|
||
}
|
||
if strings.TrimSpace(item.CallID) != "" {
|
||
accumulator.ProviderCallID = strings.TrimSpace(item.CallID)
|
||
accumulator.CallID = namespaceToolCallID(req.ModelCallID, item.CallID)
|
||
} else if strings.TrimSpace(item.ID) != "" {
|
||
accumulator.CallID = namespaceToolCallID(req.ModelCallID, item.ID)
|
||
}
|
||
if strings.TrimSpace(item.Name) != "" {
|
||
accumulator.Name = strings.TrimSpace(item.Name)
|
||
}
|
||
argsTextDelta := ""
|
||
if item.Arguments != "" && accumulator.Args.Len() == 0 {
|
||
_, _ = accumulator.Args.WriteString(item.Arguments)
|
||
argsTextDelta = item.Arguments
|
||
}
|
||
if argsTextDelta != "" || (strings.TrimSpace(accumulator.Name) == "CreatePlan" && accumulator.Args.Len() > 0) {
|
||
if err := emitOpenAIToolProgress(sink, currentModel, accumulator, argsTextDelta); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if complete {
|
||
delete(tools, key)
|
||
return completeTool(key, accumulator)
|
||
}
|
||
return nil
|
||
}
|
||
applyOutputItem := func(item openAIResponsesOutputItem, outputIndex int, complete bool) error {
|
||
switch strings.TrimSpace(item.Type) {
|
||
case "reasoning":
|
||
return emitReasoningSignature(item.EncryptedContent, item.ID, item.Status, item.Summary)
|
||
case "function_call":
|
||
return applyFunctionCallItem(item, outputIndex, complete)
|
||
case "image_generation_call":
|
||
accumulator := rememberImageGenerationItem(item, outputIndex)
|
||
if !complete {
|
||
return emitImageGenerationStarted(accumulator)
|
||
}
|
||
key := toolKey(item.ID, outputIndex)
|
||
delete(imageGenerations, key)
|
||
return completeImageGeneration(key, accumulator)
|
||
default:
|
||
return nil
|
||
}
|
||
}
|
||
errorFromEvent := func(event openAIResponsesStreamEvent) error {
|
||
if event.Error != nil && strings.TrimSpace(event.Error.Message) != "" {
|
||
return fmt.Errorf("openai responses stream error %s: %s", openAIStreamErrorDetails(event.Error.Type, event.Error.Code, event.RequestID), strings.TrimSpace(event.Error.Message))
|
||
}
|
||
if event.Response != nil && event.Response.Error != nil && strings.TrimSpace(event.Response.Error.Message) != "" {
|
||
return fmt.Errorf("openai responses stream error %s: %s", openAIStreamErrorDetails(event.Response.Error.Type, event.Response.Error.Code, event.RequestID), strings.TrimSpace(event.Response.Error.Message))
|
||
}
|
||
return fmt.Errorf("openai responses stream failed")
|
||
}
|
||
|
||
scanner := bufio.NewScanner(resp.Body)
|
||
scanner.Buffer(make([]byte, 0, 64*1024), openAIStreamMaxTokenSize)
|
||
for scanner.Scan() {
|
||
rawLine := scanner.Text()
|
||
_, _ = appendLLMResponseArtifact(req, redactOpenAIStreamArtifactLine(rawLine)+"\n")
|
||
line := strings.TrimSpace(rawLine)
|
||
if line == "" || !strings.HasPrefix(line, "data:") {
|
||
continue
|
||
}
|
||
if firstEventAt.IsZero() {
|
||
firstEventAt = time.Now().UTC()
|
||
}
|
||
payloadLine := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
|
||
if payloadLine == "[DONE]" {
|
||
if err := flushTaggedContentTail(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return fail(err)
|
||
}
|
||
for key, accumulator := range tools {
|
||
if err := completeTool(key, accumulator); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
for key, accumulator := range imageGenerations {
|
||
if err := completeImageGeneration(key, accumulator); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
if err := flushTurnFinished(); err != nil {
|
||
return fail(err)
|
||
}
|
||
break
|
||
}
|
||
|
||
var event openAIResponsesStreamEvent
|
||
if err := json.Unmarshal([]byte(payloadLine), &event); err != nil {
|
||
return fail(err)
|
||
}
|
||
if event.Response != nil {
|
||
if strings.TrimSpace(event.Response.Model) != "" {
|
||
currentModel = strings.TrimSpace(event.Response.Model)
|
||
}
|
||
applyUsage(event.Response.Usage)
|
||
}
|
||
|
||
switch strings.TrimSpace(event.Type) {
|
||
case "response.output_text.delta":
|
||
if err := emitTaggedContentParts(thinkParser.Consume(event.Delta)); err != nil {
|
||
return fail(err)
|
||
}
|
||
case "response.output_item.added":
|
||
if event.Item != nil {
|
||
if err := applyOutputItem(*event.Item, event.OutputIndex, false); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
case "response.function_call_arguments.delta":
|
||
key := toolKey(event.ItemID, event.OutputIndex)
|
||
accumulator, ok := tools[key]
|
||
if !ok {
|
||
accumulator = &openAIToolAccumulator{}
|
||
tools[key] = accumulator
|
||
}
|
||
if event.Delta != "" {
|
||
_, _ = accumulator.Args.WriteString(event.Delta)
|
||
streamIdle.MarkEffectiveContent()
|
||
if err := emitOpenAIToolProgress(sink, currentModel, accumulator, event.Delta); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
case "response.function_call_arguments.done":
|
||
key := toolKey(event.ItemID, event.OutputIndex)
|
||
accumulator, ok := tools[key]
|
||
if !ok {
|
||
accumulator = &openAIToolAccumulator{}
|
||
tools[key] = accumulator
|
||
}
|
||
if event.Arguments != "" && accumulator.Args.Len() == 0 {
|
||
_, _ = accumulator.Args.WriteString(event.Arguments)
|
||
streamIdle.MarkEffectiveContent()
|
||
if err := emitOpenAIToolProgress(sink, currentModel, accumulator, event.Arguments); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
case "response.image_generation_call.partial_image":
|
||
key := toolKey(event.ItemID, event.OutputIndex)
|
||
accumulator, ok := imageGenerations[key]
|
||
if !ok {
|
||
accumulator = &openAIImageGenerationAccumulator{}
|
||
imageGenerations[key] = accumulator
|
||
}
|
||
if itemID := strings.TrimSpace(event.ItemID); itemID != "" {
|
||
accumulator.ProviderItemID = itemID
|
||
accumulator.CallID = namespaceToolCallID(req.ModelCallID, itemID)
|
||
}
|
||
if strings.TrimSpace(accumulator.CallID) == "" {
|
||
accumulator.CallID = namespaceToolCallID(req.ModelCallID, key)
|
||
}
|
||
if err := emitImageGenerationStarted(accumulator); err != nil {
|
||
return fail(err)
|
||
}
|
||
if imageData := strings.TrimSpace(event.PartialImageB64); imageData != "" {
|
||
accumulator.ImageData = imageData
|
||
streamIdle.MarkEffectiveContent()
|
||
}
|
||
if outputFormat := strings.TrimSpace(event.OutputFormat); outputFormat != "" {
|
||
accumulator.OutputFormat = outputFormat
|
||
}
|
||
case "response.output_item.done":
|
||
if event.Item != nil {
|
||
if err := applyOutputItem(*event.Item, event.OutputIndex, true); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
case "response.reasoning_summary_text.delta", "response.reasoning_text.delta":
|
||
if err := emitThinkingDelta(event.Delta); err != nil {
|
||
return fail(err)
|
||
}
|
||
case "response.completed", "response.incomplete":
|
||
if event.Response != nil && !emittedText {
|
||
if strings.TrimSpace(event.Response.OutputText) != "" {
|
||
if err := emitTaggedContentParts(thinkParser.Consume(event.Response.OutputText)); err != nil {
|
||
return fail(err)
|
||
}
|
||
} else {
|
||
for _, item := range event.Response.Output {
|
||
for _, content := range item.Content {
|
||
if strings.TrimSpace(content.Type) != "output_text" && strings.TrimSpace(content.Type) != "text" {
|
||
continue
|
||
}
|
||
if err := emitTaggedContentParts(thinkParser.Consume(content.Text)); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if err := flushTaggedContentTail(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if event.Response != nil {
|
||
for index, item := range event.Response.Output {
|
||
if err := applyOutputItem(item, index, true); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
finishReason = strings.TrimSpace(event.Response.Status)
|
||
if event.Response.IncompleteDetails != nil && strings.TrimSpace(event.Response.IncompleteDetails.Reason) != "" {
|
||
finishReason = strings.TrimSpace(event.Response.IncompleteDetails.Reason)
|
||
}
|
||
}
|
||
turnFinishedPending = true
|
||
case "response.failed", "error":
|
||
return fail(errorFromEvent(event))
|
||
}
|
||
}
|
||
for key, accumulator := range tools {
|
||
if err := completeTool(key, accumulator); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
for key, accumulator := range imageGenerations {
|
||
if err := completeImageGeneration(key, accumulator); err != nil {
|
||
return fail(err)
|
||
}
|
||
}
|
||
if err := scanner.Err(); err != nil {
|
||
if idleErr := streamIdle.Err(); idleErr != nil {
|
||
return fail(idleErr)
|
||
}
|
||
return fail(err)
|
||
}
|
||
if err := flushTaggedContentTail(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushThinkingCompleted(); err != nil {
|
||
return fail(err)
|
||
}
|
||
if err := flushTurnFinished(); err != nil {
|
||
return fail(err)
|
||
}
|
||
finishedAt = time.Now().UTC()
|
||
recordLLMSummaryArtifact(req, buildLLMSummaryPayload(req, "openai", currentModel, startedAt, firstEventAt, finishedAt, effectiveFinishReason(), inputTokens, outputTokens, cacheReadTokens, cacheWriteTokens, nil))
|
||
return nil
|
||
}
|
||
|
||
func trailingTagPrefixLength(text string, tag string) int {
|
||
maxLen := len(text)
|
||
if len(tag)-1 < maxLen {
|
||
maxLen = len(tag) - 1
|
||
}
|
||
for size := maxLen; size > 0; size-- {
|
||
if strings.HasSuffix(text, tag[:size]) {
|
||
return size
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func maxInt64(value int64, floor int64) int64 {
|
||
if value < floor {
|
||
return floor
|
||
}
|
||
return value
|
||
}
|
||
|
||
func cloneRawJSON(raw json.RawMessage) json.RawMessage {
|
||
if len(raw) == 0 {
|
||
return nil
|
||
}
|
||
return append(json.RawMessage(nil), raw...)
|
||
}
|
||
|
||
func redactOpenAIStreamArtifactLine(rawLine string) string {
|
||
line := strings.TrimSpace(rawLine)
|
||
if !strings.HasPrefix(line, "data:") {
|
||
return rawLine
|
||
}
|
||
payloadLine := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
|
||
if !strings.Contains(payloadLine, "partial_image_b64") && !strings.Contains(payloadLine, "image_data") && !strings.Contains(payloadLine, "imageData") {
|
||
return rawLine
|
||
}
|
||
var payload any
|
||
if err := json.Unmarshal([]byte(payloadLine), &payload); err != nil {
|
||
return rawLine
|
||
}
|
||
if !redactOpenAIImagePayloadFields(payload) {
|
||
return rawLine
|
||
}
|
||
encoded, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return rawLine
|
||
}
|
||
return "data: " + string(encoded)
|
||
}
|
||
|
||
func redactOpenAIImagePayloadFields(value any) bool {
|
||
changed := false
|
||
switch item := value.(type) {
|
||
case map[string]any:
|
||
for key, child := range item {
|
||
if text, ok := child.(string); ok {
|
||
switch key {
|
||
case "partial_image_b64", "image_data", "imageData":
|
||
item[key] = fmt.Sprintf("[base64 image data omitted from debug log; bytes=%d]", len(strings.TrimSpace(text)))
|
||
changed = true
|
||
continue
|
||
}
|
||
}
|
||
if redactOpenAIImagePayloadFields(child) {
|
||
changed = true
|
||
}
|
||
}
|
||
case []any:
|
||
for _, child := range item {
|
||
if redactOpenAIImagePayloadFields(child) {
|
||
changed = true
|
||
}
|
||
}
|
||
}
|
||
return changed
|
||
}
|
||
|
||
func emitOpenAIToolProgress(
|
||
sink func(ModelEvent) error,
|
||
model string,
|
||
accumulator *openAIToolAccumulator,
|
||
argsTextDelta string,
|
||
) error {
|
||
if accumulator == nil {
|
||
return nil
|
||
}
|
||
toolName := strings.TrimSpace(accumulator.Name)
|
||
if toolName == "CreatePlan" {
|
||
return emitCreatePlanToolProgress(
|
||
sink,
|
||
"openai",
|
||
model,
|
||
accumulator.CallID,
|
||
accumulator.Args.String(),
|
||
argsTextDelta,
|
||
&accumulator.LastCreatePlanSnapshot,
|
||
)
|
||
}
|
||
if toolName != "Write" && toolName != "PatchEdit" {
|
||
return nil
|
||
}
|
||
|
||
rawArgs := accumulator.Args.String()
|
||
path, pathFound, pathComplete := extractJSONStringFieldPrefix(rawArgs, "path")
|
||
if !pathFound {
|
||
path, pathFound, pathComplete = extractJSONStringFieldPrefix(rawArgs, "file_path")
|
||
}
|
||
if pathFound && pathComplete {
|
||
trimmedPath := strings.TrimSpace(path)
|
||
if trimmedPath != "" {
|
||
pathChanged := trimmedPath != accumulator.LastEmittedPath
|
||
accumulator.LastEmittedPath = trimmedPath
|
||
if toolName == "PatchEdit" && pathChanged {
|
||
if err := sink(ModelEvent{
|
||
Kind: ModelEventKindPartialToolCall,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: model,
|
||
ToolCallID: strings.TrimSpace(accumulator.CallID),
|
||
ToolCall: &agentv1.ToolCall{
|
||
Tool: &agentv1.ToolCall_EditToolCall{
|
||
EditToolCall: &agentv1.EditToolCall{
|
||
Args: &agentv1.EditArgs{Path: trimmedPath},
|
||
},
|
||
},
|
||
},
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if toolName == "Write" && pathChanged {
|
||
if err := sink(ModelEvent{
|
||
Kind: ModelEventKindPartialToolCall,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: model,
|
||
ToolCallID: strings.TrimSpace(accumulator.CallID),
|
||
ArgsTextDelta: argsTextDelta,
|
||
ToolCall: &agentv1.ToolCall{
|
||
Tool: &agentv1.ToolCall_EditToolCall{
|
||
EditToolCall: &agentv1.EditToolCall{
|
||
Args: &agentv1.EditArgs{Path: trimmedPath},
|
||
},
|
||
},
|
||
},
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
}
|
||
streamContent, streamFound := extractToolStreamContentPrefix(rawArgs, toolName)
|
||
if !streamFound {
|
||
return nil
|
||
}
|
||
delta := suffixAfterCommonPrefix(accumulator.LastStreamContent, streamContent)
|
||
if delta == "" {
|
||
return nil
|
||
}
|
||
accumulator.LastStreamContent = streamContent
|
||
return sink(ModelEvent{
|
||
Kind: ModelEventKindToolCallDelta,
|
||
OccurredAt: time.Now().UTC(),
|
||
Provider: "openai",
|
||
Model: model,
|
||
ToolCallID: strings.TrimSpace(accumulator.CallID),
|
||
ToolCallDelta: &agentv1.ToolCallDelta{
|
||
Delta: &agentv1.ToolCallDelta_EditToolCallDelta{
|
||
EditToolCallDelta: &agentv1.EditToolCallDelta{
|
||
StreamContentDelta: delta,
|
||
},
|
||
},
|
||
},
|
||
})
|
||
}
|
||
|
||
func normalizeOpenAIProviderMessages(messages []Message, thinkingEnabled bool) ([]map[string]any, error) {
|
||
if len(messages) == 0 {
|
||
return nil, nil
|
||
}
|
||
items := make([]map[string]any, 0, len(messages))
|
||
for _, message := range messages {
|
||
content, err := openAIContentValue(message)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
item := map[string]any{
|
||
"role": strings.TrimSpace(message.Role),
|
||
"content": content,
|
||
}
|
||
// 开启 thinking 时，tool_calls 对应的 assistant message 也要显式携带空 reasoning_content。
|
||
if shouldIncludeOpenAIReasoningContent(message, thinkingEnabled) {
|
||
item["reasoning_content"] = message.ReasoningContent
|
||
}
|
||
if len(message.ToolCalls) > 0 {
|
||
item["tool_calls"] = normalizeToolCallDescriptors(message.ToolCalls)
|
||
}
|
||
if strings.TrimSpace(message.ToolCallID) != "" {
|
||
item["tool_call_id"] = providerToolCallID(message.ToolCallID)
|
||
}
|
||
if strings.TrimSpace(message.Name) != "" {
|
||
item["name"] = strings.TrimSpace(message.Name)
|
||
}
|
||
items = append(items, item)
|
||
}
|
||
return items, nil
|
||
}
|
||
|
||
func shouldSendOpenAIMaxOutputTokens(modelID string) bool {
|
||
return !strings.Contains(strings.ToLower(strings.TrimSpace(modelID)), "gpt")
|
||
}
|
||
|
||
func shouldIncludeOpenAIReasoningContent(message Message, thinkingEnabled bool) bool {
|
||
if strings.TrimSpace(message.ReasoningContent) != "" {
|
||
return true
|
||
}
|
||
if !thinkingEnabled {
|
||
return false
|
||
}
|
||
if strings.TrimSpace(message.Role) != "assistant" {
|
||
return false
|
||
}
|
||
return len(message.ToolCalls) > 0
|
||
}
|
||
|
||
func applyOpenAIThinkingDisable(body map[string]any, req StreamRequest, baseURL string, modelID string, endpoint string) {
|
||
if len(body) == 0 || normalizeRuntimeThinkingEffort(req.ThinkingEffort) != "disabled" {
|
||
return
|
||
}
|
||
switch openAIThinkingDisableKind(baseURL, modelID, endpoint) {
|
||
case "thinking_type":
|
||
body["thinking"] = map[string]any{"type": "disabled"}
|
||
delete(body, "reasoning_effort")
|
||
setRequestKnob(req, "thinking_disabled_provider_param", "thinking.type")
|
||
case "enable_thinking":
|
||
body["enable_thinking"] = false
|
||
delete(body, "reasoning_effort")
|
||
setRequestKnob(req, "thinking_disabled_provider_param", "enable_thinking")
|
||
case "reasoning_none":
|
||
if modelchannel.OpenAIEndpointShape(endpoint) == "responses" {
|
||
body["reasoning"] = map[string]any{"effort": "none"}
|
||
} else {
|
||
body["reasoning_effort"] = "none"
|
||
}
|
||
setRequestKnob(req, "thinking_disabled_provider_param", "reasoning.effort")
|
||
}
|
||
}
|
||
|
||
func openAIThinkingDisableKind(baseURL string, modelID string, endpoint string) string {
|
||
base := strings.ToLower(strings.TrimSpace(baseURL))
|
||
model := strings.ToLower(strings.TrimSpace(modelID))
|
||
switch {
|
||
case strings.Contains(base, "dashscope") ||
|
||
strings.Contains(base, "qwen") ||
|
||
strings.Contains(base, "aliyun") ||
|
||
strings.Contains(model, "qwen"):
|
||
return "enable_thinking"
|
||
case strings.Contains(base, "deepseek") ||
|
||
strings.Contains(base, "bigmodel") ||
|
||
strings.Contains(base, "z.ai") ||
|
||
strings.Contains(base, "zhipu") ||
|
||
strings.Contains(base, "xiaomimimo") ||
|
||
strings.Contains(base, "mimo") ||
|
||
strings.Contains(base, "minimax") ||
|
||
strings.Contains(model, "deepseek") ||
|
||
strings.Contains(model, "glm") ||
|
||
strings.Contains(model, "zai") ||
|
||
strings.Contains(model, "zhipu") ||
|
||
strings.Contains(model, "mimo") ||
|
||
strings.Contains(model, "minimax"):
|
||
return "thinking_type"
|
||
case openAIModelSupportsReasoningNone(model):
|
||
return "reasoning_none"
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func openAIModelSupportsReasoningNone(model string) bool {
|
||
model = strings.ToLower(strings.TrimSpace(model))
|
||
if strings.HasPrefix(model, "gpt-6") {
|
||
return true
|
||
}
|
||
if strings.Contains(model, "gpt-5.1") {
|
||
return true
|
||
}
|
||
if !strings.HasPrefix(model, "gpt-5.") {
|
||
return false
|
||
}
|
||
minorText := strings.TrimPrefix(model, "gpt-5.")
|
||
minorEnd := 0
|
||
for minorEnd < len(minorText) && minorText[minorEnd] >= '0' && minorText[minorEnd] <= '9' {
|
||
minorEnd++
|
||
}
|
||
if minorEnd == 0 {
|
||
return false
|
||
}
|
||
minor, err := strconv.Atoi(minorText[:minorEnd])
|
||
return err == nil && minor >= 1
|
||
}
|
||
|
||
func setRequestKnob(req StreamRequest, key string, value any) {
|
||
if req.RequestKnobs == nil {
|
||
return
|
||
}
|
||
req.RequestKnobs[key] = value
|
||
}
|
||
|
||
func normalizeOpenAIResponsesInput(messages []Message) (string, []map[string]any, error) {
|
||
if len(messages) == 0 {
|
||
return "", nil, nil
|
||
}
|
||
instructionParts := make([]string, 0, 2)
|
||
items := make([]map[string]any, 0, len(messages))
|
||
responsesCallIDs := make(map[string]string)
|
||
emittedCallIDs := make(map[string]struct{})
|
||
activeAssistantReasoningKey := ""
|
||
for _, message := range messages {
|
||
role := strings.TrimSpace(message.Role)
|
||
if role == "system" {
|
||
if text := openAIResponsesMessageText(message); strings.TrimSpace(text) != "" {
|
||
instructionParts = append(instructionParts, strings.TrimSpace(text))
|
||
}
|
||
activeAssistantReasoningKey = ""
|
||
continue
|
||
}
|
||
if role == "tool" && strings.TrimSpace(message.ToolCallID) != "" {
|
||
callID := openAIResponsesToolMessageCallID(message, responsesCallIDs)
|
||
if strings.TrimSpace(callID) != "" {
|
||
if _, ok := emittedCallIDs[callID]; !ok {
|
||
// 历史损坏时可能出现没有配对 function_call 的工具结果
|
||
// （例如旧版本回放逻辑剥离了 assistant 调用但保留了结果）。
|
||
// Responses API 会直接拒绝这种 input，这里补一个占位 function_call
|
||
// 让旧会话可以继续；无法补齐时丢弃该结果。
|
||
if name := strings.TrimSpace(message.Name); name != "" {
|
||
items = append(items, map[string]any{
|
||
"type": "function_call",
|
||
"call_id": callID,
|
||
"name": name,
|
||
"arguments": "{}",
|
||
"status": "completed",
|
||
})
|
||
emittedCallIDs[callID] = struct{}{}
|
||
} else {
|
||
continue
|
||
}
|
||
}
|
||
}
|
||
var output any = openAIResponsesMessageText(message)
|
||
if hasImageContentParts(message.ContentParts) {
|
||
content, err := openAIResponsesMessageContent(message, false)
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
output = content
|
||
}
|
||
items = append(items, map[string]any{
|
||
"type": "function_call_output",
|
||
"call_id": callID,
|
||
"output": output,
|
||
})
|
||
activeAssistantReasoningKey = ""
|
||
continue
|
||
}
|
||
if role != "assistant" {
|
||
activeAssistantReasoningKey = ""
|
||
}
|
||
if shouldIncludeOpenAIResponsesReasoningItem(message) {
|
||
reasoningKey := openAIResponsesReasoningReplayKey(message)
|
||
if reasoningKey != activeAssistantReasoningKey {
|
||
items = append(items, openAIResponsesReasoningItem(message))
|
||
activeAssistantReasoningKey = reasoningKey
|
||
}
|
||
}
|
||
if strings.TrimSpace(message.Content) != "" || len(message.ContentParts) > 0 {
|
||
content, err := openAIResponsesMessageContent(message, role == "assistant")
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
if len(content) > 0 {
|
||
items = append(items, map[string]any{
|
||
"role": openAIResponsesMessageRole(role),
|
||
"content": content,
|
||
})
|
||
}
|
||
}
|
||
if role == "assistant" && len(message.ToolCalls) > 0 {
|
||
for _, toolCall := range message.ToolCalls {
|
||
name := strings.TrimSpace(toolCall.Function.Name)
|
||
if name == "" {
|
||
continue
|
||
}
|
||
callID := openAIResponsesToolCallCallID(toolCall)
|
||
if strings.TrimSpace(callID) == "" {
|
||
callID = openAIResponsesProviderCallID(name)
|
||
}
|
||
if internalID := strings.TrimSpace(toolCall.ID); internalID != "" && strings.TrimSpace(callID) != "" {
|
||
responsesCallIDs[internalID] = strings.TrimSpace(callID)
|
||
}
|
||
toolItem := map[string]any{
|
||
"type": "function_call",
|
||
"call_id": callID,
|
||
"name": name,
|
||
"arguments": toolCall.Function.Arguments,
|
||
}
|
||
if itemID := strings.TrimSpace(toolCall.OpenAIResponsesID); itemID != "" {
|
||
toolItem["id"] = itemID
|
||
}
|
||
if status := strings.TrimSpace(toolCall.OpenAIResponsesStatus); status != "" {
|
||
toolItem["status"] = status
|
||
} else {
|
||
toolItem["status"] = "completed"
|
||
}
|
||
items = append(items, toolItem)
|
||
if strings.TrimSpace(callID) != "" {
|
||
emittedCallIDs[strings.TrimSpace(callID)] = struct{}{}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return strings.Join(instructionParts, "\n\n"), items, nil
|
||
}
|
||
|
||
func openAIResponsesReasoningReplayKey(message Message) string {
|
||
return strings.Join([]string{
|
||
strings.TrimSpace(message.ReasoningSignature),
|
||
strings.TrimSpace(message.OpenAIResponsesReasoningID),
|
||
strings.TrimSpace(message.OpenAIResponsesReasoningStatus),
|
||
string(message.OpenAIResponsesReasoningSummary),
|
||
}, "\x00")
|
||
}
|
||
|
||
func openAIResponsesReasoningItem(message Message) map[string]any {
|
||
reasoningItem := map[string]any{
|
||
"type": "reasoning",
|
||
"encrypted_content": strings.TrimSpace(message.ReasoningSignature),
|
||
}
|
||
if reasoningID := strings.TrimSpace(message.OpenAIResponsesReasoningID); reasoningID != "" {
|
||
reasoningItem["id"] = reasoningID
|
||
}
|
||
if reasoningStatus := strings.TrimSpace(message.OpenAIResponsesReasoningStatus); reasoningStatus != "" {
|
||
reasoningItem["status"] = reasoningStatus
|
||
}
|
||
if len(message.OpenAIResponsesReasoningSummary) > 0 {
|
||
reasoningItem["summary"] = json.RawMessage(append([]byte(nil), message.OpenAIResponsesReasoningSummary...))
|
||
} else {
|
||
reasoningItem["summary"] = []any{}
|
||
}
|
||
return reasoningItem
|
||
}
|
||
|
||
func shouldIncludeOpenAIResponsesReasoningItem(message Message) bool {
|
||
if strings.TrimSpace(message.Role) != "assistant" || strings.TrimSpace(message.ReasoningSignature) == "" {
|
||
return false
|
||
}
|
||
return strings.TrimSpace(message.ReasoningSignatureSource) == ReasoningSignatureSourceOpenAIResponses
|
||
}
|
||
|
||
func openAIResponsesToolMessageCallID(message Message, responsesCallIDs map[string]string) string {
|
||
internalID := strings.TrimSpace(message.ToolCallID)
|
||
if internalID == "" {
|
||
return ""
|
||
}
|
||
if callID := strings.TrimSpace(responsesCallIDs[internalID]); callID != "" {
|
||
return callID
|
||
}
|
||
return openAIResponsesProviderCallID(internalID)
|
||
}
|
||
|
||
func openAIResponsesToolCallCallID(toolCall ToolCallDescriptor) string {
|
||
if callID := strings.TrimSpace(toolCall.OpenAIResponsesCallID); callID != "" {
|
||
return callID
|
||
}
|
||
return openAIResponsesProviderCallID(toolCall.ID)
|
||
}
|
||
|
||
func openAIResponsesProviderCallID(toolCallID string) string {
|
||
trimmed := strings.TrimSpace(toolCallID)
|
||
if trimmed == "" {
|
||
return ""
|
||
}
|
||
if _, raw, ok := splitLegacyToolCallID(trimmed); ok {
|
||
return raw
|
||
}
|
||
if strings.HasPrefix(trimmed, "tc_") {
|
||
parts := strings.SplitN(trimmed, "_", 3)
|
||
if len(parts) == 3 && strings.TrimSpace(parts[2]) != "" {
|
||
return strings.TrimSpace(parts[2])
|
||
}
|
||
}
|
||
return providerToolCallID(trimmed)
|
||
}
|
||
|
||
func openAIResponsesMessageRole(role string) string {
|
||
switch strings.TrimSpace(role) {
|
||
case "assistant":
|
||
return "assistant"
|
||
default:
|
||
return "user"
|
||
}
|
||
}
|
||
|
||
func openAIResponsesMessageText(message Message) string {
|
||
if strings.TrimSpace(message.Content) != "" {
|
||
return message.Content
|
||
}
|
||
if len(message.ContentParts) > 0 {
|
||
return collapseTextContentParts(message.ContentParts)
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func openAIResponsesMessageContent(message Message, assistant bool) ([]map[string]any, error) {
|
||
textType := "input_text"
|
||
if assistant {
|
||
textType = "output_text"
|
||
}
|
||
if !hasImageContentParts(message.ContentParts) {
|
||
text := openAIResponsesMessageText(message)
|
||
if text == "" {
|
||
return nil, nil
|
||
}
|
||
return []map[string]any{{
|
||
"type": textType,
|
||
"text": text,
|
||
}}, nil
|
||
}
|
||
parts := make([]map[string]any, 0, len(message.ContentParts)+1)
|
||
if len(message.ContentParts) == 0 && strings.TrimSpace(message.Content) != "" {
|
||
parts = append(parts, map[string]any{
|
||
"type": textType,
|
||
"text": message.Content,
|
||
})
|
||
}
|
||
for _, part := range message.ContentParts {
|
||
switch normalizeContentPartType(part.Type) {
|
||
case contentPartTypeText:
|
||
if part.Text == "" {
|
||
continue
|
||
}
|
||
parts = append(parts, map[string]any{
|
||
"type": textType,
|
||
"text": part.Text,
|
||
})
|
||
case contentPartTypeImage:
|
||
dataURL, err := imageContentDataURL(part.Image)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
parts = append(parts, map[string]any{
|
||
"type": "input_image",
|
||
"image_url": dataURL,
|
||
})
|
||
default:
|
||
return nil, fmt.Errorf("unsupported openai responses content part type: %s", strings.TrimSpace(part.Type))
|
||
}
|
||
}
|
||
if len(parts) == 0 {
|
||
return nil, nil
|
||
}
|
||
return parts, nil
|
||
}
|
||
|
||
func normalizeOpenAIResponsesTools(items []json.RawMessage) ([]map[string]any, error) {
|
||
if len(items) == 0 {
|
||
return nil, nil
|
||
}
|
||
tools := make([]map[string]any, 0, len(items))
|
||
for _, item := range items {
|
||
var raw map[string]any
|
||
if err := json.Unmarshal(item, &raw); err != nil {
|
||
return nil, fmt.Errorf("decode openai responses tool descriptor failed: %w", err)
|
||
}
|
||
source := raw
|
||
if functionShape, ok := raw["function"].(map[string]any); ok {
|
||
source = functionShape
|
||
}
|
||
name := strings.TrimSpace(asStringMapValue(source, "name"))
|
||
if name == "" {
|
||
return nil, fmt.Errorf("openai responses tool descriptor name is required")
|
||
}
|
||
tool := map[string]any{
|
||
"type": "function",
|
||
"name": name,
|
||
}
|
||
if description := strings.TrimSpace(asStringMapValue(source, "description")); description != "" {
|
||
tool["description"] = description
|
||
}
|
||
if parameters, ok := source["parameters"]; ok && parameters != nil {
|
||
tool["parameters"] = parameters
|
||
} else {
|
||
tool["parameters"] = map[string]any{"type": "object", "properties": map[string]any{}}
|
||
}
|
||
if strict, ok := source["strict"]; ok {
|
||
tool["strict"] = strict
|
||
} else if strict, ok := raw["strict"]; ok {
|
||
tool["strict"] = strict
|
||
}
|
||
tools = append(tools, tool)
|
||
}
|
||
return tools, nil
|
||
}
|
||
|
||
func asStringMapValue(source map[string]any, key string) string {
|
||
if len(source) == 0 {
|
||
return ""
|
||
}
|
||
switch value := source[key].(type) {
|
||
case string:
|
||
return value
|
||
case fmt.Stringer:
|
||
return value.String()
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func openAIStreamErrorDetails(errorType string, code string, requestID string) string {
|
||
parts := make([]string, 0, 3)
|
||
if value := strings.TrimSpace(errorType); value != "" {
|
||
parts = append(parts, "type="+value)
|
||
}
|
||
if value := strings.TrimSpace(code); value != "" {
|
||
parts = append(parts, "code="+value)
|
||
}
|
||
if value := strings.TrimSpace(requestID); value != "" {
|
||
parts = append(parts, "request_id="+value)
|
||
}
|
||
if len(parts) == 0 {
|
||
return "provider_error"
|
||
}
|
||
return strings.Join(parts, " ")
|
||
}
|