feat: complete commercial developer workflowspublish-commercial-control-plane

Add tenant-safe usage observability, prepaid billing controls, API key lifecycle management, Embeddings metering, configurable billing alerts, and resilient provider health propagation. Harden Stripe failure handling, migrations, readiness, and the authenticated control-plane UI with end-to-end verification evidence.

1 files changed, 152 insertions, 0 deletions

diff --git a/internal/providerhealth/tracker_test.go b/internal/providerhealth/tracker_test.go
index 73bd6d5..19752df 100644
--- a/internal/providerhealth/tracker_test.go
+++ b/internal/providerhealth/tracker_test.go
@@ -1,11 +1,23 @@
 package providerhealth
 
 import (
+	"context"
+	"fmt"
+	"log/slog"
+	"os"
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/redis/go-redis/v9"
 )
 
+type captureSink struct {
+	events []Event
+}
+
+func (s *captureSink) Enqueue(event Event) { s.events = append(s.events, event) }
+
 func TestTrackerOpensAndRecoversCircuit(t *testing.T) {
 	now := time.Date(2026, time.August, 6, 0, 0, 0, 0, time.UTC)
 	tracker := New(Options{FailureThreshold: 3, OpenDuration: 30 * time.Second, Now: func() time.Time { return now }})
@@ -31,6 +43,146 @@ func TestTrackerOpensAndRecoversCircuit(t *testing.T) {
 	}
 }
 
