package ws

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"io"
	"log/slog"
	"strings"
	"sync"
	"testing"
	"time"

	"fiko-ai/internal/agent"
	"fiko-ai/internal/conversation"
	"fiko-ai/internal/llm"
	"fiko-ai/internal/protocol"
	"fiko-ai/internal/stt"
)

// textStream, sabit metin üreten akış sağlar.
func textStream(s string) streamFn {
	return func(_ context.Context, _ llm.Request, on llm.DeltaFunc) (llm.Result, error) {
		if on != nil {
			if err := on(s); err != nil {
				return llm.Result{}, err
			}
		}
		return llm.Result{Content: s}, nil
	}
}

// ---- sahte LLM ----

type streamFn func(ctx context.Context, req llm.Request, on llm.DeltaFunc) (llm.Result, error)

type fakeLLM struct {
	stream streamFn
	mu     sync.Mutex
	calls  int
}

var _ llm.Provider = (*fakeLLM)(nil)

func (f *fakeLLM) Name() string { return "fake" }

func (f *fakeLLM) Generate(context.Context, llm.Request) (llm.Result, error) {
	return llm.Result{}, errors.New("kullanılmıyor")
}

func (f *fakeLLM) ListModels(context.Context) ([]llm.ModelInfo, error) { return nil, nil }

func (f *fakeLLM) GenerateStream(ctx context.Context, req llm.Request, on llm.DeltaFunc) (llm.Result, error) {
	f.mu.Lock()
	f.calls++
	f.mu.Unlock()
	return f.stream(ctx, req, on)
}

func (f *fakeLLM) callCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.calls
}

// ---- sahte bağlantı ----

// fakeConn Conn arayüzünü kanal üzerinden uygular.
// Yazma/ping hataları enjekte edilebilir (yavaş/ölü istemci senaryoları için).
type fakeConn struct {
	in  chan []byte // istemciden sunucuya
	out chan []byte // sunucudan istemciye

	pings  chan struct{}
	writes chan []byte

	writeErr error
	pingErr  error

	closeOnce sync.Once
	closed    chan struct{}

	mu        sync.Mutex
	pongH     func()
	deadlines int
}

var _ Conn = (*fakeConn)(nil)

func newFakeConn() *fakeConn {
	return &fakeConn{
		in:       make(chan []byte, 64),
		out:      make(chan []byte, 64),
		pings:    make(chan struct{}, 8),
		writes:   make(chan []byte, 256),
		closed:   make(chan struct{}),
		pongH:    func() {},
		writeErr: nil,
		pingErr:  nil,
	}
}

func (c *fakeConn) Read() ([]byte, error) {
	select {
	case b, ok := <-c.in:
		if !ok {
			return nil, io.EOF
		}
		return b, nil
	case <-c.closed:
		return nil, errors.New("bağlantı kapalı")
	}
}

func (c *fakeConn) Write(data []byte, _ time.Time) error {
	if c.writeErr != nil {
		return c.writeErr
	}
	select {
	case c.writes <- append([]byte(nil), data...):
	default:
	}
	return nil
}

func (c *fakeConn) Ping(time.Time) error {
	if c.pingErr != nil {
		return c.pingErr
	}
	select {
	case c.pings <- struct{}{}:
	default:
	}
	return nil
}

func (c *fakeConn) SetReadDeadline(time.Time) error {
	c.mu.Lock()
	c.deadlines++
	c.mu.Unlock()
	return nil
}

func (c *fakeConn) SetPongHandler(h func()) {
	c.mu.Lock()
	c.pongH = h
	c.mu.Unlock()
}

// readDeadlineCount bağlantıya kaç kez okuma sonu verildiğini döndürür.
func (c *fakeConn) readDeadlineCount() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.deadlines
}

func (c *fakeConn) pong() {
	c.mu.Lock()
	h := c.pongH
	c.mu.Unlock()
	h()
}

func (c *fakeConn) Close() error {
	c.closeOnce.Do(func() { close(c.closed) })
	return nil
}

func (c *fakeConn) isClosed() bool {
	select {
	case <-c.closed:
		return true
	default:
		return false
	}
}

