fix(gateway): extend compact keepalive to non-passthrough OAuth path

The initial implementation only covered the passthrough branch, but
production traffic showed compact requests also flow through the
standard OAuth HTTP forward path. Add keepalive to both Forward and
forwardOpenAIPassthrough, with structured logging for diagnostics.

Author: bud-primordium

backend/internal/service/openai_gateway_service.go

@@ -443,9 +443,21 @@ func NewOpenAIGatewayService(
 		openAITokenProvider.SetAccountRuntimeBlocker(svc)
 	}
 	svc.logOpenAIWSModeBootstrap()
+	svc.logOpenAICompactNonstreamKeepaliveBootstrap()
 	return svc
 }
 
+func (s *OpenAIGatewayService) logOpenAICompactNonstreamKeepaliveBootstrap() {
+	interval := s.compactNonstreamKeepaliveInterval()
+	if interval <= 0 {
+		return
+	}
+	logger.L().With(
+		zap.String("component", "service.openai_gateway"),
+		zap.Int("interval_seconds", int(interval.Seconds())),
+	).Info("OpenAI compact non-stream keepalive enabled")
+}
+
 // ResolveChannelMapping 解析渠道级模型映射（代理到 ChannelService）
 func (s *OpenAIGatewayService) ResolveChannelMapping(ctx context.Context, groupID int64, model string) ChannelMappingResult {
 	if s.channelService == nil {
@@ -2982,25 +2994,42 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 		}
 
 		// Send request
+		stopCompactKeepalive := func() {}
+		if !reqStream {
+			stopCompactKeepalive = s.startCompactNonstreamKeepalive(ctx, c)
+		}
+
 		upstreamStart := time.Now()
 		resp, err := s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency)
 		SetOpsLatencyMs(c, OpsUpstreamLatencyMsKey, time.Since(upstreamStart).Milliseconds())
 		if err != nil {
+			stopCompactKeepalive()
 			// Transport-level failure (proxy/DNS/TCP/TLS — no HTTP response). Convert to
 			// a failover so the handler switches to a healthy account, and temporarily
 			// unschedule the account on durable faults (e.g. rejected proxy credentials).
-			return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false)
+			transportErr := s.handleOpenAIUpstreamTransportError(ctx, c, account, err, false)
+			if openAICompactKeepaliveCommitted(c) {
+				logOpenAICompactKeepaliveCommitted(ctx, c, account, nil)
+				writeOpenAICommittedTransportError(c)
+				return nil, fmt.Errorf("upstream request failed after compact keepalive: %s", sanitizeUpstreamErrorMessage(err.Error()))
+			}
+			return nil, transportErr
 		}
 
 		// Handle error response
 		if resp.StatusCode >= 400 {
+			stopCompactKeepalive()
 			respBody := s.readUpstreamErrorBody(resp)
 			_ = resp.Body.Close()
 			resp.Body = io.NopCloser(bytes.NewReader(respBody))
 
 			upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
 			upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
 			upstreamCode := extractUpstreamErrorCode(respBody)
+			if openAICompactKeepaliveCommitted(c) {
+				logOpenAICompactKeepaliveCommitted(ctx, c, account, resp)
+				return s.handleErrorResponse(ctx, resp, c, account, body, billingModel)
+			}
 			if !httpInvalidEncryptedContentRetryTried && resp.StatusCode == http.StatusBadRequest && upstreamCode == "invalid_encrypted_content" {
 				decoded, decodeErr := ensureReqBody()
 				if decodeErr != nil {
@@ -3063,6 +3092,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 		imageCount := 0
 		var imageOutputSizes []string
 		if reqStream {
+			stopCompactKeepalive()
 			streamResult, err := s.handleStreamingResponse(ctx, resp, c, account, startTime, originalModel, upstreamModel)
 			if err != nil {
 				return nil, err
@@ -3073,7 +3103,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 			imageCount = streamResult.imageCount
 			imageOutputSizes = streamResult.imageOutputSizes
 		} else {
-			nonStreamResult, err := s.handleNonStreamingResponse(ctx, resp, c, account, originalModel, upstreamModel)
+			nonStreamResult, err := s.handleNonStreamingResponse(ctx, resp, c, account, originalModel, upstreamModel, stopCompactKeepalive)
 			if err != nil {
 				return nil, err
 			}
@@ -3686,6 +3716,16 @@ func (s *OpenAIGatewayService) startCompactNonstreamKeepalive(ctx context.Contex
 	if ctx == nil {
 		ctx = context.Background()
 	}
+	path := ""
+	if c.Request != nil && c.Request.URL != nil {
+		path = strings.TrimSpace(c.Request.URL.Path)
+	}
+	log := logger.FromContext(ctx).With(
+		zap.String("component", "service.openai_gateway"),
+		zap.String("request_path", path),
+		zap.Int("interval_seconds", int(interval.Seconds())),
+	)
+	log.Info("OpenAI compact non-stream keepalive started")
 
 	headers := c.Writer.Header()
 	headers.Set("Content-Type", "application/json")
@@ -3701,15 +3741,23 @@ func (s *OpenAIGatewayService) startCompactNonstreamKeepalive(ctx context.Contex
 		defer wg.Done()
 		ticker := time.NewTicker(interval)
 		defer ticker.Stop()
+		flushedLogged := false
 		for {
 			select {
 			case <-stopCh:
 				return
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
-				_, _ = c.Writer.Write([]byte("\n"))
+				if _, err := c.Writer.Write([]byte("\n")); err != nil {
+					log.Warn("OpenAI compact non-stream keepalive write failed", zap.Error(err))
+					return
+				}
 				flusher.Flush()
+				if !flushedLogged {
+					log.Info("OpenAI compact non-stream keepalive flushed")
+					flushedLogged = true
+				}
 			}
 		}
 	}()
@@ -5385,17 +5433,22 @@ func openAIUsageFromGJSON(value gjson.Result) (OpenAIUsage, bool) {
 	}, true
 }
 
-func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, resp *http.Response, c *gin.Context, account *Account, originalModel, mappedModel string) (*openaiNonStreamingResult, error) {
-	body, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, openAITooLargeError)
+func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, resp *http.Response, c *gin.Context, account *Account, originalModel, mappedModel string, stopBeforeWrite ...func()) (*openaiNonStreamingResult, error) {
+	stop := compactStopFunc(stopBeforeWrite...)
+	body, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, func(c *gin.Context) {
+		stop()
+		openAITooLargeError(c)
+	})
 	if err != nil {
+		stop()
 		return nil, err
 	}
 
 	// Detect SSE responses for ALL account types via Content-Type header.
 	// Some OpenAI-compatible upstreams (including other sub2api instances)
 	// may return SSE even when stream=false was requested.
 	if isEventStreamResponse(resp.Header) {
-		return s.handleSSEToJSON(resp, c, body, originalModel, mappedModel)
+		return s.handleSSEToJSON(resp, c, body, originalModel, mappedModel, stop)
 	}
 	bodyLooksLikeSSE := bytes.Contains(body, []byte("data:")) || bytes.Contains(body, []byte("event:"))
 
@@ -5405,14 +5458,15 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r
 	// positives on JSON responses that coincidentally contain "data:" or
 	// "event:" in their text content.
 	if account.Type == AccountTypeOAuth && bodyLooksLikeSSE {
-		return s.handleSSEToJSON(resp, c, body, originalModel, mappedModel)
+		return s.handleSSEToJSON(resp, c, body, originalModel, mappedModel, stop)
 	}
 
 	usageValue, usageOK := extractOpenAIUsageFromJSONBytes(body)
 	if !usageOK {
 		if bodyLooksLikeSSE {
-			return s.handleSSEToJSON(resp, c, body, originalModel, mappedModel)
+			return s.handleSSEToJSON(resp, c, body, originalModel, mappedModel, stop)
 		}
+		stop()
 		return nil, fmt.Errorf("parse response: invalid json response")
 	}
 	usage := &usageValue
@@ -5431,6 +5485,7 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r
 		}
 	}
 
+	stop()
 	c.Data(resp.StatusCode, contentType, body)
 
 	return &openaiNonStreamingResult{
@@ -5442,12 +5497,20 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r
 	}, nil
 }
 
