Changeset 0.22.4 (#146)

server/health/accumulator.go

@@ -44,6 +44,13 @@ type Store interface {
 
 	// Prune deletes rows older than the given time.
 	Prune(ctx context.Context, before time.Time) (int64, error)
+
+	// ── Tool Health (v0.22.4) ──────────────
+	// UpsertToolWindow merges tool health counters into the hourly bucket.
+	UpsertToolWindow(ctx context.Context, w *models.ToolHealthWindow) error
+
+	// ListAllToolCurrent returns the current-hour bucket for every tool.
+	ListAllToolCurrent(ctx context.Context) ([]models.ToolHealthWindow, error)
 }
 
 // ── In-Memory Bucket ────────────────────────
@@ -54,6 +61,7 @@ type bucket struct {
 	requestCount     int
 	errorCount       int
 	timeoutCount     int
+	rateLimitCount   int
 	totalLatencyMs   int64
 	maxLatencyMs     int
 	lastError        string
@@ -67,23 +75,86 @@ type bucket struct {
 type Accumulator struct {
 	mu      sync.Mutex
 	buckets map[string]*bucket // keyed by provider_config_id
-	store   Store
-	stopCh  chan struct{}
-	wg      sync.WaitGroup
+
+	// Tool health (v0.22.4)
+	toolMu      sync.Mutex
+	toolBuckets map[string]*toolBucket // keyed by tool_name
+
+	store    Store
+	stopCh   chan struct{}
+	wg       sync.WaitGroup
+
+	// Auto-disable (v0.22.4): if set, providers that are "down" for
+	// N consecutive windows get auto-deactivated.
+	autoDisabler         AutoDisabler
+	autoDisableThreshold int // consecutive down windows to trigger (0 = disabled)
+}
+
+// AutoDisabler marks a provider config as inactive. Implemented by the
+// provider store so the accumulator doesn't need to import the store package.
+type AutoDisabler interface {
+	DeactivateProvider(ctx context.Context, configID string) error
+}
+
+// toolBucket holds counters for one tool within one flush interval.
+type toolBucket struct {
+	toolName       string
+	requestCount   int
+	errorCount     int
+	totalLatencyMs int64
+	maxLatencyMs   int
+	lastError      string
+	lastErrorAt    time.Time
 }
 
 // NewAccumulator creates and starts the background flush loop.
 func NewAccumulator(store Store) *Accumulator {
 	a := &Accumulator{
-		buckets: make(map[string]*bucket),
-		store:   store,
-		stopCh:  make(chan struct{}),
+		buckets:     make(map[string]*bucket),
+		toolBuckets: make(map[string]*toolBucket),
+		store:       store,
+		stopCh:      make(chan struct{}),
 	}
 	a.wg.Add(1)
 	go a.flushLoop()
 	return a
 }
 
+// SetAutoDisable configures the auto-disable policy. When a provider
+// has been "down" for N consecutive hourly windows, it is deactivated.
+// N=0 disables the feature.
+func (a *Accumulator) SetAutoDisable(disabler AutoDisabler, threshold int) {
+	a.autoDisabler = disabler
+	a.autoDisableThreshold = threshold
+}
+
+// RecordToolSuccess records a successful tool execution.
+func (a *Accumulator) RecordToolSuccess(toolName string, latencyMs int) {
+	a.toolMu.Lock()
+	defer a.toolMu.Unlock()
+	b := a.getToolBucket(toolName)
+	b.requestCount++
+	b.totalLatencyMs += int64(latencyMs)
+	if latencyMs > b.maxLatencyMs {
+		b.maxLatencyMs = latencyMs
+	}
+}
+
+// RecordToolError records a failed tool execution.
+func (a *Accumulator) RecordToolError(toolName string, latencyMs int, errMsg string) {
+	a.toolMu.Lock()
+	defer a.toolMu.Unlock()
+	b := a.getToolBucket(toolName)
+	b.requestCount++
+	b.errorCount++
+	b.totalLatencyMs += int64(latencyMs)
+	if latencyMs > b.maxLatencyMs {
+		b.maxLatencyMs = latencyMs
+	}
+	b.lastError = errMsg
+	b.lastErrorAt = time.Now().UTC()
+}
+
 // RecordSuccess records a successful provider call.
 func (a *Accumulator) RecordSuccess(providerConfigID string, latencyMs int) {
 	a.mu.Lock()
@@ -127,6 +198,47 @@ func (a *Accumulator) RecordTimeout(providerConfigID string, latencyMs int, errM
 	b.lastErrorAt = time.Now().UTC()
 }
 
+// RecordRateLimit records a rate-limited provider call (HTTP 429).
+// Counted as both an error and a rate limit event.
+func (a *Accumulator) RecordRateLimit(providerConfigID string, latencyMs int, errMsg string) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	b := a.getBucket(providerConfigID)
+	b.requestCount++
+	b.errorCount++
+	b.rateLimitCount++
+	b.totalLatencyMs += int64(latencyMs)
+	if latencyMs > b.maxLatencyMs {
+		b.maxLatencyMs = latencyMs
+	}
+	b.lastError = errMsg
+	b.lastErrorAt = time.Now().UTC()
+}
+
+// IsRateLimitError returns true if the error message indicates an HTTP 429.
+func IsRateLimitError(errMsg string) bool {
+	return len(errMsg) > 0 && (contains(errMsg, "HTTP 429") || contains(errMsg, "rate limit") || contains(errMsg, "too many requests"))
+}
+
+func contains(s, sub string) bool {
+	return len(s) >= len(sub) && (s == sub || len(s) > 0 && containsLower(s, sub))
+}
+
+func containsLower(s, sub string) bool {
+	// Simple case-insensitive contains for short substrings.
+	for i := 0; i <= len(s)-len(sub); i++ {
+		match := true
+		for j := 0; j < len(sub); j++ {
+			sc, tc := s[i+j], sub[j]
+			if sc >= 'A' && sc <= 'Z' { sc += 32 }
+			if tc >= 'A' && tc <= 'Z' { tc += 32 }
+			if sc != tc { match = false; break }
+		}
+		if match { return true }
+	}
+	return false
+}
+
 // Stop halts the flush loop and performs a final flush.
 func (a *Accumulator) Stop() {
 	close(a.stopCh)
@@ -158,6 +270,15 @@ func (a *Accumulator) getBucket(providerConfigID string) *bucket {
 	return b
 }
 
+func (a *Accumulator) getToolBucket(toolName string) *toolBucket {
+	b, ok := a.toolBuckets[toolName]
+	if !ok {
+		b = &toolBucket{toolName: toolName}
+		a.toolBuckets[toolName] = b
+	}
+	return b
+}
+
 func (a *Accumulator) flushLoop() {
 	defer a.wg.Done()
 	ticker := time.NewTicker(FlushInterval)
@@ -167,8 +288,11 @@ func (a *Accumulator) flushLoop() {
 		select {
 		case <-ticker.C:
 			a.flush()
+			a.flushTools()
+			a.checkAutoDisable()
 		case <-a.stopCh:
 			a.flush() // final flush
+			a.flushTools()
 			return
 		}
 	}
@@ -199,6 +323,7 @@ func (a *Accumulator) flush() {
 			RequestCount:     b.requestCount,
 			ErrorCount:       b.errorCount,
 			TimeoutCount:     b.timeoutCount,
+			RateLimitCount:   b.rateLimitCount,
 			TotalLatencyMs:   b.totalLatencyMs,
 			MaxLatencyMs:     b.maxLatencyMs,
 		}
@@ -213,6 +338,89 @@ func (a *Accumulator) flush() {
 	}
 }
 
+// flushTools writes tool health buckets to the database.
+func (a *Accumulator) flushTools() {
+	a.toolMu.Lock()
+	if len(a.toolBuckets) == 0 {
+		a.toolMu.Unlock()
+		return
+	}
+	snap := a.toolBuckets
+	a.toolBuckets = make(map[string]*toolBucket, len(snap))
+	a.toolMu.Unlock()
+
+	windowStart := hourFloor(time.Now().UTC())
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	for _, b := range snap {
+		if b.requestCount == 0 {
+			continue
+		}
+		w := &models.ToolHealthWindow{
+			ToolName:       b.toolName,
+			WindowStart:    windowStart,
+			RequestCount:   b.requestCount,
+			ErrorCount:     b.errorCount,
+			TotalLatencyMs: b.totalLatencyMs,
+			MaxLatencyMs:   b.maxLatencyMs,
+		}
+		if b.lastError != "" {
+			w.LastError = &b.lastError
+			ts := b.lastErrorAt.Format(time.RFC3339)
+			w.LastErrorAt = &ts
+		}
+		if err := a.store.UpsertToolWindow(ctx, w); err != nil {
+			log.Printf("⚠ health: tool flush failed for %s: %v", b.toolName, err)
+		}
+	}
+}
+
+// checkAutoDisable inspects recent health windows and deactivates providers
+// that have been "down" for too many consecutive hours.
+func (a *Accumulator) checkAutoDisable() {
+	if a.autoDisabler == nil || a.autoDisableThreshold <= 0 {
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	// Get current windows for all providers
+	windows, err := a.store.ListAllCurrent(ctx)
+	if err != nil {
+		return
+	}
+
+	for _, w := range windows {
+		status := DeriveStatus(w.ErrorRate(), w.RequestCount)
+		if status != models.StatusDown {
+			continue
+		}
+
+		// Check last N windows — are they all down?
+		recent, err := a.store.ListWindows(ctx, w.ProviderConfigID, a.autoDisableThreshold)
+		if err != nil || len(recent) < a.autoDisableThreshold {
+			continue // not enough data
+		}
+
+		allDown := true
+		for _, rw := range recent {
+			if DeriveStatus(rw.ErrorRate(), rw.RequestCount) != models.StatusDown {
+				allDown = false
+				break
+			}
+		}
+
+		if allDown {
+			log.Printf("⚠ health: auto-disabling provider %s — down for %d consecutive windows", w.ProviderConfigID, a.autoDisableThreshold)
+			if err := a.autoDisabler.DeactivateProvider(ctx, w.ProviderConfigID); err != nil {
+				log.Printf("⚠ health: auto-disable failed for %s: %v", w.ProviderConfigID, err)
+			}
+		}
+	}
+}
+
 // hourFloor truncates a time to the start of its hour.
 func hourFloor(t time.Time) time.Time {
 	return t.Truncate(time.Hour)

server/health/accumulator_test.go

@@ -70,6 +70,14 @@ func (m *mockStore) ListAllCurrent(_ context.Context) ([]models.ProviderHealthWi
 
 func (m *mockStore) Prune(_ context.Context, _ time.Time) (int64, error) { return 0, nil }
 
+func (m *mockStore) UpsertToolWindow(_ context.Context, _ *models.ToolHealthWindow) error {
+	return nil
+}
+
+func (m *mockStore) ListAllToolCurrent(_ context.Context) ([]models.ToolHealthWindow, error) {
+	return nil, nil
+}
+
 // ── Tests ───────────────────────────────────
 
 func TestRecordSuccess(t *testing.T) {