// send istemci → sunucu yönünde ham çerçeve yazar.
func (c *fakeConn) send(t *testing.T, v any) {
	t.Helper()
	var raw []byte
	switch x := v.(type) {
	case string:
		raw = []byte(x)
	default:
		b, err := json.Marshal(x)
		if err != nil {
			t.Fatalf("zarf serileştirilemedi: %v", err)
		}
		raw = b
	}
	select {
	case c.in <- raw:
	case <-time.After(2 * time.Second):
		t.Fatalf("istemci çerçevesi gönderilemedi (okuma döngüsü tıkandı mı?)")
	}
}

func (c *fakeConn) inbound(id, typ, conv string, payload any) any {
	in := protocol.Inbound{Type: protocol.ClientType(typ), ID: id, ConversationID: conv}
	if payload != nil {
		raw, _ := json.Marshal(payload)
		in.Payload = raw
	}
	return in
}

// next sunucudan gelen sıradaki olayı bekler (yazma kuyruğundan).
func (c *fakeConn) next(t *testing.T) protocol.Outbound {
	t.Helper()
	return c.decode(t, c.recv(t))
}

// recv yazma kuyruğundan sıradaki ham çerçeveyi alır.
func (c *fakeConn) recv(t *testing.T) []byte {
	t.Helper()
	select {
	case b := <-c.writes:
		return b
	case <-time.After(3 * time.Second):
		t.Fatal("zaman aşımı: sunucudan olay gelmedi")
		return nil
	}
}

// decode çerçeveyi Outbound'a çevirir. DİKKAT: WS sınırında payload JSON'dan geçtiği için
// Go tipine değil map[string]any'e iner. Bu yüzden payload alanları T üzerinden okunmalıdır.
func (c *fakeConn) decode(t *testing.T, b []byte) protocol.Outbound {
	t.Helper()
	var o protocol.Outbound
	if err := json.Unmarshal(b, &o); err != nil {
		t.Fatalf("giden olay çözümlenemedi: %v (çerçeve: %s)", err, b)
	}
	return o
}

// payload alanını T tipine çevirir (yoksa test hatası verir).
func payloadAs[T any](t *testing.T, o protocol.Outbound) T {
	t.Helper()
	var v T
	raw, err := json.Marshal(o.Payload)
	if err != nil {
		t.Fatalf("payload serileştirilemedi: %v", err)
	}
	if err := json.Unmarshal(raw, &v); err != nil {
		t.Fatalf("payload %T tipine çevrilemedi: %v", v, err)
	}
	return v
}

// nextOf t türünde bir olayı bekler; araya karışan diğer olayları atar.
func (c *fakeConn) nextOf(t *testing.T, typ protocol.ServerType) protocol.Outbound {
	t.Helper()
	deadline := time.After(3 * time.Second)
	for {
		select {
		case b := <-c.writes:
			o := c.decode(t, b)
			if o.Type == typ {
				return o
			}
		case <-deadline:
			t.Fatalf("zaman aşımı: %s olayı gelmedi", typ)
			return protocol.Outbound{}
		}
	}
}

// ---- test kurulumu ----

// fixture hub + ajan + konuşma + bağlantı üretir; bağlantı zaten bağlı (Serve) gelir.
type fixture struct {
	hub  *Hub
	ag   *agent.Agent
	svc  *conversation.Service
	conv string
	conn *fakeConn
	llm  *fakeLLM
	done chan struct{}
}

// testLogger test çıktısını çöpe atar.
// ---- ses akışı (Phase 6) ----

// fakeSTT ses verisini sabit metne çevirir.
type fakeSTT struct {
	text  string
	err   error
	got   []byte
	calls int
}

func (f *fakeSTT) Transcribe(_ context.Context, audio []byte, _ string) (string, error) {
	f.calls++
	f.got = append([]byte(nil), audio...)
	return f.text, f.err
}

