// 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 = "" openAIThinkCloseTag = "" 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 的 标签拆回 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 := "" 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) 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) items = append(items, map[string]any{ "type": "function_call_output", "call_id": callID, "output": openAIResponsesMessageText(message), }) 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) } } } 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, " ") }