From b1b6408625f5b1975b117d1115e5844c1a6c82c4 Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Tue, 25 Aug 2026 09:10:51 +0000 Subject: [PATCH 1/4] fix: switch OpenAI backend to Responses API Genkit routes OpenAI through the legacy Chat Completions endpoint, preventing Captain from using newer Responses API capabilities. Register a direct official SDK adapter that preserves Captain's streaming, structured output, caller-tool, usage, and durable approval contracts while leaving the other API backends on Genkit. Amp-Thread-ID: https://ampcode.com/threads/T-01a037fc-c3f8-73b6-89ca-149adf6cf1f2 --- pkg/ai/provider/genkit/genkit.go | 12 +- pkg/ai/provider/init.go | 6 +- pkg/ai/provider/openai/approval.go | 225 ++++++++++++++++++++++ pkg/ai/provider/openai/input.go | 264 +++++++++++++++++++++++++ pkg/ai/provider/openai/provider.go | 300 +++++++++++++++++++++++++++++ pkg/ai/provider/openai/tools.go | 188 ++++++++++++++++++ pkg/aimock/e2e_codex_cli_test.go | 5 +- pkg/aimock/genkit_test.go | 4 +- pkg/aimock/openaimock/server.go | 2 +- 9 files changed, 989 insertions(+), 17 deletions(-) create mode 100644 pkg/ai/provider/openai/approval.go create mode 100644 pkg/ai/provider/openai/input.go create mode 100644 pkg/ai/provider/openai/provider.go create mode 100644 pkg/ai/provider/openai/tools.go diff --git a/pkg/ai/provider/genkit/genkit.go b/pkg/ai/provider/genkit/genkit.go index 3e856030..a108b499 100644 --- a/pkg/ai/provider/genkit/genkit.go +++ b/pkg/ai/provider/genkit/genkit.go @@ -1,11 +1,7 @@ -// Package genkit implements captain's API-backed providers (Anthropic, OpenAI, -// Gemini) on top of Firebase Genkit, replacing the per-SDK providers. One -// Provider type serves all three backends; the plugin and model ref are chosen -// from ai.Config.Backend/Model. -// -// The exported signatures (New, the four interface methods) are FIXED — -// pkg/ai/provider/init.go registers genkit.New for the API backends against -// these signatures. +// Package genkit implements Captain's Genkit-backed API providers. The runtime +// registry uses it for Anthropic, Gemini, and DeepSeek; OpenAI compatibility is +// retained for direct consumers while Captain's OpenAI runtime uses its official +// SDK and Responses API adapter. package genkit import ( diff --git a/pkg/ai/provider/init.go b/pkg/ai/provider/init.go index 12236799..2f7b3830 100644 --- a/pkg/ai/provider/init.go +++ b/pkg/ai/provider/init.go @@ -5,6 +5,7 @@ import ( "github.com/flanksource/captain/pkg/ai/provider/claudeagent" "github.com/flanksource/captain/pkg/ai/provider/cmux" "github.com/flanksource/captain/pkg/ai/provider/genkit" + "github.com/flanksource/captain/pkg/ai/provider/openai" // Register the sandbox adapters so api.NewSandbox can construct them for // the CLI exec seam (newSandboxedCommand). @@ -13,9 +14,10 @@ import ( func init() { ai.RegisterRuntimeProbe(ai.BackendClaudeAgent, claudeagent.ProbeRuntime) - // API backends are served by Firebase Genkit (replaces the per-SDK providers). + // Anthropic, Gemini, and DeepSeek remain on Genkit; OpenAI uses its official + // SDK directly so Captain can use the Responses API. ai.RegisterProvider(ai.BackendAnthropic, func(cfg ai.Config) (ai.Provider, error) { return genkit.New(cfg) }) - ai.RegisterProvider(ai.BackendOpenAI, func(cfg ai.Config) (ai.Provider, error) { return genkit.New(cfg) }) + ai.RegisterProvider(ai.BackendOpenAI, func(cfg ai.Config) (ai.Provider, error) { return openai.New(cfg) }) ai.RegisterProvider(ai.BackendGemini, func(cfg ai.Config) (ai.Provider, error) { return genkit.New(cfg) }) // DeepSeek exposes an OpenAI-compatible API; genkit serves it via compat_oai // with a custom base URL (see pluginFor). diff --git a/pkg/ai/provider/openai/approval.go b/pkg/ai/provider/openai/approval.go new file mode 100644 index 00000000..f106496d --- /dev/null +++ b/pkg/ai/provider/openai/approval.go @@ -0,0 +1,225 @@ +package openai + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/flanksource/captain/pkg/ai" + "github.com/flanksource/captain/pkg/api" + + openaisdk "github.com/openai/openai-go" + "github.com/openai/openai-go/packages/param" + "github.com/openai/openai-go/responses" +) + +const ( + approvalCheckpointCodec = "openai-responses-input-json" + approvalCheckpointVersion = 1 +) + +// approvalCheckpoint is the durable, stateless Responses input needed to resume +// after Captain persists a tool-approval interruption. +type approvalCheckpoint struct { + Instructions string `json:"instructions,omitempty"` + Input responses.ResponseInputParam `json:"input"` +} + +func approvalState( + req ai.Request, + instructions param.Opt[string], + history responses.ResponseInputParam, + response *responses.Response, + calls []functionCall, + resolved []resolvedCall, +) (*api.ToolApprovalState, error) { + messages, err := approvalRequestMessages(req) + if err != nil { + return nil, err + } + assistant := responseAssistantMessage(response.Output) + checkpoint, err := encodeApprovalCheckpoint(instructions, history) + if err != nil { + return nil, err + } + state := &api.ToolApprovalState{ + Messages: append(messages, assistant), + Calls: make([]api.ToolApprovalCall, 0, len(calls)), + ProviderCheckpoint: checkpoint, + } + for i, call := range calls { + entry := api.ToolApprovalCall{Request: api.ToolApprovalRequest{ + ToolCallID: call.ID, Tool: call.Name, Input: json.RawMessage(call.Arguments), + }} + if resolved[i].result != nil { + entry.Result = resolved[i].result + } + state.Calls = append(state.Calls, entry) + } + if err := state.Validate(); err != nil { + return nil, fmt.Errorf("openai approval state: %w", err) + } + return state, nil +} + +func approvalRequestMessages(req ai.Request) ([]api.Message, error) { + if len(req.Messages) > 0 { + return append([]api.Message(nil), req.Messages...), nil + } + messages := make([]api.Message, 0, 2) + if req.Prompt.System != "" { + messages = append(messages, api.Message{Role: api.RoleSystem, Parts: []api.Part{{Type: api.PartText, Text: req.Prompt.System}}}) + } + parts := make([]api.Part, 0, len(req.Prompt.Attachments)+1) + if req.Prompt.User != "" { + parts = append(parts, api.Part{Type: api.PartText, Text: req.Prompt.User}) + } + for i := range req.Prompt.Attachments { + attachment := req.Prompt.Attachments[i] + parts = append(parts, api.Part{Type: api.PartAttachment, Attachment: &attachment}) + } + if len(parts) == 0 { + return nil, fmt.Errorf("openai approval state has no user prompt") + } + return append(messages, api.Message{Role: api.RoleUser, Parts: parts}), nil +} + +func responseAssistantMessage(items []responses.ResponseOutputItemUnion) api.Message { + message := api.Message{Role: api.RoleAssistant} + for _, item := range items { + switch item.Type { + case "message": + for _, content := range item.Content { + if content.Type == "output_text" && content.Text != "" { + message.Parts = append(message.Parts, api.Part{Type: api.PartText, Text: content.Text}) + } + } + case "reasoning": + for _, summary := range item.Summary { + if summary.Text != "" { + message.Parts = append(message.Parts, api.Part{Type: api.PartReasoning, Text: summary.Text}) + } + } + case "function_call": + message.Parts = append(message.Parts, api.Part{Type: api.PartToolRequest, ToolRequest: &api.ToolRequest{ + ToolCallID: item.CallID, Name: item.Name, Input: json.RawMessage(item.Arguments), + }}) + } + } + return message +} + +// The checkpoint persists the complete native input, including encrypted +// reasoning items, so an approval can resume without provider-side storage. +func encodeApprovalCheckpoint(instructions param.Opt[string], input responses.ResponseInputParam) (*api.ProviderCheckpoint, error) { + checkpoint := approvalCheckpoint{Input: input} + if instructions.Valid() { + checkpoint.Instructions = instructions.Value + } + payload, err := json.Marshal(checkpoint) + if err != nil { + return nil, fmt.Errorf("encode OpenAI approval checkpoint: %w", err) + } + return &api.ProviderCheckpoint{Codec: approvalCheckpointCodec, Version: approvalCheckpointVersion, Payload: payload}, nil +} + +func decodeApprovalCheckpoint(resume *api.ToolApprovalResume) (param.Opt[string], responses.ResponseInputParam, error) { + if resume == nil { + return param.Opt[string]{}, nil, fmt.Errorf("OpenAI tool approval resume is required") + } + if err := resume.Validate(); err != nil { + return param.Opt[string]{}, nil, err + } + checkpoint := resume.State.ProviderCheckpoint + if checkpoint == nil { + return param.Opt[string]{}, nil, fmt.Errorf("OpenAI approval checkpoint is missing") + } + if checkpoint.Codec != approvalCheckpointCodec || checkpoint.Version != approvalCheckpointVersion { + return param.Opt[string]{}, nil, fmt.Errorf("unsupported OpenAI approval checkpoint %q version %d", checkpoint.Codec, checkpoint.Version) + } + var decoded approvalCheckpoint + if err := json.Unmarshal(checkpoint.Payload, &decoded); err != nil { + return param.Opt[string]{}, nil, fmt.Errorf("decode OpenAI approval checkpoint: %w", err) + } + var instructions param.Opt[string] + if decoded.Instructions != "" { + instructions = openaisdk.String(decoded.Instructions) + } + return instructions, decoded.Input, nil +} + +// resumeCalls converts persisted decisions into function outputs. Only approved +// pending calls execute locally; resolved siblings and external responses are +// never run a second time. +func (p *Provider) resumeCalls(ctx context.Context, resume *api.ToolApprovalResume, state *requestState, out chan<- ai.Event) ([]responses.ResponseInputItemUnionParam, error) { + decisions := make(map[string]api.ToolApprovalDecision, len(resume.Decisions)) + for _, decision := range resume.Decisions { + decisions[decision.ToolCallID] = decision + } + checkpointCalls := checkpointFunctionCalls(state.history) + outputs := make([]responses.ResponseInputItemUnionParam, 0, len(resume.State.Calls)) + for _, call := range resume.State.Calls { + native, ok := checkpointCalls[call.Request.ToolCallID] + if !ok || native.Name != call.Request.Tool || !equalJSON([]byte(native.Arguments), call.Request.Input) { + return nil, fmt.Errorf("OpenAI approval checkpoint does not match tool call %q", call.Request.ToolCallID) + } + if call.Result != nil { + outputs = append(outputs, responses.ResponseInputItemParamOfFunctionCallOutput(call.Request.ToolCallID, resultOutput(call.Result))) + continue + } + decision := decisions[call.Request.ToolCallID] + switch decision.Action { + case api.ToolApprovalDeny: + reason := decision.Message + if reason == "" { + reason = "tool call denied" + } + outputs = append(outputs, responses.ResponseInputItemParamOfFunctionCallOutput(call.Request.ToolCallID, jsonText(map[string]any{"denied": true, "reason": reason}))) + case api.ToolApprovalRespond: + outputs = append(outputs, responses.ResponseInputItemParamOfFunctionCallOutput(call.Request.ToolCallID, resultOutput(decision.Result))) + case api.ToolApprovalApprove: + definition, ok := state.byName[call.Request.Tool] + if !ok { + return nil, fmt.Errorf("approved OpenAI function %q is no longer available", call.Request.Tool) + } + raw := call.Request.Input + if len(decision.Input) > 0 { + raw = decision.Input + } + args, err := callArguments(string(raw)) + if err != nil { + return nil, fmt.Errorf("approved OpenAI function %q: %w", call.Request.Tool, err) + } + if out != nil { + emit(ctx, out, ai.Event{Kind: ai.EventToolUse, Tool: call.Request.Tool, Input: args, ToolCallID: call.Request.ToolCallID, Model: p.model}) + } + value, err := definition.Handler(ctx, args) + if err != nil { + if out != nil { + emit(ctx, out, ai.Event{Kind: ai.EventToolResult, Tool: call.Request.Tool, ToolCallID: call.Request.ToolCallID, Success: false, Text: err.Error(), Model: p.model}) + } + value = map[string]any{"error": err.Error()} + } else if out != nil { + emit(ctx, out, ai.Event{Kind: ai.EventToolResult, Tool: call.Request.Tool, ToolCallID: call.Request.ToolCallID, Success: true, Text: toolOutputText(value), Model: p.model}) + } + outputs = append(outputs, responses.ResponseInputItemParamOfFunctionCallOutput(call.Request.ToolCallID, modelOutput(value))) + } + } + return outputs, nil +} + +func checkpointFunctionCalls(input responses.ResponseInputParam) map[string]functionCall { + calls := map[string]functionCall{} + for _, item := range input { + if item.OfFunctionCall != nil { + call := item.OfFunctionCall + calls[call.CallID] = functionCall{ID: call.CallID, Name: call.Name, Arguments: call.Arguments} + } + } + return calls +} + +func equalJSON(left, right []byte) bool { + var a, b any + return json.Unmarshal(left, &a) == nil && json.Unmarshal(right, &b) == nil && jsonText(a) == jsonText(b) +} diff --git a/pkg/ai/provider/openai/input.go b/pkg/ai/provider/openai/input.go new file mode 100644 index 00000000..e60213fb --- /dev/null +++ b/pkg/ai/provider/openai/input.go @@ -0,0 +1,264 @@ +package openai + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "os" + "strings" + + "github.com/flanksource/captain/pkg/ai" + captools "github.com/flanksource/captain/pkg/ai/tools" + "github.com/flanksource/captain/pkg/api" + + openaisdk "github.com/openai/openai-go" + "github.com/openai/openai-go/responses" +) + +type requestState struct { + params responses.ResponseNewParams + history responses.ResponseInputParam + byName map[string]api.ToolDefinition + resume *api.ToolApprovalResume +} + +func (p *Provider) prepare(req ai.Request) (*requestState, error) { + if err := req.ValidateRequestMode(); err != nil { + return nil, err + } + if err := api.RequireToolPolicySupport(ai.BackendOpenAI, req.Permissions); err != nil { + return nil, err + } + + definitions, err := captools.ResolveDefinitions(p.cfg.Tools, captools.ResolveOptions{ + Preferences: req.ToolPreferences, + Policy: req.ToolPolicy, + }) + if err != nil { + return nil, err + } + + state := &requestState{ + params: responses.ResponseNewParams{ + Model: openaisdk.ResponsesModel(p.model), + Store: openaisdk.Bool(false), + Include: []responses.ResponseIncludable{responses.ResponseIncludableReasoningEncryptedContent}, + }, + byName: make(map[string]api.ToolDefinition, len(definitions)), + } + for _, definition := range definitions { + state.byName[definition.Name] = definition + tool := responses.ToolParamOfFunction(definition.Name, toolSchema(definition), definition.Strict != nil && *definition.Strict) + if definition.Description != "" { + tool.OfFunction.Description = openaisdk.String(definition.Description) + } + state.params.Tools = append(state.params.Tools, tool) + } + + generation := ai.EffortConfig(ai.BackendOpenAI, p.model, req.Effort, req.Budget.MaxTokens, req.Temperature) + if value, ok := generation["temperature"].(float64); ok { + state.params.Temperature = openaisdk.Float(value) + } + if value, ok := generation["reasoning_effort"].(string); ok { + state.params.Reasoning.Effort = responses.ReasoningEffort(value) + state.params.Reasoning.Summary = responses.ReasoningSummaryAuto + } + if len(definitions) > 0 && (p.model == "gpt-5.6" || strings.HasPrefix(p.model, "gpt-5.6-")) { + state.params.Reasoning.Effort = responses.ReasoningEffort("none") + } + if req.Budget.MaxTokens > 0 { + state.params.MaxOutputTokens = openaisdk.Int(int64(req.Budget.MaxTokens)) + } + if req.Prompt.HasSchema() { + schema, err := ai.SchemaJSONForBackend(ai.BackendOpenAI, req.Prompt) + if err != nil { + return nil, fmt.Errorf("openai responses: cannot derive Prompt schema: %w", err) + } + var decoded map[string]any + if err := json.Unmarshal(schema, &decoded); err != nil { + return nil, fmt.Errorf("openai responses: invalid Prompt schema: %w", err) + } + format := responses.ResponseFormatTextConfigParamOfJSONSchema("captain_response", decoded) + format.OfJSONSchema.Strict = openaisdk.Bool(true) + state.params.Text.Format = format + } + + if req.ToolApproval != nil { + if err := validateMessageAttachments(p.cfg.Model, req.ToolApproval.State.Messages); err != nil { + return nil, err + } + instructions, history, err := decodeApprovalCheckpoint(req.ToolApproval) + if err != nil { + return nil, err + } + state.params.Instructions = instructions + state.history = history + state.resume = req.ToolApproval + return state, nil + } + + if len(req.Messages) > 0 { + if err := api.ValidateMessages(req.Messages); err != nil { + return nil, fmt.Errorf("canonical messages: %w", err) + } + if err := validateMessageAttachments(p.cfg.Model, req.Messages); err != nil { + return nil, err + } + state.history, err = conversationInput(req.Messages) + if err != nil { + return nil, err + } + return state, nil + } + + if err := ai.ValidateAttachmentCompatibility([]api.Model{p.cfg.Model}, req.Prompt.Attachments); err != nil { + return nil, err + } + if req.Prompt.System != "" { + state.params.Instructions = openaisdk.String(req.Prompt.System) + } + content, err := promptContent(req.Prompt) + if err != nil { + return nil, err + } + if len(content) > 0 { + state.history = append(state.history, messageInput(content, responses.EasyInputMessageRoleUser)) + } + return state, nil +} + +func toolSchema(definition api.ToolDefinition) map[string]any { + if definition.InputSchema != nil { + return definition.InputSchema + } + return map[string]any{"type": "object", "properties": map[string]any{}} +} + +func promptContent(prompt api.Prompt) (responses.ResponseInputMessageContentListParam, error) { + content := make(responses.ResponseInputMessageContentListParam, 0, len(prompt.Attachments)+1) + if prompt.User != "" { + content = append(content, responses.ResponseInputContentParamOfInputText(prompt.User)) + } + for i, attachment := range prompt.Attachments { + part, err := attachmentContent(attachment, fmt.Sprintf("attachment %d", i+1)) + if err != nil { + return nil, err + } + content = append(content, part) + } + return content, nil +} + +func conversationInput(messages []api.Message) (responses.ResponseInputParam, error) { + var input responses.ResponseInputParam + for i, message := range messages { + content := responses.ResponseInputMessageContentListParam{} + var calls []responses.ResponseInputItemUnionParam + for j, part := range message.Parts { + switch part.Type { + case api.PartText: + content = append(content, responses.ResponseInputContentParamOfInputText(part.Text)) + case api.PartReasoning: + // Provider summaries cannot be replayed as trusted reasoning without + // the Responses API's encrypted reasoning item. + case api.PartAttachment: + item, err := attachmentContent(*part.Attachment, fmt.Sprintf("message %d part %d attachment", i+1, j+1)) + if err != nil { + return nil, err + } + content = append(content, item) + case api.PartToolRequest: + calls = append(calls, responses.ResponseInputItemParamOfFunctionCall( + string(part.ToolRequest.Input), part.ToolRequest.ToolCallID, part.ToolRequest.Name, + )) + case api.PartToolResult: + calls = append(calls, responses.ResponseInputItemParamOfFunctionCallOutput( + part.ToolResult.ToolCallID, resultOutput(part.ToolResult), + )) + } + } + if len(content) > 0 { + input = append(input, messageInput(content, responseRole(message.Role))) + } + input = append(input, calls...) + } + return input, nil +} + +// messageInput writes the otherwise optional type discriminator because the +// durable approval checkpoint must be able to decode this SDK union later. +func messageInput(content responses.ResponseInputMessageContentListParam, role responses.EasyInputMessageRole) responses.ResponseInputItemUnionParam { + item := responses.ResponseInputItemParamOfMessage(content, role) + item.OfMessage.Type = responses.EasyInputMessageTypeMessage + return item +} + +func responseRole(role api.MessageRole) responses.EasyInputMessageRole { + switch role { + case api.RoleSystem: + return responses.EasyInputMessageRoleSystem + case api.RoleAssistant: + return responses.EasyInputMessageRoleAssistant + default: + return responses.EasyInputMessageRoleUser + } +} + +func attachmentContent(attachment api.AttachmentRef, label string) (responses.ResponseInputContentUnionParam, error) { + content, ok := attachment.PreparedContent() + if !ok { + return responses.ResponseInputContentUnionParam{}, fmt.Errorf("%s (%s) is not prepared", label, attachment.ID) + } + data := content.Bytes + if data == nil && content.Path != "" { + var err error + data, err = os.ReadFile(content.Path) + if err != nil { + return responses.ResponseInputContentUnionParam{}, fmt.Errorf("read prepared attachment %s: %w", attachment.ID, err) + } + } + uri := "data:" + attachment.MediaType + ";base64," + base64.StdEncoding.EncodeToString(data) + if strings.HasPrefix(attachment.MediaType, "image/") { + image := responses.ResponseInputContentParamOfInputImage(responses.ResponseInputImageDetailAuto) + image.OfInputImage.ImageURL = openaisdk.String(uri) + return image, nil + } + file := &responses.ResponseInputFileParam{FileData: openaisdk.String(uri)} + if attachment.Filename != "" { + file.Filename = openaisdk.String(attachment.Filename) + } + return responses.ResponseInputContentUnionParam{OfInputFile: file}, nil +} + +func validateMessageAttachments(model api.Model, messages []api.Message) error { + var attachments []api.AttachmentRef + for _, message := range messages { + for _, part := range message.Parts { + if part.Type == api.PartAttachment && part.Attachment != nil { + attachments = append(attachments, *part.Attachment) + } + } + } + return ai.ValidateAttachmentCompatibility([]api.Model{model}, attachments) +} + +func resultOutput(result *api.ToolResult) string { + if result == nil { + return "null" + } + if result.Error != "" { + return jsonText(map[string]any{"error": result.Error}) + } + return rawOutput(result.Output) +} + +func rawOutput(raw json.RawMessage) string { + if len(raw) == 0 { + return "null" + } + var text string + if json.Unmarshal(raw, &text) == nil { + return text + } + return string(raw) +} diff --git a/pkg/ai/provider/openai/provider.go b/pkg/ai/provider/openai/provider.go new file mode 100644 index 00000000..31d41142 --- /dev/null +++ b/pkg/ai/provider/openai/provider.go @@ -0,0 +1,300 @@ +// Package openai implements Captain's direct OpenAI API adapter. The OpenAI +// SDK stays behind Captain's provider-neutral contracts, and every model turn +// is sent through the Responses API with stateless input history. +package openai + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/flanksource/captain/pkg/ai" + "github.com/flanksource/captain/pkg/api" + + openaisdk "github.com/openai/openai-go" + "github.com/openai/openai-go/option" + "github.com/openai/openai-go/responses" +) + +const maxToolTurns = 16 + +// Provider implements Captain's streaming API provider over OpenAI Responses. +type Provider struct { + cfg ai.Config + client openaisdk.Client + model string +} + +var _ ai.StreamingProvider = (*Provider)(nil) +var _ api.ToolCapableProvider = (*Provider)(nil) + +// New constructs a direct OpenAI Responses provider. +func New(cfg ai.Config) (*Provider, error) { + backend := cfg.Model.Backend + if backend == "" { + var err error + backend, err = ai.InferBackend(cfg.Model.Name) + if err != nil { + return nil, err + } + } + if backend != ai.BackendOpenAI { + return nil, fmt.Errorf("openai provider does not support backend %q", backend) + } + if cfg.Model.Name == "" { + return nil, fmt.Errorf("openai provider: model cannot be empty") + } + + apiKey := cfg.APIKey + if apiKey == "" { + resolved, err := ai.ResolveAPIKey(backend) + if err != nil { + return nil, err + } + apiKey = resolved.Token + } + if apiKey == "" { + return nil, fmt.Errorf("%w: openai provider has no API key (set OPENAI_API_KEY)", ai.ErrNoAPIKey) + } + + cfg.Model.Backend = backend + cfg.Model.Name = ai.NormalizeModelForBackend(backend, cfg.Model.Name) + opts := []option.RequestOption{option.WithAPIKey(apiKey)} + if cfg.APIURL != "" { + opts = append(opts, option.WithBaseURL(cfg.APIURL)) + } + return &Provider{cfg: cfg, client: openaisdk.NewClient(opts...), model: cfg.Model.Name}, nil +} + +func (p *Provider) GetModel() string { return p.model } +func (p *Provider) GetBackend() ai.Backend { return ai.BackendOpenAI } + +// SupportsCallerTools reports that Captain can expose and execute caller tools +// in-process for this provider. +func (p *Provider) SupportsCallerTools() bool { return true } + +// Execute collects the streaming implementation into Captain's buffered +// response, binding structured JSON into the caller's Go target when present. +func (p *Provider) Execute(ctx context.Context, req ai.Request) (*ai.Response, error) { + start := time.Now() + events, err := p.ExecuteStream(ctx, req) + if err != nil { + return nil, err + } + + var text strings.Builder + var result *ai.Event + for event := range events { + switch event.Kind { + case ai.EventText: + text.WriteString(event.Text) + case ai.EventError: + if streamErr, ok := event.Raw.(error); ok { + return nil, fmt.Errorf("openai responses: %w", streamErr) + } + return nil, fmt.Errorf("openai responses: %s", event.Error) + case ai.EventResult: + copy := event + result = © + } + } + if result == nil { + return nil, fmt.Errorf("openai responses stream closed without a result") + } + + out := &ai.Response{ + Text: text.String(), + Model: p.model, + Backend: ai.BackendOpenAI, + CostUSD: result.CostUSD, + Duration: time.Since(start), + Raw: result.Raw, + ToolApproval: result.ToolApproval, + } + if result.Usage != nil { + out.Usage = *result.Usage + } + if len(result.StructuredData) > 0 { + if req.Prompt.Schema != nil { + if err := json.Unmarshal(result.StructuredData, req.Prompt.Schema); err != nil { + return nil, fmt.Errorf("%w: %v", ai.ErrSchemaValidation, err) + } + out.StructuredData = req.Prompt.Schema + out.Text = "" + } else { + out.StructuredData = result.StructuredData + out.Text = string(result.StructuredData) + } + } + return out, nil +} + +// ExecuteStream starts a Responses API run and drives any model/function loop +// locally, emitting Captain's provider-neutral lifecycle events. +func (p *Provider) ExecuteStream(ctx context.Context, req ai.Request) (<-chan ai.Event, error) { + prepared, err := p.prepare(req) + if err != nil { + return nil, err + } + + out := make(chan ai.Event, 16) + go func() { + defer close(out) + if err := p.run(ctx, req, prepared, out); err != nil { + emit(ctx, out, ai.Event{Kind: ai.EventError, Error: err.Error(), Model: p.model, Raw: err}) + } + }() + return out, nil +} + +func (p *Provider) run(ctx context.Context, req ai.Request, state *requestState, out chan<- ai.Event) error { + var usage ai.Usage + if state.resume != nil { + outputs, err := p.resumeCalls(ctx, state.resume, state, out) + if err != nil { + return err + } + state.history = append(state.history, outputs...) + state.resume = nil + } + for turn := 0; turn < maxToolTurns; turn++ { + state.params.Input.OfInputItemList = state.history + response, err := p.streamResponse(ctx, state.params, req.Prompt.HasSchema(), out) + if err != nil { + return err + } + usage = addUsage(usage, responseUsage(response.Usage)) + + output, err := responseOutputParams(response.Output) + if err != nil { + return err + } + state.history = append(state.history, output...) + calls := functionCalls(response.Output) + if len(calls) == 0 { + if refusal := responseRefusal(response.Output); refusal != "" { + return fmt.Errorf("openai response refused: %s", refusal) + } + var structured json.RawMessage + if req.Prompt.HasSchema() { + structured = json.RawMessage(response.OutputText()) + if !json.Valid(structured) { + return fmt.Errorf("%w: OpenAI returned invalid structured JSON", ai.ErrSchemaValidation) + } + } + cost := ai.PriceUsage(ai.BackendOpenAI, p.model, usage, 0).Total() + emit(ctx, out, ai.Event{ + Kind: ai.EventResult, Success: true, Usage: &usage, CostUSD: cost, + Model: p.model, StructuredData: structured, Raw: response, + }) + return nil + } + + outputs, approval, err := p.resolveCalls(ctx, req, state, response, calls, out) + if err != nil { + return err + } + if approval != nil { + cost := ai.PriceUsage(ai.BackendOpenAI, p.model, usage, 0).Total() + emit(ctx, out, ai.Event{ + Kind: ai.EventResult, Success: true, Usage: &usage, CostUSD: cost, + Model: p.model, ToolApproval: approval, Raw: response, + }) + return nil + } + state.history = append(state.history, outputs...) + } + return fmt.Errorf("openai responses exceeded the %d-turn tool limit", maxToolTurns) +} + +func (p *Provider) streamResponse(ctx context.Context, params responses.ResponseNewParams, structured bool, out chan<- ai.Event) (*responses.Response, error) { + stream := p.client.Responses.NewStreaming(ctx, params) + defer stream.Close() + + var completed *responses.Response + for stream.Next() { + event := stream.Current() + switch value := event.AsAny().(type) { + case responses.ResponseTextDeltaEvent: + if !structured && value.Delta != "" { + if !emit(ctx, out, ai.Event{Kind: ai.EventText, Text: value.Delta, Model: p.model}) { + return nil, ctx.Err() + } + } + case responses.ResponseReasoningSummaryTextDeltaEvent: + if value.Delta != "" { + if !emit(ctx, out, ai.Event{Kind: ai.EventThinking, Text: value.Delta, Model: p.model}) { + return nil, ctx.Err() + } + } + case responses.ResponseCompletedEvent: + response := value.Response + completed = &response + case responses.ResponseFailedEvent: + return nil, responseFailure(value.Response) + case responses.ResponseIncompleteEvent: + return nil, responseFailure(value.Response) + case responses.ResponseErrorEvent: + return nil, fmt.Errorf("OpenAI %s: %s", value.Code, value.Message) + } + } + if err := stream.Err(); err != nil { + return nil, p.normalizeError(ctx, err) + } + if completed == nil { + return nil, fmt.Errorf("OpenAI Responses API stream closed before response.completed") + } + return completed, nil +} + +func (p *Provider) normalizeError(ctx context.Context, err error) error { + if ctx.Err() != nil { + return fmt.Errorf("%w: %v", ai.ErrTimeout, ctx.Err()) + } + return fmt.Errorf("OpenAI Responses API: %w", err) +} + +func responseFailure(response responses.Response) error { + if response.Error.Message != "" { + return fmt.Errorf("OpenAI %s: %s", response.Error.Code, response.Error.Message) + } + if response.IncompleteDetails.Reason != "" { + return fmt.Errorf("OpenAI response incomplete: %s", response.IncompleteDetails.Reason) + } + if response.Status != "" { + return fmt.Errorf("OpenAI response ended with status %s", response.Status) + } + return fmt.Errorf("OpenAI response failed") +} + +func responseUsage(value responses.ResponseUsage) ai.Usage { + cached := int(value.InputTokensDetails.CachedTokens) + reasoning := int(value.OutputTokensDetails.ReasoningTokens) + return ai.Usage{ + InputTokens: ai.NetInputTokens(int(value.InputTokens), cached), + OutputTokens: ai.NetOutputTokens(int(value.OutputTokens), reasoning), + ReasoningTokens: reasoning, + CacheReadTokens: cached, + } +} + +func addUsage(left, right ai.Usage) ai.Usage { + return ai.Usage{ + InputTokens: left.InputTokens + right.InputTokens, + OutputTokens: left.OutputTokens + right.OutputTokens, + ReasoningTokens: left.ReasoningTokens + right.ReasoningTokens, + CacheReadTokens: left.CacheReadTokens + right.CacheReadTokens, + CacheWriteTokens: left.CacheWriteTokens + right.CacheWriteTokens, + } +} + +func emit(ctx context.Context, out chan<- ai.Event, event ai.Event) bool { + select { + case out <- event: + return true + case <-ctx.Done(): + return false + } +} diff --git a/pkg/ai/provider/openai/tools.go b/pkg/ai/provider/openai/tools.go new file mode 100644 index 00000000..91668bcd --- /dev/null +++ b/pkg/ai/provider/openai/tools.go @@ -0,0 +1,188 @@ +package openai + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/flanksource/captain/pkg/ai" + "github.com/flanksource/captain/pkg/api" + + "github.com/openai/openai-go/responses" +) + +type functionCall struct { + ID string + Name string + Arguments string +} + +type resolvedCall struct { + wire responses.ResponseInputItemUnionParam + result *api.ToolResult +} + +func functionCalls(items []responses.ResponseOutputItemUnion) []functionCall { + calls := make([]functionCall, 0) + for _, item := range items { + if item.Type == "function_call" { + calls = append(calls, functionCall{ID: item.CallID, Name: item.Name, Arguments: item.Arguments}) + } + } + return calls +} + +func (p *Provider) resolveCalls( + ctx context.Context, + req ai.Request, + state *requestState, + response *responses.Response, + calls []functionCall, + out chan<- ai.Event, +) ([]responses.ResponseInputItemUnionParam, *api.ToolApprovalState, error) { + resolved := make([]resolvedCall, 0, len(calls)) + pending := false + for _, call := range calls { + definition, ok := state.byName[call.Name] + if !ok { + return nil, nil, fmt.Errorf("OpenAI called unknown function %q", call.Name) + } + args, err := callArguments(call.Arguments) + if err != nil { + value := map[string]any{"error": err.Error()} + resolved = append(resolved, callResult(call.ID, value)) + continue + } + + emit(ctx, out, ai.Event{Kind: ai.EventToolUse, Tool: call.Name, Input: args, ToolCallID: call.ID, Model: p.model}) + if definition.NeedsApproval() { + emit(ctx, out, ai.Event{Kind: ai.EventPermission, Tool: call.Name, Input: args, ToolCallID: call.ID, Model: p.model}) + if p.cfg.CanUseTool == nil { + pending = true + resolved = append(resolved, resolvedCall{}) + continue + } + decision, decisionErr := p.cfg.CanUseTool(ctx, api.PermissionRequest{ + Tool: call.Name, Input: args, ToolUseID: call.ID, SessionID: p.cfg.SessionID, + }) + if decisionErr != nil || !decision.Allow { + reason := "tool call denied" + if decisionErr != nil { + reason = decisionErr.Error() + } else if decision.Message != "" { + reason = decision.Message + } + emit(ctx, out, ai.Event{Kind: ai.EventToolResult, Tool: call.Name, ToolCallID: call.ID, Success: false, Text: reason, Model: p.model}) + resolved = append(resolved, callResult(call.ID, map[string]any{"denied": true, "reason": reason})) + continue + } + if decision.UpdatedInput != nil { + args = decision.UpdatedInput + } + } + + value, err := definition.Handler(ctx, args) + if err != nil { + emit(ctx, out, ai.Event{Kind: ai.EventToolResult, Tool: call.Name, ToolCallID: call.ID, Success: false, Text: err.Error(), Model: p.model}) + resolved = append(resolved, callResult(call.ID, map[string]any{"error": err.Error()})) + continue + } + emit(ctx, out, ai.Event{Kind: ai.EventToolResult, Tool: call.Name, ToolCallID: call.ID, Success: true, Text: toolOutputText(value), Model: p.model}) + resolved = append(resolved, callResult(call.ID, value)) + } + + if pending { + approval, err := approvalState(req, state.params.Instructions, state.history, response, calls, resolved) + return nil, approval, err + } + outputs := make([]responses.ResponseInputItemUnionParam, 0, len(resolved)) + for _, call := range resolved { + outputs = append(outputs, call.wire) + } + return outputs, nil, nil +} + +func callArguments(raw string) (map[string]any, error) { + if raw == "" { + return map[string]any{}, nil + } + var args map[string]any + if err := json.Unmarshal([]byte(raw), &args); err != nil { + return nil, fmt.Errorf("invalid function arguments: %w", err) + } + if args == nil { + args = map[string]any{} + } + return args, nil +} + +func callResult(callID string, value any) resolvedCall { + encoded, err := json.Marshal(value) + if err != nil { + message := fmt.Sprintf("marshal tool result: %v", err) + value = map[string]any{"error": message} + encoded, _ = json.Marshal(value) + } + return resolvedCall{ + wire: responses.ResponseInputItemParamOfFunctionCallOutput(callID, modelOutput(value)), + result: &api.ToolResult{ToolCallID: callID, Output: encoded}, + } +} + +func modelOutput(value any) string { + if text, ok := value.(string); ok { + return text + } + return jsonText(value) +} + +func jsonText(value any) string { + data, err := json.Marshal(value) + if err != nil { + return fmt.Sprintf("%v", value) + } + return string(data) +} + +func toolOutputText(value any) string { + if value == nil { + return "" + } + return modelOutput(value) +} + +func responseRefusal(items []responses.ResponseOutputItemUnion) string { + for _, item := range items { + if item.Type != "message" { + continue + } + for _, content := range item.Content { + if content.Type == "refusal" && content.Refusal != "" { + return content.Refusal + } + } + } + return "" +} + +// responseOutputParams preserves message, function, and encrypted reasoning +// items so subsequent tool turns remain stateless without losing model context. +func responseOutputParams(items []responses.ResponseOutputItemUnion) ([]responses.ResponseInputItemUnionParam, error) { + output := make([]responses.ResponseInputItemUnionParam, 0, len(items)) + for _, item := range items { + switch item.Type { + case "message": + value := item.AsMessage().ToParam() + output = append(output, responses.ResponseInputItemUnionParam{OfOutputMessage: &value}) + case "function_call": + value := item.AsFunctionCall().ToParam() + output = append(output, responses.ResponseInputItemUnionParam{OfFunctionCall: &value}) + case "reasoning": + value := item.AsReasoning().ToParam() + output = append(output, responses.ResponseInputItemUnionParam{OfReasoning: &value}) + default: + return nil, fmt.Errorf("OpenAI returned unsupported output item %q", item.Type) + } + } + return output, nil +} diff --git a/pkg/aimock/e2e_codex_cli_test.go b/pkg/aimock/e2e_codex_cli_test.go index d94e4613..0cea4ce0 100644 --- a/pkg/aimock/e2e_codex_cli_test.go +++ b/pkg/aimock/e2e_codex_cli_test.go @@ -36,9 +36,8 @@ func TestE2ECodexCLI(t *testing.T) { assert.Contains(t, report(events), capitalAnswer) - // codex speaks the Responses API; genkit's openai plugin speaks Chat - // Completions. Both are served from the same scenario section, and this is - // the only spec that exercises the former. + // Codex and Captain's direct OpenAI backend both speak the Responses API; + // this spec exercises the external Codex binary path. servedPromptContaining(t, srv.Requests(), "/v1/responses", capitalPrompt) assert.Empty(t, srv.Remaining(), "the scenario must be played out") } diff --git a/pkg/aimock/genkit_test.go b/pkg/aimock/genkit_test.go index 2a6130f6..1a297a96 100644 --- a/pkg/aimock/genkit_test.go +++ b/pkg/aimock/genkit_test.go @@ -113,9 +113,7 @@ func TestOpenAIBackendHonoursAPIURL(t *testing.T) { assert.Equal(t, capitalAnswer, text) require.NotNil(t, result, "the stream must end with a result event") - // genkit's openai plugin speaks Chat Completions; /v1/responses is codex's - // wire API, exercised by the codex-cli e2e instead. - assert.Equal(t, capitalPrompt, servedPrompt(t, srv.Requests(), "/v1/chat/completions")) + assert.Equal(t, capitalPrompt, servedPrompt(t, srv.Requests(), "/v1/responses")) assert.Empty(t, srv.Remaining(), "the scenario must be played out") } diff --git a/pkg/aimock/openaimock/server.go b/pkg/aimock/openaimock/server.go index 634ec4aa..ef7658d5 100644 --- a/pkg/aimock/openaimock/server.go +++ b/pkg/aimock/openaimock/server.go @@ -1,4 +1,4 @@ -// ABOUTME: A mock OpenAI API — the endpoint codex and captain's genkit openai/deepseek backends talk to. +// ABOUTME: A mock OpenAI API — the endpoint codex and Captain's OpenAI-compatible API backends talk to. // ABOUTME: Standalone: Start it, export Env(), and a real `codex` binary runs against scripted replies. package openaimock From cffc0893e40e8fff6114448dc644e9020761c76f Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Tue, 25 Aug 2026 09:45:18 +0000 Subject: [PATCH 2/4] fix(ai): address OpenAI Responses review findings Preserve configured GPT-5.6 reasoning effort when caller tools are present and avoid overflow-prone slice capacity arithmetic flagged by CodeQL. Poll the mock journal for completed streaming requests so the E2E surface assertion cannot race the server's final journal write. Amp-Thread-ID: https://ampcode.com/threads/T-01a037fc-c3f8-73b6-89ca-149adf6cf1f2 --- pkg/ai/provider/genkit/options.go | 8 -------- pkg/ai/provider/openai/approval.go | 2 +- pkg/ai/provider/openai/input.go | 5 +---- pkg/aimock/e2e_claude_agent_test.go | 2 +- pkg/aimock/e2e_claude_cli_test.go | 2 +- pkg/aimock/e2e_codex_cli_test.go | 2 +- pkg/aimock/e2e_surfaces_test.go | 2 +- pkg/aimock/e2e_test.go | 4 ++-- pkg/aimock/genkit_test.go | 26 +++++++++++++++++--------- 9 files changed, 25 insertions(+), 28 deletions(-) diff --git a/pkg/ai/provider/genkit/options.go b/pkg/ai/provider/genkit/options.go index 6aaa44a0..c708f0a2 100644 --- a/pkg/ai/provider/genkit/options.go +++ b/pkg/ai/provider/genkit/options.go @@ -6,7 +6,6 @@ import ( "encoding/json" "fmt" "os" - "strings" "github.com/flanksource/captain/pkg/ai" captools "github.com/flanksource/captain/pkg/ai/tools" @@ -115,13 +114,6 @@ func generateOptions(p *Provider, req ai.Request, stream gkai.ModelStreamCallbac modelToken = req.ID } cfg := ai.EffortConfig(p.backend, modelToken, req.Effort, req.Budget.MaxTokens, req.Temperature) - model := bareModel(modelToken) - if p.backend == ai.BackendOpenAI && len(toolOptions) > 0 && (model == "gpt-5.6" || strings.HasPrefix(model, "gpt-5.6-")) { - if cfg == nil { - cfg = map[string]any{} - } - cfg["reasoning_effort"] = "none" - } if cfg != nil { opts = append(opts, gkai.WithConfig(cfg)) } diff --git a/pkg/ai/provider/openai/approval.go b/pkg/ai/provider/openai/approval.go index f106496d..5d70a8e0 100644 --- a/pkg/ai/provider/openai/approval.go +++ b/pkg/ai/provider/openai/approval.go @@ -70,7 +70,7 @@ func approvalRequestMessages(req ai.Request) ([]api.Message, error) { if req.Prompt.System != "" { messages = append(messages, api.Message{Role: api.RoleSystem, Parts: []api.Part{{Type: api.PartText, Text: req.Prompt.System}}}) } - parts := make([]api.Part, 0, len(req.Prompt.Attachments)+1) + parts := make([]api.Part, 0, len(req.Prompt.Attachments)) if req.Prompt.User != "" { parts = append(parts, api.Part{Type: api.PartText, Text: req.Prompt.User}) } diff --git a/pkg/ai/provider/openai/input.go b/pkg/ai/provider/openai/input.go index e60213fb..6665df6a 100644 --- a/pkg/ai/provider/openai/input.go +++ b/pkg/ai/provider/openai/input.go @@ -63,9 +63,6 @@ func (p *Provider) prepare(req ai.Request) (*requestState, error) { state.params.Reasoning.Effort = responses.ReasoningEffort(value) state.params.Reasoning.Summary = responses.ReasoningSummaryAuto } - if len(definitions) > 0 && (p.model == "gpt-5.6" || strings.HasPrefix(p.model, "gpt-5.6-")) { - state.params.Reasoning.Effort = responses.ReasoningEffort("none") - } if req.Budget.MaxTokens > 0 { state.params.MaxOutputTokens = openaisdk.Int(int64(req.Budget.MaxTokens)) } @@ -135,7 +132,7 @@ func toolSchema(definition api.ToolDefinition) map[string]any { } func promptContent(prompt api.Prompt) (responses.ResponseInputMessageContentListParam, error) { - content := make(responses.ResponseInputMessageContentListParam, 0, len(prompt.Attachments)+1) + content := make(responses.ResponseInputMessageContentListParam, 0, len(prompt.Attachments)) if prompt.User != "" { content = append(content, responses.ResponseInputContentParamOfInputText(prompt.User)) } diff --git a/pkg/aimock/e2e_claude_agent_test.go b/pkg/aimock/e2e_claude_agent_test.go index fa82ebe8..fdbfa0ca 100644 --- a/pkg/aimock/e2e_claude_agent_test.go +++ b/pkg/aimock/e2e_claude_agent_test.go @@ -34,6 +34,6 @@ func TestE2EClaudeAgent(t *testing.T) { require.NoError(t, err) assert.Contains(t, report(events), capitalAnswer) - servedPromptContaining(t, srv.Requests(), "/v1/messages", capitalPrompt) + servedPromptContaining(t, srv.Requests, "/v1/messages", capitalPrompt) assert.Empty(t, srv.Remaining(), "the scenario must be played out") } diff --git a/pkg/aimock/e2e_claude_cli_test.go b/pkg/aimock/e2e_claude_cli_test.go index 49857aa9..872b6528 100644 --- a/pkg/aimock/e2e_claude_cli_test.go +++ b/pkg/aimock/e2e_claude_cli_test.go @@ -49,7 +49,7 @@ func TestE2EClaudeCLI(t *testing.T) { assert.Equal(t, 8, result.Usage.OutputTokens) assert.NotEmpty(t, result.SessionID, "a real claude run always reports a session id") - servedPromptContaining(t, srv.Requests(), "/v1/messages", capitalPrompt) + servedPromptContaining(t, srv.Requests, "/v1/messages", capitalPrompt) assert.Empty(t, srv.Remaining(), "the scenario must be played out") } diff --git a/pkg/aimock/e2e_codex_cli_test.go b/pkg/aimock/e2e_codex_cli_test.go index 0cea4ce0..8348d611 100644 --- a/pkg/aimock/e2e_codex_cli_test.go +++ b/pkg/aimock/e2e_codex_cli_test.go @@ -38,7 +38,7 @@ func TestE2ECodexCLI(t *testing.T) { // Codex and Captain's direct OpenAI backend both speak the Responses API; // this spec exercises the external Codex binary path. - servedPromptContaining(t, srv.Requests(), "/v1/responses", capitalPrompt) + servedPromptContaining(t, srv.Requests, "/v1/responses", capitalPrompt) assert.Empty(t, srv.Remaining(), "the scenario must be played out") } diff --git a/pkg/aimock/e2e_surfaces_test.go b/pkg/aimock/e2e_surfaces_test.go index e34ed264..665b6b6d 100644 --- a/pkg/aimock/e2e_surfaces_test.go +++ b/pkg/aimock/e2e_surfaces_test.go @@ -79,7 +79,7 @@ func TestE2ESurfacesAgree(t *testing.T) { srv := startAnthropic(t, "text-only.yaml") assert.Contains(t, s.run(t, srv), capitalAnswer) - servedPromptContaining(t, srv.Requests(), "/v1/messages", capitalPrompt) + servedPromptContaining(t, srv.Requests, "/v1/messages", capitalPrompt) assert.Empty(t, srv.Remaining(), "the scenario must be played out") }) } diff --git a/pkg/aimock/e2e_test.go b/pkg/aimock/e2e_test.go index bcccdb5b..4fd100b1 100644 --- a/pkg/aimock/e2e_test.go +++ b/pkg/aimock/e2e_test.go @@ -61,9 +61,9 @@ func exportEnv(t *testing.T, env []string) { // carries want, and reports the whole served text. An agent CLI wraps the user's // question in system reminders and project context, so the served prompt is a // superset of what the caller asked for — never an exact match. -func servedPromptContaining(t *testing.T, served []aimock.Recorded, path, want string) string { +func servedPromptContaining(t *testing.T, requests func() []aimock.Recorded, path, want string) string { t.Helper() - prompt := servedPrompt(t, served, path) + prompt := servedPrompt(t, requests, path) require.Contains(t, prompt, want, "the mock must have seen the caller's question") return prompt } diff --git a/pkg/aimock/genkit_test.go b/pkg/aimock/genkit_test.go index 1a297a96..6eabd545 100644 --- a/pkg/aimock/genkit_test.go +++ b/pkg/aimock/genkit_test.go @@ -7,6 +7,7 @@ import ( "context" "strings" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -60,16 +61,23 @@ func drain(t *testing.T, events <-chan api.Event) (text string, result *api.Even // servedPrompt returns the last user text the mock saw on path, failing when the // generation never reached it — i.e. when the client called the real API instead. -func servedPrompt(t *testing.T, served []aimock.Recorded, path string) string { +func servedPrompt(t *testing.T, requests func() []aimock.Recorded, path string) string { t.Helper() - for i := len(served) - 1; i >= 0; i-- { - if served[i].Path == path { - require.Empty(t, served[i].Miss, "the mock answered %s with a miss", path) - return served[i].Request.LastUserText() + deadline := time.Now().Add(time.Second) + for { + served := requests() + for i := len(served) - 1; i >= 0; i-- { + if served[i].Path == path { + require.Empty(t, served[i].Miss, "the mock answered %s with a miss", path) + return served[i].Request.LastUserText() + } } + if time.Now().After(deadline) { + t.Fatalf("no request reached %s; the mock saw %+v", path, served) + return "" + } + time.Sleep(10 * time.Millisecond) } - t.Fatalf("no request reached %s; the mock saw %+v", path, served) - return "" } const capitalPrompt = "What is the capital of France?" @@ -95,7 +103,7 @@ func TestAnthropicBackendHonoursAPIURL(t *testing.T) { // The plugin lists models on init, so the generation is the last request, not // the only one. - assert.Equal(t, capitalPrompt, servedPrompt(t, srv.Requests(), "/v1/messages")) + assert.Equal(t, capitalPrompt, servedPrompt(t, srv.Requests, "/v1/messages")) assert.Empty(t, srv.Remaining(), "the scenario must be played out") } @@ -113,7 +121,7 @@ func TestOpenAIBackendHonoursAPIURL(t *testing.T) { assert.Equal(t, capitalAnswer, text) require.NotNil(t, result, "the stream must end with a result event") - assert.Equal(t, capitalPrompt, servedPrompt(t, srv.Requests(), "/v1/responses")) + assert.Equal(t, capitalPrompt, servedPrompt(t, srv.Requests, "/v1/responses")) assert.Empty(t, srv.Remaining(), "the scenario must be played out") } From 86d14237423238d382d2b1796154aba1e31f42ae Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Tue, 25 Aug 2026 16:05:17 +0000 Subject: [PATCH 3/4] fix(ai): preserve OpenAI output across approval resume The OpenAI SDK decodes persisted response messages as input messages because both share type message. This drops the output content and makes the resumed Responses request invalid. Tag checkpoint items by request or response union and restore response items from their raw wire form. Read restored function calls from the wire payload because SDK ToParam values keep their fields in override metadata. Amp-Thread-ID: https://ampcode.com/threads/T-01a037fc-c3f8-73b6-89ca-149adf6cf1f2 --- pkg/ai/provider/openai/approval.go | 70 ++++++++++++++++++++++++++---- 1 file changed, 62 insertions(+), 8 deletions(-) diff --git a/pkg/ai/provider/openai/approval.go b/pkg/ai/provider/openai/approval.go index 5d70a8e0..ccda1482 100644 --- a/pkg/ai/provider/openai/approval.go +++ b/pkg/ai/provider/openai/approval.go @@ -19,10 +19,18 @@ const ( ) // approvalCheckpoint is the durable, stateless Responses input needed to resume -// after Captain persists a tool-approval interruption. +// after Captain persists a tool-approval interruption. Output items remain raw +// so the SDK decodes them through its response union rather than its ambiguous +// request union. type approvalCheckpoint struct { - Instructions string `json:"instructions,omitempty"` - Input responses.ResponseInputParam `json:"input"` + Instructions string `json:"instructions,omitempty"` + Input []approvalCheckpointItem `json:"input"` +} + +// approvalCheckpointItem records which SDK union must decode each wire item. +type approvalCheckpointItem struct { + Request json.RawMessage `json:"request,omitempty"` + Output json.RawMessage `json:"output,omitempty"` } func approvalState( @@ -111,11 +119,24 @@ func responseAssistantMessage(items []responses.ResponseOutputItemUnion) api.Mes // The checkpoint persists the complete native input, including encrypted // reasoning items, so an approval can resume without provider-side storage. +// Response output stays tagged because message request and response variants +// share type:"message" and cannot safely round-trip through the SDK param union. func encodeApprovalCheckpoint(instructions param.Opt[string], input responses.ResponseInputParam) (*api.ProviderCheckpoint, error) { - checkpoint := approvalCheckpoint{Input: input} + checkpoint := approvalCheckpoint{Input: make([]approvalCheckpointItem, len(input))} if instructions.Valid() { checkpoint.Instructions = instructions.Value } + for i, item := range input { + payload, err := json.Marshal(item) + if err != nil { + return nil, fmt.Errorf("encode OpenAI approval checkpoint item %d: %w", i+1, err) + } + if item.OfOutputMessage != nil || item.OfFunctionCall != nil || item.OfReasoning != nil { + checkpoint.Input[i].Output = payload + } else { + checkpoint.Input[i].Request = payload + } + } payload, err := json.Marshal(checkpoint) if err != nil { return nil, fmt.Errorf("encode OpenAI approval checkpoint: %w", err) @@ -145,7 +166,30 @@ func decodeApprovalCheckpoint(resume *api.ToolApprovalResume) (param.Opt[string] if decoded.Instructions != "" { instructions = openaisdk.String(decoded.Instructions) } - return instructions, decoded.Input, nil + input := make(responses.ResponseInputParam, 0, len(decoded.Input)) + for i, item := range decoded.Input { + switch { + case len(item.Request) > 0 && len(item.Output) == 0: + var request responses.ResponseInputItemUnionParam + if err := json.Unmarshal(item.Request, &request); err != nil { + return param.Opt[string]{}, nil, fmt.Errorf("decode OpenAI approval checkpoint request item %d: %w", i+1, err) + } + input = append(input, request) + case len(item.Output) > 0 && len(item.Request) == 0: + var output responses.ResponseOutputItemUnion + if err := json.Unmarshal(item.Output, &output); err != nil { + return param.Opt[string]{}, nil, fmt.Errorf("decode OpenAI approval checkpoint output item %d: %w", i+1, err) + } + params, err := responseOutputParams([]responses.ResponseOutputItemUnion{output}) + if err != nil { + return param.Opt[string]{}, nil, fmt.Errorf("decode OpenAI approval checkpoint output item %d: %w", i+1, err) + } + input = append(input, params[0]) + default: + return param.Opt[string]{}, nil, fmt.Errorf("decode OpenAI approval checkpoint item %d: expected exactly one request or output payload", i+1) + } + } + return instructions, input, nil } // resumeCalls converts persisted decisions into function outputs. Only approved @@ -208,13 +252,23 @@ func (p *Provider) resumeCalls(ctx context.Context, resume *api.ToolApprovalResu return outputs, nil } +// checkpointFunctionCalls reads the wire form because SDK response parameters +// keep their values in override metadata rather than their exported fields. func checkpointFunctionCalls(input responses.ResponseInputParam) map[string]functionCall { calls := map[string]functionCall{} for _, item := range input { - if item.OfFunctionCall != nil { - call := item.OfFunctionCall - calls[call.CallID] = functionCall{ID: call.CallID, Name: call.Name, Arguments: call.Arguments} + if item.OfFunctionCall == nil { + continue + } + payload, err := json.Marshal(item) + if err != nil { + continue + } + var call responses.ResponseFunctionToolCall + if err := json.Unmarshal(payload, &call); err != nil || call.CallID == "" { + continue } + calls[call.CallID] = functionCall{ID: call.CallID, Name: call.Name, Arguments: call.Arguments} } return calls } From 4d3d06e0af9b3270c6cafc6586c5e6b39a56d6af Mon Sep 17 00:00:00 2001 From: Aditya Thebe Date: Wed, 26 Aug 2026 05:02:44 +0000 Subject: [PATCH 4/4] fix(ai): encode assistant Responses history correctly Canonical conversation replay encoded assistant text as input_text, which the Responses API rejects for the assistant role. Use the SDK's supported string content form for assistant history while retaining typed input_text parts for user and system messages. Amp-Thread-ID: https://ampcode.com/threads/T-01a037fc-c3f8-73b6-89ca-149adf6cf1f2 --- pkg/ai/provider/openai/input.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/pkg/ai/provider/openai/input.go b/pkg/ai/provider/openai/input.go index 6665df6a..f4c54a24 100644 --- a/pkg/ai/provider/openai/input.go +++ b/pkg/ai/provider/openai/input.go @@ -150,11 +150,16 @@ func conversationInput(messages []api.Message) (responses.ResponseInputParam, er var input responses.ResponseInputParam for i, message := range messages { content := responses.ResponseInputMessageContentListParam{} + var assistantText strings.Builder var calls []responses.ResponseInputItemUnionParam for j, part := range message.Parts { switch part.Type { case api.PartText: - content = append(content, responses.ResponseInputContentParamOfInputText(part.Text)) + if message.Role == api.RoleAssistant { + assistantText.WriteString(part.Text) + } else { + content = append(content, responses.ResponseInputContentParamOfInputText(part.Text)) + } case api.PartReasoning: // Provider summaries cannot be replayed as trusted reasoning without // the Responses API's encrypted reasoning item. @@ -174,6 +179,9 @@ func conversationInput(messages []api.Message) (responses.ResponseInputParam, er )) } } + if assistantText.Len() > 0 { + input = append(input, messageInput(assistantText.String(), responses.EasyInputMessageRoleAssistant)) + } if len(content) > 0 { input = append(input, messageInput(content, responseRole(message.Role))) } @@ -184,7 +192,7 @@ func conversationInput(messages []api.Message) (responses.ResponseInputParam, er // messageInput writes the otherwise optional type discriminator because the // durable approval checkpoint must be able to decode this SDK union later. -func messageInput(content responses.ResponseInputMessageContentListParam, role responses.EasyInputMessageRole) responses.ResponseInputItemUnionParam { +func messageInput[T string | responses.ResponseInputMessageContentListParam](content T, role responses.EasyInputMessageRole) responses.ResponseInputItemUnionParam { item := responses.ResponseInputItemParamOfMessage(content, role) item.OfMessage.Type = responses.EasyInputMessageTypeMessage return item