package httpapi

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
	"github.com/jackc/pgx/v5/pgxpool"
)

type failurePolicyAcceptanceFixture struct {
	ctx        context.Context
	cancel     context.CancelFunc
	db         *store.Store
	pool       *pgxpool.Pool
	server     *httptest.Server
	adminToken string
	apiKey     string
	suffix     string
}

type controlledProviderStub struct {
	server *httptest.Server
	calls  atomic.Int64
}

type acceptancePlatform struct {
	ID string `json:"id"`
}

type acceptancePlatformModel struct {
	ID string `json:"id"`
}

type acceptanceTaskDetail struct {
	ID                string         `json:"id"`
	Status            string         `json:"status"`
	ErrorCode         string         `json:"errorCode"`
	ErrorMessage      string         `json:"errorMessage"`
	FinalChargeAmount float64        `json:"finalChargeAmount"`
	BillingStatus     string         `json:"billingStatus"`
	Metrics           map[string]any `json:"metrics"`
	Attempts          []struct {
		AttemptNo      int              `json:"attemptNo"`
		PlatformID     string           `json:"platformId"`
		PlatformName   string           `json:"platformName"`
		PlatformModel  string           `json:"platformModelId"`
		Status         string           `json:"status"`
		ErrorCode      string           `json:"errorCode"`
		ErrorMessage   string           `json:"errorMessage"`
		DynamicMetrics map[string]any   `json:"metrics"`
		Trace          []map[string]any `json:"trace"`
	} `json:"attempts"`
}

func TestFailurePolicyHTTPAcceptance(t *testing.T) {
	fixture := newFailurePolicyAcceptanceFixture(t)
	defer fixture.close()

	originalPolicy, err := fixture.db.GetActiveRunnerPolicy(fixture.ctx)
	if err != nil {
		t.Fatalf("read original runner policy: %v", err)
	}
	t.Cleanup(func() {
		_, _ = fixture.db.UpsertDefaultRunnerPolicy(context.Background(), store.RunnerPolicyInput{
			PolicyKey:            originalPolicy.PolicyKey,
			Name:                 originalPolicy.Name,
			Description:          originalPolicy.Description,
			FailoverPolicy:       originalPolicy.FailoverPolicy,
			HardStopPolicy:       originalPolicy.HardStopPolicy,
			PriorityDemotePolicy: originalPolicy.PriorityDemotePolicy,
			SingleSourcePolicy:   originalPolicy.SingleSourcePolicy,
			CacheAffinityPolicy:  originalPolicy.CacheAffinityPolicy,
			Metadata:             originalPolicy.Metadata,
			Status:               originalPolicy.Status,
		})
	})
	fixture.setRunnerPolicy(t, true)

	t.Run("Gemini 429 后轮转且只冷却模型一次", fixture.testGemini429ImageFailover)
	t.Run("Gemini 空图片响应返回具体原因并失败", fixture.testGeminiMissingImageFails)
	t.Run("5xx 同平台重试耗尽后降权并轮转", fixture.testServerErrorRetryThenFailover)
	t.Run("401 立即禁用平台并轮转", fixture.testAuthFailureDisableThenFailover)
	t.Run("400 参数错误硬停止", fixture.testBadRequestStops)
	t.Run("旧 degradePolicy 仅转换一次", fixture.testLegacyDegradeDoesNotDoubleApply)
	t.Run("并发单源保护至少保留一个来源", fixture.testConcurrentSingleSourceProtection)
	t.Run("全部冷却同步返回脱敏 429", fixture.testAllCoolingSynchronousContract)
	t.Run("全部冷却异步零 attempt 重排并恢复", fixture.testAllCoolingAsyncRecovery)
	t.Run("流式首包边界控制轮转", fixture.testStreamingBoundary)
	t.Run("管理接口严格校验且不覆盖原策略", fixture.testRunnerPolicyHTTPValidation)
}

func newFailurePolicyAcceptanceFixture(t *testing.T) *failurePolicyAcceptanceFixture {
	t.Helper()
	databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
	if databaseURL == "" {
		t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run failure-policy HTTP acceptance")
	}
	ctx, cancel := context.WithCancel(context.Background())
	applyMigration(t, ctx, databaseURL)
	db, err := store.Connect(ctx, databaseURL)
	if err != nil {
		cancel()
		t.Fatalf("connect acceptance store: %v", err)
	}
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		db.Close()
		cancel()
		t.Fatalf("connect acceptance pool: %v", err)
	}
	handler := NewServerWithContext(ctx, config.Config{
		AppEnv:                            "test",
		HTTPAddr:                          ":0",
		DatabaseURL:                       databaseURL,
		IdentityMode:                      "hybrid",
		JWTSecret:                         "failure-policy-http-acceptance-secret",
		BillingEngineMode:                 "observe",
		CORSAllowedOrigin:                 "*",
		AsyncQueueWorkerEnabled:           true,
		AsyncWorkerHardLimit:              32,
		AsyncWorkerInstanceHardLimit:      32,
		AsyncWorkerRefreshIntervalSeconds: 1,
	}, db, slog.New(slog.NewTextHandler(io.Discard, nil)))
	server := httptest.NewServer(handler)
	fixture := &failurePolicyAcceptanceFixture{
		ctx:    ctx,
		cancel: cancel,
		db:     db,
		pool:   pool,
		server: server,
		suffix: strconv.FormatInt(time.Now().UnixNano(), 10),
	}

	username := "failure_acceptance_" + fixture.suffix
	password := "password123"
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{
		"username": username,
		"email":    username + "@example.com",
		"password": password,
	}, http.StatusCreated, &map[string]any{})
	if _, err := pool.Exec(ctx, `UPDATE gateway_users SET roles = '["admin"]'::jsonb WHERE username = $1`, username); err != nil {
		fixture.close()
		t.Fatalf("promote acceptance admin: %v", err)
	}
	var login struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
		"account": username, "password": password,
	}, http.StatusOK, &login)
	fixture.adminToken = login.AccessToken
	var gatewayUserID string
	if err := pool.QueryRow(ctx, `SELECT id::text FROM gateway_users WHERE username=$1`, username).Scan(&gatewayUserID); err != nil {
		fixture.close()
		t.Fatalf("read acceptance user: %v", err)
	}
	doJSON(t, server.URL, http.MethodPatch, "/api/admin/users/"+gatewayUserID+"/wallet", fixture.adminToken, map[string]any{
		"currency": "resource",
		"balance":  1000000,
		"reason":   "seed isolated failure-policy acceptance wallet",
	}, http.StatusOK, &map[string]any{})
	var key struct {
		Secret string `json:"secret"`
	}
	doJSON(t, server.URL, http.MethodPost, "/api/v1/api-keys", fixture.adminToken, map[string]any{
		"name": "failure policy acceptance",
	}, http.StatusCreated, &key)
	fixture.apiKey = key.Secret
	return fixture
}

