fix(auth): harden failover recovery

config.example.yaml

@@ -283,6 +283,7 @@ nonstream-keepalive-interval: 0
 # streaming:
 #   keepalive-seconds: 15   # Default: 0 (disabled). <= 0 disables keep-alives.
 #   bootstrap-retries: 1    # Default: 0 (disabled). Retries before first byte is sent.
+#   stream-connect-timeout-seconds: 20 # Default: 0 (disabled). Optional maximum wait for connection/stream establishment before failover. (Deprecated alias: stream-first-chunk-timeout-seconds).
 
 # Signature cache validation for thinking blocks (Antigravity/Claude).
 # When true (default), cached signatures are preferred and validated.

internal/api/modules/amp/secret_test.go

@@ -69,7 +69,7 @@ func TestMultiSourceSecret_CacheBehavior(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	s := NewMultiSourceSecretWithPath("", p, 50*time.Millisecond)
+	s := NewMultiSourceSecretWithPath("", p, 500*time.Millisecond)
 
 	// First read - should return v1
 	got1, err := s.Get(ctx)
@@ -90,7 +90,7 @@ func TestMultiSourceSecret_CacheBehavior(t *testing.T) {
 	}
 
 	// After TTL expires, should see v2
-	time.Sleep(60 * time.Millisecond)
+	time.Sleep(600 * time.Millisecond)
 	got3, _ := s.Get(ctx)
 	if got3 != "v2" {
 		t.Fatalf("cache miss expected v2, got %s", got3)

internal/clienterror/client_error.go

@@ -80,6 +80,20 @@ func IsRequestFault(status int, err error) bool {
 			status = statusErr.StatusCode()
 		}
 	}
+	// Payment and rate-limit statuses are authoritative even when an upstream
+	// pairs them with a generic invalid_request_error body. The credential must
+	// remain eligible for cooldown and rotation.
+	if status == http.StatusPaymentRequired || status == http.StatusTooManyRequests {
+		return false
+	}
+	// Authentication and invalid-or-expired-credential failures are caused by
+	// the credential, not the request: they must remain eligible for rotation
+	// by the shared mixed-auth loop even when the provider pairs them with a
+	// generic invalid-request identifier in the body. This must be checked
+	// before hasRequestFaultBody so the generic classifier cannot misfile them.
+	if (status == http.StatusUnauthorized || status == http.StatusForbidden) && hasAuthenticationErrorBody(err) {
+		return false
+	}
 	if hasRequestFaultBody(err) {
 		return true
 	}
@@ -112,6 +126,37 @@ func IsItemNotPersisted(message string) bool {
 		strings.Contains(lower, "items are not persisted when `store` is set to false")
 }
 
+// hasAuthenticationErrorBody reports whether err is a structured credential
+// failure: an authentication_error type, an invalid or expired API-key code,
+// or a Gemini UNAUTHENTICATED status. These are credential faults, not request
+// faults, so they must never be classified as request faults on 401/403.
+func hasAuthenticationErrorBody(err error) bool {
+	if err == nil {
+		return false
+	}
+	body := strings.TrimSpace(err.Error())
+	if body == "" || !json.Valid([]byte(body)) {
+		return false
+	}
+	for _, path := range []string{"error.code", "code", "response.error.code", "body.error.code"} {
+		switch strings.ToLower(strings.TrimSpace(gjson.Get(body, path).String())) {
+		case "invalid_api_key", "incorrect_api_key", "expired_api_key":
+			return true
+		}
+	}
+	for _, path := range []string{"error.type", "type", "response.error.type", "body.error.type"} {
+		if errType := strings.ToLower(strings.TrimSpace(gjson.Get(body, path).String())); errType == "authentication_error" {
+			return true
+		}
+	}
+	for _, path := range []string{"error.status", "status", "response.error.status", "body.error.status"} {
+		if status := strings.ToLower(strings.TrimSpace(gjson.Get(body, path).String())); status == "unauthenticated" {
+			return true
+		}
+	}
+	return false
+}
+
 func hasRequestFaultBody(err error) bool {
 	if err == nil {
 		return false

internal/clienterror/client_error_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"strings"
 	"testing"
 )
 
@@ -195,6 +196,124 @@ func TestIsRequestFault(t *testing.T) {
 		{name: "transport", status: http.StatusBadGateway, err: errors.New("unexpected EOF")},
 		{name: "invalid JSON body", status: http.StatusBadGateway, err: errors.New(`{"error":`)},
 		{name: "nil", status: 0},
+		{
+			// DeepSeek-style: 401 with authentication_error type plus the generic
+			// invalid_request_error code. Credential fault, not a request fault.
+			name:   "401 authentication_error body",
+			status: http.StatusUnauthorized,
+			err:    errors.New(`{"error":{"message":"Authentication Fails","type":"authentication_error","code":"invalid_request_error"}}`),
+		},
+		{
+			name:   "403 authentication_error body",
+			status: http.StatusForbidden,
+			err:    errors.New(`{"error":{"type":"authentication_error","message":"Invalid token."}}`),
+		},
+		{
+			name:   "401 invalid_api_key code",
+			status: http.StatusUnauthorized,
+			err:    errors.New(`{"error":{"message":"Invalid API key","type":"invalid_request_error","code":"invalid_api_key"}}`),
+		},
+		{
+			name:   "403 incorrect_api_key code",
+			status: http.StatusForbidden,
+			err:    errors.New(`{"error":{"code":"incorrect_api_key","message":"bad key"}}`),
+		},
+		{
+			name:   "401 gemini unauthenticated status",
+			status: http.StatusUnauthorized,
+			err:    errors.New(`{"error":{"code":16,"message":"Request had invalid authentication credentials.","status":"UNAUTHENTICATED"}}`),
+		},
+		{
+			// Authentication marker with an invalid-request-type body: the auth
+			// carve-out must win over the generic classifier.
+			name:   "401 auth marker beats generic invalid_request",
+			status: http.StatusUnauthorized,
+			err:    errors.New(`{"error":{"type":"invalid_request","code":"invalid_api_key"}}`),
+		},
+		{
+			// Same body on a non-auth status stays a request fault.
+			name:   "authentication marker only honored on 401/403",
+			status: http.StatusBadRequest,
+			err:    errors.New(`{"error":{"type":"invalid_request","code":"invalid_api_key"}}`),
+			want:   true,
+		},
+		{
+			// An auth-styled body behind a server status is not a request fault.
+			name:   "authentication_error body behind upstream error",
+			status: http.StatusInternalServerError,
+			err:    errors.New(`{"error":{"type":"authentication_error","message":"server glitch"}}`),
+		},
+		{
+			// Wrapped status code: the 401 is extracted via errors.As, and even a
+			// non-splittable (prefixed) body stays a non-request credential fault.
+			name: "wrapped 401 authentication_error",
+			err:  fmt.Errorf("upstream: %w", statusError{status: http.StatusUnauthorized, body: `{"error":{"type":"authentication_error"}}`}),
+		},
+		{
+			name: "wrapped 400 invalid request stays request fault",
+			err:  fmt.Errorf("upstream: %w", statusError{status: http.StatusBadRequest, body: "bad input"}),
+			want: true,
+		},
+		{
+			// statusCoder form: the error carries both the 401 and the structured
+			// auth body directly (injected status stays 0, so it is read off the
+			// error). This mirrors how the shared loop feeds a Result error.
+			name: "statusCoder auth body",
+			err:  statusError{status: http.StatusUnauthorized, body: `{"error":{"type":"authentication_error","code":"invalid_api_key"}}`},
+		},
+		{
+			// Negative: a statusCoder 400 with an auth-looking body is still a
+			// request fault once status wins over the body.
+			name: "statusCoder 400 auth body still request fault",
+			err:  statusError{status: http.StatusBadRequest, body: `{"error":{"type":"authentication_error"}}`},
+			want: true,
+		},
+		{
+			// Large but valid bodies are still classified safely.
+			name:   "large authentication_error body",
+			status: http.StatusForbidden,
+			err:    errors.New(`{"error":{"type":"authentication_error","message":"` + strings.Repeat("x", 1<<16) + `"}}`),
+		},
+		{
+			name:   "429 stays credential domain",
+			status: http.StatusTooManyRequests,
+			err:    errors.New(`{"error":{"type":"rate_limit_error","message":"slow down"}}`),
+		},
+		{
+			// CPA parity: a rate-limit status is authoritative even when the body
+			// carries a generic invalid_request_error code. Quota, not request.
+			name:   "429 with generic invalid_request_error code stays quota",
+			status: http.StatusTooManyRequests,
+			err:    errors.New(`{"error":{"code":"invalid_request_error","message":"Rate Limit Reached","param":null,"type":"unknown_error"}}`),
+		},
+		{
+			// CPA parity: a payment-required status is authoritative even when the
+			// body carries the generic invalid_request_error code. Quota/balance,
+			// not request.
+			name:   "402 with generic invalid_request_error code stays payment",
+			status: http.StatusPaymentRequired,
+			err:    errors.New(`{"error":{"message":"Insufficient Balance","type":"unknown_error","param":null,"code":"invalid_request_error"}}`),
+		},
+		{
+			// A 402 surface with the invalid_request_error type spelling is still
+			// payment-scoped, never a request fault.
+			name:   "402 with invalid_request_error type stays payment",
+			status: http.StatusPaymentRequired,
+			err:    errors.New(`{"error":{"type":"invalid_request_error","code":"insufficient_balance","message":"low balance"}}`),
+		},
+		{
+			// statusCoder form: quota/payment status is read off the error and the
+			// precedence still wins over the generic body code.
+			name: "statusCoder 429 with generic code stays quota",
+			err:  statusError{status: http.StatusTooManyRequests, body: `{"error":{"code":"invalid_request_error","message":"quota"}}`},
+		},
+		{
+			// A 429 that is genuinely a 429 request-fault-looking body never flips
+			// to a request fault either, because the status stays authoritative.
+			name:   "429 status stays authoritative over auth-looking body",
+			status: http.StatusTooManyRequests,
+			err:    errors.New(`{"error":{"type":"authentication_error","code":"invalid_request_error"}}`),
+		},
 	}
 
 	for _, tc := range tests {

internal/config/sdk_config.go

@@ -82,4 +82,11 @@ type StreamingConfig struct {
 	// to allow auth rotation / transient recovery.
 	// <= 0 disables bootstrap retries. Default is 0.
 	BootstrapRetries int `yaml:"bootstrap-retries,omitempty" json:"bootstrap-retries,omitempty"`
+
+	// StreamConnectTimeoutSeconds controls the maximum time to wait for connection/stream establishment from an upstream stream before timing out and failing over.
+	// <= 0 disables stream connect timeout. Default is 0.
+	StreamConnectTimeoutSeconds int `yaml:"stream-connect-timeout-seconds,omitempty" json:"stream-connect-timeout-seconds,omitempty"`
+
+	// StreamFirstChunkTimeoutSeconds is a deprecated alias for StreamConnectTimeoutSeconds.
+	StreamFirstChunkTimeoutSeconds int `yaml:"stream-first-chunk-timeout-seconds,omitempty" json:"stream-first-chunk-timeout-seconds,omitempty"`
 }

internal/home/client.go

@@ -1618,9 +1618,8 @@ func newPluginSyncCancelableConn(ctx context.Context, conn net.Conn) net.Conn {
 	go func() {
 		select {
 		case <-ctx.Done():
-			if errDeadline := conn.SetDeadline(time.Now()); errDeadline != nil {
-				_ = conn.Close()
-			}
+			_ = conn.SetDeadline(time.Now())
+			_ = conn.Close()
 		case <-wrapped.done:
 		}
 	}()

internal/interfaces/error_message.go

@@ -21,6 +21,11 @@ type ErrorMessage struct {
 	// DirectResponse reports that Body and Headers were explicitly supplied by a trusted in-process component.
 	DirectResponse bool
 
+	// TrustedDirectResponse reports that a DirectResponse originated locally
+	// (plugin/interceptor) and is safe to preserve through OpenAI sanitizers.
+	// Zero value false means the DirectResponse must be treated as untrusted.
+	TrustedDirectResponse bool
+
 	// Body contains a preformatted downstream response when DirectResponse is true.
 	Body []byte
 

internal/pluginhost/executor_route.go

@@ -159,17 +159,28 @@ func wrapStreamEmptyCompletion(ctx context.Context, streamResult *coreexecutor.S
 			case chunk, ok = <-src:
 			}
 			if !ok {
-				if !forwarding && len(buffered) == 0 {
-					_ = forward(coreexecutor.StreamChunk{Err: &coreauth.Error{
-						Code:      "empty_stream",
-						Message:   "upstream stream closed before first payload",
-						Retryable: true,
-					}})
-					return
-				}
-				if !forwarding && coreauth.IsEmptyCompletionPayload(streamChunkPayload(buffered)) {
-					_ = forward(coreexecutor.StreamChunk{Err: coreauth.EmptyCompletionError()})
-					return
+				if !forwarding {
+					payloadBytes := 0
+					for _, c := range buffered {
+						payloadBytes += len(c.Payload)
+					}
+					if payloadBytes == 0 {
+						// Zero-payload chunks are dropped downstream; a stream of only
+						// such chunks is an empty stream, not a successful completion.
+						_ = forward(coreexecutor.StreamChunk{Err: &coreauth.Error{
+							Code:      "empty_stream",
+							Message:   "upstream stream closed before first payload",
+							Retryable: true,
+						}})
+						return
+					}
+					// Judge with the incremental detector state instead of re-parsing
+					// the concatenated payload: separately chunked SSE frames do not
+					// reassemble into valid input for the payload-level check.
+					if detector.Finish() {
+						_ = forward(coreexecutor.StreamChunk{Err: coreauth.EmptyCompletionError()})
+						return
+					}
 				}
 				_ = flush()
 				return
@@ -202,11 +213,26 @@ func wrapStreamEmptyCompletion(ctx context.Context, streamResult *coreexecutor.S
 					return
 				}
 			}
+			if detector.IsTerminalEmpty() {
+				discardStreamChunks(src)
+				_ = forward(coreexecutor.StreamChunk{Err: coreauth.EmptyCompletionError()})
+				return
+			}
 		}
 	}()
 	return &coreexecutor.StreamResult{Chunks: wrapped, Headers: streamResult.Headers}
 }
 
+func discardStreamChunks(ch <-chan coreexecutor.StreamChunk) {
+	if ch == nil {
+		return
+	}
+	go func() {
+		for range ch {
+		}
+	}()
+}
+
 func streamChunkPayload(chunks []coreexecutor.StreamChunk) []byte {
 	var payload []byte
 	for _, chunk := range chunks {

internal/pluginhost/executor_route_stream_codex_test.go

@@ -0,0 +1,67 @@
+package pluginhost
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth"
+	coreexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor"
+)
+
+// Regression tests mirrored from CLIProxyAPI PR #4881 follow-up
+// (codex pullrequestreview-4943660625, pluginhost stream wrapper EOF
+// handling).
+
+// TestWrapStreamEmptyCompletionRejectsZeroPayloadChunkStream is a regression
+// guard for the zero-payload finding: zero-payload chunks made the buffer
+// non-empty, so the EOF branch skipped the empty_stream error and flushed a
+// client-invisible stream as success.
+func TestWrapStreamEmptyCompletionRejectsZeroPayloadChunkStream(t *testing.T) {
+	src := make(chan coreexecutor.StreamChunk, 2)
+	src <- coreexecutor.StreamChunk{Payload: nil}
+	src <- coreexecutor.StreamChunk{Payload: []byte{}}
+	close(src)
+
+	wrapped := wrapStreamEmptyCompletion(context.Background(), &coreexecutor.StreamResult{Chunks: src})
+	first, ok := <-wrapped.Chunks
+	if !ok {
+		t.Fatal("wrapped stream closed without empty_stream error")
+	}
+	var authErr *coreauth.Error
+	if !errors.As(first.Err, &authErr) || authErr.Code != "empty_stream" || !authErr.Retryable {
+		t.Fatalf("first error = %#v, want retryable empty_stream", first.Err)
+	}
+	if len(first.Payload) != 0 {
+		t.Fatalf("first payload = %q, want error before client-visible bytes", first.Payload)
+	}
+	if _, ok = <-wrapped.Chunks; ok {
+		t.Fatal("wrapped stream emitted chunks after empty_stream error")
+	}
+}
+
+// TestWrapStreamEmptyCompletionDetectsSplitUsageOnlyStream is a regression
+// guard for the detector.Finish finding: the EOF branch used to re-parse the
+// concatenated payload, and separately chunked SSE frames without trailing
+// newlines concatenated into invalid input, so the empty check failed and an
+// empty plugin stream was flushed as success. The incremental detector state
+// now decides at EOF.
+func TestWrapStreamEmptyCompletionDetectsSplitUsageOnlyStream(t *testing.T) {
+	src := make(chan coreexecutor.StreamChunk, 2)
+	src <- coreexecutor.StreamChunk{Payload: []byte("data: {\"choices\":[],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":0}}")}
+	src <- coreexecutor.StreamChunk{Payload: []byte("data: [DONE]")}
+	close(src)
+
+	wrapped := wrapStreamEmptyCompletion(context.Background(), &coreexecutor.StreamResult{Chunks: src})
+	first, ok := <-wrapped.Chunks
+	if !ok {
+		t.Fatal("wrapped split usage-only stream closed without empty_completion error")
+	}
+	var authErr *coreauth.Error
+	if !errors.As(first.Err, &authErr) || authErr.Code != "empty_completion" {
+		t.Fatalf("first error = %v, want empty_completion", first.Err)
+	}
+	if len(first.Payload) != 0 {
+		t.Fatalf("first payload = %q, want no client-visible bytes before error", first.Payload)
+	}
+}