+func compactStopFunc(stops ...func()) func() {
+	if len(stops) == 0 || stops[0] == nil {
+		return func() {}
+	}
+	return stops[0]
+}
+
 func isEventStreamResponse(header http.Header) bool {
 	contentType := strings.ToLower(header.Get("Content-Type"))
 	return strings.Contains(contentType, "text/event-stream")
 }
 
-func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Context, body []byte, originalModel, mappedModel string) (*openaiNonStreamingResult, error) {
+func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Context, body []byte, originalModel, mappedModel string, stopBeforeWrite ...func()) (*openaiNonStreamingResult, error) {
+	stop := compactStopFunc(stopBeforeWrite...)
 	bodyText := string(body)
 	finalResponse, ok := extractCodexFinalResponse(bodyText)
 
@@ -5479,6 +5542,7 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte
 			if msg == "" {
 				msg = "Upstream compact response failed"
 			}
+			stop()
 			return nil, s.writeOpenAINonStreamingProtocolError(resp, c, msg)
 		}
 		usage = s.parseSSEUsageFromBody(bodyText)
@@ -5497,6 +5561,7 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte
 			contentType = "text/event-stream"
 		}
 	}
+	stop()
 	c.Data(resp.StatusCode, contentType, body)
 
 	return &openaiNonStreamingResult{

backend/internal/service/openai_oauth_passthrough_test.go

@@ -47,12 +47,12 @@ func (u *httpUpstreamRecorder) Do(req *http.Request, proxyURL string, accountID
 		req.Body = io.NopCloser(bytes.NewReader(b))
 	}
 	u.requests = append(u.requests, req)
-	if u.err != nil {
-		return nil, u.err
-	}
 	if u.delay > 0 {
 		time.Sleep(u.delay)
 	}
+	if u.err != nil {
+		return nil, u.err
+	}
 	if len(u.responses) > 0 {
 		resp := u.responses[0]
 		u.responses = u.responses[1:]
@@ -497,6 +497,55 @@ func TestOpenAIGatewayService_OAuthPassthrough_CompactNonstreamKeepaliveWritesLe
 	require.Contains(t, string(body), `"id":"cmp_keepalive"`)
 }
 
+func TestOpenAIGatewayService_OAuthCompactNonPassthroughKeepaliveWritesLeadingNewline(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses/compact", bytes.NewReader(nil))
+	c.Request.Header.Set("User-Agent", "Codex Desktop/0.135.0-alpha.1")
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	originalBody := []byte(`{"model":"gpt-5.5","stream":false,"instructions":"local-test-instructions","input":[{"type":"message","role":"user","content":"compact me"}]}`)
+	resp := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid-compact-native-keepalive"}},
+		Body:       io.NopCloser(strings.NewReader(`{"id":"cmp_native_keepalive","usage":{"input_tokens":11,"output_tokens":22}}`)),
+	}
+	upstream := &httpUpstreamRecorder{resp: resp, delay: 1200 * time.Millisecond}
+
+	svc := &OpenAIGatewayService{
+		cfg: &config.Config{Gateway: config.GatewayConfig{
+			OpenAICompactNonstreamKeepaliveInterval: 1,
+		}},
+		httpUpstream: upstream,
+	}
+	account := &Account{
+		ID:          124,
+		Name:        "acc-native",
+		Platform:    PlatformOpenAI,
+		Type:        AccountTypeOAuth,
+		Concurrency: 1,
+		Credentials: map[string]any{
+			"access_token":       "oauth-token",
+			"chatgpt_account_id": "chatgpt-acc",
+		},
+		Status:         StatusActive,
+		Schedulable:    true,
+		RateMultiplier: f64p(1),
+	}
+
+	result, err := svc.Forward(context.Background(), c, account, originalBody)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	body := rec.Body.Bytes()
+	require.True(t, bytes.HasPrefix(body, []byte("\n")), "native compact keepalive should write a leading blank line")
+	require.True(t, json.Valid(bytes.TrimSpace(body)))
+	require.Contains(t, string(body), `"id":"cmp_native_keepalive"`)
+	require.NotNil(t, upstream.lastReq)
+	require.Equal(t, chatgptCodexURL+"/compact", upstream.lastReq.URL.String())
+}
+
 func TestOpenAIGatewayService_OAuthPassthrough_CompactNonstreamKeepaliveDisabledHasNoLeadingWhitespace(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 
@@ -623,6 +672,71 @@ func TestOpenAIGatewayService_OAuthPassthrough_CompactNonstreamKeepaliveCommitte
 	}
 }
 
