// Package ws WebSocket oturumlarını yönetir: okuma/yazma döngüleri, heartbeat, geri basınç (backpressure)
// ve istemci olaylarının agent'a yönlendirilmesi. Taşıma katmanından bağımsızdır (bkz. Conn);
// Fiber adaptörü fiber.go içindedir.
//
// Tasarım kararları:
//   - Giden olaylar tamponlu bir kuyruğa yazılır; kuyruk dolarsa istemci "yavaş" sayılır ve bağlantı
//     kapatılır (delta kaybetmektense bağlantıyı koparmak doğrudur; istemci yeniden bağlanıp geçmişi okur).
//   - Bağlantı koparsa o bağlantının başlattığı üretimler iptal edilir (sebep: "disconnect");
//     kısmi içerik agent tarafından "interrupted" olarak kaydedilir.
//   - Bir bağlantı yalnızca kendi başlattığı konuşmaları iptal edebilir.
package ws

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"regexp"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"fiko-ai/internal/agent"
	"fiko-ai/internal/memory"
	"fiko-ai/internal/notify"
	"fiko-ai/internal/protocol"
	"fiko-ai/internal/stt"
	"fiko-ai/internal/tts"
	"fiko-ai/pkg/id"
)

// Conn tek bir WebSocket bağlantısının ihtiyaç duyulan asgari yüzeyidir.
// Write ve Ping yalnızca yazma döngüsünden çağrılır; Close herhangi bir goroutine'den çağrılabilir.
type Conn interface {
	// Read bir sonraki metin mesajını döndürür; bağlantı kapanınca hata verir.
	Read() ([]byte, error)
	Write(data []byte, deadline time.Time) error
	Ping(deadline time.Time) error
	SetReadDeadline(t time.Time) error
	// SetPongHandler pong geldiğinde çağrılacak fonksiyonu kaydeder (Read goroutine'inde çalışır).
	SetPongHandler(h func())
	Close() error
}

// Options zaman aşımı ve kapasite ayarlarıdır; sıfır değerler varsayılanla değiştirilir.
type Options struct {
	PingInterval   time.Duration // varsayılan 20 sn
	PongWait       time.Duration // varsayılan 50 sn (PingInterval'dan büyük olmalı)
	WriteWait      time.Duration // varsayılan 10 sn
	OutboundBuffer int           // varsayılan 256
	MaxConcurrent  int           // bağlantı başına eşzamanlı üretim sayısı, varsayılan 8
}

func (o *Options) defaults() {
	if o.PingInterval <= 0 {
		o.PingInterval = 20 * time.Second
	}
	if o.PongWait <= o.PingInterval {
		o.PongWait = 50 * time.Second
		if o.PongWait <= o.PingInterval {
			o.PongWait = 3 * o.PingInterval
		}
	}
	if o.WriteWait <= 0 {
		o.WriteWait = 10 * time.Second
	}
	if o.OutboundBuffer < 1 {
		o.OutboundBuffer = 256
	}
	if o.MaxConcurrent < 1 {
		o.MaxConcurrent = 8
	}
}

// Hub açık oturumları izler ve kapanışta hepsini düzgünce sonlandırır.
type Hub struct {
	agent *agent.Agent
	stt   *stt.Service    // nil ise sesli giriş kapalı (motor yapılandırılmamış)
	tts   tts.Synthesizer // nil ise sesli çıkış kapalı
	// memory soruları LLM'in emin olmadığı kalıcı bilgiler için kullanıcıya sorulur.
	// nil ise bellek onay akışı kapalıdır (memory.answer "yapılandırılmamış" döner).
	memory *memoryBridge
	// notify kullanıcılar arası bildirim köprüsüdür. nil ise bildirim akışı kapalı
	// (notify.answer "yapılandırılmamış" hatası verir).
	notify *notifyBridge
	log    *slog.Logger
	opts   Options

	mu       sync.Mutex
	closed   bool
	sessions map[*session]struct{}
	wg       sync.WaitGroup
}

// memoryBridge, ws katmanının bellek onay akışı için ihtiyaç duyduğu asgari yüzeydir.
// ws paketi memory paketinin TÜMÜNÜ bilmez (agent ile aynı desen).
type memoryBridge struct {
	questions *memory.QuestionStore
	// answer, soru kimliğini kullanıcı kimliğiyle birlikte uygular; ikisi de
	// DIŞARIDAN (ws katmanından) gelir. Böylece sahiplik kontrolü hem burada hem
	// ApplyAnswer içinde yapılır — ikinci katman, ilkini atlayan bir çağrıyı yakalar.
	answer func(ctx context.Context, userID, questionID string, a memory.Answer) error
}

// ---- Kullanıcılar arası bildirim (notify) ----

// notifyBridge, ws katmanının bildirim akışı için ihtiyaç duyduğu asgari yüzeydir.
//
// DIŞA AÇIK ALANLAR: main.go bu yapıyı DOLDURMAK zorunda (notify paketi ws'e
// bağımlı olamaz, dolayısıyla ws.notifyBridge'i dolduran tek yer main.go'dur —
// bkz. cmd/server/notify.go notifyBridgeFor). Alanlar dışa açık olmasa orada
// yapı oluşturulamaz; bu yüzden burada YAZILIR, okunmaz (ws içinde kullanılır).
type notifyBridge struct {
	// List, kullanıcının kutusundaki bildirimleri okur (bağlantı kurulunca
	// kaçırılmışları geri yüklemek için).
	List func(ctx context.Context, userID string, box string) ([]notify.Message, error)
	// MarkRead, alıcının bildirimi okundu işaretler.
	MarkRead func(ctx context.Context, userID, id string) (notify.Message, error)
	// Answer, soruyu yanıtlar: katılım denetimi depoda YAPILIR (sadece alıcı).
	Answer func(ctx context.Context, userID, questionID, answer string) error
	// Counts, zil sayaçlarını döndürür (unread + bekleyen soru).
	Counts func(ctx context.Context, userID string) (unread, pending int)
	// DisplayName, yanıtlayanın görünen adını çözer ("X sana yanıt verdi").
	DisplayName func(ctx context.Context, userID string) string
}

// NotifyBridge, ws katmanının bildirim akışı için ihtiyaç duyduğu asgari yüzeydir
// (main.go köprüsü için dışa açık ad).
type NotifyBridge = notifyBridge

// SetNotify, kullanıcılar arası bildirim akışını bağlar.
//
// LIST NİLEN ZORUNLU? list nil ise yalnızca ANLIK bildirimler gider; sekme
// yenilenince kaçırılmış bildirimler geri YÜKLENMEZ (bellek sorularında olduğu
// gibi sessizce kaybolur). Bu yüzden list verilmezse akış tamamen kapalı sayılır.
func (h *Hub) SetNotify(b notifyBridge) {
	if h == nil {
		return
	}
	if b.List == nil || b.Answer == nil || b.Counts == nil {
		h.notify = nil
		return
	}
	h.notify = &b
}