func (f *failurePolicyAcceptanceFixture) close() {
	if f.server != nil {
		f.server.Close()
	}
	if f.cancel != nil {
		f.cancel()
	}
	if f.pool != nil {
		f.pool.Close()
	}
	if f.db != nil {
		f.db.Close()
	}
}

func (f *failurePolicyAcceptanceFixture) setRunnerPolicy(t *testing.T, singleSource bool) {
	t.Helper()
	doJSON(t, f.server.URL, http.MethodPatch, "/api/admin/runtime/runner-policy", f.adminToken, standardFailureRunnerPolicy(singleSource), http.StatusOK, &map[string]any{})
}

func standardFailureRunnerPolicy(singleSource bool) map[string]any {
	return map[string]any{
		"policyKey":   "default-runner-v1",
		"name":        "失败策略 HTTP 验收",
		"status":      "active",
		"description": "真实 provider HTTP client 的失败策略验收",
		"failoverPolicy": map[string]any{
			"enabled":            true,
			"maxPlatforms":       99,
			"maxDurationSeconds": 600,
			"actionRules": []any{
				map[string]any{"categories": []any{"rate_limit"}, "statusCodes": []any{429}, "action": "cooldown_and_next", "target": "model", "cooldownSeconds": 30},
				map[string]any{"categories": []any{"auth_error"}, "statusCodes": []any{401, 403}, "action": "disable_and_next", "target": "platform"},
				map[string]any{"categories": []any{"timeout", "provider_5xx", "provider_overloaded"}, "statusCodes": []any{408, 500, 502, 503, 504}, "action": "demote_and_next", "target": "platform", "demoteSteps": 1},
				map[string]any{"categories": []any{"network", "stream_error"}, "action": "next"},
				map[string]any{"categories": []any{"request_error", "unsupported_model", "user_permission", "insufficient_balance"}, "statusCodes": []any{400, 404, 409, 422}, "action": "stop"},
			},
		},
		"singleSourcePolicy":  map[string]any{"enabled": singleSource},
		"cacheAffinityPolicy": map[string]any{"enabled": false},
	}
}

func (f *failurePolicyAcceptanceFixture) createBaseModel(t *testing.T, model string, modelType string) string {
	t.Helper()
	var created struct {
		ID string `json:"id"`
	}
	doJSON(t, f.server.URL, http.MethodPost, "/api/admin/catalog/base-models", f.adminToken, map[string]any{
		"providerKey":       "openai",
		"canonicalModelKey": "openai:" + model,
		"invocationName":    model,
		"providerModelName": model,
		"modelType":         []string{modelType},
		"displayName":       model,
		"status":            "active",
		"capabilities":      map[string]any{},
		"baseBillingConfig": map[string]any{},
	}, http.StatusCreated, &created)
	return created.ID
}

func (f *failurePolicyAcceptanceFixture) createPlatformModel(
	t *testing.T,
	name string,
	provider string,
	baseURL string,
	priority int,
	baseModelID string,
	providerModelName string,
	modelType string,
	retryAttempts int,
	runtimePolicySetID string,
	runtimeOverride map[string]any,
) (acceptancePlatform, acceptancePlatformModel) {
	t.Helper()
	var platform acceptancePlatform
	doJSON(t, f.server.URL, http.MethodPost, "/api/admin/platforms", f.adminToken, map[string]any{
		"provider":    provider,
		"platformKey": "acceptance-" + name + "-" + f.suffix,
		"name":        "acceptance-" + name,
		"baseUrl":     strings.TrimRight(baseURL, "/"),
		"authType":    "bearer",
		"credentials": map[string]any{"apiKey": "controlled-upstream-test-key"},
		"priority":    priority,
		"status":      "enabled",
	}, http.StatusCreated, &platform)
	var model acceptancePlatformModel
	payload := map[string]any{
		"baseModelId":       baseModelID,
		"providerModelName": providerModelName,
		"modelType":         []string{modelType},
		"retryPolicy":       map[string]any{"enabled": true, "maxAttempts": retryAttempts},
	}
	if runtimePolicySetID != "" {
		payload["runtimePolicySetId"] = runtimePolicySetID
	}
	if len(runtimeOverride) > 0 {
		payload["runtimePolicyOverride"] = runtimeOverride
	}
	doJSON(t, f.server.URL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", f.adminToken, payload, http.StatusCreated, &model)
	return platform, model
}

func newControlledProviderStub(handler func(call int, w http.ResponseWriter, r *http.Request)) *controlledProviderStub {
	stub := &controlledProviderStub{}
	stub.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		call := int(stub.calls.Add(1))
		handler(call, w, r)
	}))
	return stub
}

func (s *controlledProviderStub) close() {
	if s != nil && s.server != nil {
		s.server.Close()
	}
}

func writeStubJSON(w http.ResponseWriter, status int, payload any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(payload)
}

