OpenAI GPT-Live Full-Duplex Voice System Deep Dive: Speech and Reasoning Layer Decoupling

"""
Full-Duplex Audio Pipeline Implementation
Simulating the core audio stream processing logic of GPT-Live
"""

import asyncio
import collections
import struct
import time
import wave
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Callable, Awaitable


class AudioState(Enum):
    """Audio stream states"""
    SILENCE = "silence"
    LISTENING = "listening"
    SPEAKING = "speaking"
    BARGE_IN = "barge_in"


@dataclass
class AudioFrame:
    """Audio frame data structure"""
    timestamp: float
    data: bytes
    sample_rate: int = 16000
    channels: int = 1
    sample_width: int = 2  # 16-bit PCM
    
    @property
    def duration_seconds(self) -> float:
        return len(self.data) / (self.sample_rate * self.channels * self.sample_width)
    
    @property
    def rms(self) -> float:
        """Calculate RMS energy of the audio frame"""
        if not self.data:
            return 0.0
        samples = struct.unpack_from(f"<{len(self.data)//2}h", self.data)
        sum_squares = sum(s * s for s in samples)
        return (sum_squares / len(samples)) ** 0.5


class AudioRingBuffer:
    """
    Audio ring buffer
    Manages continuously incoming audio data with sliding window reads
    """
    
    def __init__(self, max_duration_seconds: float = 30.0, sample_rate: int = 16000):
        self.sample_rate = sample_rate
        self.max_samples = int(max_duration_seconds * sample_rate)
        self.buffer = collections.deque(maxlen=self.max_samples)
        self._lock = asyncio.Lock()
    
    async def write(self, samples: list[int]) -> None:
        """Write audio samples"""
        async with self._lock:
            self.buffer.extend(samples)
    
    async def read(self, num_samples: int) -> list[int]:
        """Read the most recent N samples"""
        async with self._lock:
            if len(self.buffer) < num_samples:
                return list(self.buffer)
            return list(self.buffer)[-num_samples:]
    
    async def clear(self) -> None:
        async with self._lock:
            self.buffer.clear()
    
    @property
    async def duration_seconds(self) -> float:
        async with self._lock:
            return len(self.buffer) / self.sample_rate


class VoiceActivityDetector:
    """
    Voice Activity Detector (VAD)
    Based on energy levels and adaptive thresholds
    Simulates GPT-Live's built-in VAD logic (no external turn detector)
    """
    
    def __init__(
        self,
        frame_duration_ms: int = 30,
        silence_ratio_threshold: float = 0.3,
        min_speech_frames: int = 3,
        min_silence_frames: int = 15,  # ~450ms silence considered a pause
        energy_threshold: float = 100.0,
    ):
        self.frame_duration_ms = frame_duration_ms
        self.silence_ratio_threshold = silence_ratio_threshold
        self.min_speech_frames = min_speech_frames
        self.min_silence_frames = min_silence_frames
        self.energy_threshold = energy_threshold
        
        self._speech_frames = 0
        self._silence_frames = 0
        self._is_speech = False
        self._recent_frames: list[bool] = []
    
    def is_speech_frame(self, frame: AudioFrame) -> bool:
        """Determine if a frame contains speech"""
        return frame.rms > self.energy_threshold
    
    def process_frame(self, frame: AudioFrame) -> AudioState:
        """
        Process an audio frame and return the current state
        GPT-Live's core innovation: the model makes interaction decisions
        at every frame, rather than relying on a binary turn detector
        """
        is_speech = self.is_speech_frame(frame)
        self._recent_frames.append(is_speech)
        
        # Smoothing window
        window_size = 10
        if len(self._recent_frames) > window_size:
            self._recent_frames.pop(0)
        
        speech_ratio = sum(self._recent_frames) / len(self._recent_frames)
        
        if speech_ratio > self.silence_ratio_threshold:
            self._speech_frames += 1
            self._silence_frames = 0
        else:
            self._silence_frames += 1
            self._speech_frames = 0
        
        # State transitions
        if not self._is_speech and self._speech_frames >= self.min_speech_frames:
            self._is_speech = True
            return AudioState.LISTENING
        
        if self._is_speech and self._silence_frames >= self.min_silence_frames:
            self._is_speech = False
            return AudioState.SILENCE
        
        return AudioState.LISTENING if self._is_speech else AudioState.SILENCE
    
    def reset(self) -> None:
        self._speech_frames = 0
        self._silence_frames = 0
        self._is_speech = False
        self._recent_frames.clear()


class EchoCanceller:
    """
    Simplified echo canceller
    Conceptual implementation using adaptive NLMS filter
    """
    
    def __init__(self, filter_length: int = 512, mu: float = 0.01):
        self.filter_length = filter_length
        self.mu = mu
        self.weights = [0.0] * filter_length
        self.reference_buffer: list[float] = [0.0] * filter_length
    
    def process(self, mic_signal: list[float], ref_signal: list[float]) -> list[float]:
        """
        Process microphone signal, subtract echo component from reference signal
        mic_signal: microphone input
        ref_signal: speaker output reference
        """
        output = []
        for n in range(len(mic_signal)):
            # Update reference buffer
            self.reference_buffer.pop(0)
            self.reference_buffer.append(ref_signal[n] if n < len(ref_signal) else 0.0)
            
            # Estimate echo
            echo_estimate = sum(
                self.weights[i] * self.reference_buffer[self.filter_length - 1 - i]
                for i in range(self.filter_length)
            )
            
            # Error signal (echo-cancelled signal)
            error = mic_signal[n] - echo_estimate
            
            # Update filter weights (NLMS adaptation)
            norm = sum(x * x for x in self.reference_buffer)
            if norm > 1e-10:
                for i in range(self.filter_length):
                    self.weights[i] += (
                        self.mu * error * self.reference_buffer[self.filter_length - 1 - i] / norm
                    )
            
            output.append(error)
        
        return output


class FullDuplexAudioPipeline:
    """
    Full-duplex audio pipeline (complete implementation)
    Simulating GPT-Live's core audio stream processing
    """
    
    def __init__(
        self,
        sample_rate: int = 16000,
        frame_duration_ms: int = 20,  # 20ms per frame, matching Opus
        vad: Optional[VoiceActivityDetector] = None,
        echo_canceller: Optional[EchoCanceller] = None,
    ):
        self.sample_rate = sample_rate
        self.frame_duration_ms = frame_duration_ms
        self.frame_size = int(sample_rate * frame_duration_ms / 1000)
        
        self.vad = vad or VoiceActivityDetector()
        self.echo_canceller = echo_canceller or EchoCanceller()
        
        self.input_buffer = AudioRingBuffer(max_duration_seconds=60.0)
        self.output_buffer = AudioRingBuffer(max_duration_seconds=10.0)
        
        self.state = AudioState.SILENCE
        self._running = False
        self._on_audio_frame: Optional[Callable[[AudioFrame], Awaitable[None]]] = None
        self._on_state_change: Optional[Callable[[AudioState], Awaitable[None]]] = None
    
    async def push_input_frame(self, frame: AudioFrame) -> None:
        """
        Process incoming audio frame (from microphone)
        Core of full-duplex: input processing does not block output
        """
        # 1. Echo cancellation
        ref_samples = await self.output_buffer.read(self.frame_size)
        if ref_samples:
            mic_samples = struct.unpack_from(f"<{len(frame.data)//2}h", frame.data)
            cleaned = self.echo_canceller.process(
                list(mic_samples), ref_samples
            )
            frame.data = struct.pack(f"<{len(cleaned)}h", *[int(s) for s in cleaned])
        
        # 2. Write to input buffer
        samples = struct.unpack_from(f"<{len(frame.data)//2}h", frame.data)
        await self.input_buffer.write(list(samples))
        
        # 3. VAD detection (simulated - in GPT-Live this is built into the model)
        new_state = self.vad.process_frame(frame)
        if new_state != self.state:
            self.state = new_state
            if self._on_state_change:
                await self._on_state_change(new_state)
        
        # 4. Callback notification
        if self._on_audio_frame:
            await self._on_audio_frame(frame)
    
    async def push_output_frame(self, frame: AudioFrame) -> None:
        """
        Process output audio frame (from model-generated speech)
        Also written to output buffer for echo cancellation
        """
        samples = struct.unpack_from(f"<{len(frame.data)//2}h", frame.data)
        await self.output_buffer.write(list(samples))
    
    async def start(self) -> None:
        self._running = True
        self.state = AudioState.LISTENING
    
    async def stop(self) -> None:
        self._running = False
        await self.input_buffer.clear()
        await self.output_buffer.clear()
    
    def on_audio_frame(self, callback: Callable[[AudioFrame], Awaitable[None]]) -> None:
        self._on_audio_frame = callback
    
    def on_state_change(self, callback: Callable[[AudioState], Awaitable[None]]) -> None:
        self._on_state_change = callback

// Full-duplex WebRTC audio session management
// Simulating GPT-Live's Go-based media frontend

package media

import (
	"context"
	"encoding/binary"
	"io"
	"log"
	"sync"
	"time"

	"github.com/pion/webrtc/v4"
	"github.com/pion/rtp"
)

// AudioCodec defines audio codec parameters
type AudioCodec struct {
	SampleRate   int
	Channels     int
	FrameSize    int // Samples per frame
	Bitrate      int
}

// DefaultOpusCodec Opus default configuration (codec used by GPT-Live)
var DefaultOpusCodec = AudioCodec{
	SampleRate: 48000,
	Channels:   1,
	FrameSize:  960, // 20ms @ 48kHz
	Bitrate:    32000,
}

// AudioFrame represents an audio frame
type AudioFrame struct {
	Timestamp time.Time
	Sequence  uint16
	Data      []byte
	Duration  time.Duration
}

// AudioTrack wraps a WebRTC audio track
type AudioTrack struct {
	track      *webrtc.TrackLocalStaticSample
	codec      AudioCodec
	seqNum     uint16
	mu         sync.Mutex
	onFrame    func(AudioFrame)
	writeCh    chan AudioFrame
	ctx        context.Context
	cancel     context.CancelFunc
}

// NewAudioTrack creates a new audio track
func NewAudioTrack(codec AudioCodec) (*AudioTrack, error) {
	opusCodec := webrtc.RTPCodecCapability{
		MimeType:    webrtc.MimeTypeOpus,
		ClockRate:   uint32(codec.SampleRate),
		Channels:    uint16(codec.Channels),
	}

	track, err := webrtc.NewTrackLocalStaticSample(
		opusCodec, "audio", "voice-stream",
	)
	if err != nil {
		return nil, err
	}

	ctx, cancel := context.WithCancel(context.Background())
	return &AudioTrack{
		track:   track,
		codec:   codec,
		seqNum:  0,
		writeCh: make(chan AudioFrame, 256), // Buffer 256 frames
		ctx:     ctx,
		cancel:  cancel,
	}, nil
}

// WriteFrame writes an audio frame (non-blocking, sent by background goroutine)
func (at *AudioTrack) WriteFrame(frame AudioFrame) error {
	select {
	case at.writeCh <- frame:
		return nil
	default:
		// Buffer full, drop oldest frame (maintain real-time)
		select {
		case <-at.writeCh:
			// Drop one frame
		default:
		}
		at.writeCh <- frame
		return nil
	}
}

// Start begins the audio transmission loop
func (at *AudioTrack) Start() {
	go func() {
		ticker := time.NewTicker(20 * time.Millisecond)
		defer ticker.Stop()

		for {
			select {
			case <-at.ctx.Done():
				return
			case frame := <-at.writeCh:
				at.mu.Lock()
				at.seqNum++
				seq := at.seqNum
				at.mu.Unlock()

				// Build RTP packet
				rtpPacket := &rtp.Packet{
					Header: rtp.Header{
						Version:        2,
						PayloadType:    111,
						SequenceNumber: seq,
						Timestamp:      uint32(frame.Timestamp.UnixMicro() * at.codec.SampleRate / 1_000_000),
						SSRC:           0xABCDEF01,
						Marker:         false,
					},
					Payload: frame.Data,
				}

				raw, err := rtpPacket.Marshal()
				if err != nil {
					log.Printf("failed to marshal RTP packet: %v", err)
					continue
				}

				if _, err := at.track.Write(raw); err != nil {
					log.Printf("failed to write to track: %v", err)
				}

			case <-ticker.C:
				// Heartbeat: ensure connection stays active
			}
		}
	}()
}

// Stop stops the audio track
func (at *AudioTrack) Stop() {
	at.cancel()
}

// MediaSession represents a full-duplex media session
type MediaSession struct {
	pc          *webrtc.PeerConnection
	inputTrack  *webrtc.TrackRemote
	outputTrack *AudioTrack
	clientID    string
	onFrame     func(AudioFrame)
	mu          sync.RWMutex
}

// NewMediaSession creates a new full-duplex media session
func NewMediaSession(clientID string, api *webrtc.API) (*MediaSession, error) {
	config := webrtc.Configuration{
		ICEServers: []webrtc.ICEServer{
			{
				URLs: []string{"stun:stun.l.google.com:19302"},
			},
		},
	}

	pc, err := api.NewPeerConnection(config)
	if err != nil {
		return nil, err
	}

	session := &MediaSession{
		pc:       pc,
		clientID: clientID,
	}

	// Set up incoming track handler
	pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
		session.mu.Lock()
		session.inputTrack = track
		session.mu.Unlock()

		go session.handleIncomingTrack(track)
	})

	// Create output track
	outputTrack, err := NewAudioTrack(DefaultOpusCodec)
	if err != nil {
		pc.Close()
		return nil, err
	}

	if _, err := pc.AddTrack(outputTrack.track); err != nil {
		pc.Close()
		return nil, err
	}

	session.outputTrack = outputTrack
	return session, nil
}

// handleIncomingTrack processes incoming audio stream
func (ms *MediaSession) handleIncomingTrack(track *webrtc.TrackRemote) {
	// Jitter buffer: 200ms jitter tolerance
	jitterBuffer := NewJitterBuffer(200 * time.Millisecond)

	for {
		rtpPacket, _, err := track.ReadRTP()
		if err != nil {
			if err == io.EOF {
				return
			}
			log.Printf("read RTP error: %v", err)
			continue
		}

		frame := AudioFrame{
			Timestamp: time.Now(),
			Sequence:  rtpPacket.Header.SequenceNumber,
			Data:      rtpPacket.Payload,
			Duration:  20 * time.Millisecond,
		}

		// Process through jitter buffer
		jitterBuffer.Push(frame)
	}
}

// Start starts the media session
func (ms *MediaSession) Start() {
	ms.outputTrack.Start()
}

// Stop stops the media session
func (ms *MediaSession) Stop() {
	ms.outputTrack.Stop()
	ms.pc.Close()
}

// JitterBuffer handles network delay variation
// GPT-Live uses the same mechanism
type JitterBuffer struct {
	buffer    []AudioFrame
	maxSize   int
	cond      *sync.Cond
	mu        sync.Mutex
}

func NewJitterBuffer(maxDuration time.Duration) *JitterBuffer {
	maxSize := int(maxDuration / (20 * time.Millisecond))
	return &JitterBuffer{
		buffer:  make([]AudioFrame, 0, maxSize),
		maxSize: maxSize,
		cond:    sync.NewCond(&sync.Mutex{}),
	}
}

func (jb *JitterBuffer) Push(frame AudioFrame) {
	jb.mu.Lock()
	defer jb.mu.Unlock()

	if len(jb.buffer) >= jb.maxSize {
		jb.buffer = jb.buffer[1:]
	}
	jb.buffer = append(jb.buffer, frame)
	jb.cond.Signal()
}

func (jb *JitterBuffer) Pop() (AudioFrame, bool) {
	jb.mu.Lock()
	defer jb.mu.Unlock()

	for len(jb.buffer) == 0 {
		jb.cond.Wait()
	}

	frame := jb.buffer[0]
	jb.buffer = jb.buffer[1:]
	return frame, true
}

// Async Inference Scheduler
// Manages concurrent task scheduling between the frontend voice model
// and the backend reasoning model

package scheduler

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"
)

// DelegationRequest represents a delegation request
type DelegationRequest struct {
	ID        string
	SessionID string
	Query     string
	Context   *ConversationContext
	Tools     []ToolDefinition
	Effort    ReasoningEffort
	CreatedAt time.Time
	ResultCh  chan *DelegationResult
}

// DelegationResult represents a delegation result
type DelegationResult struct {
	RequestID   string
	Content     string
	ToolResults []ToolResult
	TokenUsage  TokenUsage
	Latency     time.Duration
	Error       error
}

// ReasoningEffort defines reasoning intensity
type ReasoningEffort int

const (
	EffortInstant ReasoningEffort = iota
	EffortMedium
	EffortHigh
)

// TokenUsage tracks token consumption
type TokenUsage struct {
	PromptTokens     int
	CompletionTokens int
	TotalTokens      int
}

// ToolDefinition defines a tool
type ToolDefinition struct {
	Name        string
	Description string
	Parameters  map[string]interface{}
}

// ToolResult represents a tool call result
type ToolResult struct {
	ToolName string
	Content  string
	Success  bool
}

// ConversationContext holds the conversation state
type ConversationContext struct {
	Messages     []Message
	TokenCount   int
	LastActivity time.Time
}

// Message represents a conversation message
type Message struct {
	Role    string
	Content string
	AudioID string
	Time    time.Time
}

// SessionState tracks a session's state
type SessionState struct {
	ID              string
	VoiceModelID    string
	FrontierModelID string
	Context         *ConversationContext
	PendingRequests map[string]*DelegationRequest
	CreatedAt       time.Time
	mu              sync.RWMutex
}

// InferenceScheduler manages async task delegation
// Core responsibility: scheduling async tasks between voice model
// and frontier reasoning model
type InferenceScheduler struct {
	mu            sync.RWMutex
	sessions      map[string]*SessionState
	frontierModel FrontierModelClient
	promptCache   *PromptCache
	workerPool    *WorkerPool
	maxRetries    int
}

// FrontierModelClient interface for frontier model inference
type FrontierModelClient interface {
	Infer(ctx context.Context, req *DelegationRequest) (*DelegationResult, error)
	PrefillContext(ctx context.Context, sessionID string, context *ConversationContext) error
	HealthCheck(ctx context.Context) bool
}

// NewInferenceScheduler creates a new inference scheduler
func NewInferenceScheduler(
	frontier FrontierModelClient,
	maxWorkers int,
	cacheSize int,
) *InferenceScheduler {
	return &InferenceScheduler{
		sessions:      make(map[string]*SessionState),
		frontierModel: frontier,
		promptCache:   NewPromptCache(cacheSize),
		workerPool:    NewWorkerPool(maxWorkers),
		maxRetries:    3,
	}
}

// RegisterSession registers a new session
// Pre-creates a frontier inference session and pre-fills context
// when the voice session starts
func (s *InferenceScheduler) RegisterSession(
	sessionID string,
	initialContext *ConversationContext,
) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	state := &SessionState{
		ID:              sessionID,
		Context:         initialContext,
		PendingRequests: make(map[string]*DelegationRequest),
		CreatedAt:       time.Now(),
	}

	// Pre-fill frontier model context (key GPT-Live optimization)
	// Ensures prompt is processed before first delegation request
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := s.frontierModel.PrefillContext(ctx, sessionID, initialContext); err != nil {
		log.Printf("failed to prefill context for session %s: %v", sessionID, err)
		// Non-fatal; continue registration
	}

	s.sessions[sessionID] = state
	log.Printf("registered session %s with prefilled frontier context", sessionID)
	return nil
}

// DelegateTask asynchronously delegates a task
// Core GPT-Live operation: voice model delegates complex tasks
// to the backend reasoning model
// Returns a channel allowing the voice model to continue talking while waiting
func (s *InferenceScheduler) DelegateTask(
	sessionID string,
	query string,
	tools []ToolDefinition,
	effort ReasoningEffort,
) (<-chan *DelegationResult, error) {
	s.mu.RLock()
	session, exists := s.sessions[sessionID]
	s.mu.RUnlock()

	if !exists {
		return nil, fmt.Errorf("session %s not found", sessionID)
	}

	req := &DelegationRequest{
		ID:        generateRequestID(),
		SessionID: sessionID,
		Query:     query,
		Context:   session.Context,
		Tools:     tools,
		Effort:    effort,
		CreatedAt: time.Now(),
		ResultCh:  make(chan *DelegationResult, 1),
	}

	session.mu.Lock()
	session.PendingRequests[req.ID] = req
	session.mu.Unlock()

	// Submit to worker pool asynchronously
	s.workerPool.Submit(func() {
		result := s.executeDelegationWithRetry(req)
		
		session.mu.Lock()
		delete(session.PendingRequests, req.ID)
		session.mu.Unlock()
		
		req.ResultCh <- result
		close(req.ResultCh)
	})

	return req.ResultCh, nil
}

// executeDelegationWithRetry executes delegation with retry logic
func (s *InferenceScheduler) executeDelegationWithRetry(req *DelegationRequest) *DelegationResult {
	var lastErr error
	
	for attempt := 0; attempt <= s.maxRetries; attempt++ {
		if attempt > 0 {
			backoff := time.Duration(100*attempt) * time.Millisecond
			time.Sleep(backoff)
		}

		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		start := time.Now()
		
		result, err := s.frontierModel.Infer(ctx, req)
		cancel()
		
		if err == nil {
			result.Latency = time.Since(start)
			return result
		}
		
		lastErr = err
		log.Printf("delegation attempt %d failed for request %s: %v",
			attempt+1, req.ID, err)
	}

	return &DelegationResult{
		RequestID: req.ID,
		Error:     fmt.Errorf("all retries failed: %v", lastErr),
	}
}

// GetSessionContext retrieves session context
func (s *InferenceScheduler) GetSessionContext(sessionID string) *ConversationContext {
	s.mu.RLock()
	defer s.mu.RUnlock()
	
	if session, exists := s.sessions[sessionID]; exists {
		return session.Context
	}
	return nil
}

// UpdateSessionContext updates session context
// Handles synchronization between background task results
// and real-time user corrections
func (s *InferenceScheduler) UpdateSessionContext(
	sessionID string,
	messages []Message,
) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	session, exists := s.sessions[sessionID]
	if !exists {
		return fmt.Errorf("session %s not found", sessionID)
	}

	session.mu.Lock()
	defer session.mu.Unlock()

	session.Context.Messages = append(session.Context.Messages, messages...)
	session.Context.TokenCount = calculateTokenCount(session.Context.Messages)
	session.Context.LastActivity = time.Now()

	return nil
}

// UnregisterSession removes a session
func (s *InferenceScheduler) UnregisterSession(sessionID string) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if session, exists := s.sessions[sessionID]; exists {
		session.mu.Lock()
		for _, req := range session.PendingRequests {
			close(req.ResultCh)
		}
		session.mu.Unlock()
		
		delete(s.sessions, sessionID)
		log.Printf("unregistered session %s", sessionID)
	}
}

// WorkerPool goroutine worker pool
type WorkerPool struct {
	workers chan struct{}
	wg      sync.WaitGroup
}

func NewWorkerPool(maxWorkers int) *WorkerPool {
	return &WorkerPool{
		workers: make(chan struct{}, maxWorkers),
	}
}

func (wp *WorkerPool) Submit(task func()) {
	wp.workers <- struct{}{}
	wp.wg.Add(1)
	go func() {
		defer func() {
			<-wp.workers
			wp.wg.Done()
		}()
		task()
	}()
}

func (wp *WorkerPool) Wait() {
	wp.wg.Wait()
}

// PromptCache caches prompt prefixes with KV state
type PromptCache struct {
	mu    sync.RWMutex
	items map[string]*CachedPrompt
	size  int
}

type CachedPrompt struct {
	Prefix    string
	KVState   []byte // Cached KV cache state
	CreatedAt time.Time
	HitCount  int
}

func NewPromptCache(size int) *PromptCache {
	return &PromptCache{
		items: make(map[string]*CachedPrompt),
		size:  size,
	}
}

func (pc *PromptCache) Get(key string) (*CachedPrompt, bool) {
	pc.mu.RLock()
	defer pc.mu.RUnlock()
	item, ok := pc.items[key]
	if ok {
		item.HitCount++
	}
	return item, ok
}

func (pc *PromptCache) Set(key string, item *CachedPrompt) {
	pc.mu.Lock()
	defer pc.mu.Unlock()
	
	if len(pc.items) >= pc.size {
		// LRU eviction
		var minKey string
		minHits := int(^uint(0) >> 1)
		for k, v := range pc.items {
			if v.HitCount < minHits {
				minHits = v.HitCount
				minKey = k
			}
		}
		delete(pc.items, minKey)
	}
	
	pc.items[key] = item
}

// Helper functions
func generateRequestID() string {
	return fmt.Sprintf("req_%d_%d", time.Now().UnixNano(), randInt(1000, 9999))
}

func randInt(min, max int) int {
	return min + int(time.Now().UnixNano()%int64(max-min+1))
}

func calculateTokenCount(messages []Message) int {
	total := 0
	for _, msg := range messages {
		total += len(msg.Content) / 4 // Rough estimate
	}
	return total
}

// Reasoning effort configuration
func (s *InferenceScheduler) getEffortConfig(effort ReasoningEffort) EffortConfig {
	switch effort {
	case EffortInstant:
		return EffortConfig{
			ModelName:       "gpt-5.5-instant",
			MaxTokens:       512,
			ReasoningEffort: 0,
			Temperature:     0.7,
			Timeout:         10 * time.Second,
		}
	case EffortMedium:
		return EffortConfig{
			ModelName:       "gpt-5.5-thinking",
			MaxTokens:       2048,
			ReasoningEffort: 1,
			Temperature:     0.5,
			Timeout:         30 * time.Second,
		}
	case EffortHigh:
		return EffortConfig{
			ModelName:       "gpt-5.5-thinking",
			MaxTokens:       4096,
			ReasoningEffort: 2,
			Temperature:     0.3,
			Timeout:         60 * time.Second,
		}
	default:
		return EffortConfig{}
	}
}

type EffortConfig struct {
	ModelName       string
	MaxTokens       int
	ReasoningEffort int
	Temperature     float64
	Timeout         time.Duration
}

// Context Manager
// Handles session context compaction, handoff, and state synchronization

package context

import (
	"context"
	"log"
	"sync"
	"time"
)

// ContextState represents context processing state
type ContextState int

const (
	StateActive    ContextState = iota
	StateCompacting
	StateReady
	StateSwitching
)

// SessionContext holds session context data
type SessionContext struct {
	ID              string
	Messages        []Message
	TokenCount      int
	KVState         []byte
	LastActivity    time.Time
	State           ContextState
	StateChangedAt  time.Time
	
	mu              sync.RWMutex
}

// Message represents a conversation message
type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
	Time    time.Time `json:"time"`
}

// ContextManager manages session contexts
type ContextManager struct {
	mu               sync.RWMutex
	sessions         map[string]*SessionContext
	maxContextTokens int
	compactionRatio  float64
	compactionFunc   CompactionFunc
}

// CompactionFunc defines the context compaction function signature
type CompactionFunc func(messages []Message, maxTokens int) ([]Message, int)

// NewContextManager creates a new context manager
func NewContextManager(
	maxContextTokens int,
	compactionRatio float64,
	compactionFunc CompactionFunc,
) *ContextManager {
	if compactionFunc == nil {
		compactionFunc = defaultCompactionFunc
	}
	
	return &ContextManager{
		sessions:         make(map[string]*SessionContext),
		maxContextTokens: maxContextTokens,
		compactionRatio:  compactionRatio,
		compactionFunc:   compactionFunc,
	}
}

// defaultCompactionFunc implements the default compaction strategy
// Strategy: keep system prompt and recent messages, summarize middle history
func defaultCompactionFunc(messages []Message, maxTokens int) ([]Message, int) {
	if len(messages) == 0 {
		return messages, 0
	}
	
	// Keep system prompt (first message)
	systemMsg := messages[0]
	
	// Keep recent N messages
	keepRecent := 10
	if keepRecent >= len(messages) {
		keepRecent = len(messages) - 1
	}
	
	compacted := make([]Message, 0, keepRecent+2)
	compacted = append(compacted, systemMsg)
	
	// Summarize middle messages
	midMessages := messages[1 : len(messages)-keepRecent]
	if len(midMessages) > 0 {
		summary := summarizeMessages(midMessages)
		compacted = append(compacted, Message{
			Role:    "system",
			Content: summary,
			Time:    time.Now(),
		})
	}
	
	// Append recent messages
	compacted = append(compacted, messages[len(messages)-keepRecent:]...)
	
	totalTokens := 0
	for _, msg := range compacted {
		totalTokens += len(msg.Content) / 4
	}
	
	return compacted, totalTokens
}

// summarizeMessages creates a summary of messages (simplified)
func summarizeMessages(messages []Message) string {
	if len(messages) == 0 {
		return ""
	}
	
	summary := "[Compressed conversation summary: "
	summary += "The conversation covers "
	if len(messages) <= 5 {
		summary += "a brief exchange"
	} else {
		summary += "an extended discussion"
	}
	summary += " with " + messages[0].Role + " starting the conversation"
	summary += " ending with " + messages[len(messages)-1].Role + "'s last message at "
	summary += messages[len(messages)-1].Time.Format("15:04:05")
	summary += "]"
	
	return summary
}

// RegisterSession registers a new session
func (cm *ContextManager) RegisterSession(id string, initialMessages []Message) {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	
	tokenCount := 0
	for _, msg := range initialMessages {
		tokenCount += len(msg.Content) / 4
	}
	
	cm.sessions[id] = &SessionContext{
		ID:             id,
		Messages:       initialMessages,
		TokenCount:     tokenCount,
		LastActivity:   time.Now(),
		State:          StateActive,
		StateChangedAt: time.Now(),
	}
	
	log.Printf("registered session context: %s (%d tokens)", id, tokenCount)
}

// AppendMessage appends a message to the session
func (cm *ContextManager) AppendMessage(sessionID string, msg Message) error {
	cm.mu.Lock()
	session, exists := cm.sessions[sessionID]
	cm.mu.Unlock()
	
	if !exists {
		return nil
	}
	
	session.mu.Lock()
	defer session.mu.Unlock()
	
	session.Messages = append(session.Messages, msg)
	session.TokenCount += len(msg.Content) / 4
	session.LastActivity = time.Now()
	
	// Check if compaction is needed
	if session.TokenCount > cm.maxContextTokens {
		go cm.CompactContext(sessionID)
	}
	
	return nil
}

// CompactContext asynchronously compacts context
// Core logic: keep original instance running while preparing new instance
func (cm *ContextManager) CompactContext(sessionID string) {
	cm.mu.RLock()
	session, exists := cm.sessions[sessionID]
	cm.mu.RUnlock()
	
	if !exists {
		return
	}
	
	session.mu.Lock()
	if session.State == StateCompacting || session.State == StateSwitching {
		session.mu.Unlock()
		return
	}
	session.State = StateCompacting
	session.StateChangedAt = time.Now()
	
	currentMessages := make([]Message, len(session.Messages))
	copy(currentMessages, session.Messages)
	session.mu.Unlock()
	
	// Execute compaction in background (does not block media path)
	targetTokens := int(float64(cm.maxContextTokens) * cm.compactionRatio)
	compactedMessages, newTokenCount := cm.compactionFunc(currentMessages, targetTokens)
	
	log.Printf("compacted session %s: %d -> %d tokens",
		sessionID, session.TokenCount, newTokenCount)
	
	// Update context
	session.mu.Lock()
	session.Messages = compactedMessages
	session.TokenCount = newTokenCount
	session.KVState = nil // KV cache invalidated, needs rebuild
	session.State = StateReady
	session.StateChangedAt = time.Now()
	session.mu.Unlock()
	
	// Notify scheduler to prepare new instance
	log.Printf("context compaction complete for session %s, ready for handoff", sessionID)
}

