Merge pull request #4 from feat/chat-completions-compat

apps/api/internal/httpapi/chat_completions_mode_test.go

@@ -50,7 +50,7 @@ func TestWriteCompatibleTaskResponseReturnsJSONWhenStreamIsFalse(t *testing.T) {
 	req := httptest.NewRequest(http.MethodPost, "/api/v1/chat/completions", nil)
 	recorder := httptest.NewRecorder()
 
-	writeCompatibleTaskResponse(context.Background(), recorder, req, executor, "chat.completions", "gpt-test", store.GatewayTask{ID: "task-test"}, &auth.User{}, false)
+	writeCompatibleTaskResponse(context.Background(), recorder, req, executor, "chat.completions", "gpt-test", store.GatewayTask{ID: "task-test"}, &auth.User{}, false, false)
 
 	if recorder.Code != http.StatusOK {
 		t.Fatalf("status=%d want=%d body=%s", recorder.Code, http.StatusOK, recorder.Body.String())
@@ -69,13 +69,13 @@ func TestWriteCompatibleTaskResponseReturnsJSONWhenStreamIsFalse(t *testing.T) {
 
 func TestWriteCompatibleTaskResponseReturnsSSEWhenStreamIsTrue(t *testing.T) {
 	executor := &fakeTaskExecutor{
-		deltas: []string{"hel", "lo"},
-		output: map[string]any{"id": "chatcmpl-test", "object": "chat.completion"},
+		deltas: []clients.StreamDeltaEvent{{Text: "hel"}, {Text: "lo"}},
+		output: map[string]any{"id": "chatcmpl-test", "object": "chat.completion", "usage": map[string]any{"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3}},
 	}
 	req := httptest.NewRequest(http.MethodPost, "/api/v1/chat/completions", nil)
 	recorder := httptest.NewRecorder()
 
-	writeCompatibleTaskResponse(context.Background(), recorder, req, executor, "chat.completions", "gpt-test", store.GatewayTask{ID: "task-test"}, &auth.User{}, true)
+	writeCompatibleTaskResponse(context.Background(), recorder, req, executor, "chat.completions", "gpt-test", store.GatewayTask{ID: "task-test"}, &auth.User{}, true, true)
 
 	if executor.executeCalls != 0 || executor.streamCalls != 1 {
 		t.Fatalf("expected stream execute only, got execute=%d stream=%d", executor.executeCalls, executor.streamCalls)
@@ -84,17 +84,53 @@ func TestWriteCompatibleTaskResponseReturnsSSEWhenStreamIsTrue(t *testing.T) {
 		t.Fatalf("Content-Type=%q want text/event-stream", contentType)
 	}
 	body := recorder.Body.String()
-	for _, want := range []string{"event: message", `"content":"hel"`, `"content":"lo"`, `"finish_reason":"stop"`} {
+	for _, want := range []string{`data: {`, `"role":"assistant"`, `"created":`, `"system_fingerprint":`, `"content":"hel"`, `"content":"lo"`, `"finish_reason":"stop"`, `"usage":{"completion_tokens":2,"prompt_tokens":1,"total_tokens":3}`, "data: [DONE]"} {
 		if !strings.Contains(body, want) {
 			t.Fatalf("SSE body missing %s: %s", want, body)
 		}
 	}
+	if strings.Contains(body, "event: message") {
+		t.Fatalf("chat completions stream should use OpenAI data-only SSE frames: %s", body)
+	}
+}
+
+func TestWriteCompatibleTaskResponseStreamsStructuredToolAndReasoningDeltas(t *testing.T) {
+	executor := &fakeTaskExecutor{
+		deltas: []clients.StreamDeltaEvent{
+			{Event: map[string]any{"id": "chatcmpl-upstream", "object": "chat.completion.chunk", "created": float64(1710000000), "model": "deepseek-v4", "system_fingerprint": "fp-test", "choices": []any{map[string]any{"index": float64(0), "delta": map[string]any{"reasoning_details": []any{map[string]any{"type": "reasoning.text", "text": "detail-"}, map[string]any{"type": "reasoning.summary", "summary": "summary"}, map[string]any{"type": "reasoning.encrypted", "data": "secret"}}}, "finish_reason": nil}}}},
+			{Event: map[string]any{"id": "chatcmpl-upstream", "object": "chat.completion.chunk", "created": float64(1710000000), "model": "deepseek-v4", "system_fingerprint": "fp-test", "choices": []any{map[string]any{"index": float64(0), "delta": map[string]any{"content": "<think>tagged</think>answer"}, "finish_reason": nil}}}},
+			{Event: map[string]any{"id": "chatcmpl-upstream", "object": "chat.completion.chunk", "created": float64(1710000000), "model": "deepseek-v4", "system_fingerprint": "fp-test", "choices": []any{map[string]any{"index": float64(0), "delta": map[string]any{"functionCall": map[string]any{"name": "legacy_lookup", "arguments": "{\"city\":\"Boston\"}"}}, "finish_reason": nil}}}},
+			{Event: map[string]any{"id": "chatcmpl-upstream", "object": "chat.completion.chunk", "created": float64(1710000000), "model": "deepseek-v4", "system_fingerprint": "fp-test", "choices": []any{map[string]any{"index": float64(0), "delta": map[string]any{"tool_calls": []any{map[string]any{"index": float64(0), "id": "call_1", "type": "function", "function": map[string]any{"name": "lookup", "arguments": "{\"q\":"}}}}, "finish_reason": nil}}}},
+			{Event: map[string]any{"id": "chatcmpl-upstream", "object": "chat.completion.chunk", "created": float64(1710000000), "model": "deepseek-v4", "system_fingerprint": "fp-test", "choices": []any{map[string]any{"index": float64(0), "delta": map[string]any{"tool_calls": []any{map[string]any{"index": float64(0), "function": map[string]any{"arguments": "\"weather\"}"}}}}, "finish_reason": "tool_calls"}}}},
+			{Event: map[string]any{"id": "chatcmpl-upstream", "object": "chat.completion.chunk", "created": float64(1710000000), "model": "deepseek-v4", "choices": []any{}, "usage": map[string]any{"prompt_tokens": float64(4), "completion_tokens": float64(5), "total_tokens": float64(9)}}},
+		},
+		output: map[string]any{"id": "chatcmpl-upstream", "object": "chat.completion", "model": "deepseek-v4"},
+	}
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/chat/completions", nil)
+	recorder := httptest.NewRecorder()
+
+	writeCompatibleTaskResponse(context.Background(), recorder, req, executor, "chat.completions", "gpt-test", store.GatewayTask{ID: "task-test"}, &auth.User{}, true, true)
+
+	body := recorder.Body.String()
+	roleIndex := strings.Index(body, `"role":"assistant"`)
+	reasoningIndex := strings.Index(body, `"reasoning_content":"detail-summary"`)
+	if roleIndex < 0 || reasoningIndex < 0 || roleIndex > reasoningIndex {
+		t.Fatalf("assistant role should be emitted before structured deltas: %s", body)
+	}
+	for _, want := range []string{`"system_fingerprint":"fp-test"`, `"created":1710000000`, `"reasoning_content":"tagged"`, `"content":"answer"`, `"tool_calls":[{"function":{"arguments":"{\"city\":\"Boston\"}","name":"legacy_lookup"}`, `"tool_calls":[{"function":{"arguments":"{\"q\":"`, `"finish_reason":"tool_calls"`, `"choices":[],"created":1710000000`, `"usage":{"completion_tokens":5,"prompt_tokens":4,"total_tokens":9}`, "data: [DONE]"} {
+		if !strings.Contains(body, want) {
+			t.Fatalf("SSE body missing %s: %s", want, body)
+		}
+	}
+	if strings.Contains(body, "reasoning_details") || strings.Contains(body, "<think>") || strings.Contains(body, "functionCall") {
+		t.Fatalf("provider-specific reasoning/tool fields should be converted away: %s", body)
+	}
 }
 
 type fakeTaskExecutor struct {
 	executeCalls int
 	streamCalls  int
-	deltas       []string
+	deltas       []clients.StreamDeltaEvent
 	output       map[string]any
 }
 

apps/api/internal/httpapi/handlers.go

@@ -934,7 +934,7 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler {
 		runCtx, cancelRun := s.requestExecutionContext(r)
 		defer cancelRun()
 		if responsePlan.compatibleMode {
-			writeCompatibleTaskResponse(runCtx, w, r, s.runner, kind, model, task, user, responsePlan.streamMode)
+			writeCompatibleTaskResponse(runCtx, w, r, s.runner, kind, model, task, user, responsePlan.streamMode, streamIncludeUsage(body))
 			return
 		}
 		result, runErr := s.runner.Execute(runCtx, task, user)
@@ -1002,14 +1002,15 @@ type taskExecutor interface {
 	ExecuteStream(context.Context, store.GatewayTask, *auth.User, clients.StreamDelta) (runner.Result, error)
 }
 
-func writeCompatibleTaskResponse(runCtx context.Context, w http.ResponseWriter, r *http.Request, executor taskExecutor, kind string, model string, task store.GatewayTask, user *auth.User, streamMode bool) {
+func writeCompatibleTaskResponse(runCtx context.Context, w http.ResponseWriter, r *http.Request, executor taskExecutor, kind string, model string, task store.GatewayTask, user *auth.User, streamMode bool, includeUsage bool) {
 	if streamMode {
 		flusher := prepareCompatibleStream(w)
-		result, runErr := executor.ExecuteStream(runCtx, task, user, func(delta string) error {
+		streamWriter := newCompatibleStreamWriter(kind, model, includeUsage)
+		result, runErr := executor.ExecuteStream(runCtx, task, user, func(delta clients.StreamDeltaEvent) error {
 			if !requestStillConnected(r) {
 				return nil
 			}
-			writeCompatibleDelta(w, kind, model, delta)
+			streamWriter.writeDelta(w, delta)
 			if flusher != nil {
 				flusher.Flush()
 			}
@@ -1043,7 +1044,7 @@ func writeCompatibleTaskResponse(runCtx context.Context, w http.ResponseWriter,
 		if !requestStillConnected(r) {
 			return
 		}
-		writeCompatibleDone(w, kind, model, result.Output)
+		streamWriter.writeDone(w, result.Output)
 		if flusher != nil {
 			flusher.Flush()
 		}
@@ -1064,6 +1065,12 @@ func writeCompatibleTaskResponse(runCtx context.Context, w http.ResponseWriter,
 	writeJSON(w, http.StatusOK, result.Output)
 }
 
+func streamIncludeUsage(body map[string]any) bool {
+	streamOptions, _ := body["stream_options"].(map[string]any)
+	includeUsage, _ := streamOptions["include_usage"].(bool)
+	return includeUsage
+}
+
 func asyncRequest(r *http.Request) bool {
 	value := strings.TrimSpace(strings.ToLower(r.Header.Get("x-async")))
 	return value == "1" || value == "true" || value == "yes" || value == "on"

apps/api/internal/httpapi/openapi_models.go

@@ -172,16 +172,18 @@ type PricingEstimateResponse struct {
 }
 
 type TaskRequest struct {
-	Model      string        `json:"model" example:"gpt-4o-mini"`
-	Messages   []ChatMessage `json:"messages,omitempty"`
-	Input      string        `json:"input,omitempty" example:"Tell me a short story"`
-	Prompt     string        `json:"prompt,omitempty" example:"A watercolor robot reading a book"`
-	Stream     bool          `json:"stream,omitempty" example:"false"`
-	RunMode    string        `json:"runMode,omitempty" example:"simulation"`
-	MaxTokens  int           `json:"max_tokens,omitempty" example:"512"`
-	Size       string        `json:"size,omitempty" example:"1024x1024"`
-	Duration   int           `json:"duration,omitempty" example:"5"`
-	Resolution string        `json:"resolution,omitempty" example:"720p"`
+	Model     string        `json:"model" example:"gpt-4o-mini"`
+	Messages  []ChatMessage `json:"messages,omitempty"`
+	Input     string        `json:"input,omitempty" example:"Tell me a short story"`
+	Prompt    string        `json:"prompt,omitempty" example:"A watercolor robot reading a book"`
+	Stream    bool          `json:"stream,omitempty" example:"false"`
+	RunMode   string        `json:"runMode,omitempty" example:"simulation"`
+	MaxTokens int           `json:"max_tokens,omitempty" example:"512"`
+	// ReasoningEffort 推理深度，OpenAI-compatible 请求字段；开放字符串，取值随 provider 和模型能力而定，常见值为 none、minimal、low、medium、high、xhigh，也可配置 max 等供应商自定义值。
+	ReasoningEffort string `json:"reasoning_effort,omitempty" example:"medium"`
+	Size            string `json:"size,omitempty" example:"1024x1024"`
+	Duration        int    `json:"duration,omitempty" example:"5"`
+	Resolution      string `json:"resolution,omitempty" example:"720p"`
 }
 
 type ChatCompletionRequest struct {
@@ -189,8 +191,10 @@ type ChatCompletionRequest struct {
 	Messages    []ChatMessage `json:"messages"`
 	Temperature float64       `json:"temperature,omitempty" example:"0.7"`
 	MaxTokens   int           `json:"max_tokens,omitempty" example:"512"`
-	Stream      bool          `json:"stream,omitempty" example:"false"`
-	RunMode     string        `json:"runMode,omitempty" example:"simulation"`
+	// ReasoningEffort 推理深度，OpenAI-compatible 请求字段；开放字符串，取值随 provider 和模型能力而定，常见值为 none、minimal、low、medium、high、xhigh，也可配置 max 等供应商自定义值。
+	ReasoningEffort string `json:"reasoning_effort,omitempty" example:"medium"`
+	Stream          bool   `json:"stream,omitempty" example:"false"`
+	RunMode         string `json:"runMode,omitempty" example:"simulation"`
 }
 
 type ChatMessage struct {

apps/api/internal/httpapi/streaming.go

@@ -1,6 +1,13 @@
 package httpapi
 
-import "net/http"
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
+)
 
 func prepareCompatibleStream(w http.ResponseWriter) http.Flusher {
 	w.Header().Set("Content-Type", "text/event-stream")
@@ -10,55 +17,180 @@ func prepareCompatibleStream(w http.ResponseWriter) http.Flusher {
 	return flusher
 }
 
-func writeCompatibleDelta(w http.ResponseWriter, kind string, model string, content string) {
-	if kind == "responses" {
-		sendSSE(w, "response.output_text.delta", map[string]any{"type": "response.output_text.delta", "delta": content})
-		return
-	}
-	sendSSE(w, "message", map[string]any{
-		"id":      "chatcmpl-stream",
-		"object":  "chat.completion.chunk",
-		"model":   model,
-		"choices": []any{map[string]any{"index": 0, "delta": map[string]any{"content": content}, "finish_reason": nil}},
-	})
+type compatibleStreamWriter struct {
+	kind         string
+	model        string
+	includeUsage bool
+
+	id                string
+	created           int64
+	systemFingerprint any
+	sentRole          bool
+	sentFinish        bool
+	sentUsage         bool
 }
 
-func writeCompatibleDone(w http.ResponseWriter, kind string, model string, output map[string]any) {
-	if kind == "responses" {
+func newCompatibleStreamWriter(kind string, model string, includeUsage bool) *compatibleStreamWriter {
+	return &compatibleStreamWriter{
+		kind:         kind,
+		model:        model,
+		includeUsage: includeUsage,
+		id:           "chatcmpl-stream",
+		created:      time.Now().Unix(),
+	}
+}
+
+func (s *compatibleStreamWriter) writeDelta(w http.ResponseWriter, event clients.StreamDeltaEvent) {
+	if s.kind == "responses" {
+		if event.Text != "" {
+			sendSSE(w, "response.output_text.delta", map[string]any{"type": "response.output_text.delta", "delta": event.Text})
+		}
+		return
+	}
+
+	if event.Event != nil && isChatCompletionChunk(event.Event) {
+		s.writeChatChunk(w, event.Event)
+		return
+	}
+
+	if event.Text == "" && event.ReasoningContent == "" {
+		return
+	}
+	s.ensureRoleChunk(w)
+	if event.ReasoningContent != "" {
+		s.writeChatData(w, s.chatChunk([]any{map[string]any{"index": 0, "delta": map[string]any{"reasoning_content": event.ReasoningContent}, "finish_reason": nil}}, nil))
+	}
+	if event.Text != "" {
+		s.writeChatData(w, s.chatChunk([]any{map[string]any{"index": 0, "delta": map[string]any{"content": event.Text}, "finish_reason": nil}}, nil))
+	}
+}
+
+func (s *compatibleStreamWriter) writeDone(w http.ResponseWriter, output map[string]any) {
+	if s.kind == "responses" {
 		sendSSE(w, "response.completed", map[string]any{"type": "response.completed", "response": output})
 		return
 	}
-	sendSSE(w, "message", map[string]any{
-		"id":      firstString(output["id"], "chatcmpl-stream"),
-		"object":  "chat.completion.chunk",
-		"model":   model,
-		"choices": []any{map[string]any{"index": 0, "delta": map[string]any{}, "finish_reason": "stop"}},
-	})
+	s.captureOutputMetadata(output)
+	if !s.sentRole {
+		s.ensureRoleChunk(w)
+	}
+	if !s.sentFinish {
+		s.writeChatData(w, s.chatChunk([]any{map[string]any{"index": 0, "delta": map[string]any{}, "finish_reason": finishReasonFromOutput(output)}}, nil))
+		s.sentFinish = true
+	}
+	if s.includeUsage && !s.sentUsage {
+		if usage, ok := output["usage"].(map[string]any); ok && len(usage) > 0 {
+			s.writeChatData(w, s.chatChunk([]any{}, usage))
+			s.sentUsage = true
+		}
+	}
+	s.writeDoneMarker(w)
+}
+
+func (s *compatibleStreamWriter) writeChatChunk(w http.ResponseWriter, chunk map[string]any) {
+	chunk = clients.NormalizeChatCompletionStreamEvent(chunk)
+	s.captureChunkMetadata(chunk)
+	choices, _ := chunk["choices"].([]any)
+	usage, hasUsage := chunk["usage"].(map[string]any)
+	if len(choices) == 0 && hasUsage {
+		if !s.includeUsage {
+			return
+		}
+		s.writeChatData(w, s.chatChunk([]any{}, usage))
+		s.sentUsage = true
+		return
+	}
+	if len(choices) > 0 && !chunkHasRole(choices) && !s.sentRole {
+		s.ensureRoleChunk(w)
+	}
+	if chunkHasRole(choices) {
+		s.sentRole = true
+	}
+	if chunkHasFinishReason(choices) {
+		s.sentFinish = true
+	}
+	normalized := cloneMap(chunk)
+	normalized["id"] = s.id
+	normalized["object"] = "chat.completion.chunk"
+	normalized["created"] = s.created
+	normalized["model"] = firstString(normalized["model"], s.model)
+	normalized["system_fingerprint"] = s.systemFingerprint
+	s.writeChatData(w, normalized)
+}
+
+func (s *compatibleStreamWriter) ensureRoleChunk(w http.ResponseWriter) {
+	if s.sentRole {
+		return
+	}
+	s.writeChatData(w, s.chatChunk([]any{map[string]any{"index": 0, "delta": map[string]any{"role": "assistant"}, "finish_reason": nil}}, nil))
+	s.sentRole = true
+}
+
+func (s *compatibleStreamWriter) chatChunk(choices []any, usage map[string]any) map[string]any {
+	chunk := map[string]any{
+		"id":                 s.id,
+		"object":             "chat.completion.chunk",
+		"created":            s.created,
+		"model":              s.model,
+		"system_fingerprint": s.systemFingerprint,
+		"choices":            choices,
+	}
+	if usage != nil {
+		chunk["usage"] = usage
+	} else {
+		chunk["usage"] = nil
+	}
+	return chunk
+}
+
+func (s *compatibleStreamWriter) writeChatData(w http.ResponseWriter, payload map[string]any) {
+	bytes, _ := json.Marshal(payload)
+	_, _ = fmt.Fprintf(w, "data: %s\n\n", bytes)
+}
+
+func (s *compatibleStreamWriter) writeDoneMarker(w http.ResponseWriter) {
+	_, _ = fmt.Fprint(w, "data: [DONE]\n\n")
+}
+
+func (s *compatibleStreamWriter) captureChunkMetadata(chunk map[string]any) {
+	if id := firstString(chunk["id"], ""); id != "" {
+		s.id = id
+	}
+	if model := firstString(chunk["model"], ""); model != "" {
+		s.model = model
+	}
+	if created := int64FromAny(chunk["created"]); created > 0 {
+		s.created = created
+	}
+	if value, ok := chunk["system_fingerprint"]; ok {
+		s.systemFingerprint = value
+	}
+}
+
+func (s *compatibleStreamWriter) captureOutputMetadata(output map[string]any) {
+	if id := firstString(output["id"], ""); id != "" {
+		s.id = id
+	}
+	if model := firstString(output["model"], ""); model != "" {
+		s.model = model
+	}
+	if created := int64FromAny(output["created"]); created > 0 {
+		s.created = created
+	}
+	if value, ok := output["system_fingerprint"]; ok {
+		s.systemFingerprint = value
+	}
 }
 
 func writeCompatibleStream(w http.ResponseWriter, kind string, model string, output map[string]any) {
 	prepareCompatibleStream(w)
+	writer := newCompatibleStreamWriter(kind, model, true)
 	content := extractOutputText(output)
 	if content == "" {
 		content = "done"
 	}
-	if kind == "responses" {
-		sendSSE(w, "response.output_text.delta", map[string]any{"type": "response.output_text.delta", "delta": content})
-		sendSSE(w, "response.completed", map[string]any{"type": "response.completed", "response": output})
-		return
-	}
-	sendSSE(w, "message", map[string]any{
-		"id":      output["id"],
-		"object":  "chat.completion.chunk",
-		"model":   model,
-		"choices": []any{map[string]any{"index": 0, "delta": map[string]any{"content": content}, "finish_reason": nil}},
-	})
-	sendSSE(w, "message", map[string]any{
-		"id":      output["id"],
-		"object":  "chat.completion.chunk",
-		"model":   model,
-		"choices": []any{map[string]any{"index": 0, "delta": map[string]any{}, "finish_reason": "stop"}},
-	})
+	writer.writeDelta(w, clients.StreamDeltaEvent{Text: content})
+	writer.writeDone(w, output)
 }
 
 func firstString(value any, fallback string) string {
@@ -68,6 +200,68 @@ func firstString(value any, fallback string) string {
 	return fallback
 }
 
+func int64FromAny(value any) int64 {
+	switch typed := value.(type) {
+	case int64:
+		return typed
+	case int:
+		return int64(typed)
+	case float64:
+		return int64(typed)
+	default:
+		return 0
+	}
+}
+
+func isChatCompletionChunk(event map[string]any) bool {
+	object, _ := event["object"].(string)
+	if object == "chat.completion.chunk" {
+		return true
+	}
+	_, hasChoices := event["choices"].([]any)
+	return hasChoices
+}
+
+func chunkHasRole(choices []any) bool {
+	for _, rawChoice := range choices {
+		choice, _ := rawChoice.(map[string]any)
+		delta, _ := choice["delta"].(map[string]any)
+		if role, ok := delta["role"].(string); ok && role != "" {
+			return true
+		}
+	}
+	return false
+}
+
+func chunkHasFinishReason(choices []any) bool {
+	for _, rawChoice := range choices {
+		choice, _ := rawChoice.(map[string]any)
+		if reason, ok := choice["finish_reason"].(string); ok && reason != "" {
+			return true
+		}
+	}
+	return false
+}
+
+func finishReasonFromOutput(output map[string]any) string {
+	choices, _ := output["choices"].([]any)
+	for _, rawChoice := range choices {
+		choice, _ := rawChoice.(map[string]any)
+		if reason, ok := choice["finish_reason"].(string); ok && reason != "" {
+			return reason
+		}
+	}
+	return "stop"
+}
+
+func cloneMap(value map[string]any) map[string]any {
+	out := map[string]any{}
+	for key, item := range value {
+		out[key] = item
+	}
+	return out
+}
+
 func extractOutputText(output map[string]any) string {
 	if text, ok := output["output_text"].(string); ok {
 		return text