refactor: 将格式转换逻辑从 Adapter 移至 Executor 层 (#120)

将请求/响应格式转换职责从各 Provider Adapter 集中到 Executor 层：

- 在 Executor 中检测客户端类型与 Provider 支持的类型是否匹配
- 新增 ConvertingResponseWriter 用于流式/非流式响应的格式转换
- 简化 CustomAdapter，移除其格式转换逻辑
- 更新 AntigravityAdapter 支持的客户端类型（移除 OpenAI，由 Executor 转换处理）
- 新增 OriginalClientType 上下文字段以追踪原始客户端类型
- 实时处理 Adapter 事件以提升请求可见性

Author: Bowl42

internal/adapter/provider/antigravity/adapter.go

@@ -49,8 +49,9 @@ func NewAdapter(p *domain.Provider) (provider.ProviderAdapter, error) {
 }
 
 func (a *AntigravityAdapter) SupportedClientTypes() []domain.ClientType {
-	// Antigravity natively supports Claude, OpenAI, and Gemini by converting to Gemini/v1internal API
-	return []domain.ClientType{domain.ClientTypeClaude, domain.ClientTypeOpenAI, domain.ClientTypeGemini}
+	// Antigravity natively supports Claude and Gemini by converting to Gemini/v1internal API
+	// OpenAI requests will be converted to Claude format by Executor before reaching this adapter
+	return []domain.ClientType{domain.ClientTypeClaude, domain.ClientTypeGemini}
 }
 
 func (a *AntigravityAdapter) Execute(ctx context.Context, w http.ResponseWriter, req *http.Request, provider *domain.Provider) error {

internal/adapter/provider/custom/adapter.go

@@ -14,7 +14,6 @@ import (
 
 	"github.com/awsl-project/maxx/internal/adapter/provider"
 	ctxutil "github.com/awsl-project/maxx/internal/context"
-	"github.com/awsl-project/maxx/internal/converter"
 	"github.com/awsl-project/maxx/internal/domain"
 	"github.com/awsl-project/maxx/internal/usage"
 )
@@ -24,17 +23,15 @@ func init() {
 }
 
 type CustomAdapter struct {
-	provider  *domain.Provider
-	converter *converter.Registry
+	provider *domain.Provider
 }
 
 func NewAdapter(p *domain.Provider) (provider.ProviderAdapter, error) {
 	if p.Config == nil || p.Config.Custom == nil {
 		return nil, fmt.Errorf("provider %s missing custom config", p.Name)
 	}
 	return &CustomAdapter{
-		provider:  p,
-		converter: converter.NewRegistry(),
+		provider: p,
 	}, nil
 }
 
@@ -50,22 +47,12 @@ func (a *CustomAdapter) Execute(ctx context.Context, w http.ResponseWriter, req
 	// Determine if streaming
 	stream := isStreamRequest(requestBody)
 
-	// Determine target client type for the provider
-	// If provider supports the client's type natively, use it directly
-	// Otherwise, find a supported type and convert
-	targetType := clientType
-	needsConversion := false
-	if !a.supportsClientType(clientType) {
-		// Find a supported type (prefer OpenAI as it's most common)
-		for _, supported := range a.provider.SupportedClientTypes {
-			targetType = supported
-			break
-		}
-		needsConversion = true
-	}
+	// Note: Format conversion is now handled by Executor layer
+	// The clientType in context is already the correct type that this provider supports
+	// We use clientType directly for URL building and auth header selection
 
 	// Build upstream URL
-	baseURL := a.getBaseURL(targetType)
+	baseURL := a.getBaseURL(clientType)
 	requestURI := ctxutil.GetRequestURI(ctx)
 
 	// For Gemini, update model in URL path if mapping is configured
@@ -87,7 +74,7 @@ func (a *CustomAdapter) Execute(ctx context.Context, w http.ResponseWriter, req
 
 	// Override auth headers with provider's credentials
 	if a.provider.Config.Custom.APIKey != "" {
-		setAuthHeader(upstreamReq, targetType, a.provider.Config.Custom.APIKey)
+		setAuthHeader(upstreamReq, clientType, a.provider.Config.Custom.APIKey)
 	}
 
 	// Send request info via EventChannel
@@ -100,12 +87,14 @@ func (a *CustomAdapter) Execute(ctx context.Context, w http.ResponseWriter, req
 		})
 	}
 
-	// Execute request
-	client := &http.Client{}
+	// Execute request with reasonable timeout
+	client := &http.Client{
+		Timeout: 10 * time.Minute, // Long timeout for LLM requests
+	}
 	resp, err := client.Do(upstreamReq)
 	if err != nil {
 		proxyErr := domain.NewProxyErrorWithMessage(domain.ErrUpstreamError, true, "failed to connect to upstream")
-		proxyErr.IsNetworkError = true // Mark as network error (connection timeout, DNS failure, etc.)
+		proxyErr.IsNetworkError = true
 		return proxyErr
 	}
 	defer resp.Body.Close()
@@ -144,10 +133,12 @@ func (a *CustomAdapter) Execute(ctx context.Context, w http.ResponseWriter, req
 	}
 
 	// Handle response
+	// Note: Response format conversion is handled by Executor's ConvertingResponseWriter
+	// Adapters simply pass through the upstream response
 	if stream {
-		return a.handleStreamResponse(ctx, w, resp, clientType, targetType, needsConversion)
+		return a.handleStreamResponse(ctx, w, resp, clientType)
 	}
-	return a.handleNonStreamResponse(ctx, w, resp, clientType, targetType, needsConversion)
+	return a.handleNonStreamResponse(ctx, w, resp, clientType)
 }
 
 func (a *CustomAdapter) supportsClientType(ct domain.ClientType) bool {
@@ -167,7 +158,7 @@ func (a *CustomAdapter) getBaseURL(clientType domain.ClientType) string {
 	return config.BaseURL
 }
 
-func (a *CustomAdapter) handleNonStreamResponse(ctx context.Context, w http.ResponseWriter, resp *http.Response, clientType, targetType domain.ClientType, needsConversion bool) error {
+func (a *CustomAdapter) handleNonStreamResponse(ctx context.Context, w http.ResponseWriter, resp *http.Response, clientType domain.ClientType) error {
 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return domain.NewProxyErrorWithMessage(domain.ErrUpstreamError, true, "failed to read upstream response")
@@ -197,28 +188,21 @@ func (a *CustomAdapter) handleNonStreamResponse(ctx context.Context, w http.Resp
 	}
 
 	// Extract and send responseModel
-	if responseModel := extractResponseModel(body, targetType); responseModel != "" {
+	if responseModel := extractResponseModel(body, clientType); responseModel != "" {
 		eventChan.SendResponseModel(responseModel)
 	}
 
-	var responseBody []byte
-	if needsConversion {
-		responseBody, err = a.converter.TransformResponse(targetType, clientType, body)
-		if err != nil {
-			return domain.NewProxyErrorWithMessage(domain.ErrFormatConversion, false, "failed to transform response")
-		}
-	} else {
-		responseBody = body
-	}
+	// Note: Response format conversion is handled by Executor's ConvertingResponseWriter
+	// Adapter simply passes through the upstream response body
 
 	// Copy upstream headers (except those we override)
 	copyResponseHeaders(w.Header(), resp.Header)
 	w.WriteHeader(resp.StatusCode)
-	_, _ = w.Write(responseBody)
+	_, _ = w.Write(body)
 	return nil
 }
 
-func (a *CustomAdapter) handleStreamResponse(ctx context.Context, w http.ResponseWriter, resp *http.Response, clientType, targetType domain.ClientType, needsConversion bool) error {
+func (a *CustomAdapter) handleStreamResponse(ctx context.Context, w http.ResponseWriter, resp *http.Response, clientType domain.ClientType) error {
 	eventChan := ctxutil.GetEventChan(ctx)
 
 	// Send initial response info (for streaming, we only capture status and headers)
@@ -251,10 +235,8 @@ func (a *CustomAdapter) handleStreamResponse(ctx context.Context, w http.Respons
 		return domain.NewProxyErrorWithMessage(domain.ErrUpstreamError, false, "streaming not supported")
 	}
 
-	var state *converter.TransformState
-	if needsConversion {
-		state = converter.NewTransformState()
-	}
+	// Note: Response format conversion is handled by Executor's ConvertingResponseWriter
+	// Adapter simply passes through the upstream SSE data
 
 	// Collect all SSE events for response body and token extraction
 	var sseBuffer strings.Builder
@@ -285,7 +267,7 @@ func (a *CustomAdapter) handleStreamResponse(ctx context.Context, w http.Respons
 			}
 
 			// Extract and send responseModel
-			if responseModel := extractResponseModelFromSSE(sseBuffer.String(), targetType); responseModel != "" {
+			if responseModel := extractResponseModelFromSSE(sseBuffer.String(), clientType); responseModel != "" {
 				eventChan.SendResponseModel(responseModel)
 			}
 		}
@@ -369,20 +351,10 @@ func (a *CustomAdapter) handleStreamResponse(ctx context.Context, w http.Respons
 					}
 				}
 
-				var output []byte
-				if needsConversion {
-					// Transform the chunk
-					transformed, transformErr := a.converter.TransformStreamChunk(targetType, clientType, []byte(line), state)
-					if transformErr != nil {
-						continue // Skip malformed chunks
-					}
-					output = transformed
-				} else {
-					output = []byte(line)
-				}
-
-				if len(output) > 0 {
-					_, writeErr := w.Write(output)
+				// Note: Response format conversion is handled by Executor's ConvertingResponseWriter
+				// Adapter simply passes through the upstream SSE data
+				if len(line) > 0 {
+					_, writeErr := w.Write([]byte(line))
 					if writeErr != nil {
 						// Client disconnected
 						sendFinalEvents()

internal/context/context.go

@@ -11,28 +11,33 @@ import (
 type contextKey string
 
 const (
-	CtxKeyClientType      contextKey = "client_type"
-	CtxKeySessionID       contextKey = "session_id"
-	CtxKeyProjectID       contextKey = "project_id"
-	CtxKeyRequestModel    contextKey = "request_model"
-	CtxKeyMappedModel     contextKey = "mapped_model"
-	CtxKeyResponseModel   contextKey = "response_model"
-	CtxKeyProxyRequest    contextKey = "proxy_request"
-	CtxKeyRequestBody     contextKey = "request_body"
-	CtxKeyUpstreamAttempt contextKey = "upstream_attempt"
-	CtxKeyRequestHeaders  contextKey = "request_headers"
-	CtxKeyRequestURI      contextKey = "request_uri"
-	CtxKeyBroadcaster     contextKey = "broadcaster"
-	CtxKeyIsStream        contextKey = "is_stream"
-	CtxKeyAPITokenID      contextKey = "api_token_id"
-	CtxKeyEventChan       contextKey = "event_chan"
+	CtxKeyClientType         contextKey = "client_type"
+	CtxKeyOriginalClientType contextKey = "original_client_type" // Original client type before format conversion
+	CtxKeySessionID          contextKey = "session_id"
+	CtxKeyProjectID          contextKey = "project_id"
+	CtxKeyRequestModel       contextKey = "request_model"
+	CtxKeyMappedModel        contextKey = "mapped_model"
+	CtxKeyResponseModel      contextKey = "response_model"
+	CtxKeyProxyRequest       contextKey = "proxy_request"
+	CtxKeyRequestBody        contextKey = "request_body"
+	CtxKeyUpstreamAttempt    contextKey = "upstream_attempt"
+	CtxKeyRequestHeaders     contextKey = "request_headers"
+	CtxKeyRequestURI         contextKey = "request_uri"
+	CtxKeyBroadcaster        contextKey = "broadcaster"
+	CtxKeyIsStream           contextKey = "is_stream"
+	CtxKeyAPITokenID         contextKey = "api_token_id"
+	CtxKeyEventChan          contextKey = "event_chan"
 )
 
 // Setters
 func WithClientType(ctx context.Context, ct domain.ClientType) context.Context {
 	return context.WithValue(ctx, CtxKeyClientType, ct)
 }
 
+func WithOriginalClientType(ctx context.Context, ct domain.ClientType) context.Context {
+	return context.WithValue(ctx, CtxKeyOriginalClientType, ct)
+}
+
 func WithSessionID(ctx context.Context, sid string) context.Context {
 	return context.WithValue(ctx, CtxKeySessionID, sid)
 }
@@ -81,6 +86,13 @@ func GetClientType(ctx context.Context) domain.ClientType {
 	return ""
 }
 
+func GetOriginalClientType(ctx context.Context) domain.ClientType {
+	if v, ok := ctx.Value(CtxKeyOriginalClientType).(domain.ClientType); ok {
+		return v
+	}
+	return ""
+}
+
 func GetSessionID(ctx context.Context) string {
 	if v, ok := ctx.Value(CtxKeySessionID).(string); ok {
 		return v