openai: support web search in Responses API (#17686)

This commit is contained in:
Parth Sareen
2026-08-12 11:51:54 -07:00
committed by GitHub
parent 4138e853d5
commit b6b1b258c3
14 changed files with 2657 additions and 250 deletions
+59
View File
@@ -2,6 +2,7 @@ package api
import (
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
@@ -388,6 +389,64 @@ func TestClientWebSearchExperimentalUsesLocalRoute(t *testing.T) {
}
}
func TestClientWebSearchExperimentalErrors(t *testing.T) {
tests := []struct {
name string
status int
body string
assertError func(*testing.T, error)
}{
{
name: "unauthorized retains sign in URL",
status: http.StatusUnauthorized,
body: `{"error":"unauthorized","signin_url":"https://ollama.com/signin/example"}`,
assertError: func(t *testing.T, err error) {
t.Helper()
var authErr AuthorizationError
if !errors.As(err, &authErr) {
t.Fatalf("error = %T, want AuthorizationError", err)
}
if authErr.StatusCode != http.StatusUnauthorized || authErr.SigninURL != "https://ollama.com/signin/example" {
t.Fatalf("authorization error = %#v", authErr)
}
},
},
{
name: "rate limit retains status",
status: http.StatusTooManyRequests,
body: `{"error":"rate limit exceeded"}`,
assertError: func(t *testing.T, err error) {
t.Helper()
var statusErr StatusError
if !errors.As(err, &statusErr) {
t.Fatalf("error = %T, want StatusError", err)
}
if statusErr.StatusCode != http.StatusTooManyRequests || statusErr.ErrorMessage != "rate limit exceeded" {
t.Fatalf("status error = %#v", statusErr)
}
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(tt.status)
_, _ = w.Write([]byte(tt.body))
}))
defer ts.Close()
client := NewClient(&url.URL{Scheme: "http", Host: ts.Listener.Addr().String()}, http.DefaultClient)
_, err := client.WebSearchExperimental(t.Context(), &WebSearchRequest{Query: "ollama"})
if err == nil {
t.Fatal("expected error")
}
tt.assertError(t, err)
})
}
}
func TestClientWebFetchExperimentalUsesLocalRoute(t *testing.T) {
var gotPath string
var gotMethod string
+12
View File
@@ -76,3 +76,15 @@ Then run:
```
codex --profile ollama-launch
```
## Web search
Codex web-search requests sent through the Ollama profile are executed by
Ollama for both local and cloud models. Sign in with `ollama signin` to use the
web-search service.
To disable web search for a Codex session:
```shell
codex --profile ollama-launch -c 'web_search="disabled"'
```
+1 -95
View File
@@ -15,7 +15,6 @@ import (
"github.com/ollama/ollama/anthropic"
"github.com/ollama/ollama/api"
"github.com/ollama/ollama/envconfig"
internalcloud "github.com/ollama/ollama/internal/cloud"
"github.com/ollama/ollama/internal/modelref"
"github.com/ollama/ollama/logutil"
@@ -115,8 +114,6 @@ type WebSearchAnthropicWriter struct {
streamNextIndex int
}
const maxWebSearchLoops = 3
type webSearchLoopResult struct {
response anthropic.MessagesResponse
loopErr *webSearchLoopError
@@ -485,20 +482,6 @@ func (w *WebSearchAnthropicWriter) combineServerAndFinalContent(serverContent []
}
}
func buildWebSearchAssistantMessage(response api.ChatResponse, webSearchCall api.ToolCall) api.Message {
assistantMsg := api.Message{
Role: "assistant",
ToolCalls: []api.ToolCall{webSearchCall},
}
if response.Message.Content != "" {
assistantMsg.Content = response.Message.Content
}
if response.Message.Thinking != "" {
assistantMsg.Thinking = response.Message.Thinking
}
return assistantMsg
}
func formatWebSearchResultsForToolMessage(results []anthropic.OllamaWebSearchResult) string {
var resultText strings.Builder
for _, r := range results {
@@ -511,25 +494,6 @@ func formatWebSearchResultsForToolMessage(results []anthropic.OllamaWebSearchRes
return resultText.String()
}
func findWebSearchToolCall(toolCalls []api.ToolCall) (api.ToolCall, bool, bool) {
var webSearchCall api.ToolCall
hasWebSearch := false
hasOtherTools := false
for _, toolCall := range toolCalls {
if toolCall.Function.Name == "web_search" {
if !hasWebSearch {
webSearchCall = toolCall
hasWebSearch = true
}
continue
}
hasOtherTools = true
}
return webSearchCall, hasWebSearch, hasOtherTools
}
func loopServerToolUseID(messageID string, loop int) string {
base := serverToolUseID(messageID)
if loop <= 1 {
@@ -539,53 +503,7 @@ func loopServerToolUseID(messageID string, loop int) string {
}
func (w *WebSearchAnthropicWriter) callFollowUpChat(ctx context.Context, messages []api.Message, tools api.Tools) (api.ChatResponse, error) {
streaming := false
followUp := api.ChatRequest{
Model: w.chatReq.Model,
Messages: messages,
Stream: &streaming,
Tools: tools,
Options: w.chatReq.Options,
}
body, err := json.Marshal(followUp)
if err != nil {
return api.ChatResponse{}, err
}
chatURL := envconfig.Host().String() + "/api/chat"
logutil.TraceContext(ctx, "anthropic middleware: followup request",
"url", chatURL,
"req", anthropic.TraceChatRequest(&followUp),
)
httpReq, err := http.NewRequestWithContext(ctx, "POST", chatURL, bytes.NewReader(body))
if err != nil {
return api.ChatResponse{}, err
}
httpReq.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(httpReq)
if err != nil {
return api.ChatResponse{}, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
respBody, _ := io.ReadAll(resp.Body)
logutil.TraceContext(ctx, "anthropic middleware: followup non-200 response",
"status", resp.StatusCode,
"response", strings.TrimSpace(string(respBody)),
)
return api.ChatResponse{}, fmt.Errorf("followup /api/chat returned status %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
}
var chatResp api.ChatResponse
if err := json.NewDecoder(resp.Body).Decode(&chatResp); err != nil {
return api.ChatResponse{}, err
}
logutil.TraceContext(ctx, "anthropic middleware: followup decoded", "resp", anthropic.TraceChatResponse(chatResp))
return chatResp, nil
return doFollowUpChat(ctx, *w.chatReq, messages, tools)
}
func (w *WebSearchAnthropicWriter) writePassthroughStreamChunk(chatResponse api.ChatResponse) error {
@@ -915,18 +833,6 @@ func isCloudModelName(name string) bool {
return modelref.HasExplicitCloudSource(name)
}
// extractQueryFromToolCall extracts the search query from a web_search tool call
func extractQueryFromToolCall(tc *api.ToolCall) string {
q, ok := tc.Function.Arguments.Get("query")
if !ok {
return ""
}
if s, ok := q.(string); ok {
return s
}
return ""
}
// writeSSE writes a Server-Sent Event
func writeSSE(w http.ResponseWriter, eventType string, data any) error {
d, err := json.Marshal(data)
-47
View File
@@ -653,53 +653,6 @@ func TestHasWebSearchTool(t *testing.T) {
}
}
func TestExtractQueryFromToolCall(t *testing.T) {
tests := []struct {
name string
tc *api.ToolCall
expected string
}{
{
name: "valid query",
tc: &api.ToolCall{
Function: api.ToolCallFunction{
Name: "web_search",
Arguments: makeArgs("query", "test search"),
},
},
expected: "test search",
},
{
name: "empty arguments",
tc: &api.ToolCall{
Function: api.ToolCallFunction{
Name: "web_search",
},
},
expected: "",
},
{
name: "no query key",
tc: &api.ToolCall{
Function: api.ToolCallFunction{
Name: "web_search",
Arguments: makeArgs("other", "value"),
},
},
expected: "",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
result := extractQueryFromToolCall(tt.tc)
if result != tt.expected {
t.Errorf("expected %q, got %q", tt.expected, result)
}
})
}
}
// makeArgs is a test helper that creates ToolCallFunctionArguments
func makeArgs(key string, value any) api.ToolCallFunctionArguments {
args := api.NewToolCallFunctionArguments()
+626 -1
View File
@@ -2,9 +2,12 @@ package middleware
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"math/rand"
"net/http"
"strings"
@@ -541,8 +544,616 @@ func (w *ResponsesWriter) Write(data []byte) (int, error) {
return w.writeResponse(data)
}
// WebSearchResponsesWriter runs the built-in Responses web_search tool on the
// server. The model sees it as an ordinary function; callers only see the
// native web_search_call items which describe the searches we actually ran.
type WebSearchResponsesWriter struct {
BaseWriter
inner *ResponsesWriter
req openai.ResponsesRequest
chat *api.ChatRequest
// The functions are injectable so the protocol lifecycle can be tested
// without a running server or cloud credentials.
search func(context.Context, string) (*api.WebSearchResponse, error)
followUpChat func(context.Context, []api.Message, api.Tools) (api.ChatResponse, error)
followUpStream func(context.Context, []api.Message, api.Tools, func(api.ChatResponse) error) error
newContext func() (context.Context, context.CancelFunc)
// Keep the initial model response for the follow-up context while streaming
// ordinary output immediately. Once web_search appears, its private function
// call and terminal chunk are intercepted and replaced by native events.
buffered []api.ChatResponse
webSearchPending bool
streamedInitialOutput bool
status int
done bool
// Accumulated across loop iterations by runLoop, consumed by
// writeWebSearchResponse / writeWebSearchStream.
preSearchThinking string // reasoning the model emitted before calling web_search
preSearchContent string // text the model emitted before calling web_search
otherToolCalls []api.ToolCall // non-web_search tool calls from mixed responses
finalOutputStreamed bool
}
func (w *WebSearchResponsesWriter) WriteHeader(code int) {
w.status = code
}
func (w *WebSearchResponsesWriter) WriteHeaderNow() {
if w.status != 0 {
w.ResponseWriter.WriteHeader(w.status)
}
w.ResponseWriter.WriteHeaderNow()
}
func (w *WebSearchResponsesWriter) Status() int {
if w.status != 0 {
return w.status
}
return w.ResponseWriter.Status()
}
func (w *WebSearchResponsesWriter) Write(data []byte) (int, error) {
if w.done {
return len(data), nil
}
if w.Status() != http.StatusOK {
return len(data), w.writeWebSearchError(decodeWebSearchResponseError(w.Status(), data), api.Metrics{})
}
var response api.ChatResponse
if err := json.Unmarshal(data, &response); err != nil {
return 0, err
}
if w.inner.stream {
w.buffered = append(w.buffered, response)
_, hasWebSearch, _ := findWebSearchToolCall(response.Message.ToolCalls)
if hasWebSearch {
w.webSearchPending = true
}
if !w.webSearchPending && len(response.Message.ToolCalls) == 0 {
if _, err := w.inner.writeResponse(data); err != nil {
return 0, err
}
if response.Message.Content != "" || response.Message.Thinking != "" {
w.streamedInitialOutput = true
}
if response.Done {
w.buffered = nil
w.done = true
}
return len(data), nil
}
// Tool-bearing chunks never pass through Process: its normal tool path
// latches text off for the rest of the stream. Stream ordinary output,
// then emit client tools through the latch-free path.
if response.Message.Content != "" || response.Message.Thinking != "" {
streamed := response
streamed.Message.ToolCalls = nil
streamed.Done = false
streamedData, err := json.Marshal(streamed)
if err != nil {
return 0, err
}
if _, err := w.inner.writeResponse(streamedData); err != nil {
return 0, err
}
w.streamedInitialOutput = true
}
var otherToolCalls []api.ToolCall
for _, tc := range response.Message.ToolCalls {
if tc.Function.Name != "web_search" {
otherToolCalls = append(otherToolCalls, tc)
}
}
if len(otherToolCalls) > 0 {
for _, event := range w.inner.converter.Process(api.ChatResponse{}) {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return 0, err
}
}
for _, event := range w.inner.converter.FinishMessageItem() {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return 0, err
}
}
for _, event := range w.inner.converter.EmitFunctionCallItems(otherToolCalls) {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return 0, err
}
}
w.streamedInitialOutput = true
}
if response.Done {
if w.webSearchPending {
return len(data), w.finishStream()
}
response.Message = api.Message{}
terminal, err := json.Marshal(response)
if err != nil {
return 0, err
}
if _, err := w.inner.writeResponse(terminal); err != nil {
return 0, err
}
w.buffered = nil
w.done = true
}
return len(data), nil
}
call, found, mixed := findWebSearchToolCall(response.Message.ToolCalls)
if !found {
return w.inner.writeResponse(data)
}
if mixed {
slog.Debug("preferring web_search tool call over client tool calls in mixed Responses response")
}
return len(data), w.runAndWrite(response, call)
}
func (w *WebSearchResponsesWriter) finishStream() error {
var initial api.ChatResponse
var call api.ToolCall
var found bool
var observed api.Metrics
var contentBuilder strings.Builder
var thinkingBuilder strings.Builder
var toolCalls []api.ToolCall
for _, response := range w.buffered {
observed.PromptEvalCount = max(observed.PromptEvalCount, response.Metrics.PromptEvalCount)
observed.EvalCount = max(observed.EvalCount, response.Metrics.EvalCount)
if response.Message.Content != "" {
contentBuilder.WriteString(response.Message.Content)
}
if response.Message.Thinking != "" {
thinkingBuilder.WriteString(response.Message.Thinking)
}
toolCalls = append(toolCalls, response.Message.ToolCalls...)
if candidate, ok, mixed := findWebSearchToolCall(response.Message.ToolCalls); ok && !found {
if mixed {
slog.Debug("preferring web_search tool call over client tool calls in mixed Responses response")
}
initial, call, found = response, candidate, true
}
}
if !found {
return fmt.Errorf("web_search call disappeared before the terminal chunk")
}
// Combine model output from all streamed chunks into the initial response so
// runLoop can preserve it before the web search events and in the follow-up.
initial.Message.Content = contentBuilder.String()
initial.Message.Thinking = thinkingBuilder.String()
initial.Message.ToolCalls = toolCalls
initial.Metrics = observed
return w.runAndWrite(initial, call)
}
func (w *WebSearchResponsesWriter) runAndWrite(initial api.ChatResponse, call api.ToolCall) error {
ctx, cancel := w.loopContext()
defer cancel()
if w.inner.stream {
w.ResponseWriter.Header().Set("Content-Type", "text/event-stream")
}
final, calls, usage, err := w.runLoop(ctx, initial, call)
if err != nil {
return w.writeWebSearchError(err, usage)
}
if w.inner.stream {
return w.writeWebSearchStream(final, usage)
}
return w.writeWebSearchResponse(final, calls, usage)
}
func (w *WebSearchResponsesWriter) runLoop(ctx context.Context, initial api.ChatResponse, call api.ToolCall) (api.ChatResponse, []openai.ResponsesWebSearchCall, api.Metrics, error) {
messages := append([]api.Message(nil), w.chat.Messages...)
tools := append(api.Tools(nil), w.chat.Tools...)
usage := initial.Metrics
current, currentCall := initial, call
calls := make([]openai.ResponsesWebSearchCall, 0, maxWebSearchLoops)
var preSearchThinking strings.Builder
var preSearchContent strings.Builder
var otherToolCalls []api.ToolCall
currentOutputStreamed := w.streamedInitialOutput
// Emit response.created / response.in_progress once, before the loop.
if w.inner.stream {
for _, event := range w.inner.converter.Process(api.ChatResponse{}) {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return api.ChatResponse{}, calls, usage, err
}
}
}
for loop := 1; loop <= maxWebSearchLoops; loop++ {
// Collect non-web_search tool calls from mixed responses so they can
// be surfaced to the client instead of silently dropped.
var currentOtherToolCalls []api.ToolCall
for _, tc := range current.Message.ToolCalls {
if tc.Function.Name != "web_search" {
currentOtherToolCalls = append(currentOtherToolCalls, tc)
}
}
if !w.inner.stream {
otherToolCalls = append(otherToolCalls, currentOtherToolCalls...)
}
if w.inner.stream && currentOutputStreamed {
for _, event := range w.inner.converter.FinishMessageItem() {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return api.ChatResponse{}, calls, usage, err
}
}
}
// Emit pre-search content (text the model produced before calling
// web_search) as a completed message item before the search events.
if current.Message.Thinking != "" && w.inner.stream && !currentOutputStreamed {
thinkingResponse := api.ChatResponse{Message: api.Message{Role: "assistant", Thinking: current.Message.Thinking}}
for _, event := range w.inner.converter.Process(thinkingResponse) {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return api.ChatResponse{}, calls, usage, err
}
}
}
if current.Message.Thinking != "" && !w.inner.stream {
if preSearchThinking.Len() > 0 {
preSearchThinking.WriteString("\n")
}
preSearchThinking.WriteString(current.Message.Thinking)
}
if current.Message.Content != "" {
if w.inner.stream && !currentOutputStreamed {
contentResponse := api.ChatResponse{Message: api.Message{Role: "assistant", Content: current.Message.Content}}
for _, event := range w.inner.converter.Process(contentResponse) {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return api.ChatResponse{}, calls, usage, err
}
}
for _, event := range w.inner.converter.FinishMessageItem() {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return api.ChatResponse{}, calls, usage, err
}
}
} else if !w.inner.stream {
if preSearchContent.Len() > 0 {
preSearchContent.WriteString("\n")
}
preSearchContent.WriteString(current.Message.Content)
}
}
query := extractQueryFromToolCall(&currentCall)
if strings.TrimSpace(query) == "" {
return api.ChatResponse{}, calls, usage, fmt.Errorf("web_search requires a non-empty string query")
}
responseCall := openai.ResponsesWebSearchCall{
ID: fmt.Sprintf("ws_%s_%d", strings.TrimPrefix(w.inner.responseID, "resp_"), loop),
Type: "web_search_call",
Status: "completed",
Action: &openai.ResponsesWebSearchAction{Type: "search", Query: query},
}
outputIndex := 0
if w.inner.stream {
var events []openai.ResponsesStreamEvent
outputIndex, events = w.inner.converter.StartWebSearchCall(responseCall)
for _, event := range events {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return api.ChatResponse{}, calls, usage, err
}
}
}
slog.Debug("executing Responses web search", "loop", loop)
searchResponse, err := w.webSearch(ctx, query)
if err != nil {
return api.ChatResponse{}, calls, usage, err
}
slog.Debug("completed Responses web search", "loop", loop, "results", len(searchResponse.Results))
if w.inner.stream {
for _, event := range w.inner.converter.FinishWebSearchCall(responseCall, outputIndex) {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return api.ChatResponse{}, calls, usage, err
}
}
}
calls = append(calls, responseCall)
messages = append(messages,
buildWebSearchAssistantMessage(current, currentCall),
api.Message{Role: "tool", ToolCallID: currentCall.ID, Content: formatResponsesWebSearchResults(searchResponse.Results)},
)
var followUp api.ChatResponse
var followUpOutputStreamed bool
if w.inner.stream {
followUp, followUpOutputStreamed, err = w.callFollowUpStream(ctx, messages, tools)
} else {
followUp, err = w.callFollowUp(ctx, messages, tools)
}
if err != nil {
return api.ChatResponse{}, calls, usage, err
}
usage.PromptEvalCount += followUp.Metrics.PromptEvalCount
usage.EvalCount += followUp.Metrics.EvalCount
next, hasWebSearch, mixed := findWebSearchToolCall(followUp.Message.ToolCalls)
if mixed {
slog.Debug("preferring web_search tool call over client tool calls in mixed Responses followup")
}
if !hasWebSearch {
w.preSearchThinking = preSearchThinking.String()
w.preSearchContent = preSearchContent.String()
w.otherToolCalls = otherToolCalls
w.finalOutputStreamed = followUpOutputStreamed
followUp.Metrics = usage
return followUp, calls, usage, nil
}
current, currentCall = followUp, next
currentOutputStreamed = followUpOutputStreamed
}
w.preSearchThinking = preSearchThinking.String()
w.preSearchContent = preSearchContent.String()
w.otherToolCalls = otherToolCalls
return current, calls, usage, fmt.Errorf("web_search exceeded the maximum of %d calls", maxWebSearchLoops)
}
func (w *WebSearchResponsesWriter) loopContext() (context.Context, context.CancelFunc) {
if w.newContext != nil {
return w.newContext()
}
return context.WithTimeout(context.Background(), 5*time.Minute)
}
func (w *WebSearchResponsesWriter) webSearch(ctx context.Context, query string) (*api.WebSearchResponse, error) {
if w.search != nil {
return w.search(ctx, query)
}
client, err := api.ClientFromEnvironment()
if err != nil {
return nil, err
}
return client.WebSearchExperimental(ctx, &api.WebSearchRequest{Query: query, MaxResults: 5})
}
func (w *WebSearchResponsesWriter) callFollowUp(ctx context.Context, messages []api.Message, tools api.Tools) (api.ChatResponse, error) {
if w.followUpChat != nil {
return w.followUpChat(ctx, messages, tools)
}
return doFollowUpChat(ctx, *w.chat, messages, tools)
}
func (w *WebSearchResponsesWriter) callFollowUpStream(ctx context.Context, messages []api.Message, tools api.Tools) (api.ChatResponse, bool, error) {
var final api.ChatResponse
var content strings.Builder
var thinking strings.Builder
var role string
var toolCalls []api.ToolCall
outputStreamed := false
yield := func(response api.ChatResponse) error {
final = response
if response.Message.Role != "" {
role = response.Message.Role
}
content.WriteString(response.Message.Content)
thinking.WriteString(response.Message.Thinking)
toolCalls = append(toolCalls, response.Message.ToolCalls...)
streamed := response
streamed.Message.ToolCalls = nil
streamed.Done = false
if streamed.Message.Content != "" || streamed.Message.Thinking != "" {
events := w.inner.converter.Process(streamed)
for _, event := range events {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return err
}
}
outputStreamed = outputStreamed || len(events) > 0
}
var otherToolCalls []api.ToolCall
for _, tc := range response.Message.ToolCalls {
if tc.Function.Name != "web_search" {
otherToolCalls = append(otherToolCalls, tc)
}
}
if len(otherToolCalls) > 0 {
for _, event := range w.inner.converter.FinishMessageItem() {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return err
}
}
for _, event := range w.inner.converter.EmitFunctionCallItems(otherToolCalls) {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return err
}
}
outputStreamed = true
}
return nil
}
var err error
switch {
case w.followUpStream != nil:
err = w.followUpStream(ctx, messages, tools, yield)
case w.followUpChat != nil:
var response api.ChatResponse
response, err = w.followUpChat(ctx, messages, tools)
if err == nil {
err = yield(response)
}
default:
err = streamFollowUpChat(ctx, *w.chat, messages, tools, yield)
}
if err != nil {
return api.ChatResponse{}, outputStreamed, err
}
final.Message.Role = role
final.Message.Content = content.String()
final.Message.Thinking = thinking.String()
final.Message.ToolCalls = toolCalls
return final, outputStreamed, nil
}
func formatResponsesWebSearchResults(results []api.WebSearchResult) string {
var text strings.Builder
for _, result := range results {
fmt.Fprintf(&text, "Title: %s\nURL: %s\n", result.Title, result.URL)
if result.Content != "" {
fmt.Fprintf(&text, "Content: %s\n", result.Content)
}
text.WriteByte('\n')
}
return text.String()
}
func (w *WebSearchResponsesWriter) writeWebSearchResponse(final api.ChatResponse, calls []openai.ResponsesWebSearchCall, usage api.Metrics) error {
response := openai.ToResponse(w.inner.model, w.inner.responseID, w.inner.itemID, final, w.req)
completedAt := time.Now().Unix()
response.CompletedAt = &completedAt
response.Output = buildResponsesWebSearchOutput(response.Output, w.preSearchThinking, w.preSearchContent, calls, w.otherToolCalls)
if response.Usage != nil {
response.Usage.InputTokens = usage.PromptEvalCount
response.Usage.OutputTokens = usage.EvalCount
response.Usage.TotalTokens = usage.PromptEvalCount + usage.EvalCount
}
w.ResponseWriter.Header().Set("Content-Type", "application/json")
w.done = true
return json.NewEncoder(w.ResponseWriter).Encode(response)
}
// buildResponsesWebSearchOutput assembles the final non-streaming output in
// model-leg order: pre-search reasoning/text, server and mixed tool calls, then
// the final model output.
func buildResponsesWebSearchOutput(output []openai.ResponsesOutputItem, preSearchThinking, preSearchContent string, searchCalls []openai.ResponsesWebSearchCall, otherToolCalls []api.ToolCall) []openai.ResponsesOutputItem {
items := make([]openai.ResponsesOutputItem, 0, len(output)+len(searchCalls)+len(otherToolCalls)+2)
if preSearchThinking != "" {
items = append(items, openai.ResponsesOutputItem{
ID: "rs_presearch",
Type: "reasoning",
Summary: []openai.ResponsesReasoningSummary{
{Type: "summary_text", Text: preSearchThinking},
},
EncryptedContent: preSearchThinking,
})
}
// pre-search message (if the model emitted content before calling web_search)
if preSearchContent != "" {
items = append(items, openai.ResponsesOutputItem{
ID: "msg_presearch",
Type: "message",
Status: "completed",
Role: "assistant",
Content: []openai.ResponsesOutputContent{
{Type: "output_text", Text: preSearchContent, Annotations: []any{}, Logprobs: []any{}},
},
})
}
// web_search_call items
for _, call := range searchCalls {
items = append(items, openai.WebSearchCallOutputItem(call))
}
// function_call items from mixed responses
convertedCalls := openai.ToToolCalls(otherToolCalls)
for i, tc := range convertedCalls {
items = append(items, openai.ResponsesOutputItem{
ID: fmt.Sprintf("fc_mixed_%d", i),
Type: "function_call",
Status: "completed",
CallID: tc.ID,
Name: tc.Function.Name,
Arguments: tc.Function.Arguments,
})
}
// remaining items (final reasoning, message, or function calls)
items = append(items, output...)
return items
}
func (w *WebSearchResponsesWriter) writeWebSearchStream(final api.ChatResponse, usage api.Metrics) error {
if w.finalOutputStreamed {
final.Message = api.Message{}
}
final.Metrics = usage
final.Done = true
for _, event := range w.inner.converter.Process(final) {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return err
}
}
w.done = true
return nil
}
func (w *WebSearchResponsesWriter) writeWebSearchError(err error, usage api.Metrics) error {
message := err.Error()
status := http.StatusBadGateway
errorCode := "api_error"
var authorizationError api.AuthorizationError
var statusError api.StatusError
switch {
case errors.As(err, &authorizationError):
status = authorizationError.StatusCode
errorCode = "authentication_error"
if authorizationError.SigninURL != "" {
message += "; sign in at " + authorizationError.SigninURL
}
case errors.As(err, &statusError):
status = statusError.StatusCode
if status == http.StatusTooManyRequests {
errorCode = "rate_limit_exceeded"
}
}
if !w.inner.stream {
w.ResponseWriter.Header().Set("Content-Type", "application/json")
w.ResponseWriter.WriteHeader(status)
w.done = true
return json.NewEncoder(w.ResponseWriter).Encode(openai.NewError(status, message))
}
w.ResponseWriter.Header().Set("Content-Type", "text/event-stream")
response := map[string]any{
"id": w.inner.responseID, "object": "response", "status": "failed", "model": w.req.Model,
"output": []any{}, "error": map[string]any{"code": errorCode, "message": message},
"usage": map[string]any{"input_tokens": usage.PromptEvalCount, "output_tokens": usage.EvalCount, "total_tokens": usage.PromptEvalCount + usage.EvalCount},
}
initialEvents := w.inner.converter.Process(api.ChatResponse{})
for _, event := range initialEvents {
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return err
}
}
event := w.inner.converter.ResponseFailed(response)
if err := w.inner.writeEvent(event.Event, event.Data); err != nil {
return err
}
w.done = true
return nil
}
func decodeWebSearchResponseError(status int, data []byte) error {
var response struct {
Error string `json:"error"`
SigninURL string `json:"signin_url"`
}
if err := json.Unmarshal(data, &response); err != nil {
response.Error = string(data)
}
if status == http.StatusUnauthorized {
return api.AuthorizationError{StatusCode: status, Status: response.Error, SigninURL: response.SigninURL}
}
return api.StatusError{StatusCode: status, ErrorMessage: response.Error}
}
func ResponsesMiddleware() gin.HandlerFunc {
return func(c *gin.Context) {
requestCtx := c.Request.Context()
if c.GetHeader("Content-Encoding") == "zstd" {
reader, err := zstd.NewReader(c.Request.Body, zstd.WithDecoderMaxMemory(8<<20))
if err != nil {
@@ -600,7 +1211,21 @@ func ResponsesMiddleware() gin.HandlerFunc {
c.Writer.Header().Set("Connection", "keep-alive")
}
c.Writer = w
hasWebSearch := openai.HasWebSearchTool(req.Tools)
slog.Debug("parsed Responses tools", "count", len(req.Tools), "web_search", hasWebSearch)
if hasWebSearch {
c.Writer = &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: c.Writer},
inner: w,
req: req,
chat: chatReq,
newContext: func() (context.Context, context.CancelFunc) {
return context.WithTimeout(requestCtx, 5*time.Minute)
},
}
} else {
c.Writer = w
}
c.Next()
}
}
+903
View File
@@ -0,0 +1,903 @@
package middleware
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/gin-gonic/gin"
"github.com/ollama/ollama/api"
"github.com/ollama/ollama/openai"
)
func TestWebSearchResponsesWriterNonStreaming(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
request := openai.ResponsesRequest{
Model: "test-model",
Tools: []openai.ResponsesTool{{Type: "web_search"}},
}
inner := &ResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer},
model: request.Model,
responseID: "resp_test",
itemID: "msg_test",
request: request,
}
followUps := 0
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer},
inner: inner,
req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(_ context.Context, query string) (*api.WebSearchResponse, error) {
if query != "ollama news" {
t.Fatalf("search query = %q", query)
}
return &api.WebSearchResponse{Results: []api.WebSearchResult{{Title: "Ollama", URL: "https://ollama.com/news", Content: "news"}}}, nil
},
followUpChat: func(_ context.Context, messages []api.Message, _ api.Tools) (api.ChatResponse, error) {
followUps++
if len(messages) != 2 || messages[1].Role != "tool" {
t.Fatalf("follow-up messages = %#v", messages)
}
if strings.Contains(messages[1].Content, "Cite") || !strings.Contains(messages[1].Content, "URL: https://ollama.com/news") {
t.Fatalf("unexpected search result content: %q", messages[1].Content)
}
return api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "Read [Ollama](https://ollama.com/news)."}, Metrics: api.Metrics{PromptEvalCount: 7, EvalCount: 3}}, nil
},
}
initial := api.ChatResponse{Done: true, Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "ollama news"})}}}}, Metrics: api.Metrics{PromptEvalCount: 5, EvalCount: 2}}
data, err := json.Marshal(initial)
if err != nil {
t.Fatal(err)
}
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
if followUps != 1 {
t.Fatalf("follow-up calls = %d", followUps)
}
var response openai.ResponsesResponse
if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
t.Fatalf("decode response: %v: %s", err, recorder.Body.String())
}
if len(response.Output) != 2 || response.Output[0].Type != "web_search_call" || response.Output[1].Type != "message" {
t.Fatalf("output = %#v", response.Output)
}
if response.Output[0].Action == nil || response.Output[0].Action.Query != "ollama news" {
t.Fatalf("search action = %#v", response.Output[0].Action)
}
if response.Usage == nil || response.Usage.InputTokens != 12 || response.Usage.OutputTokens != 5 {
t.Fatalf("usage = %#v", response.Usage)
}
if len(response.Output[1].Content[0].Annotations) != 0 {
t.Fatalf("annotations = %#v, want none", response.Output[1].Content[0].Annotations)
}
}
func TestWebSearchResponsesWriterStreamingNoSearchStreamsImmediately(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request}
chunk, _ := json.Marshal(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "hello"}})
if _, err := writer.Write(chunk); err != nil {
t.Fatal(err)
}
if body := recorder.Body.String(); !strings.Contains(body, "response.output_text.delta") || !strings.Contains(body, "hello") {
t.Fatalf("content was not streamed immediately: %s", body)
} else if strings.Contains(body, "response.completed") {
t.Fatalf("response completed before terminal chunk: %s", body)
}
done, _ := json.Marshal(api.ChatResponse{Done: true})
if _, err := writer.Write(done); err != nil {
t.Fatal(err)
}
if body := recorder.Body.String(); !strings.Contains(body, "response.completed") || !strings.Contains(body, "hello") {
t.Fatalf("missing completed event: %s", recorder.Body.String())
}
}
func TestWebSearchResponsesWriterNonStreamingAuthorizationError(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
request := openai.ResponsesRequest{Model: "test-model", Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, model: request.Model, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request}
writer.WriteHeader(http.StatusUnauthorized)
data := []byte(`{"error":"sign in required","signin_url":"https://ollama.com/signin"}`)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
if recorder.Code != http.StatusUnauthorized {
t.Fatalf("status = %d, want %d", recorder.Code, http.StatusUnauthorized)
}
if body := recorder.Body.String(); !strings.Contains(body, "https://ollama.com/signin") {
t.Fatalf("missing sign-in URL: %s", body)
}
}
func TestWebSearchResponsesWriterStreamingAuthorizationError(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model:cloud", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request}
writer.WriteHeader(http.StatusUnauthorized)
data := []byte(`{"error":"sign in required","signin_url":"https://ollama.com/signin"}`)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
if recorder.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", recorder.Code, http.StatusOK)
}
body := recorder.Body.String()
if !strings.Contains(body, "response.failed") || !strings.Contains(body, "https://ollama.com/signin") {
t.Fatalf("missing streaming authorization error: %s", body)
}
}
func TestWebSearchResponsesWriterStreamingRateLimitError(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request}
if err := writer.writeWebSearchError(api.StatusError{StatusCode: http.StatusTooManyRequests, ErrorMessage: "slow down"}, api.Metrics{}); err != nil {
t.Fatal(err)
}
if recorder.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", recorder.Code, http.StatusOK)
}
body := recorder.Body.String()
if !strings.Contains(body, "response.failed") || !strings.Contains(body, "rate_limit_exceeded") {
t.Fatalf("unexpected rate-limit response: %s", body)
}
if !strings.Contains(recorder.Header().Get("Content-Type"), "text/event-stream") {
t.Fatalf("content type = %q", recorder.Header().Get("Content-Type"))
}
}
func TestResponsesMiddlewareWebSearchStatusOnlyResponse(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
router := gin.New()
router.POST("/v1/responses", ResponsesMiddleware(), func(c *gin.Context) {
c.AbortWithStatus(http.StatusServiceUnavailable)
})
request := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{
"model":"test-model",
"input":"hello",
"tools":[{"type":"web_search"}]
}`))
request.Header.Set("Content-Type", "application/json")
router.ServeHTTP(recorder, request)
if recorder.Code != http.StatusServiceUnavailable {
t.Fatalf("status = %d, want %d", recorder.Code, http.StatusServiceUnavailable)
}
}
func TestWebSearchResponsesWriterPreservesFollowUpErrors(t *testing.T) {
tests := []struct {
name string
status int
body string
wantInBody string
}{
{name: "authorization", status: http.StatusUnauthorized, body: `{"error":"sign in required","signin_url":"https://ollama.com/signin/followup"}`, wantInBody: "https://ollama.com/signin/followup"},
{name: "rate limit", status: http.StatusTooManyRequests, body: `{"error":"slow down"}`, wantInBody: "slow down"},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
if request.URL.Path != "/api/chat" {
t.Errorf("follow-up path = %q", request.URL.Path)
}
w.WriteHeader(test.status)
_, _ = w.Write([]byte(test.body))
}))
defer server.Close()
t.Setenv("OLLAMA_HOST", server.URL)
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
request := openai.ResponsesRequest{Model: "test-model:cloud", Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, model: request.Model, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) { return &api.WebSearchResponse{}, nil },
}
initial := api.ChatResponse{Done: true, Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "test"})}}}}}
data, _ := json.Marshal(initial)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
if recorder.Code != test.status || !strings.Contains(recorder.Body.String(), test.wantInBody) {
t.Fatalf("response status=%d body=%s", recorder.Code, recorder.Body.String())
}
})
}
}
func TestWebSearchResponsesWriterStreamingHidesInternalFunction(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) {
body := recorder.Body.String()
if !strings.Contains(body, "response.web_search_call.searching") || strings.Contains(body, "response.web_search_call.completed") {
t.Fatalf("search lifecycle before execution = %s", body)
}
return &api.WebSearchResponse{}, nil
},
followUpChat: func(context.Context, []api.Message, api.Tools) (api.ChatResponse, error) {
body := recorder.Body.String()
if !strings.Contains(body, "response.web_search_call.completed") || strings.Contains(body, "response.completed") {
t.Fatalf("search lifecycle before follow-up = %s", body)
}
return api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "done"}}, nil
},
}
initial := api.ChatResponse{Done: true, Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "test"})}}}}}
data, _ := json.Marshal(initial)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
body := recorder.Body.String()
if !strings.Contains(body, "response.web_search_call.completed") {
t.Fatalf("missing completed web search event: %s", body)
}
if strings.Contains(body, "response.function_call_arguments") || strings.Contains(body, `"type":"function_call"`) {
t.Fatalf("internal function call leaked: %s", body)
}
}
func TestWebSearchResponsesWriterStreamingToolCallBeforeDoneChunk(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) { return &api.WebSearchResponse{}, nil },
followUpChat: func(context.Context, []api.Message, api.Tools) (api.ChatResponse, error) {
return api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "done"}}, nil
},
}
toolChunk := api.ChatResponse{Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "test"})}}}}}
data, _ := json.Marshal(toolChunk)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
if recorder.Body.Len() != 0 {
t.Fatalf("tool chunk leaked before done: %s", recorder.Body.String())
}
done, _ := json.Marshal(api.ChatResponse{Done: true, Metrics: api.Metrics{PromptEvalCount: 9, EvalCount: 4}})
if _, err := writer.Write(done); err != nil {
t.Fatal(err)
}
body := recorder.Body.String()
if !strings.Contains(body, "response.web_search_call.completed") || strings.Contains(body, "response.function_call_arguments") {
t.Fatalf("unexpected response stream: %s", body)
}
}
func TestWebSearchResponsesWriterStreamingPreservesContentBeforeToolCall(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) { return &api.WebSearchResponse{}, nil },
followUpChat: func(context.Context, []api.Message, api.Tools) (api.ChatResponse, error) {
return api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "done"}}, nil
},
}
content, _ := json.Marshal(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "I will search."}})
if _, err := writer.Write(content); err != nil {
t.Fatal(err)
}
if body := recorder.Body.String(); !strings.Contains(body, "response.output_text.delta") || !strings.Contains(body, "I will search.") {
t.Fatalf("pre-search content was not streamed immediately: %s", body)
} else if strings.Contains(body, "response.web_search_call.in_progress") {
t.Fatalf("search started before its tool call: %s", body)
}
toolChunk, _ := json.Marshal(api.ChatResponse{Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "test"})}}}}})
if _, err := writer.Write(toolChunk); err != nil {
t.Fatal(err)
}
done, _ := json.Marshal(api.ChatResponse{Done: true})
if _, err := writer.Write(done); err != nil {
t.Fatal(err)
}
body := recorder.Body.String()
// Pre-search content must be emitted as a completed message item before
// the web_search_call events, and the private function must not leak.
if !strings.Contains(body, "I will search.") {
t.Fatalf("pre-search content was discarded: %s", body)
}
if !strings.Contains(body, "response.web_search_call.completed") {
t.Fatalf("missing web search completed event: %s", body)
}
if strings.Contains(body, "response.function_call_arguments") {
t.Fatalf("private web_search function leaked: %s", body)
}
if strings.Count(body, "event: response.output_text.delta") != 2 {
t.Fatalf("unexpected output delta count; pre-search content may have been replayed: %s", body)
}
}
func TestWebSearchResponsesWriterStreamingPreservesThinkingBeforeToolCall(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) { return &api.WebSearchResponse{}, nil },
followUpChat: func(_ context.Context, messages []api.Message, _ api.Tools) (api.ChatResponse, error) {
assistant := messages[len(messages)-2]
if assistant.Thinking != "I should search first." {
t.Fatalf("follow-up thinking = %q", assistant.Thinking)
}
return api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "done"}}, nil
},
}
thinking, _ := json.Marshal(api.ChatResponse{Message: api.Message{Role: "assistant", Thinking: "I should search first."}})
if _, err := writer.Write(thinking); err != nil {
t.Fatal(err)
}
if body := recorder.Body.String(); !strings.Contains(body, "response.reasoning_summary_text.delta") || !strings.Contains(body, "I should search first.") {
t.Fatalf("pre-search reasoning was not streamed immediately: %s", body)
} else if strings.Contains(body, "response.web_search_call.in_progress") {
t.Fatalf("search started before its tool call: %s", body)
}
toolChunk, _ := json.Marshal(api.ChatResponse{Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "test"})}}}}})
if _, err := writer.Write(toolChunk); err != nil {
t.Fatal(err)
}
done, _ := json.Marshal(api.ChatResponse{Done: true})
if _, err := writer.Write(done); err != nil {
t.Fatal(err)
}
body := recorder.Body.String()
reasoningDelta := strings.Index(body, "response.reasoning_summary_text.delta")
reasoningDone := strings.Index(body, "response.reasoning_summary_text.done")
searchStarted := strings.Index(body, "response.web_search_call.in_progress")
if reasoningDelta < 0 || reasoningDone < reasoningDelta || searchStarted < reasoningDone {
t.Fatalf("reasoning/search lifecycle is out of order: %s", body)
}
}
func TestWebSearchResponsesWriterStreamingContentAndToolCallInSameChunk(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) { return &api.WebSearchResponse{}, nil },
followUpChat: func(context.Context, []api.Message, api.Tools) (api.ChatResponse, error) {
return api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "done"}}, nil
},
}
chunk, _ := json.Marshal(api.ChatResponse{Message: api.Message{
Role: "assistant",
Content: "Let me check.",
ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{
Name: "web_search", Arguments: testArgs(map[string]any{"query": "test"}),
}}},
}})
if _, err := writer.Write(chunk); err != nil {
t.Fatal(err)
}
if body := recorder.Body.String(); !strings.Contains(body, "Let me check.") || strings.Contains(body, "response.function_call_arguments") {
t.Fatalf("same-chunk content was not streamed safely: %s", body)
}
done, _ := json.Marshal(api.ChatResponse{Done: true})
if _, err := writer.Write(done); err != nil {
t.Fatal(err)
}
body := recorder.Body.String()
if !strings.Contains(body, "response.web_search_call.completed") || strings.Count(body, "event: response.output_text.delta") != 2 {
t.Fatalf("unexpected response stream: %s", body)
}
}
func TestWebSearchResponsesWriterStreamsFollowUpAsProduced(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) { return &api.WebSearchResponse{}, nil },
followUpStream: func(_ context.Context, _ []api.Message, _ api.Tools, yield func(api.ChatResponse) error) error {
if err := yield(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "streamed "}}); err != nil {
return err
}
if body := recorder.Body.String(); !strings.Contains(body, `"delta":"streamed "`) || strings.Contains(body, "response.completed") {
t.Fatalf("first follow-up chunk was not flushed immediately: %s", body)
}
if err := yield(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "answer"}}); err != nil {
return err
}
return yield(api.ChatResponse{Done: true, Metrics: api.Metrics{PromptEvalCount: 7, EvalCount: 3}})
},
}
initial := api.ChatResponse{Done: true, Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "test"})}}}}, Metrics: api.Metrics{PromptEvalCount: 5, EvalCount: 2}}
data, _ := json.Marshal(initial)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
body := recorder.Body.String()
searchDone := strings.Index(body, "response.web_search_call.completed")
firstDelta := strings.Index(body, `"delta":"streamed "`)
secondDelta := strings.Index(body, `"delta":"answer"`)
completed := strings.Index(body, "response.completed")
if searchDone < 0 || firstDelta < searchDone || secondDelta < firstDelta || completed < secondDelta {
t.Fatalf("follow-up stream lifecycle is out of order: %s", body)
}
output := completedResponseOutput(t, body)
if len(output) != 2 || output[0]["type"] != "web_search_call" || output[1]["type"] != "message" {
t.Fatalf("terminal output = %#v", output)
}
content := output[1]["content"].([]any)[0].(map[string]any)
if content["text"] != "streamed answer" {
t.Fatalf("final text = %#v", content["text"])
}
}
func TestWebSearchResponsesWriterStreamsFollowUpBeforeSecondSearch(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
searches := 0
followUps := 0
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(_ context.Context, query string) (*api.WebSearchResponse, error) {
searches++
if query != []string{"first", "second"}[searches-1] {
t.Fatalf("search %d query = %q", searches, query)
}
return &api.WebSearchResponse{}, nil
},
followUpStream: func(_ context.Context, messages []api.Message, _ api.Tools, yield func(api.ChatResponse) error) error {
followUps++
if followUps == 1 {
if err := yield(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "Need another search."}}); err != nil {
return err
}
if !strings.Contains(recorder.Body.String(), `"delta":"Need another search."`) {
t.Fatalf("intermediate content was not streamed: %s", recorder.Body.String())
}
if err := yield(api.ChatResponse{Message: api.Message{Role: "assistant", ToolCalls: []api.ToolCall{{ID: "call_2", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "second"})}}}}}); err != nil {
return err
}
return yield(api.ChatResponse{Done: true, Metrics: api.Metrics{PromptEvalCount: 3, EvalCount: 4}})
}
assistant := messages[len(messages)-2]
if assistant.Content != "Need another search." || len(assistant.ToolCalls) != 1 || assistant.ToolCalls[0].Function.Name != "web_search" {
t.Fatalf("second-search assistant context = %#v", assistant)
}
if err := yield(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "Final "}}); err != nil {
return err
}
if err := yield(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "answer."}}); err != nil {
return err
}
return yield(api.ChatResponse{Done: true, Metrics: api.Metrics{PromptEvalCount: 5, EvalCount: 6}})
},
}
initial := api.ChatResponse{Done: true, Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "first"})}}}}, Metrics: api.Metrics{PromptEvalCount: 1, EvalCount: 2}}
data, _ := json.Marshal(initial)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
if searches != 2 || followUps != 2 {
t.Fatalf("searches=%d follow-ups=%d, want 2 each", searches, followUps)
}
body := recorder.Body.String()
firstSearchDone := strings.Index(body, "response.web_search_call.completed")
intermediate := strings.Index(body, `"delta":"Need another search."`)
secondSearch := strings.Index(body, `"query":"second"`)
finalDelta := strings.Index(body, `"delta":"Final "`)
completed := strings.Index(body, "response.completed")
if firstSearchDone < 0 || intermediate < firstSearchDone || secondSearch < intermediate || finalDelta < secondSearch || completed < finalDelta {
t.Fatalf("repeated search lifecycle is out of order: %s", body)
}
if strings.Count(body, "event: response.web_search_call.completed") != 2 || strings.Contains(body, "response.function_call_arguments") {
t.Fatalf("unexpected search events: %s", body)
}
output := completedResponseOutput(t, body)
wantTypes := []string{"web_search_call", "message", "web_search_call", "message"}
if len(output) != len(wantTypes) {
t.Fatalf("terminal output = %#v", output)
}
for i, want := range wantTypes {
if output[i]["type"] != want {
t.Fatalf("output[%d] type = %v, want %s: %#v", i, output[i]["type"], want, output)
}
}
}
func TestWebSearchResponsesWriterStreamingMixedFollowUpDoesNotLatchText(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}, {Type: "function", Name: "get_weather", Description: ptr("weather"), Parameters: map[string]any{"type": "object"}}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) { return &api.WebSearchResponse{}, nil },
followUpStream: func(_ context.Context, _ []api.Message, _ api.Tools, yield func(api.ChatResponse) error) error {
chunks := []api.ChatResponse{
{Message: api.Message{Role: "assistant", Content: "before"}},
{Message: api.Message{Role: "assistant", ToolCalls: []api.ToolCall{{ID: "call_weather", Function: api.ToolCallFunction{Name: "get_weather", Arguments: testArgs(map[string]any{"city": "SF"})}}}}},
{Message: api.Message{Role: "assistant", Content: " after"}},
{Done: true},
}
for _, chunk := range chunks {
if err := yield(chunk); err != nil {
return err
}
}
return nil
},
}
initial := api.ChatResponse{Done: true, Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "test"})}}}}}
data, _ := json.Marshal(initial)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
body := recorder.Body.String()
if !strings.Contains(body, `"delta":"before"`) || !strings.Contains(body, `"delta":" after"`) {
t.Fatalf("follow-up text was dropped: %s", body)
}
if strings.Count(body, "event: response.function_call_arguments.delta") != 1 {
t.Fatalf("function call should be emitted once: %s", body)
}
output := completedResponseOutput(t, body)
wantTypes := []string{"web_search_call", "message", "function_call", "message"}
if len(output) != len(wantTypes) {
t.Fatalf("terminal output = %#v", output)
}
for i, want := range wantTypes {
if output[i]["type"] != want {
t.Fatalf("output[%d] type = %v, want %s: %#v", i, output[i]["type"], want, output)
}
}
}
func TestWebSearchResponsesWriterStreamingSplitInitialToolsDoNotLatchFinalText(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}, {Type: "function", Name: "get_weather", Description: ptr("weather"), Parameters: map[string]any{"type": "object"}}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) { return &api.WebSearchResponse{}, nil },
followUpStream: func(_ context.Context, _ []api.Message, _ api.Tools, yield func(api.ChatResponse) error) error {
if err := yield(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "final answer"}}); err != nil {
return err
}
return yield(api.ChatResponse{Done: true})
},
}
weather, _ := json.Marshal(api.ChatResponse{Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_weather", Function: api.ToolCallFunction{Name: "get_weather", Arguments: testArgs(map[string]any{"city": "SF"})}}}}})
search, _ := json.Marshal(api.ChatResponse{Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_search", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "weather"})}}}}})
done, _ := json.Marshal(api.ChatResponse{Done: true})
for _, chunk := range [][]byte{weather, search, done} {
if _, err := writer.Write(chunk); err != nil {
t.Fatal(err)
}
}
body := recorder.Body.String()
if !strings.Contains(body, `"delta":"final answer"`) || strings.Count(body, "event: response.function_call_arguments.delta") != 1 {
t.Fatalf("split initial tools corrupted stream: %s", body)
}
output := completedResponseOutput(t, body)
wantTypes := []string{"function_call", "web_search_call", "message"}
if len(output) != len(wantTypes) {
t.Fatalf("terminal output = %#v", output)
}
for i, want := range wantTypes {
if output[i]["type"] != want {
t.Fatalf("output[%d] type = %v, want %s: %#v", i, output[i]["type"], want, output)
}
}
}
func TestWebSearchResponsesWriterStreamingLoopExhaustionFails(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
searches := 0
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) {
searches++
return &api.WebSearchResponse{}, nil
},
followUpStream: func(_ context.Context, _ []api.Message, _ api.Tools, yield func(api.ChatResponse) error) error {
call := api.ToolCall{ID: "call_next", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "again"})}}
if err := yield(api.ChatResponse{Message: api.Message{Role: "assistant", ToolCalls: []api.ToolCall{call}}}); err != nil {
return err
}
return yield(api.ChatResponse{Done: true})
},
}
initial := api.ChatResponse{Done: true, Message: api.Message{ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "first"})}}}}}
data, _ := json.Marshal(initial)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
body := recorder.Body.String()
if searches != maxWebSearchLoops || !strings.Contains(body, "response.failed") || !strings.Contains(body, "exceeded the maximum") || strings.Contains(body, "response.completed") {
t.Fatalf("loop exhaustion was not a terminal failure: searches=%d body=%s", searches, body)
}
}
func TestWebSearchResponsesWriterNonStreamingPreservesContentBeforeToolCall(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
request := openai.ResponsesRequest{Model: "test-model", Tools: []openai.ResponsesTool{{Type: "web_search"}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, model: request.Model, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(_ context.Context, query string) (*api.WebSearchResponse, error) {
return &api.WebSearchResponse{Results: []api.WebSearchResult{{Title: "Result", URL: "https://example.com", Content: "info"}}}, nil
},
followUpChat: func(_ context.Context, messages []api.Message, _ api.Tools) (api.ChatResponse, error) {
return api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "Here is the answer."}, Metrics: api.Metrics{PromptEvalCount: 7, EvalCount: 3}}, nil
},
}
// Non-streaming response with both content and a web_search tool call.
initial := api.ChatResponse{Done: true, Message: api.Message{Content: "Let me look this up.", ToolCalls: []api.ToolCall{{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "test"})}}}}, Metrics: api.Metrics{PromptEvalCount: 5, EvalCount: 2}}
data, err := json.Marshal(initial)
if err != nil {
t.Fatal(err)
}
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
var response openai.ResponsesResponse
if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
t.Fatalf("decode response: %v: %s", err, recorder.Body.String())
}
// Output should be: [pre-search message, web_search_call, final message]
if len(response.Output) != 3 {
t.Fatalf("output count = %d, want 3: %#v", len(response.Output), response.Output)
}
if response.Output[0].Type != "message" || response.Output[0].Content[0].Text != "Let me look this up." {
t.Fatalf("pre-search message = %#v", response.Output[0])
}
if response.Output[1].Type != "web_search_call" {
t.Fatalf("web_search_call = %#v", response.Output[1])
}
if response.Output[2].Type != "message" || response.Output[2].Content[0].Text != "Here is the answer." {
t.Fatalf("final message = %#v", response.Output[2])
}
}
func TestWebSearchResponsesWriterNonStreamingSurfacesMixedToolCalls(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
request := openai.ResponsesRequest{Model: "test-model", Tools: []openai.ResponsesTool{{Type: "web_search"}, {Type: "function", Name: "get_weather", Description: ptr("weather"), Parameters: map[string]any{"type": "object"}}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, model: request.Model, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) {
return &api.WebSearchResponse{}, nil
},
followUpChat: func(_ context.Context, messages []api.Message, _ api.Tools) (api.ChatResponse, error) {
// Verify the assistant message only contains the web_search tool call,
// not the get_weather tool call.
if len(messages) < 2 {
t.Fatalf("expected at least 2 messages, got %d", len(messages))
}
assistant := messages[len(messages)-2]
if len(assistant.ToolCalls) != 1 || assistant.ToolCalls[0].Function.Name != "web_search" {
t.Fatalf("assistant message should only have web_search tool call, got %#v", assistant.ToolCalls)
}
return api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "done"}}, nil
},
}
// Non-streaming response with both web_search and get_weather tool calls.
initial := api.ChatResponse{Done: true, Message: api.Message{ToolCalls: []api.ToolCall{
{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "weather"})}},
{ID: "call_2", Function: api.ToolCallFunction{Name: "get_weather", Arguments: testArgs(map[string]any{"city": "SF"})}},
}}}
data, _ := json.Marshal(initial)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
var response openai.ResponsesResponse
if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
t.Fatalf("decode response: %v: %s", err, recorder.Body.String())
}
// Output should include a function_call item for get_weather.
var hasFunctionCall bool
for _, item := range response.Output {
if item.Type == "function_call" && item.Name == "get_weather" {
hasFunctionCall = true
}
}
if !hasFunctionCall {
t.Fatalf("mixed tool call (get_weather) was not surfaced: %#v", response.Output)
}
}
func TestWebSearchResponsesWriterStreamingSurfacesMixedToolCalls(t *testing.T) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
ctx, _ := gin.CreateTestContext(recorder)
stream := true
request := openai.ResponsesRequest{Model: "test-model", Stream: &stream, Tools: []openai.ResponsesTool{{Type: "web_search"}, {Type: "function", Name: "get_weather", Description: ptr("weather"), Parameters: map[string]any{"type": "object"}}}}
inner := &ResponsesWriter{BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, converter: openai.NewResponsesStreamConverter("resp_test", "msg_test", request.Model, request), model: request.Model, stream: true, responseID: "resp_test", itemID: "msg_test", request: request}
writer := &WebSearchResponsesWriter{
BaseWriter: BaseWriter{ResponseWriter: ctx.Writer}, inner: inner, req: request,
chat: &api.ChatRequest{Model: request.Model, Tools: api.Tools{openai.WebSearchFunctionTool()}},
search: func(context.Context, string) (*api.WebSearchResponse, error) {
return &api.WebSearchResponse{}, nil
},
followUpChat: func(_ context.Context, messages []api.Message, _ api.Tools) (api.ChatResponse, error) {
assistant := messages[len(messages)-2]
if len(assistant.ToolCalls) != 1 || assistant.ToolCalls[0].Function.Name != "web_search" {
t.Fatalf("assistant message should only have web_search tool call, got %#v", assistant.ToolCalls)
}
return api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "done"}}, nil
},
}
// Streaming: initial response has both web_search and get_weather tool calls.
initial := api.ChatResponse{Done: true, Message: api.Message{ToolCalls: []api.ToolCall{
{ID: "call_1", Function: api.ToolCallFunction{Name: "web_search", Arguments: testArgs(map[string]any{"query": "weather"})}},
{ID: "call_2", Function: api.ToolCallFunction{Name: "get_weather", Arguments: testArgs(map[string]any{"city": "SF"})}},
}}}
data, _ := json.Marshal(initial)
if _, err := writer.Write(data); err != nil {
t.Fatal(err)
}
body := recorder.Body.String()
if !strings.Contains(body, "response.web_search_call.completed") {
t.Fatalf("missing web search event: %s", body)
}
if !strings.Contains(body, "response.function_call_arguments") {
t.Fatalf("mixed function call (get_weather) was not emitted: %s", body)
}
if !strings.Contains(body, "get_weather") {
t.Fatalf("get_weather function name not found: %s", body)
}
output := completedResponseOutput(t, body)
var hasFunctionCall, hasFinalMessage bool
for _, item := range output {
switch item["type"] {
case "function_call":
hasFunctionCall = item["name"] == "get_weather"
case "message":
content := item["content"].([]any)
part := content[0].(map[string]any)
hasFinalMessage = part["text"] == "done"
}
}
if !hasFunctionCall || !hasFinalMessage {
t.Fatalf("terminal output missing mixed call or final message: %#v", output)
}
}
func completedResponseOutput(t *testing.T, body string) []map[string]any {
t.Helper()
for _, block := range strings.Split(body, "\n\n") {
if !strings.HasPrefix(block, "event: response.completed\n") {
continue
}
dataAt := strings.Index(block, "\ndata: ")
if dataAt < 0 {
t.Fatalf("response.completed event has no data: %s", block)
}
var payload struct {
Response struct {
Output []map[string]any `json:"output"`
} `json:"response"`
}
if err := json.Unmarshal([]byte(block[dataAt+7:]), &payload); err != nil {
t.Fatalf("decode response.completed: %v: %s", err, block)
}
return payload.Response.Output
}
t.Fatalf("response.completed event not found: %s", body)
return nil
}
func ptr[T any](v T) *T { return &v }
+82
View File
@@ -0,0 +1,82 @@
package middleware
import (
"context"
"github.com/ollama/ollama/api"
)
const maxWebSearchLoops = 3
// doFollowUpChat sends a non-streaming /api/chat request with the accumulated
// messages and tools so the model can continue after a web search result.
func doFollowUpChat(ctx context.Context, base api.ChatRequest, messages []api.Message, tools api.Tools) (api.ChatResponse, error) {
stream := false
client, err := api.ClientFromEnvironment()
if err != nil {
return api.ChatResponse{}, err
}
var chatResponse api.ChatResponse
request := base
request.Messages = messages
request.Stream = &stream
request.Tools = tools
if err := client.Chat(ctx, &request, func(response api.ChatResponse) error {
chatResponse = response
return nil
}); err != nil {
return api.ChatResponse{}, err
}
return chatResponse, nil
}
// streamFollowUpChat streams the model response after a web search result.
func streamFollowUpChat(ctx context.Context, base api.ChatRequest, messages []api.Message, tools api.Tools, yield func(api.ChatResponse) error) error {
stream := true
client, err := api.ClientFromEnvironment()
if err != nil {
return err
}
request := base
request.Messages = messages
request.Stream = &stream
request.Tools = tools
return client.Chat(ctx, &request, yield)
}
func buildWebSearchAssistantMessage(response api.ChatResponse, webSearchCall api.ToolCall) api.Message {
assistant := api.Message{
Role: "assistant",
ToolCalls: []api.ToolCall{webSearchCall},
}
assistant.Content = response.Message.Content
assistant.Thinking = response.Message.Thinking
return assistant
}
func findWebSearchToolCall(toolCalls []api.ToolCall) (api.ToolCall, bool, bool) {
var webSearchCall api.ToolCall
var hasWebSearch, hasOtherTools bool
for _, toolCall := range toolCalls {
if toolCall.Function.Name == "web_search" {
if !hasWebSearch {
webSearchCall = toolCall
hasWebSearch = true
}
continue
}
hasOtherTools = true
}
return webSearchCall, hasWebSearch, hasOtherTools
}
func extractQueryFromToolCall(toolCall *api.ToolCall) string {
query, ok := toolCall.Function.Arguments.Get("query")
if !ok {
return ""
}
value, _ := query.(string)
return value
}
+152
View File
@@ -0,0 +1,152 @@
package middleware
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"testing"
"github.com/ollama/ollama/api"
)
func TestStreamFollowUpChat(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api/chat" {
t.Fatalf("path = %q", r.URL.Path)
}
var request api.ChatRequest
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
t.Fatal(err)
}
if request.Stream == nil || !*request.Stream {
t.Fatalf("stream = %#v, want true", request.Stream)
}
if string(request.Format) != `{"type":"object"}` || request.Think == nil || request.Think.Value != "high" {
t.Fatalf("follow-up controls were not preserved: format=%s think=%#v", request.Format, request.Think)
}
encoder := json.NewEncoder(w)
if err := encoder.Encode(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "one"}}); err != nil {
t.Fatal(err)
}
if err := encoder.Encode(api.ChatResponse{Done: true, Message: api.Message{Role: "assistant", Content: "two"}}); err != nil {
t.Fatal(err)
}
}))
defer server.Close()
t.Setenv("OLLAMA_HOST", server.URL)
var chunks []string
base := api.ChatRequest{Model: "test-model", Format: json.RawMessage(`{"type":"object"}`), Think: &api.ThinkValue{Value: "high"}}
if err := streamFollowUpChat(context.Background(), base, nil, nil, func(response api.ChatResponse) error {
chunks = append(chunks, response.Message.Content)
return nil
}); err != nil {
t.Fatal(err)
}
if len(chunks) != 2 || chunks[0] != "one" || chunks[1] != "two" {
t.Fatalf("chunks = %#v", chunks)
}
}
func TestFindWebSearchToolCall(t *testing.T) {
first := api.ToolCall{ID: "search_1", Function: api.ToolCallFunction{Name: "web_search"}}
calls := []api.ToolCall{
{ID: "client_1", Function: api.ToolCallFunction{Name: "get_weather"}},
first,
{ID: "search_2", Function: api.ToolCallFunction{Name: "web_search"}},
}
got, found, mixed := findWebSearchToolCall(calls)
if !found || !mixed || got.ID != first.ID {
t.Fatalf("call = %#v, found = %v, mixed = %v", got, found, mixed)
}
}
func TestExtractQueryFromToolCall(t *testing.T) {
tests := []struct {
name string
args api.ToolCallFunctionArguments
want string
}{
{name: "valid", args: webSearchTestArgs("query", "test search"), want: "test search"},
{name: "missing"},
{name: "wrong key", args: webSearchTestArgs("other", "value")},
{name: "wrong type", args: webSearchTestArgs("query", 42)},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
call := api.ToolCall{Function: api.ToolCallFunction{Name: "web_search", Arguments: test.args}}
if got := extractQueryFromToolCall(&call); got != test.want {
t.Fatalf("query = %q, want %q", got, test.want)
}
})
}
}
func webSearchTestArgs(key string, value any) api.ToolCallFunctionArguments {
args := api.NewToolCallFunctionArguments()
args.Set(key, value)
return args
}
func TestBuildWebSearchAssistantMessage(t *testing.T) {
call := api.ToolCall{ID: "search_1", Function: api.ToolCallFunction{Name: "web_search"}}
response := api.ChatResponse{Message: api.Message{Content: "searching", Thinking: "need current data"}}
message := buildWebSearchAssistantMessage(response, call)
if message.Role != "assistant" || message.Content != response.Message.Content || message.Thinking != response.Message.Thinking || len(message.ToolCalls) != 1 || message.ToolCalls[0].ID != call.ID {
t.Fatalf("message = %#v", message)
}
}
func TestDoFollowUpChatPreservesHTTPErrorTypes(t *testing.T) {
tests := []struct {
name string
status int
body string
check func(*testing.T, error)
}{
{
name: "authorization",
status: http.StatusUnauthorized,
body: `{"error":"unauthorized","signin_url":"https://ollama.com/signin/followup"}`,
check: func(t *testing.T, err error) {
var authorizationError api.AuthorizationError
if !errors.As(err, &authorizationError) || authorizationError.SigninURL != "https://ollama.com/signin/followup" {
t.Fatalf("error = %#v, want AuthorizationError with sign-in URL", err)
}
},
},
{
name: "rate limit",
status: http.StatusTooManyRequests,
body: `{"error":"slow down"}`,
check: func(t *testing.T, err error) {
var statusError api.StatusError
if !errors.As(err, &statusError) || statusError.StatusCode != http.StatusTooManyRequests {
t.Fatalf("error = %#v, want 429 StatusError", err)
}
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(test.status)
_, _ = w.Write([]byte(test.body))
}))
defer server.Close()
t.Setenv("OLLAMA_HOST", server.URL)
_, err := doFollowUpChat(context.Background(), api.ChatRequest{Model: "test-model"}, nil, nil)
if err == nil {
t.Fatal("expected error")
}
test.check(t, err)
})
}
}
+302 -89
View File
@@ -280,6 +280,12 @@ func unmarshalResponsesInputItem(data []byte) (ResponsesInputItem, error) {
return nil, err
}
return reasoning, nil
case "web_search_call":
var call ResponsesWebSearchCall
if err := json.Unmarshal(data, &call); err != nil {
return nil, err
}
return call, nil
default:
if itemType == "" {
return nil, fmt.Errorf("input item missing required 'type' field")
@@ -347,7 +353,7 @@ type ResponsesText struct {
// ResponsesTool represents a tool in the Responses API format.
// Note: This differs from api.Tool which nests fields under "function".
type ResponsesTool struct {
Type string `json:"type"` // "function" or "namespace"
Type string `json:"type"` // "function", "namespace", or "web_search"
Name string `json:"name"`
Description *string `json:"description"` // nullable but required
Strict *bool `json:"strict"` // nullable but required
@@ -430,7 +436,7 @@ func FromResponsesRequest(r ResponsesRequest) (*api.ChatRequest, error) {
// Track pending reasoning to merge with the next assistant message
var pendingThinking string
for _, item := range r.Input.Items {
for i, item := range r.Input.Items {
switch v := item.(type) {
case ResponsesReasoningInput:
// Store thinking to merge with the next assistant message
@@ -445,6 +451,39 @@ func FromResponsesRequest(r ResponsesRequest) (*api.ChatRequest, error) {
msg.Thinking = pendingThinking
pendingThinking = ""
}
// Responses may replay an assistant message between a function_call and
// its function_call_output. Keep those items in one Chat assistant message
// so the tool result immediately follows the call it answers.
var outputCallID string
if i+1 < len(r.Input.Items) {
if output, ok := r.Input.Items[i+1].(ResponsesFunctionCallOutput); ok {
outputCallID = output.CallID
}
}
if msg.Role == "assistant" && outputCallID != "" && len(messages) > 0 {
lastMsg := &messages[len(messages)-1]
if lastMsg.Role == "assistant" && len(lastMsg.ToolCalls) > 0 {
merged := false
for _, call := range lastMsg.ToolCalls {
if call.ID != outputCallID {
continue
}
if lastMsg.Content != "" && msg.Content != "" {
lastMsg.Content += "\n"
}
lastMsg.Content += msg.Content
lastMsg.Images = append(lastMsg.Images, msg.Images...)
if msg.Thinking != "" {
lastMsg.Thinking = msg.Thinking
}
merged = true
break
}
if merged {
continue
}
}
}
messages = append(messages, msg)
case ResponsesFunctionCall:
// Convert function call to assistant message with tool calls
@@ -497,6 +536,9 @@ func FromResponsesRequest(r ResponsesRequest) (*api.ChatRequest, error) {
Images: images,
ToolCallID: v.CallID,
})
case ResponsesWebSearchCall:
// Built-in tool calls are history metadata. The assistant message
// that follows carries the model-visible result of the prior search.
}
}
@@ -544,12 +586,24 @@ func FromResponsesRequest(r ResponsesRequest) (*api.ChatRequest, error) {
// Convert tools from Responses API format to api.Tool format
var tools []api.Tool
hasWebSearch := HasWebSearchTool(r.Tools)
for _, t := range r.Tools {
if isWebSearchTool(t) {
tools = append(tools, WebSearchFunctionTool())
continue
}
expanded, err := convertTools(t)
if err != nil {
return nil, err
}
tools = append(tools, expanded...)
for _, tool := range expanded {
// The built-in tool owns this name. Keeping a user-declared function
// with the same name makes a model call ambiguous.
if hasWebSearch && tool.Function.Name == "web_search" {
continue
}
tools = append(tools, tool)
}
}
// Handle text format (e.g. json_schema)
@@ -573,6 +627,39 @@ func FromResponsesRequest(r ResponsesRequest) (*api.ChatRequest, error) {
}, nil
}
func isWebSearchTool(t ResponsesTool) bool { return t.Type == "web_search" }
// HasWebSearchTool reports whether a request declares the built-in Responses
// web-search tool.
func HasWebSearchTool(tools []ResponsesTool) bool {
for _, tool := range tools {
if isWebSearchTool(tool) {
return true
}
}
return false
}
// WebSearchFunctionTool is the private function contract passed to local
// models. Responses clients must only ever see their original web_search
// declaration echoed back.
func WebSearchFunctionTool() api.Tool {
properties := api.NewToolPropertiesMap()
properties.Set("query", api.ToolProperty{Type: api.PropertyType{"string"}, Description: "The search query."})
return api.Tool{
Type: "function",
Function: api.ToolFunction{
Name: "web_search",
Description: "Search the web for current information.",
Parameters: api.ToolFunctionParameters{
Type: "object",
Required: []string{"query"},
Properties: properties,
},
},
}
}
// convertTools converts one Responses-API tool declaration to api.Tools. A
// "namespace" declaration groups member functions under a common name; it
// expands to those members with namespace-qualified names, since api.Tool
@@ -734,20 +821,49 @@ type ResponsesResponse struct {
}
type ResponsesOutputItem struct {
ID string `json:"id"`
Type string `json:"type"` // "message", "function_call", or "reasoning"
Status string `json:"status,omitempty"`
Role string `json:"role,omitempty"` // for message
Content []ResponsesOutputContent `json:"content,omitempty"` // for message
CallID string `json:"call_id,omitempty"` // for function_call
Name string `json:"name,omitempty"` // for function_call
Arguments string `json:"arguments,omitempty"` // for function_call
ID string `json:"id"`
Type string `json:"type"` // "message", "function_call", or "reasoning"
Status string `json:"status,omitempty"`
Role string `json:"role,omitempty"` // for message
Content []ResponsesOutputContent `json:"content,omitempty"` // for message
CallID string `json:"call_id,omitempty"` // for function_call
Name string `json:"name,omitempty"` // for function_call
Arguments string `json:"arguments,omitempty"` // for function_call
Action *ResponsesWebSearchAction `json:"action,omitempty"` // for web_search_call
// Reasoning fields
Summary []ResponsesReasoningSummary `json:"summary,omitempty"` // for reasoning
EncryptedContent string `json:"encrypted_content,omitempty"` // for reasoning
}
// ResponsesWebSearchCall is the native output item emitted for an executed
// built-in web search. It is intentionally separate from function calls: the
// internal web_search function is never exposed through the Responses API.
type ResponsesWebSearchCall struct {
ID string `json:"id"`
Type string `json:"type"` // always "web_search_call"
Status string `json:"status"`
Action *ResponsesWebSearchAction `json:"action,omitempty"`
}
func (ResponsesWebSearchCall) responsesInputItem() {}
type ResponsesWebSearchAction struct {
Type string `json:"type"` // always "search"
Query string `json:"query"`
}
// WebSearchCallOutputItem converts a web search call into the generic output
// item used by non-streaming Responses responses.
func WebSearchCallOutputItem(call ResponsesWebSearchCall) ResponsesOutputItem {
return ResponsesOutputItem{
ID: call.ID,
Type: call.Type,
Status: call.Status,
Action: call.Action,
}
}
type ResponsesReasoningSummary struct {
Type string `json:"type"` // "summary_text"
Text string `json:"text"`
@@ -942,10 +1058,9 @@ type ResponsesStreamConverter struct {
accumulatedThinking string
reasoningItemID string
reasoningStarted bool
reasoningDone bool
// Tool calls state (for final output)
toolCallItems []map[string]any
// Items completed before the final message, in streamed output-index order.
completedItems []any
}
// newEvent creates a ResponsesStreamEvent with the sequence number included in the data.
@@ -1165,45 +1280,54 @@ func (c *ResponsesStreamConverter) processThinking(thinking string) []ResponsesS
}
func (c *ResponsesStreamConverter) finishReasoning() []ResponsesStreamEvent {
if !c.reasoningStarted || c.reasoningDone {
if !c.reasoningStarted {
return nil
}
c.reasoningDone = true
itemID := c.reasoningItemID
thinking := c.accumulatedThinking
outputIndex := c.outputIndex
item := map[string]any{
"id": itemID,
"type": "reasoning",
"summary": []map[string]any{{"type": "summary_text", "text": thinking}},
"encrypted_content": thinking,
}
c.completedItems = append(c.completedItems, item)
c.accumulatedThinking = ""
c.reasoningItemID = ""
c.reasoningStarted = false
c.outputIndex++
events := []ResponsesStreamEvent{
c.newEvent("response.reasoning_summary_text.done", map[string]any{
"item_id": c.reasoningItemID,
"output_index": c.outputIndex,
"item_id": itemID,
"output_index": outputIndex,
"summary_index": 0,
"text": c.accumulatedThinking,
"text": thinking,
}),
c.newEvent("response.output_item.done", map[string]any{
"output_index": c.outputIndex,
"item": map[string]any{
"id": c.reasoningItemID,
"type": "reasoning",
"summary": []map[string]any{{"type": "summary_text", "text": c.accumulatedThinking}},
"encrypted_content": c.accumulatedThinking, // Plain text for now
},
"output_index": outputIndex,
"item": item,
}),
}
c.outputIndex++
return events
}
func (c *ResponsesStreamConverter) processToolCalls(toolCalls []api.ToolCall) []ResponsesStreamEvent {
return append(c.finishReasoning(), c.emitFunctionCallEvents(toolCalls)...)
}
// emitFunctionCallEvents emits function_call stream events for the given tool
// calls, stores them for the final output, and advances the output index.
func (c *ResponsesStreamConverter) emitFunctionCallEvents(toolCalls []api.ToolCall) []ResponsesStreamEvent {
var events []ResponsesStreamEvent
// Finish reasoning first if it was started
events = append(events, c.finishReasoning()...)
converted := ToToolCalls(toolCalls)
for i, tc := range converted {
outputIndex := c.outputIndex + i
fcItemID := fmt.Sprintf("fc_%d_%d", rand.Intn(999999), i)
// Store for final output (with status: completed)
toolCallItem := map[string]any{
"id": fcItemID,
"type": "function_call",
@@ -1212,52 +1336,157 @@ func (c *ResponsesStreamConverter) processToolCalls(toolCalls []api.ToolCall) []
"name": tc.Function.Name,
"arguments": tc.Function.Arguments,
}
c.toolCallItems = append(c.toolCallItems, toolCallItem)
c.completedItems = append(c.completedItems, toolCallItem)
// response.output_item.added for function call
events = append(events, c.newEvent("response.output_item.added", map[string]any{
"output_index": c.outputIndex + i,
"item": map[string]any{
"id": fcItemID,
"type": "function_call",
"status": "in_progress",
"call_id": tc.ID,
"name": tc.Function.Name,
"arguments": "",
},
}))
// response.function_call_arguments.delta
if tc.Function.Arguments != "" {
events = append(events, c.newEvent("response.function_call_arguments.delta", map[string]any{
events = append(events,
c.newEvent("response.output_item.added", map[string]any{
"output_index": outputIndex,
"item": map[string]any{
"id": fcItemID,
"type": "function_call",
"status": "in_progress",
"call_id": tc.ID,
"name": tc.Function.Name,
"arguments": "",
},
}),
c.newEvent("response.function_call_arguments.delta", map[string]any{
"item_id": fcItemID,
"output_index": c.outputIndex + i,
"output_index": outputIndex,
"delta": tc.Function.Arguments,
}))
}
}),
c.newEvent("response.function_call_arguments.done", map[string]any{
"item_id": fcItemID,
"output_index": outputIndex,
"arguments": tc.Function.Arguments,
}),
c.newEvent("response.output_item.done", map[string]any{
"output_index": outputIndex,
"item": toolCallItem,
}),
)
}
c.outputIndex += len(converted)
return events
}
// response.function_call_arguments.done
events = append(events, c.newEvent("response.function_call_arguments.done", map[string]any{
"item_id": fcItemID,
"output_index": c.outputIndex + i,
"arguments": tc.Function.Arguments,
}))
// response.output_item.done for function call
events = append(events, c.newEvent("response.output_item.done", map[string]any{
"output_index": c.outputIndex + i,
// StartWebSearchCall emits the events that precede a server-side search and
// reserves its position in the response output.
func (c *ResponsesStreamConverter) StartWebSearchCall(call ResponsesWebSearchCall) (int, []ResponsesStreamEvent) {
events := c.finishReasoning()
outputIndex := c.outputIndex
c.outputIndex++
events = append(events,
c.newEvent("response.output_item.added", map[string]any{
"output_index": outputIndex,
"item": map[string]any{
"id": fcItemID,
"type": "function_call",
"status": "completed",
"call_id": tc.ID,
"name": tc.Function.Name,
"arguments": tc.Function.Arguments,
"id": call.ID, "type": "web_search_call", "status": "in_progress",
"action": map[string]any{"type": call.Action.Type, "query": call.Action.Query},
},
}))
}),
c.newEvent("response.web_search_call.in_progress", map[string]any{
"item_id": call.ID, "output_index": outputIndex,
}),
c.newEvent("response.web_search_call.searching", map[string]any{
"item_id": call.ID, "output_index": outputIndex,
}),
)
return outputIndex, events
}
// FinishWebSearchCall emits the events that follow a successful server-side search.
func (c *ResponsesStreamConverter) FinishWebSearchCall(call ResponsesWebSearchCall, outputIndex int) []ResponsesStreamEvent {
item := webSearchCallMap(call)
c.completedItems = append(c.completedItems, item)
return []ResponsesStreamEvent{
c.newEvent("response.web_search_call.completed", map[string]any{
"item_id": call.ID, "output_index": outputIndex,
}),
c.newEvent("response.output_item.done", map[string]any{
"output_index": outputIndex,
"item": item,
}),
}
}
// ResponseFailed emits a terminal failure using this stream's sequence counter.
func (c *ResponsesStreamConverter) ResponseFailed(response map[string]any) ResponsesStreamEvent {
return c.newEvent("response.failed", map[string]any{"response": response})
}
func webSearchCallMap(call ResponsesWebSearchCall) map[string]any {
item := map[string]any{
"id": call.ID, "type": "web_search_call", "status": call.Status,
}
if call.Action != nil {
item["action"] = map[string]any{"type": call.Action.Type, "query": call.Action.Query}
}
return item
}
// FinishMessageItem closes the current text message item (if one was started)
// and reserves its place in the final output. This allows pre-search content
// to be emitted as a distinct message item before web_search_call events.
func (c *ResponsesStreamConverter) FinishMessageItem() []ResponsesStreamEvent {
if !c.contentStarted {
return nil
}
return events
c.contentStarted = false
text := c.accumulatedText
c.accumulatedText = ""
c.contentIndex = 0
itemID := c.itemID
item := map[string]any{
"id": itemID,
"type": "message",
"status": "completed",
"role": "assistant",
"content": []map[string]any{{
"type": "output_text",
"text": text,
"annotations": []any{},
"logprobs": []any{},
}},
}
c.completedItems = append(c.completedItems, item)
outputIndex := c.outputIndex
c.outputIndex++
c.itemID = fmt.Sprintf("msg_%s_%d", strings.TrimPrefix(c.responseID, "resp_"), c.outputIndex)
return []ResponsesStreamEvent{
c.newEvent("response.output_text.done", map[string]any{
"item_id": itemID,
"output_index": outputIndex,
"content_index": 0,
"text": text,
"logprobs": []any{},
}),
c.newEvent("response.content_part.done", map[string]any{
"item_id": itemID,
"output_index": outputIndex,
"content_index": 0,
"part": map[string]any{
"type": "output_text",
"text": text,
"annotations": []any{},
"logprobs": []any{},
},
}),
c.newEvent("response.output_item.done", map[string]any{
"output_index": outputIndex,
"item": item,
}),
}
}
// EmitFunctionCallItems emits function_call events for client-provided tool
// calls that accompanied a web_search call (mixed responses). Unlike
// processToolCalls, this does not set toolCallsSent, so subsequent text
// content can still be processed.
func (c *ResponsesStreamConverter) EmitFunctionCallItems(toolCalls []api.ToolCall) []ResponsesStreamEvent {
return c.emitFunctionCallEvents(toolCalls)
}
func (c *ResponsesStreamConverter) processTextContent(content string) []ResponsesStreamEvent {
@@ -1312,24 +1541,8 @@ func (c *ResponsesStreamConverter) processTextContent(content string) []Response
}
func (c *ResponsesStreamConverter) buildFinalOutput() []any {
var output []any
// Add reasoning item if present
if c.reasoningStarted {
output = append(output, map[string]any{
"id": c.reasoningItemID,
"type": "reasoning",
"summary": []map[string]any{{"type": "summary_text", "text": c.accumulatedThinking}},
"encrypted_content": c.accumulatedThinking,
})
}
// Add tool calls if present
if len(c.toolCallItems) > 0 {
for _, item := range c.toolCallItems {
output = append(output, item)
}
} else if c.contentStarted {
output := append([]any(nil), c.completedItems...)
if c.contentStarted {
// Add message item if we had text content
output = append(output, map[string]any{
"id": c.itemID,
+287
View File
@@ -248,6 +248,21 @@ func TestUnmarshalResponsesInputItem(t *testing.T) {
}
})
t.Run("web_search_call item", func(t *testing.T) {
got, err := unmarshalResponsesInputItem([]byte(`{"type":"web_search_call","id":"ws_123","status":"completed","action":{"type":"search","query":"Parth Sareen"}}`))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
call, ok := got.(ResponsesWebSearchCall)
if !ok {
t.Fatalf("got type %T, want ResponsesWebSearchCall", got)
}
if call.ID != "ws_123" || call.Action == nil || call.Action.Query != "Parth Sareen" {
t.Fatalf("call = %#v", call)
}
})
t.Run("unknown item type", func(t *testing.T) {
_, err := unmarshalResponsesInputItem([]byte(`{"type": "unknown_type"}`))
if err == nil {
@@ -279,6 +294,67 @@ func TestUnmarshalResponsesInputItem(t *testing.T) {
})
}
func TestFromResponsesRequestIgnoresReplayedWebSearchCall(t *testing.T) {
req := ResponsesRequest{
Model: "test",
Input: ResponsesInput{Items: []ResponsesInputItem{
ResponsesInputMessage{Type: "message", Role: "user", Content: []ResponsesContent{ResponsesTextContent{Type: "input_text", Text: "Who is Parth Sareen?"}}},
ResponsesWebSearchCall{ID: "ws_123", Type: "web_search_call", Status: "completed", Action: &ResponsesWebSearchAction{Type: "search", Query: "Parth Sareen"}},
ResponsesInputMessage{Type: "message", Role: "assistant", Content: []ResponsesContent{ResponsesOutputTextContent{Type: "output_text", Text: "He works at Ollama."}}},
ResponsesInputMessage{Type: "message", Role: "user", Content: []ResponsesContent{ResponsesTextContent{Type: "input_text", Text: "What do you think of him?"}}},
}},
}
chat, err := FromResponsesRequest(req)
if err != nil {
t.Fatal(err)
}
if len(chat.Messages) != 3 {
t.Fatalf("messages = %#v", chat.Messages)
}
if chat.Messages[0].Role != "user" || chat.Messages[1].Role != "assistant" || chat.Messages[1].Content != "He works at Ollama." || chat.Messages[2].Role != "user" {
t.Fatalf("messages = %#v", chat.Messages)
}
}
func TestFromResponsesRequestMergesMessageAfterFunctionCall(t *testing.T) {
var req ResponsesRequest
err := json.Unmarshal([]byte(`{
"model": "kimi-k3:cloud",
"input": [
{"role": "user", "content": "Find Ollama and inspect the current directory."},
{"type": "web_search_call", "id": "ws_test", "status": "completed", "action": {"type": "search", "query": "Ollama"}},
{"type": "function_call", "call_id": "call_test", "name": "exec_command", "arguments": "{\"cmd\":\"pwd\"}"},
{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "I found Ollama and will inspect the directory."}]},
{"type": "function_call_output", "call_id": "call_test", "output": "/tmp"}
]
}`), &req)
if err != nil {
t.Fatal(err)
}
chat, err := FromResponsesRequest(req)
if err != nil {
t.Fatal(err)
}
if len(chat.Messages) != 3 {
t.Fatalf("messages = %#v", chat.Messages)
}
assistant := chat.Messages[1]
if assistant.Role != "assistant" || assistant.Content != "I found Ollama and will inspect the directory." || len(assistant.ToolCalls) != 1 {
t.Fatalf("assistant message = %#v", assistant)
}
if assistant.ToolCalls[0].ID != "call_test" || assistant.ToolCalls[0].Function.Name != "exec_command" {
t.Fatalf("tool call = %#v", assistant.ToolCalls[0])
}
tool := chat.Messages[2]
if tool.Role != "tool" || tool.ToolCallID != "call_test" || tool.Content != "/tmp" {
t.Fatalf("tool message = %#v", tool)
}
}
func TestResponsesRequest_UnmarshalJSON(t *testing.T) {
tests := []struct {
name string
@@ -415,6 +491,61 @@ func TestFromResponsesRequest_Tools(t *testing.T) {
}
}
func TestFromResponsesRequest_WebSearchTool(t *testing.T) {
var req ResponsesRequest
if err := json.Unmarshal([]byte(`{
"model":"gpt-oss:20b", "input":"latest news",
"tools":[
{"type":"web_search"},
{"type":"function", "name":"web_search", "description":"client collision", "parameters":{"type":"object"}},
{"type":"function", "name":"weather", "description":"weather", "parameters":{"type":"object"}}
]
}`), &req); err != nil {
t.Fatal(err)
}
if !HasWebSearchTool(req.Tools) {
t.Fatal("built-in web_search was not detected")
}
chat, err := FromResponsesRequest(req)
if err != nil {
t.Fatal(err)
}
if len(chat.Tools) != 2 {
t.Fatalf("converted tool count = %d, want 2", len(chat.Tools))
}
if chat.Tools[0].Function.Name != "web_search" {
t.Errorf("first tool = %q, want web_search", chat.Tools[0].Function.Name)
}
if chat.Tools[0].Function.Parameters.Type != "object" || len(chat.Tools[0].Function.Parameters.Required) != 1 || chat.Tools[0].Function.Parameters.Required[0] != "query" {
t.Errorf("web_search parameters = %#v, want required query object", chat.Tools[0].Function.Parameters)
}
if chat.Tools[1].Function.Name != "weather" {
t.Errorf("second tool = %q, want weather", chat.Tools[1].Function.Name)
}
}
func TestFromResponsesRequest_WebSearchIgnoresUnknownControls(t *testing.T) {
for _, declaration := range []string{
`{"type":"web_search","filters":{"allowed_domains":["example.com"]}}`,
`{"type":"web_search","user_location":{"type":"approximate"}}`,
`{"type":"web_search","search_context_size":"high"}`,
} {
t.Run(declaration, func(t *testing.T) {
var tool ResponsesTool
if err := json.Unmarshal([]byte(declaration), &tool); err != nil {
t.Fatal(err)
}
request, err := FromResponsesRequest(ResponsesRequest{Model: "test", Input: ResponsesInput{Text: "hi"}, Tools: []ResponsesTool{tool}})
if err != nil {
t.Fatal(err)
}
if len(request.Tools) != 1 || request.Tools[0].Function.Name != "web_search" {
t.Fatalf("tools = %#v, want web_search", request.Tools)
}
})
}
}
// TestFromResponsesRequest_NamespaceTools covers the "namespace" tool
// declaration: one namespace whose members are the real functions must
// expand to namespace-qualified function tools with their schemas intact,
@@ -2140,3 +2271,159 @@ func TestResponsesStreamConverter_FunctionCallStatus(t *testing.T) {
t.Errorf("output_item.done status = %q, want %q", doneItem["status"], "completed")
}
}
func TestResponsesStreamConverter_WebSearchCall(t *testing.T) {
converter := NewResponsesStreamConverter("resp_123", "msg_456", "gpt-oss:20b", ResponsesRequest{})
call := ResponsesWebSearchCall{
ID: "ws_123",
Type: "web_search_call",
Status: "completed",
Action: &ResponsesWebSearchAction{
Type: "search",
Query: "Ollama news",
},
}
outputIndex, events := converter.StartWebSearchCall(call)
events = append(events, converter.FinishWebSearchCall(call, outputIndex)...)
wantTypes := []string{
"response.output_item.added",
"response.web_search_call.in_progress",
"response.web_search_call.searching",
"response.web_search_call.completed",
"response.output_item.done",
}
if len(events) != len(wantTypes) {
t.Fatalf("event count = %d, want %d", len(events), len(wantTypes))
}
for i, event := range events {
if event.Event != wantTypes[i] {
t.Errorf("event[%d] = %q, want %q", i, event.Event, wantTypes[i])
}
if got := event.Data.(map[string]any)["sequence_number"]; got != i {
t.Errorf("event[%d] sequence = %v, want %d", i, got, i)
}
}
done := events[len(events)-1].Data.(map[string]any)["item"].(map[string]any)
action := done["action"].(map[string]any)
if action["query"] != "Ollama news" {
t.Errorf("done query = %q", action["query"])
}
}
func TestResponsesStreamConverter_FinishMessageItem(t *testing.T) {
converter := NewResponsesStreamConverter("resp_123", "msg_456", "gpt-oss:20b", ResponsesRequest{})
// Process some text content first
textEvents := converter.Process(api.ChatResponse{Message: api.Message{Role: "assistant", Content: "pre-search text"}})
if len(textEvents) == 0 {
t.Fatal("expected events from Process")
}
// FinishMessageItem should close the message item
finishEvents := converter.FinishMessageItem()
wantTypes := []string{
"response.output_text.done",
"response.content_part.done",
"response.output_item.done",
}
if len(finishEvents) != len(wantTypes) {
t.Fatalf("event count = %d, want %d", len(finishEvents), len(wantTypes))
}
for i, event := range finishEvents {
if event.Event != wantTypes[i] {
t.Errorf("event[%d] = %q, want %q", i, event.Event, wantTypes[i])
}
}
// buildFinalOutput should include the completed message item
output := converter.buildFinalOutput()
if len(output) != 1 {
t.Fatalf("output count = %d, want 1", len(output))
}
item := output[0].(map[string]any)
if item["type"] != "message" {
t.Fatalf("output type = %v, want message", item["type"])
}
content := item["content"].([]map[string]any)
if content[0]["text"] != "pre-search text" {
t.Fatalf("text = %v, want 'pre-search text'", content[0]["text"])
}
preSearchID := item["id"]
// Calling FinishMessageItem again without content should be a no-op
if events := converter.FinishMessageItem(); len(events) != 0 {
t.Fatalf("expected no events, got %d", len(events))
}
// A later model leg must use a fresh message item ID, and both message
// items must be present in the terminal response output.
finalEvents := converter.Process(api.ChatResponse{
Message: api.Message{Role: "assistant", Content: "final text"},
Done: true,
})
var finalMessageID any
var finalOutput []any
for _, event := range finalEvents {
data := event.Data.(map[string]any)
switch event.Event {
case "response.output_item.added":
added := data["item"].(map[string]any)
if added["type"] == "message" {
finalMessageID = added["id"]
}
case "response.completed":
response := data["response"].(map[string]any)
finalOutput = response["output"].([]any)
}
}
if finalMessageID == nil || finalMessageID == preSearchID {
t.Fatalf("message item IDs were not rotated: pre-search=%v final=%v", preSearchID, finalMessageID)
}
if len(finalOutput) != 2 {
t.Fatalf("terminal output count = %d, want 2: %#v", len(finalOutput), finalOutput)
}
if got := finalOutput[1].(map[string]any)["id"]; got != finalMessageID {
t.Fatalf("terminal final message ID = %v, want %v", got, finalMessageID)
}
}
func TestResponsesStreamConverter_FinalOutputKeepsStreamedItemOrder(t *testing.T) {
converter := NewResponsesStreamConverter("resp_order", "msg_order", "test-model", ResponsesRequest{})
var events []ResponsesStreamEvent
events = append(events, converter.Process(api.ChatResponse{Message: api.Message{Thinking: "before search"}})...)
call := ResponsesWebSearchCall{
ID: "ws_order_1",
Type: "web_search_call",
Status: "completed",
Action: &ResponsesWebSearchAction{Type: "search", Query: "test"},
}
outputIndex, searchEvents := converter.StartWebSearchCall(call)
events = append(events, searchEvents...)
events = append(events, converter.FinishWebSearchCall(call, outputIndex)...)
events = append(events, converter.Process(api.ChatResponse{Message: api.Message{Thinking: "after search"}})...)
events = append(events, converter.Process(api.ChatResponse{Message: api.Message{Content: "answer"}, Done: true})...)
doneIDs := map[int]any{}
var finalOutput []any
for _, event := range events {
data := event.Data.(map[string]any)
switch event.Event {
case "response.output_item.done":
item := data["item"].(map[string]any)
doneIDs[data["output_index"].(int)] = item["id"]
case "response.completed":
response := data["response"].(map[string]any)
finalOutput = response["output"].([]any)
}
}
if len(finalOutput) != 4 {
t.Fatalf("terminal output count = %d, want 4: %#v", len(finalOutput), finalOutput)
}
for outputIndex, raw := range finalOutput {
item := raw.(map[string]any)
if got, want := item["id"], doneIDs[outputIndex]; got != want {
t.Fatalf("terminal output[%d] ID = %v, streamed done ID = %v; output=%#v", outputIndex, got, want, finalOutput)
}
}
}
+27 -15
View File
@@ -21,15 +21,16 @@ import (
"github.com/ollama/ollama/auth"
"github.com/ollama/ollama/envconfig"
internalcloud "github.com/ollama/ollama/internal/cloud"
"github.com/ollama/ollama/openai"
"github.com/ollama/ollama/version"
)
const (
defaultCloudProxyBaseURL = "https://ollama.com:443"
defaultCloudProxySigningHost = "ollama.com"
cloudProxyBaseURLEnv = "OLLAMA_CLOUD_BASE_URL"
legacyCloudAnthropicKey = "legacy_cloud_anthropic_web_search"
cloudProxyClientVersionHeader = "X-Ollama-Client-Version"
defaultCloudProxyBaseURL = "https://ollama.com:443"
defaultCloudProxySigningHost = "ollama.com"
cloudProxyBaseURLEnv = "OLLAMA_CLOUD_BASE_URL"
cloudWebSearchOrchestrationKey = "cloud_web_search_orchestration"
cloudProxyClientVersionHeader = "X-Ollama-Client-Version"
// maxDecompressedBodySize limits the size of a decompressed request body
maxDecompressedBodySize = 20 << 20
@@ -120,14 +121,14 @@ func cloudPassthroughMiddleware(disabledOperation string) gin.HandlerFunc {
return
}
// TEMP(drifkin): keep Anthropic web search requests on the local middleware
// path so WebSearchAnthropicWriter can orchestrate follow-up calls.
if c.Request.URL.Path == "/v1/messages" {
if hasAnthropicWebSearchTool(body) {
c.Set(legacyCloudAnthropicKey, true)
c.Next()
return
}
// Keep server-side web search on the local compatibility middleware path.
// The converted model requests use Ollama's /api/chat contract, including
// for cloud models; all other cloud compatibility traffic remains raw
// passthrough.
if hasWebSearchTool(c.Request.URL.Path, body) {
c.Set(cloudWebSearchOrchestrationKey, true)
c.Next()
return
}
proxyCloudRequest(c, normalizedBody, disabledOperation)
@@ -234,7 +235,7 @@ func proxyCloudRequestWithPath(c *gin.Context, body []byte, path string, disable
// into WebSearchAnthropicWriter, but this proxy copy loop may coalesce
// multiple jsonl records into one Write. WebSearchAnthropicWriter currently
// unmarshals one JSON value per Write.
if path == "/api/chat" && resp.StatusCode == http.StatusOK && c.GetBool(legacyCloudAnthropicKey) {
if path == "/api/chat" && resp.StatusCode == http.StatusOK && c.GetBool(cloudWebSearchOrchestrationKey) {
framedWriter = &jsonlFramingResponseWriter{ResponseWriter: c.Writer}
bodyWriter = framedWriter
}
@@ -324,11 +325,22 @@ func extractModelField(body []byte) (string, bool) {
return model, model != ""
}
func hasAnthropicWebSearchTool(body []byte) bool {
func hasWebSearchTool(path string, body []byte) bool {
if len(body) == 0 {
return false
}
if path == "/v1/responses" {
var payload struct {
Tools []openai.ResponsesTool `json:"tools"`
}
return json.Unmarshal(body, &payload) == nil && openai.HasWebSearchTool(payload.Tools)
}
if path != "/v1/messages" {
return false
}
var payload struct {
Tools []struct {
Type string `json:"type"`
+6 -1
View File
@@ -2144,6 +2144,11 @@ func (s *Server) WebFetchExperimentalHandler(c *gin.Context) {
}
func (s *Server) webExperimentalProxyHandler(c *gin.Context, proxyPath, disabledOperation string) {
// This endpoint is authenticated by the server's cloud signature. A client
// may have supplied an unrelated provider credential (for example, Codex's
// Responses API key); it must not be sent to the web-search service.
c.Request.Header.Del("Authorization")
body, err := readRequestBody(c.Request)
if err != nil {
c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": err.Error()})
@@ -2439,7 +2444,7 @@ func (s *Server) ChatHandler(c *gin.Context) {
if modelRef.Source == modelSourceCloud {
req.Model = modelRef.Base
if c.GetBool(legacyCloudAnthropicKey) {
if c.GetBool(cloudWebSearchOrchestrationKey) {
proxyCloudJSONRequestWithPath(c, req, "/api/chat", cloudErrRemoteInferenceUnavailable)
return
}
+152
View File
@@ -6,6 +6,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
@@ -797,6 +798,157 @@ func TestExplicitCloudPassthroughAPIAndV1(t *testing.T) {
})
}
func TestCloudResponsesWebSearchUsesLocalOrchestration(t *testing.T) {
gin.SetMode(gin.TestMode)
setTestHome(t, t.TempDir())
chatCalls := 0
searchCalls := 0
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/api/chat":
chatCalls++
w.Header().Set("Content-Type", "application/x-ndjson")
if chatCalls == 1 {
_, _ = io.WriteString(w, `{"message":{"role":"assistant","tool_calls":[{"id":"call_1","function":{"name":"web_search","arguments":{"query":"latest Ollama release"}}}]},"done":false}`+"\n")
_, _ = io.WriteString(w, `{"message":{"role":"assistant"},"done":true,"prompt_eval_count":12,"eval_count":4}`+"\n")
return
}
_, _ = io.WriteString(w, `{"message":{"role":"assistant","content":"Ollama [release](https://ollama.com/release)."},"done":true,"prompt_eval_count":20,"eval_count":6}`)
case "/api/web_search":
searchCalls++
w.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(w, `{"results":[{"title":"Ollama release","url":"https://ollama.com/release","content":"current release"}]}`)
default:
t.Fatalf("unexpected upstream path %q", r.URL.Path)
}
}))
defer upstream.Close()
originalBaseURL := cloudProxyBaseURL
originalSignRequest := cloudProxySignRequest
cloudProxyBaseURL = upstream.URL
cloudProxySignRequest = func(context.Context, *http.Request) error { return nil }
t.Cleanup(func() {
cloudProxyBaseURL = originalBaseURL
cloudProxySignRequest = originalSignRequest
})
s := &Server{}
router, err := s.GenerateRoutes()
if err != nil {
t.Fatal(err)
}
local := httptest.NewServer(router)
defer local.Close()
t.Setenv("OLLAMA_HOST", local.URL)
reqBody := `{
"model":"kimi-k2.5:cloud",
"input":"Find the latest Ollama release",
"stream":true,
"tools":[{"type":"web_search","external_web_access":false}]
}`
req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, local.URL+"/v1/responses", bytes.NewBufferString(reqBody))
if err != nil {
t.Fatal(err)
}
req.Header.Set("Content-Type", "application/json")
resp, err := local.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
if resp.StatusCode != http.StatusOK {
t.Fatalf("status = %d, want 200: %s", resp.StatusCode, body)
}
if chatCalls != 2 || searchCalls != 1 {
t.Fatalf("chat calls = %d, search calls = %d; want 2 and 1", chatCalls, searchCalls)
}
if !bytes.Contains(body, []byte("response.web_search_call.completed")) || !bytes.Contains(body, []byte("https://ollama.com/release")) {
t.Fatalf("missing native web search result and citation: %s", body)
}
if bytes.Contains(body, []byte("response.function_call_arguments")) || bytes.Contains(body, []byte(`"type":"function_call"`)) {
t.Fatalf("private web_search function leaked: %s", body)
}
}
func TestCloudResponsesUnsupportedWebSearchPassthrough(t *testing.T) {
gin.SetMode(gin.TestMode)
setTestHome(t, t.TempDir())
type upstreamCapture struct {
path string
body string
}
capture := &upstreamCapture{}
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
capture.path = r.URL.Path
body, err := io.ReadAll(r.Body)
if err != nil {
t.Fatal(err)
}
capture.body = string(body)
w.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(w, `{"id":"resp_1","object":"response"}`)
}))
defer upstream.Close()
originalBaseURL := cloudProxyBaseURL
originalSignRequest := cloudProxySignRequest
cloudProxyBaseURL = upstream.URL
cloudProxySignRequest = func(context.Context, *http.Request) error { return nil }
t.Cleanup(func() {
cloudProxyBaseURL = originalBaseURL
cloudProxySignRequest = originalSignRequest
})
router := gin.New()
router.POST(
"/v1/responses",
cloudPassthroughMiddleware(cloudErrRemoteInferenceUnavailable),
middleware.ResponsesMiddleware(),
func(c *gin.Context) { c.Status(http.StatusTeapot) },
)
local := httptest.NewServer(router)
defer local.Close()
for _, toolType := range []string{"web_search_preview", "web_search_invalid"} {
t.Run(toolType, func(t *testing.T) {
capture.path = ""
capture.body = ""
reqBody := fmt.Sprintf(`{
"model":"kimi-k2.5:cloud",
"input":"Find the latest Ollama release",
"tools":[{"type":%q}]
}`, toolType)
req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, local.URL+"/v1/responses", bytes.NewBufferString(reqBody))
if err != nil {
t.Fatal(err)
}
req.Header.Set("Content-Type", "application/json")
resp, err := local.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
t.Fatalf("status = %d, want %d: %s", resp.StatusCode, http.StatusOK, body)
}
if capture.path != "/v1/responses" {
t.Fatalf("upstream path = %q, want /v1/responses", capture.path)
}
if !strings.Contains(capture.body, fmt.Sprintf(`"type":%q`, toolType)) {
t.Fatalf("unsupported web-search tool was not passed through: %s", capture.body)
}
})
}
}
func TestCloudDisabledBlocksExplicitCloudPassthrough(t *testing.T) {
gin.SetMode(gin.TestMode)
setTestHome(t, t.TempDir())
+48 -2
View File
@@ -111,8 +111,8 @@ func TestExperimentalWebEndpointsPassthrough(t *testing.T) {
if !bytes.Contains([]byte(capture.body), []byte(tt.assertBody)) {
t.Fatalf("expected upstream body to contain %q, got %q", tt.assertBody, capture.body)
}
if got := capture.header.Get("Authorization"); got != "Bearer should-forward" {
t.Fatalf("expected forwarded Authorization header, got %q", got)
if got := capture.header.Get("Authorization"); got != "" {
t.Fatalf("expected Authorization header to be stripped, got %q", got)
}
if got := capture.header.Get("X-Test-Header"); got != "web-experimental" {
t.Fatalf("expected forwarded X-Test-Header=web-experimental, got %q", got)
@@ -124,6 +124,52 @@ func TestExperimentalWebEndpointsPassthrough(t *testing.T) {
}
}
func TestExperimentalWebEndpointPreservesUpstreamRateLimit(t *testing.T) {
gin.SetMode(gin.TestMode)
setTestHome(t, t.TempDir())
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if got := r.Header.Get("Authorization"); got != "" {
t.Fatalf("unexpected forwarded Authorization header: %q", got)
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusTooManyRequests)
_, _ = w.Write([]byte(`{"error":"rate limit exceeded"}`))
}))
defer upstream.Close()
original := cloudProxyBaseURL
cloudProxyBaseURL = upstream.URL
t.Cleanup(func() { cloudProxyBaseURL = original })
s := &Server{}
router, err := s.GenerateRoutes()
if err != nil {
t.Fatal(err)
}
local := httptest.NewServer(router)
defer local.Close()
req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, local.URL+"/api/experimental/web_search", bytes.NewBufferString(`{"query":"hello"}`))
if err != nil {
t.Fatal(err)
}
req.Header.Set("Authorization", "Bearer codex-credential")
resp, err := local.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
if resp.StatusCode != http.StatusTooManyRequests {
t.Fatalf("status = %d, want 429 (%s)", resp.StatusCode, body)
}
if string(body) != `{"error":"rate limit exceeded"}` {
t.Fatalf("body = %s", body)
}
}
func TestExperimentalWebEndpointsMissingBody(t *testing.T) {
gin.SetMode(gin.TestMode)
setTestHome(t, t.TempDir())