// newVoiceFixture ses bağlı hub kurar.
func newVoiceFixture(t *testing.T, f *fakeLLM, rec stt.Recognizer) *fixture {
	t.Helper()
	svc := conversation.NewService(conversation.NewMemoryRepo())
	c, err := svc.Create(context.Background(), "u")
	if err != nil {
		t.Fatal(err)
	}
	log := testLogger()
	ag := agent.New(svc, f, nil, nil, log, agent.Config{})
	var svc2 *stt.Service
	if rec != nil {
		svc2 = stt.New(rec, stt.Limits{})
	}
	hub := NewHubWithSTT(ag, svc2, log, Options{})
	conn := newFakeConn()
	fx := &fixture{hub: hub, ag: ag, svc: svc, conv: c.ID, conn: conn, llm: f, done: make(chan struct{})}
	go func() { defer close(fx.done); hub.Serve(context.Background(), conn, "u") }()
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		_ = ag.Close(ctx)
		_ = hub.Close(ctx)
		select {
		case <-fx.done:
		case <-time.After(3 * time.Second):
		}
	})
	return fx
}

// waitForType, verilen türde bir olay gelene kadar bekler.
func waitForType(t *testing.T, fx *fixture, want protocol.ServerType) protocol.Outbound {
	t.Helper()
	deadline := time.After(5 * time.Second)
	for {
		select {
		case b := <-fx.conn.writes:
			var o protocol.Outbound
			if err := json.Unmarshal(b, &o); err != nil {
				continue
			}
			if o.Type == want {
				return o
			}
		case <-deadline:
			t.Fatalf("%s olayı gelmedi", want)
		}
	}
}

func TestVoice_StopTranscribesAndSends(t *testing.T) {
	f := &fakeLLM{stream: textStream("merhaba dedin")}
	rec := &fakeSTT{text: "merhaba"}
	fx := newVoiceFixture(t, f, rec)

	fx.conn.in <- inbound(t, "v1", protocol.ClientVoiceStart,
		protocol.VoiceStartPayload{ConversationID: fx.conv, Mime: "audio/webm"})
	fx.conn.in <- inbound(t, "v2", protocol.ClientVoiceAudio,
		protocol.VoiceAudioPayload{ConversationID: fx.conv, Data: base64.StdEncoding.EncodeToString([]byte("AUDIO"))})
	fx.conn.in <- inbound(t, "v3", protocol.ClientVoiceStop,
		protocol.VoiceStopPayload{ConversationID: fx.conv})

	o := waitForType(t, fx, protocol.ServerVoiceDetected)
	p := payloadAs[protocol.VoiceDetectedPayload](t, o)
	if p.Text != "merhaba" || !p.Sent {
		t.Errorf("tespit edilen metin: %+v", p)
	}
	if string(rec.got) != "AUDIO" {
		t.Errorf("ses verisi motora gitmedi: %q", rec.got)
	}
	// Tanınan metin normal mesaj gibi işlenmeli: asistan cevabı gelmeli.
	waitForType(t, fx, protocol.ServerAssistantTextFinal)
}

func TestVoice_EmptyTranscriptIsNotSent(t *testing.T) {
	// Motor boş döndürürse BOŞ mesaj gönderilmez; yalnızca teslim olayı gider.
	f := &fakeLLM{stream: textStream("x")}
	rec := &fakeSTT{text: "   "}
	fx := newVoiceFixture(t, f, rec)

	fx.conn.in <- inbound(t, "v1", protocol.ClientVoiceStart,
		protocol.VoiceStartPayload{ConversationID: fx.conv, Mime: "audio/webm"})
	fx.conn.in <- inbound(t, "v2", protocol.ClientVoiceAudio,
		protocol.VoiceAudioPayload{ConversationID: fx.conv, Data: base64.StdEncoding.EncodeToString([]byte("A"))})
	fx.conn.in <- inbound(t, "v3", protocol.ClientVoiceStop, protocol.VoiceStopPayload{ConversationID: fx.conv})

	o := waitForType(t, fx, protocol.ServerVoiceDetected)
	p := payloadAs[protocol.VoiceDetectedPayload](t, o)
	if p.Text != "" || p.Sent {
		t.Errorf("boş transcript gönderilmemeliydi: %+v", p)
	}
}

func TestVoice_NotConfigured(t *testing.T) {
	// Motor yoksa sessizce yutulmaz: açık hata döner.
	f := &fakeLLM{stream: textStream("x")}
	fx := newVoiceFixture(t, f, nil)

	fx.conn.in <- inbound(t, "v1", protocol.ClientVoiceStart,
		protocol.VoiceStartPayload{ConversationID: fx.conv, Mime: "audio/webm"})

	o := waitForType(t, fx, protocol.ServerError)
	p := payloadAs[protocol.ErrorPayload](t, o)
	if p.Code != "not_implemented" {
		t.Errorf("beklenmeyen hata kodu: %+v", p)
	}
}