// sendNotification, bir bildirimi/kaydı kullanıcının TÜM bağlantılarına iter.
//
// TÜR SEÇİMİ: yanıt (`answer`) kaydı sorana gider ve zil ÇALMAZ; bildirim/soru
// alıcıya gider ve zil çalar. Bu ayrım sunucuda yapılır: arayüz olayı alıp
// karar verirse, WS düşmesinde zil yanlış çalar.
func (h *Hub) sendNotification(userID string, m notify.Message) {
	if h == nil || userID == "" {
		return
	}
	if m.Kind == notify.KindAnswer {
		o := protocol.NewOutbound(protocol.ServerNotifyAnswered, "",
			protocol.NotifyAnswerPayload{
				QuestionID: m.QuestionID,
				Question:   m.Text,
				Answer:     m.AnswerText,
				FromUserID: m.FromUserID,
				FromName:   h.notifyDisplayName(m.FromUserID, m.FromName),
				At:         m.CreatedAt.UTC().Format(time.RFC3339),
			})
		h.BroadcastToUser(userID, "", protocol.ServerNotifyAnswered, o.Payload)
		return
	}
	// Soru KENDİ kimliğini (m.ID) taşır; notifyPayloadOf bunu ayarlar.
	h.BroadcastToUser(userID, m.ConversationID, protocol.ServerNotifyReceived,
		notifyPayloadOf(m))
	// Zil sayacı tazelenir: yeni bildirim geldi.
	if h.notify != nil && h.notify.Counts != nil {
		unread, pending := h.notify.Counts(context.Background(), userID)
		h.BroadcastToUser(userID, "", protocol.ServerNotifyUnread,
			protocol.NotifyUnreadPayload{Unread: unread, Pending: pending})
	}
}

// notifyDisplayName, ad boşsa köprü üzerinden çözmeyi dener.
//
// Neden? Yanıt kaydı depoda adsız yazılır (bkz. memstore.Answer): kim cevapladıysa
// OnUserID = soran, FromUserID = cevaplayan. Kaydın FromName'i boş olduğu için
// arayüz "Veli sana yanıt verdi" diyebilmelidir.
func (h *Hub) notifyDisplayName(userID, known string) string {
	if known != "" {
		return known
	}
	if h != nil && h.notify != nil && h.notify.DisplayName != nil {
		return h.notify.DisplayName(context.Background(), userID)
	}
	return ""
}

// notifyPayloadOf, bir bildirim kaydını protokol biçimine çevirir.
//
// SORU KİMLİĞİ NEREDEN? `m.ID` — çünkü soru kaydının kendi kimliği QuestionID
// ALANINDA değil (o alan yalnızca YANIT kaydında "hangi soruya ait" bilgisini taşır,
// bkz. memstore.Answer). Soruyu `m.QuestionID` üzerinden gönderirsek alıcının
// cevabı boş soru kimliğiyle gider ve hep "bulunamadı" hatası alır.
func notifyPayloadOf(m notify.Message) protocol.NotifyPayload {
	qid := m.QuestionID
	if m.Kind == notify.KindQuestion {
		qid = m.ID
	}
	p := protocol.NotifyPayload{
		ID: m.ID, QuestionID: qid,
		Kind: string(m.Kind), Text: m.Text,
		FromUserID: m.FromUserID, FromName: m.FromName,
		Pending: m.IsPending(),
		At:      m.CreatedAt.UTC().Format(time.RFC3339),
	}
	if m.ExpiresAt != nil {
		p.ExpiresAt = m.ExpiresAt.UTC().Format(time.RFC3339)
	}
	for _, o := range m.Options {
		p.Options = append(p.Options, protocol.NotifyOption{Text: o.Text})
	}
	return p
}

// SetMemory, kalıcı bellek onay akışını bağlar. questions/answer nil ise akış kapalı kalır.
//
// Kullanıcı isteği: LLM karar veremediğinde kullanıcıya SOR; cevabı o verir.
func (h *Hub) SetMemory(questions *memory.QuestionStore, answer func(ctx context.Context, userID, questionID string, a memory.Answer) error) {
	if h == nil {
		return
	}
	if questions == nil || answer == nil {
		h.memory = nil
		return
	}
	h.memory = &memoryBridge{questions: questions, answer: answer}
}

// Deliver, notify.Sink arayüzünün WS uyarlamasıdır: bildirimi alıcıya iter ve
// zil sayacını tazeller.
//
// VARSAYILAN OLARAK KULLANILMAZ: notify paketi kendi Sink'ini main.go'da alır.
// Bu metot, hub'ın kendi içinde kullanılan kısa yol (testler ve ileride REST
// üzerinden anlık gönderim için) olarak durur.
func (h *Hub) Deliver(userID string, m notify.Message) {
	if h == nil {
		return
	}
	h.sendNotification(userID, m)
}

// markNotificationRead, tek bir bildirimi okundu işaretler ve güncel sayacı
// kullanıcının TÜM bağlantılarına yollar.
//
// Neden tüm bağlantılara? Okundu işareti hesap genelindedir: kullanıcı bir sekmede
// açtı diye diğer sekmedeki rozet eskimiş kalmasın. Sadece o bağlantıya gönderilseydi
// sayaçlar birbirinden kopardı.
func (s *session) markNotificationRead(id string) error {
	if s == nil || s.hub == nil || s.hub.notify == nil || s.hub.notify.MarkRead == nil {
		return fmt.Errorf("bildirim akışı kapalı")
	}
	if _, err := s.hub.notify.MarkRead(context.Background(), s.userID, id); err != nil {
		return err
	}
	// Sayaç TÜM bağlantılara gider: okundu işareti hesap genelindedir.
	if s.hub.notify.Counts != nil {
		unread, pending := s.hub.notify.Counts(context.Background(), s.userID)
		s.hub.BroadcastToUser(s.userID, "", protocol.ServerNotifyUnread,
			protocol.NotifyUnreadPayload{Unread: unread, Pending: pending})
	}
	return nil
}

// NewHub hub kurar. stt nil ise ses olayları "yapılandırılmamış" hatasıyla döner.
func NewHub(a *agent.Agent, log *slog.Logger, opts Options) *Hub {
	return NewHubWithSTT(a, nil, log, opts)
}

// NewHubWithSTT ses tanımayı da bağlar (Phase 6).
func NewHubWithSTT(a *agent.Agent, s *stt.Service, log *slog.Logger, opts Options) *Hub {
	return NewHubWithVoice(a, s, nil, log, opts)
}

// NewHubWithVoice ses tanımayı (STT) ve ses üretimini (TTS) bağlar (Phase 6 + 7).
// Her ikisi de nil olabilir: o zaman ses olayları "yapılandırılmamış" hatası döner.
func NewHubWithVoice(a *agent.Agent, s *stt.Service, t tts.Synthesizer, log *slog.Logger, opts Options) *Hub {
	if log == nil {
		log = slog.Default()
	}
	opts.defaults()
	return &Hub{agent: a, stt: s, tts: t, log: log, opts: opts, sessions: make(map[*session]struct{})}
}