func writeOpenAIChatSuccess(w http.ResponseWriter, content string) {
	writeStubJSON(w, http.StatusOK, map[string]any{
		"id": "chatcmpl-controlled", "object": "chat.completion", "created": time.Now().Unix(), "model": "controlled-model",
		"choices": []any{map[string]any{
			"index": 0, "message": map[string]any{"role": "assistant", "content": content}, "finish_reason": "stop",
		}},
		"usage": map[string]any{"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
	})
}

func writeOpenAIImageSuccess(w http.ResponseWriter) {
	writeStubJSON(w, http.StatusOK, map[string]any{
		"created": time.Now().Unix(),
		"data":    []any{map[string]any{"url": "https://example.invalid/controlled-test.png"}},
	})
}

func (f *failurePolicyAcceptanceFixture) postJSON(t *testing.T, path string, payload any, headers map[string]string) (int, http.Header, []byte) {
	t.Helper()
	raw, err := json.Marshal(payload)
	if err != nil {
		t.Fatalf("marshal acceptance payload: %v", err)
	}
	request, err := http.NewRequest(http.MethodPost, f.server.URL+path, bytes.NewReader(raw))
	if err != nil {
		t.Fatalf("build acceptance request: %v", err)
	}
	request.Header.Set("Authorization", "Bearer "+f.apiKey)
	request.Header.Set("Content-Type", "application/json")
	for key, value := range headers {
		request.Header.Set(key, value)
	}
	response, err := http.DefaultClient.Do(request)
	if err != nil {
		t.Fatalf("acceptance request %s: %v", path, err)
	}
	defer response.Body.Close()
	body, _ := io.ReadAll(response.Body)
	return response.StatusCode, response.Header.Clone(), body
}

func (f *failurePolicyAcceptanceFixture) taskIDForModel(t *testing.T, model string, startedAt time.Time) string {
	t.Helper()
	var taskID string
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		err := f.pool.QueryRow(f.ctx, `
SELECT id::text
FROM gateway_tasks
WHERE model = $1
  AND created_at >= $2
ORDER BY created_at DESC
LIMIT 1`, model, startedAt.Add(-time.Second)).Scan(&taskID)
		if err == nil && taskID != "" {
			return taskID
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatalf("task for model %q was not persisted", model)
	return ""
}

func (f *failurePolicyAcceptanceFixture) loadTask(t *testing.T, taskID string) acceptanceTaskDetail {
	t.Helper()
	var detail acceptanceTaskDetail
	doJSON(t, f.server.URL, http.MethodGet, "/api/v1/tasks/"+taskID, f.apiKey, nil, http.StatusOK, &detail)
	return detail
}

func countFailureEffects(detail acceptanceTaskDetail, effect string) int {
	count := 0
	for _, attempt := range detail.Attempts {
		trace := attempt.Trace
		for _, raw := range anySlice(attempt.DynamicMetrics["trace"]) {
			if entry, ok := raw.(map[string]any); ok {
				trace = append(trace, entry)
			}
		}
		for _, entry := range trace {
			if entry["event"] == "failure_decision" && entry["effect"] == effect {
				count++
			}
		}
	}
	return count
}

func anySlice(value any) []any {
	items, _ := value.([]any)
	return items
}

func (f *failurePolicyAcceptanceFixture) testGemini429ImageFailover(t *testing.T) {
	model := "acceptance-gemini-image-" + f.suffix
	baseModelID := f.createBaseModel(t, model, "image_generate")
	gemini := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
		writeStubJSON(w, http.StatusTooManyRequests, map[string]any{
			"error": map[string]any{"code": 429, "message": "RESOURCE_EXHAUSTED controlled failure", "status": "RESOURCE_EXHAUSTED"},
		})
	})
	defer gemini.close()
	openai := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIImageSuccess(w) })
	defer openai.close()
	platformA, modelA := f.createPlatformModel(t, "gemini-image-a", "gemini", gemini.server.URL, 10, baseModelID, "gemini-image-controlled", "image_generate", 3, "", nil)
	platformB, _ := f.createPlatformModel(t, "openai-image-b", "openai", openai.server.URL+"/v1", 20, baseModelID, "openai-image-controlled", "image_generate", 1, "", nil)

	startedAt := time.Now()
	status, _, body := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "controlled image"}, nil)
	if status != http.StatusOK {
		t.Fatalf("status=%d body=%s", status, body)
	}
	if gemini.calls.Load() != 1 || openai.calls.Load() != 1 {
		t.Fatalf("upstream calls gemini=%d openai=%d, want 1/1", gemini.calls.Load(), openai.calls.Load())
	}
	var modelCooling, platformCooling bool
	if err := f.pool.QueryRow(f.ctx, `
SELECT COALESCE(model.cooldown_until > now(), false), COALESCE(platform.cooldown_until > now(), false)
FROM platform_models model
JOIN integration_platforms platform ON platform.id = model.platform_id
WHERE model.id = $1::uuid`, modelA.ID).Scan(&modelCooling, &platformCooling); err != nil {
		t.Fatalf("read Gemini cooldown state: %v", err)
	}
	if !modelCooling || platformCooling {
		t.Fatalf("cooldown state model=%v platform=%v, want true/false for platform %s", modelCooling, platformCooling, platformA.ID)
	}
	detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt))
	if len(detail.Attempts) != 2 || detail.Attempts[0].PlatformID != platformA.ID || detail.Attempts[1].PlatformID != platformB.ID || countFailureEffects(detail, "cooldown") != 1 {
		t.Fatalf("unexpected task attempts/trace: %+v", detail)
	}
}

