Fix cloud model refresh fallbacks

Refresh OpenCSG models on startup and recover from empty or unauthorized cloud model lists so Chat keeps showing available OpenCSG and provider models.

Author: ganisback

internal/cloud/opencsg.go

@@ -197,29 +197,67 @@ func (s *Service) cachedModels() ([]api.ModelInfo, bool) {
 
 func (s *Service) refresh(ctx context.Context) ([]api.ModelInfo, error) {
 	baseURL := s.BaseURL()
+	token := s.currentAccessToken()
+	models, limits, err := fetchCloudModels(ctx, s.client, baseURL, token)
+	if err != nil && token != "" && isUnauthorizedStatus(err) {
+		models, limits, err = fetchCloudModels(ctx, s.client, baseURL, "")
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	s.mu.Lock()
+	s.cached = cloneModels(models)
+	s.limits = limits
+	s.cachedAt = time.Now()
+	s.mu.Unlock()
+
+	return models, nil
+}
+
+type cloudModelListStatusError struct {
+	statusCode int
+	body       string
+}
+
+func (e cloudModelListStatusError) Error() string {
+	return fmt.Sprintf("cloud model list returned %d: %s", e.statusCode, e.body)
+}
+
+func isUnauthorizedStatus(err error) bool {
+	if statusErr, ok := err.(cloudModelListStatusError); ok {
+		return statusErr.statusCode == http.StatusUnauthorized || statusErr.statusCode == http.StatusForbidden
+	}
+	return false
+}
+
+func fetchCloudModels(ctx context.Context, client *http.Client, baseURL, token string) ([]api.ModelInfo, map[string]ModelTokenLimits, error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, baseURL+"/v1/models?page="+cloudModelListPage+"&per="+cloudModelListPer, nil)
 	if err != nil {
-		return nil, fmt.Errorf("creating cloud model request: %w", err)
+		return nil, nil, fmt.Errorf("creating cloud model request: %w", err)
 	}
 	req.Header.Set("Accept", "application/json")
-	if token := s.currentAccessToken(); token != "" {
+	if token != "" {
 		req.Header.Set("Authorization", "Bearer "+token)
 	}
 
-	resp, err := s.client.Do(req)
+	resp, err := client.Do(req)
 	if err != nil {
-		return nil, fmt.Errorf("fetching cloud models: %w", err)
+		return nil, nil, fmt.Errorf("fetching cloud models: %w", err)
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
 		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
-		return nil, fmt.Errorf("cloud model list returned %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
+		return nil, nil, cloudModelListStatusError{
+			statusCode: resp.StatusCode,
+			body:       strings.TrimSpace(string(body)),
+		}
 	}
 
 	var payload modelListResponse
 	if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil {
-		return nil, fmt.Errorf("decoding cloud model list: %w", err)
+		return nil, nil, fmt.Errorf("decoding cloud model list: %w", err)
 	}
 
 	models := make([]api.ModelInfo, 0, len(payload.Data))
@@ -233,13 +271,7 @@ func (s *Service) refresh(ctx context.Context) ([]api.ModelInfo, error) {
 		limits[strings.TrimSpace(info.Model)] = modelTokenLimitsFromRemote(item)
 	}
 
-	s.mu.Lock()
-	s.cached = cloneModels(models)
-	s.limits = limits
-	s.cachedAt = time.Now()
-	s.mu.Unlock()
-
-	return models, nil
+	return models, limits, nil
 }
 
 func (s *Service) currentAccessToken() string {

internal/cloud/opencsg_test.go

@@ -312,6 +312,42 @@ func TestRefreshChatModelsOmitsAuthorizationWithoutAccessToken(t *testing.T) {
 	}
 }
 
+func TestRefreshChatModelsFallsBackToPublicListWhenAccessTokenUnauthorized(t *testing.T) {
+	requests := 0
+	apiServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests++
+		if got := r.Header.Get("Authorization"); got != "" {
+			http.Error(w, "expired token", http.StatusUnauthorized)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"object": "list",
+			"data": []map[string]any{
+				{
+					"id":   "public/model",
+					"task": "text-generation",
+				},
+			},
+		})
+	}))
+	defer apiServer.Close()
+
+	svc := NewService(apiServer.URL)
+	svc.SetAccessToken("expired-token")
+
+	models, err := svc.RefreshChatModels(context.Background())
+	if err != nil {
+		t.Fatalf("RefreshChatModels returned error: %v", err)
+	}
+	if requests != 2 {
+		t.Fatalf("requests = %d, want authenticated request plus public fallback", requests)
+	}
+	if len(models) != 1 || models[0].Model != "public/model" {
+		t.Fatalf("models = %#v, want public/model", models)
+	}
+}
+
 func TestRefreshChatModelsBypassesCache(t *testing.T) {
 	requests := 0
 	apiServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

internal/server/cloud_models.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"context"
+	"log"
 	"net/http"
 	"sort"
 	"strings"
@@ -12,6 +13,7 @@ import (
 )
 
 const forcedCloudModelRefreshInterval = 30 * time.Second
+const startupCloudModelRefreshTimeout = 20 * time.Second
 
 func requestWantsModelRefresh(r *http.Request) bool {
 	value := strings.TrimSpace(strings.ToLower(r.URL.Query().Get("refresh")))
@@ -23,6 +25,21 @@ func requestWantsModelRefresh(r *http.Request) bool {
 	}
 }
 
+func (s *Server) refreshCloudModelsOnStartup(parent context.Context) {
+	if s == nil || s.cloud == nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(parent, startupCloudModelRefreshTimeout)
+	defer cancel()
+
+	models, err := s.refreshCloudChatModels(ctx)
+	if err != nil {
+		log.Printf("startup cloud model refresh failed: %v", err)
+		return
+	}
+	log.Printf("startup cloud model refresh complete: %d models", len(models))
+}
+
 func (s *Server) listAvailableModelsWithRefresh(ctx context.Context, refreshCloud bool) ([]api.ModelInfo, error) {
 	localModels, err := s.listLocalModelInfos()
 	if err != nil {
@@ -56,6 +73,8 @@ func (s *Server) listAvailableModelsWithRefresh(ctx context.Context, refreshClou
 			seen[modelID] = struct{}{}
 			out = append(out, item)
 		}
+	} else {
+		log.Printf("cloud model list unavailable: %v", err)
 	}
 
 	for _, item := range s.listSelectedThirdPartyProviderModels(ctx) {
@@ -88,6 +107,9 @@ func (s *Server) listCloudModels(ctx context.Context, refresh bool) ([]api.Model
 		return s.withConfiguredCloudProvider(models), err
 	}
 	models, err := s.cloud.ListChatModels(ctx)
+	if err == nil && len(models) == 0 {
+		models, err = s.refreshCloudChatModels(ctx)
+	}
 	return s.withConfiguredCloudProvider(models), err
 }
 

internal/server/cloud_models_test.go

@@ -75,6 +75,47 @@ func TestHandleTagsWithoutTokenIncludesAndRefreshesCloudModels(t *testing.T) {
 	}
 }
 
+func TestHandleTagsRefreshesCloudModelsWhenCacheIsEmpty(t *testing.T) {
+	requests := 0
+	apiServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests++
+		w.Header().Set("Content-Type", "application/json")
+		data := []map[string]any{}
+		if requests > 1 {
+			data = append(data, map[string]any{
+				"id":   "fallback/model",
+				"task": "text-generation",
+			})
+		}
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"object": "list",
+			"data":   data,
+		})
+	}))
+	defer apiServer.Close()
+
+	s := newTestServer(t)
+	s.cloud = cloud.NewService(apiServer.URL)
+
+	req := httptest.NewRequest(http.MethodGet, "/api/tags", nil)
+	w := httptest.NewRecorder()
+	s.handleTags(w, req)
+	if w.Code != http.StatusOK {
+		t.Fatalf("status = %d, want %d", w.Code, http.StatusOK)
+	}
+
+	var resp api.TagsResponse
+	if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
+		t.Fatalf("decode tags response: %v", err)
+	}
+	if requests != 2 {
+		t.Fatalf("requests = %d, want initial empty fetch plus fallback refresh", requests)
+	}
+	if len(resp.Models) != 1 || resp.Models[0].Model != "fallback/model" {
+		t.Fatalf("models = %#v, want fallback/model", resp.Models)
+	}
+}
+
 func TestHandleTagsSendsLoginTokenToCloudGateway(t *testing.T) {
 	apiServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if got := r.Header.Get("Authorization"); got != "Bearer access-token" {

internal/server/handlers_audio.go

@@ -132,6 +132,7 @@ func (s *Server) streamAudioTranscription(w http.ResponseWriter, r *http.Request
 	})
 	if err != nil {
 		log.Printf("MODEL %s: ASR stream transcription failed: %v", modelID, err)
+		s.closeASREngine(modelID)
 		writeSSE(w, map[string]interface{}{
 			"error": err.Error(),
 			"done":  true,

internal/server/handlers_audio_test.go

@@ -2,6 +2,8 @@ package server
 
 import (
 	"bytes"
+	"context"
+	"errors"
 	"io"
 	"mime/multipart"
 	"net/http"
@@ -12,6 +14,7 @@ import (
 	"testing"
 
 	"github.com/opencsgs/csghub-lite/internal/config"
+	"github.com/opencsgs/csghub-lite/pkg/api"
 )
 
 type zeroReader struct{}
@@ -21,6 +24,27 @@ func (zeroReader) Read(p []byte) (int, error) {
 	return len(p), nil
 }
 
+type failingStreamASREngine struct {
+	closed bool
+}
+
+func (e *failingStreamASREngine) Transcribe(context.Context, api.OpenAIAudioTranscriptionRequest) (*api.OpenAIAudioTranscriptionResponse, error) {
+	return nil, errors.New("not used")
+}
+
+func (e *failingStreamASREngine) TranscribeStream(context.Context, api.OpenAIAudioTranscriptionRequest, func(api.OpenAIAudioTranscriptionResponse) error) error {
+	return errors.New("worker exited")
+}
+
+func (e *failingStreamASREngine) Close() error {
+	e.closed = true
+	return nil
+}
+
+func (e *failingStreamASREngine) ModelName() string {
+	return "test-asr"
+}
+
 func TestHandleOpenAIAudioTranscriptionsUsesLiteTempDir(t *testing.T) {
 	missingTempDir := filepath.Join(t.TempDir(), "missing-temp")
 	t.Setenv("TMPDIR", missingTempDir)
@@ -100,3 +124,28 @@ func TestHandleOpenAIAudioTranscriptionsParsesFieldsFromStreamedMultipart(t *tes
 		t.Fatalf("expected streamed multipart fields to be parsed, got status=%d body=%s", w.Code, w.Body.String())
 	}
 }
+
+func TestStreamAudioTranscriptionClosesFailedWorker(t *testing.T) {
+	modelID := "AIWizards/Fun-ASR-Nano-2512"
+	engine := &failingStreamASREngine{}
+	s := New(&config.Config{
+		ModelDir:   config.ModelDirForStorage(t.TempDir()),
+		DatasetDir: config.DatasetDirForStorage(t.TempDir()),
+	}, "test")
+	s.asrEngines[modelID] = &managedASREngine{engine: engine}
+
+	req := httptest.NewRequest(http.MethodPost, "/v1/audio/transcriptions", nil)
+	w := httptest.NewRecorder()
+
+	s.streamAudioTranscription(w, req, modelID, engine, api.OpenAIAudioTranscriptionRequest{})
+
+	if !engine.closed {
+		t.Fatal("expected failed ASR stream worker to be closed")
+	}
+	if _, ok := s.asrEngines[modelID]; ok {
+		t.Fatal("expected failed ASR stream worker to be removed from cache")
+	}
+	if !strings.Contains(w.Body.String(), "worker exited") {
+		t.Fatalf("expected SSE error response, got %q", w.Body.String())
+	}
+}

internal/server/handlers_cloud.go

@@ -192,6 +192,9 @@ func (s *Server) getChatEngine(ctx context.Context, modelID, source string, numC
 
 	models, cloudErr := s.listCloudModels(ctx, false)
 	if cloudErr != nil {
+		if providerSource := s.thirdPartyProviderSourceForModel(ctx, modelID); providerSource != "" {
+			return newThirdPartyProviderEngine(providerSource, modelID)
+		}
 		return nil, err
 	}
 	if modelInfoListContains(models, modelID) {
@@ -203,6 +206,9 @@ func (s *Server) getChatEngine(ctx context.Context, modelID, source string, numC
 	}
 	models, cloudErr = s.cloud.RefreshChatModels(ctx)
 	if cloudErr != nil {
+		if providerSource := s.thirdPartyProviderSourceForModel(ctx, modelID); providerSource != "" {
+			return newThirdPartyProviderEngine(providerSource, modelID)
+		}
 		return nil, err
 	}
 	if modelInfoListContains(models, modelID) {

internal/server/handlers_providers_test.go

@@ -1041,6 +1041,59 @@ func TestGetChatEnginePrefersThirdPartyWhenCloudLoginMissing(t *testing.T) {
 	}
 }
 
+func TestGetChatEngineFallsBackToThirdPartyWhenCloudListFails(t *testing.T) {
+	home := t.TempDir()
+	t.Setenv("HOME", home)
+	t.Setenv("USERPROFILE", home)
+	config.ResetProviders()
+	config.ResetProviderModelAllowlist()
+	t.Cleanup(config.ResetProviders)
+	t.Cleanup(config.ResetProviderModelAllowlist)
+
+	providerServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/v1/chat/completions":
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = fmt.Fprint(w, `{"choices":[{"message":{"role":"assistant","content":"provider ok"}}]}`)
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	defer providerServer.Close()
+	cloudServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "cloud unavailable", http.StatusInternalServerError)
+	}))
+	defer cloudServer.Close()
+
+	s := newTestServer(t)
+	s.cfg.OpenCSGAPIKey = "cloud-key"
+	s.cloud = cloud.NewService(cloudServer.URL)
+	if err := config.SaveProviders([]config.ThirdPartyProvider{{
+		ID:      "provider1",
+		Name:    "OpenAI",
+		BaseURL: providerServer.URL + "/v1",
+		APIKey:  "secret",
+		Enabled: true,
+	}}); err != nil {
+		t.Fatalf("save providers: %v", err)
+	}
+	if err := config.ReplaceProviderModelAllowlist("provider1", []string{"deepseek-v4-pro"}); err != nil {
+		t.Fatalf("save provider model allowlist: %v", err)
+	}
+
+	eng, err := s.getChatEngine(context.Background(), "deepseek-v4-pro", "", 0, 0, -1, "", "", "")
+	if err != nil {
+		t.Fatalf("getChatEngine returned error: %v", err)
+	}
+	got, err := eng.Chat(context.Background(), nil, inference.DefaultOptions(), nil)
+	if err != nil {
+		t.Fatalf("chat returned error: %v", err)
+	}
+	if got != "provider ok" {
+		t.Fatalf("chat = %q, want provider ok", got)
+	}
+}
+
 func TestDisabledProviderExcludedFromTagsAndEngine(t *testing.T) {
 	home := t.TempDir()
 	t.Setenv("HOME", home)