// HasSTT sesli giriş (mikrofon) motorunun bağlı olup olmadığını bildirir.
// Arayüz bu bilgiyle mikrofon düğmesini gizler: motor yoksa voice.start
// istekleri "yapılandırılmamış" hatası dönerdi.
func (h *Hub) HasSTT() bool { return h != nil && h.stt != nil }

// HasTTS sesli çıkış (sesli okuma) motorunun bağlı olup olmadığını bildirir.
// Arayüz bu bilgiyle hoparlör düğmesini gizler.
func (h *Hub) HasTTS() bool { return h != nil && h.tts != nil }

// Count açık oturum sayısıdır.
func (h *Hub) Count() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return len(h.sessions)
}

// BroadcastToUser, bir olayı kullanıcının TÜM açık bağlantılarına gönderir.
//
// Neden kullanıcı bazlı? Bellek sorusu bir sohbetle bağlı değildir: kullanıcı hangi
// sekmede olursa olsun görmelidir (birden çok sekme açık olabilir).
func (h *Hub) BroadcastToUser(userID, conversationID string, t protocol.ServerType, payload any) {
	if h == nil || userID == "" {
		return
	}
	o := protocol.NewOutbound(t, conversationID, payload)
	h.mu.Lock()
	list := make([]*session, 0, len(h.sessions))
	for s := range h.sessions {
		if s.userID == userID {
			list = append(list, s)
		}
	}
	h.mu.Unlock()
	// send() kuyruğa güvenle yazar (doluysa bağlantıyı kapatır) ve asla bloklamaz.
	for _, s := range list {
		s.send(o)
	}
}

// Serve bağlantı kapanana kadar bloklar. ctx sunucu ömrüdür; iptali bağlantıyı kapatır.
func (h *Hub) Serve(ctx context.Context, conn Conn, userID string) {
	sctx, cancel := context.WithCancel(ctx)
	s := &session{
		hub:    h,
		conn:   conn,
		userID: userID,
		ctx:    sctx,
		cancel: cancel,
		out:    make(chan []byte, h.opts.OutboundBuffer),
		sem:    make(chan struct{}, h.opts.MaxConcurrent),
		owned:  make(map[string]struct{}),
		id:     id.New("ws"),
	}
	if !h.register(s) {
		cancel()
		_ = conn.Close()
		return
	}
	defer h.unregister(s)

	h.log.Debug("ws bağlandı", "user", userID)
	writerDone := make(chan struct{})
	go func() { defer close(writerDone); s.writeLoop() }()

	// BEKLEYEN BELLEK SORULARINI GERİ YÜKLE (kullanıcı isteği).
	//
	// Soru yalnızca üretildiği AN `memory.question` olayıyla gider. Sekme
	// yenilenince ya da bağlantı kopup tekrar kurulunca o anlık olay kaçmış olur:
	// soru depoda TTL boyunca dursa da kimse sormuyor, kullanıcı kartı hiç
	// görmüyor ve bilgisi sessizce kayboluyor.
	//
	// Bu yüzden her bağlantıda kullanıcının hâlâ bekleyen soruları TEKRAR gönderilir.
	// conversationID boş bırakılır: kullanıcı hangi sekmede olursa olsun görmelidir.
	s.replayPendingQuestions()

	// Bildirim tarafı da aynı mantıkla geri yüklenir: bildirimler KALICI
	// kayıttır (kullanıcı çevrimdışıyken de gider); bağlantı kurulunca gelen
	// kutusu, gönendiler ve zil sayacı tazelenir.
	s.replayNotifications()

	s.readLoop()
	s.close()
	<-writerDone
	s.jobs.Wait() // iptal edilen üretimlerin kısmi içeriği kaydetmesini bekle
	h.log.Debug("ws kapandı", "user", userID)
}

// Close yeni bağlantıları reddeder, bekleyen giden olayları göndermeye çalışıp (ctx süresince) tüm
// oturumları kapatır ve Serve çağrılarının dönmesini bekler.
// Önce agent.Close çağrılmalıdır; böylece "interrupted" olayları kuyruğa girmiş olur.
func (h *Hub) Close(ctx context.Context) error {
	h.mu.Lock()
	h.closed = true
	list := make([]*session, 0, len(h.sessions))
	for s := range h.sessions {
		list = append(list, s)
	}
	h.mu.Unlock()

	for _, s := range list {
		go s.drainAndClose(ctx)
	}
	done := make(chan struct{})
	go func() { h.wg.Wait(); close(done) }()
	select {
	case <-done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (h *Hub) register(s *session) bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.closed {
		return false
	}
	h.sessions[s] = struct{}{}
	h.wg.Add(1)
	return true
}

func (h *Hub) unregister(s *session) {
	h.mu.Lock()
	delete(h.sessions, s)
	h.mu.Unlock()
	h.wg.Done()
}

// ---- oturum ----

type session struct {
	hub    *Hub
	conn   Conn
	userID string
	ctx    context.Context
	cancel context.CancelFunc

	out     chan []byte
	pending atomic.Int64 // kuyruğa girmiş ama henüz yazılmamış mesaj sayısı
	sem     chan struct{}
	jobs    sync.WaitGroup

	mu     sync.Mutex
	closed bool
	id     string              // bu bağlantının kimliği (ses kaydı oturuma bağlanır)
	owned  map[string]struct{} // bu bağlantının üretim başlattığı konuşmalar

	// --- TTS (Phase 7) ---
	// Sesli mod oturum bazlıdır ve kullanıcıya aittir: sunucu bir kez açılır,
	// her mesajda otomatik konuşmaz. ttsCancel, üretim sürerken kullanıcı
	// konuşmaya başlarsa (barge-in) veya tts.stop gelirse üretimi keser.
	ttsOn     bool
	ttsConv   string   // sesli modun açıldığı konuşma
	tts       *speaker // akışlı seslendirme durumu (nil = bu üretimde ses yok)
	ttsJobs   chan ttsJob
	ttsWait   chan struct{} // sözlü bitince kapanır
	ttsCancel context.CancelFunc
	ttsMu     sync.Mutex // yalnızca tts alanlarını korur (tüm oturumu kilitlemez)
	// speechBuf: delta'larla gelen metin. Cümle bittiğinde TTS'e gönderilir
	// (bkz. feedSpeech). Boş metinde konuşacak bir şey kalmadığını gösterir.
	speechBuf string
}

// ttsJob, seslendirilecek tek bir cümledir. Kanal üzerinden seri olarak işlenir:
// tek GPU'da iki üretimi aynı anda çalıştırmak motoru yavaşlatır.
type ttsJob struct {
	text string
	// index bu mesajdaki sıra numarasıdır (istemci sırayla oynatır).
	index int
}

// sessionID, ses kaydının bağlandığı oturum kimliğidir.
// Bağlantı başına tek kayıt tutulur: iki mikrofon aynı anda açılamaz (aynı kayıt üstüne yazar).
func (s *session) sessionID() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.id
}

// close oturumu bir kez kapatır: bağlamı iptal eder ve bağlantıyı koparır.
func (s *session) close() {
	s.mu.Lock()
	already := s.closed
	s.closed = true
	s.mu.Unlock()
	if already {
		return
	}
	s.cancel()
	// Açık ses kaydı bağlantıyla birlikte düşer (yarım kayıt sunucuda kalmasın).
	if s.hub.stt != nil {
		s.hub.stt.Cancel(s.sessionID())
	}
	_ = s.conn.Close()
}