func (f *failurePolicyAcceptanceFixture) testGeminiMissingImageFails(t *testing.T) {
	model := "acceptance-gemini-empty-image-" + f.suffix
	baseModelID := f.createBaseModel(t, model, "image_generate")
	gemini := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("X-Request-Id", "acceptance-gemini-empty-image")
		writeStubJSON(w, http.StatusOK, map[string]any{
			"promptFeedback": map[string]any{
				"blockReason":        "SAFETY",
				"blockReasonMessage": "The request was blocked by the image safety policy.",
			},
			"usageMetadata": map[string]any{"promptTokenCount": 3, "totalTokenCount": 3},
		})
	})
	defer gemini.close()
	f.createPlatformModel(t, "gemini-empty-image", "gemini", gemini.server.URL, 10, baseModelID, "gemini-image-controlled", "image_generate", 3, "", nil)

	startedAt := time.Now()
	status, _, body := f.postJSON(t, "/api/v1/images/generations", map[string]any{
		"model": model, "prompt": "controlled blocked image",
	}, nil)
	if status != http.StatusBadGateway {
		t.Fatalf("status=%d body=%s", status, body)
	}
	var response struct {
		Error struct {
			Code    string         `json:"code"`
			Message string         `json:"message"`
			Details map[string]any `json:"details"`
		} `json:"error"`
	}
	if err := json.Unmarshal(body, &response); err != nil {
		t.Fatalf("decode missing image response: %v body=%s", err, body)
	}
	if response.Error.Code != "gemini_image_output_missing" ||
		!strings.Contains(response.Error.Message, "blockReason=SAFETY") ||
		response.Error.Details["reason"] != "prompt_blocked" {
		t.Fatalf("missing image response should expose the concrete reason: %+v", response.Error)
	}
	if gemini.calls.Load() != 1 {
		t.Fatalf("non-retryable missing image response calls=%d, want 1", gemini.calls.Load())
	}
	detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt))
	if detail.Status != "failed" ||
		detail.ErrorCode != "gemini_image_output_missing" ||
		!strings.Contains(detail.ErrorMessage, "blockReason=SAFETY") ||
		detail.FinalChargeAmount != 0 {
		t.Fatalf("missing image task should fail without charge: %+v", detail)
	}
	if len(detail.Attempts) != 1 ||
		detail.Attempts[0].Status != "failed" ||
		detail.Attempts[0].ErrorCode != "gemini_image_output_missing" ||
		!strings.Contains(detail.Attempts[0].ErrorMessage, "blockReason=SAFETY") {
		t.Fatalf("missing image attempt should be recorded as failed: %+v", detail.Attempts)
	}
	errorDetails, _ := detail.Metrics["errorDetails"].(map[string]any)
	if errorDetails["reason"] != "prompt_blocked" || errorDetails["promptBlockReason"] != "SAFETY" {
		t.Fatalf("missing image task lost diagnostic details: %+v", detail.Metrics)
	}
}

func (f *failurePolicyAcceptanceFixture) testServerErrorRetryThenFailover(t *testing.T) {
	model := "acceptance-5xx-" + f.suffix
	baseModelID := f.createBaseModel(t, model, "text_generate")
	failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
		writeStubJSON(w, http.StatusInternalServerError, map[string]any{"error": map[string]any{"message": "controlled 500"}})
	})
	defer failed.close()
	success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "fallback ok") })
	defer success.close()
	platformA, _ := f.createPlatformModel(t, "5xx-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "controlled-5xx", "text_generate", 2, "", nil)
	platformB, _ := f.createPlatformModel(t, "5xx-b", "openai", success.server.URL+"/v1", 20, baseModelID, "controlled-success", "text_generate", 1, "", nil)

	startedAt := time.Now()
	status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{
		"model": model, "messages": []any{map[string]any{"role": "user", "content": "retry"}},
	}, nil)
	if status != http.StatusOK {
		t.Fatalf("status=%d body=%s", status, body)
	}
	if failed.calls.Load() != 2 || success.calls.Load() != 1 {
		t.Fatalf("upstream calls failed=%d success=%d, want 2/1", failed.calls.Load(), success.calls.Load())
	}
	detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt))
	if len(detail.Attempts) != 3 ||
		detail.Attempts[0].PlatformID != platformA.ID ||
		detail.Attempts[1].PlatformID != platformA.ID ||
		detail.Attempts[2].PlatformID != platformB.ID ||
		countFailureEffects(detail, "demote") != 1 {
		t.Fatalf("unexpected retry/failover chain: %+v", detail)
	}
	var dynamicPriority *int
	if err := f.pool.QueryRow(f.ctx, `SELECT dynamic_priority FROM integration_platforms WHERE id=$1::uuid`, platformA.ID).Scan(&dynamicPriority); err != nil {
		t.Fatalf("read demoted priority: %v", err)
	}
	if dynamicPriority == nil {
		t.Fatal("failed platform should be demoted exactly once after retry exhaustion")
	}
}

func (f *failurePolicyAcceptanceFixture) testAuthFailureDisableThenFailover(t *testing.T) {
	model := "acceptance-auth-" + f.suffix
	baseModelID := f.createBaseModel(t, model, "text_generate")
	failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
		writeStubJSON(w, http.StatusUnauthorized, map[string]any{"error": map[string]any{"message": "invalid api key"}})
	})
	defer failed.close()
	success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "auth fallback") })
	defer success.close()
	platformA, _ := f.createPlatformModel(t, "auth-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "controlled-auth", "text_generate", 3, "", nil)
	_, _ = f.createPlatformModel(t, "auth-b", "openai", success.server.URL+"/v1", 20, baseModelID, "controlled-success", "text_generate", 1, "", nil)

	startedAt := time.Now()
	status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{
		"model": model, "messages": []any{map[string]any{"role": "user", "content": "auth"}},
	}, nil)
	if status != http.StatusOK {
		t.Fatalf("status=%d body=%s", status, body)
	}
	if failed.calls.Load() != 1 || success.calls.Load() != 1 {
		t.Fatalf("upstream calls failed=%d success=%d, want 1/1", failed.calls.Load(), success.calls.Load())
	}
	var statusValue string
	if err := f.pool.QueryRow(f.ctx, `SELECT status FROM integration_platforms WHERE id=$1::uuid`, platformA.ID).Scan(&statusValue); err != nil {
		t.Fatalf("read disabled platform: %v", err)
	}
	detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt))
	if statusValue != "disabled" || countFailureEffects(detail, "disable") != 1 || len(detail.Attempts) != 2 {
		t.Fatalf("unexpected auth disable result status=%s detail=%+v", statusValue, detail)
	}
}