func TestVoice_BadBase64Rejected(t *testing.T) {
	f := &fakeLLM{stream: textStream("x")}
	fx := newVoiceFixture(t, f, &fakeSTT{text: "x"})

	fx.conn.in <- inbound(t, "v1", protocol.ClientVoiceStart,
		protocol.VoiceStartPayload{ConversationID: fx.conv, Mime: "audio/webm"})
	fx.conn.in <- inbound(t, "v2", protocol.ClientVoiceAudio,
		protocol.VoiceAudioPayload{ConversationID: fx.conv, Data: "bu base64 değil!!"})

	o := waitForType(t, fx, protocol.ServerError)
	p := payloadAs[protocol.ErrorPayload](t, o)
	if p.Code != protocol.ErrCodeBadRequest {
		t.Errorf("bozuk base64 reddedilmeliydi: %+v", p)
	}
}

// inbound, test karesini üretir.
func inbound(t *testing.T, id string, typ protocol.ClientType, payload any) []byte {
	t.Helper()
	raw, err := json.Marshal(payload)
	if err != nil {
		t.Fatal(err)
	}
	b, err := json.Marshal(protocol.Inbound{Type: typ, ID: id, Payload: raw})
	if err != nil {
		t.Fatal(err)
	}
	return b
}

func testLogger() *slog.Logger {
	return slog.New(slog.NewTextHandler(io.Discard, nil))
}

// newFixture ajanı ve hub'ı kurar; Serve ayrı goroutine'de çalışır.
func newFixture(t *testing.T, f *fakeLLM, opts Options) *fixture {
	t.Helper()
	svc := conversation.NewService(conversation.NewMemoryRepo())
	c, err := svc.Create(context.Background(), "u")
	if err != nil {
		t.Fatalf("konuşma oluşturulamadı: %v", err)
	}
	log := testLogger()
	ag := agent.New(svc, f, nil, nil, log, agent.Config{})
	hub := NewHub(ag, log, opts)
	conn := newFakeConn()
	fx := &fixture{hub: hub, ag: ag, svc: svc, conv: c.ID, conn: conn, llm: f, done: make(chan struct{})}
	go func() { defer close(fx.done); hub.Serve(context.Background(), conn, "u") }()
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		_ = ag.Close(ctx)
		_ = hub.Close(ctx)
		select {
		case <-fx.done:
		case <-time.After(3 * time.Second):
		}
	})
	return fx
}

// defaultOpts testlerde hızlı tepki veren ama macera gerektirmeyen ayarlar.
func defaultOpts() Options {
	return Options{
		PingInterval:   50 * time.Millisecond,
		PongWait:       2 * time.Second,
		WriteWait:      time.Second,
		OutboundBuffer: 8,
		MaxConcurrent:  2,
	}
}

// echoStream iki delta yollar ve tamamlanır.
func echoStream(text string) streamFn {
	return func(_ context.Context, _ llm.Request, on llm.DeltaFunc) (llm.Result, error) {
		_ = on(text[:1])
		_ = on(text[1:])
		return llm.Result{Model: "m1", Content: text}, nil
	}
}

// blockingStream bir delta yollar, started'ı kapatır ve iptal/serbest bekler.
func blockingStream(started chan struct{}, release <-chan struct{}, partial string) streamFn {
	return func(ctx context.Context, _ llm.Request, on llm.DeltaFunc) (llm.Result, error) {
		_ = on(partial)
		close(started)
		select {
		case <-release:
			return llm.Result{Model: "m", Content: partial}, nil
		case <-ctx.Done():
			return llm.Result{Model: "m", Content: partial}, ctx.Err()
		}
	}
}

// ---- testler ----