func (s *session) drainAndClose(ctx context.Context) {
	tk := time.NewTicker(10 * time.Millisecond)
	defer tk.Stop()
	for s.pending.Load() > 0 {
		select {
		case <-ctx.Done():
			s.close()
			return
		case <-tk.C:
		}
	}
	s.close()
}

// send bir olayı kuyruğa yazar; asla bloklamaz. Kuyruk doluysa bağlantı kapatılır.
func (s *session) send(o protocol.Outbound) {
	// --- Akışlı seslendirme ---
	// ÖNCEKİ DAVRANIŞ: seslendirme yalnızca `text.final` gelince başlardı, yani
	// kullanıcı cevabın TAMAMINI okuduktan sonra ilk sesi duyardu.
	// ŞİMDİ: cümle metin AKIŞI SIRASINDA kuyruğa giriyor. LLM 3. cümleyi yazarken
	// 1. cümle seslendirilmiş olur → ilk ses, cevap süresinden bağımsız gelir.
	if s.speaking() {
		switch o.Type {
		case protocol.ServerAssistantThinking:
			// Asistan balonu açıldı: yeni cevap için kuyruk başlat (öncekini kes).
			if p, ok := o.Payload.(protocol.ThinkingPayload); ok {
				s.startSpeaker(o.ConversationID, p.MessageID)
			}
		case protocol.ServerAssistantTextDelta:
			if p, ok := o.Payload.(protocol.TextDeltaPayload); ok {
				s.feedSpeech(p.Delta)
			}
		case protocol.ServerAssistantTextFinal:
			if p, ok := o.Payload.(protocol.TextFinalPayload); ok && p.Status == "complete" {
				// Kalan metni boşalt ve işleyicinin bitmesini bekle (tts.done buradan çıkar).
				s.finishSpeech()
			}
		}
	}
	b, err := json.Marshal(o)
	if err != nil {
		s.hub.log.Error("ws olayı serileştirilemedi", "type", o.Type, "err", err)
		return
	}
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return
	}
	s.pending.Add(1)
	select {
	case s.out <- b:
		s.mu.Unlock()
	default:
		s.pending.Add(-1)
		s.mu.Unlock()
		s.hub.log.Warn("ws istemcisi yavaş: giden kuyruk doldu, bağlantı kapatılıyor", "user", s.userID)
		s.close()
	}
}

func (s *session) sendError(convID, replyTo, code, msg string, retryable bool) {
	o := protocol.NewOutbound(protocol.ServerError, convID, protocol.ErrorPayload{Code: code, Message: msg, Retryable: retryable})
	o.ReplyTo = replyTo
	s.send(o)
}

func (s *session) writeLoop() {
	opts := s.hub.opts
	tk := time.NewTicker(opts.PingInterval)
	defer tk.Stop()
	for {
		select {
		case <-s.ctx.Done():
			return
		case b := <-s.out:
			err := s.conn.Write(b, time.Now().Add(opts.WriteWait))
			s.pending.Add(-1)
			if err != nil {
				s.hub.log.Debug("ws yazma hatası", "err", err)
				s.close()
				return
			}
		case <-tk.C:
			if err := s.conn.Ping(time.Now().Add(opts.WriteWait)); err != nil {
				s.hub.log.Debug("ws ping hatası", "err", err)
				s.close()
				return
			}
		}
	}
}

func (s *session) readLoop() {
	wait := s.hub.opts.PongWait
	extend := func() { _ = s.conn.SetReadDeadline(time.Now().Add(wait)) }
	extend()
	s.conn.SetPongHandler(extend)
	for {
		data, err := s.conn.Read()
		if err != nil {
			s.hub.log.Debug("ws okuma bitti", "err", err)
			return
		}
		extend()
		s.handle(data)
	}
}

// handle tek bir istemci olayını işler. Yavaş işler ayrı goroutine'de yürür; okuma döngüsü bloklanmaz.
func (s *session) handle(data []byte) {
	var in protocol.Inbound
	if err := json.Unmarshal(data, &in); err != nil {
		s.sendError("", "", protocol.ErrCodeBadRequest, "geçersiz JSON", false)
		return
	}
	if !in.Type.Valid() {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeUnknownEvent, fmt.Sprintf("bilinmeyen olay türü: %q", in.Type), false)
		return
	}

	switch in.Type {
	case protocol.ClientPing:
		o := protocol.NewOutbound(protocol.ServerPong, in.ConversationID, nil)
		o.ReplyTo = in.ID
		s.send(o)

	case protocol.ClientTextMessage:
		var p protocol.TextMessagePayload
		if len(in.Payload) == 0 || json.Unmarshal(in.Payload, &p) != nil {
			s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, "text.message payload'ı geçersiz", false)
			return
		}
		s.startGeneration(in, p)

	case protocol.ClientVoiceStart:
		s.handleVoiceStart(in)

	case protocol.ClientVoiceAudio:
		s.handleVoiceAudio(in)

	case protocol.ClientVoiceStop:
		s.handleVoiceStop(in)

	case protocol.ClientTTSMode:
		s.handleTTSMode(in)

	case protocol.ClientTTSStop:
		s.handleTTSStop(in)

	case protocol.ClientConversationCancel, protocol.ClientInterrupt:
		if in.ConversationID == "" {
			s.sendError("", in.ID, protocol.ErrCodeBadRequest, "conversationId gerekli", false)
			return
		}
		s.mu.Lock()
		_, mine := s.owned[in.ConversationID]
		s.mu.Unlock()
		if !mine {
			return // bu bağlantının başlattığı bir üretim değil: sessizce yoksay
		}
		var p protocol.CancelPayload
		_ = json.Unmarshal(in.Payload, &p)
		reason := p.Reason
		if reason == "" {
			reason = "user"
		}
		s.hub.agent.Cancel(in.ConversationID, reason)

	case protocol.ClientMemoryAnswer:
		s.handleMemoryAnswer(in)

	case protocol.ClientNotifyAnswer:
		s.handleNotifyAnswer(in)

	case protocol.ClientToolConfirmAnswer:
		s.handleConfirmAnswer(in)

	default: // voice.* — sonraki fazlar
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeNotImplemented, "bu olay türü henüz desteklenmiyor", false)
	}
}

