Merge branch 'main' into feat/ci-coverage-reporting

Author: atulmgupta

internal/ai/dispatch/dispatch.go

@@ -408,8 +408,8 @@ type turnResult struct {
 }
 
 // completeTurn uses the provider's real streaming path whenever advertised.
-// Capability drift falls back to Chat only when Stream refuses synchronously,
-// before any frame has been consumed.
+// Any Stream failure that happens before a frame is consumed falls back
+// to Chat (capability drift, Foundry stream 404, etc.).
 func (d *Dispatcher) completeTurn(
 	ctx context.Context,
 	req provider.ChatRequest,
@@ -420,7 +420,10 @@ func (d *Dispatcher) completeTurn(
 		if err == nil {
 			return turn, "stream", nil
 		}
-		if consumed || !errors.Is(err, provider.ErrCapabilityNotSupported) {
+		// Fall back to Chat only when Stream refused before any
+		// frame — including Azure Foundry 404 DeploymentNotFound
+		// on stream:true while the same Chat probe succeeds.
+		if consumed {
 			return turnResult{}, "stream", err
 		}
 	}

internal/ai/dispatch/dispatch_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 	"sync"
 	"testing"
@@ -420,6 +421,29 @@ func TestDispatcher_FallsBackWhenStreamCapabilityDrifts(t *testing.T) {
 	}
 }
 
+func TestDispatcher_FallsBackWhenStream404sBeforeFrames(t *testing.T) {
+	t.Parallel()
+
+	p := &streamingScriptedProvider{
+		streamErr: fmt.Errorf("%w: azure stream status 404: DeploymentNotFound", provider.ErrUpstream),
+		chatResp: &provider.ChatResponse{
+			Message: provider.Message{Role: provider.RoleAssistant, Content: "chat-ok"},
+		},
+	}
+	d := New(tools.NewRegistry(), p, nil, 0)
+	w := NewCaptureWriter()
+
+	if err := d.Run(context.Background(), fakeStrategy{}, strategy.StrategyInput{}, w); err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+	if got := strings.Join(w.Deltas(), ""); got != "chat-ok" {
+		t.Fatalf("joined deltas = %q", got)
+	}
+	if p.streamCalls != 1 || p.chatCalls != 1 {
+		t.Fatalf("provider calls: stream=%d chat=%d", p.streamCalls, p.chatCalls)
+	}
+}
+
 func TestDispatcher_SingleToolCall(t *testing.T) {
 	t.Parallel()
 	p := newScripted(

internal/ai/provider/azure/azure.go

@@ -206,13 +206,60 @@ func (a *Adapter) Stream(ctx context.Context, req provider.ChatRequest) (<-chan
 	if resp.StatusCode/100 != 2 {
 		raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
 		_ = resp.Body.Close()
-		return nil, fmt.Errorf("%w: azure stream status %d: %s", provider.ErrUpstream, resp.StatusCode, string(raw))
+		streamErr := fmt.Errorf("%w: azure stream status %d: %s", provider.ErrUpstream, resp.StatusCode, string(raw))
+		// Foundry OpenAI v1 (especially model-router) accepts the
+		// same chat/completions URL for non-stream Chat that
+		// Validate uses, but returns 404 DeploymentNotFound when
+		// the body has stream:true. Retry as Chat and synthesize
+		// chunks so Helix still completes.
+		if resp.StatusCode == http.StatusNotFound {
+			chatResp, chatErr := a.Chat(ctx, req)
+			if chatErr == nil {
+				return chatResponseAsStream(ctx, chatResp), nil
+			}
+		}
+		return nil, streamErr
 	}
 	out := make(chan provider.Chunk, 8)
 	go relayStream(ctx, resp.Body, out)
 	return out, nil
 }
 
+// chatResponseAsStream turns a completed Chat response into the
+// Stream channel shape dispatch already consumes.
+func chatResponseAsStream(ctx context.Context, resp *provider.ChatResponse) <-chan provider.Chunk {
+	out := make(chan provider.Chunk, 8)
+	go func() {
+		defer close(out)
+		if resp == nil {
+			send(ctx, out, provider.Chunk{Err: fmt.Errorf("%w: azure stream fallback returned nil chat", provider.ErrUpstream)})
+			return
+		}
+		if resp.Message.Content != "" {
+			send(ctx, out, provider.Chunk{Delta: resp.Message.Content})
+		}
+		for i := range resp.ToolCalls {
+			call := resp.ToolCalls[i]
+			send(ctx, out, provider.Chunk{ToolDelta: &call})
+		}
+		finish := resp.FinishReason
+		if finish == "" {
+			if len(resp.ToolCalls) > 0 {
+				finish = provider.FinishToolCalls
+			} else {
+				finish = provider.FinishStop
+			}
+		}
+		send(ctx, out, provider.Chunk{
+			Done:         true,
+			FinishReason: finish,
+			InputTokens:  resp.InputTokens,
+			OutputTokens: resp.OutputTokens,
+		})
+	}()
+	return out
+}
+
 // Embed uses the Azure embeddings route. URL shape depends on flavor:
 //
 //   - OpenAI flavor:  {base}/openai/deployments/{depl}/embeddings?api-version=...

internal/ai/provider/azure/azure_test.go

@@ -658,6 +658,61 @@ func TestChatEndpoint_V1DoesNotDuplicatePath(t *testing.T) {
 	}
 }
 
+func TestStream_V1NotFoundFallsBackToChat(t *testing.T) {
+	t.Parallel()
+	var streamHits, chatHits int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		var probe map[string]any
+		_ = json.Unmarshal(body, &probe)
+		stream, _ := probe["stream"].(bool)
+		if stream {
+			streamHits++
+			w.WriteHeader(http.StatusNotFound)
+			_, _ = io.WriteString(w, `{ "error": { "type": "invalid_request_error", "code": "DeploymentNotFound", "message": "The API deployment for this resource does not exist." } }`)
+			return
+		}
+		chatHits++
+		_, _ = io.WriteString(w, `{"choices":[{"index":0,"finish_reason":"stop","message":{"role":"assistant","content":"from-chat"}}],"usage":{"prompt_tokens":4,"completion_tokens":2}}`)
+	}))
+	t.Cleanup(srv.Close)
+	a, err := New(provider.ProviderConfig{
+		BaseURL: srv.URL + "/openai/v1",
+		Model:   "model-router",
+		APIKey:  "k",
+		Flavor:  provider.AzureFlavorOpenAI,
+	}, WithHTTPClient(srv.Client()))
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	ch, err := a.Stream(context.Background(), provider.ChatRequest{
+		Messages: []provider.Message{{Role: provider.RoleUser, Content: "hi"}},
+	})
+	if err != nil {
+		t.Fatalf("Stream: %v", err)
+	}
+	var content string
+	var terminal provider.Chunk
+	for c := range ch {
+		if c.Err != nil {
+			t.Fatalf("chunk err: %v", c.Err)
+		}
+		content += c.Delta
+		if c.Done {
+			terminal = c
+		}
+	}
+	if streamHits != 1 || chatHits != 1 {
+		t.Fatalf("hits stream=%d chat=%d", streamHits, chatHits)
+	}
+	if content != "from-chat" {
+		t.Fatalf("content=%q", content)
+	}
+	if !terminal.Done || terminal.FinishReason != provider.FinishStop {
+		t.Fatalf("terminal = %+v", terminal)
+	}
+}
+
 func TestOpenAIV1_Chat_URLAuthAndBody(t *testing.T) {
 	t.Parallel()
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {