package wire

// Pre-stream HTTP retry for the openai-compat path. These tests drive the real
// retryer.Do loop through a live httptest server and COUNT actual HTTP attempts
// -- a classification-only assertion would go green even if nothing retried.
// See Stream / doStreamOnce.
//
// openai-compat 路径的 pre-stream HTTP 重试. 这些测试经真 httptest server 驱动
// 真实 retryer.Do 循环并**数真实 HTTP 尝试次数** -- 只断言分类会绿即使没重试.

import (
	"context"
	"errors"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"
	"time"

	"git.flytoex.net/yuanwei/flyto-agent/core/internal/transport/retry"
	"git.flytoex.net/yuanwei/flyto-agent/core/pkg/flyto"
)

// fastRetryPolicy mirrors the default composite policy but with a near-zero
// backoff so the tests do not sleep. BaseDelay must be non-zero: ExponentialBackoff
// treats a zero BaseDelay as "unset" and falls back to 500ms, which would make
// the exhaustion test take seconds. maxRetries is explicit so the exhaustion
// count is deterministic.
//
// fastRetryPolicy 镜像默认 composite 策略但近零退避, 测试不睡眠. BaseDelay 必须
// 非零: ExponentialBackoff 把零 BaseDelay 当"未设"回退到 500ms, 会让耗尽测试拖
// 数秒. maxRetries 显式给定使耗尽次数确定.
func fastRetryPolicy(maxRetries int) retry.RetryPolicy {
	return retry.NewCompositeRetryPolicy(
		&retry.ForegroundOnly{},
		&retry.ServerDirective{},
		&retry.ExponentialBackoff{BaseDelay: time.Nanosecond, MaxDelay: time.Millisecond, MaxRetries: maxRetries},
	)
}

const okSSEBody = "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n" +
	"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1}}\n\n" +
	"data: [DONE]\n\n"

func retryTestRequest() *StreamRequest {
	return &StreamRequest{
		Model:    "test-model",
		Messages: []flyto.Message{{Role: flyto.RoleUser, Blocks: []flyto.Block{flyto.TextBlock("hi")}}},
	}
}

// TestOpenAICompatClient_Retries5xxThenSucceeds: first two attempts return 503,
// third returns 200 SSE. The run must retry twice and finally succeed -- proving
// the pre-stream retry loop is wired (not a no-op) and that a 200 stops retrying.
func TestOpenAICompatClient_Retries5xxThenSucceeds(t *testing.T) {
	var calls atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		n := calls.Add(1)
		if int(n) <= 2 {
			w.WriteHeader(http.StatusServiceUnavailable)
			_, _ = w.Write([]byte(`{"error":{"message":"upstream boom"}}`))
			return
		}
		w.Header().Set("Content-Type", "text/event-stream")
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte(okSSEBody))
	}))
	defer srv.Close()

	c := NewOpenAICompatClient("k", srv.URL, WithRetryPolicy(fastRetryPolicy(4)))
	ch, err := c.Stream(context.Background(), retryTestRequest())
	if err != nil {
		t.Fatalf("expected success after 2 retries, got error: %v", err)
	}
	for range ch { // drain to let the SSE goroutine finish
	}
	if got := calls.Load(); got != 3 {
		t.Errorf("expected 3 HTTP attempts (2 retries + success), got %d", got)
	}
}

// TestOpenAICompatClient_4xxNoRetry: a 400 is terminal (not retryable). The run
// must fail on the first attempt with a typed ErrProviderHTTPStatus (ADR-0006
// fail-loud preserved through toEngineError) and NOT retry.
func TestOpenAICompatClient_4xxNoRetry(t *testing.T) {
	var calls atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		calls.Add(1)
		w.WriteHeader(http.StatusBadRequest)
		_, _ = w.Write([]byte(`{"error":{"message":"bad request"}}`))
	}))
	defer srv.Close()

	c := NewOpenAICompatClient("k", srv.URL, WithRetryPolicy(fastRetryPolicy(4)))
	_, err := c.Stream(context.Background(), retryTestRequest())
	if err == nil {
		t.Fatal("expected error on 400, got nil")
	}
	var engErr *flyto.EngineError
	if !errors.As(err, &engErr) {
		t.Fatalf("expected *flyto.EngineError, got %T: %v", err, err)
	}
	if engErr.Code != flyto.ErrProviderHTTPStatus {
		t.Errorf("expected code %q, got %q", flyto.ErrProviderHTTPStatus, engErr.Code)
	}
	if got := calls.Load(); got != 1 {
		t.Errorf("expected 1 attempt (no retry on 4xx), got %d", got)
	}
}

// TestOpenAICompatClient_RetryExhausted: a persistent 503 must retry up to the
// limit then fail loud. With MaxRetries=4 the total is 5 attempts (1 initial +
// 4 retries). Confirms the loop terminates and reports a typed error.
func TestOpenAICompatClient_RetryExhausted(t *testing.T) {
	var calls atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		calls.Add(1)
		w.WriteHeader(http.StatusServiceUnavailable)
		_, _ = w.Write([]byte(`{"error":{"message":"still down"}}`))
	}))
	defer srv.Close()

	c := NewOpenAICompatClient("k", srv.URL, WithRetryPolicy(fastRetryPolicy(4)))
	_, err := c.Stream(context.Background(), retryTestRequest())
	if err == nil {
		t.Fatal("expected error after retries exhausted, got nil")
	}
	var engErr *flyto.EngineError
	if !errors.As(err, &engErr) {
		t.Fatalf("expected *flyto.EngineError, got %T: %v", err, err)
	}
	if got := calls.Load(); got != 5 {
		t.Errorf("expected 5 attempts (1 + 4 retries), got %d", got)
	}
}

// TestOpenAICompatClient_200NonSSENoRetry: some providers (e.g. MiniMax on auth
// failure) return 200 + a JSON error body instead of an SSE stream. That is a
// terminal protocol error, not transient -- doStreamOnce returns a plain
// *flyto.EngineError (ErrProviderNonSSE), which is NOT a retry.RetryError, so
// retryer.Do surfaces it unretried. Guards this live path's reroute through the
// retry loop against a future refactor that might make it retryable.
func TestOpenAICompatClient_200NonSSENoRetry(t *testing.T) {
	var calls atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		calls.Add(1)
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte(`{"base_resp":{"status_code":2049,"status_msg":"invalid api key"}}`))
	}))
	defer srv.Close()

	c := NewOpenAICompatClient("k", srv.URL, WithRetryPolicy(fastRetryPolicy(4)))
	_, err := c.Stream(context.Background(), retryTestRequest())
	if err == nil {
		t.Fatal("expected error on 200-non-SSE, got nil")
	}
	var engErr *flyto.EngineError
	if !errors.As(err, &engErr) {
		t.Fatalf("expected *flyto.EngineError, got %T: %v", err, err)
	}
	if engErr.Code != flyto.ErrProviderNonSSE {
		t.Errorf("expected code %q, got %q", flyto.ErrProviderNonSSE, engErr.Code)
	}
	if got := calls.Load(); got != 1 {
		t.Errorf("expected 1 attempt (no retry on 200-non-SSE), got %d", got)
	}
}

// TestOpenAICompatClient_MidStreamRetryableSurfaces: a mid-stream error chunk
// (200 handshake OK, then SSE {"error":{"code":429}}) must surface as an
// ErrorEvent carrying Retryable=true through the full consumeSSE + StreamGuard
// path. This is the write+transport half of gap B (engine reads ev.Retryable to
// retry mid-stream): it proves the field survives to the channel the engine
// consumes -- StreamGuard could have dropped it (Gate 1). Without this guard the
// engine-side gap-B branch could be a silent no-op.
func TestOpenAICompatClient_MidStreamRetryableSurfaces(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		w.WriteHeader(http.StatusOK)
		// 200 handshake OK, then a mid-stream 429 error chunk before any content.
		_, _ = w.Write([]byte("data: {\"error\":{\"code\":429,\"message\":\"rate limited mid-stream\"}}\n\n"))
	}))
	defer srv.Close()

	c := NewOpenAICompatClient("k", srv.URL)
	ch, err := c.Stream(context.Background(), retryTestRequest())
	if err != nil {
		t.Fatalf("Stream should not error on 200 handshake: %v", err)
	}
	var sawErrorEvent, gotRetryable bool
	for ev := range ch {
		if ee, ok := ev.(*flyto.ErrorEvent); ok {
			sawErrorEvent = true
			gotRetryable = ee.Retryable
		}
	}
	if !sawErrorEvent {
		t.Fatal("expected an ErrorEvent from the mid-stream 429 chunk")
	}
	if !gotRetryable {
		t.Error("mid-stream 429 must carry Retryable=true through consumeSSE+StreamGuard (else engine gap-B retry is a no-op)")
	}
}