// handleConfirmAnswer, kullanıcının yazma/silme onay formunu cevabını uygular.
//
// Güvenlik: form, bu bağlantının KULLANICISINA ait olmalıdır; aksi hâlde başka birinin
// onayını göndererek onun adına silme yaptırılabilirdi. Onay deposu kullanıcı kimliğini
// taşımadığı için burada konuşma sahipliği denetlenir (bkz. agent.ConfirmAnswer).
func (s *session) handleConfirmAnswer(in protocol.Inbound) {
	if s.hub.agent == nil {
		s.sendError("", in.ID, protocol.ErrCodeNotImplemented, "onay akışı kapalı", false)
		return
	}
	var p protocol.ToolConfirmAnswerPayload
	if len(in.Payload) == 0 || json.Unmarshal(in.Payload, &p) != nil {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, "tool.confirm.answer payload'ı geçersiz", false)
		return
	}
	if strings.TrimSpace(p.ConfirmID) == "" {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, "confirmId zorunlu", false)
		return
	}
	convID := in.ConversationID
	if convID == "" {
		// Konuşma kimliği payload'da yoksa formu bu bağlantının sahip olduğu son
		// konuşmada çalıştır (owned = bu bağlantının üretim başlattığı konuşmalar).
		convID = s.lastOwnedConversation()
	}
	if convID == "" {
		s.sendError("", in.ID, protocol.ErrCodeBadRequest, "konuşma bulunamadı", false)
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	// Sonucu BU oturum gönderir: form tur bittikten sonra cevaplanır, aktif tur (emit) yoktur;
	// ajan üzerinden yayınlansa sessizce düşer ve kart "işleniyor"da kalırdı.
	res := s.hub.agent.ResolveConfirm(ctx, convID, s.userID, p)
	s.send(protocol.NewOutbound(protocol.ServerToolConfirmResolved, convID, res))
}

// lastOwnedConversation bu bağlantının sahip olduğu bir konuşma kimliğini döndürür
// (yoksa ""). Onay cevabında conversationId gelmezse devam edecek yer buradan bulunur.
func (s *session) lastOwnedConversation() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	for id := range s.owned {
		return id
	}
	return ""
}

// ---- ses (Phase 6) ----

// handleMemoryAnswer, kullanıcının bellek sorusuna verdiği yanıtı uygular.
//
// Güvenlik: soru, bu bağlantının kendi KULLANICISINA ait olmalıdır; aksi hâlde başka
// birinin sorusunu yanıtlamak mümkün olurdu.
func (s *session) handleMemoryAnswer(in protocol.Inbound) {
	if s.hub.memory == nil {
		s.sendError("", in.ID, protocol.ErrCodeNotImplemented, "bellek onay akışı kapalı", false)
		return
	}
	var p protocol.MemoryAnswerPayload
	if len(in.Payload) == 0 || json.Unmarshal(in.Payload, &p) != nil {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, "memory.answer payload'ı geçersiz", false)
		return
	}
	q, ok := s.hub.memory.questions.Get(p.QuestionID)
	if !ok {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest,
			"soru bulunamadı veya süresi doldu", false)
		return
	}
	if q.UserID != s.userID {
		// Başka kullanıcının sorusu: bilgi sızdırma riski, sessizce yok sayılır.
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodePermissionDenied, "yetkisiz", false)
		return
	}
	if err := s.hub.memory.answer(context.Background(), s.userID, p.QuestionID, memory.Answer{
		Confirmed: p.Confirmed, SaveAsNew: p.SaveAsNew,
	}); err != nil {
		if errors.Is(err, memory.ErrQuestionNotFound) {
			s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, "soru bulunamadı", false)
			return
		}
		// Hata YUTULMAZ: "bellek güncellenemedi" tek cümlesi nedeni söylemiyordu ve
		// canlıda bu satırda üretilen hata görünmez şekilde kayboluyordu.
		//
		// Doğrulama hatası kullanıcının girdisiyle ilgilidir (ör. içerik boş/hassas);
		// 500 değil 400'tür ve sebebi kullanıcıya gösterilebilir. Diğer hatalar
		// sunucu tarafıdır: loglanır, kullanıcıya genel mesaj gider.
		var ve *memory.ValidationError
		if errors.As(err, &ve) {
			s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest,
				"bu bilgi kaydedilemedi: "+ve.Error(), false)
			return
		}
		if s.hub != nil && s.hub.log != nil {
			s.hub.log.Error("bellek onayı uygulanamadı", "err", err,
				"soru", p.QuestionID, "user", s.userID, "conversation", in.ConversationID)
		}
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeInternal, "bellek güncellenemedi", true)
		return
	}
	// Onay uygulandı; soru listeden düşer. İstemci `MemoryUpdated`/`MemoryAdded`
	// olaylarını zaten alır, ayrıca onay göndermeye gerek yoktur.
}

// replayPendingQuestions, bu bağlantı kurulduğunda kullanıcının hâlâ yanıt bekleyen
// bellek sorularını yeniden gönderir.
//
// Neden gerekir? `memory.question` olayı ANLIKTIR. Sekme yenilenince ya da ağ
// dalgalanınca o olay kaçar; soru depoda TTL boyunca dursa da kullanıcı kartı
// hiç görmez ve verdiği kararı kaybeder.
//
// Güvenlik: sorular yalnızca BAĞLANTININ KENDİ kullanıcısına gider (PendingFor
// userID ile filtreler); başkasının sorusu sızmaz.
func (s *session) replayPendingQuestions() {
	if s == nil || s.hub == nil || s.hub.memory == nil || s.userID == "" {
		return
	}
	pending := s.hub.memory.questions.PendingFor(s.userID, "")
	if len(pending) == 0 {
		return
	}
	for _, q := range pending {
		o := protocol.NewOutbound(protocol.ServerMemoryQuestion, q.ConversationID,
			protocol.MemoryQuestionPayload{
				QuestionID: q.ID, Text: q.Text, Options: wsQuestionOptions(q.Options),
				Previous: q.Previous, NewValue: q.Content, TargetID: q.TargetID,
				ExpiresAt: q.ExpiresAt.UTC().Format(time.RFC3339),
			})
		s.send(o)
	}
	s.hub.log.Debug("bekleyen bellek soruları geri yüklendi",
		"adet", len(pending), "user", s.userID)
}

// handleNotifyAnswer, kullanıcının bildirim sorusuna verdiği yanıtı uygular.
//
// GÜVENLİK: katılım denetimi notify DEPOSUNDA yapılır (yalnızca sorunun alıcısı
// yanıtlar) — ws katmanı ikinci kez kontrol etmez çünkü aynı soruyu iki yerden
// denetlemek iki yerde yanlışlık riski demektir. Ancak hata KULLANICIYA özgü
// olduğu için "yetkisiz" ayrımı korunur (403).
func (s *session) handleNotifyAnswer(in protocol.Inbound) {
	if s.hub == nil || s.hub.notify == nil {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeNotImplemented,
			"bildirim akışı kapalı", false)
		return
	}
	var p protocol.NotifyAnswerInput
	if len(in.Payload) == 0 || json.Unmarshal(in.Payload, &p) != nil {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest,
			"notify.answer payload'ı geçersiz", false)
		return
	}
	if strings.TrimSpace(p.Answer) == "" {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest,
			"yanıt boş olamaz", false)
		return
	}
	if err := s.hub.notify.Answer(context.Background(), s.userID, p.QuestionID, p.Answer); err != nil {
		switch {
		case errors.Is(err, notify.ErrForbidden):
			// Başkasının sorusu: bilgi sızdırma riski nedeniyle sessizce yok sayılır.
			s.sendError(in.ConversationID, in.ID, protocol.ErrCodePermissionDenied, "yetkisiz", false)
		case errors.Is(err, notify.ErrNotFound):
			s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest,
				"soru bulunamadı veya süresi doldu", false)
		case errors.Is(err, notify.ErrValidation):
			// Doğrulama hatası kullanıcının GİRDİSİYLE ilgilidir → 400 + sebep.
			s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, err.Error(), false)
		default:
			if s.hub.log != nil {
				s.hub.log.Error("bildirim yanıtı uygulanamadı", "err", err,
					"soru", p.QuestionID, "user", s.userID)
			}
			s.sendError(in.ConversationID, in.ID, protocol.ErrCodeInternal,
				"yanıt kaydedilemedi", true)
		}
		return
	}
	// Yanıt uygulandı: soru kartı "cevaplandı" olmalı ve zil AZALMALI.
	// notify.answer olayı SORANA zaten gitti; alıcıya göndermedeğimiz tek şey
	// kendi durumu ve güncel sayaç.
	s.sendUnread()
}

