From 3f77cb6dfb6f6ae6d2ab132fa32d09414efd1279 Mon Sep 17 00:00:00 2001 From: Parth Sareen Date: Fri, 4 Sep 2026 13:07:33 -0700 Subject: [PATCH] openai: add Codex compaction support (#18224) --- openai/responses.go | 17 + openai/responses_compact.go | 699 +++++++++++++++++++++++++++++++ openai/responses_compact_test.go | 313 ++++++++++++++ server/responses_compact.go | 219 ++++++++++ server/responses_compact_test.go | 282 +++++++++++++ server/routes.go | 3 +- 6 files changed, 1532 insertions(+), 1 deletion(-) create mode 100644 openai/responses_compact.go create mode 100644 openai/responses_compact_test.go create mode 100644 server/responses_compact.go create mode 100644 server/responses_compact_test.go diff --git a/openai/responses.go b/openai/responses.go index b5a739d2b..c08dd3cb2 100644 --- a/openai/responses.go +++ b/openai/responses.go @@ -2,6 +2,7 @@ package openai import ( "encoding/json" + "errors" "fmt" "math/rand" "strings" @@ -323,6 +324,18 @@ func unmarshalResponsesInputItem(data []byte) (ResponsesInputItem, error) { return nil, err } return call, nil + case "compaction": + var compaction ResponsesCompactionItem + if err := json.Unmarshal(data, &compaction); err != nil { + return nil, err + } + return compaction, nil + case "compaction_trigger": + var trigger ResponsesCompactionTrigger + if err := json.Unmarshal(data, &trigger); err != nil { + return nil, err + } + return trigger, nil default: if itemType == "" { return nil, fmt.Errorf("input item missing required 'type' field") @@ -594,6 +607,10 @@ func FromResponsesRequest(r ResponsesRequest) (*api.ChatRequest, error) { case ResponsesWebSearchCall: // Built-in tool calls are history metadata. The assistant message // that follows carries the model-visible result of the prior search. + case ResponsesCompactionItem: + return nil, errors.New("compaction items must be expanded before Responses conversion") + case ResponsesCompactionTrigger: + return nil, errors.New("compaction_trigger must be handled before Responses conversion") } } diff --git a/openai/responses_compact.go b/openai/responses_compact.go new file mode 100644 index 000000000..f8c7e9de7 --- /dev/null +++ b/openai/responses_compact.go @@ -0,0 +1,699 @@ +package openai + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "strings" + "time" + + "github.com/ollama/ollama/api" +) + +const ( + OllamaCompactionPayloadType = "ollama_compaction" + OllamaCompactionPayloadVersion = 1 + CreateSummaryToolName = "create_summary" + compactionSummaryToolName = "ollama_compaction_summary" +) + +// ResponsesCompactionTrigger is the terminal input item sent by current Codex +// clients when they request remote compaction through POST /v1/responses. +type ResponsesCompactionTrigger struct { + Type string `json:"type"` +} + +func (ResponsesCompactionTrigger) responsesInputItem() {} + +// ResponsesCompactionItem is the opaque continuation item understood by Codex. +// Ollama stores a versioned JSON payload in EncryptedContent; the payload is not +// encrypted and must be consumed by Ollama before another model request. +type ResponsesCompactionItem struct { + Type string `json:"type"` + EncryptedContent string `json:"encrypted_content"` +} + +func (ResponsesCompactionItem) responsesInputItem() {} + +// OllamaCompactionPayload is Ollama's stateless continuation format. +type OllamaCompactionPayload struct { + Type string `json:"type"` + Version int `json:"version"` + Summary string `json:"summary"` + Retained []api.Message `json:"retained"` +} + +// CompactionTranscriptItem is one ordered input item shown to the compaction +// model. Ref is request-local and is the only value the model may select. +type CompactionTranscriptItem struct { + Ref string `json:"ref"` + Type string `json:"type"` + Message api.Message `json:"message"` +} + +type compactionToolMetadata struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` +} + +type compactionTranscript struct { + Items []CompactionTranscriptItem `json:"items"` + Tools []compactionToolMetadata `json:"tools,omitempty"` +} + +type compactionToolGroup struct { + CallRef string + ResultRef string +} + +// ResponsesCompactionPlan contains the validated, ordered state required to +// make and verify a compaction-model request. +type ResponsesCompactionPlan struct { + Model string + Stream bool + + items []CompactionTranscriptItem + tools []compactionToolMetadata + groups []compactionToolGroup + forcedRefs map[string]struct{} +} + +// ResponsesCompactionResult is the validated result of a compaction-model call. +type ResponsesCompactionResult struct { + Item ResponsesCompactionItem + Usage *ResponsesUsage +} + +type summarySelection struct { + Summary string `json:"summary"` + RetainItemIDs []string `json:"retain_item_ids"` +} + +type rawResponsesRequest struct { + Model string `json:"model"` + Input json.RawMessage `json:"input"` + Stream *bool `json:"stream,omitempty"` + Tools []ResponsesTool `json:"tools,omitempty"` + Fields map[string]json.RawMessage `json:"-"` +} + +// PrepareTriggeredCompaction recognizes the Codex v2 terminal trigger. It +// returns requested=false without changing ordinary Responses requests. +func PrepareTriggeredCompaction(body []byte) (*ResponsesCompactionPlan, bool, error) { + req, items, err := decodeRawResponsesRequest(body) + if err != nil { + return nil, false, err + } + if len(items) == 0 || rawInputItemType(items[len(items)-1]) != "compaction_trigger" { + return nil, false, nil + } + if req.Stream == nil || !*req.Stream { + return nil, true, errors.New("compaction_trigger requires stream=true") + } + for _, item := range items[:len(items)-1] { + if rawInputItemType(item) == "compaction_trigger" { + return nil, true, errors.New("compaction_trigger must be the final and only compaction trigger") + } + } + + items, _, err = expandOllamaCompactionItems(items[:len(items)-1]) + if err != nil { + return nil, true, err + } + plan, err := newResponsesCompactionPlan(req, items) + return plan, true, err +} + +// PrepareStandaloneCompaction validates a POST /v1/responses/compact body. +func PrepareStandaloneCompaction(body []byte) (*ResponsesCompactionPlan, error) { + req, items, err := decodeRawResponsesRequest(body) + if err != nil { + return nil, err + } + for _, item := range items { + if rawInputItemType(item) == "compaction_trigger" { + return nil, errors.New("compaction_trigger is only valid on POST /v1/responses") + } + } + items, _, err = expandOllamaCompactionItems(items) + if err != nil { + return nil, err + } + return newResponsesCompactionPlan(req, items) +} + +// ExpandResponsesCompactionInput replaces the newest Ollama compaction item +// with its summary and retained messages. Items on either side of the +// compaction item remain in their original order. +func ExpandResponsesCompactionInput(body []byte) ([]byte, bool, error) { + req, items, err := decodeRawResponsesRequest(body) + if err != nil { + return nil, false, err + } + items, changed, err := expandOllamaCompactionItems(items) + if err != nil || !changed { + return body, changed, err + } + + input, err := json.Marshal(items) + if err != nil { + return nil, false, err + } + req.Fields["input"] = input + rewritten, err := json.Marshal(req.Fields) + return rewritten, true, err +} + +func decodeRawResponsesRequest(body []byte) (rawResponsesRequest, []json.RawMessage, error) { + var fields map[string]json.RawMessage + if err := json.Unmarshal(body, &fields); err != nil { + return rawResponsesRequest{}, nil, fmt.Errorf("invalid Responses request: %w", err) + } + + var req rawResponsesRequest + if raw, ok := fields["model"]; !ok || json.Unmarshal(raw, &req.Model) != nil || strings.TrimSpace(req.Model) == "" { + return rawResponsesRequest{}, nil, errors.New("model is required") + } + req.Model = strings.TrimSpace(req.Model) + req.Fields = fields + req.Input = fields["input"] + if raw := fields["stream"]; len(raw) > 0 { + if err := json.Unmarshal(raw, &req.Stream); err != nil { + return rawResponsesRequest{}, nil, fmt.Errorf("invalid stream value: %w", err) + } + } + if raw := fields["tools"]; len(raw) > 0 { + if err := json.Unmarshal(raw, &req.Tools); err != nil { + return rawResponsesRequest{}, nil, fmt.Errorf("invalid tools: %w", err) + } + } + + items, err := decodeRawResponsesInput(req.Input) + if err != nil { + return rawResponsesRequest{}, nil, err + } + return req, items, nil +} + +func decodeRawResponsesInput(input json.RawMessage) ([]json.RawMessage, error) { + if len(input) == 0 || bytes.Equal(bytes.TrimSpace(input), []byte("null")) { + return nil, errors.New("input is required") + } + + var text string + if err := json.Unmarshal(input, &text); err == nil { + item, err := json.Marshal(map[string]any{ + "type": "message", "role": "user", "content": text, + }) + if err != nil { + return nil, err + } + return []json.RawMessage{item}, nil + } + + var items []json.RawMessage + if err := json.Unmarshal(input, &items); err != nil { + return nil, fmt.Errorf("input must be a string or array: %w", err) + } + if items == nil { + return nil, errors.New("input must be a string or array") + } + return items, nil +} + +func rawInputItemType(item json.RawMessage) string { + var header struct { + Type string `json:"type"` + Role string `json:"role"` + } + if json.Unmarshal(item, &header) != nil { + return "" + } + if header.Type == "" && header.Role != "" { + return "message" + } + return header.Type +} + +func expandOllamaCompactionItems(items []json.RawMessage) ([]json.RawMessage, bool, error) { + boundary := -1 + for i := range items { + if rawInputItemType(items[i]) == "compaction" { + boundary = i + } + } + if boundary < 0 { + return items, false, nil + } + + payload, err := decodeOllamaCompactionItem(items[boundary]) + if err != nil { + return nil, false, err + } + expanded, err := payloadToResponsesItems(payload) + if err != nil { + return nil, false, err + } + rewritten := make([]json.RawMessage, 0, len(items)-1+len(expanded)) + rewritten = append(rewritten, items[:boundary]...) + rewritten = append(rewritten, expanded...) + rewritten = append(rewritten, items[boundary+1:]...) + return rewritten, true, nil +} + +func decodeOllamaCompactionItem(item json.RawMessage) (OllamaCompactionPayload, error) { + var wire ResponsesCompactionItem + if err := json.Unmarshal(item, &wire); err != nil { + return OllamaCompactionPayload{}, fmt.Errorf("invalid compaction item: %w", err) + } + if wire.Type != "compaction" || wire.EncryptedContent == "" { + return OllamaCompactionPayload{}, errors.New("invalid compaction item") + } + + var payload OllamaCompactionPayload + decoder := json.NewDecoder(strings.NewReader(wire.EncryptedContent)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&payload); err != nil { + return OllamaCompactionPayload{}, errors.New("unsupported compaction item: encrypted_content is not an Ollama payload") + } + if payload.Type != OllamaCompactionPayloadType || payload.Version != OllamaCompactionPayloadVersion { + return OllamaCompactionPayload{}, fmt.Errorf("unsupported Ollama compaction payload type or version") + } + if strings.TrimSpace(payload.Summary) == "" { + return OllamaCompactionPayload{}, errors.New("Ollama compaction payload has an empty summary") + } + if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { + return OllamaCompactionPayload{}, errors.New("unsupported compaction item: encrypted_content contains more than one JSON value") + } + return payload, nil +} + +func payloadToResponsesItems(payload OllamaCompactionPayload) ([]json.RawMessage, error) { + b, err := json.Marshal(payload) + if err != nil { + return nil, err + } + hash := sha256.Sum256(b) + callID := "call_ollama_compaction_" + hex.EncodeToString(hash[:6]) + args, _ := json.Marshal(map[string]any{"version": payload.Version}) + + items := make([]json.RawMessage, 0, 2+len(payload.Retained)) + call, err := json.Marshal(map[string]any{ + "type": "function_call", "call_id": callID, "name": compactionSummaryToolName, "arguments": string(args), + }) + if err != nil { + return nil, err + } + result, err := json.Marshal(map[string]any{ + "type": "function_call_output", "call_id": callID, "output": payload.Summary, + }) + if err != nil { + return nil, err + } + items = append(items, call, result) + for _, message := range payload.Retained { + converted, err := messageToResponsesItems(message) + if err != nil { + return nil, fmt.Errorf("invalid retained message: %w", err) + } + items = append(items, converted...) + } + return items, nil +} + +func messageToResponsesItems(message api.Message) ([]json.RawMessage, error) { + if len(message.Images) > 0 { + return nil, errors.New("retained images are not supported") + } + + var values []any + if message.Thinking != "" { + values = append(values, map[string]any{ + "type": "reasoning", + "summary": []map[string]string{{"type": "summary_text", "text": message.Thinking}}, + }) + } + if message.Role == "tool" { + if message.ToolCallID == "" { + return nil, errors.New("retained tool message is missing tool_call_id") + } + values = append(values, map[string]any{ + "type": "function_call_output", "call_id": message.ToolCallID, "output": message.Content, + }) + } else if message.Content != "" || len(message.ToolCalls) == 0 { + values = append(values, map[string]any{ + "type": "message", "role": message.Role, "content": message.Content, + }) + } + for _, call := range message.ToolCalls { + if call.ID == "" || call.Function.Name == "" { + return nil, errors.New("retained function call is missing call_id or name") + } + arguments, err := json.Marshal(call.Function.Arguments) + if err != nil { + return nil, err + } + values = append(values, map[string]any{ + "type": "function_call", "call_id": call.ID, "name": call.Function.Name, "arguments": string(arguments), + }) + } + + items := make([]json.RawMessage, 0, len(values)) + for _, value := range values { + item, err := json.Marshal(value) + if err != nil { + return nil, err + } + items = append(items, item) + } + return items, nil +} + +func newResponsesCompactionPlan(req rawResponsesRequest, rawItems []json.RawMessage) (*ResponsesCompactionPlan, error) { + if len(rawItems) == 0 { + return nil, errors.New("compaction input is empty") + } + + items := make([]CompactionTranscriptItem, 0, len(rawItems)) + for i, raw := range rawItems { + item, err := unmarshalResponsesInputItem(raw) + if err != nil { + return nil, fmt.Errorf("input[%d]: %w", i, err) + } + message, kind, err := compactionMessage(item) + if err != nil { + return nil, fmt.Errorf("input[%d]: %w", i, err) + } + if len(message.Images) > 0 { + return nil, fmt.Errorf("input[%d]: image inputs are not supported by compaction", i) + } + items = append(items, CompactionTranscriptItem{ + Ref: fmt.Sprintf("item_%06d", i+1), Type: kind, Message: message, + }) + } + + groups, forced, err := analyzeCompactionToolState(items) + if err != nil { + return nil, err + } + stream := req.Stream != nil && *req.Stream + return &ResponsesCompactionPlan{ + Model: req.Model, Stream: stream, items: items, tools: collectCompactionToolMetadata(req.Tools), groups: groups, forcedRefs: forced, + }, nil +} + +func compactionMessage(item ResponsesInputItem) (api.Message, string, error) { + switch value := item.(type) { + case ResponsesInputMessage: + message, err := convertInputMessage(value) + return message, "message", err + case ResponsesFunctionCall: + var arguments api.ToolCallFunctionArguments + if value.Arguments != "" { + if err := json.Unmarshal([]byte(value.Arguments), &arguments); err != nil { + return api.Message{}, "", fmt.Errorf("invalid function arguments: %w", err) + } + } + return api.Message{Role: "assistant", ToolCalls: []api.ToolCall{{ + ID: value.CallID, Function: api.ToolCallFunction{Name: value.Name, Arguments: arguments}, + }}}, "function_call", nil + case ResponsesFunctionCallOutput: + content := value.Output + var images []api.ImageData + if len(value.OutputItems) > 0 { + var err error + content, images, err = convertResponsesContent(value.OutputItems) + if err != nil { + return api.Message{}, "", err + } + } + return api.Message{Role: "tool", Content: content, Images: images, ToolCallID: value.CallID}, "function_call_output", nil + case ResponsesReasoningInput: + var summary strings.Builder + for _, part := range value.Summary { + summary.WriteString(part.Text) + } + thinking := summary.String() + if thinking == "" && value.EncryptedContent != "" { + thinking = "[opaque reasoning state omitted during Ollama compaction]" + } + return api.Message{Role: "assistant", Thinking: thinking}, "reasoning", nil + case ResponsesCompactionItem, ResponsesCompactionTrigger: + return api.Message{}, "", errors.New("unexpected compaction control item") + default: + return api.Message{}, "", fmt.Errorf("unsupported compaction input type %T", item) + } +} + +func analyzeCompactionToolState(items []CompactionTranscriptItem) ([]compactionToolGroup, map[string]struct{}, error) { + type pendingGroup struct { + callIndex int + resultIndex int + group compactionToolGroup + } + byCallID := make(map[string]*pendingGroup) + ignoredCallIDs := make(map[string]struct{}) + var ordered []*pendingGroup + + for i, item := range items { + switch item.Type { + case "function_call": + call := item.Message.ToolCalls[0] + if call.ID == "" { + return nil, nil, fmt.Errorf("%s: function call is missing call_id", item.Ref) + } + if call.Function.Name == compactionSummaryToolName && strings.HasPrefix(call.ID, "call_ollama_compaction_") { + ignoredCallIDs[call.ID] = struct{}{} + continue + } + if _, exists := byCallID[call.ID]; exists { + return nil, nil, fmt.Errorf("%s: duplicate function call_id %q", item.Ref, call.ID) + } + group := &pendingGroup{callIndex: i, resultIndex: -1, group: compactionToolGroup{CallRef: item.Ref}} + byCallID[call.ID] = group + ordered = append(ordered, group) + case "function_call_output": + callID := item.Message.ToolCallID + if _, ignored := ignoredCallIDs[callID]; ignored { + continue + } + group := byCallID[callID] + if group == nil { + return nil, nil, fmt.Errorf("%s: function output has no matching call %q", item.Ref, callID) + } + if group.resultIndex >= 0 { + return nil, nil, fmt.Errorf("%s: duplicate function output for call %q", item.Ref, callID) + } + group.resultIndex = i + group.group.ResultRef = item.Ref + } + } + + forced := make(map[string]struct{}) + groups := make([]compactionToolGroup, 0, len(ordered)) + for _, candidate := range ordered { + groups = append(groups, candidate.group) + active := candidate.resultIndex < 0 + if !active { + active = true + for _, item := range items[candidate.resultIndex+1:] { + if item.Type == "reasoning" || item.Type == "function_call" || (item.Type == "message" && item.Message.Role == "assistant") { + active = false + break + } + } + } + if active { + forced[candidate.group.CallRef] = struct{}{} + if candidate.group.ResultRef != "" { + forced[candidate.group.ResultRef] = struct{}{} + } + } + } + return groups, forced, nil +} + +func collectCompactionToolMetadata(tools []ResponsesTool) []compactionToolMetadata { + var metadata []compactionToolMetadata + var add func(prefix string, tools []ResponsesTool) + add = func(prefix string, tools []ResponsesTool) { + for _, tool := range tools { + name := tool.Name + if prefix != "" && !strings.HasPrefix(name, prefix+".") { + name = prefix + "." + name + } + if tool.Type == "namespace" { + add(name, tool.Tools) + continue + } + description := "" + if tool.Description != nil { + description = *tool.Description + } + metadata = append(metadata, compactionToolMetadata{Name: name, Description: description}) + } + } + add("", tools) + return metadata +} + +// SummaryRequest returns an ordinary non-streaming Responses request. A repair +// request includes the validation error from the first model response. +func (p *ResponsesCompactionPlan) SummaryRequest(repairError string) ([]byte, error) { + // TODO(compaction): enforce the 40% retained-context target after the + // selected model's prompt renderer and token budget are available here. + transcript, err := json.Marshal(compactionTranscript{Items: p.items, Tools: p.tools}) + if err != nil { + return nil, err + } + prompt := `Summarize the conversation for another coding agent. Preserve the goal, decisions, constraints, repository state, changed files, test results, failures, active work, and next actions. Use retain_item_ids only for exact source items that cannot safely be paraphrased. Tool calls and results are execution state; do not invent or edit them. Call create_summary exactly once.` + if repairError != "" { + prompt += " Your previous create_summary call was invalid: " + repairError + ". Return one corrected create_summary call." + } + + description := "Return the compact summary and the exact input item references that must remain verbatim." + strict := true + request := map[string]any{ + "model": p.Model, + "input": []any{ + map[string]any{"type": "message", "role": "system", "content": prompt}, + map[string]any{"type": "message", "role": "user", "content": string(transcript)}, + }, + "tools": []ResponsesTool{{ + Type: "function", Name: CreateSummaryToolName, Description: &description, Strict: &strict, + Parameters: map[string]any{ + "type": "object", + "properties": map[string]any{ + "summary": map[string]any{"type": "string"}, + "retain_item_ids": map[string]any{"type": "array", "items": map[string]any{"type": "string"}}, + }, + "required": []string{"summary", "retain_item_ids"}, + "additionalProperties": false, + }, + }}, + "tool_choice": map[string]any{"type": "function", "name": CreateSummaryToolName}, + "parallel_tool_calls": false, + "store": false, + "stream": false, + } + return json.Marshal(request) +} + +// Complete validates create_summary and builds the stateless continuation. +func (p *ResponsesCompactionPlan) Complete(body []byte) (ResponsesCompactionResult, error) { + var response ResponsesResponse + if err := json.Unmarshal(body, &response); err != nil { + return ResponsesCompactionResult{}, fmt.Errorf("invalid summary response: %w", err) + } + + var calls []ResponsesOutputItem + for _, item := range response.Output { + if item.Type == "function_call" && item.Name == CreateSummaryToolName { + calls = append(calls, item) + } + } + if len(calls) != 1 { + return ResponsesCompactionResult{}, fmt.Errorf("expected one %s call, got %d", CreateSummaryToolName, len(calls)) + } + + var selection summarySelection + if err := json.Unmarshal([]byte(calls[0].Arguments), &selection); err != nil { + return ResponsesCompactionResult{}, fmt.Errorf("invalid %s arguments: %w", CreateSummaryToolName, err) + } + selection.Summary = strings.TrimSpace(selection.Summary) + if selection.Summary == "" { + return ResponsesCompactionResult{}, errors.New("create_summary returned an empty summary") + } + + known := make(map[string]struct{}, len(p.items)) + for _, item := range p.items { + known[item.Ref] = struct{}{} + } + selected := make(map[string]struct{}, len(selection.RetainItemIDs)+len(p.forcedRefs)) + for _, ref := range selection.RetainItemIDs { + if _, ok := known[ref]; !ok { + return ResponsesCompactionResult{}, fmt.Errorf("create_summary selected unknown item %q", ref) + } + if _, duplicate := selected[ref]; duplicate { + return ResponsesCompactionResult{}, fmt.Errorf("create_summary selected duplicate item %q", ref) + } + selected[ref] = struct{}{} + } + for ref := range p.forcedRefs { + selected[ref] = struct{}{} + } + for _, group := range p.groups { + _, callSelected := selected[group.CallRef] + _, resultSelected := selected[group.ResultRef] + if callSelected || resultSelected { + selected[group.CallRef] = struct{}{} + if group.ResultRef != "" { + selected[group.ResultRef] = struct{}{} + } + } + } + + retained := make([]api.Message, 0, len(selected)) + for _, item := range p.items { + if _, ok := selected[item.Ref]; ok { + retained = append(retained, item.Message) + } + } + payload := OllamaCompactionPayload{ + Type: OllamaCompactionPayloadType, Version: OllamaCompactionPayloadVersion, Summary: selection.Summary, Retained: retained, + } + payloadJSON, err := json.Marshal(payload) + if err != nil { + return ResponsesCompactionResult{}, err + } + return ResponsesCompactionResult{ + Item: ResponsesCompactionItem{Type: "compaction", EncryptedContent: string(payloadJSON)}, Usage: response.Usage, + }, nil +} + +// ResponsesCompactedResponse is returned by POST /v1/responses/compact. +type ResponsesCompactedResponse struct { + ID string `json:"id"` + Object string `json:"object"` + CreatedAt int64 `json:"created_at"` + Output []ResponsesCompactionItem `json:"output"` + Usage *ResponsesUsage `json:"usage"` +} + +// NewResponsesCompactedResponse builds the standalone compact response. +func NewResponsesCompactedResponse(id string, result ResponsesCompactionResult) ResponsesCompactedResponse { + return ResponsesCompactedResponse{ + ID: id, Object: "response.compaction", CreatedAt: time.Now().Unix(), Output: []ResponsesCompactionItem{result.Item}, Usage: result.Usage, + } +} + +// NewResponsesCompactionStreamEvents builds the Codex v2 stream. Codex uses +// output_item.done and requires exactly one compaction item before completed. +func NewResponsesCompactionStreamEvents(id, model string, result ResponsesCompactionResult) []ResponsesStreamEvent { + converter := NewResponsesStreamConverter(id, "", model, ResponsesRequest{Model: model}) + created := converter.buildResponseObject("in_progress", []any{}, nil) + completedUsage := map[string]any(nil) + if result.Usage != nil { + completedUsage = map[string]any{ + "input_tokens": result.Usage.InputTokens, + "output_tokens": result.Usage.OutputTokens, + "total_tokens": result.Usage.TotalTokens, + "input_tokens_details": result.Usage.InputTokensDetails, + "output_tokens_details": result.Usage.OutputTokensDetails, + } + } + completed := converter.buildResponseObject("completed", []any{result.Item}, completedUsage) + completed["completed_at"] = time.Now().Unix() + + return []ResponsesStreamEvent{ + converter.newEvent("response.created", map[string]any{"response": created}), + converter.newEvent("response.in_progress", map[string]any{"response": created}), + converter.newEvent("response.output_item.added", map[string]any{"output_index": 0, "item": result.Item}), + converter.newEvent("response.output_item.done", map[string]any{"output_index": 0, "item": result.Item}), + converter.newEvent("response.completed", map[string]any{"response": completed}), + } +} diff --git a/openai/responses_compact_test.go b/openai/responses_compact_test.go new file mode 100644 index 000000000..eb08c35e5 --- /dev/null +++ b/openai/responses_compact_test.go @@ -0,0 +1,313 @@ +package openai + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/ollama/ollama/api" +) + +func compactionResponseBody(t *testing.T, selection map[string]any) []byte { + t.Helper() + arguments, err := json.Marshal(selection) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(map[string]any{ + "id": "resp_summary", + "object": "response", + "status": "completed", + "output": []any{map[string]any{ + "id": "fc_summary", "type": "function_call", "status": "completed", + "call_id": "call_summary", "name": CreateSummaryToolName, "arguments": string(arguments), + }}, + "usage": map[string]any{"input_tokens": 100, "output_tokens": 20, "total_tokens": 120}, + }) + if err != nil { + t.Fatal(err) + } + return body +} + +func decodeResultPayload(t *testing.T, result ResponsesCompactionResult) OllamaCompactionPayload { + t.Helper() + item, err := json.Marshal(result.Item) + if err != nil { + t.Fatal(err) + } + payload, err := decodeOllamaCompactionItem(item) + if err != nil { + t.Fatal(err) + } + return payload +} + +func TestPrepareTriggeredCompactionBuildsSummaryRequest(t *testing.T) { + description := "Read a file" + body := []byte(`{ + "model":"qwen3:4b", + "stream":true, + "instructions":"do not send this to the compactor", + "tools":[{"type":"function","name":"read_file","description":"Read a file","strict":false,"parameters":{"type":"object"}}], + "input":[ + {"type":"message","role":"user","content":"inspect main.go"}, + {"type":"function_call","call_id":"call_1","name":"read_file","arguments":"{\"path\":\"main.go\"}"}, + {"type":"function_call_output","call_id":"call_1","output":"package main"}, + {"type":"compaction_trigger"} + ] + }`) + + plan, requested, err := PrepareTriggeredCompaction(body) + if err != nil { + t.Fatal(err) + } + if !requested { + t.Fatal("expected compaction request") + } + if plan.Model != "qwen3:4b" || !plan.Stream { + t.Fatalf("unexpected plan: model=%q stream=%v", plan.Model, plan.Stream) + } + + requestBody, err := plan.SummaryRequest("") + if err != nil { + t.Fatal(err) + } + var request struct { + Input []struct { + Role string `json:"role"` + Content string `json:"content"` + } `json:"input"` + Tools []ResponsesTool `json:"tools"` + Instructions string `json:"instructions"` + Stream bool `json:"stream"` + } + if err := json.Unmarshal(requestBody, &request); err != nil { + t.Fatal(err) + } + if request.Stream { + t.Fatal("summary inference must be non-streaming") + } + if request.Instructions != "" { + t.Fatalf("original instructions leaked into summary request: %q", request.Instructions) + } + if len(request.Tools) != 1 || request.Tools[0].Name != CreateSummaryToolName { + t.Fatalf("unexpected callable tools: %+v", request.Tools) + } + if len(request.Input) != 2 || !strings.Contains(request.Input[1].Content, description) || !strings.Contains(request.Input[1].Content, "read_file") { + t.Fatalf("summary transcript is missing tool metadata: %+v", request.Input) + } + if strings.Contains(request.Input[1].Content, `"instructions"`) { + t.Fatalf("original instructions leaked into transcript: %s", request.Input[1].Content) + } + + result, err := plan.Complete(compactionResponseBody(t, map[string]any{ + "summary": "Continue inspecting main.go.", "retain_item_ids": []string{}, + })) + if err != nil { + t.Fatal(err) + } + payload := decodeResultPayload(t, result) + if payload.Summary != "Continue inspecting main.go." { + t.Fatalf("unexpected summary: %q", payload.Summary) + } + if len(payload.Retained) != 2 { + t.Fatalf("expected active call and result to be forced, got %+v", payload.Retained) + } + if payload.Retained[0].ToolCalls[0].ID != "call_1" || payload.Retained[1].ToolCallID != "call_1" { + t.Fatalf("tool state was not retained in order: %+v", payload.Retained) + } +} + +func TestPrepareTriggeredCompactionRequiresStreaming(t *testing.T) { + body := []byte(`{"model":"test","input":[{"type":"message","role":"user","content":"hi"},{"type":"compaction_trigger"}]}`) + _, requested, err := PrepareTriggeredCompaction(body) + if !requested || err == nil || !strings.Contains(err.Error(), "stream=true") { + t.Fatalf("requested=%v err=%v", requested, err) + } +} + +func TestCompactionSelectionKeepsCompleteToolGroupInOriginalOrder(t *testing.T) { + body := []byte(`{ + "model":"test", + "stream":true, + "input":[ + {"type":"message","role":"user","content":"run it"}, + {"type":"function_call","call_id":"call_1","name":"shell","arguments":"{}"}, + {"type":"function_call_output","call_id":"call_1","output":"ok"}, + {"type":"message","role":"assistant","content":"done"}, + {"type":"compaction_trigger"} + ] + }`) + plan, requested, err := PrepareTriggeredCompaction(body) + if err != nil || !requested { + t.Fatalf("prepare: requested=%v err=%v", requested, err) + } + + result, err := plan.Complete(compactionResponseBody(t, map[string]any{ + "summary": "The command completed.", "retain_item_ids": []string{"item_000003"}, + })) + if err != nil { + t.Fatal(err) + } + payload := decodeResultPayload(t, result) + if len(payload.Retained) != 2 { + t.Fatalf("expected complete tool group, got %+v", payload.Retained) + } + if len(payload.Retained[0].ToolCalls) != 1 || payload.Retained[1].Role != "tool" { + t.Fatalf("tool group order changed: %+v", payload.Retained) + } +} + +func TestCompactionRejectsInvalidToolStateAndSelections(t *testing.T) { + t.Run("orphan result", func(t *testing.T) { + body := []byte(`{"model":"test","stream":true,"input":[{"type":"function_call_output","call_id":"missing","output":"x"},{"type":"compaction_trigger"}]}`) + _, requested, err := PrepareTriggeredCompaction(body) + if !requested || err == nil || !strings.Contains(err.Error(), "no matching call") { + t.Fatalf("requested=%v err=%v", requested, err) + } + }) + + t.Run("unknown ref", func(t *testing.T) { + body := []byte(`{"model":"test","stream":true,"input":[{"type":"message","role":"user","content":"hi"},{"type":"compaction_trigger"}]}`) + plan, _, err := PrepareTriggeredCompaction(body) + if err != nil { + t.Fatal(err) + } + _, err = plan.Complete(compactionResponseBody(t, map[string]any{ + "summary": "hi", "retain_item_ids": []string{"missing"}, + })) + if err == nil || !strings.Contains(err.Error(), "unknown item") { + t.Fatalf("expected unknown item error, got %v", err) + } + }) +} + +func TestExpandResponsesCompactionInputPreservesCodexRetainedPrefix(t *testing.T) { + payload := OllamaCompactionPayload{ + Type: OllamaCompactionPayloadType, Version: OllamaCompactionPayloadVersion, + Summary: "The user chose option A.", + Retained: []api.Message{{Role: "assistant", ToolCalls: []api.ToolCall{{ + ID: "call_live", Function: api.ToolCallFunction{Name: "shell", Arguments: api.NewToolCallFunctionArguments()}, + }}}}, + } + payloadJSON, err := json.Marshal(payload) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(map[string]any{ + "model": "test", + "input": []any{ + map[string]any{"type": "message", "role": "user", "content": "first user turn retained by Codex"}, + map[string]any{"type": "message", "role": "user", "content": "second user turn retained by Codex"}, + ResponsesCompactionItem{Type: "compaction", EncryptedContent: string(payloadJSON)}, + map[string]any{"type": "message", "role": "user", "content": "new turn"}, + }, + }) + if err != nil { + t.Fatal(err) + } + + rewritten, changed, err := ExpandResponsesCompactionInput(body) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("expected rewritten input") + } + var request struct { + Input []json.RawMessage `json:"input"` + } + if err := json.Unmarshal(rewritten, &request); err != nil { + t.Fatal(err) + } + if len(request.Input) != 6 { + t.Fatalf("expected retained prefix, summary pair, retained call, and new turn; got %s", rewritten) + } + var types []string + for _, item := range request.Input { + types = append(types, rawInputItemType(item)) + } + want := []string{"message", "message", "function_call", "function_call_output", "function_call", "message"} + for i := range want { + if types[i] != want[i] { + t.Fatalf("types=%v want=%v", types, want) + } + } + for _, marker := range []string{"first user turn retained by Codex", "second user turn retained by Codex"} { + if strings.Count(string(rewritten), marker) != 1 { + t.Fatalf("retained user message %q was lost or duplicated: %s", marker, rewritten) + } + } + if !strings.Contains(string(rewritten), "new turn") { + t.Fatalf("post-boundary input was lost: %s", rewritten) + } +} + +func TestExpandResponsesCompactionInputRejectsForeignPayload(t *testing.T) { + body := []byte(`{"model":"test","input":[{"type":"compaction","encrypted_content":"opaque-provider-state"}]}`) + _, changed, err := ExpandResponsesCompactionInput(body) + if changed || err == nil || !strings.Contains(err.Error(), "not an Ollama payload") { + t.Fatalf("changed=%v err=%v", changed, err) + } +} + +func TestRepeatedCompactionReplacesRatherThanNestsPayload(t *testing.T) { + oldPayload, err := json.Marshal(OllamaCompactionPayload{ + Type: OllamaCompactionPayloadType, Version: OllamaCompactionPayloadVersion, Summary: "old summary", + }) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(map[string]any{ + "model": "test", "stream": true, + "input": []any{ + map[string]any{"type": "message", "role": "user", "content": "retained prefix"}, + ResponsesCompactionItem{Type: "compaction", EncryptedContent: string(oldPayload)}, + map[string]any{"type": "message", "role": "user", "content": "new work"}, + ResponsesCompactionTrigger{Type: "compaction_trigger"}, + }, + }) + if err != nil { + t.Fatal(err) + } + plan, requested, err := PrepareTriggeredCompaction(body) + if err != nil || !requested { + t.Fatalf("prepare: requested=%v err=%v", requested, err) + } + result, err := plan.Complete(compactionResponseBody(t, map[string]any{ + "summary": "replacement summary", "retain_item_ids": []string{}, + })) + if err != nil { + t.Fatal(err) + } + payload := decodeResultPayload(t, result) + if len(payload.Retained) != 0 { + t.Fatalf("old synthetic summary pair was retained: %+v", payload.Retained) + } + if strings.Contains(result.Item.EncryptedContent, "old summary") || strings.Contains(result.Item.EncryptedContent, "retained prefix") { + t.Fatalf("old compaction was nested: %s", result.Item.EncryptedContent) + } +} + +func TestCompactionStreamContainsExactlyOneCompletedItem(t *testing.T) { + result := ResponsesCompactionResult{Item: ResponsesCompactionItem{Type: "compaction", EncryptedContent: `{"type":"ollama_compaction","version":1}`}} + events := NewResponsesCompactionStreamEvents("resp_1", "test", result) + var done, completed int + for _, event := range events { + switch event.Event { + case "response.output_item.done": + done++ + data := event.Data.(map[string]any) + item := data["item"].(ResponsesCompactionItem) + if item.Type != "compaction" { + t.Fatalf("unexpected item: %+v", item) + } + case "response.completed": + completed++ + } + } + if done != 1 || completed != 1 { + t.Fatalf("done=%d completed=%d events=%+v", done, completed, events) + } +} diff --git a/server/responses_compact.go b/server/responses_compact.go new file mode 100644 index 000000000..8aae4d334 --- /dev/null +++ b/server/responses_compact.go @@ -0,0 +1,219 @@ +package server + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "time" + + "github.com/gin-gonic/gin" + "github.com/klauspost/compress/zstd" + + "github.com/ollama/ollama/middleware" + "github.com/ollama/ollama/openai" +) + +// responsesCompactionMiddleware intercepts only Codex compaction control items. +// Ordinary Responses requests continue through the existing route unchanged. +func (s *Server) responsesCompactionMiddleware() gin.HandlerFunc { + return func(c *gin.Context) { + body, err := readResponsesCompactionBody(c) + if err != nil { + writeResponsesCompactionError(c, http.StatusBadRequest, "invalid_request_error", err.Error()) + return + } + + plan, requested, err := openai.PrepareTriggeredCompaction(body) + if err != nil { + writeResponsesCompactionError(c, http.StatusBadRequest, "invalid_request_error", err.Error()) + return + } + if requested { + c.Abort() + s.handleResponsesCompaction(c, plan, true) + return + } + + rewritten, changed, err := openai.ExpandResponsesCompactionInput(body) + if err != nil { + writeResponsesCompactionError(c, http.StatusBadRequest, "invalid_request_error", err.Error()) + return + } + if changed { + resetResponsesRequestBody(c.Request, rewritten) + } + c.Next() + } +} + +// ResponsesCompactHandler implements POST /v1/responses/compact with an +// Ollama-owned ordinary inference request rather than upstream passthrough. +func (s *Server) ResponsesCompactHandler(c *gin.Context) { + body, err := readResponsesCompactionBody(c) + if err != nil { + writeResponsesCompactionError(c, http.StatusBadRequest, "invalid_request_error", err.Error()) + return + } + plan, err := openai.PrepareStandaloneCompaction(body) + if err != nil { + writeResponsesCompactionError(c, http.StatusBadRequest, "invalid_request_error", err.Error()) + return + } + s.handleResponsesCompaction(c, plan, false) +} + +func readResponsesCompactionBody(c *gin.Context) ([]byte, error) { + if c.GetHeader("Content-Encoding") == "zstd" { + reader, err := zstd.NewReader(c.Request.Body, zstd.WithDecoderMaxMemory(8<<20)) + if err != nil { + return nil, fmt.Errorf("failed to decompress zstd body") + } + decompressed, err := io.ReadAll(http.MaxBytesReader(c.Writer, io.NopCloser(reader), maxDecompressedBodySize)) + reader.Close() + if err != nil { + return nil, err + } + c.Request.Header.Del("Content-Encoding") + resetResponsesRequestBody(c.Request, decompressed) + } + + body, err := readRequestBody(c.Request) + if err != nil { + return nil, err + } + if len(bytes.TrimSpace(body)) == 0 { + return nil, fmt.Errorf("missing request body") + } + return body, nil +} + +func resetResponsesRequestBody(r *http.Request, body []byte) { + r.Body = io.NopCloser(bytes.NewReader(body)) + r.ContentLength = int64(len(body)) + r.Header.Set("Content-Length", strconv.Itoa(len(body))) +} + +func (s *Server) handleResponsesCompaction(c *gin.Context, plan *openai.ResponsesCompactionPlan, stream bool) { + var validationErr error + for range 2 { + repair := "" + if validationErr != nil { + repair = validationErr.Error() + } + request, err := plan.SummaryRequest(repair) + if err != nil { + writeResponsesCompactionError(c, http.StatusInternalServerError, "compaction_failed", "compaction failed; the original conversation is unchanged") + return + } + + response := s.runResponsesCompactionInference(c, request) + if response.status < http.StatusOK || response.status >= http.StatusMultipleChoices { + copyResponsesCompactionResponse(c, response) + return + } + + result, err := plan.Complete(response.body.Bytes()) + if err != nil { + validationErr = err + continue + } + + id := fmt.Sprintf("resp_compact_%d", time.Now().UnixNano()) + if stream { + writeResponsesCompactionStream(c, openai.NewResponsesCompactionStreamEvents(id, plan.Model, result)) + return + } + c.JSON(http.StatusOK, openai.NewResponsesCompactedResponse(id, result)) + return + } + + writeResponsesCompactionError(c, http.StatusInternalServerError, "compaction_failed", "compaction failed; the selected model did not return a valid summary and the original conversation is unchanged") +} + +// runResponsesCompactionInference uses the normal Responses stack without the +// compaction dispatcher. This keeps local and cloud model selection identical +// to an ordinary request and permits one isolated repair retry. +func (s *Server) runResponsesCompactionInference(c *gin.Context, body []byte) *responsesInferenceRecorder { + router := gin.New() + router.POST("/v1/responses", + cloudPassthroughMiddleware(cloudErrRemoteInferenceUnavailable), + middleware.ResponsesMiddleware(), + s.ChatHandler, + ) + + req, err := http.NewRequestWithContext(c.Request.Context(), http.MethodPost, "/v1/responses", bytes.NewReader(body)) + if err != nil { + return &responsesInferenceRecorder{header: make(http.Header), status: http.StatusInternalServerError, body: *bytes.NewBufferString(err.Error())} + } + req.Header = c.Request.Header.Clone() + req.Header.Del("Content-Encoding") + req.Header.Set("Content-Type", "application/json") + req.ContentLength = int64(len(body)) + + recorder := &responsesInferenceRecorder{header: make(http.Header)} + router.ServeHTTP(recorder, req) + return recorder +} + +type responsesInferenceRecorder struct { + header http.Header + body bytes.Buffer + status int +} + +func (r *responsesInferenceRecorder) Header() http.Header { + return r.header +} + +func (r *responsesInferenceRecorder) WriteHeader(status int) { + if r.status == 0 { + r.status = status + } +} + +func (r *responsesInferenceRecorder) Write(data []byte) (int, error) { + if r.status == 0 { + r.status = http.StatusOK + } + return r.body.Write(data) +} + +func (r *responsesInferenceRecorder) Flush() {} + +func copyResponsesCompactionResponse(c *gin.Context, response *responsesInferenceRecorder) { + for key, values := range response.header { + if key == "Content-Length" { + continue + } + for _, value := range values { + c.Header(key, value) + } + } + c.Data(response.status, response.header.Get("Content-Type"), response.body.Bytes()) +} + +func writeResponsesCompactionStream(c *gin.Context, events []openai.ResponsesStreamEvent) { + c.Header("Content-Type", "text/event-stream") + c.Header("Cache-Control", "no-cache") + c.Header("Connection", "keep-alive") + c.Status(http.StatusOK) + for _, event := range events { + data, err := json.Marshal(event.Data) + if err != nil { + return + } + _, _ = fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event.Event, data) + } + if flusher, ok := c.Writer.(http.Flusher); ok { + flusher.Flush() + } +} + +func writeResponsesCompactionError(c *gin.Context, status int, code, message string) { + response := openai.NewError(status, message) + response.Error.Code = &code + c.AbortWithStatusJSON(status, response) +} diff --git a/server/responses_compact_test.go b/server/responses_compact_test.go new file mode 100644 index 000000000..02d654419 --- /dev/null +++ b/server/responses_compact_test.go @@ -0,0 +1,282 @@ +package server + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/gin-gonic/gin" + + "github.com/ollama/ollama/openai" +) + +func summaryResponse(t *testing.T, summary string, retained []string) []byte { + t.Helper() + arguments, err := json.Marshal(map[string]any{"summary": summary, "retain_item_ids": retained}) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(map[string]any{ + "id": "resp_summary", "object": "response", "status": "completed", "model": "fixture", + "output": []any{map[string]any{ + "id": "fc_summary", "type": "function_call", "status": "completed", "call_id": "call_summary", + "name": openai.CreateSummaryToolName, "arguments": string(arguments), + }}, + "usage": map[string]any{ + "input_tokens": 100, "output_tokens": 20, "total_tokens": 120, + "input_tokens_details": map[string]any{"cached_tokens": 0}, + "output_tokens_details": map[string]any{"reasoning_tokens": 0}, + }, + }) + if err != nil { + t.Fatal(err) + } + return body +} + +type compactionUpstreamCapture struct { + mu sync.Mutex + paths []string + bodies [][]byte +} + +func (c *compactionUpstreamCapture) add(path string, body []byte) int { + c.mu.Lock() + defer c.mu.Unlock() + c.paths = append(c.paths, path) + c.bodies = append(c.bodies, append([]byte(nil), body...)) + return len(c.bodies) +} + +func (c *compactionUpstreamCapture) snapshot() ([]string, [][]byte) { + c.mu.Lock() + defer c.mu.Unlock() + paths := append([]string(nil), c.paths...) + bodies := make([][]byte, len(c.bodies)) + for i := range c.bodies { + bodies[i] = append([]byte(nil), c.bodies[i]...) + } + return paths, bodies +} + +func newCompactionTestServer(t *testing.T, handler func(int, http.ResponseWriter, *http.Request, []byte)) (*httptest.Server, *compactionUpstreamCapture) { + t.Helper() + gin.SetMode(gin.TestMode) + setTestHome(t, t.TempDir()) + + capture := &compactionUpstreamCapture{} + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + attempt := capture.add(r.URL.Path, body) + handler(attempt, w, r, body) + })) + t.Cleanup(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) + t.Cleanup(local.Close) + return local, capture +} + +func postCompactionRequest(t *testing.T, server *httptest.Server, path, body string) (int, http.Header, []byte) { + t.Helper() + req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, server.URL+path, strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + response, err := server.Client().Do(req) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + responseBody, err := io.ReadAll(response.Body) + if err != nil { + t.Fatal(err) + } + return response.StatusCode, response.Header.Clone(), responseBody +} + +func TestResponsesCompactUsesOrdinarySelectedCloudModel(t *testing.T) { + local, capture := newCompactionTestServer(t, func(_ int, w http.ResponseWriter, _ *http.Request, _ []byte) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(summaryResponse(t, "Continue the task.", nil)) + }) + + status, _, body := postCompactionRequest(t, local, "/v1/responses/compact", `{ + "model":"fixture:cloud", + "instructions":"original agent instructions", + "input":[{"type":"message","role":"user","content":"hello"}], + "tools":[{"type":"function","name":"shell","description":"Run a command","strict":false,"parameters":{"type":"object"}}] + }`) + if status != http.StatusOK { + t.Fatalf("status=%d body=%s", status, body) + } + var compacted openai.ResponsesCompactedResponse + if err := json.Unmarshal(body, &compacted); err != nil { + t.Fatal(err) + } + if compacted.Object != "response.compaction" || len(compacted.Output) != 1 || compacted.Output[0].Type != "compaction" { + t.Fatalf("unexpected compact response: %+v", compacted) + } + + paths, bodies := capture.snapshot() + if len(paths) != 1 || paths[0] != "/v1/responses" { + t.Fatalf("compaction must use one ordinary Responses inference call, paths=%v", paths) + } + if bytes.Contains(bodies[0], []byte("original agent instructions")) { + t.Fatalf("top-level instructions leaked to compactor: %s", bodies[0]) + } + var summaryRequest struct { + Model string `json:"model"` + Stream bool `json:"stream"` + Tools []openai.ResponsesTool `json:"tools"` + } + if err := json.Unmarshal(bodies[0], &summaryRequest); err != nil { + t.Fatal(err) + } + if summaryRequest.Model != "fixture" || summaryRequest.Stream { + t.Fatalf("unexpected upstream summary request: %+v", summaryRequest) + } + if len(summaryRequest.Tools) != 1 || summaryRequest.Tools[0].Name != openai.CreateSummaryToolName { + t.Fatalf("unexpected callable tools: %+v", summaryRequest.Tools) + } + if !bytes.Contains(bodies[0], []byte("shell")) { + t.Fatalf("original tool metadata missing from transcript: %s", bodies[0]) + } +} + +func TestResponsesCompactionTriggerReturnsCodexStream(t *testing.T) { + local, capture := newCompactionTestServer(t, func(_ int, w http.ResponseWriter, _ *http.Request, _ []byte) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(summaryResponse(t, "Compact summary.", nil)) + }) + + status, header, body := postCompactionRequest(t, local, "/v1/responses", `{ + "model":"fixture:cloud","stream":true, + "input":[{"type":"message","role":"user","content":"hello"},{"type":"compaction_trigger"}] + }`) + if status != http.StatusOK { + t.Fatalf("status=%d body=%s", status, body) + } + if got := header.Get("Content-Type"); !strings.HasPrefix(got, "text/event-stream") { + t.Fatalf("content-type=%q", got) + } + text := string(body) + if strings.Count(text, "event: response.output_item.done") != 1 || strings.Count(text, `"type":"compaction"`) == 0 { + t.Fatalf("missing single compaction output item: %s", text) + } + if strings.Count(text, "event: response.completed") != 1 { + t.Fatalf("missing response.completed: %s", text) + } + paths, requests := capture.snapshot() + if len(paths) != 1 || paths[0] != "/v1/responses" { + t.Fatalf("paths=%v requests=%s", paths, requests) + } +} + +func TestResponsesCompactionRepairsMalformedSummaryOnce(t *testing.T) { + local, capture := newCompactionTestServer(t, func(attempt int, w http.ResponseWriter, _ *http.Request, _ []byte) { + w.Header().Set("Content-Type", "application/json") + if attempt == 1 { + _, _ = w.Write([]byte(`{"id":"bad","object":"response","output":[{"type":"message","role":"assistant","content":[]}]}`)) + return + } + _, _ = w.Write(summaryResponse(t, "Repaired summary.", nil)) + }) + + status, _, body := postCompactionRequest(t, local, "/v1/responses/compact", `{"model":"fixture:cloud","input":"hello"}`) + if status != http.StatusOK { + t.Fatalf("status=%d body=%s", status, body) + } + _, bodies := capture.snapshot() + if len(bodies) != 2 { + t.Fatalf("expected one repair retry, got %d requests", len(bodies)) + } + if !bytes.Contains(bodies[1], []byte("previous create_summary call was invalid")) { + t.Fatalf("repair request does not explain the validation error: %s", bodies[1]) + } +} + +func TestResponsesCompactionFailsAfterOneRepair(t *testing.T) { + local, capture := newCompactionTestServer(t, func(_ int, w http.ResponseWriter, _ *http.Request, _ []byte) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"bad","object":"response","output":[]}`)) + }) + + status, _, body := postCompactionRequest(t, local, "/v1/responses/compact", `{"model":"fixture:cloud","input":"hello"}`) + if status != http.StatusInternalServerError { + t.Fatalf("status=%d body=%s", status, body) + } + var errorResponse openai.ErrorResponse + if err := json.Unmarshal(body, &errorResponse); err != nil { + t.Fatal(err) + } + if errorResponse.Error.Code == nil || *errorResponse.Error.Code != "compaction_failed" { + t.Fatalf("unexpected error: %+v", errorResponse) + } + paths, _ := capture.snapshot() + if len(paths) != 2 { + t.Fatalf("expected exactly two attempts, got %d", len(paths)) + } +} + +func TestResponsesCompactionPayloadIsExpandedBeforeCloudPassthrough(t *testing.T) { + local, capture := newCompactionTestServer(t, func(_ int, w http.ResponseWriter, _ *http.Request, _ []byte) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"resp_next","object":"response","status":"completed","model":"fixture","output":[],"usage":null}`)) + }) + payload, err := json.Marshal(openai.OllamaCompactionPayload{ + Type: openai.OllamaCompactionPayloadType, Version: openai.OllamaCompactionPayloadVersion, + Summary: "The build is ready.", + }) + if err != nil { + t.Fatal(err) + } + request, err := json.Marshal(map[string]any{ + "model": "fixture:cloud", "stream": false, + "input": []any{ + map[string]any{"type": "message", "role": "user", "content": "first retained user turn"}, + map[string]any{"type": "message", "role": "user", "content": "second retained user turn"}, + openai.ResponsesCompactionItem{Type: "compaction", EncryptedContent: string(payload)}, + map[string]any{"type": "message", "role": "user", "content": "new turn"}, + }, + }) + if err != nil { + t.Fatal(err) + } + + status, _, body := postCompactionRequest(t, local, "/v1/responses", string(request)) + if status != http.StatusOK { + t.Fatalf("status=%d body=%s", status, body) + } + paths, bodies := capture.snapshot() + if len(paths) != 1 || paths[0] != "/v1/responses" { + t.Fatalf("paths=%v", paths) + } + forwarded := string(bodies[0]) + if strings.Contains(forwarded, `"type":"compaction"`) { + t.Fatalf("compaction boundary was forwarded: %s", forwarded) + } + for _, marker := range []string{"first retained user turn", "second retained user turn"} { + if strings.Count(forwarded, marker) != 1 { + t.Fatalf("retained user message %q was lost or duplicated: %s", marker, forwarded) + } + } + if !strings.Contains(forwarded, "The build is ready.") || !strings.Contains(forwarded, "new turn") || !strings.Contains(forwarded, "ollama_compaction_summary") { + t.Fatalf("expanded state is incomplete: %s", forwarded) + } +} diff --git a/server/routes.go b/server/routes.go index 263c7e27a..63a889761 100644 --- a/server/routes.go +++ b/server/routes.go @@ -1922,7 +1922,8 @@ func (s *Server) GenerateRoutes() (http.Handler, error) { r.POST("/v1/embeddings", cloudPassthroughMiddleware(cloudErrRemoteInferenceUnavailable), middleware.EmbeddingsMiddleware(), s.EmbedHandler) r.GET("/v1/models", middleware.ListMiddleware(), s.ListHandler) r.GET("/v1/models/:model", cloudModelPathPassthroughMiddleware(cloudErrRemoteModelDetailsUnavailable), middleware.RetrieveMiddleware(), s.ShowHandler) - r.POST("/v1/responses", s.withInferenceRequestLogging("/v1/responses", cloudPassthroughMiddleware(cloudErrRemoteInferenceUnavailable), middleware.ResponsesMiddleware(), s.ChatHandler)...) + r.POST("/v1/responses", s.withInferenceRequestLogging("/v1/responses", s.responsesCompactionMiddleware(), cloudPassthroughMiddleware(cloudErrRemoteInferenceUnavailable), middleware.ResponsesMiddleware(), s.ChatHandler)...) + r.POST("/v1/responses/compact", s.ResponsesCompactHandler) // OpenAI-compatible audio endpoint r.POST("/v1/audio/transcriptions", middleware.TranscriptionMiddleware(), s.ChatHandler)