Skip to content

fix(openresponses): support Codex WebSocket warm-up #11608

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aim9sour wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix(openresponses): support Codex WebSocket warm-up #11608

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 87 additions & 30 deletions core/http/endpoints/openresponses/responses.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,33 +58,17 @@ func ResponsesEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, eval
shouldStore = false
}

// Handle previous_response_id if provided
var previousResponse *schema.ORResponseResource
// Handle previous_response_id if provided.
var messages []schema.Message
if input.PreviousResponseID != "" {
stored, err := store.Get(input.PreviousResponseID)
previousMessages, err := resolvePreviousResponseMessages(store, input.PreviousResponseID, cfg)
if err != nil {
return sendOpenResponsesError(c, 404, "not_found", fmt.Sprintf("previous response not found: %s", input.PreviousResponseID), "previous_response_id")
}
previousResponse = stored.Response

// Also convert previous response input to messages
previousInputMessages, err := convertORInputToMessages(stored.Request.Input, cfg)
if err != nil {
return sendOpenResponsesError(c, 400, "invalid_request", fmt.Sprintf("failed to convert previous input: %v", err), "")
}

// Convert previous response output items to messages
previousOutputMessages, err := convertOROutputItemsToMessages(previousResponse.Output)
if err != nil {
return sendOpenResponsesError(c, 400, "invalid_request", fmt.Sprintf("failed to convert previous response: %v", err), "")
if notFound, ok := err.(*previousResponseNotFoundError); ok {
return sendOpenResponsesError(c, 404, "not_found", notFound.Error(), "previous_response_id")
}
return sendOpenResponsesError(c, 400, "invalid_request", err.Error(), "")
}

// Concatenate: previous_input + previous_output + new_input
// Start with previous input messages
messages = previousInputMessages
// Add previous output as assistant messages
messages = append(messages, previousOutputMessages...)
messages = previousMessages
}

// Convert Open Responses input to internal Messages
Expand Down Expand Up @@ -266,7 +250,7 @@ func ResponsesEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, eval