// sendUnread, kullanıcıya güncel zil sayaçlarını gönderir.
func (s *session) sendUnread() {
	if s == nil || s.hub == nil || s.hub.notify == nil || s.hub.notify.Counts == nil {
		return
	}
	unread, pending := s.hub.notify.Counts(context.Background(), s.userID)
	s.send(protocol.NewOutbound(protocol.ServerNotifyUnread, "",
		protocol.NotifyUnreadPayload{Unread: unread, Pending: pending}))
}

// replayNotifications, bağlantı kurulduğunda kaçırılmış bildirimleri geri yükler.
//
// NEDEN GEREKLİ? `notify.received` ANLIK olaydır. Kullanıcı çevrimdışıyken gönderilir,
// WS düşer ve o bildirim kaybolur — kalıcı kayıttan geri yüklenmezse kullanıcı
// hiç göremez. Aynı sebepten zil sayacı da bağlantıda gönderilir.
//
// GÜVENLİK: list yalnızca BAĞLANTININ KENDİ kullanıcısının kutusunu okur; başkasının
// bildirimi sızmaz.
func (s *session) replayNotifications() {
	if s == nil || s.hub == nil || s.hub.notify == nil || s.userID == "" {
		return
	}
	ctx := context.Background()

	// 1) Gelen kutusu: bana gönderilen bildirim/sorular.
	inbox, err := s.hub.notify.List(ctx, s.userID, notify.BoxInbox)
	if err != nil {
		s.hub.log.Warn("bildirim geri yüklenemedi (gelen kutusu)", "user", s.userID, "err", err)
	} else {
		// ESKİDEN YENİYE: liste yeniye önce sıralı gelir; kullanıcı en son
		// bildirimi üstte görsün diye TERS ÇEVİRİLİR.
		for i := len(inbox) - 1; i >= 0; i-- {
			m := inbox[i]
			s.send(protocol.NewOutbound(protocol.ServerNotifyReceived, m.ConversationID,
				notifyPayloadOf(m)))
		}
	}

	// 2) Gönendiler: SORULARIM VE BANA GELEN YANITLARIM.
	sent, err := s.hub.notify.List(ctx, s.userID, notify.BoxSent)
	if err != nil {
		s.hub.log.Warn("bildirim geri yüklenemedi (gönendiler)", "user", s.userID, "err", err)
	} else {
		for i := len(sent) - 1; i >= 0; i-- {
			m := sent[i]
			if m.Kind == notify.KindAnswer {
				s.send(protocol.NewOutbound(protocol.ServerNotifyAnswered, "",
					protocol.NotifyAnswerPayload{
						QuestionID: m.QuestionID, Question: m.Text, Answer: m.AnswerText,
						FromUserID: m.FromUserID,
						FromName:   s.hub.notifyDisplayName(m.FromUserID, m.FromName),
						At:         m.CreatedAt.UTC().Format(time.RFC3339),
					}))
			} else {
				s.send(protocol.NewOutbound(protocol.ServerNotifyReceived, m.ConversationID,
					notifyPayloadOf(m)))
			}
		}
	}

	// 3) Zil sayacı.
	s.sendUnread()
	if s.hub.log != nil {
		s.hub.log.Debug("bildirimler geri yüklendi",
			"gelen", len(inbox), "gonderilen", len(sent), "user", s.userID)
	}
}

// wsQuestionOptions, bellek seçeneklerini protokol biçimine çevirir.
func wsQuestionOptions(opts []memory.QuestionOption) []protocol.MemoryQuestionOption {
	if len(opts) == 0 {
		return nil
	}
	out := make([]protocol.MemoryQuestionOption, 0, len(opts))
	for _, o := range opts {
		out = append(out, protocol.MemoryQuestionOption{
			Text: o.Text, Confirmed: o.Confirmed, SaveAsNew: o.SaveAsNew,
		})
	}
	return out
}

// handleVoiceStart kaydı başlatır. Motor yapılandırılmamışsa sessizce GEÇİLMEZ:
// istemciye açık hata döner, mikrofon rozetinde "kapalı" görünür.
func (s *session) handleVoiceStart(in protocol.Inbound) {
	// BARGE-IN: kullanıcı konuşmaya başladıysa asistan susmalı. Aksi hâlde
	// mikrofon kendi sesini kaydeder ve kullanıcı iki sesi birden duyar.
	if s.stopSpeaking() {
		s.send(protocol.NewOutbound(protocol.ServerTTSDone, voiceConvID(in, ""), protocol.TTSDonePayload{
			ConversationID: voiceConvID(in, ""), Stopped: true,
		}))
	}
	if s.hub.stt == nil || !s.hub.stt.Enabled() {
		s.sendError(in.ConversationID, in.ID, "not_implemented", "sesli giriş yapılandırılmamış (STT_URL)", false)
		return
	}
	var p protocol.VoiceStartPayload
	if len(in.Payload) == 0 || json.Unmarshal(in.Payload, &p) != nil {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, "voice.start payload'ı geçersiz", false)
		return
	}
	s.hub.stt.Start(s.sessionID(), p.Mime)
	o := protocol.NewOutbound(protocol.ServerConversationState, voiceConvID(in, p.ConversationID), protocol.StatePayload{State: protocol.StateThinking})
	o.ReplyTo = in.ID
	s.send(o)
}

// voiceConvID, konuşma kimliğini kare üzerinden okur (payload yedeği).
//
// NÖRİN: Konuşma kimliği `Inbound.ConversationID` alanındadır; payload içinde değil.
// İlk yazımda payload'dan okuyunca kimlik BOŞ geliyor, otomatik gönderim
// "conversationId gerekli" ile düşüyordu (canlı test, 18. oturum).
func voiceConvID(in protocol.Inbound, fallback string) string {
	if in.ConversationID != "" {
		return in.ConversationID
	}
	return fallback
}