// İstemci ping → sunucu pong, replyTo korunur. Okuma sonu verilmiş olmalıdır.
func TestSession_PingPong(t *testing.T) {
	fx := newFixture(t, &fakeLLM{stream: echoStream("x")}, defaultOpts())
	fx.conn.send(t, fx.conn.inbound("r1", string(protocol.ClientPing), fx.conv, nil))

	o := fx.conn.nextOf(t, protocol.ServerPong)
	if o.ReplyTo != "r1" || o.ConversationID != fx.conv {
		t.Fatalf("pong alanları: %+v", o)
	}
	if n := fx.conn.readDeadlineCount(); n < 1 {
		t.Error("okuma sonu verilmemiş olmalı (heartbeat için)")
	}
}

// Bu bağlantının başlattığı üretim conversation.cancel ile kesilir; kısmi içerik kaydedilir.
func TestSession_CancelByOwner(t *testing.T) {
	started := make(chan struct{})
	fx := newFixture(t, &fakeLLM{stream: blockingStream(started, nil, "yarım")}, defaultOpts())

	fx.conn.send(t, fx.conn.inbound("r1", string(protocol.ClientTextMessage), fx.conv,
		protocol.TextMessagePayload{Text: "uzun iş"}))
	fx.conn.nextOf(t, protocol.ServerAssistantThinking) // üretim başladı

	fx.conn.send(t, fx.conn.inbound("r2", string(protocol.ClientConversationCancel), fx.conv,
		protocol.CancelPayload{Reason: "barge-in"}))
	intr := fx.conn.nextOf(t, protocol.ServerAssistantInterrupted)
	if p := payloadAs[protocol.InterruptedPayload](t, intr); p.Reason != "barge-in" || p.Partial != "yarım" {
		t.Fatalf("interrupted: %+v", p)
	}
	hist, _ := fx.svc.Messages(context.Background(), "u", fx.conv, 10)
	if len(hist) != 2 || hist[1].Status != conversation.StatusInterrupted || hist[1].Content != "yarım" {
		t.Fatalf("kısmi içerik interrupted olarak kaydedilmeli: %+v", hist)
	}
}

// conversationId yoksa 400; sahibi olmayan konuşma sessizce yoksayılır (olay GÖNDERİLMEZ).
func TestSession_CancelOwnership(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	fx := newFixture(t, &fakeLLM{stream: blockingStream(started, release, "yarım")}, defaultOpts())
	fx.conn.send(t, fx.conn.inbound("r1", string(protocol.ClientTextMessage), fx.conv,
		protocol.TextMessagePayload{Text: "iş"}))
	fx.conn.nextOf(t, protocol.ServerAssistantThinking)

	// 1) conversationId yok → bad_request
	fx.conn.send(t, fx.conn.inbound("r2", string(protocol.ClientConversationCancel), "", nil))
	e := fx.conn.nextOf(t, protocol.ServerError)
	if p := payloadAs[protocol.ErrorPayload](t, e); p.Code != protocol.ErrCodeBadRequest {
		t.Fatalf("konuşmasız iptal kodu = %q", p.Code)
	}

	// 2) bu bağlantının sahibi olmadığı konuşma → sessizce yok sayılır.
	fx.conn.send(t, fx.conn.inbound("r3", string(protocol.ClientConversationCancel), "conv_baska", nil))
	select {
	case o := <-fx.conn.writes:
		t.Fatalf("sahipli olmayan konuşma için olay gönderilmemeliydi, gelen: %s", o)
	case <-time.After(200 * time.Millisecond):
	}
	if fx.conn.isClosed() {
		t.Fatal("sahiplik dışı iptal bağlantıyı kapatmamalı")
	}
	// Üretim hâlâ akıyor olmalı: sahiplik dışı iptal hiçbir şeyi durdurmaz.
	// (Kontrollü olarak bırakılır: ajan.Close ve hub.Close t.Cleanup'ta bekler.)
}