func (f *failurePolicyAcceptanceFixture) testBadRequestStops(t *testing.T) {
	model := "acceptance-400-" + f.suffix
	baseModelID := f.createBaseModel(t, model, "text_generate")
	failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
		writeStubJSON(w, http.StatusBadRequest, map[string]any{"error": map[string]any{"message": "controlled invalid parameter", "code": "invalid_parameter"}})
	})
	defer failed.close()
	success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "must not run") })
	defer success.close()
	platformA, modelA := f.createPlatformModel(t, "400-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "controlled-bad", "text_generate", 3, "", nil)
	_, _ = f.createPlatformModel(t, "400-b", "openai", success.server.URL+"/v1", 20, baseModelID, "controlled-success", "text_generate", 1, "", nil)

	status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{
		"model": model, "messages": []any{map[string]any{"role": "user", "content": "bad"}},
	}, nil)
	if status != http.StatusBadRequest || success.calls.Load() != 0 || failed.calls.Load() != 1 {
		t.Fatalf("status=%d calls=%d/%d body=%s", status, failed.calls.Load(), success.calls.Load(), body)
	}
	var platformStatus string
	var modelEnabled bool
	var platformCooldown, modelCooldown *time.Time
	if err := f.pool.QueryRow(f.ctx, `
SELECT platform.status, model.enabled, platform.cooldown_until, model.cooldown_until
FROM integration_platforms platform
JOIN platform_models model ON model.platform_id=platform.id
WHERE platform.id=$1::uuid AND model.id=$2::uuid`, platformA.ID, modelA.ID).Scan(&platformStatus, &modelEnabled, &platformCooldown, &modelCooldown); err != nil {
		t.Fatalf("read hard-stop state: %v", err)
	}
	if platformStatus != "enabled" || !modelEnabled || platformCooldown != nil || modelCooldown != nil {
		t.Fatalf("hard stop mutated runtime state: status=%s enabled=%v platformCooldown=%v modelCooldown=%v", platformStatus, modelEnabled, platformCooldown, modelCooldown)
	}
}

func (f *failurePolicyAcceptanceFixture) testLegacyDegradeDoesNotDoubleApply(t *testing.T) {
	model := "acceptance-legacy-" + f.suffix
	baseModelID := f.createBaseModel(t, model, "text_generate")
	var policySet struct {
		ID string `json:"id"`
	}
	doJSON(t, f.server.URL, http.MethodPost, "/api/admin/runtime/policy-sets", f.adminToken, map[string]any{
		"policyKey": "acceptance-legacy-" + f.suffix,
		"name":      "acceptance legacy degrade",
		"degradePolicy": map[string]any{
			"enabled": true, "keywords": []any{"rate_limit"}, "cooldownSeconds": 45,
		},
	}, http.StatusCreated, &policySet)
	failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
		writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "rate_limit controlled"}})
	})
	defer failed.close()
	success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "legacy fallback") })
	defer success.close()
	_, modelA := f.createPlatformModel(t, "legacy-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "legacy-failed", "text_generate", 3, policySet.ID, nil)
	_, _ = f.createPlatformModel(t, "legacy-b", "openai", success.server.URL+"/v1", 20, baseModelID, "legacy-success", "text_generate", 1, "", nil)

	startedAt := time.Now()
	status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{
		"model": model, "messages": []any{map[string]any{"role": "user", "content": "legacy"}},
	}, nil)
	if status != http.StatusOK || failed.calls.Load() != 1 || success.calls.Load() != 1 {
		t.Fatalf("status=%d calls=%d/%d body=%s", status, failed.calls.Load(), success.calls.Load(), body)
	}
	detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt))
	if countFailureEffects(detail, "cooldown") != 1 {
		t.Fatalf("legacy and global policy must not both execute: %+v", detail)
	}
	var remaining float64
	if err := f.pool.QueryRow(f.ctx, `SELECT EXTRACT(EPOCH FROM cooldown_until-now())::float8 FROM platform_models WHERE id=$1::uuid`, modelA.ID).Scan(&remaining); err != nil {
		t.Fatalf("read legacy cooldown: %v", err)
	}
	if remaining < 35 || remaining > 50 {
		t.Fatalf("legacy cooldown should be applied once with 45 seconds, remaining=%f", remaining)
	}
}

func (f *failurePolicyAcceptanceFixture) testConcurrentSingleSourceProtection(t *testing.T) {
	f.setRunnerPolicy(t, true)
	model := "acceptance-concurrent-" + f.suffix
	baseModelID := f.createBaseModel(t, model, "text_generate")
	stubs := make([]*controlledProviderStub, 0, 3)
	modelIDs := make([]string, 0, 3)
	for index := 0; index < 3; index++ {
		stub := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
			writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "controlled 429"}})
		})
		stubs = append(stubs, stub)
		_, platformModel := f.createPlatformModel(t, fmt.Sprintf("concurrent-%d", index), "openai", stub.server.URL+"/v1", 10+index*10, baseModelID, fmt.Sprintf("concurrent-%d", index), "text_generate", 1, "", nil)
		modelIDs = append(modelIDs, platformModel.ID)
	}
	defer func() {
		for _, stub := range stubs {
			stub.close()
		}
	}()

	var wait sync.WaitGroup
	errorsFound := make(chan error, 12)
	for index := 0; index < 12; index++ {
		wait.Add(1)
		go func() {
			defer wait.Done()
			status, _, body, err := f.postJSONNoFail("/api/v1/chat/completions", map[string]any{
				"model": model, "messages": []any{map[string]any{"role": "user", "content": "concurrent"}},
			})
			if err != nil {
				errorsFound <- err
				return
			}
			if status != http.StatusTooManyRequests {
				errorsFound <- fmt.Errorf("status=%d body=%s", status, body)
			}
		}()
	}
	wait.Wait()
	close(errorsFound)
	for err := range errorsFound {
		t.Error(err)
	}

	var available int
	if err := f.pool.QueryRow(f.ctx, `
SELECT count(*)
FROM platform_models model
JOIN integration_platforms platform ON platform.id=model.platform_id
WHERE model.id = ANY($1::uuid[])
  AND model.enabled=true
  AND platform.status='enabled'
  AND (model.cooldown_until IS NULL OR model.cooldown_until <= now())
  AND (platform.cooldown_until IS NULL OR platform.cooldown_until <= now())`, modelIDs).Scan(&available); err != nil {
		t.Fatalf("count available sources: %v", err)
	}
	if available < 1 {
		t.Fatal("concurrent cooldown decisions disabled or cooled every source")
	}
	before := make([]int64, len(stubs))
	for index, stub := range stubs {
		before[index] = stub.calls.Load()
	}
	status, _, _, err := f.postJSONNoFail("/api/v1/chat/completions", map[string]any{
		"model": model, "messages": []any{map[string]any{"role": "user", "content": "post-cooldown"}},
	})
	if err != nil || status != http.StatusTooManyRequests {
		t.Fatalf("post-cooldown request status=%d err=%v", status, err)
	}
	for index, modelID := range modelIDs {
		var cooled bool
		if err := f.pool.QueryRow(f.ctx, `SELECT COALESCE(cooldown_until > now(), false) FROM platform_models WHERE id=$1::uuid`, modelID).Scan(&cooled); err != nil {
			t.Fatalf("read source cooldown: %v", err)
		}
		if cooled && stubs[index].calls.Load() != before[index] {
			t.Fatalf("cooled source %d was called again: before=%d after=%d", index, before[index], stubs[index].calls.Load())
		}
	}
}