// GetSessionContext retrieves a snapshot of the session context
func (cm *ContextManager) GetSessionContext(sessionID string) *SessionContext {
	cm.mu.RLock()
	session, exists := cm.sessions[sessionID]
	cm.mu.RUnlock()
	
	if !exists {
		return nil
	}
	
	session.mu.RLock()
	defer session.mu.RUnlock()
	
	ctxCopy := &SessionContext{
		ID:            session.ID,
		Messages:      make([]Message, len(session.Messages)),
		TokenCount:    session.TokenCount,
		LastActivity:  session.LastActivity,
		State:         session.State,
	}
	copy(ctxCopy.Messages, session.Messages)
	
	return ctxCopy
}

// UnregisterSession removes a session
func (cm *ContextManager) UnregisterSession(sessionID string) {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	delete(cm.sessions, sessionID)
	log.Printf("unregistered session context: %s", sessionID)
}

"""
Extracting discrete messages from continuous speech streams
Simulating GPT-Live's speculative message queue design
"""

import asyncio
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class Speaker(Enum):
    USER = "user"
    ASSISTANT = "assistant"
    UNKNOWN = "unknown"


class MessageStatus(Enum):
    PROVISIONAL = "provisional"
    FINALIZED = "finalized"


@dataclass
class TranscriptSegment:
    text: str
    speaker: Speaker
    start_time: float
    end_time: float
    confidence: float
    is_acknowledgment: bool = False


@dataclass
class DiscreteMessage:
    id: str
    speaker: Speaker
    text: str
    start_time: float
    end_time: float
    status: MessageStatus
    segments: list[TranscriptSegment] = field(default_factory=list)
    is_acknowledgment: bool = False


class SpeculativeMessageQueue:
    """
    Speculative message queue
    Extracts discrete messages from continuous speech streams
    Maintains two views: speculative (UI) and authoritative (logging)
    """
    
    def __init__(
        self,
        finalize_after_silence_ms: float = 800.0,
        min_utterance_ms: float = 300.0,
        acknowledgment_threshold_ms: float = 500.0,
    ):
        self.finalize_after_silence_ms = finalize_after_silence_ms
        self.min_utterance_ms = min_utterance_ms
        self.acknowledgment_threshold_ms = acknowledgment_threshold_ms
        
        self._queue: list[DiscreteMessage] = []
        self._current_message: Optional[DiscreteMessage] = None
        self._last_segment_time: float = 0.0
        self._message_counter = 0
        
        self._speculative_view: list[DiscreteMessage] = []
        self._authoritative_view: list[DiscreteMessage] = []
    
    def process_segment(self, segment: TranscriptSegment) -> list[DiscreteMessage]:
        now = segment.start_time
        changes = []
        
        time_since_last = (now - self._last_segment_time) * 1000
        
        if time_since_last > self.finalize_after_silence_ms:
            if self._current_message and self._current_message.status == MessageStatus.PROVISIONAL:
                self._finalize_current_message()
                changes.append(self._current_message)
            
            self._start_new_message(segment)
            changes.append(self._current_message)
        
        elif self._current_message is None:
            self._start_new_message(segment)
            changes.append(self._current_message)
        
        elif segment.speaker != self._current_message.speaker:
            if self._current_message.status == MessageStatus.PROVISIONAL:
                if self._is_acknowledgment(segment):
                    segment.is_acknowledgment = True
                    self._current_message.segments.append(segment)
                    self._current_message.is_acknowledgment = True
                else:
                    self._finalize_current_message()
                    changes.append(self._current_message)
                    self._start_new_message(segment)
                    changes.append(self._current_message)
        
        else:
            self._current_message.segments.append(segment)
            self._current_message.text = self._build_text(self._current_message.segments)
            self._current_message.end_time = segment.end_time
            self._update_speculative_view()
        
        self._last_segment_time = segment.end_time
        return changes
    
    def _start_new_message(self, segment: TranscriptSegment) -> None:
        self._message_counter += 1
        self._current_message = DiscreteMessage(
            id=f"msg_{self._message_counter}",
            speaker=segment.speaker,
            text=segment.text,
            start_time=segment.start_time,
            end_time=segment.end_time,
            status=MessageStatus.PROVISIONAL,
            segments=[segment],
            is_acknowledgment=segment.is_acknowledgment,
        )
        self._queue.append(self._current_message)
        self._update_speculative_view()
    
    def _finalize_current_message(self) -> None:
        if self._current_message is None:
            return
        
        duration = (self._current_message.end_time - self._current_message.start_time) * 1000
        
        if duration < self.min_utterance_ms and not self._current_message.is_acknowledgment:
            self._queue.remove(self._current_message)
            self._update_speculative_view()
            self._current_message = None
            return
        
        self._current_message.status = MessageStatus.FINALIZED
        self._authoritative_view.append(self._current_message)
        self._current_message = None
    
    def _is_acknowledgment(self, segment: TranscriptSegment) -> bool:
        acknowledgment_words = {
            "mhmm", "uh-huh", "yeah", "okay", "got it", "right",
            "i see", "aha", "mm", "hmm", "嗯", "嗯哼", "好的",
        }
        return segment.text.strip().lower() in acknowledgment_words
    
    def _build_text(self, segments: list[TranscriptSegment]) -> str:
        return " ".join(seg.text for seg in segments)
    
    def _update_speculative_view(self) -> None:
        self._speculative_view = [
            msg for msg in self._queue
            if msg.status == MessageStatus.PROVISIONAL
        ]
    
    def get_speculative_view(self) -> list[DiscreteMessage]:
        return list(self._speculative_view)
    
    def get_authoritative_view(self) -> list[DiscreteMessage]:
        return list(self._authoritative_view)
    
    def get_all_messages(self) -> list[DiscreteMessage]:
        return list(self._queue)


