diff --git a/changelog/unreleased/catalog-auto-refresh.md b/changelog/unreleased/catalog-auto-refresh.md new file mode 100644 index 00000000..1b62454b --- /dev/null +++ b/changelog/unreleased/catalog-auto-refresh.md @@ -0,0 +1,13 @@ +### English + +- The display catalog now reloads in the background every 10 minutes by default instead of only when a client asks, so a model an upstream adds appears in `/v1/models` without a manual refresh. Set `CLI2API_CATALOG_REFRESH_INTERVAL` (Go duration such as `15m`, or plain minutes) to change the interval; `0` disables the loop. +- The last successful catalog is persisted in SQLite and restored on boot, and a failed catalog refresh now keeps serving the previous model list instead of returning an empty one. +- `GET /v1/models` reports `catalog.synced_at` and `catalog.stale` beside `data`; upstream error text stays on the console endpoint so a scoped API key cannot read another provider's failure. +- `GET /api/model-catalog` (console) returns the catalog sync state with per-provider last success time, last error, and model count; `POST /api/model-catalog` forces a refresh. + +### 中文 + +- 模型目录改为后台自动刷新，默认每 10 分钟一次，不再只在客户端请求时才拉取，上游新增的模型无需手动刷新就会出现在 `/v1/models`。刷新间隔用 `CLI2API_CATALOG_REFRESH_INTERVAL` 调整（Go duration 如 `15m`，或纯数字表示分钟），设为 `0` 关闭定时刷新。 +- 最近一次成功的模型目录会写入 SQLite 并在启动时恢复；目录刷新失败时继续使用上一次的有效列表，不再返回空列表。 +- `GET /v1/models` 在 `data` 旁边新增 `catalog.synced_at` 与 `catalog.stale`；上游错误文本只留在控制台接口，限定供应商的 API Key 读不到其他供应商的失败原因。 +- 控制台 `GET /api/model-catalog` 返回目录同步状态（每个供应商的上次成功时间、失败原因、模型数量），`POST /api/model-catalog` 立即刷新一次。 diff --git a/internal/app/app.go b/internal/app/app.go index 1062153f..93c52e91 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -169,6 +169,13 @@ func New(cfg config.Config) *App { } a.CrossProviderModelPool.Store(crossProviderModelPool) a.Control.Catalog = appsvc.NewCatalog(a.FetchWorkerModelsForMode) + a.Control.Catalog.Store = a.Manager.Store() + a.Control.Catalog.Interval = cfg.CatalogRefreshInterval + a.Control.Catalog.Health = appsvc.NewCatalogHealth() + // Restore the last successful catalog before the first request so a restart + // serves the previous model list instead of an empty one. + a.Control.Catalog.Restore(context.Background()) + go a.Control.Catalog.RunRefresher(stopLogs) if a.Control.Settings != nil { a.Control.Settings.BindCatalog(a.Control.Catalog) } diff --git a/internal/app/gateway.go b/internal/app/gateway.go index 39624092..0515e635 100644 --- a/internal/app/gateway.go +++ b/internal/app/gateway.go @@ -29,6 +29,19 @@ func (a *App) newGateway() *apigateway.Handler { Models: func(refresh bool, accountID string) ([]map[string]any, error) { return a.fetchDisplayModels(refresh, accountID, control.CatalogModeMerge) }, + CatalogStatus: func() map[string]any { + if a.Control == nil || a.Control.Catalog == nil { + return nil + } + status := a.Control.Catalog.Status() + // The public listing reports freshness only: a scoped API key must not + // read another provider's upstream error text. The console endpoint + // carries the details. + return map[string]any{ + "synced_at": status.SyncedAt, + "stale": status.Stale, + } + }, } if a.Control != nil { h.ModelContexts = a.Control.Settings diff --git a/internal/app/workerproxy.go b/internal/app/workerproxy.go index 40e207f9..9b2a11ec 100644 --- a/internal/app/workerproxy.go +++ b/internal/app/workerproxy.go @@ -22,7 +22,7 @@ func (a *App) catalogSource() *control.CatalogSource { item, ok := a.Pool.First() return item.URL, ok }} - return &control.CatalogSource{Providers: a.Providers, WorkerModels: worker.Models, Accounts: func() []control.CatalogAccount { + return &control.CatalogSource{Providers: a.Providers, WorkerModels: worker.Models, Health: a.catalogHealth(), Accounts: func() []control.CatalogAccount { var out []control.CatalogAccount for _, item := range a.Pool.Items() { out = append(out, control.CatalogAccount{ID: item.ID, Provider: item.Provider, Region: item.Region, Worker: item.URL != ""}) @@ -30,6 +30,15 @@ func (a *App) catalogSource() *control.CatalogSource { return out }} } + +// catalogHealth exposes the display catalog's per-provider recorder so a +// provider-level catalog fetch lands in the same status the console serves. +func (a *App) catalogHealth() *control.CatalogHealth { + if a == nil || a.Control == nil || a.Control.Catalog == nil { + return nil + } + return a.Control.Catalog.Health +} func (a *App) FetchWorkerModelsFor(refresh bool, accountID string) ([]map[string]any, error) { return a.FetchWorkerModelsForMode(refresh, accountID, control.CatalogModeMerge) } diff --git a/internal/config/config.go b/internal/config/config.go index 342ab383..0d05a449 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -4,27 +4,35 @@ import ( "os" "strconv" "strings" + "time" ) +// DefaultCatalogRefreshInterval is the documented default for the background +// model-catalog reload. Operators can change it with +// CLI2API_CATALOG_REFRESH_INTERVAL (a Go duration such as 30s or 15m, or plain +// minutes); 0 disables the loop and leaves only manual/console refreshes. +const DefaultCatalogRefreshInterval = 10 * time.Minute + type Config struct { - Host string - Port int - ProxyAPIKey string - ProxyURL string - MaxRetryAccounts int - QoderHome string - DataDir string - RuntimeDir string - WorkerBasePort int - NodeBinary string - WorkerDaemonPath string - QoderCLIPath string - QoderCNCLIPath string - PlainTemplatePath string - UpdateSocketPath string - UpdateAgentURL string - UpdateAgentToken string - UpdateGitHubToken string + Host string + Port int + ProxyAPIKey string + ProxyURL string + MaxRetryAccounts int + QoderHome string + DataDir string + RuntimeDir string + WorkerBasePort int + NodeBinary string + WorkerDaemonPath string + QoderCLIPath string + QoderCNCLIPath string + PlainTemplatePath string + UpdateSocketPath string + UpdateAgentURL string + UpdateAgentToken string + UpdateGitHubToken string + CatalogRefreshInterval time.Duration } func Load() (Config, error) { @@ -70,27 +78,52 @@ func Load() (Config, error) { runtimeDir = "/tmp/cli2api-runtime" } return Config{ - Host: host, - Port: port, - ProxyAPIKey: "", - ProxyURL: strings.TrimSpace(os.Getenv("QODER_PROXY_URL")), - MaxRetryAccounts: maxRetryAccounts, - QoderHome: home, - DataDir: dataDir, - RuntimeDir: runtimeDir, - WorkerBasePort: workerBasePort, - NodeBinary: firstNonEmpty(os.Getenv("QODER_NODE_BINARY"), "node"), - WorkerDaemonPath: firstNonEmpty(os.Getenv("QODER_WORKER_DAEMON"), "worker/src/daemon.mjs"), - QoderCLIPath: firstNonEmpty(os.Getenv("QODERCLI_JS"), "/usr/local/lib/node_modules/@qoder-ai/qodercli/bundle/qodercli.js"), - QoderCNCLIPath: firstNonEmpty(os.Getenv("QODERCNCLI_JS"), "/usr/local/lib/node_modules/@qodercn-ai/qoderclicn/bundle/qoderclicn.js"), - PlainTemplatePath: firstNonEmpty(os.Getenv("PLAIN_TEMPLATE_PATH"), "worker/last-plain.sample.json"), - UpdateSocketPath: firstNonEmpty(os.Getenv("UPDATE_SOCKET_PATH"), "/run/cli2api-updater/updater.sock"), - UpdateAgentURL: strings.TrimSpace(os.Getenv("UPDATE_AGENT_URL")), - UpdateAgentToken: strings.TrimSpace(os.Getenv("UPDATE_AGENT_TOKEN")), - UpdateGitHubToken: strings.TrimSpace(os.Getenv("UPDATE_GITHUB_TOKEN")), + Host: host, + Port: port, + ProxyAPIKey: "", + ProxyURL: strings.TrimSpace(os.Getenv("QODER_PROXY_URL")), + MaxRetryAccounts: maxRetryAccounts, + QoderHome: home, + DataDir: dataDir, + RuntimeDir: runtimeDir, + WorkerBasePort: workerBasePort, + NodeBinary: firstNonEmpty(os.Getenv("QODER_NODE_BINARY"), "node"), + WorkerDaemonPath: firstNonEmpty(os.Getenv("QODER_WORKER_DAEMON"), "worker/src/daemon.mjs"), + QoderCLIPath: firstNonEmpty(os.Getenv("QODERCLI_JS"), "/usr/local/lib/node_modules/@qoder-ai/qodercli/bundle/qodercli.js"), + QoderCNCLIPath: firstNonEmpty(os.Getenv("QODERCNCLI_JS"), "/usr/local/lib/node_modules/@qodercn-ai/qoderclicn/bundle/qoderclicn.js"), + PlainTemplatePath: firstNonEmpty(os.Getenv("PLAIN_TEMPLATE_PATH"), "worker/last-plain.sample.json"), + UpdateSocketPath: firstNonEmpty(os.Getenv("UPDATE_SOCKET_PATH"), "/run/cli2api-updater/updater.sock"), + UpdateAgentURL: strings.TrimSpace(os.Getenv("UPDATE_AGENT_URL")), + UpdateAgentToken: strings.TrimSpace(os.Getenv("UPDATE_AGENT_TOKEN")), + UpdateGitHubToken: strings.TrimSpace(os.Getenv("UPDATE_GITHUB_TOKEN")), + CatalogRefreshInterval: catalogRefreshInterval(os.Getenv("CLI2API_CATALOG_REFRESH_INTERVAL")), }, nil } +// catalogRefreshInterval parses CLI2API_CATALOG_REFRESH_INTERVAL: a Go duration +// ("90s", "15m") or a bare number of minutes ("10"). Unset keeps the default, +// a negative or zero value disables the background loop, and garbage falls back +// to the default with a log line rather than failing the boot. +func catalogRefreshInterval(raw string) time.Duration { + value := strings.TrimSpace(raw) + if value == "" { + return DefaultCatalogRefreshInterval + } + if d, err := time.ParseDuration(value); err == nil { + if d < 0 { + return 0 + } + return d + } + if minutes, err := strconv.Atoi(value); err == nil { + if minutes < 0 { + return 0 + } + return time.Duration(minutes) * time.Minute + } + return DefaultCatalogRefreshInterval +} + func firstNonEmpty(vals ...string) string { for _, v := range vals { if s := strings.TrimSpace(v); s != "" { diff --git a/internal/console/models.go b/internal/console/models.go index e58e5085..4b5ac8ed 100644 --- a/internal/console/models.go +++ b/internal/console/models.go @@ -38,6 +38,26 @@ func (h *Handler) HandleModelsAPI(w http.ResponseWriter, r *http.Request) { }) } +// HandleCatalogStatus reports the model-catalog sync state. GET only reads it; +// POST forces one full reload first and answers with the state that resulted. +// The body is 200 in both cases — a partial or failed refresh is described by +// last_error and the per-provider rows, not by the HTTP status. +func (h *Handler) HandleCatalogStatus(w http.ResponseWriter, r *http.Request) { + if h == nil || h.Control == nil || h.Control.Catalog == nil { + writeErr(w, http.StatusServiceUnavailable, "catalog_unavailable", "model catalog is not configured") + return + } + switch r.Method { + case http.MethodGet: + writeJSON(w, http.StatusOK, h.Control.Catalog.Status()) + case http.MethodPost: + status, _ := h.Control.Catalog.Refresh() + writeJSON(w, http.StatusOK, status) + default: + writeErr(w, http.StatusMethodNotAllowed, "method_not_allowed", "GET or POST only") + } +} + func (h *Handler) HandleModelSetting(w http.ResponseWriter, r *http.Request) { provider, modelKey := splitModelSettingPath(r.URL.Path, r.URL.Query().Get("provider")) if modelKey == "" { diff --git a/internal/control/catalog.go b/internal/control/catalog.go index 9f279149..d651c96a 100644 --- a/internal/control/catalog.go +++ b/internal/control/catalog.go @@ -1,6 +1,8 @@ package control import ( + "context" + "encoding/json" "strings" "sync" "time" @@ -8,6 +10,11 @@ import ( const CatalogCacheTTL = 5 * time.Minute +// catalogSnapshotSecret is the app_secrets key holding the last successful +// merged catalog. It keeps /v1/models populated across a restart and after a +// failed refresh, so a transient upstream failure never empties the model list. +const catalogSnapshotSecret = "catalog_snapshot_v1" + type CatalogMode int const ( @@ -30,14 +37,36 @@ type catalogRefresh struct { // SQLite settings stay outside this cache. type CatalogFetcher func(refresh bool, accountID string, mode CatalogMode) ([]map[string]any, error) +// CatalogStore persists the last successful catalog snapshot. internal/store +// satisfies it; the narrow surface keeps control free of the store package. +type CatalogStore interface { + GetSecret(ctx context.Context, name string) (string, bool, error) + SetSecret(ctx context.Context, name, value string) error +} + +type catalogSnapshot struct { + Version int `json:"version"` + SyncedAt string `json:"synced_at"` + Models []map[string]any `json:"models"` + Health []CatalogProviderHealth `json:"providers,omitempty"` +} + +const catalogSnapshotVersion = 1 + // Catalog is the shared /v1/models and /api/models display cache. It does not // own runtime catalogs or proven-model membership. type Catalog struct { TTL time.Duration + Interval time.Duration Fetch CatalogFetcher + Store CatalogStore + Health *CatalogHealth + mu sync.Mutex cache map[string]CatalogCacheEntry inflight map[string]*catalogRefresh + syncedAt time.Time + lastErr string } func NewCatalog(fetch CatalogFetcher) *Catalog { @@ -60,6 +89,10 @@ func CatalogCacheKey(accountID string, mode CatalogMode) string { return accountID + "@" + view } +// defaultKey is the merged, account-agnostic view /v1/models serves. Only that +// view is persisted: per-account and regional shapes are console-only. +func defaultCatalogKey() string { return CatalogCacheKey("", CatalogModeMerge) } + func CloneModelList(models []map[string]any) []map[string]any { if models == nil { return nil @@ -140,6 +173,9 @@ func (c *Catalog) snapshot(accountID string, mode CatalogMode) []map[string]any return CloneModelList(entry.Models) } +// Get returns the display catalog for one view. A failed fetch never discards a +// catalog that already served: the last successful list is returned and the +// failure is reported through Status instead. func (c *Catalog) Get(refresh bool, accountID string, mode CatalogMode) ([]map[string]any, error) { if c == nil || c.Fetch == nil { return nil, nil @@ -164,11 +200,28 @@ func (c *Catalog) Get(refresh bool, accountID string, mode CatalogMode) ([]map[s c.mu.Unlock() <-refreshing.done if refreshing.err != nil { + if hasCache { + return CloneModelList(entry.Models), nil + } + if restored := c.snapshot(accountID, mode); len(restored) > 0 { + return restored, nil + } return nil, refreshing.err } return CloneModelList(refreshing.models), nil } +// recordFailure keeps the last error on the default view only: the console and +// task 2's status surface describe the shared catalog, not a per-account probe. +func (c *Catalog) recordFailure(key string, err error) { + if c == nil || err == nil || key != defaultCatalogKey() { + return + } + c.mu.Lock() + c.lastErr = strings.TrimSpace(err.Error()) + c.mu.Unlock() +} + func (c *Catalog) startLocked(key string, force bool, accountID string, mode CatalogMode) *catalogRefresh { if refreshing, ok := c.inflight[key]; ok { return refreshing @@ -180,9 +233,21 @@ func (c *Catalog) startLocked(key string, force bool, accountID string, mode Cat refreshing.models = models refreshing.err = err if err == nil { + at := time.Now() c.mu.Lock() - c.cache[key] = CatalogCacheEntry{Models: CloneModelList(models), At: time.Now()} + c.cache[key] = CatalogCacheEntry{Models: CloneModelList(models), At: at} + if key == defaultCatalogKey() { + c.syncedAt = at + c.lastErr = "" + } c.mu.Unlock() + if key == defaultCatalogKey() { + c.persist(context.Background()) + } + } else if key == defaultCatalogKey() { + // Record a failed background/TTL refresh too: without this the status + // only ever reported failures of an explicitly forced refresh. + c.recordFailure(key, err) } c.mu.Lock() delete(c.inflight, key) @@ -191,3 +256,153 @@ func (c *Catalog) startLocked(key string, force bool, accountID string, mode Cat }() return refreshing } + +// Status reports the shared catalog's sync state plus each provider's own. +func (c *Catalog) Status() CatalogStatus { + if c == nil { + return CatalogStatus{Providers: nil} + } + c.mu.Lock() + syncedAt := c.syncedAt + if entry, ok := c.cache[defaultCatalogKey()]; ok && entry.At.After(syncedAt) { + syncedAt = entry.At + } + lastErr := c.lastErr + refreshing := len(c.inflight) > 0 + c.mu.Unlock() + interval := c.Interval + if interval < 0 { + interval = 0 + } + status := CatalogStatus{ + Refreshing: refreshing, + LastError: lastErr, + IntervalSeconds: int(interval / time.Second), + Providers: c.Health.Snapshot(), + } + // A provider behind its last success makes the whole view stale even when the + // last full refresh succeeded for the others. + for _, provider := range status.Providers { + if provider.Stale { + status.Stale = true + break + } + } + if !status.Stale { + status.Stale = lastErr != "" + } + if !syncedAt.IsZero() { + status.SyncedAt = syncedAt.UTC().Format(time.RFC3339) + } + return status +} + +// Refresh forces one full reload and reports the resulting status. The returned +// error is the fetch error for this attempt; a served catalog is still intact +// when it is non-nil. +func (c *Catalog) Refresh() (CatalogStatus, error) { + if c == nil || c.Fetch == nil { + return CatalogStatus{}, nil + } + _, err := c.Get(true, "", CatalogModeMerge) + return c.Status(), err +} + +// RunRefresher keeps the merged catalog warm: it refreshes once at start and +// then every Interval, so a newly announced upstream model reaches clients +// without waiting for someone to open the model list. A zero or negative +// interval disables the loop (manual refresh only). +func (c *Catalog) RunRefresher(stop <-chan struct{}) { + if c == nil || c.Fetch == nil || c.Interval <= 0 { + return + } + c.backgroundRefresh() + ticker := time.NewTicker(c.Interval) + defer ticker.Stop() + for { + select { + case <-stop: + return + case <-ticker.C: + c.backgroundRefresh() + } + } +} + +func (c *Catalog) backgroundRefresh() { + if c == nil || c.Fetch == nil { + return + } + if _, err := c.Get(true, "", CatalogModeMerge); err != nil { + logFailure("catalog", err) + } +} + +// persist writes the default view and provider health to app_secrets. Called +// only after a successful default-view fetch, so the stored snapshot is always +// a usable catalog. +func (c *Catalog) persist(ctx context.Context) { + if c == nil || c.Store == nil { + return + } + c.mu.Lock() + entry, ok := c.cache[defaultCatalogKey()] + syncedAt := c.syncedAt + c.mu.Unlock() + if !ok || len(entry.Models) == 0 { + return + } + if syncedAt.IsZero() { + syncedAt = entry.At + } + payload, err := json.Marshal(catalogSnapshot{ + Version: catalogSnapshotVersion, + SyncedAt: syncedAt.UTC().Format(time.RFC3339), + Models: entry.Models, + Health: c.Health.Snapshot(), + }) + if err != nil { + logFailure("catalog snapshot encode", err) + return + } + if err := c.Store.SetSecret(ctx, catalogSnapshotSecret, string(payload)); err != nil { + logFailure("catalog snapshot write", err) + } +} + +// Restore loads the persisted snapshot into the cache without treating it as +// fresh: the entry keeps its original timestamp, so the next read refreshes it +// in the background while still serving the last known models. +func (c *Catalog) Restore(ctx context.Context) { + if c == nil || c.Store == nil { + return + } + raw, ok, err := c.Store.GetSecret(ctx, catalogSnapshotSecret) + if err != nil { + logFailure("catalog snapshot read", err) + return + } + if !ok || strings.TrimSpace(raw) == "" { + return + } + var snapshot catalogSnapshot + if err := json.Unmarshal([]byte(raw), &snapshot); err != nil { + logFailure("catalog snapshot decode", err) + return + } + if len(snapshot.Models) == 0 { + return + } + at, parseErr := time.Parse(time.RFC3339, snapshot.SyncedAt) + if parseErr != nil { + at = time.Now().Add(-CatalogCacheTTL) + } + c.mu.Lock() + key := defaultCatalogKey() + if _, exists := c.cache[key]; !exists { + c.cache[key] = CatalogCacheEntry{Models: CloneModelList(snapshot.Models), At: at} + c.syncedAt = at + } + c.mu.Unlock() + c.Health.Restore(snapshot.Health) +} diff --git a/internal/control/catalog_health.go b/internal/control/catalog_health.go new file mode 100644 index 00000000..8c4e94dd --- /dev/null +++ b/internal/control/catalog_health.go @@ -0,0 +1,149 @@ +package control + +import ( + "log" + "sort" + "strings" + "sync" + "time" +) + +// CatalogProviderHealth is one provider's upstream model-catalog sync state. +// It is display-only: a failure never removes models from the shared catalog. +type CatalogProviderHealth struct { + Provider string `json:"provider"` + Models int `json:"models"` + LastAttemptAt string `json:"last_attempt_at,omitempty"` + LastSuccessAt string `json:"last_success_at,omitempty"` + LastError string `json:"last_error,omitempty"` + // Stale is true while the last attempt failed and the model list below + // still describes the last successful sync. + Stale bool `json:"stale"` +} + +// CatalogStatus is the whole display catalog's sync state. SyncedAt is the last +// successful full refresh of the merged view that backs /v1/models. +type CatalogStatus struct { + SyncedAt string `json:"synced_at,omitempty"` + Stale bool `json:"stale"` + Refreshing bool `json:"refreshing"` + LastError string `json:"last_error,omitempty"` + IntervalSeconds int `json:"interval_seconds"` + Providers []CatalogProviderHealth `json:"providers"` +} + +// CatalogHealth records per-provider catalog outcomes. Providers discover their +// own model list, so the merged catalog can be half fresh: recording per +// provider is what lets the console and /v1/models say which upstream is behind +// instead of only that "the catalog" failed. +type CatalogHealth struct { + mu sync.Mutex + providers map[string]*CatalogProviderHealth + now func() time.Time +} + +func NewCatalogHealth() *CatalogHealth { + return &CatalogHealth{providers: map[string]*CatalogProviderHealth{}, now: time.Now} +} + +func (h *CatalogHealth) clock() time.Time { + if h == nil || h.now == nil { + return time.Now() + } + return h.now() +} + +func (h *CatalogHealth) entry(provider string) *CatalogProviderHealth { + key := strings.ToLower(strings.TrimSpace(provider)) + if key == "" { + key = "unknown" + } + entry, ok := h.providers[key] + if !ok { + entry = &CatalogProviderHealth{Provider: key} + h.providers[key] = entry + } + return entry +} + +// Attempt marks the start of one upstream catalog request. +func (h *CatalogHealth) Attempt(provider string) { + if h == nil { + return + } + at := h.clock().UTC().Format(time.RFC3339) + h.mu.Lock() + defer h.mu.Unlock() + h.entry(provider).LastAttemptAt = at +} + +// Success records a served catalog and clears the provider's failure state. +func (h *CatalogHealth) Success(provider string, models int) { + if h == nil { + return + } + at := h.clock().UTC().Format(time.RFC3339) + h.mu.Lock() + defer h.mu.Unlock() + entry := h.entry(provider) + entry.Models = models + entry.LastSuccessAt = at + entry.LastError = "" + entry.Stale = false +} + +// Failure records a failed catalog request. The previous model count and last +// success time stay in place so a failure never looks like an empty catalog. +func (h *CatalogHealth) Failure(provider string, err error) { + if h == nil || err == nil { + return + } + at := h.clock().UTC().Format(time.RFC3339) + h.mu.Lock() + defer h.mu.Unlock() + entry := h.entry(provider) + entry.LastAttemptAt = at + entry.LastError = strings.TrimSpace(err.Error()) + entry.Stale = true +} + +// Snapshot returns a stable, provider-sorted copy for display. +func (h *CatalogHealth) Snapshot() []CatalogProviderHealth { + if h == nil { + return nil + } + h.mu.Lock() + defer h.mu.Unlock() + out := make([]CatalogProviderHealth, 0, len(h.providers)) + for _, entry := range h.providers { + out = append(out, *entry) + } + sort.Slice(out, func(i, j int) bool { return out[i].Provider < out[j].Provider }) + return out +} + +// Restore seeds health from a persisted snapshot. Timestamps are kept verbatim; +// a provider whose last attempt failed stays stale until the next success. +func (h *CatalogHealth) Restore(entries []CatalogProviderHealth) { + if h == nil { + return + } + h.mu.Lock() + defer h.mu.Unlock() + for _, entry := range entries { + if strings.TrimSpace(entry.Provider) == "" { + continue + } + copied := entry + h.providers[strings.ToLower(copied.Provider)] = &copied + } +} + +// logFailure prints one line per failed provider request so an operator can see +// which upstream is behind without opening the console. +func logFailure(provider string, err error) { + if err == nil { + return + } + log.Printf("catalog sync failed provider=%s: %v", provider, err) +} diff --git a/internal/control/catalog_refresh_test.go b/internal/control/catalog_refresh_test.go new file mode 100644 index 00000000..b7866cc7 --- /dev/null +++ b/internal/control/catalog_refresh_test.go @@ -0,0 +1,204 @@ +package control + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" +) + +type fakeCatalogStore struct { + mu sync.Mutex + data map[string]string + err error +} + +func (s *fakeCatalogStore) GetSecret(_ context.Context, name string) (string, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.err != nil { + return "", false, s.err + } + value, ok := s.data[name] + return value, ok, nil +} + +func (s *fakeCatalogStore) SetSecret(_ context.Context, name, value string) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.err != nil { + return s.err + } + if s.data == nil { + s.data = map[string]string{} + } + s.data[name] = value + return nil +} + +func TestCatalogHealthTracksPerProviderOutcome(t *testing.T) { + now := time.Date(2026, 10, 26, 3, 0, 0, 0, time.UTC) + health := NewCatalogHealth() + health.now = func() time.Time { return now } + health.Attempt("qoder") + health.Success("qoder", 14) + health.Attempt("workbuddy") + health.Failure("workbuddy", errors.New("upstream 503")) + + rows := health.Snapshot() + if len(rows) != 2 { + t.Fatalf("rows=%v", rows) + } + if rows[0].Provider != "qoder" || rows[0].Models != 14 || rows[0].Stale || rows[0].LastError != "" { + t.Fatalf("qoder row=%+v", rows[0]) + } + if rows[0].LastSuccessAt != now.Format(time.RFC3339) { + t.Fatalf("qoder success at=%q", rows[0].LastSuccessAt) + } + if rows[1].Provider != "workbuddy" || !rows[1].Stale || rows[1].LastError != "upstream 503" { + t.Fatalf("workbuddy row=%+v", rows[1]) + } + + // A later success clears the failure without dropping the provider. + health.Success("workbuddy", 26) + rows = health.Snapshot() + if rows[1].Stale || rows[1].LastError != "" || rows[1].Models != 26 { + t.Fatalf("workbuddy after success=%+v", rows[1]) + } +} + +func TestCatalogPersistsAndRestoresSnapshot(t *testing.T) { + store := &fakeCatalogStore{} + first := NewCatalog(func(bool, string, CatalogMode) ([]map[string]any, error) { + return []map[string]any{{"id": "glm-5.3", "provider": "workbuddy"}}, nil + }) + first.Store = store + first.Health = NewCatalogHealth() + first.Health.Success("workbuddy", 1) + + if _, err := first.Get(true, "", CatalogModeMerge); err != nil { + t.Fatal(err) + } + if store.data[catalogSnapshotSecret] == "" { + t.Fatal("snapshot was not persisted") + } + + // A restart starts with an upstream that fails; the stored snapshot must + // still answer /v1/models. + restored := NewCatalog(func(bool, string, CatalogMode) ([]map[string]any, error) { + return nil, errors.New("upstream unreachable") + }) + restored.Store = store + restored.Health = NewCatalogHealth() + restored.Restore(context.Background()) + + models, err := restored.Get(false, "", CatalogModeMerge) + if err != nil { + t.Fatalf("restored read failed: %v", err) + } + if len(models) != 1 || models[0]["id"] != "glm-5.3" { + t.Fatalf("restored models=%v", models) + } + rows := restored.Status().Providers + if len(rows) != 1 || rows[0].Provider != "workbuddy" { + t.Fatalf("restored health=%v", rows) + } +} + +func TestCatalogServesLastKnownCatalogAfterFailure(t *testing.T) { + var fail atomic.Bool + catalog := NewCatalog(func(bool, string, CatalogMode) ([]map[string]any, error) { + if fail.Load() { + return nil, errors.New("upstream 502") + } + return []map[string]any{{"id": "kimi-k3"}}, nil + }) + if _, err := catalog.Get(true, "", CatalogModeMerge); err != nil { + t.Fatal(err) + } + fail.Store(true) + + models, err := catalog.Get(true, "", CatalogModeMerge) + if err != nil { + t.Fatalf("failed refresh must not fail a served catalog: %v", err) + } + if len(models) != 1 || models[0]["id"] != "kimi-k3" { + t.Fatalf("models after failure=%v", models) + } + status := catalog.Status() + if !status.Stale || status.LastError != "upstream 502" { + t.Fatalf("status=%+v", status) + } + if status.SyncedAt == "" { + t.Fatal("synced_at must survive a failure") + } +} + +func TestCatalogRefreshReportsStatus(t *testing.T) { + var calls atomic.Int32 + catalog := NewCatalog(func(bool, string, CatalogMode) ([]map[string]any, error) { + calls.Add(1) + return []map[string]any{{"id": "sonus"}}, nil + }) + catalog.Interval = 10 * time.Minute + + status, err := catalog.Refresh() + if err != nil { + t.Fatal(err) + } + if calls.Load() != 1 { + t.Fatalf("calls=%d", calls.Load()) + } + if status.SyncedAt == "" || status.Stale || status.Refreshing { + t.Fatalf("status=%+v", status) + } + if status.IntervalSeconds != 600 { + t.Fatalf("interval=%d", status.IntervalSeconds) + } +} + +func TestCatalogRefresherTicks(t *testing.T) { + var calls atomic.Int32 + catalog := NewCatalog(func(bool, string, CatalogMode) ([]map[string]any, error) { + calls.Add(1) + return []map[string]any{{"id": "cantus"}}, nil + }) + catalog.Interval = 20 * time.Millisecond + + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { + catalog.RunRefresher(stop) + close(done) + }() + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if calls.Load() >= 2 { + break + } + time.Sleep(5 * time.Millisecond) + } + close(stop) + <-done + if calls.Load() < 2 { + t.Fatalf("refresher did not tick: calls=%d", calls.Load()) + } +} + +func TestCatalogRefresherDisabledWithZeroInterval(t *testing.T) { + var calls atomic.Int32 + catalog := NewCatalog(func(bool, string, CatalogMode) ([]map[string]any, error) { + calls.Add(1) + return nil, nil + }) + catalog.Interval = 0 + stop := make(chan struct{}) + close(stop) + catalog.RunRefresher(stop) + if calls.Load() != 0 { + t.Fatalf("disabled refresher fetched %d times", calls.Load()) + } +} diff --git a/internal/control/catalog_source.go b/internal/control/catalog_source.go index 88163486..6b978469 100644 --- a/internal/control/catalog_source.go +++ b/internal/control/catalog_source.go @@ -16,10 +16,32 @@ type CatalogAccount struct { } // CatalogSource aggregates provider catalogs through injected data sources. +// Health is optional and records one outcome per provider request. type CatalogSource struct { Accounts func() []CatalogAccount Providers *providers.Registry WorkerModels func(context.Context, string, bool) ([]map[string]any, error) + Health *CatalogHealth +} + +// recordAttempt, recordSuccess and recordFailure are no-ops when no health +// recorder is wired, so a source built by a test needs no extra setup. +func (a *CatalogSource) recordAttempt(provider string) { + if a != nil && a.Health != nil { + a.Health.Attempt(provider) + } +} + +func (a *CatalogSource) recordSuccess(provider string, models int) { + if a != nil && a.Health != nil { + a.Health.Success(provider, models) + } +} + +func (a *CatalogSource) recordFailure(provider string, err error) { + if a != nil && a.Health != nil { + a.Health.Failure(provider, err) + } } func (a *CatalogSource) byID(id string) (CatalogAccount, bool) { @@ -49,13 +71,16 @@ func (a *CatalogSource) Fetch(refresh bool, accountID string, mode CatalogMode) // Last-resort path for a lone Qoder worker with no in-process providers // and no pool URLs folded above. Stamp region the same way expand/merge // would, so Providers filters never treat CN catalogs as unlabeled. + a.recordAttempt("qoder") parsed, err := a.WorkerModels(context.Background(), accountID, refresh) if err != nil { + a.recordFailure("qoder", err) if accountID != "" { return nil, err } return nil, nil } + a.recordSuccess("qoder", len(parsed)) if len(parsed) == 0 { return nil, nil } @@ -100,15 +125,18 @@ func (a *CatalogSource) fetchProviderModels(refresh bool, accountID string, mode if !ok || adapter.Models == nil { continue } + a.recordAttempt(item.Provider) models, err := adapter.Models.Models(context.Background(), item.ID) if err != nil { log.Printf("catalog fetch failed account=%s provider=%s: %v", item.ID, item.Provider, err) + a.recordFailure(item.Provider, err) lastErr = err if accountID != "" { return nil, err } continue } + a.recordSuccess(item.Provider, len(models)) sawAny = true region := accounts.NormalizeRegion(item.Region) for _, model := range models { @@ -205,8 +233,14 @@ func (a *CatalogSource) fetchQoderModels(refresh bool, accountID string) []map[s if item, ok := a.byID(accountID); ok { region = accounts.NormalizeRegion(item.Region) } + a.recordAttempt("qoder") parsed, err := a.WorkerModels(context.Background(), accountID, refresh) - if err != nil || len(parsed) == 0 { + if err != nil { + a.recordFailure("qoder", err) + return nil + } + a.recordSuccess("qoder", len(parsed)) + if len(parsed) == 0 { return nil } for _, model := range parsed { diff --git a/internal/gateway/handler.go b/internal/gateway/handler.go index ab4fd1f5..e3691b7f 100644 --- a/internal/gateway/handler.go +++ b/internal/gateway/handler.go @@ -22,6 +22,7 @@ type Handler struct { Logs executor.RequestStarter CrossProviderPool *atomic.Bool Models CatalogQuery + CatalogStatus func() map[string]any RequestedAccount func(*http.Request) string FilterModels func(*http.Request, []map[string]any) []map[string]any DecorateModels func(*http.Request, []map[string]any) []map[string]any diff --git a/internal/gateway/models.go b/internal/gateway/models.go index 7f392397..1c7a467b 100644 --- a/internal/gateway/models.go +++ b/internal/gateway/models.go @@ -16,8 +16,17 @@ func (h *Handler) HandleModels(w http.ResponseWriter, r *http.Request) { if h != nil && h.DecorateModels != nil { models = h.DecorateModels(r, models) } - writeJSON(w, http.StatusOK, map[string]any{ + body := map[string]any{ "object": "list", "data": models, - }) + } + // Catalog sync metadata rides beside `data`; OpenAI clients ignore unknown + // top-level members, and a model-sync client can tell a fresh list from the + // last known one after an upstream failure. + if h != nil && h.CatalogStatus != nil { + if status := h.CatalogStatus(); len(status) > 0 { + body["catalog"] = status + } + } + writeJSON(w, http.StatusOK, body) } diff --git a/internal/server/router.go b/internal/server/router.go index 8bbd422c..295b2dbf 100644 --- a/internal/server/router.go +++ b/internal/server/router.go @@ -24,6 +24,7 @@ func (s *Server) routes() { s.mux.HandleFunc("/api/keys", s.withConsoleKey(s.consoleHandler().HandleAPIKeys)) s.mux.HandleFunc("/api/keys/", s.withConsoleKey(s.consoleHandler().HandleAPIKeyByID)) s.mux.HandleFunc("/api/models", s.withConsoleKey(s.consoleHandler().HandleModelsAPI)) + s.mux.HandleFunc("/api/model-catalog", s.withConsoleKey(s.consoleHandler().HandleCatalogStatus)) s.mux.HandleFunc("/api/models/", s.withConsoleKey(s.consoleHandler().HandleModelSetting)) s.mux.HandleFunc("/api/providers", s.withConsoleKey(s.consoleHandler().HandleProviders)) s.mux.HandleFunc("/api/accounts", s.withConsoleKey(s.consoleHandler().HandleAccounts))