fix(client): route OpenCode Go through dual-protocol client with fallback (#32)

Use OpenCodeGoClient for deployment adapters so MiniMax/Qwen models hit
Anthropic /v1/messages and transparently fall back to OpenAI chat when the
stream is reasoning-only.

Author: Patel230

client/opencodego.go

@@ -13,6 +13,8 @@ type OpenCodeGoClient struct {
 	anthropic *AnthropicClient
 }
 
+type openCodeGoStreamOpener func(context.Context, []EyrieMessage, ChatOptions) (*StreamResult, error)
+
 // NewOpenCodeGoClient builds a dual-protocol OpenCode Go provider client.
 func NewOpenCodeGoClient(apiKey, baseURL string, opts ...ClientOption) *OpenCodeGoClient {
 	openBase := strings.TrimRight(strings.TrimSpace(baseURL), "/")
@@ -30,7 +32,21 @@ func (c *OpenCodeGoClient) Name() string { return "opencodego" }
 
 func (c *OpenCodeGoClient) Chat(ctx context.Context, messages []EyrieMessage, opts ChatOptions) (*EyrieResponse, error) {
 	if openCodeGoUsesAnthropicMessages(opts.Model) {
-		return c.anthropic.Chat(ctx, messages, opts)
+		resp, err := c.anthropic.Chat(ctx, messages, opts)
+		if err != nil {
+			return nil, err
+		}
+		if resp != nil && strings.TrimSpace(resp.Content) != "" {
+			return resp, nil
+		}
+		if c.openAI == nil {
+			return resp, err
+		}
+		openResp, openErr := c.openAI.Chat(ctx, messages, opts)
+		if openErr == nil && openResp != nil && strings.TrimSpace(openResp.Content) != "" {
+			return openResp, nil
+		}
+		return resp, err
 	}
 	resp, err := c.openAI.Chat(ctx, messages, opts)
 	if err != nil || c.anthropic == nil || !openCodeGoMightNeedAnthropicFallback(opts.Model) {
@@ -48,13 +64,17 @@ func (c *OpenCodeGoClient) Chat(ctx context.Context, messages []EyrieMessage, op
 
 func (c *OpenCodeGoClient) StreamChat(ctx context.Context, messages []EyrieMessage, opts ChatOptions) (*StreamResult, error) {
 	if openCodeGoUsesAnthropicMessages(opts.Model) {
-		return c.anthropic.StreamChat(ctx, messages, opts)
+		primary, err := c.anthropic.StreamChat(ctx, messages, opts)
+		if err != nil || c.openAI == nil {
+			return primary, err
+		}
+		return newOpenCodeGoStreamWithFallback(ctx, messages, opts, primary, c.openAI.StreamChat), nil
 	}
 	primary, err := c.openAI.StreamChat(ctx, messages, opts)
 	if err != nil || c.anthropic == nil || !openCodeGoMightNeedAnthropicFallback(opts.Model) {
 		return primary, err
 	}
-	return newOpenCodeGoFallbackStream(ctx, c.anthropic, messages, opts, primary), nil
+	return newOpenCodeGoStreamWithFallback(ctx, messages, opts, primary, c.anthropic.StreamChat), nil
 }
 
 func (c *OpenCodeGoClient) Ping(ctx context.Context) error {
@@ -64,9 +84,10 @@ func (c *OpenCodeGoClient) Ping(ctx context.Context) error {
 	return c.anthropic.Ping(ctx)
 }
 
-// newOpenCodeGoFallbackStream watches an OpenAI-format stream; if it ends with
-// reasoning tokens but no answer, transparently retries via /v1/messages.
-func newOpenCodeGoFallbackStream(ctx context.Context, anthropic *AnthropicClient, messages []EyrieMessage, opts ChatOptions, primary *StreamResult) *StreamResult {
+// newOpenCodeGoStreamWithFallback watches a primary stream; if it ends with
+// reasoning tokens but no answer, transparently retries via the fallback opener.
+// Used for OpenAI→Anthropic and Anthropic→OpenAI on MiniMax/Qwen models.
+func newOpenCodeGoStreamWithFallback(ctx context.Context, messages []EyrieMessage, opts ChatOptions, primary *StreamResult, fallback openCodeGoStreamOpener) *StreamResult {
 	out := make(chan EyrieStreamEvent, streamChannelBuffer)
 	cancelCtx, cancel := context.WithCancel(ctx)
 	go func() {
@@ -126,7 +147,7 @@ func newOpenCodeGoFallbackStream(ctx context.Context, anthropic *AnthropicClient
 			return
 		}
 
-		fallback, err := anthropic.StreamChat(cancelCtx, messages, opts)
+		fallbackResult, err := fallback(cancelCtx, messages, opts)
 		if err != nil {
 			flush()
 			select {
@@ -135,8 +156,8 @@ func newOpenCodeGoFallbackStream(ctx context.Context, anthropic *AnthropicClient
 			}
 			return
 		}
-		defer fallback.Close()
-		for ev := range fallback.Events {
+		defer fallbackResult.Close()
+		for ev := range fallbackResult.Events {
 			select {
 			case out <- ev:
 			case <-cancelCtx.Done():

client/opencodego_test.go

@@ -2,7 +2,9 @@ package client
 
 import (
 	"context"
+	"fmt"
 	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 )
@@ -101,3 +103,51 @@ func TestOpenCodeGoClient_RoutesKimiToOpenAI(t *testing.T) {
 		t.Fatalf("path = %q, want suffix /chat/completions", gotPath)
 	}
 }
+
+func TestOpenCodeGoClient_AnthropicReasoningOnlyFallsBackToOpenAI(t *testing.T) {
+	var paths []string
+	transport := roundTripFunc(func(r *http.Request) (*http.Response, error) {
+		paths = append(paths, r.URL.Path)
+		if strings.HasSuffix(r.URL.Path, "/messages") {
+			w := httptest.NewRecorder()
+			w.Header().Set("Content-Type", "text/event-stream")
+			_, _ = fmt.Fprintf(w, "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"thinking\"}}\n\n")
+			_, _ = fmt.Fprintf(w, "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"thinking_delta\",\"text\":\"hmm\"}}\n\n")
+			_, _ = fmt.Fprintf(w, "event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n")
+			_, _ = fmt.Fprintf(w, "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n")
+			return w.Result(), nil
+		}
+		w := httptest.NewRecorder()
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"content\":\"Hello!\"},\"finish_reason\":null}]}\n\n")
+		_, _ = fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n")
+		_, _ = fmt.Fprintf(w, "data: [DONE]\n\n")
+		return w.Result(), nil
+	})
+
+	c := NewOpenCodeGoClient("ocg-test-key", "https://opencode.example/zen/go/v1")
+	c.anthropic.httpClient = &http.Client{Transport: transport}
+	c.openAI.httpClient = &http.Client{Transport: transport}
+
+	sr, err := c.StreamChat(context.Background(), []EyrieMessage{{Role: "user", Content: "Hi"}}, ChatOptions{
+		Model:     "minimax-m3",
+		MaxTokens: 256,
+	})
+	if err != nil {
+		t.Fatalf("StreamChat: %v", err)
+	}
+	defer sr.Close()
+
+	var content string
+	for ev := range sr.Events {
+		if ev.Type == "content" {
+			content += ev.Content
+		}
+	}
+	if content != "Hello!" {
+		t.Fatalf("content = %q, want Hello!; paths=%v", content, paths)
+	}
+	if len(paths) < 2 {
+		t.Fatalf("expected anthropic then openai paths, got %v", paths)
+	}
+}

client/stream.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
@@ -68,6 +69,14 @@ func parseSSEStream(ctx context.Context, body io.ReadCloser, logger *slog.Logger
 			}
 		}
 		if err := scanner.Err(); err != nil {
+			// Context cancellation produces "context canceled" from the
+			// scanner when the body is closed; that is the expected
+			// shutdown path, not a stream error. Skip the warning and
+			// the synthetic error event so callers (and operators) don't
+			// see noise for every normal cancel/close.
+			if ctxErr := ctx.Err(); ctxErr != nil || errors.Is(err, context.Canceled) {
+				return
+			}
 			logger.Warn("SSE stream read error", "error", err)
 			select {
 			case ch <- SSEEvent{Event: "error", Data: fmt.Sprintf("stream read error: %v", err)}:
@@ -109,6 +118,7 @@ func processAnthropicStream(ctx context.Context, sseEvents <-chan SSEEvent, logg
 			jsonBuf  strings.Builder
 		}
 		var currentTool *toolAccum
+		blockTypes := make(map[int]string)
 		var stopReason string
 
 		for {
@@ -145,6 +155,7 @@ func processAnthropicStream(ctx context.Context, sseEvents <-chan SSEEvent, logg
 				switch ae.Type {
 				case "content_block_start":
 					if ae.ContentBlock != nil {
+						blockTypes[ae.Index] = ae.ContentBlock.Type
 						switch ae.ContentBlock.Type {
 						case "tool_use":
 							currentTool = &toolAccum{id: ae.ContentBlock.ID, name: ae.ContentBlock.Name}
@@ -160,7 +171,11 @@ func processAnthropicStream(ctx context.Context, sseEvents <-chan SSEEvent, logg
 					switch ae.Delta.Type {
 					case "text_delta":
 						if ae.Delta.Text != "" {
-							emit(ctx, ch, EyrieStreamEvent{Type: "content", Content: ae.Delta.Text})
+							if strings.Contains(blockTypes[ae.Index], "thinking") {
+								emit(ctx, ch, EyrieStreamEvent{Type: "thinking", Thinking: ae.Delta.Text})
+							} else {
+								emit(ctx, ch, EyrieStreamEvent{Type: "content", Content: ae.Delta.Text})
+							}
 						}
 					case "input_json_delta":
 						if currentTool != nil && ae.Delta.PartialJSON != "" {
@@ -173,6 +188,7 @@ func processAnthropicStream(ctx context.Context, sseEvents <-chan SSEEvent, logg
 					}
 
 				case "content_block_stop":
+					delete(blockTypes, ae.Index)
 					if currentTool != nil {
 						rawJSON := currentTool.jsonBuf.String()
 						var args map[string]interface{}

client/stream_test.go

@@ -217,6 +217,36 @@ func TestSSEAnthropicThinkingDelta(t *testing.T) {
 	}
 }
 
+func TestSSEAnthropicThinkingTextDeltaHidden(t *testing.T) {
+	events := make(chan SSEEvent, 10)
+	events <- SSEEvent{Event: "content_block_start", Data: `{"type":"content_block_start","index":0,"content_block":{"type":"thinking"}}`}
+	events <- SSEEvent{Event: "content_block_delta", Data: `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"private reasoning"}}`}
+	events <- SSEEvent{Event: "content_block_stop", Data: `{"type":"content_block_stop","index":0}`}
+	events <- SSEEvent{Event: "message_stop", Data: `{"type":"message_stop"}`}
+	close(events)
+
+	ctx := context.Background()
+	ch := processAnthropicStream(ctx, events, testLogger())
+
+	var sawThinking, sawContent bool
+	for evt := range ch {
+		switch evt.Type {
+		case "thinking":
+			if evt.Thinking == "private reasoning" {
+				sawThinking = true
+			}
+		case "content":
+			sawContent = true
+		}
+	}
+	if !sawThinking {
+		t.Fatal("expected thinking event for text_delta inside thinking block")
+	}
+	if sawContent {
+		t.Fatal("did not expect visible content from thinking block text_delta")
+	}
+}
+
 func TestSSEAnthropicStopReason(t *testing.T) {
 	events := make(chan SSEEvent, 10)
 	events <- SSEEvent{Event: "message_delta", Data: `{"type":"message_delta","delta":{"stop_reason":"max_tokens"}}`}

setup/deployment.go

@@ -232,7 +232,7 @@ func ProviderForDeployment(id string, deployment config.DeploymentConfig) (clien
 		if apiKey == "" {
 			return nil, false
 		}
-		return client.NewOpenAIClient(apiKey, FirstNonEmpty(deployment.BaseURL, config.DefaultOpenCodeGoBaseURL), &client.OpenCodeGoCompat), true
+		return client.NewOpenCodeGoClient(apiKey, FirstNonEmpty(deployment.BaseURL, config.DefaultOpenCodeGoBaseURL)), true
 	case "kimi-direct":
 		apiKey := FirstNonEmpty(deployment.APIKey, storeSecret("MOONSHOT_API_KEY"))
 		if apiKey == "" {

setup/deployment_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"testing"
 
+	"github.com/GrayCodeAI/eyrie/client"
 	"github.com/GrayCodeAI/eyrie/config"
 	"github.com/GrayCodeAI/eyrie/credentials"
 )
@@ -593,7 +594,10 @@ func TestProviderForDeployment_OpenCodeGo(t *testing.T) {
 	if !ok {
 		t.Fatal("expected opencodego to be configured")
 	}
-	if p.Name() != "openai" {
-		t.Fatalf("provider name = %q, want openai", p.Name())
+	if p.Name() != "opencodego" {
+		t.Fatalf("provider name = %q, want opencodego", p.Name())
+	}
+	if _, ok := p.(*client.OpenCodeGoClient); !ok {
+		t.Fatalf("provider type = %T, want *client.OpenCodeGoClient", p)
 	}
 }