func (f *failurePolicyAcceptanceFixture) postJSONNoFail(path string, payload any) (int, http.Header, []byte, error) {
	raw, err := json.Marshal(payload)
	if err != nil {
		return 0, nil, nil, err
	}
	request, err := http.NewRequest(http.MethodPost, f.server.URL+path, bytes.NewReader(raw))
	if err != nil {
		return 0, nil, nil, err
	}
	request.Header.Set("Authorization", "Bearer "+f.apiKey)
	request.Header.Set("Content-Type", "application/json")
	response, err := http.DefaultClient.Do(request)
	if err != nil {
		return 0, nil, nil, err
	}
	defer response.Body.Close()
	body, err := io.ReadAll(response.Body)
	return response.StatusCode, response.Header.Clone(), body, err
}

func (f *failurePolicyAcceptanceFixture) testAllCoolingSynchronousContract(t *testing.T) {
	f.setRunnerPolicy(t, false)
	defer f.setRunnerPolicy(t, true)
	model := "acceptance-all-cooling-" + f.suffix
	baseModelID := f.createBaseModel(t, model, "image_generate")
	stubA := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
		writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "private-a-raw-error"}})
	})
	defer stubA.close()
	stubB := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
		writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "private-b-raw-error"}})
	})
	defer stubB.close()
	override := func(seconds int) map[string]any {
		return map[string]any{"failoverPolicy": map[string]any{"actionRules": []any{
			map[string]any{"statusCodes": []any{429}, "action": "cooldown_and_next", "target": "model", "cooldownSeconds": seconds},
		}}}
	}
	_, modelA := f.createPlatformModel(t, "cooling-a", "openai", stubA.server.URL+"/v1", 10, baseModelID, "cooling-a", "image_generate", 3, "", override(30))
	_, modelB := f.createPlatformModel(t, "cooling-b", "openai", stubB.server.URL+"/v1", 20, baseModelID, "cooling-b", "image_generate", 3, "", override(60))
	defer func() {
		_, _ = f.pool.Exec(context.Background(), `UPDATE platform_models SET cooldown_until=NULL WHERE id IN ($1::uuid,$2::uuid)`, modelA.ID, modelB.ID)
	}()

	status, _, _ := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "cool both"}, nil)
	if status != http.StatusTooManyRequests || stubA.calls.Load() != 1 || stubB.calls.Load() != 1 {
		t.Fatalf("initial cooling request status=%d calls=%d/%d", status, stubA.calls.Load(), stubB.calls.Load())
	}
	status, headers, body := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "all cooling"}, nil)
	if status != http.StatusTooManyRequests {
		t.Fatalf("cooldown contract status=%d body=%s", status, body)
	}
	retryAfter, err := strconv.Atoi(headers.Get("Retry-After"))
	if err != nil || retryAfter < 20 || retryAfter > 31 {
		t.Fatalf("Retry-After=%q, want earliest recovery near 30 seconds", headers.Get("Retry-After"))
	}
	bodyText := string(body)
	for _, forbidden := range []string{"acceptance-cooling-a", "acceptance-cooling-b", stubA.server.URL, stubB.server.URL, "private-a-raw-error", "private-b-raw-error"} {
		if strings.Contains(bodyText, forbidden) {
			t.Fatalf("cooldown response leaked %q: %s", forbidden, bodyText)
		}
	}
	var envelope struct {
		Error struct {
			Code    string         `json:"code"`
			Details map[string]any `json:"details"`
		} `json:"error"`
	}
	if err := json.Unmarshal(body, &envelope); err != nil {
		t.Fatalf("decode cooldown response: %v body=%s", err, body)
	}
	if envelope.Error.Code != "model_cooling_down" || len(envelope.Error.Details) != 2 {
		t.Fatalf("unexpected cooldown envelope: %+v", envelope)
	}
	if stubA.calls.Load() != 1 || stubB.calls.Load() != 1 {
		t.Fatalf("candidate selection cooldown must not create provider attempts: calls=%d/%d", stubA.calls.Load(), stubB.calls.Load())
	}
}