// handleVoiceAudio bir ses parçasını biriktirir.
func (s *session) handleVoiceAudio(in protocol.Inbound) {
	if s.hub.stt == nil || !s.hub.stt.Enabled() {
		s.sendError(in.ConversationID, in.ID, "not_implemented", "sesli giriş yapılandırılmamış", false)
		return
	}
	var p protocol.VoiceAudioPayload
	if len(in.Payload) == 0 || json.Unmarshal(in.Payload, &p) != nil {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, "voice.audio payload'ı geçersiz", false)
		return
	}
	raw, err := base64.StdEncoding.DecodeString(p.Data)
	if err != nil {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, "ses verisi base64 değil", false)
		return
	}
	s.hub.stt.Append(s.sessionID(), raw)
}

// handleVoiceStop kaydı bitirir, metne çevirir ve metni kullanıcı mesajı gibi gönderir.
func (s *session) handleVoiceStop(in protocol.Inbound) {
	if s.hub.stt == nil || !s.hub.stt.Enabled() {
		s.sendError(in.ConversationID, in.ID, "not_implemented", "sesli giriş yapılandırılmamış", false)
		return
	}
	var p protocol.VoiceStopPayload
	if len(in.Payload) == 0 || json.Unmarshal(in.Payload, &p) != nil {
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBadRequest, "voice.stop payload'ı geçersiz", false)
		return
	}

	// Tanıma YAVAŞ olabilir: okuma döngüsünü bloklamayız.
	s.jobs.Add(1)
	go func() {
		defer s.jobs.Done()
		text, dur, err := s.hub.stt.Stop(s.ctx, s.sessionID())
		convID := voiceConvID(in, p.ConversationID)
		if err != nil {
			s.hub.log.Warn("ses tanınmadı", "err", err)
			code, msg, retry := voiceError(err)
			s.sendError(convID, in.ID, code, msg, retry)
			return
		}
		out := protocol.VoiceDetectedPayload{
			ConversationID: convID, Text: text, DurationMs: dur.Milliseconds(), Sent: text != "",
		}
		o := protocol.NewOutbound(protocol.ServerVoiceDetected, convID, out)
		o.ReplyTo = in.ID
		s.send(o)

		if text == "" {
			// Motor boş döndü: boş mesaj göndermek yerine kullanıcıya bırakılır.
			return
		}
		// Tanınan metin, kullanıcının yazdığı mesaj gibi işlenir.
		s.startGeneration(protocol.Inbound{Type: protocol.ClientTextMessage, ID: in.ID, ConversationID: convID},
			protocol.TextMessagePayload{Text: text})
	}()
}

// ---- TTS (Phase 7) ----

// speaking, oturumda sesli modun açık olup olmadığını döner (motor yoksa her zaman false).
func (s *session) speaking() bool {
	if s.hub.tts == nil {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.ttsOn
}

// handleTTSMode sesli modu açar/kapatır.
//
// Neden otomatik değil? Her mesajda seslendirme pahalıdır (motor çağrısı + veri) ve
// rahatsız edicidir; kullanıcı kendi düğmesiyle kontrol eder.
func (s *session) handleTTSMode(in protocol.Inbound) {
	if s.hub.tts == nil {
		s.sendError(in.ConversationID, in.ID, "not_implemented", "sesli çıkış yapılandırılmamış (TTS_URL)", false)
		return
	}
	var p protocol.TTSModePayload
	if len(in.Payload) > 0 {
		_ = json.Unmarshal(in.Payload, &p) // payload yoksa kapatma olarak yorumlanır
	}
	s.mu.Lock()
	s.ttsOn = p.Enabled
	if p.Enabled {
		s.ttsConv = p.ConversationID
	} else {
		s.ttsConv = ""
	}
	s.mu.Unlock()

	// Kapatma: yarım kalan seslendirmeyi durdur (stopSpeaking kendi kilidini kullanır;
	// s.mu TUTULMADAN çağrılmalı — aksi hâlde kilit sırası çıkmaz).
	if !p.Enabled {
		s.stopSpeaking()
	}
}

// handleTTSStop konuşmayı keser (barge-in). Hata DEĞİLDİR: tts.done(stopped) döner.
func (s *session) handleTTSStop(in protocol.Inbound) {
	convID := voiceConvID(in, "")
	canceled := s.stopSpeaking()
	if canceled {
		s.send(protocol.NewOutbound(protocol.ServerTTSDone, convID, protocol.TTSDonePayload{
			ConversationID: convID, Stopped: true,
		}))
	}
}

// ---- akışlı seslendirme (gerçek zaman) ----

// speaker, bir yanıtın seslendirilmesini kuyrukta tutar ve seri olarak işler.
// Tasarım: cümle METİN AKIŞI SIRASINDA kuyruğa girer (text.final'i beklemez),
// böylece kullanıcı cevabın bitmesini beklemeden ilk sesi duyar.
type speaker struct {
	convID string
	msgID  string
	index  int
}

// cümleSonu, bir cümlenin bittiği konumu bulur.
//
// Neden regex? Kısmi metin gelir; "Merhaba. Nasılsın" görüldüğünde ilk cümle
// hemen gönderilebilir. Kural: noktalama işareti + BOŞLUK (veya metin sonu).
// "3.14" gibi noktaları yanlışlıkla cümle sonu saymamak için boşluk şart.
var cumleSonu = regexp.MustCompile(`[.!?…]["'»)\]]?\s`)

// feedSpeech, gelen metin parçasını tampona ekler ve TAMAMLANMIŞ cümleleri kuyruğa gönderir.
//
// Neden tamamlanmış cümle beklenir? Yarım cümle ("Bugün hava çok") seslendirilirse
// telaffuz bozulur ve dinleyici cümlenin devamını bekler. Bu yüzden noktalama beklenir.
func (s *session) feedSpeech(delta string) {
	if s.hub.tts == nil || delta == "" {
		return
	}
	s.ttsMu.Lock()
	s.speechBuf += delta
	// Yalnızca KAPANMIŞ cümleleri al: son eşleşme + sonrasındaki boşluk.
	m := cumleSonu.FindAllStringIndex(s.speechBuf, -1)
	if len(m) == 0 {
		s.ttsMu.Unlock()
		return
	}
	last := m[len(m)-1]
	ready := s.speechBuf[:last[1]]
	s.speechBuf = s.speechBuf[last[1]:]
	s.ttsMu.Unlock()

	s.enqueueSpeech(ready)
}

// finishSpeech, kalan (noktalamasız olabilir) metni kuyruğa alır ve bitişi bekler.
// text.final geldiğinde çağrılır: bu andan sonra yeni metin GELMEYECEK.
func (s *session) finishSpeech() {
	s.ttsMu.Lock()
	rest := s.speechBuf
	s.speechBuf = ""
	s.ttsMu.Unlock()
	if strings.TrimSpace(rest) != "" {
		s.enqueueSpeech(rest)
	}
	s.closeSpeechQueue()
}

// enqueueSpeech metni okunabilir cümlelere böler ve kuyruğa ekler (bloklamadan).
func (s *session) enqueueSpeech(text string) {
	s.ttsMu.Lock()
	jobs := s.ttsJobs
	s.ttsMu.Unlock()
	if jobs == nil {
		return
	}
	for _, sent := range tts.Split(text, tts.DefaultMaxChars) {
		if strings.TrimSpace(sent) == "" {
			continue
		}
		s.ttsMu.Lock()
		idx := 0
		if s.tts != nil {
			idx = s.tts.index
			s.tts.index++
		}
		s.ttsMu.Unlock()
		select {
		case jobs <- ttsJob{text: sent, index: idx}:
		default:
			// Kuyruk dolu: motor yavaş demektir, parçayı at (kullanıcı yine de
			// sonraki cümleleri duyar; bellek şişmesi daha kötü).
			s.hub.log.Warn("tts kuyruğu dolu, cümle atlandı")
		}
	}
}

// closeSpeechQueue kanalı kapatır: işleyici kalan işleri bitirip tts.done yollar.
// İki kez çağrılırsa güvenlidir (kanal zaten kapalıysa bir şey olmaz).
func (s *session) closeSpeechQueue() {
	s.ttsMu.Lock()
	jobs := s.ttsJobs
	s.ttsJobs = nil
	s.ttsMu.Unlock()
	if jobs != nil {
		close(jobs)
	}
}

// runSpeaker kuyruktaki cümleleri sırayla seslendirir; kanal kapanınca biter.
// Tek goroutine: tek GPU'da iki üretimi yarıştırmak yalnızca yavaşlatır.
func (s *session) runSpeaker(ctx context.Context, sp *speaker, jobs <-chan ttsJob, wait chan struct{}) {
	defer close(wait)
	var count int
	var total int64
	start := time.Now()
	for job := range jobs {
		if ctx.Err() != nil {
			return // kullanıcı kesti: yarım kalmış sesi at
		}
		speech, err := s.hub.tts.Speak(ctx, job.text)
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			code, msg, retry := ttsError(err)
			s.hub.log.Warn("ses üretilemedi", "err", err, "cümle", job.index)
			s.sendError(sp.convID, "", code, msg, retry)
			return // motor çalışmıyor; kalan cümleleri denemek anlamsız
		}
		s.send(protocol.NewOutbound(protocol.ServerTTSAudio, sp.convID, protocol.TTSAudioPayload{
			ConversationID: sp.convID,
			MessageID:      sp.msgID,
			Index:          job.index,
			Mime:           speech.Mime,
			Data:           base64.StdEncoding.EncodeToString(speech.Audio),
		}))
		count++
		total += int64(len(speech.Audio))
	}
	s.send(protocol.NewOutbound(protocol.ServerTTSDone, sp.convID, protocol.TTSDonePayload{
		ConversationID: sp.convID,
		MessageID:      sp.msgID,
		Sentences:      count,
		Bytes:          total,
		DurationMs:     time.Since(start).Milliseconds(),
	}))
}

