"""Async bridge from frozen live features to speaker-relative ASR turns."""
import asyncio,json,time,logging,re
from collections import deque
from pathlib import Path
import numpy as np
from speaker_runtime import LivePredictor
from speaker_turns import SpeakerMemory,SpeakerTurns,self_name,WAKE,wake_intent,choose_turn_text
from identity import unit

class SpeakerSession:
    def __init__(self,conversation,long_context='long',use_diarization=True):
        self.c=conversation;self.predictor=LivePredictor(long_context)
        config=Path(__file__).resolve().parent/'speaker_live/calibration.json'
        values=json.loads(config.read_text()) if config.exists() else {}
        self.memory=SpeakerMemory(threshold=float(values.get('threshold',.55)),margin=float(values.get('margin',.04)),
            center=self.predictor.fast.teacher_mean.detach().numpy().copy() if values.get('centered',True) else None)
        self.turns=SpeakerTurns(self.memory);self.tasks=set();self.queue=asyncio.Queue(maxsize=4)
        self.asr_lock=asyncio.Lock();self.worker=None;self.closed=False;self.last_notice=0.;self.last_barge=-100.
        self.pending_names={};self.decode_count=0;self.last_activity=None
        self.epoch=0;self.records=deque(maxlen=80);self.diar_offset=None;self.owners=deque(maxlen=2048)
        self.overlaps=deque(maxlen=512)
        self.asr_wakes=deque(maxlen=64)
        self.background_channels=set();self.channel_levels={};self.bridged_seconds=0.
        from diarization_client import DiarizationClient
        self.diar=DiarizationClient() if use_diarization else None
        self.diarization_active=False
        self.seen_channels=set();self.foreground_channel=None
        self.utterance_threshold=.50;self.utterance_margin=.08
        self.interrupt_checked_at=-100.;self.interrupt_reference=None
    def start(self):
        if self.worker is None:self.worker=asyncio.create_task(self.deliver_loop())
    async def close(self):
        self.closed=True
        for task in [*self.tasks,*([self.worker] if self.worker else [])]:task.cancel()
        await asyncio.gather(*self.tasks,*([self.worker] if self.worker else []),return_exceptions=True)
        if self.diar is not None:await self.diar.close()
        self.turns.clear();self.pending_names.clear();self.records.clear();self.owners.clear();self.overlaps.clear();self.asr_wakes.clear();self.background_channels.clear();self.channel_levels.clear()
        while not self.queue.empty():self.queue.get_nowait()
        self.worker=None;self.last_activity=None
        self.interrupt_reference=None
    async def clear(self):
        self.epoch+=1
        for task in list(self.tasks):task.cancel()
        await asyncio.gather(*list(self.tasks),return_exceptions=True)
        self.tasks.clear();self.turns.clear();self.pending_names.clear()
        while not self.queue.empty():self.queue.get_nowait()
        self.records.clear();self.diar_offset=None
        self.owners.clear()
        self.overlaps.clear()
        self.asr_wakes.clear()
        self.background_channels.clear();self.channel_levels.clear();self.bridged_seconds=0.
        self.seen_channels.clear();self.foreground_channel=None
        self.interrupt_reference=None;self.interrupt_checked_at=-100.
        if self.diar is not None:
            await self.diar.close()
            from diarization_client import DiarizationClient
            self.diar=DiarizationClient()
    def spawn(self,coroutine):
        task=asyncio.create_task(coroutine);self.tasks.add(task);task.add_done_callback(self.tasks.discard)
    async def notice(self,**fields):
        if not self.c.ws.closed:
            await self.c.ws.send_json({'type':'identity','enabled':self.c.identity_enabled,'mode':self.c.identity_mode,
                **self.turns.status(),'diarization_active':self.diarization_active,'utterance_threshold':self.utterance_threshold,**fields})
            if fields.get('warning'):await self.c.ws.send_json({'type':'warning','text':fields['warning']})
    async def transcribe(self,wave):
        if len(wave)<2560:return ''
        # A known person's short "yes" is valid. Right padding is silence,
        # never borrowed background audio from a different speaker.
        wave=np.pad(wave,(0,max(0,10240-len(wave))))
        async with self.asr_lock:
            self.decode_count+=1
            texts=await asyncio.to_thread(self.c.ears.service.engine.asr.transcribe,[np.asarray(wave,np.float32)])
            return (texts[0] if texts else '').strip()[:3000]

    def note_asr(self,text,at):
        # Keep explicit, same-name activation evidence across ASR revisions.
        # A later full-window decode may lose "Hey" although the incremental
        # decoder already recognized it. This is not an LLM-inferred wake.
        name=self_name(text)
        activated=wake_intent(text)
        if name and activated:self.asr_wakes.append((float(at),name.casefold()))
        active=[t for t in self.turns.tracks.values() if t.sid and not t.closed and at-t.last_speech<1.6]
        if len(active)==1:
            track=active[0]
            if activated:track.wake=True
            user=self.memory.users.get(track.sid)
            same_name=bool(name and (self_name(track.partial) or (user.name if user and user.named else '')).casefold()==name.casefold())
            # A room-wide partial cannot give a background-channel Guest the
            # name spoken by a different, not-yet-diarized foreground person.
            if same_name and (activated or track.wake):track.name_heard=True

    def observed_wake(self,track,text):
        name=self_name(text)
        return bool(name and any(track.start-.32<=at<=track.last_speech+.96 and value==name.casefold()
            for at,value in self.asr_wakes))

    async def confirm_interrupt(self,proposal,hop_end):
        """Confirm a native GRU proposal with known-user vs actual bot voice.

        No enrollment, ASR or invented BARGE. A reusable diarization channel
        alone is not sufficient: echoes can occupy an old user's channel.
        """
        end=hop_end/self.c.ears.sr
        if end-self.interrupt_checked_at<.30:return None
        self.interrupt_checked_at=end
        references=[(u.id,u.teacher.copy()) for u in self.memory.users.values() if u.teacher is not None]
        if not references:return None
        begin=max(0.,end-1.92,float(getattr(self.c,'speaker_reply_start_s',0.)))
        wave=self.audio_between(begin,end)
        if len(wave)<7680:return None
        # Remove quiet leading leakage before speaker verification, not from
        # the captured turn or native GRU input. Double talk is not separated.
        n=len(wave)//320;levels=np.sqrt(np.mean(wave[:n*320].reshape(n,320)**2,axis=1))
        active=np.flatnonzero(levels>=max(.003,float(levels.max())*.25))
        if len(active)<24:return None # at least 480 ms of audible evidence
        query=wave[max(0,int(active[0])*320-160):min(len(wave),(int(active[-1])+1)*320+160)]
        query=np.pad(query,(0,max(0,10240-len(query))))
        playback=self.c.playback;rid=playback.rid;epoch=self.epoch
        size=min(playback.n16,4*16000)
        reference=playback.audio[:size].copy()
        cached=self.interrupt_reference
        def score():
            from speaker_teacher import get_cached_teacher
            teacher=get_cached_teacher()
            reuse=bool(cached and cached[0]==rid and size<cached[1]*1.8)
            # While waiting for admission/first TTS audio there is no new
            # playback to echo. Retain the previous bot cue for acoustic tails
            # if available, without preventing known users cancelling a wait.
            bot=(cached[2] if cached else None) if size<10240 else cached[2] if reuse else unit(teacher.embed(reference))
            user=unit(teacher.embed(query))
            values=sorted(((float(user@v),sid) for sid,v in references),reverse=True)
            return bot,cached[1] if reuse else size,float(user@bot) if bot is not None else -1.,values
        try:bot,bot_size,bot_score,values=await asyncio.to_thread(score)
        except Exception:return None
        if self.closed or self.epoch!=epoch or not self.c.speaking or playback.rid!=rid:return None
        if size>=10240:self.interrupt_reference=(rid,bot_size,bot)
        best,sid=values[0];runner=max(bot_score,values[1][0] if len(values)>1 else -1.)
        # Preserve quick, strong native interruptions when the already-bound
        # person's independent fast cue agrees. Use the same bounded acoustic
        # continuity floor as full-turn verification, never for new users.
        recent=self.turns.tracks.get(sid);user=self.memory.users.get(sid)
        recently_owned=bool(recent is not None and not recent.closed and end-recent.last_speech<=.64
            and recent.speech_s>=.48)
        # A short ECAPA window alone can confuse two male voices. Require
        # independent live ownership, or a much stronger full known-voice match
        # when the diarizer assigned a genuinely new channel.
        accepted=bool(best>=self.utterance_threshold and best-runner>=self.utterance_margin
            and (recently_owned or (best>=max(.70,self.utterance_threshold) and best-runner>=.20)))
        continuity=bool(float(proposal.get('yield_prob',0))>=.8 and recent is not None and user is not None
            and not recent.closed and end-recent.last_speech<=.64 and len(recent.fast)>=2
            and self.memory.similarity(recent.vector(),user.fast)>=self.memory.threshold-.12
            and best>=max(.4,self.utterance_threshold-.08) and best-runner>=max(.12,self.utterance_margin))
        accepted=accepted or continuity
        if getattr(self.c,'synthetic_diagnostics',False):
            await self.c.ws.send_json({'type':'synthetic_barge_verification','accepted':accepted,
                'user_cosine':best,'bot_cosine':bot_score,'margin':best-runner,'evidence_seconds':len(query)/16000,
                'strong_native_continuity':continuity})
        return sid if accepted and sid in self.memory.users else None

    async def playback_voice(self):
        playback=getattr(getattr(self,'c',None),'playback',None)
        if playback is None or playback.n16<10240:return None
        rid=playback.rid;size=min(playback.n16,64000);epoch=self.epoch
        cached=self.interrupt_reference
        if cached and cached[0]==rid and size<cached[1]*1.8:return cached[2]
        reference=playback.audio[:size].copy()
        def embed():
            from speaker_teacher import get_cached_teacher
            return unit(get_cached_teacher().embed(reference))
        try:bot=await asyncio.to_thread(embed)
        except Exception:return None
        if self.closed or self.epoch!=epoch or playback.rid!=rid:return None
        self.interrupt_reference=(rid,size,bot);return bot

    async def transcribe_turn(self,track):
        wave=self.submission_wave(track)
        start=max(track.start,track.recognition_start or track.start)
        # Post-roll rescues word endings, but must not turn quiet trailing TV
        # into the enrollment voiceprint. Identity keeps the original span.
        identity_length=max(0,round((track.last_speech-start)*16000))
        track.verification_audio=wave[:identity_length].copy()
        # A reused channel can become confident only after the wake phrase.
        # Decode several bounded views of that one-second pre-roll as ONE
        # native batch. Prefix interference otherwise hides a perfectly
        # audible self-introduction and silently maps the new person to the
        # old channel. Never invent a wake/name or remove ordinary content.
        preroll=(track.voiced_start-track.start) if track.voiced_start is not None else 0.
        offsets=[0.,.16,.32,.80] if preroll>=.8 and len(wave)>=25600 else [0.]
        views=[np.pad(wave[round(offset*16000):],(0,max(0,10240-len(wave[round(offset*16000):])))) for offset in offsets]
        owned=track.owned_waveform();owned=np.pad(owned,(0,max(0,10240-len(owned))))
        def decode():
            output=[];batch=views+[owned]
            for at in range(0,len(batch),4):output.extend(self.c.ears.service.engine.asr.transcribe(batch[at:at+4]))
            return output
        async with self.asr_lock:
            self.decode_count+=len(views)+1
            results=await asyncio.to_thread(decode)
        texts=[str(text).strip()[:3000] for text in results[:len(views)]]
        owned_text=str(results[-1]).strip()[:3000] if results else ''
        text,chosen=choose_turn_text(texts,owned_text)
        track.asr_crop_verified=bool(text and chosen is not None and chosen>0)
        if chosen is not None and chosen>0:track.submission_audio=views[chosen].copy()
        if chosen==-1:track.submission_audio=owned.copy()
        if getattr(self.c,'synthetic_diagnostics',False):
            await self.c.ws.send_json({'type':'synthetic_asr_views','texts':texts,'owned_text':owned_text,
                'offsets':offsets,'chosen':chosen,'word_origin_consistent':text is not None})
        return text
    async def candidate(self,track,wave,coverage=0.,continuous=False):
        try:
            cut=0.
            if continuous and len(wave)>=25600:
                offsets=[0.,.32,.64,.96]
                views=[np.pad(wave[round(offset*16000):],(0,max(0,10240-len(wave[round(offset*16000):])))) for offset in offsets]
                async with self.asr_lock:
                    self.decode_count+=len(views)
                    texts=await asyncio.to_thread(self.c.ears.service.engine.asr.transcribe,views)
                # Trim only verified pre-wake context; take the latest view
                # that STILL contains the greeting, never fabricate one.
                choices=[i for i,text in enumerate(texts) if wake_intent(text) and self_name(text)]
                if not choices:choices=[i for i,text in enumerate(texts) if wake_intent(text)]
                chosen=choices[-1] if choices else 0;text=(texts[chosen] if texts else '').strip()[:3000];cut=offsets[chosen]
            else:text=await self.transcribe(wave)
            if self.closed or not self.c.identity_enabled:return
            was_background=track.channel in self.background_channels
            recovered_start=None
            if continuous and coverage>=.4 and not was_background and self_name(text) and not wake_intent(text):
                recovered=await self.recover_intro_prefix(track,text)
                if recovered is not None:text,recovered_start=recovered
            background_wake_only=track.channel in self.background_channels and wake_intent(text) and not self_name(text)
            eligible=coverage>=.4
            if was_background and wake_intent(text) and self_name(text) and eligible and self.records:
                raw=self.audio_between(max(track.start,track.last_speech-8.),min(self.records[-1][1],track.last_speech+.8))
                room_text=await self.transcribe(raw)
                eligible=bool(self_name(room_text)==self_name(text) and wake_intent(room_text))
            # A long-observed TV channel can briefly own a new person's "Hey"
            # during a channel transition. Wait for an addressed complete
            # introduction instead of enrolling the TV as a permanent Guest.
            observed_wake=not was_background and continuous and coverage>=.4 and self.observed_wake(track,text)
            decision=self.turns.candidate_text(track.key,'' if background_wake_only or not eligible else text,observed_wake=observed_wake and eligible)
            if not decision and not wake_intent(text) and len(text.split())>=2 and track.speech_s>=1.28:
                decision=await self.recover_existing_channel(track,was_background)
            if not decision and track.channel is not None and track.channel not in self.turns.channels:
                if coverage<.35 or (not wake_intent(text) and len(text.split())>=2 and track.speech_s>=1.28):
                    self.background_channels.add(track.channel)
            if decision and continuous:
                if recovered_start is not None:
                    track.start=min(track.start,recovered_start);track.recognition_start=recovered_start
                else:track.recognition_start=track.start+cut
            await self.c.ws.send_json({'type':'asr_preview','text':text,
                    'status':'enrolled' if decision else 'awaiting_wake_or_stable_voice',
                    'source':'speaker_candidate','track':track.key,'audio_seconds':round(len(wave)/16000,2),
                    'voiced_seconds':round(track.speech_s,2),'channel':track.channel,
                    'energy_coverage':round(coverage,3),'continuous_audio':continuous})
            if decision:await self.notice(decision=decision)
        except asyncio.CancelledError:raise
        except Exception:
            track.decoding=False
            await self.notice(warning='Speaker-owned ASR was unavailable; enrollment not guessed.')

    async def recover_intro_prefix(self,track,text):
        """Verify a real greeting immediately before a new channel's name.

        A channel transition can split "Hey Robbie" from "I am ...". Do not
        invent a wake or transfer an unrelated room-wide partial: re-decode
        the bounded original microphone prefix and require the SAME explicit
        self-introduction. Only energy-eligible non-background candidates call
        this, and final owned-word/acoustic checks remain mandatory.
        """
        if not self.records:return None
        start=max(self.records[0][0],track.start-1.6)
        end=min(self.records[-1][1],track.last_speech+.8)
        if any(a<end and b>start for a,b in self.overlaps):return None
        wave=self.audio_between(start,end)
        if len(wave)<10240:return None
        offsets=[0.,.32,.64,.96]
        views=[np.pad(wave[round(offset*16000):],(0,max(0,10240-len(wave[round(offset*16000):])))) for offset in offsets]
        async with self.asr_lock:
            self.decode_count+=len(views)
            texts=await asyncio.to_thread(self.c.ears.service.engine.asr.transcribe,views)
        name=self_name(text)
        choices=[i for i,value in enumerate(texts) if wake_intent(value) and self_name(value)==name]
        if not choices:return None
        chosen=choices[-1]
        return texts[chosen].strip()[:3000],start+offsets[chosen]

    async def recover_existing_channel(self,track,was_background=False):
        """A returning named voice may acquire a new diarizer channel.

        Reassociate a known, acoustically verified identity only; never enroll
        an unrelated speaker from ordinary speech without a wake phrase.
        """
        if self.c.speaking or track.channel is None:return None
        references=[u for u in self.memory.users.values() if u.named and u.teacher is not None]
        if not references:return None
        query=track.owned_waveform()[-40960:]
        if len(query)<20480:return None
        levels=np.sqrt(np.mean(query[:len(query)//320*320].reshape(-1,320)**2,axis=1))
        if was_background:
            history=self.channel_levels.get(track.channel,deque())
            floor=float(np.median(history)) if history else 0.
            if float(np.quantile(levels,.75))<max(.008,floor*2.5):return None
        from speaker_teacher import get_cached_teacher
        epoch=self.epoch
        def embed():return unit(get_cached_teacher().embed(query))
        try:embedding=await asyncio.to_thread(embed)
        except Exception:return None
        if self.closed or epoch!=self.epoch or track.key not in self.turns.tracks:return None
        scores=sorted(((float(embedding@u.teacher),u.id) for u in references),reverse=True)
        best,sid=scores[0];runner=scores[1][0] if len(scores)>1 else -1.
        if best<max(.55,self.utterance_threshold) or best-runner<max(.12,self.utterance_margin):return None
        self.rebind(track,sid)
        # A channel that was already observed as TV/background cannot retain
        # this person's identity after a verified move to a new channel.
        for channel,who in list(self.turns.channels.items()):
            if who==sid and channel!=track.channel and channel in self.background_channels:self.turns.channels.pop(channel,None)
        self.background_channels.discard(track.channel)
        return {'kind':'enrolled','id':sid,'name':self.memory.users[sid].name,'new_user':False,
            'accepted':True,'source':'Full known-voice match on a new diarization channel','cosine':best}
    async def observe(self,event,hop_end):
        c=self.c;at=hop_end/c.ears.sr
        prediction=await asyncio.to_thread(self.predictor.predict,event)
        begin=at-len(event['audio16'])/16000
        self.records.append((begin,at,prediction,event['audio16'].copy()))
        diarized=False
        if self.diar is not None:
            if self.diar_offset is None:self.diar_offset=begin
            try:
                response=await self.diar.feed(event['audio16']);self.diarization_active=True;diarized=True
                values=np.asarray(response['probabilities'],np.float32).reshape(-1,8)
                if len(values):
                    labels=[]
                    for row in values:
                        active=[int(ch) for ch in np.flatnonzero(row>=.55)]
                        new=[ch for ch in active if ch not in self.seen_channels]
                        self.seen_channels.update(active)
                        if len(new)==1:self.foreground_channel=new[0]
                        known=[ch for ch in active if self.turns.channels.get(ch) in self.memory.users]
                        known_ids={self.turns.channels[ch] for ch in known}
                        if len(known_ids)==1:label=max(known,key=lambda ch:row[ch])
                        elif len(known_ids)>1:label=-1 # two enrolled people: don't guess mixed words
                        elif self.foreground_channel in active:label=self.foreground_channel
                        elif len(active)==1:label=active[0]
                        else:label=-1
                        labels.append(label)
                    labels=np.asarray(labels)
                    boundaries=np.r_[0,np.flatnonzero(np.diff(labels)!=0)+1,len(labels)]
                    for a,b in zip(boundaries[:-1],boundaries[1:]):
                        start=self.diar_offset+(response['start_frame']+int(a))*.01
                        end=self.diar_offset+(response['start_frame']+int(b))*.01
                        source=next((p for s,t,p,x in self.records if s<end<=t+.0001),prediction)
                        cue=dict(source);label=int(labels[a]);wave=self.audio_between(start,end)
                        bridged=self.bridge_gap(label,cue,wave,start,end,values[a:b].mean(0))
                        if bridged is not None:
                            label=bridged;cue['provisional_bridge']=True;self.bridged_seconds+=end-start
                        if label>=0:
                            cue.update(kind='speech',confidence=max(.65,cue.get('confidence',0.)),diarization_channel=label)
                        else:
                            cue.update(kind='silence',confidence=0.,embedding=None)
                            # Ambiguous overlaps do not put mixed speech into
                            # ASR, but a genuinely active enrolled person has
                            # not yet stopped talking just because they overlap.
                            active=np.flatnonzero(values[a:b].mean(0)>=.55)
                            if len(active)>1:
                                known_ids=[self.turns.channels.get(int(ch)) for ch in active
                                    if self.turns.channels.get(int(ch)) in self.memory.users]
                                if len(set(known_ids))>1:self.overlaps.append((start,end))
                                for ch in active:
                                    sid=self.turns.channels.get(int(ch));track=self.turns.tracks.get(sid)
                                    if track is not None:track.last_speech=end
                        await self.step(cue,wave,end)
            except asyncio.CancelledError:raise
            except Exception:
                await self.diar.close();self.diar=None;self.diarization_active=False
                await c.ws.send_json({'type':'warning','text':'Diarization unavailable: using the less reliable V4 embedding-only fallback. Speaker strictness is adjustable.'})
        if not diarized:await self.step(prediction,event['audio16'],at)
        if time.monotonic()-self.last_notice>.65:
            self.last_notice=time.monotonic();await self.notice(prediction=self.last_activity or {'kind':prediction['kind']})

    def bridge_gap(self,label,cue,wave,start,end,probabilities):
        """Repair a short foreground-channel dropout, never permanent binding.

        Only established quiet background channels (or silence with a strong
        speech cue) are eligible. Require louder energy, same recent fast
        speaker vector and <=1.6 s since DIRECT diarizer evidence. This avoids
        both splitting a sentence and letting room noise hold its clock open.
        Full-utterance identity and word-origin checks still decide submission.
        """
        rms=float(np.sqrt(np.mean(wave*wave))) if len(wave) else 0.
        previous=self.channel_levels.get(label,deque())
        floor=float(np.median(previous)) if previous else 0.
        if label>=0:
            self.channel_levels.setdefault(label,deque(maxlen=128)).append(rms)
            if label not in self.background_channels or label in self.turns.channels:return None
        elif cue.get('kind')!='speech' or cue.get('confidence',0)<.8:return None
        vector=cue.get('raw_embedding',cue.get('embedding'))
        if vector is None:return None
        active=[t for t in self.turns.tracks.values() if t.sid in self.memory.users and t.channel is not None
            and not t.closed and t.fast and end-t.last_direct_speech<=1.6 and start>=t.start]
        if len(active)!=1:return None
        track=active[0];reference=max(track.levels,default=.01)
        similarity=self.memory.similarity(vector,track.vector())
        if rms<max(.004,floor*2.5,reference*.3) or similarity<.55:return None
        if probabilities[track.channel]<.15 and similarity<.75:return None
        return track.channel

    def audio_between(self,start,end):
        output=np.zeros(max(0,round((end-start)*16000)),np.float32)
        for a,b,_,wave in self.records:
            lo=max(start,a);hi=min(end,b)
            if hi<=lo:continue
            at=round((lo-start)*16000);source=round((lo-a)*16000);n=min(round((hi-lo)*16000),len(output)-at,len(wave)-source)
            if n>0:output[at:at+n]=wave[source:source+n]
        return output

    def submission_wave(self,track):
        """Keep phonemes through short diarizer dropouts, not source separation.

        Channel activity owns the *turn clock*, but zeroing every uncertain
        10-ms frame corrupts wake words/names and biases ECAPA. Reconstruct the
        bounded continuous sentence, excluding another enrolled user's frames
        and ambiguous enrolled overlap. Freeze it before the RAM ring rotates.
        """
        if track.submission_audio is not None:return track.submission_audio
        start=max(track.start,track.recognition_start or track.start)
        # The diarizer can stop owning a quiet final consonant before the
        # sentence ends. Use already-received quiet-clock audio as bounded
        # post-roll; do not delay the speaker-relative one-second endpoint.
        # Other known users/ambiguous overlaps are still masked below, and
        # independent owned/raw recognition still guards background words.
        available=self.records[-1][1] if self.records else track.last_speech
        end=min(available,track.last_speech+.80)
        wave=self.audio_between(start,end)
        provisional=self.memory.users.get(track.sid)
        introduction=bool(provisional and provisional.teacher is None and track.wake and self_name(track.partial))
        excluded=[] if introduction else [(owner[0],owner[1]) for owner in self.owners if owner[2] and owner[2]!=track.sid]
        excluded.extend(self.overlaps)
        for a,b in excluded:
            lo=max(0,round((a-start)*16000));hi=min(len(wave),round((b-start)*16000))
            if hi>lo:wave[lo:hi]=0
        return wave if len(wave) else track.waveform()

    def candidate_wave(self,track):
        selected=track.owned_waveform().copy();continuous=self.submission_wave(track).copy()
        energy=float(np.sum(continuous.astype(np.float64)**2))
        coverage=min(1.,float(np.sum(selected.astype(np.float64)**2))/max(energy,1e-10))
        # Only the candidate that owns most of the acoustic energy may repair
        # its internal channel gaps. A quiet TV/background channel cannot
        # borrow a louder person's wake phrase from the same raw microphone.
        coherent=bool(len(track.fast)>=2 and np.mean([self.memory.similarity(v,track.vector()) for v in track.fast])>=.8)
        competitor=max((float(np.sum(t.owned_waveform().astype(np.float64)**2)) for t in self.turns.tracks.values()
            if t is not track and not t.closed),default=0.)
        dominant=float(np.sum(selected.astype(np.float64)**2))>=1.25*competitor
        use_continuous=energy>1e-8 and (coverage>=.65 or
            (coverage>=.4 and coherent and dominant and track.channel not in self.background_channels))
        if use_continuous:
            # A coherent energy-winning new foreground can span an old/new
            # diarizer channel transition. Recover its full greeting; known
            # name/acoustic checks still decide whether enrollment is valid.
            if self.records:
                end=min(self.records[-1][1],track.last_speech+.8)
                continuous=self.audio_between(max(track.start,track.last_speech-8.),end)
                # Eligibility must describe the audio actually sent to ASR,
                # not an easier masked view that omitted a competing voice.
                energy=float(np.sum(continuous.astype(np.float64)**2))
                coverage=min(1.,float(np.sum(selected.astype(np.float64)**2))/max(energy,1e-10))
                if coverage<.4:use_continuous=False
        return (continuous if use_continuous else selected),coverage,use_continuous

    def rebind(self,track,sid):
        old=track.sid;spans=self.turns.rebind_channel(track.channel,sid) if track.channel is not None else []
        spans.append((old,track.voiced_start if track.voiced_start is not None else track.start,track.last_speech))
        updated=deque(maxlen=self.owners.maxlen)
        for owner in self.owners:
            a,b,who=owner[:3];channel=owner[3] if len(owner)>3 else track.channel
            if channel==track.channel and any(who==previous and a<end and b>start for previous,start,end in spans):who=sid
            updated.append((a,b,who,channel))
        self.owners=updated;track.sid=sid

    async def step(self,prediction,wave,at):
        c=self.c
        old_keys=set(self.turns.tracks)
        match=self.turns.update(prediction,wave,at,speaking=c.speaking)
        for key in set(self.turns.tracks)-old_keys:
            track=self.turns.tracks[key];begin=at-len(wave)/16000
            pre=max(self.records[0][0] if self.records else begin,min(track.start,begin-1.))
            # Preserve a clipped initial "Hey" while avoiding another known
            # person's preceding words. The pre-roll is not identity evidence.
            # The streaming diarizer can recognize a new foreground channel
            # late. Preserve its wake phrase, but never borrow a previous
            # enrolled person's words (including already closed turns).
            for owner in self.owners:
                a,b,sid=owner[:3]
                if sid and sid!=track.sid and b>pre and a<begin:pre=max(pre,b)
            conflict=any(t.sid and t.key!=key and t.last_speech>pre for t in self.turns.tracks.values())
            if pre<begin and not conflict:
                track.chunks.appendleft((pre,begin,self.audio_between(pre,begin)));track.start=pre
        self.owners.append((at-len(wave)/16000,at,match.get('id') if match else None,prediction.get('diarization_channel')))
        self.last_activity=match
        # Original TYGER/second-ear proposals own interruptions. Identity only
        # gates/replays a recent native proposal in server.py; it must never
        # create its own barge from a diarizer channel (including bot echo).
        for track in self.turns.decode_candidates():
            wave,coverage,continuous=self.candidate_wave(track)
            self.spawn(self.candidate(track,wave,coverage,continuous))
        for track in self.turns.due():
            try:
                track.submission_audio=self.submission_wave(track)
                self.queue.put_nowait(track)
            except asyncio.QueueFull:await c.ws.send_json({'type':'warning','text':'Several people spoke at once; please repeat the last turn.'})
    async def deliver_loop(self):
        try:
            while True:
                track=await self.queue.get()
                epoch=self.epoch
                try:
                    if getattr(self.c,'synthetic_diagnostics',False):
                        import base64
                        await self.c.ws.send_json({'type':'synthetic_audio_debug','start':track.start,'end':track.last_speech,
                            'audio':base64.b64encode(self.submission_wave(track).astype('<f2').tobytes()).decode(),
                            'track':track.key,'channel':track.channel})
                    text=await self.transcribe_turn(track)
                    if self.closed or not self.c.identity_enabled or epoch!=self.epoch:continue
                    if text is None:
                        await self.c.ws.send_json({'type':'ignored','reason':'Speech could not be separated confidently from background words. Please repeat; nothing was sent to Luna.'})
                        continue
                    text=text or track.partial
                    if self.observed_wake(track,text):track.wake=True
                    await self.c.ws.send_json({'type':'asr_preview','text':text,'status':'checking_identity',
                        'source':'speaker_turn','track':track.key,'audio_seconds':round(len(self.submission_wave(track))/16000,2),
                        'voiced_seconds':round(track.speech_s,2),'channel':track.channel})
                    await self.refine_identity(track,text,epoch=epoch)
                    if self.closed or epoch!=self.epoch:continue
                    context=self.turns.speaker_context(track,text)
                    if not text or context is None:
                        await self.c.ws.send_json({'type':'ignored','reason':'Speaker identity or words were uncertain. Please repeat or say Hey Robbie and introduce yourself again.'})
                        continue
                    # Wait for a halt acknowledgment rather than racing another
                    # generation. Microphone hops continue in their own task.
                    while (self.c.speaking or self.c.halting) and not self.closed:
                        await asyncio.sleep(.05)
                    if self.closed or epoch!=self.epoch or context['speaker_id'] not in self.memory.users:continue
                    self.c.current_speaker=context;self.pending_names[context['speaker_id']]=text
                    self.c.session.reset()
                    await self.notice(decision={'accepted':True,'kind':'enrolled','id':context['speaker_id'],'name':context['display_name'],
                        'reason':'One second without this speaker; background voices do not extend the turn'})
                    await self.c.ws.send_json({'type':'endpoint','transcript':text,'speaker_id':context['speaker_id'],
                        'speaker_name':context['display_name'],'by':'speaker_silence_1s','turn_s':round(track.last_speech-track.start,2)})
                    await self.c._send_turn(text)
                except asyncio.CancelledError:raise
                except Exception as exc:
                    logging.getLogger('robo.speakers').warning('Speaker submit failed: %s',type(exc).__name__)
                    await self.c.ws.send_json({'type':'warning','text':'Speaker turn could not be submitted; please repeat.'})
        except asyncio.CancelledError:pass

    async def refine_identity(self,track,text,epoch=None):
        """Full-utterance ECAPA checks channel reuse/merges, not fast timing.

        A diarization channel is NOT a permanent biometric identity: two similar
        voices can share it. Never overwrite Christoph's name with Robert's
        introduction just because the diarizer reused a channel.
        """
        current=self.memory.users.get(track.sid)
        if current is None:return
        introduction_name=self_name(text)
        enrollment_probe=current.teacher is None or bool(introduction_name and (wake_intent(text) or track.wake))
        embedding=None
        view_cosine=None;view_count=0
        try:
            from speaker_teacher import get_cached_teacher
            wave=self.submission_wave(track);wave=np.pad(wave,(0,max(0,10240-len(wave))))
            whole=track.verification_audio
            def embed_views():
                teacher=get_cached_teacher()
                if enrollment_probe and whole is not None and len(whole)>=10240:
                    return unit(teacher.embed(whole)),None,1
                vectors=[unit(teacher.embed(wave))]
                if whole is not None and len(whole)>=10240 and (len(whole)!=len(wave) or not np.array_equal(whole,wave)):
                    vectors.append(unit(teacher.embed(whole)))
                cosine=float(vectors[0]@vectors[1]) if len(vectors)>1 else None
                # When independent owned-word recognition verified the ASR
                # crop, the discarded prefix is not equally good identity
                # evidence. Preserve a 20% whole-sentence cross-check rather
                # than averaging a TV-heavy pre-roll at the same weight.
                if len(vectors)==2 and getattr(track,'asr_crop_verified',False) and len(wave)>=20480:
                    merged=.8*vectors[0]+.2*vectors[1]
                else:merged=np.mean(vectors,axis=0)
                return unit(merged),cosine,len(vectors)
            embedding,view_cosine,view_count=await asyncio.to_thread(embed_views)
        except Exception:pass
        if epoch is not None and (self.closed or epoch!=self.epoch):track.sid=None;return
        if view_cosine is not None and view_cosine<.5:
            track.sid=None
            await self.notice(warning='Speaker cues disagree across the full sentence and ASR crop. Please repeat with less overlap.')
            return
        if embedding is None:
            track.sid=None
            await self.notice(warning='Full-utterance speaker verification is unavailable. Please retry or explicitly use Talk without speaker filter.')
            return
        bot=await self.playback_voice();bot_score=float(embedding@bot) if bot is not None else None
        if epoch is not None and (self.closed or epoch!=self.epoch):track.sid=None;return
        known_best=max((float(embedding@u.teacher) for u in self.memory.users.values() if u.teacher is not None),default=-1.)
        if bot_score is not None and bot_score>=.65 and bot_score-known_best>=self.utterance_margin:
            track.sid=None
            await self.notice(warning='Likely playback echo rather than an enrolled user; no turn was submitted.')
            return
        quality_score=None;quality_margin=None
        name=self_name(text);wake=bool(wake_intent(text) or track.wake)
        incomplete=bool(re.search(r'\b(?:ich\s+(?:bin|heiße)|i\s+am|my\s+name\s+is)\s*[.!?…]*$',text,re.I))
        other_introduction=any(t.sid and t.sid!=current.id and t.wake and t.name_heard and not t.closed
            and t.start<=track.last_speech+1. and t.last_speech>=track.start-.32 for t in self.turns.tracks.values())
        other_introduction=other_introduction or any(track.start-.32<=at<=track.last_speech+1.6
            and value!=current.name.casefold() for at,value in self.asr_wakes)
        if wake and not name and (incomplete or (current.named and other_introduction)):
            # One introduction can straddle old/new diarizer channels. Its
            # clipped "Hey, I am…" prefix is not an additional anonymous user.
            # Leave the complete foreground introduction to establish its ID.
            track.sid=None;return
        # Incremental ASR labels are provisional until the first complete
        # acoustic reference exists (e.g. partial "Selber" -> final "Selma").
        # Correct that label in place, not by creating an orphan second user.
        if current.teacher is None and name and wake:
            existing=next((u for u in self.memory.users.values() if u.id!=current.id and u.named
                and u.name.casefold()==name.casefold() and u.teacher is not None),None)
            if existing is not None:
                if float(embedding@existing.teacher)<self.utterance_threshold:track.sid=None;return
                previous=current.id;self.rebind(track,existing.id);self.memory.users.pop(previous,None)
            else:
                self.memory.rename(current.id,name);current.teacher=unit(embedding)
            return
        # A wake-only Guest is provisional. Use the immediately following
        # full self-introduction as its reliable teacher reference instead of
        # rejecting it against a 640-ms wake fragment. Never overwrite a named
        # person. A repeated known name must still agree acoustically.
        if not current.named and name and (wake or self.memory.clock()-current.created<=15):
            existing=next((u for u in self.memory.users.values() if u.id!=current.id and u.named and u.name.casefold()==name.casefold()),None)
            if existing is not None:
                if existing.teacher is None or float(unit(embedding)@existing.teacher)<self.utterance_threshold:track.sid=None;return
                old=current.id;current=existing;self.rebind(track,existing.id);self.memory.users.pop(old,None)
            else:self.memory.rename(current.id,name);current.teacher=unit(embedding)
            return
        if name and current.named and current.name.casefold()!=name.casefold() and not wake:
            track.sid=None
            await self.notice(warning='A different name was heard without the wake phrase. Say “Hey Robbie, I am …” to join; an existing user will not be renamed.')
            return
        student_vote=self.memory.match(track.vector(),track.long_vector())
        scores=[{'name':u.name,'cosine':round(float(unit(embedding)@u.teacher),4)} for u in self.memory.users.values()
            if embedding is not None and u.teacher is not None]
        await self.notice(verification={'candidate':current.name,'reference_scores':scores,'wake':wake,
            'self_introduction':bool(name),'threshold':self.utterance_threshold,'views':view_count,'view_cosine':view_cosine,
            'student_vote':student_vote})
        if name and wake and current.named and current.name.casefold()!=name.casefold():
            existing=next((u for u in self.memory.users.values() if u.named and u.name.casefold()==name.casefold()),None)
            if existing is not None:
                # A repeated self-introduction may reuse a name only if its
                # acoustic reference also agrees; a spoken name is not proof.
                agree=embedding is not None and existing.teacher is not None and float(unit(embedding)@existing.teacher)>=self.utterance_threshold
                if not agree:track.sid=None;return
                user=existing
            else:
                user,_=self.memory.enroll(track.vector(),track.long_vector(),name,force_new=True)
                if user is None:track.sid=None;return
            self.rebind(track,user.id)
            current=user
        elif embedding is not None:
            references=[(float(unit(embedding)@u.teacher),u) for u in self.memory.users.values() if u.teacher is not None]
            references.sort(key=lambda x:x[0],reverse=True)
            if references:
                score,user=references[0];runner=references[1][0] if len(references)>1 else -1.
                if wake and not name and current.named and score<self.utterance_threshold:
                    guest,_=self.memory.enroll(track.vector(),track.long_vector(),force_new=True)
                    if guest is None:track.sid=None;return
                    guest.teacher=unit(embedding);self.rebind(track,guest.id)
                    return
                continuity=(len(references)>=2 and user.id==current.id and track.channel is not None
                    and self.turns.channels.get(track.channel)==current.id and
                    score>=max(.4,self.utterance_threshold-.08) and score-runner>=max(.12,self.utterance_margin))
                consensus=(score>=max(.55,self.utterance_threshold) and score-runner>=max(.04,self.utterance_margin/2)
                    and view_cosine is not None and view_cosine>=.9 and track.speech_s>=1.5
                    and (bot_score is None or score-bot_score>=.12))
                # A nearly tied teacher must not be resolved by reusable
                # diarizer-channel ownership. A separately calibrated student
                # can corroborate the SAME teacher winner, never overrule it.
                corroborated=(score>=self.utterance_threshold and score-runner>=0 and
                    view_cosine is not None and view_cosine>=.85 and track.speech_s>=1.28 and
                    student_vote.get('id')==user.id and
                    student_vote.get('similarity',-1)>=max(.70,self.memory.threshold) and
                    student_vote.get('margin',0)>=.10 and
                    (bot_score is None or score-bot_score>=.12))
                if (score>=self.utterance_threshold and score-runner>=self.utterance_margin) or continuity or consensus or corroborated:
                    if continuity and score<self.utterance_threshold:
                        await self.notice(continuity={'name':user.name,'cosine':score,'margin':score-runner,
                            'reason':'Established diarization channel + separated reference winner; bounded threshold relaxation'})
                    current=user;self.rebind(track,user.id)
                    quality_score=score;quality_margin=score-runner
                elif current.teacher is not None:
                    track.sid=None
                    await self.notice(warning='Full-utterance speaker cue was ambiguous; please repeat or reintroduce yourself.')
                    return
        if embedding is not None and current.teacher is None:current.teacher=unit(embedding)
        elif (quality_score is not None and current.named and track.speech_s>=1.5
              and quality_score>=max(.4,self.utterance_threshold-.08) and quality_margin>=max(.20,self.utterance_margin)
              and view_cosine is not None and (view_cosine>=.9 or
                  (quality_score>=max(.55,self.utterance_threshold) and view_cosine>=.85))
              and (bot_score is None or quality_score-bot_score>=.12)):
            anchor=current.teacher_anchor if current.teacher_anchor is not None else current.teacher
            if float(embedding@anchor)>=max(.4,self.utterance_threshold-.08):current.adapt_teacher(embedding)
    async def metadata(self,data):
        update=data.get('speaker_update')
        if not isinstance(update,dict):return
        sid=update.get('speaker_id');name=update.get('display_name');text=self.pending_names.pop(sid,None)
        user=self.memory.users.get(sid)
        if text and name==self_name(text) and user is not None and (not user.named or user.name.casefold()==name.casefold()):
            self.memory.rename(sid,name);await self.notice(metadata=update)