fix(auth): treat empty upstream completions as retriable failures

internal/clienterror/client_error.go

@@ -87,9 +87,12 @@ func IsRequestFault(status int, err error) bool {
 		return false
 	}
 	// DeepSeek reports an invalid API key as 401 with the authentication_error
-	// type alongside the same generic code. Preserve that credential failure
-	// classification without weakening generic request-fault handling.
-	if status == http.StatusUnauthorized && hasAuthenticationErrorBody(err) {
+	// type alongside the same generic code. Other providers also surface an
+	// invalid or expired credential as the authentication_error body on 403.
+	// Preserve that credential failure classification without weakening generic
+	// request-fault handling: a request-fault-looking code on the same body does
+	// not turn a credential rejection into a request fault.
+	if (status == http.StatusUnauthorized || status == http.StatusForbidden) && hasAuthenticationErrorBody(err) {
 		return false
 	}
 	if hasRequestFaultBody(err) {

internal/clienterror/client_error_test.go

@@ -197,6 +197,23 @@ func TestIsRequestFault(t *testing.T) {
 			err:    errors.New(`{"error":{"code":"invalid_request_error","message":"Authentication Fails, Your api key: ****heck is invalid","param":null,"type":"authentication_error"}}`),
 			want:   false,
 		},
+		{
+			// Codex surfaces an invalid or expired API key as 403 with the
+			// authentication_error body. The credential must stay eligible for
+			// rotation, so this is not a request fault.
+			name:   "codex invalid or expired key on 403 is credential failure",
+			status: http.StatusForbidden,
+			err:    errors.New(`{"error":{"message":"invalid or expired token","type":"authentication_error","code":"invalid_api_key"}}`),
+			want:   false,
+		},
+		{
+			// A credential rejection must not be reclassified as a request fault
+			// merely because the same body carries a request-fault-looking code.
+			name:   "authentication body with generic code on 403 is credential failure",
+			status: http.StatusForbidden,
+			err:    errors.New(`{"error":{"code":"invalid_request_error","message":"Authentication Fails, Your api key: ****heck is invalid","param":null,"type":"authentication_error"}}`),
+			want:   false,
+		},
 		{
 			name:   "deepseek insufficient balance is payment failure",
 			status: http.StatusPaymentRequired,

internal/config/sdk_config.go

@@ -79,4 +79,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.
+	// Zero or a negative value disables the timeout.
+	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/interfaces/error_message.go

@@ -21,6 +21,10 @@ 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 from a
+	// trusted local interceptor rather than an untrusted upstream error.
+	TrustedDirectResponse bool
+
 	// Body contains a preformatted downstream response when DirectResponse is true.
 	Body []byte
 

internal/pluginhost/executor_route.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth"
 	coreexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor"
@@ -88,7 +89,14 @@ func (h *Host) ExecutePluginExecutor(ctx context.Context, pluginID string, req c
 	if errAdapter != nil {
 		return coreexecutor.Response{}, errAdapter
 	}
-	return adapter.Execute(ctx, (*coreauth.Auth)(nil), req, opts)
+	resp, err := adapter.Execute(ctx, (*coreauth.Auth)(nil), req, opts)
+	if err != nil {
+		return coreexecutor.Response{}, err
+	}
+	if coreauth.IsEmptyCompletionPayload(resp.Payload) {
+		return coreexecutor.Response{}, coreauth.EmptyCompletionError()
+	}
+	return resp, nil
 }
 
 // ExecutePluginExecutorStream executes a streaming request with the named plugin executor without changing the requested model.
@@ -97,7 +105,212 @@ func (h *Host) ExecutePluginExecutorStream(ctx context.Context, pluginID string,
 	if errAdapter != nil {
 		return nil, errAdapter
 	}
-	return adapter.ExecuteStream(ctx, (*coreauth.Auth)(nil), req, opts)
+	streamResult, err := adapter.ExecuteStream(ctx, (*coreauth.Auth)(nil), req, opts)
+	if err != nil {
+		return nil, err
+	}
+	return wrapStreamEmptyCompletion(ctx, streamResult, req.Payload, opts.OriginalRequest), nil
+}
+
+// wrapStreamEmptyCompletion wraps a plugin stream so that a terminal but empty
+// completion (no content, no tool calls) surfaces as an empty-completion error
+// instead of a clean stream end, mirroring the conductor's aggregate-at-close
+// judgment. Recognized protocol framing is buffered only until meaningful output
+// appears or the stream closes; unrecognized streams remain pass-through.
+func wrapStreamEmptyCompletion(ctx context.Context, streamResult *coreexecutor.StreamResult, requestPayloads ...[]byte) *coreexecutor.StreamResult {
+	if streamResult == nil || streamResult.Chunks == nil {
+		errChunks := make(chan coreexecutor.StreamChunk, 1)
+		errChunks <- coreexecutor.StreamChunk{Err: &coreauth.Error{
+			Code:      "empty_stream",
+			Message:   "upstream stream has no source",
+			Retryable: true,
+		}}
+		close(errChunks)
+		wrapped := &coreexecutor.StreamResult{Chunks: errChunks}
+		if streamResult != nil {
+			wrapped.Headers = streamResult.Headers
+		}
+		return wrapped
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	src := streamResult.Chunks
+	wrapped := make(chan coreexecutor.StreamChunk)
+	go func() {
+		defer close(wrapped)
+		buffered := make([]coreexecutor.StreamChunk, 0, 1)
+		var detector coreauth.StreamBootstrapDetector
+		for _, p := range requestPayloads {
+			if n := coreauth.ExtractExpectedChoices(p); n > 1 {
+				detector.SetExpectedChoices(n)
+				break
+			}
+		}
+		forwarding := false
+		var payloadErrors coreauth.StreamPayloadErrorDetector
+		forward := func(chunk coreexecutor.StreamChunk) bool {
+			if chunk.Err != nil {
+				chunk.Err = coreauth.SanitizeError(chunk.Err)
+			}
+			errorPath := chunk.Err != nil || detector.StreamError() != nil
+			if len(chunk.Payload) > 0 {
+				if err := payloadErrors.Observe(chunk.Payload); err != nil {
+					errorPath = true
+				}
+				if errorPath {
+					chunk.Payload = []byte(coreauth.RedactSecrets(string(chunk.Payload)))
+				}
+			}
+			select {
+			case <-ctx.Done():
+				return false
+			case wrapped <- chunk:
+				return true
+			}
+		}
+		flush := func() bool {
+			for _, chunk := range buffered {
+				if !forward(chunk) {
+					return false
+				}
+			}
+			buffered = nil
+			return true
+		}
+
+		for {
+			var (
+				chunk coreexecutor.StreamChunk
+				ok    bool
+			)
+			select {
+			case <-ctx.Done():
+				return
+			case chunk, ok = <-src:
+			}
+			if !ok {
+				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.
+					// Finish() parses the trailing fragment, so a provider error that only
+					// lands in that final unterminated frame is not known until after it
+					// runs: consult StreamError() before reporting terminal emptiness.
+					terminalEmpty := detector.Finish()
+					if streamErr := detector.StreamError(); streamErr != nil {
+						_ = forward(coreexecutor.StreamChunk{Err: streamErr})
+						return
+					}
+					if terminalEmpty {
+						_ = forward(coreexecutor.StreamChunk{Err: coreauth.EmptyCompletionError()})
+						return
+					}
+				}
+				_ = flush()
+				return
+			}
+			if forwarding {
+				if !forward(chunk) {
+					return
+				}
+				if chunk.Err != nil {
+					return
+				}
+				continue
+			}
+
+			buffered = append(buffered, chunk)
+			if chunk.Err != nil {
+				// Before any semantic output, protocol framing is not client-visible.
+				// Surface the upstream failure first so the HTTP layer can still
+				// choose an error response instead of committing a successful stream.
+				buffered = buffered[:0]
+				forwarding = true
+				if !forward(chunk) {
+					return
+				}
+				return
+			}
+			if detector.Observe(chunk.Payload) {
+				forwarding = true
+				if !flush() {
+					return
+				}
+			}
+			if streamErr := detector.StreamError(); streamErr != nil {
+				discardStreamChunks(ctx, src)
+				_ = forward(coreexecutor.StreamChunk{Err: streamErr})
+				return
+			}
+			if detector.IsTerminalEmpty() {
+				discardStreamChunks(ctx, src)
+				_ = forward(coreexecutor.StreamChunk{Err: coreauth.EmptyCompletionError()})
+				return
+			}
+		}
+	}()
+	return &coreexecutor.StreamResult{Chunks: wrapped, Headers: streamResult.Headers}
+}
+
+var streamDrainTimeout = 5 * time.Second
+
+func discardStreamChunks(ctx context.Context, ch <-chan coreexecutor.StreamChunk) <-chan struct{} {
+	done := make(chan struct{})
+	if ch == nil {
+		close(done)
+		return done
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	go func() {
+		defer close(done)
+		timer := time.NewTimer(streamDrainTimeout)
+		defer timer.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-timer.C:
+				return
+			case _, ok := <-ch:
+				if !ok {
+					return
+				}
+				if !timer.Stop() {
+					select {
+					case <-timer.C:
+					default:
+					}
+				}
+				timer.Reset(streamDrainTimeout)
+			}
+		}
+	}()
+	return done
+}
+
+func streamChunkPayload(chunks []coreexecutor.StreamChunk) []byte {
+	var payload []byte
+	for _, chunk := range chunks {
+		payload = append(payload, chunk.Payload...)
+	}
+	return payload
 }
 
 // CountPluginExecutor executes a count-tokens request with the named plugin executor without changing the requested model.
@@ -106,7 +319,14 @@ func (h *Host) CountPluginExecutor(ctx context.Context, pluginID string, req cor
 	if errAdapter != nil {
 		return coreexecutor.Response{}, errAdapter
 	}
-	return adapter.CountTokens(ctx, (*coreauth.Auth)(nil), req, opts)
+	resp, err := adapter.CountTokens(ctx, (*coreauth.Auth)(nil), req, opts)
+	if err != nil {
+		return coreexecutor.Response{}, err
+	}
+	if coreauth.IsEmptyCompletionPayload(resp.Payload) {
+		return coreexecutor.Response{}, coreauth.EmptyCountError()
+	}
+	return resp, nil
 }
 
 func (h *Host) executorAdapterForPlugin(pluginID string) (*executorAdapter, error) {

internal/pluginhost/executor_route_close_order_test.go

@@ -0,0 +1,44 @@
+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"
+)
+
+// TestWrapStreamEmptyCompletion_PrefersDetectedErrorOverTerminalEmptiness covers a
+// stream whose recognized frames carry no content and whose real provider error
+// arrives as an SSE error event that is newline-terminated but never followed by the
+// blank separator line. flushData() only runs on that blank line or from Finish(),
+// so the detected provider error does not exist yet while the stream is being
+// observed; judging emptiness before consulting it would replace a routable
+// invalid_api_key with a generic empty_completion.
+func TestWrapStreamEmptyCompletion_PrefersDetectedErrorOverTerminalEmptiness(t *testing.T) {
+	chunks := make(chan coreexecutor.StreamChunk, 3)
+	chunks <- coreexecutor.StreamChunk{Payload: []byte("data: {\"choices\":[{\"delta\":{\"role\":\"assistant\"}}]}\n\n")}
+	chunks <- coreexecutor.StreamChunk{Payload: []byte("event: error\ndata: {\"error\":{\"code\":\"invalid_api_key\",\"message\":\"invalid api key\"}}\n")}
+	close(chunks)
+
+	res := wrapStreamEmptyCompletion(context.Background(), &coreexecutor.StreamResult{Chunks: chunks})
+	var received []coreexecutor.StreamChunk
+	for c := range res.Chunks {
+		received = append(received, c)
+	}
+
+	if len(received) != 1 {
+		t.Fatalf("expected 1 chunk carrying the detected provider error, got %d", len(received))
+	}
+	if received[0].Err == nil {
+		t.Fatalf("expected an error chunk, got payload: %s", string(received[0].Payload))
+	}
+	var authErr *coreauth.Error
+	if !errors.As(received[0].Err, &authErr) {
+		t.Fatalf("expected *coreauth.Error, got %v", received[0].Err)
+	}
+	if authErr.Code != "invalid_api_key" {
+		t.Fatalf("expected the provider error to survive terminal emptiness, got code %q (%v)", authErr.Code, received[0].Err)
+	}
+}