func (f *failurePolicyAcceptanceFixture) testAllCoolingAsyncRecovery(t *testing.T) {
	f.setRunnerPolicy(t, false)
	defer f.setRunnerPolicy(t, true)
	model := "acceptance-async-cooling-" + f.suffix
	baseModelID := f.createBaseModel(t, model, "image_generate")
	stubA := newControlledProviderStub(func(call int, w http.ResponseWriter, _ *http.Request) {
		if call == 1 {
			writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "controlled first 429"}})
			return
		}
		writeOpenAIImageSuccess(w)
	})
	defer stubA.close()
	stubB := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
		writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "controlled second 429"}})
	})
	defer stubB.close()
	override := func(seconds int) map[string]any {
		return map[string]any{"failoverPolicy": map[string]any{"actionRules": []any{
			map[string]any{"statusCodes": []any{429}, "action": "cooldown_and_next", "target": "model", "cooldownSeconds": seconds},
		}}}
	}
	_, modelA := f.createPlatformModel(t, "async-cooling-a", "openai", stubA.server.URL+"/v1", 10, baseModelID, "async-a", "image_generate", 1, "", override(4))
	_, modelB := f.createPlatformModel(t, "async-cooling-b", "openai", stubB.server.URL+"/v1", 20, baseModelID, "async-b", "image_generate", 1, "", override(8))
	defer func() {
		_, _ = f.pool.Exec(context.Background(), `UPDATE platform_models SET cooldown_until=NULL WHERE id IN ($1::uuid,$2::uuid)`, modelA.ID, modelB.ID)
	}()
	status, _, _ := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "prime cooldown"}, nil)
	if status != http.StatusTooManyRequests {
		t.Fatalf("prime cooldown status=%d", status)
	}

	startedAt := time.Now()
	status, _, body := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "async recovery"}, map[string]string{"X-Async": "true"})
	if status != http.StatusAccepted {
		t.Fatalf("async submit status=%d body=%s", status, body)
	}
	taskID := f.taskIDForModel(t, model, startedAt)
	deadline := time.Now().Add(2 * time.Second)
	queuedObserved := false
	for time.Now().Before(deadline) {
		detail := f.loadTask(t, taskID)
		if detail.Status == "queued" && len(detail.Attempts) == 0 {
			queuedObserved = true
			break
		}
		time.Sleep(25 * time.Millisecond)
	}
	if !queuedObserved {
		t.Fatal("async cooling task was not observed queued with zero provider attempts")
	}
	deadline = time.Now().Add(12 * time.Second)
	var completed acceptanceTaskDetail
	for time.Now().Before(deadline) {
		completed = f.loadTask(t, taskID)
		if completed.Status == "succeeded" {
			break
		}
		time.Sleep(100 * time.Millisecond)
	}
	if completed.Status != "succeeded" || len(completed.Attempts) != 1 || stubA.calls.Load() != 2 {
		t.Fatalf("async task did not recover on first restored candidate: task=%+v callsA=%d callsB=%d", completed, stubA.calls.Load(), stubB.calls.Load())
	}
}

func (f *failurePolicyAcceptanceFixture) testStreamingBoundary(t *testing.T) {
	t.Run("延迟 429 在首个 delta 前仍冷却并轮转", func(t *testing.T) {
		model := "acceptance-stream-delayed-429-" + f.suffix
		baseModelID := f.createBaseModel(t, model, "text_generate")
		failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
			time.Sleep(125 * time.Millisecond)
			writeStubJSON(w, http.StatusTooManyRequests, map[string]any{
				"error": map[string]any{"message": "controlled delayed 429"},
			})
		})
		defer failed.close()
		success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
			w.Header().Set("Content-Type", "text/event-stream")
			_, _ = io.WriteString(w, "data: {\"id\":\"chunk-delayed-429\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"delayed 429 fallback\"},\"finish_reason\":null}]}\n\n")
			_, _ = io.WriteString(w, "data: {\"id\":\"chunk-delayed-429\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n")
			_, _ = io.WriteString(w, "data: [DONE]\n\n")
		})
		defer success.close()
		platformA, modelA := f.createPlatformModel(t, "stream-delayed-429-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "stream-delayed-429", "text_generate", 1, "", nil)
		platformB, _ := f.createPlatformModel(t, "stream-delayed-429-b", "openai", success.server.URL+"/v1", 20, baseModelID, "stream-delayed-success", "text_generate", 1, "", nil)

		startedAt := time.Now()
		status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{
			"model": model, "stream": true, "messages": []any{map[string]any{"role": "user", "content": "stream delayed 429"}},
		}, nil)
		bodyText := string(body)
		if status != http.StatusOK || failed.calls.Load() != 1 || success.calls.Load() != 1 {
			t.Fatalf("unexpected delayed 429 failover status=%d calls=%d/%d body=%s", status, failed.calls.Load(), success.calls.Load(), body)
		}
		if strings.Count(bodyText, "delayed 429 fallback") != 1 || strings.Contains(bodyText, "controlled delayed 429") || !strings.Contains(bodyText, "[DONE]") {
			t.Fatalf("delayed 429 response leaked or duplicated candidate output: %s", bodyText)
		}

		detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt))
		if len(detail.Attempts) != 2 || detail.Attempts[0].PlatformID != platformA.ID || detail.Attempts[0].ErrorCode != "upstream_rate_limited" || detail.Attempts[1].PlatformID != platformB.ID || detail.Attempts[1].Status != "succeeded" || countFailureEffects(detail, "cooldown") != 1 {
			t.Fatalf("unexpected delayed 429 attempt chain: %+v", detail)
		}
		decisionCount := 0
		for _, attempt := range detail.Attempts {
			trace := append([]map[string]any(nil), attempt.Trace...)
			for _, raw := range anySlice(attempt.DynamicMetrics["trace"]) {
				if entry, ok := raw.(map[string]any); ok {
					trace = append(trace, entry)
				}
			}
			for _, entry := range trace {
				if entry["event"] == "failure_decision" && entry["route"] == "next" && entry["effect"] == "cooldown" && entry["category"] == "rate_limit" {
					decisionCount++
				}
			}
		}
		if decisionCount != 1 {
			t.Fatalf("delayed 429 should record one cooldown-and-next decision, got %d: %+v", decisionCount, detail)
		}
		var modelCooling bool
		if err := f.pool.QueryRow(f.ctx, `SELECT COALESCE(cooldown_until > now(), false) FROM platform_models WHERE id=$1::uuid`, modelA.ID).Scan(&modelCooling); err != nil {
			t.Fatalf("read delayed 429 cooldown state: %v", err)
		}
		if !modelCooling {
			t.Fatal("delayed 429 failed platform model was not cooled")
		}
	})

	t.Run("首个 delta 前允许轮转", func(t *testing.T) {
		model := "acceptance-stream-before-" + f.suffix
		baseModelID := f.createBaseModel(t, model, "text_generate")
		failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
			writeStubJSON(w, http.StatusInternalServerError, map[string]any{"error": map[string]any{"message": "before delta"}})
		})
		defer failed.close()
		success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
			w.Header().Set("Content-Type", "text/event-stream")
			_, _ = io.WriteString(w, "data: {\"id\":\"chunk-ok\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"stream fallback\"},\"finish_reason\":null}]}\n\n")
			_, _ = io.WriteString(w, "data: {\"id\":\"chunk-ok\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n")
			_, _ = io.WriteString(w, "data: [DONE]\n\n")
		})
		defer success.close()
		_, _ = f.createPlatformModel(t, "stream-before-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "stream-failed", "text_generate", 1, "", nil)
		_, _ = f.createPlatformModel(t, "stream-before-b", "openai", success.server.URL+"/v1", 20, baseModelID, "stream-success", "text_generate", 1, "", nil)

		status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{
			"model": model, "stream": true, "messages": []any{map[string]any{"role": "user", "content": "stream"}},
		}, nil)
		if status != http.StatusOK || failed.calls.Load() != 1 || success.calls.Load() != 1 || strings.Count(string(body), "stream fallback") != 1 || !strings.Contains(string(body), "[DONE]") {
			t.Fatalf("unexpected pre-delta failover status=%d calls=%d/%d body=%s", status, failed.calls.Load(), success.calls.Load(), body)
		}
	})

	t.Run("已输出 delta 后禁止轮转", func(t *testing.T) {
		model := "acceptance-stream-after-" + f.suffix
		baseModelID := f.createBaseModel(t, model, "text_generate")
		broken := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) {
			hijacker, ok := w.(http.Hijacker)
			if !ok {
				http.Error(w, "hijacking unavailable", http.StatusInternalServerError)
				return
			}
			connection, buffer, err := hijacker.Hijack()
			if err != nil {
				return
			}
			defer connection.Close()
			_, _ = buffer.WriteString("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: 4096\r\n\r\n")
			_, _ = buffer.WriteString("data: {\"id\":\"chunk-broken\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"first platform only\"},\"finish_reason\":null}]}\n\n")
			_ = buffer.Flush()
		})
		defer broken.close()
		fallback := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "must not mix") })
		defer fallback.close()
		_, _ = f.createPlatformModel(t, "stream-after-a", "openai", broken.server.URL+"/v1", 10, baseModelID, "stream-broken", "text_generate", 2, "", nil)
		_, _ = f.createPlatformModel(t, "stream-after-b", "openai", fallback.server.URL+"/v1", 20, baseModelID, "stream-fallback", "text_generate", 1, "", nil)

		status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{
			"model": model, "stream": true, "messages": []any{map[string]any{"role": "user", "content": "stream"}},
		}, nil)
		if status != http.StatusOK || broken.calls.Load() != 1 || fallback.calls.Load() != 0 {
			t.Fatalf("post-delta failure rotated unexpectedly status=%d calls=%d/%d body=%s", status, broken.calls.Load(), fallback.calls.Load(), body)
		}
		bodyText := string(body)
		if !strings.Contains(bodyText, "first platform only") || strings.Contains(bodyText, "must not mix") || strings.Count(bodyText, "event: error") != 1 {
			t.Fatalf("stream response mixed providers or missed terminal error: %s", bodyText)
		}
	})
}