# Usage demonstration
async def demo_speculative_queue():
    queue = SpeculativeMessageQueue()
    
    segments = [
        TranscriptSegment("well", Speaker.USER, 0.0, 0.3, 0.95),
        TranscriptSegment("I was wondering", Speaker.USER, 0.3, 0.8, 0.92),
        TranscriptSegment("how GPT-5.5 performs", Speaker.USER, 0.8, 1.8, 0.88),
        # 800ms silence → message finalized
        
        # Assistant starts answering
        TranscriptSegment("sure", Speaker.ASSISTANT, 2.8, 3.0, 0.97),
        TranscriptSegment("GPT-5.5 compared to previous", Speaker.ASSISTANT, 3.0, 3.8, 0.95),
        # User interrupts
        TranscriptSegment("specific numbers", Speaker.USER, 3.8, 4.2, 0.90),
    ]
    
    for seg in segments:
        changes = queue.process_segment(seg)
        if changes:
            for msg in changes:
                print(f"[{msg.status.value}] {msg.speaker.value}: {msg.text}")
    
    print(f"\nSpeculative view: {len(queue.get_speculative_view())} messages")
    print(f"Authoritative view: {len(queue.get_authoritative_view())} messages")
    
    for msg in queue.get_authoritative_view():
        print(f"  [{msg.speaker.value}] {msg.text}")


asyncio.run(demo_speculative_queue())

// Instant Connect Implementation
// Pre-negotiates SDP parameters for single-UDP-packet session startup

package transport

import (
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"log"
	"net"
	"time"
)

// PreNegotiatedSession represents a pre-negotiated session
type PreNegotiatedSession struct {
	SessionID     string
	Ufrag         string
	Pwd           string
	Fingerprint   string
	AudioCodec    string
	CreatedAt     time.Time
	ExpiresAt     time.Time
	IsUsed        bool
}

// InstantConnectManager manages instant connections
type InstantConnectManager struct {
	preSessions       map[string]*PreNegotiatedSession
	sessionTTL        time.Duration
	audioCodec        string
	serverUfrag       string
	serverPwd         string
	serverFingerprint string
}

// NewInstantConnectManager creates a new instant connect manager
func NewInstantConnectManager(
	sessionTTL time.Duration,
	audioCodec string,
) *InstantConnectManager {
	return &InstantConnectManager{
		preSessions:       make(map[string]*PreNegotiatedSession),
		sessionTTL:        sessionTTL,
		audioCodec:        audioCodec,
		serverUfrag:       generateICEUfrag(),
		serverPwd:         generateICEPwd(),
		serverFingerprint: generateFingerprint(),
	}
}

// GeneratePreSession generates a pre-negotiated session
// Completed before the user even clicks the button
func (m *InstantConnectManager) GeneratePreSession() *PreNegotiatedSession {
	sessionID := generateSessionID()
	
	session := &PreNegotiatedSession{
		SessionID:   sessionID,
		Ufrag:       generateICEUfrag(),
		Pwd:         generateICEPwd(),
		Fingerprint: m.serverFingerprint,
		AudioCodec:  m.audioCodec,
		CreatedAt:   time.Now(),
		ExpiresAt:   time.Now().Add(m.sessionTTL),
		IsUsed:      false,
	}
	
	m.preSessions[sessionID] = session
	log.Printf("generated pre-negotiated session: %s (expires %v)",
		sessionID, session.ExpiresAt)
	
	return session
}

// HandleFirstPacket handles the first UDP packet
// Server responds immediately when the client sends the first media packet
func (m *InstantConnectManager) HandleFirstPacket(
	conn *net.UDPConn,
	addr *net.UDPAddr,
	packet []byte,
) (*PreNegotiatedSession, error) {
	sessionID := extractSessionID(packet)
	
	session, exists := m.preSessions[sessionID]
	if !exists {
		return nil, nil // Fall back to standard signaling
	}
	
	if time.Now().After(session.ExpiresAt) {
		delete(m.preSessions, sessionID)
		return nil, nil
	}
	
	if session.IsUsed {
		return nil, nil
	}
	
	session.IsUsed = true
	
	// Respond immediately
	response := m.buildImmediateResponse(session)
	conn.WriteTo(response, addr)
	
	log.Printf("instant connect for session %s from %s", sessionID, addr)
	return session, nil
}

// buildImmediateResponse builds an immediate response
func (m *InstantConnectManager) buildImmediateResponse(
	session *PreNegotiatedSession,
) []byte {
	response := map[string]interface{}{
		"type":        "instant_connect",
		"session_id":  session.SessionID,
		"ufrag":       m.serverUfrag,
		"pwd":         m.serverPwd,
		"fingerprint": m.serverFingerprint,
		"codec":       session.AudioCodec,
		"timestamp":   time.Now().UnixMilli(),
	}
	
	data, _ := json.Marshal(response)
	return data
}

// Helper functions
func generateSessionID() string {
	b := make([]byte, 16)
	rand.Read(b)
	return hex.EncodeToString(b)
}

func generateICEUfrag() string {
	b := make([]byte, 8)
	rand.Read(b)
	return hex.EncodeToString(b)
}

func generateICEPwd() string {
	b := make([]byte, 22)
	rand.Read(b)
	return hex.EncodeToString(b)
}

func generateFingerprint() string {
	return "sha-256 " + generateSessionID()
}

func extractSessionID(packet []byte) string {
	if len(packet) < 4 {
		return ""
	}
	return string(packet[4:])
}