// Bağlantı kopunca üretim iptal edilir ve kısmi içerik "disconnect" sebebiyle kaydedilir.
func TestSession_DisconnectCancelsGeneration(t *testing.T) {
	started := make(chan struct{})
	fx := newFixture(t, &fakeLLM{stream: blockingStream(started, nil, "yarım")}, defaultOpts())

	fx.conn.send(t, fx.conn.inbound("r1", string(protocol.ClientTextMessage), fx.conv,
		protocol.TextMessagePayload{Text: "iş"}))
	fx.conn.nextOf(t, protocol.ServerAssistantThinking)

	// Serve, Read hatasıyla döner: gelen kanalı kapatmak bağlantı düşmesini simüler.
	close(fx.conn.in)

	select {
	case <-fx.done:
	case <-time.After(3 * time.Second):
		t.Fatal("zaman aşımı: Serve bağlantı düşünce dönmeli")
	}
	deadline := time.Now().Add(2 * time.Second)
	for fx.hub.Count() != 0 && time.Now().Before(deadline) {
		time.Sleep(10 * time.Millisecond)
	}
	if fx.hub.Count() != 0 {
		t.Errorf("oturum kayıtlı kaldı: %d", fx.hub.Count())
	}
	// Üretim iptal edildi → kısmi içerik kaydedilmiş olmalı.
	hist, _ := fx.svc.Messages(context.Background(), "u", fx.conv, 10)
	if len(hist) != 2 || hist[1].Status != conversation.StatusInterrupted {
		t.Fatalf("kısmi içerik kaydedilmeliydi: %+v", hist)
	}
}

// Giden kuyruk dolunca bağlantı kapatılır (backpressure: delta kaybetmektense kop).
func TestSession_SlowClientClosesOnFullQueue(t *testing.T) {
	opts := defaultOpts()
	opts.OutboundBuffer = 1
	opts.MaxConcurrent = 1
	fx := newFixture(t, &fakeLLM{stream: floodStream(500)}, opts)
	// Yazma tarafı tüketilmiyor: kanal dolu kalır, sunucu "yavaş istemci" saymalı.
	// (writes kanalı 256 kapasiteli; tüketici yok → dolu.)

	fx.conn.send(t, fx.conn.inbound("r1", string(protocol.ClientTextMessage), fx.conv,
		protocol.TextMessagePayload{Text: "selam"}))

	waitClosed(t, fx.conn)
}

// floodStream çok sayıda delta üretir (kuyruğu doldurur).
func floodStream(n int) streamFn {
	return func(ctx context.Context, _ llm.Request, on llm.DeltaFunc) (llm.Result, error) {
		var sb strings.Builder
		for i := 0; i < n; i++ {
			if err := ctx.Err(); err != nil {
				return llm.Result{Content: sb.String()}, err
			}
			sb.WriteByte('x')
			if err := on("x"); err != nil {
				return llm.Result{Content: sb.String()}, err
			}
		}
		return llm.Result{Content: sb.String()}, nil
	}
}

// Hub.Close: açık oturumları kapatır, Serve döner ve YENİ bağlantıları reddeder.
func TestHub_CloseRejectsNewAndClosesSessions(t *testing.T) {
	svc := conversation.NewService(conversation.NewMemoryRepo())
	log := testLogger()
	ag := agent.New(svc, &fakeLLM{stream: echoStream("x")}, nil, nil, log, agent.Config{})
	hub := NewHub(ag, log, defaultOpts())

	conn := newFakeConn()
	done := make(chan struct{})
	go func() { defer close(done); hub.Serve(context.Background(), conn, "u") }()
	deadline := time.Now().Add(2 * time.Second)
	for hub.Count() == 0 && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	if hub.Count() != 1 {
		t.Fatalf("oturum kaydedilmedi: %d", hub.Count())
	}

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	if err := hub.Close(ctx); err != nil {
		t.Fatalf("Hub.Close: %v", err)
	}
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("Serve, hub kapanınca dönmeli")
	}
	if !conn.isClosed() {
		t.Error("açık oturum kapatılmalı")
	}

	// Kapanmış hub yeni bağlantıyı reddeder.
	rej := newFakeConn()
	hub.Serve(context.Background(), rej, "u")
	if !rej.isClosed() {
		t.Error("kapanmış hub yeni bağlantıyı kabul etmemeli (reddedip kapatmalı)")
	}
}