// startSpeaker yeni bir seslendirme oturumu başlatır (öncekini keser).
// Bir konuşma bitmeden yenisi başlamasın diye eskisi iptal edilir.
func (s *session) startSpeaker(convID, messageID string) {
	if s.hub.tts == nil {
		return
	}
	s.stopSpeaking()
	ctx, cancel := context.WithCancel(s.ctx)
	jobs := make(chan ttsJob, 32)
	wait := make(chan struct{})
	sp := &speaker{convID: convID, msgID: messageID}
	s.ttsMu.Lock()
	s.tts = sp
	s.ttsJobs = jobs
	s.ttsWait = wait
	s.ttsCancel = cancel
	s.speechBuf = ""
	s.ttsMu.Unlock()
	go s.runSpeaker(ctx, sp, jobs, wait)
}

// stopSpeaking yarım kalan seslendirmeyi keser; kesildiyse true döner.
func (s *session) stopSpeaking() bool {
	s.ttsMu.Lock()
	cancel := s.ttsCancel
	s.ttsCancel = nil
	s.tts = nil
	s.speechBuf = ""
	jobs := s.ttsJobs
	s.ttsJobs = nil
	wait := s.ttsWait
	s.ttsWait = nil
	s.ttsMu.Unlock()
	if cancel == nil {
		return false
	}
	cancel() // mevcut cümle üretimi hemen durur
	if jobs != nil {
		close(jobs) // işleyici kalanları atıp çıkar
	}
	if wait != nil {
		<-wait // dışarı çıkmadan önce işleyici bitsin (kayıp olay olmasın)
	}
	return true
}

// ttsError, tts hatalarını WS hata gövdesine çevirir.
func ttsError(err error) (code, message string, retryable bool) {
	switch {
	case errors.Is(err, tts.ErrNoSynthesizer):
		return "not_implemented", "sesli çıkış yapılandırılmamış (TTS_URL)", false
	case errors.Is(err, tts.ErrEmptyText):
		return protocol.ErrCodeBadRequest, "okunacak metin yok", false
	case errors.Is(err, tts.ErrTooLong):
		return protocol.ErrCodeTooLarge, "seslendirilecek metin çok uzun", false
	case errors.Is(err, tts.ErrUnavailable):
		return protocol.ErrCodeLLM, "ses motoru yanıt vermedi", true
	default:
		return protocol.ErrCodeInternal, "ses üretilemedi", false
	}
}

// voiceError, stt hatalarını WS hata gövdesine çevirir.
func voiceError(err error) (code, message string, retryable bool) {
	switch {
	case errors.Is(err, stt.ErrNoRecognizer):
		return "not_implemented", "sesli giriş yapılandırılmamış (STT_URL)", false
	case errors.Is(err, stt.ErrNotStarted):
		return protocol.ErrCodeBadRequest, "kayıt başlatılmamış", false
	case errors.Is(err, stt.ErrEmptyAudio):
		return protocol.ErrCodeBadRequest, "ses kaydı boş", false
	case errors.Is(err, stt.ErrTooLong):
		return protocol.ErrCodeTooLarge, "ses kaydı çok uzun", false
	case errors.Is(err, stt.ErrUnavailable):
		return protocol.ErrCodeLLM, "konuşma tanıma motoru yanıt vermedi", true
	default:
		return protocol.ErrCodeInternal, "ses işlenemedi", false
	}
}

func (s *session) startGeneration(in protocol.Inbound, p protocol.TextMessagePayload) {
	select {
	case s.sem <- struct{}{}:
	default:
		s.sendError(in.ConversationID, in.ID, protocol.ErrCodeBusy, "bu bağlantıda çok fazla eşzamanlı yanıt üretiliyor", true)
		return
	}
	if in.ConversationID != "" {
		s.mu.Lock()
		s.owned[in.ConversationID] = struct{}{}
		s.mu.Unlock()
	}
	s.jobs.Add(1)
	go func() {
		defer s.jobs.Done()
		defer func() { <-s.sem }()
		defer func() {
			if r := recover(); r != nil {
				s.hub.log.Error("ws üretim işleyicisinde panik", "panic", r, "conversation", in.ConversationID)
				s.sendError(in.ConversationID, in.ID, protocol.ErrCodeInternal, "beklenmeyen sunucu hatası", true)
			}
		}()
		s.hub.agent.HandleText(s.ctx, agent.Input{
			UserID:         s.userID,
			ConversationID: in.ConversationID,
			RequestID:      in.ID,
			Payload:        p,
		}, s.send)
	}()
}