if input.Stream {
// Background streaming processing (buffer events)
finalResponse, bgErr = handleBackgroundStream(bgCtx, store, responseID, createdAt, input, cfg, ml, cl, appConfig, predInput, openAIReq, funcs, shouldUseFn, mcpExecutor, evaluator)
finalResponse, bgErr = handleBackgroundStream(bgCtx, store, responseID, createdAt, input, cfg, ml, cl, appConfig, predInput, openAIReq, funcs, shouldUseFn, true, mcpExecutor, evaluator)
} else {
// Background non-streaming processing
finalResponse, bgErr = handleBackgroundNonStream(bgCtx, store, responseID, createdAt, input, cfg, ml, cl, appConfig, predInput, openAIReq, funcs, shouldUseFn, mcpExecutor, evaluator)
Expand Down Expand Up @@ -515,6 +499,76 @@ func extractReasoningContentFromORItem(item *schema.ORItemField) string {
return ""
}

type previousResponseNotFoundError struct {
ResponseID string
}

func (e *previousResponseNotFoundError) Error() string {
return fmt.Sprintf("previous response not found: %s", e.ResponseID)
}

// resolvePreviousResponseMessages reconstructs the complete stored conversation
// ending at responseID. Requests are stored as incremental deltas, so replaying
// only the immediately previous request loses older turns after the first chain.
func resolvePreviousResponseMessages(store *ResponseStore, responseID string, cfg *config.ModelConfig) ([]schema.Message, error) {
return resolvePreviousResponseMessagesFromStores([]*ResponseStore{store}, responseID, cfg)
}

// resolvePreviousResponseMessagesFromStores resolves each hop against the stores
// in priority order. WebSocket mode uses this to prefer connection-local
// store=false responses while still allowing references to globally stored ones.
func resolvePreviousResponseMessagesFromStores(stores []*ResponseStore, responseID string, cfg *config.ModelConfig) ([]schema.Message, error) {
type chainEntry struct {
id string
stored *StoredResponse
}

var chain []chainEntry
seen := make(map[string]struct{})
for currentID := responseID; currentID != ""; {
if _, exists := seen[currentID]; exists {
return nil, fmt.Errorf("previous_response_id cycle detected at %s", currentID)
}
seen[currentID] = struct{}{}

var stored *StoredResponse
for _, store := range stores {
if store == nil {
continue
}
candidate, err := store.Get(currentID)
if err == nil {
stored = candidate
break
}
}
if stored == nil {
return nil, &previousResponseNotFoundError{ResponseID: currentID}
}
if stored.Request == nil || stored.Response == nil {
return nil, fmt.Errorf("stored previous response %s is incomplete", currentID)
}
chain = append(chain, chainEntry{id: currentID, stored: stored})
currentID = stored.Request.PreviousResponseID
}

var messages []schema.Message
for i := len(chain) - 1; i >= 0; i-- {
entry := chain[i]
inputMessages, err := convertORInputToMessages(entry.stored.Request.Input, cfg)
if err != nil {
return nil, fmt.Errorf("failed to convert previous input for %s: %w", entry.id, err)
}
outputMessages, err := convertOROutputItemsToMessages(entry.stored.Response.Output)
if err != nil {
return nil, fmt.Errorf("failed to convert previous response %s: %w", entry.id, err)
}
messages = append(messages, inputMessages...)
messages = append(messages, outputMessages...)
}
return messages, nil
}

// convertOROutputItemsToMessages converts Open Responses output items to internal Messages.
// Contiguous assistant items (message, reasoning, function_call) are merged into a single message.
func convertOROutputItemsToMessages(outputItems []schema.ORItemField) ([]schema.Message, error) {
Expand Down Expand Up @@ -1013,7 +1067,7 @@ func handleBackgroundNonStream(ctx context.Context, store *ResponseStore, respon
}

// handleBackgroundStream handles background streaming responses with event buffering
func handleBackgroundStream(ctx context.Context, store *ResponseStore, responseID string, createdAt int64, input *schema.OpenResponsesRequest, cfg *config.ModelConfig, ml *model.ModelLoader, cl *config.ModelConfigLoader, appConfig *config.ApplicationConfig, predInput string, openAIReq *schema.OpenAIRequest, funcs functions.Functions, shouldUseFn bool, mcpExecutor mcpTools.ToolExecutor, evaluator *templates.Evaluator) (*schema.ORResponseResource, error) {
func handleBackgroundStream(ctx context.Context, store *ResponseStore, responseID string, createdAt int64, input *schema.OpenResponsesRequest, cfg *config.ModelConfig, ml *model.ModelLoader, cl *config.ModelConfigLoader, appConfig *config.ApplicationConfig, predInput string, openAIReq *schema.OpenAIRequest, funcs functions.Functions, shouldUseFn bool, shouldStore bool, mcpExecutor mcpTools.ToolExecutor, evaluator *templates.Evaluator) (*schema.ORResponseResource, error) {
// Populate openAIReq fields for ComputeChoices
openAIReq.Tools = convertORToolsToOpenAIFormat(input.Tools)
openAIReq.ToolsChoice = input.ToolChoice
Expand All @@ -1026,7 +1080,7 @@ func handleBackgroundStream(ctx context.Context, store *ResponseStore, responseI
sequenceNumber := 0

// Emit response.created
responseCreated := buildORResponse(responseID, createdAt, nil, schema.ORStatusInProgress, input, []schema.ORItemField{}, nil, true)
responseCreated := buildORResponse(responseID, createdAt, nil, schema.ORStatusInProgress, input, []schema.ORItemField{}, nil, shouldStore)
bufferEvent(store, responseID, &schema.ORStreamEvent{
Type: "response.created",
SequenceNumber: sequenceNumber,
Expand Down Expand Up @@ -1298,7 +1352,7 @@ func handleBackgroundStream(ctx context.Context, store *ResponseStore, responseI
InputTokens: lastTokenUsage.Prompt,
OutputTokens: lastTokenUsage.Completion,
TotalTokens: lastTokenUsage.Prompt + lastTokenUsage.Completion,
}, true)
}, shouldStore)

// Emit response.completed
bufferEvent(store, responseID, &schema.ORStreamEvent{
Expand Down Expand Up @@ -2955,12 +3009,15 @@ func sendOpenResponsesError(c echo.Context, statusCode int, errorType, message,
return c.JSON(statusCode, errorResp)
}

// convertORToolsToOpenAIFormat converts Open Responses tools to OpenAI format for the backend
// Open Responses format: { type, name, description, parameters }
// OpenAI format: { type, function: { name, description, parameters } }
// convertORToolsToOpenAIFormat converts only tools that have an equivalent in
// the OpenAI-compatible function-tool representation. Native Responses tools
// such as web_search and namespace must not be rewritten as functions.
func convertORToolsToOpenAIFormat(orTools []schema.ORFunctionTool) []functions.Tool {
result := make([]functions.Tool, 0, len(orTools))
for _, t := range orTools {
if t.Type != "function" {
continue
}
result = append(result, functions.Tool{
Type: "function",
Function: functions.Function{
Expand Down
76 changes: 76 additions & 0 deletions core/http/endpoints/openresponses/responses_convert_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package openresponses

import (
"github.com/mudler/LocalAI/core/config"
"github.com/mudler/LocalAI/core/schema"

. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
Expand Down Expand Up @@ -60,3 +61,78 @@ var _ = Describe("convertORInputToMessages", func() {
Expect(msgs).To(BeEmpty())
})
})

var _ = Describe("convertORToolsToOpenAIFormat", func() {
It("only converts Responses function tools", func() {
converted := convertORToolsToOpenAIFormat([]schema.ORFunctionTool{
{Type: "function", Name: "example_function", Parameters: map[string]any{"type": "object"}},
{Type: "web_search"},
{Type: "namespace", Name: "multi_agent_v1"},
})

Expect(converted).To(HaveLen(1))
Expect(converted[0].Type).To(Equal("function"))
Expect(converted[0].Function.Name).To(Equal("example_function"))
Expect(converted[0].Function.Parameters).To(Equal(map[string]any{"type": "object"}))
})
})

var _ = Describe("resolvePreviousResponseMessages", func() {
It("replays multi-hop response history from oldest to newest", func() {
store := NewResponseStore(0)
cfg := &config.ModelConfig{}
message := func(role, text string) schema.ORItemField {
return schema.ORItemField{
Type: "message",
Role: role,
Content: []schema.ORContentPart{{Type: "output_text", Text: text}},
}
}

store.Store("resp_0", &schema.OpenResponsesRequest{Input: "base"}, &schema.ORResponseResource{
ID: "resp_0", Output: []schema.ORItemField{message("assistant", "answer-0")},
})
store.Store("resp_1", &schema.OpenResponsesRequest{PreviousResponseID: "resp_0", Input: "question-1"}, &schema.ORResponseResource{
ID: "resp_1", Output: []schema.ORItemField{message("assistant", "answer-1")},
})
store.Store("resp_2", &schema.OpenResponsesRequest{PreviousResponseID: "resp_1", Input: "question-2"}, &schema.ORResponseResource{
ID: "resp_2", Output: []schema.ORItemField{message("assistant", "answer-2")},
})

msgs, err := resolvePreviousResponseMessages(store, "resp_2", cfg)
Expect(err).NotTo(HaveOccurred())
Expect(msgs).To(HaveLen(6))
Expect([]string{
msgs[0].StringContent, msgs[1].StringContent,
msgs[2].StringContent, msgs[3].StringContent,
msgs[4].StringContent, msgs[5].StringContent,
}).To(Equal([]string{"base", "answer-0", "question-1", "answer-1", "question-2", "answer-2"}))
})

It("resolves a chain across connection-local and global stores", func() {
connectionStore := NewResponseStore(0)
globalStore := NewResponseStore(0)
cfg := &config.ModelConfig{}
message := func(text string) schema.ORItemField {
return schema.ORItemField{
Type: "message",
Role: "assistant",
Content: []schema.ORContentPart{{Type: "output_text", Text: text}},
}
}

globalStore.Store("resp_global", &schema.OpenResponsesRequest{Input: "base"}, &schema.ORResponseResource{
ID: "resp_global", Output: []schema.ORItemField{message("answer-0")},
})
connectionStore.Store("resp_local", &schema.OpenResponsesRequest{PreviousResponseID: "resp_global", Input: "question-1"}, &schema.ORResponseResource{
ID: "resp_local", Output: []schema.ORItemField{message("answer-1")},
})

msgs, err := resolvePreviousResponseMessagesFromStores([]*ResponseStore{connectionStore, globalStore}, "resp_local", cfg)
Expect(err).NotTo(HaveOccurred())
Expect(msgs).To(HaveLen(4))
Expect([]string{msgs[0].StringContent, msgs[1].StringContent, msgs[2].StringContent, msgs[3].StringContent}).To(
Equal([]string{"base", "answer-0", "question-1", "answer-1"}),
)
})
})
Loading