+func TestOpenAIGatewayService_OAuthCompactNonstreamKeepaliveCommittedTransportErrorDoesNotFailover(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	testCases := []struct {
+		name    string
+		body    []byte
+		account *Account
+	}{
+		{
+			name:    "passthrough",
+			body:    []byte(`{"model":"gpt-5.1-codex","stream":true,"instructions":"local-test-instructions","input":[{"type":"text","text":"compact me"}]}`),
+			account: newOpenAICompactPassthroughTestAccount(),
+		},
+		{
+			name: "non_passthrough",
+			body: []byte(`{"model":"gpt-5.5","stream":false,"instructions":"local-test-instructions","input":[{"type":"message","role":"user","content":"compact me"}]}`),
+			account: &Account{
+				ID:          124,
+				Name:        "acc-native",
+				Platform:    PlatformOpenAI,
+				Type:        AccountTypeOAuth,
+				Concurrency: 1,
+				Credentials: map[string]any{
+					"access_token":       "oauth-token",
+					"chatgpt_account_id": "chatgpt-acc",
+				},
+				Status:         StatusActive,
+				Schedulable:    true,
+				RateMultiplier: f64p(1),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			rec := httptest.NewRecorder()
+			c, _ := gin.CreateTestContext(rec)
+			c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses/compact", bytes.NewReader(nil))
+			c.Request.Header.Set("User-Agent", "codex_cli_rs/0.1.0")
+			c.Request.Header.Set("Content-Type", "application/json")
+
+			upstream := &httpUpstreamRecorder{
+				err:   errors.New("dial tcp: i/o timeout"),
+				delay: 1200 * time.Millisecond,
+			}
+			svc := &OpenAIGatewayService{
+				cfg: &config.Config{Gateway: config.GatewayConfig{
+					ForceCodexCLI:                           false,
+					OpenAICompactNonstreamKeepaliveInterval: 1,
+				}},
+				httpUpstream: upstream,
+			}
+
+			result, err := svc.Forward(context.Background(), c, tc.account, tc.body)
+			require.Error(t, err)
+			require.Nil(t, result)
+			var failoverErr *UpstreamFailoverError
+			require.False(t, errors.As(err, &failoverErr), "committed compact keepalive must suppress transport failover")
+			body := rec.Body.Bytes()
+			require.True(t, bytes.HasPrefix(body, []byte("\n")))
+			require.Contains(t, string(body), "Upstream request failed")
+		})
+	}
+}
+
 func newOpenAICompactPassthroughTestAccount() *Account {
 	return &Account{
 		ID:             123,