+func TestTrackerRecordsActiveProbeMetadata(t *testing.T) {
+	now := time.Date(2026, time.August, 6, 1, 0, 0, 0, time.UTC)
+	tracker := New(Options{Now: func() time.Time { return now }})
+	key := RouteKey{ModelID: "model", ProviderID: "provider", WireAPI: "embeddings"}
+	tracker.Observe(key, Observation{StatusCode: 200, Latency: 20 * time.Millisecond, Active: true})
+	status := tracker.Snapshot()[0]
+	if status.ActiveProbes != 1 || status.LastProbeAt == nil || !status.LastProbeAt.Equal(now) || status.State != "healthy" {
+		t.Fatalf("unexpected active probe status: %+v", status)
+	}
+}
+
+func TestTrackerRecordsTTFTWithoutDoubleCountingAvailability(t *testing.T) {
+	tracker := New(Options{})
+	key := RouteKey{ModelID: "model", ProviderID: "provider", WireAPI: "responses"}
+	tracker.Observe(key, Observation{StatusCode: 200, Latency: 20 * time.Millisecond})
+	tracker.ObserveTTFT(key, 100*time.Millisecond)
+	tracker.ObserveTTFT(key, 200*time.Millisecond)
+	status, found := tracker.StatusFor(key)
+	if !found || status.Attempts != 1 || status.RecentSamples != 1 || status.TTFTSamples != 2 || status.TTFTEWMA != 120 {
+		t.Fatalf("unexpected TTFT status: %+v", status)
+	}
+}
+
+func TestTrackerSharesLocalEventsWithoutRepublishingImportedEvents(t *testing.T) {
+	sink := &captureSink{}
+	tracker := New(Options{Sink: sink})
+	key := RouteKey{ModelID: "model", ProviderID: "provider", WireAPI: "responses"}
+	tracker.Observe(key, Observation{StatusCode: 200, Latency: 20 * time.Millisecond})
+	tracker.ObserveTTFT(key, 100*time.Millisecond)
+	if len(sink.events) != 2 || sink.events[0].Kind != EventOutcome || sink.events[1].Kind != EventTTFT {
+		t.Fatalf("unexpected published events %+v", sink.events)
+	}
+	tracker.ApplyShared(Event{Kind: EventOutcome, Key: key, StatusCode: 503, Failed: true, ObservedAt: time.Now()})
+	tracker.ApplyShared(Event{Kind: EventTTFT, Key: key, LatencyMillis: 250, ObservedAt: time.Now()})
+	if len(sink.events) != 2 {
+		t.Fatalf("imported observations were republished: %d events", len(sink.events))
+	}
+	status, found := tracker.StatusFor(key)
+	if !found || status.Attempts != 2 || status.SharedAttempts != 1 || status.TTFTSamples != 2 || status.SharedTTFTSamples != 1 {
+		t.Fatalf("unexpected shared status %+v", status)
+	}
+}
+
+func TestSharedFailuresOpenLocalCircuit(t *testing.T) {
+	now := time.Date(2026, time.August, 6, 2, 0, 0, 0, time.UTC)
+	tracker := New(Options{FailureThreshold: 3, OpenDuration: 30 * time.Second, Now: func() time.Time { return now }})
+	key := RouteKey{ModelID: "model", ProviderID: "provider", WireAPI: "responses"}
+	for range 3 {
+		tracker.ApplyShared(Event{Kind: EventOutcome, Key: key, StatusCode: 503, Failed: true, ObservedAt: now})
+	}
+	status, found := tracker.StatusFor(key)
+	if !found || !tracker.CircuitOpen(key) || status.State != "open" || status.Attempts != 3 || status.SharedAttempts != 3 {
+		t.Fatalf("shared failures did not open the circuit: %+v", status)
+	}
+}
+
+func TestRedisHistoryDropsWithoutBlockingWhenQueueIsFull(t *testing.T) {
+	client := redis.NewClient(&redis.Options{Addr: "127.0.0.1:1"})
+	defer client.Close()
+	history := NewRedisHistory(HistoryOptions{Enabled: true, Client: client, QueueSize: 1,
+		Logger: slog.New(slog.NewTextHandler(ioDiscard{}, nil))})
+	event := Event{Kind: EventOutcome, Key: RouteKey{ModelID: "model", ProviderID: "provider"}, ObservedAt: time.Now()}
+	history.Enqueue(event)
+	history.Enqueue(event)
+	if history.Dropped() != 1 {
+		t.Fatalf("dropped = %d, want 1", history.Dropped())
+	}
+}
+
+func TestRedisHistorySharesAndReplaysObservations(t *testing.T) {
+	redisURL := os.Getenv("AIGW_TEST_REDIS_URL")
+	if redisURL == "" {
+		t.Skip("AIGW_TEST_REDIS_URL is not set")
+	}
+	stream := fmt.Sprintf("aigw:test:provider-health:%d", time.Now().UnixNano())
+	options, err := redis.ParseURL(redisURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupClient := redis.NewClient(options)
+	t.Cleanup(func() {
+		_, _ = cleanupClient.Del(context.Background(), stream).Result()
+		_ = cleanupClient.Close()
+	})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	key := RouteKey{ModelID: "model", ProviderID: "provider", WireAPI: "responses"}
+	first := New(Options{})
+	second := New(Options{})
+	firstHistory := NewRedisHistory(HistoryOptions{Enabled: true, RedisURL: redisURL, Stream: stream, Instance: "first",
+		TTL: time.Minute, MaxEvents: 100, Logger: slog.New(slog.NewTextHandler(ioDiscard{}, nil))})
+	secondHistory := NewRedisHistory(HistoryOptions{Enabled: true, RedisURL: redisURL, Stream: stream, Instance: "second",
+		TTL: time.Minute, MaxEvents: 100, Logger: slog.New(slog.NewTextHandler(ioDiscard{}, nil))})
+	if firstHistory == nil || secondHistory == nil {
+		t.Fatal("Redis histories were not configured")
+	}
+	t.Cleanup(func() { _ = firstHistory.Close() })
+	t.Cleanup(func() { _ = secondHistory.Close() })
+	first.SetSink(firstHistory)
+	go firstHistory.Run(ctx, first)
+	go secondHistory.Run(ctx, second)
+
+	first.Observe(key, Observation{StatusCode: 200, Latency: 10 * time.Millisecond})
+	first.ObserveTTFT(key, 40*time.Millisecond)
+	waitForSharedStatus(t, second, key, 1, 1)
+	status, _ := first.StatusFor(key)
+	if status.SharedAttempts != 0 || status.SharedTTFTSamples != 0 {
+		t.Fatalf("publisher imported its own events: %+v", status)
+	}
+
+	restarted := New(Options{})
+	restartedHistory := NewRedisHistory(HistoryOptions{Enabled: true, RedisURL: redisURL, Stream: stream, Instance: "restarted",
+		TTL: time.Minute, MaxEvents: 100, Logger: slog.New(slog.NewTextHandler(ioDiscard{}, nil))})
+	if restartedHistory == nil {
+		t.Fatal("restart history was not configured")
+	}
+	t.Cleanup(func() { _ = restartedHistory.Close() })
+	go restartedHistory.Run(ctx, restarted)
+	waitForSharedStatus(t, restarted, key, 1, 1)
+}
+
+func waitForSharedStatus(t *testing.T, tracker *Tracker, key RouteKey, attempts, ttft uint64) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		status, found := tracker.StatusFor(key)
+		if found && status.SharedAttempts >= attempts && status.SharedTTFTSamples >= ttft {
+			return
+		}
+		time.Sleep(25 * time.Millisecond)
+	}
+	status, _ := tracker.StatusFor(key)
+	t.Fatalf("shared status did not converge: %+v", status)
+}
+
+type ioDiscard struct{}
+
+func (ioDiscard) Write(data []byte) (int, error) { return len(data), nil }
+
 func TestTrackerConcurrentObservations(t *testing.T) {
 	tracker := New(Options{FailureThreshold: 1000})
 	key := RouteKey{ModelID: "model", ProviderID: "provider", WireAPI: "chat_completions"}