m1saka
9aba511abe
- Extract callUpstream helper to deduplicate ~30 lines between chatCompletions and responses - Move HTML templates to internal/server/templates/ via go:embed (server.go 1749→1192 lines) - Consolidate 4 copies of firstNonEmpty into util.FirstNonEmpty - Extract chatStreamChunk named type shared by aggregateStream and aggregateResponsesStream - Fix tool-call ordering: iterate sorted map keys instead of sequential 0..N - Map upstream finish_reason to Responses API status (length/content_filter → incomplete) - Surface /v1/models errors as 401/502 instead of silently returning empty 200 - Add Secure cookie flag via isTLSRequest helper - forEachSSEChunk: use sseDataPayload parser, drop redundant json.Valid, distinguish bufio.ErrTooLong - Fix StreamIdleTimout typo → StreamIdleTimeout - sso.go: single Read → io.ReadAll(io.LimitReader), explicit unknown error fallback
564 lines
18 KiB
Go
564 lines
18 KiB
Go
package server
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"sort"
|
|
"time"
|
|
|
|
"git.misaka.ren/M1saka/zhanlu_proxy/internal/openai"
|
|
)
|
|
|
|
// responses handles POST /v1/responses — the OpenAI Responses API. It
|
|
// translates the request to a chat-completions request for the upstream
|
|
// gateway, then translates the response back into the Responses format.
|
|
func (s *Server) responses(w http.ResponseWriter, r *http.Request) {
|
|
creds, err := s.currentCredentials()
|
|
if err != nil {
|
|
writeOpenAIError(w, http.StatusUnauthorized, "zhanlu credentials are not configured; open /login first", "auth_error", "missing_credentials")
|
|
return
|
|
}
|
|
|
|
body, err := io.ReadAll(r.Body)
|
|
if err != nil {
|
|
writeOpenAIError(w, http.StatusBadRequest, err.Error(), "invalid_request_error", "bad_body")
|
|
return
|
|
}
|
|
|
|
chatReq, err := openai.ParseResponsesRequest(body)
|
|
if err != nil {
|
|
writeOpenAIError(w, http.StatusBadRequest, err.Error(), "invalid_request_error", "bad_json")
|
|
return
|
|
}
|
|
meta := openai.ParseResponsesMeta(body)
|
|
if chatReq.Model == "" {
|
|
chatReq.Model = "zhanlu/auto"
|
|
}
|
|
|
|
start := time.Now()
|
|
clientWantsStream := chatReq.Stream
|
|
chatReq.Stream = true // upstream always streams
|
|
upstreamBody, err := chatReq.MarshalForUpstream()
|
|
if err != nil {
|
|
writeOpenAIError(w, http.StatusBadRequest, err.Error(), "invalid_request_error", "bad_body")
|
|
return
|
|
}
|
|
|
|
resp, ok := s.callUpstream(w, r, creds, chatReq.Model, upstreamBody, clientWantsStream, start)
|
|
if !ok {
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if clientWantsStream {
|
|
usage, status := s.proxyResponsesStream(w, resp, chatReq.Model, meta)
|
|
s.record(chatReq.Model, true, usage, status, start)
|
|
return
|
|
}
|
|
usage, status := s.aggregateResponsesStream(w, resp, chatReq.Model, meta)
|
|
s.record(chatReq.Model, false, usage, status, start)
|
|
}
|
|
|
|
// ---------------------------------------------------------------- non-streaming
|
|
|
|
// aggregateResponsesStream reads the upstream SSE, aggregates it (same logic
|
|
// as aggregateStream), then emits a Responses API JSON object.
|
|
func (s *Server) aggregateResponsesStream(w http.ResponseWriter, resp *http.Response, model string, meta openai.ResponsesMeta) (any, string) {
|
|
var content, reasoning, id string
|
|
var usage any
|
|
finishReason := "stop"
|
|
type toolCall struct {
|
|
ID string `json:"id"`
|
|
Type string `json:"type"`
|
|
Function struct {
|
|
Name string `json:"name"`
|
|
Arguments string `json:"arguments"`
|
|
} `json:"function"`
|
|
}
|
|
toolCalls := map[int]*toolCall{}
|
|
err := forEachSSEChunk(resp.Body, func(chunk []byte) error {
|
|
var event chatStreamChunk
|
|
if err := json.Unmarshal(chunk, &event); err != nil {
|
|
return err
|
|
}
|
|
if event.ID != "" {
|
|
id = event.ID
|
|
}
|
|
if event.Usage != nil {
|
|
usage = event.Usage
|
|
}
|
|
if len(event.Choices) > 0 {
|
|
content += event.Choices[0].Delta.Content
|
|
reasoning += event.Choices[0].Delta.ReasoningContent + event.Choices[0].Delta.Reasoning
|
|
for _, part := range event.Choices[0].Delta.ToolCalls {
|
|
call := toolCalls[part.Index]
|
|
if call == nil {
|
|
call = &toolCall{Type: "function"}
|
|
toolCalls[part.Index] = call
|
|
}
|
|
if part.ID != "" {
|
|
call.ID = part.ID
|
|
}
|
|
if part.Type != "" {
|
|
call.Type = part.Type
|
|
}
|
|
call.Function.Name += part.Function.Name
|
|
call.Function.Arguments += part.Function.Arguments
|
|
}
|
|
if event.Choices[0].FinishReason != nil {
|
|
finishReason = *event.Choices[0].FinishReason
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
writeOpenAIError(w, http.StatusBadGateway, err.Error(), "upstream_error", "zhanlu_stream_error")
|
|
return nil, "upstream_error"
|
|
}
|
|
|
|
_ = id
|
|
// Map the upstream finish_reason onto the Responses API status: a
|
|
// length/content_filter cutoff means the response is incomplete, not
|
|
// completed. Default ("stop"/"tool_calls") stays "completed".
|
|
status := "completed"
|
|
switch finishReason {
|
|
case "length", "content_filter":
|
|
status = "incomplete"
|
|
}
|
|
respID := "resp_" + randomRequestID()
|
|
output := []any{}
|
|
|
|
if reasoning != "" {
|
|
output = append(output, map[string]any{
|
|
"type": "reasoning", "id": "rs_" + randomRequestID(), "status": "completed",
|
|
"content": []any{},
|
|
"summary": []map[string]any{{"type": "summary_text", "text": reasoning}},
|
|
})
|
|
}
|
|
|
|
// Message output item — included when there is text content.
|
|
if content != "" {
|
|
output = append(output, map[string]any{
|
|
"type": "message", "id": "msg_" + randomRequestID(), "status": "completed",
|
|
"role": "assistant",
|
|
"content": []map[string]any{{
|
|
"type": "output_text", "text": content,
|
|
"annotations": []any{}, "logprobs": []any{},
|
|
}},
|
|
})
|
|
}
|
|
|
|
if len(toolCalls) > 0 {
|
|
indices := make([]int, 0, len(toolCalls))
|
|
for i := range toolCalls {
|
|
indices = append(indices, i)
|
|
}
|
|
sort.Ints(indices)
|
|
ordered := make([]*toolCall, 0, len(toolCalls))
|
|
for _, i := range indices {
|
|
ordered = append(ordered, toolCalls[i])
|
|
}
|
|
for _, tc := range ordered {
|
|
output = append(output, map[string]any{
|
|
"type": "function_call", "id": "fc_" + randomRequestID(),
|
|
"call_id": tc.ID, "name": tc.Function.Name,
|
|
"arguments": tc.Function.Arguments, "status": "completed",
|
|
})
|
|
}
|
|
}
|
|
|
|
// If there was no content and no tool calls, still emit an empty message
|
|
// so the response always has at least one output item.
|
|
if len(output) == 0 {
|
|
output = append(output, map[string]any{
|
|
"type": "message", "id": "msg_" + randomRequestID(), "status": "completed",
|
|
"role": "assistant", "content": []map[string]any{},
|
|
})
|
|
}
|
|
|
|
result := map[string]any{
|
|
"id": respID, "object": "response", "created_at": time.Now().Unix(),
|
|
"model": model, "status": status, "output": output,
|
|
"parallel_tool_calls": meta.ParallelToolCalls,
|
|
"tool_choice": meta.ToolChoice,
|
|
"tools": meta.Tools,
|
|
}
|
|
if usage != nil {
|
|
result["usage"] = openai.UsageToResponses(usage)
|
|
}
|
|
writeJSON(w, http.StatusOK, result)
|
|
return usage, "success"
|
|
}
|
|
|
|
// ------------------------------------------------------------------- streaming
|
|
|
|
// responsesStreamState tracks the lifecycle of output items while converting
|
|
// Chat Completions SSE into Responses API SSE events.
|
|
type responsesStreamState struct {
|
|
w http.ResponseWriter
|
|
flusher http.Flusher
|
|
respID string
|
|
msgID string
|
|
rsID string
|
|
meta openai.ResponsesMeta
|
|
|
|
outputIdx int
|
|
reasoningOn bool
|
|
reasonPartOn bool
|
|
messageOn bool
|
|
partOn bool
|
|
fullContent string
|
|
fullReasoning string
|
|
usage any
|
|
status string
|
|
finishReason string
|
|
|
|
// tool call tracking
|
|
tools map[int]*streamToolCall
|
|
toolOrder []int
|
|
}
|
|
|
|
type streamToolCall struct {
|
|
itemID string
|
|
callID string
|
|
name string
|
|
args string
|
|
started bool
|
|
doneIdx int // assigned output index when started
|
|
}
|
|
|
|
func newResponsesStreamState(w http.ResponseWriter, model string, meta openai.ResponsesMeta) *responsesStreamState {
|
|
flusher, _ := w.(http.Flusher)
|
|
w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.Header().Set("X-Accel-Buffering", "no")
|
|
w.WriteHeader(http.StatusOK)
|
|
st := &responsesStreamState{
|
|
w: w,
|
|
flusher: flusher,
|
|
respID: "resp_" + randomRequestID(),
|
|
msgID: "msg_" + randomRequestID(),
|
|
rsID: "rs_" + randomRequestID(),
|
|
meta: meta,
|
|
tools: map[int]*streamToolCall{},
|
|
status: "success",
|
|
}
|
|
baseResp := func(st2 string) map[string]any {
|
|
return map[string]any{
|
|
"id": st.respID, "object": "response", "created_at": time.Now().Unix(),
|
|
"model": model, "status": st2, "output": []any{},
|
|
"parallel_tool_calls": meta.ParallelToolCalls,
|
|
"tool_choice": meta.ToolChoice,
|
|
"tools": meta.Tools,
|
|
}
|
|
}
|
|
st.emit("response.created", map[string]any{"type": "response.created", "response": baseResp("in_progress")})
|
|
st.emit("response.in_progress", map[string]any{"type": "response.in_progress", "response": baseResp("in_progress")})
|
|
return st
|
|
}
|
|
|
|
func (st *responsesStreamState) emit(event string, data any) {
|
|
b, _ := json.Marshal(data)
|
|
fmt.Fprintf(st.w, "event: %s\ndata: %s\n\n", event, b)
|
|
if st.flusher != nil {
|
|
st.flusher.Flush()
|
|
}
|
|
}
|
|
|
|
// closeReasoning closes the reasoning output item if it is open.
|
|
func (st *responsesStreamState) closeReasoning() {
|
|
if !st.reasoningOn {
|
|
return
|
|
}
|
|
if st.reasonPartOn {
|
|
st.emit("response.reasoning_summary_text.done", map[string]any{
|
|
"type": "response.reasoning_summary_text.done", "item_id": st.rsID,
|
|
"output_index": st.outputIdx, "summary_index": 0, "text": st.fullReasoning,
|
|
})
|
|
st.emit("response.reasoning_summary_part.done", map[string]any{
|
|
"type": "response.reasoning_summary_part.done", "item_id": st.rsID,
|
|
"output_index": st.outputIdx, "summary_index": 0,
|
|
"part": map[string]any{"type": "summary_text", "text": st.fullReasoning},
|
|
})
|
|
st.reasonPartOn = false
|
|
}
|
|
st.emit("response.output_item.done", map[string]any{
|
|
"type": "response.output_item.done", "output_index": st.outputIdx,
|
|
"item": map[string]any{
|
|
"type": "reasoning", "id": st.rsID, "status": "completed",
|
|
"content": []any{},
|
|
"summary": []map[string]any{{"type": "summary_text", "text": st.fullReasoning}},
|
|
},
|
|
})
|
|
st.reasoningOn = false
|
|
st.outputIdx++
|
|
}
|
|
|
|
// closeMessage closes the message output item if it is open.
|
|
func (st *responsesStreamState) closeMessage() {
|
|
if !st.messageOn {
|
|
return
|
|
}
|
|
if st.partOn {
|
|
st.emit("response.output_text.done", map[string]any{
|
|
"type": "response.output_text.done", "item_id": st.msgID,
|
|
"output_index": st.outputIdx, "content_index": 0, "text": st.fullContent,
|
|
})
|
|
st.emit("response.content_part.done", map[string]any{
|
|
"type": "response.content_part.done", "item_id": st.msgID,
|
|
"output_index": st.outputIdx, "content_index": 0,
|
|
"part": map[string]any{"type": "output_text", "text": st.fullContent, "annotations": []any{}, "logprobs": []any{}},
|
|
})
|
|
st.partOn = false
|
|
}
|
|
st.emit("response.output_item.done", map[string]any{
|
|
"type": "response.output_item.done", "output_index": st.outputIdx,
|
|
"item": map[string]any{
|
|
"type": "message", "id": st.msgID, "status": "completed", "role": "assistant",
|
|
"content": []map[string]any{{"type": "output_text", "text": st.fullContent, "annotations": []any{}, "logprobs": []any{}}},
|
|
},
|
|
})
|
|
st.messageOn = false
|
|
st.outputIdx++
|
|
}
|
|
|
|
// handleReasoning processes a reasoning content delta.
|
|
func (st *responsesStreamState) handleReasoning(delta string) {
|
|
if !st.reasoningOn {
|
|
st.reasoningOn = true
|
|
st.emit("response.output_item.added", map[string]any{
|
|
"type": "response.output_item.added", "output_index": st.outputIdx,
|
|
"item": map[string]any{
|
|
"type": "reasoning", "id": st.rsID, "status": "in_progress",
|
|
"content": []any{}, "summary": []any{},
|
|
},
|
|
})
|
|
st.emit("response.reasoning_summary_part.added", map[string]any{
|
|
"type": "response.reasoning_summary_part.added", "item_id": st.rsID,
|
|
"output_index": st.outputIdx, "summary_index": 0,
|
|
"part": map[string]any{"type": "summary_text", "text": ""},
|
|
})
|
|
st.reasonPartOn = true
|
|
}
|
|
st.fullReasoning += delta
|
|
st.emit("response.reasoning_summary_text.delta", map[string]any{
|
|
"type": "response.reasoning_summary_text.delta", "item_id": st.rsID,
|
|
"output_index": st.outputIdx, "summary_index": 0, "delta": delta,
|
|
})
|
|
}
|
|
|
|
// handleContent processes a text content delta.
|
|
func (st *responsesStreamState) handleContent(delta string) {
|
|
// Close reasoning if open — text content comes after reasoning.
|
|
st.closeReasoning()
|
|
if !st.messageOn {
|
|
st.messageOn = true
|
|
st.emit("response.output_item.added", map[string]any{
|
|
"type": "response.output_item.added", "output_index": st.outputIdx,
|
|
"item": map[string]any{
|
|
"type": "message", "id": st.msgID, "status": "in_progress", "role": "assistant", "content": []any{},
|
|
},
|
|
})
|
|
st.emit("response.content_part.added", map[string]any{
|
|
"type": "response.content_part.added", "item_id": st.msgID,
|
|
"output_index": st.outputIdx, "content_index": 0,
|
|
"part": map[string]any{"type": "output_text", "text": "", "annotations": []any{}, "logprobs": []any{}},
|
|
})
|
|
st.partOn = true
|
|
}
|
|
st.fullContent += delta
|
|
st.emit("response.output_text.delta", map[string]any{
|
|
"type": "response.output_text.delta", "item_id": st.msgID,
|
|
"output_index": st.outputIdx, "content_index": 0, "delta": delta,
|
|
})
|
|
}
|
|
|
|
// handleToolCall processes a tool call delta from the chat completion stream.
|
|
func (st *responsesStreamState) handleToolCall(index int, id, name, args string) {
|
|
// Close reasoning/message if open — tool calls are separate output items.
|
|
st.closeReasoning()
|
|
st.closeMessage()
|
|
|
|
call := st.tools[index]
|
|
if call == nil {
|
|
call = &streamToolCall{itemID: "fc_" + randomRequestID()}
|
|
st.tools[index] = call
|
|
st.toolOrder = append(st.toolOrder, index)
|
|
}
|
|
if id != "" {
|
|
call.callID = id
|
|
}
|
|
if name != "" {
|
|
call.name += name
|
|
}
|
|
if !call.started {
|
|
call.started = true
|
|
call.doneIdx = st.outputIdx
|
|
st.emit("response.output_item.added", map[string]any{
|
|
"type": "response.output_item.added", "output_index": st.outputIdx,
|
|
"item": map[string]any{
|
|
"type": "function_call", "id": call.itemID, "call_id": call.callID,
|
|
"name": call.name, "arguments": "", "status": "in_progress",
|
|
},
|
|
})
|
|
st.outputIdx++
|
|
}
|
|
if args != "" {
|
|
call.args += args
|
|
st.emit("response.function_call_arguments.delta", map[string]any{
|
|
"type": "response.function_call_arguments.delta", "item_id": call.itemID,
|
|
"output_index": call.doneIdx, "delta": args,
|
|
})
|
|
}
|
|
}
|
|
|
|
// finish closes all open items and sends the response.completed event.
|
|
func (st *responsesStreamState) finish(model string) {
|
|
st.closeReasoning()
|
|
st.closeMessage()
|
|
for _, idx := range st.toolOrder {
|
|
call := st.tools[idx]
|
|
st.emit("response.function_call_arguments.done", map[string]any{
|
|
"type": "response.function_call_arguments.done", "item_id": call.itemID,
|
|
"output_index": call.doneIdx, "arguments": call.args,
|
|
})
|
|
st.emit("response.output_item.done", map[string]any{
|
|
"type": "response.output_item.done", "output_index": call.doneIdx,
|
|
"item": map[string]any{
|
|
"type": "function_call", "id": call.itemID, "call_id": call.callID,
|
|
"name": call.name, "arguments": call.args, "status": "completed",
|
|
},
|
|
})
|
|
}
|
|
|
|
// Build the final output array for the completed event.
|
|
finalOutput := []any{}
|
|
if st.fullReasoning != "" {
|
|
finalOutput = append(finalOutput, map[string]any{
|
|
"type": "reasoning", "id": st.rsID, "status": "completed",
|
|
"content": []any{},
|
|
"summary": []map[string]any{{"type": "summary_text", "text": st.fullReasoning}},
|
|
})
|
|
}
|
|
if st.fullContent != "" {
|
|
finalOutput = append(finalOutput, map[string]any{
|
|
"type": "message", "id": st.msgID, "status": "completed", "role": "assistant",
|
|
"content": []map[string]any{{"type": "output_text", "text": st.fullContent, "annotations": []any{}, "logprobs": []any{}}},
|
|
})
|
|
}
|
|
for _, idx := range st.toolOrder {
|
|
call := st.tools[idx]
|
|
finalOutput = append(finalOutput, map[string]any{
|
|
"type": "function_call", "id": call.itemID, "call_id": call.callID,
|
|
"name": call.name, "arguments": call.args, "status": "completed",
|
|
})
|
|
}
|
|
// Ensure at least one output item exists.
|
|
if len(finalOutput) == 0 {
|
|
finalOutput = append(finalOutput, map[string]any{
|
|
"type": "message", "id": st.msgID, "status": "completed", "role": "assistant",
|
|
"content": []map[string]any{},
|
|
})
|
|
}
|
|
|
|
// Map the upstream finish_reason onto the Responses API status, mirroring
|
|
// the non-streaming path: a length/content_filter cutoff is "incomplete".
|
|
respStatus := "completed"
|
|
switch st.finishReason {
|
|
case "length", "content_filter":
|
|
respStatus = "incomplete"
|
|
}
|
|
completedResp := map[string]any{
|
|
"id": st.respID, "object": "response", "created_at": time.Now().Unix(),
|
|
"model": model, "status": respStatus, "output": finalOutput,
|
|
"parallel_tool_calls": st.meta.ParallelToolCalls,
|
|
"tool_choice": st.meta.ToolChoice,
|
|
"tools": st.meta.Tools,
|
|
}
|
|
if st.usage != nil {
|
|
completedResp["usage"] = openai.UsageToResponses(st.usage)
|
|
}
|
|
st.emit("response.completed", map[string]any{"type": "response.completed", "response": completedResp})
|
|
}
|
|
|
|
// proxyResponsesStream reads the upstream Chat Completions SSE and emits
|
|
// Responses API SSE events via a responsesStreamState state machine.
|
|
func (s *Server) proxyResponsesStream(w http.ResponseWriter, resp *http.Response, model string, meta openai.ResponsesMeta) (any, string) {
|
|
st := newResponsesStreamState(w, model, meta)
|
|
|
|
reader := bufio.NewReader(resp.Body)
|
|
for {
|
|
line, err := reader.ReadString('\n')
|
|
if line != "" {
|
|
payload := sseDataPayload(line)
|
|
if payload != "" && payload != "[DONE]" {
|
|
var evt struct {
|
|
State string `json:"state"`
|
|
ErrorMessage string `json:"errorMessage"`
|
|
Usage any `json:"usage"`
|
|
Choices []struct {
|
|
Delta struct {
|
|
Content string `json:"content"`
|
|
ReasoningContent string `json:"reasoning_content"`
|
|
Reasoning string `json:"reasoning"`
|
|
ToolCalls []struct {
|
|
Index int `json:"index"`
|
|
ID string `json:"id"`
|
|
Function struct {
|
|
Name string `json:"name"`
|
|
Arguments string `json:"arguments"`
|
|
} `json:"function"`
|
|
} `json:"tool_calls"`
|
|
} `json:"delta"`
|
|
FinishReason *string `json:"finish_reason"`
|
|
} `json:"choices"`
|
|
}
|
|
if jErr := json.Unmarshal([]byte(payload), &evt); jErr == nil {
|
|
if evt.State == "ERROR" {
|
|
st.status = "upstream_error"
|
|
}
|
|
if evt.Usage != nil {
|
|
st.usage = evt.Usage
|
|
}
|
|
if len(evt.Choices) > 0 {
|
|
delta := evt.Choices[0].Delta
|
|
if evt.Choices[0].FinishReason != nil {
|
|
st.finishReason = *evt.Choices[0].FinishReason
|
|
}
|
|
reasoningDelta := delta.ReasoningContent + delta.Reasoning
|
|
if reasoningDelta != "" {
|
|
st.handleReasoning(reasoningDelta)
|
|
}
|
|
if delta.Content != "" {
|
|
st.handleContent(delta.Content)
|
|
}
|
|
for _, tc := range delta.ToolCalls {
|
|
st.handleToolCall(tc.Index, tc.ID, tc.Function.Name, tc.Function.Arguments)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if err != nil {
|
|
if err != io.EOF {
|
|
b, _ := json.Marshal(map[string]any{
|
|
"message": err.Error(), "type": "upstream_error", "code": "zhanlu_stream_error",
|
|
})
|
|
fmt.Fprintf(w, "data: %s\n\n", b)
|
|
if st.flusher != nil {
|
|
st.flusher.Flush()
|
|
}
|
|
st.status = "upstream_error"
|
|
}
|
|
break
|
|
}
|
|
}
|
|
|
|
st.finish(model)
|
|
return st.usage, st.status
|
|
}
|