// Konuşma başına eşzamanlılık sınırı aşılırsa busy hatası (okuma döngüsü bloklanmaz).
func TestSession_ConcurrencyLimitBusy(t *testing.T) {
	opts := defaultOpts()
	opts.MaxConcurrent = 1
	fx := newFixture(t, &fakeLLM{stream: blockingStream(make(chan struct{}), make(chan struct{}), "x")}, opts)

	fx.conn.send(t, fx.conn.inbound("r1", string(protocol.ClientTextMessage), fx.conv,
		protocol.TextMessagePayload{Text: "birinci"}))
	fx.conn.nextOf(t, protocol.ServerAssistantThinking) // sem doldu

	fx.conn.send(t, fx.conn.inbound("r2", string(protocol.ClientTextMessage), fx.conv,
		protocol.TextMessagePayload{Text: "ikinci"}))
	e := fx.conn.nextOf(t, protocol.ServerError)
	if p := payloadAs[protocol.ErrorPayload](t, e); p.Code != protocol.ErrCodeBusy || !p.Retryable {
		t.Fatalf("eşzamanlılık sınırı busy dönmeli: %+v", p)
	}
}

// Konuşma yoksa hata döner ve LLM çağrılmaz (sahiplik ajan katmanında denetlenir).
func TestSession_UnknownConversation(t *testing.T) {
	fx := newFixture(t, &fakeLLM{stream: echoStream("x")}, defaultOpts())
	fx.conn.send(t, fx.conn.inbound("r1", string(protocol.ClientTextMessage), "conv_yok",
		protocol.TextMessagePayload{Text: "selam"}))
	e := fx.conn.nextOf(t, protocol.ServerError)
	if p := payloadAs[protocol.ErrorPayload](t, e); p.Code != protocol.ErrCodeBadRequest {
		t.Fatalf("bilinmeyen konuşma hatası: %+v", p)
	}
	if n := fx.llm.callCount(); n != 0 {
		t.Fatalf("bilinmeyen konuşmada LLM çağrılmamalıydı, çağrı = %d", n)
	}
}

// Yazma hatası bağlantıyı kapatır (yavaş/ölü istemci yönetimi).
// NOT: Hata, ÜRETİM BAŞLADIKTAN SONRA devreye alınır ve akış SÜREKLİ olmalıdır.
// Tek seferlik `echoStream` biter ve yazma döngüsü boşa düşer; bu durumda writeErr
// hiç tetiklenmez ve test niyeti dışında zaman aşımına uğrar (flaky test).
func TestSession_WriteErrorClosesConnection(t *testing.T) {
	fx := newFixture(t, &fakeLLM{stream: neverEndingStream()}, defaultOpts())

	fx.conn.send(t, fx.conn.inbound("r1", string(protocol.ClientTextMessage), fx.conv,
		protocol.TextMessagePayload{Text: "selam"}))
	fx.conn.nextOf(t, protocol.ServerAssistantThinking) // akış sağlıklı

	// Bundan sonra yazma hatası: writeLoop hata görüp oturumu kapatmalı.
	fx.conn.writeErr = errors.New("bağlantı koptu")

	waitClosed(t, fx.conn)
}

// neverEndingStream sürekli delta üretir; bağlantı kapanana kadar sürer (kuyruk hep doludur).
func neverEndingStream() streamFn {
	return func(ctx context.Context, _ llm.Request, on llm.DeltaFunc) (llm.Result, error) {
		for {
			if err := ctx.Err(); err != nil {
				return llm.Result{Content: "x"}, err
			}
			if err := on("x"); err != nil {
				return llm.Result{Content: "x"}, err
			}
			select {
			case <-ctx.Done():
				return llm.Result{Content: "x"}, ctx.Err()
			case <-time.After(2 * time.Millisecond):
			}
		}
	}
}

// Ping düzenli atılır, pong okuma sonunu uzatır, ping hatası bağlantıyı kapatır.
func TestSession_PingAndPingErrorCloses(t *testing.T) {
	fx := newFixture(t, &fakeLLM{stream: echoStream("x")}, defaultOpts())

	select {
	case <-fx.conn.pings:
	case <-time.After(3 * time.Second):
		t.Fatal("zaman aşımı: ping gönderilmedi")
	}
	before := fx.conn.readDeadlineCount()
	fx.conn.pong()
	time.Sleep(50 * time.Millisecond)
	if fx.conn.readDeadlineCount() <= before {
		t.Error("pong okuma sonunu uzatmalıydı")
	}

	fx.conn.pingErr = errors.New("ping başarısız")
	waitClosed(t, fx.conn)
}