func (f *failurePolicyAcceptanceFixture) testRunnerPolicyHTTPValidation(t *testing.T) {
	valid := standardFailureRunnerPolicy(true)
	var saved map[string]any
	doJSON(t, f.server.URL, http.MethodPatch, "/api/admin/runtime/runner-policy", f.adminToken, valid, http.StatusOK, &saved)
	var before map[string]any
	doJSON(t, f.server.URL, http.MethodGet, "/api/admin/runtime/runner-policy", f.adminToken, nil, http.StatusOK, &before)

	invalidRules := []map[string]any{
		{"categories": []any{"rate_limit"}, "action": "rotate"},
		{"categories": []any{"rate_limit"}, "action": "next", "target": "client"},
		{"action": "next"},
		{"statusCodes": []any{429}, "action": "cooldown_and_next", "target": "model", "cooldownSeconds": 0},
		{"statusCodes": []any{500}, "action": "demote_and_next", "target": "platform", "demoteSteps": -1},
	}
	for index, rule := range invalidRules {
		payload := standardFailureRunnerPolicy(true)
		payload["failoverPolicy"].(map[string]any)["actionRules"] = []any{rule}
		raw, _ := json.Marshal(payload)
		request, _ := http.NewRequest(http.MethodPatch, f.server.URL+"/api/admin/runtime/runner-policy", bytes.NewReader(raw))
		request.Header.Set("Authorization", "Bearer "+f.adminToken)
		request.Header.Set("Content-Type", "application/json")
		response, err := http.DefaultClient.Do(request)
		if err != nil {
			t.Fatalf("invalid policy request %d: %v", index, err)
		}
		body, _ := io.ReadAll(response.Body)
		response.Body.Close()
		if response.StatusCode != http.StatusBadRequest {
			t.Fatalf("invalid policy %d status=%d body=%s", index, response.StatusCode, body)
		}
	}
	var after map[string]any
	doJSON(t, f.server.URL, http.MethodGet, "/api/admin/runtime/runner-policy", f.adminToken, nil, http.StatusOK, &after)
	beforeFailover, _ := before["failoverPolicy"].(map[string]any)
	afterFailover, _ := after["failoverPolicy"].(map[string]any)
	if !jsonObjectsEqual(beforeFailover["actionRules"], afterFailover["actionRules"]) {
		t.Fatalf("invalid PATCH overwrote runner action rules: before=%+v after=%+v", beforeFailover["actionRules"], afterFailover["actionRules"])
	}
}

func jsonObjectsEqual(left any, right any) bool {
	leftRaw, _ := json.Marshal(left)
	rightRaw, _ := json.Marshal(right)
	return bytes.Equal(leftRaw, rightRaw)
}