func TestSession_InvalidInput(t *testing.T) {
	fx := newFixture(t, &fakeLLM{stream: echoStream("x")}, defaultOpts())

	fx.conn.send(t, "{bozuk json")
	e := fx.conn.nextOf(t, protocol.ServerError)
	if p := payloadAs[protocol.ErrorPayload](t, e); p.Code != protocol.ErrCodeBadRequest {
		t.Fatalf("bozuk JSON kodu = %q, beklenen bad_request", p.Code)
	}

	fx.conn.send(t, fx.conn.inbound("r2", "bilinmeyen.olay", fx.conv, nil))
	e = fx.conn.nextOf(t, protocol.ServerError)
	p := payloadAs[protocol.ErrorPayload](t, e)
	if p.Code != protocol.ErrCodeUnknownEvent || e.ReplyTo != "r2" {
		t.Fatalf("bilinmeyen olay yanıtı: %+v", p)
	}
	if fx.conn.isClosed() {
		t.Fatal("geçersiz girdi bağlantıyı kapatmamalı")
	}
	if n := fx.llm.callCount(); n != 0 {
		t.Fatalf("geçersiz girdide LLM çağrılmamalıydı, çağrı = %d", n)
	}
}

// text.message tam akışı: thinking → delta* → final → state, konuşmaya kaydedilir.
func TestSession_TextMessageFullFlow(t *testing.T) {
	fx := newFixture(t, &fakeLLM{stream: echoStream("Merhaba")}, defaultOpts())

	fx.conn.send(t, fx.conn.inbound("r9", string(protocol.ClientTextMessage), fx.conv,
		protocol.TextMessagePayload{Text: "selam"}))

	think := fx.conn.nextOf(t, protocol.ServerAssistantThinking)
	fx.conn.nextOf(t, protocol.ServerAssistantTextDelta)
	fx.conn.nextOf(t, protocol.ServerAssistantTextDelta)
	fin := fx.conn.nextOf(t, protocol.ServerAssistantTextFinal)

	fp := payloadAs[protocol.TextFinalPayload](t, fin)
	if fp.Content != "Merhaba" || fp.Status != string(conversation.StatusComplete) {
		t.Fatalf("final: %+v", fp)
	}
	if fin.ReplyTo != "r9" || fin.ConversationID != fx.conv {
		t.Fatalf("final alanları: %+v", fin)
	}
	// thinking ile aynı messageId → istemci balonu doğru kapatabilir.
	if payloadAs[protocol.ThinkingPayload](t, think).MessageID != fp.MessageID {
		t.Fatal("thinking ve final messageId uyuşmuyor")
	}

	// Konuşmaya hem kullanıcı hem asistan mesajı yazıldı.
	hist, err := fx.svc.Messages(context.Background(), "u", fx.conv, 10)
	if err != nil {
		t.Fatalf("geçmiş okunamadı: %v", err)
	}
	if len(hist) != 2 || hist[0].Role != conversation.RoleUser || hist[1].Content != "Merhaba" {
		t.Fatalf("kayıt: %+v", hist)
	}
}

// Boş/geçersiz payload reddedilir, LLM çağrılmaz.
func TestSession_TextMessageBadPayload(t *testing.T) {
	fx := newFixture(t, &fakeLLM{stream: echoStream("x")}, defaultOpts())

	fx.conn.send(t, fx.conn.inbound("r1", string(protocol.ClientTextMessage), fx.conv, nil)) // payload yok
	e := fx.conn.nextOf(t, protocol.ServerError)
	if p := payloadAs[protocol.ErrorPayload](t, e); p.Code != protocol.ErrCodeBadRequest {
		t.Fatalf("payload'sız mesaj kodu = %q", p.Code)
	}
	if n := fx.llm.callCount(); n != 0 {
		t.Fatalf("LLM çağrılmamalıydı, çağrı = %d", n)
	}
}

// waitClosed bağlantı kapanana kadar bekler.
func waitClosed(t *testing.T, c *fakeConn) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for !c.isClosed() && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	if !c.isClosed() {
		t.Fatal("zaman aşımı: bağlantı kapanmadı")
	}
}