Add per-turn timeline logging: every stage of a reply (LLM, per-chunk TTS, feature encode, speech render, WS delivery, client audio start/end) logs in order as [T<turn> +<secs>] so sync/latency problems can be read straight from a copied log

app.py

@@ -16,6 +16,8 @@ import wave
 import cv2
 import asyncio
 
+import timeline
+
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
@@ -296,7 +298,7 @@ async def generate_greeting():
                 from float_streamer import get_streamer
                 streamer = get_streamer()
                 if streamer.ready:
-                    await asyncio.to_thread(streamer.inject_speech, audio_path, audio_url)
+                    await asyncio.to_thread(streamer.inject_speech, audio_path, audio_url, tag="greeting")
                     logger.info(f"[{_ts()}] [GREETING] Injected into streamer")
             except Exception as e:
                 logger.warning(f"[GREETING] Streamer inject failed: {e}")
@@ -841,6 +843,7 @@ async def interrupt():
     (the client already stops audio playback on its own the instant this fires).
     """
     global _active_tail_task, _generation_id
+    timeline.mark("BARGE-IN: client sent /api/interrupt")
     # Invalidate in-flight chunks immediately, at the moment of interrupt rather
     # than waiting for the eventual new message — closes the race where a chunk's
     # TTS finishes and gets injected between the interrupt and the next message.
@@ -857,6 +860,25 @@ async def interrupt():
     return JSONResponse({"status": "ok"})
 
 
+@app.post("/api/timing")
+async def client_timing(request: Request):
+    """Client-side playback milestones, logged into the same turn timeline so
+    one copied log shows when audio actually started relative to its video."""
+    try:
+        b = await request.json()
+    except Exception:
+        return JSONResponse({"status": "bad"})
+    tag, turn, ev = b.get("tag"), b.get("turn"), b.get("event")
+    if ev == "audio_play":
+        timeline.mark(
+            f"{tag} CLIENT audio playing {b.get('since_start_ms', 0):.0f}ms after its speech_start arrived "
+            f"(waited {b.get('queued_ms', 0):.0f}ms for previous clip + {b.get('load_ms', 0):.0f}ms loading) "
+            f"= how far audio lags its video", turn)
+    elif ev == "audio_end":
+        timeline.mark(f"{tag} CLIENT audio ended (played {b.get('played_ms', 0)/1000:.2f}s)", turn)
+    return JSONResponse({"status": "ok"})
+
+
 @app.get("/api/greeting")
 async def get_greeting():
     if GREETING_DATA:
@@ -892,6 +914,10 @@ async def ws_video(websocket: WebSocket):
             if isinstance(item, dict):
                 # Audio trigger — send instantly, skip the 25fps delay
                 await websocket.send_json(item)
+                if item.get("type") == "speech_start":
+                    timeline.mark(f"{item.get('tag')} speech_start sent to client — its video frames follow now", item.get("turn"))
+                elif item.get("type") == "speech_end":
+                    timeline.mark("speech_end sent to client", item.get("turn"))
                 continue
             elif item is not None:
                 # Video frame
@@ -984,7 +1010,7 @@ async def ws_mic(websocket: WebSocket):
     logger.info(f"[{_ts()}] [WS-MIC] Client disconnected")
 
 
-async def _chat_stream_process_chunk(idx: int, total: int, sentence: str, streamer, gen_id: int):
+async def _chat_stream_process_chunk(idx: int, total: int, sentence: str, streamer, gen_id: int, turn: int):
     """Synthesize TTS for one chunk and inject it into the streamer. Returns (audio_url, duration).
 
     gen_id guards against a race with barge-in/new-message cancellation: asyncio
@@ -996,6 +1022,8 @@ async def _chat_stream_process_chunk(idx: int, total: int, sentence: str, stream
     instead of being spoken after the user already cut the avatar off.
     """
     from tts import generate_audio
+    tag = f"c{idx+1}/{total}"
+    timeline.mark(f"{tag} TTS start: \"{sentence[:60]}\"", turn)
     t_a = _time.time()
     audio_path = await asyncio.to_thread(generate_audio, sentence)
     t_b = _time.time()
@@ -1005,40 +1033,42 @@ async def _chat_stream_process_chunk(idx: int, total: int, sentence: str, stream
     if audio_path and os.path.exists(audio_path):
         audio_url = f"/api/audio/{os.path.basename(audio_path)}"
         audio_duration = _audio_duration(audio_path, sentence)
-    logger.info(
-        f"[{_ts()}] [CHAT-STREAM] Chunk {idx+1}/{total} TTS: {t_b-t_a:.2f}s | {audio_duration:.1f}s audio"
-    )
+    timeline.mark(
+        f"{tag} TTS done in {t_b-t_a:.2f}s ({audio_duration:.1f}s audio)" if audio_path
+        else f"{tag} TTS FAILED after {t_b-t_a:.2f}s", turn)
 
     if gen_id != _generation_id:
-        logger.info(f"[{_ts()}] [CHAT-STREAM] Chunk {idx+1}/{total} stale (interrupted mid-flight) — discarding")
+        timeline.mark(f"{tag} stale (interrupted mid-flight) — discarding", turn)
         return audio_url, audio_duration
 
     if streamer.ready and audio_path:
         await asyncio.to_thread(
             streamer.inject_speech, audio_path, audio_url,
             clear_buffer=(idx == 0), is_last=(idx == total - 1),
+            tag=tag, turn=turn,
         )
     elif streamer.ready and idx == total - 1:
         # TTS failed for the final chunk — inject_speech (and therefore the
         # is_last=True marker) never runs. Signal speech_end anyway so the
         # client's mic doesn't stay muted forever waiting for a response
         # that already partially played but will never formally "end".
-        logger.warning(f"[{_ts()}] [CHAT-STREAM] Chunk {idx+1}/{total} TTS failed — signaling speech_end anyway")
+        timeline.mark(f"{tag} was the final chunk — signaling speech_end anyway", turn)
         await asyncio.to_thread(streamer.signal_speech_end)
 
     return audio_url, audio_duration
 
 
-async def _chat_stream_tail(sentences, total, streamer, gen_id: int):
+async def _chat_stream_tail(sentences, total, streamer, gen_id: int, turn: int):
     """Synthesize + inject chunks 2..N in the background while chunk 1 plays."""
     for i in range(1, total):
         try:
-            await _chat_stream_process_chunk(i, total, sentences[i], streamer, gen_id)
+            await _chat_stream_process_chunk(i, total, sentences[i], streamer, gen_id, turn)
         except asyncio.CancelledError:
-            logger.info(f"[{_ts()}] [CHAT-STREAM] Tail cancelled (barge-in) at chunk {i+1}/{total}")
+            timeline.mark(f"tail cancelled (barge-in) at c{i+1}/{total}", turn)
             raise
         except Exception as e:
             logger.error(f"[{_ts()}] [CHAT-STREAM] Chunk {i+1}/{total} failed: {e}")
+            timeline.mark(f"c{i+1}/{total} FAILED: {e}", turn)
 
 
 @app.post("/api/chat/stream")
@@ -1066,7 +1096,7 @@ async def chat_stream(request: Request):
     _generation_id += 1
     my_gen = _generation_id
 
-    logger.info(f"[{_ts()}] [CHAT-STREAM] User: {user_text}")
+    turn = timeline.begin(f"user: \"{user_text[:80]}\"")
 
     # 1. LLM — run in worker thread so the event loop stays free to send WS frames
     t1 = _time.time()
@@ -1075,8 +1105,7 @@ async def chat_stream(request: Request):
     reply_text = llm_result['text']
     clean_text = llm_result.get('clean_text', reply_text)
     t2 = _time.time()
-    logger.info(f"[{_ts()}] [CHAT-STREAM] Reply: {clean_text[:80]}...")
-    logger.info(f"[{_ts()}] [CHAT-STREAM] LLM: {t2-t1:.2f}s")
+    timeline.mark(f"LLM done in {t2-t1:.2f}s: \"{clean_text[:80]}...\"", turn)
 
     # 2. Split into short chunks (~12 words each) so the first chunk's TTS
     # call is small and fast instead of synthesizing the entire reply.
@@ -1084,19 +1113,20 @@ async def chat_stream(request: Request):
     if not sentences:
         sentences = [reply_text]
     total = len(sentences)
+    timeline.mark(f"reply split into {total} chunk(s)", turn)
 
     from float_streamer import get_streamer
     streamer = get_streamer()
 
     # 3. First chunk — awaited inline so speech is already queued by the time
     # this handler returns.
-    audio_url, audio_duration = await _chat_stream_process_chunk(0, total, sentences[0], streamer, my_gen)
+    audio_url, audio_duration = await _chat_stream_process_chunk(0, total, sentences[0], streamer, my_gen, turn)
 
     # 4. Remaining chunks — generated in the background while chunk 1 plays.
     if total > 1:
-        _active_tail_task = asyncio.create_task(_chat_stream_tail(sentences, total, streamer, my_gen))
+        _active_tail_task = asyncio.create_task(_chat_stream_tail(sentences, total, streamer, my_gen, turn))
 
-    logger.info(f"[{_ts()}] [CHAT-STREAM] Total (first chunk ready): {_time.time()-t0:.2f}s")
+    timeline.mark(f"HTTP response sent (first chunk queued) — {_time.time()-t0:.2f}s total", turn)
 
     return JSONResponse({
         "text": clean_text,
@@ -1295,7 +1325,7 @@ function connectWebSocket(){
         if(typeof e.data==='string'){
             const msg=JSON.parse(e.data);
             if(msg.type==='speech_start' && msg.audio_url){
-                enqueueAudio(msg.audio_url);
+                enqueueAudio(msg.audio_url, msg);
             } else if(msg.type==='speech_end'){
                 handleSpeechEnd();
             }
@@ -1336,16 +1366,30 @@ function waitForAvatarDone(){
     return new Promise(resolve=>doneResolvers.push(resolve));
 }
 
+// Report playback milestones into the server's per-turn timeline log.
+function reportTiming(data){
+    fetch('/api/timing',{method:'POST',headers:{'Content-Type':'application/json'},
+        body:JSON.stringify(data),keepalive:true}).catch(()=>{});
+}
+
 function playNextAudio(){
     if(audioPlaying || audioQueue.length===0) return;
     audioPlaying=true;
-    const url=audioQueue.shift();
-    const audio=new Audio(url);
+    const item=audioQueue.shift();
+    const loadAt=performance.now();
+    const audio=new Audio(item.url);
     currentAudioEl=audio;
     audio.addEventListener('canplaythrough', ()=>{
         audio.play().catch(err=>console.warn('[AUDIO]',err));
     }, {once:true});
+    let playAt=0;
+    audio.addEventListener('playing', ()=>{
+        playAt=performance.now();
+        reportTiming({event:'audio_play', tag:item.tag, turn:item.turn,
+            since_start_ms:playAt-item.recvAt, queued_ms:loadAt-item.recvAt, load_ms:playAt-loadAt});
+    }, {once:true});
     audio.addEventListener('ended', ()=>{
+        reportTiming({event:'audio_end', tag:item.tag, turn:item.turn, played_ms:playAt?performance.now()-playAt:0});
         audioPlaying=false;
         if(currentAudioEl===audio) currentAudioEl=null;
         if(audioQueue.length>0) playNextAudio();
@@ -1354,10 +1398,10 @@ function playNextAudio(){
     audio.load();
 }
 
-function enqueueAudio(url){
+function enqueueAudio(url, msg){
     expectingAudio=false; avatarSpeaking=true;
     updateMicIndicator();
-    audioQueue.push(url);
+    audioQueue.push({url, tag:msg&&msg.tag, turn:msg&&msg.turn, recvAt:performance.now()});
     playNextAudio();
 }
 

float_streamer.py

@@ -20,6 +20,8 @@ import torch.nn.functional as F
 import cv2
 import librosa
 
+import timeline
+
 logger = logging.getLogger(__name__)
 
 FLOAT_REPO_PATH = "/app/float_repo"
@@ -121,12 +123,16 @@ class FloatStreamer:
         self._running = False
         if self._thread: self._thread.join(timeout=5)
 
-    def inject_speech(self, audio_path: str, audio_url: str = None, clear_buffer: bool = True, is_last: bool = True):
+    def inject_speech(self, audio_path: str, audio_url: str = None, clear_buffer: bool = True, is_last: bool = True,
+                      tag: str = None, turn: int = None):
+        """tag ("c2/4") and turn label this TTS chunk's timeline entries all the
+        way through render, WS delivery and client playback."""
         t0 = time.time()
         
         if clear_buffer:
+            flushed = self._frame_buffer.qsize()
             self.drain_buffer()
-            logger.info("[STREAMER] Buffer flushed for immediate speech playback")
+            timeline.mark(f"{tag} flushed {flushed} buffered idle frames so speech plays immediately", turn)
 
         speech_array, sr = librosa.load(audio_path, sr=self.sampling_rate)
 
@@ -154,19 +160,23 @@ class FloatStreamer:
             chunk_data = {
                 "wa": wa_chunk, "we": we, "chunk_index": i,
                 "total_chunks": num_chunks, "actual_frames": end_frame - start_frame,
+                "tag": tag, "turn": turn,
             }
 
             if i == 0:
                 chunk_data["speech_start"] = {
                     "type": "speech_start", "audio_url": audio_url,
                     "duration": len(speech_array) / self.sampling_rate, "num_chunks": num_chunks,
+                    "tag": tag, "turn": turn,
                 }
             self._speech_queue.put(chunk_data)
 
         if is_last:
-            self._speech_queue.put({"type": "speech_end"})
-            
-        logger.info(f"[STREAMER] Speech injected: {T} frames, {num_chunks} chunks, is_last={is_last}")
+            self._speech_queue.put({"type": "speech_end", "turn": turn})
+
+        timeline.mark(
+            f"{tag} audio features encoded in {time.time()-t0:.2f}s → {num_chunks} render chunk(s) "
+            f"({T} frames) queued{' + speech_end' if is_last else ''}", turn)
 
     def get_frame(self, timeout=0.1):
         try: return self._frame_buffer.get(timeout=timeout)
@@ -185,9 +195,11 @@ class FloatStreamer:
                 if speech_data and speech_data.get("type") == "speech_end":
                     self._is_speaking = False
                     self._idle_cfg_scale = self.opt.a_cfg_scale
-                    logger.info("[STREAMER] Speech ended, transitioning to idle")
+                    timeline.mark(
+                        f"speech_end queued behind {self._frame_buffer.qsize()} buffered frames — back to idle",
+                        speech_data.get("turn"))
                     try:
-                        self._frame_buffer.put({"type": "speech_end"}, timeout=2.0)
+                        self._frame_buffer.put({"type": "speech_end", "turn": speech_data.get("turn")}, timeout=2.0)
                     except queue.Full:
                         logger.warning("[STREAMER] Frame buffer full, dropping speech_end event")
                     continue
@@ -200,7 +212,9 @@ class FloatStreamer:
                         wa=speech_data["wa"], we=speech_data["we"],
                         actual_frames=speech_data.get("actual_frames", self.frames_per_chunk),
                         is_speech=True, nfe=5,
-                        speech_start_event=speech_start_event
+                        speech_start_event=speech_start_event,
+                        tag=f"{speech_data.get('tag')} render {speech_data['chunk_index']+1}/{speech_data['total_chunks']}",
+                        turn=speech_data.get("turn"),
                     )
 
                 else:
@@ -253,7 +267,7 @@ class FloatStreamer:
         )
 
     @torch.no_grad()
-    def _generate_chunk(self, wa, we, actual_frames=None, a_cfg_scale=None, e_cfg_scale=None, is_speech=False, nfe=None, speech_start_event=None, dynamic_dampening=None):
+    def _generate_chunk(self, wa, we, actual_frames=None, a_cfg_scale=None, e_cfg_scale=None, is_speech=False, nfe=None, speech_start_event=None, dynamic_dampening=None, tag=None, turn=None):
         if actual_frames is None: actual_frames = self.frames_per_chunk
         if a_cfg_scale is None: a_cfg_scale = self.opt.a_cfg_scale
         if e_cfg_scale is None: e_cfg_scale = self.opt.e_cfg_scale
@@ -323,6 +337,12 @@ class FloatStreamer:
         frames_pushed = 0
 
         if speech_start_event:
+            # Frames already buffered ahead of speech_start play before this
+            # chunk's audio is even requested — that's added reply latency.
+            ahead = self._frame_buffer.qsize()
+            timeline.mark(
+                f"{speech_start_event.get('tag')} speech_start queued behind {ahead} buffered frames "
+                f"(~{ahead / self.fps:.2f}s of video before it)", turn)
             try:
                 self._frame_buffer.put(speech_start_event, timeout=2.0)
             except queue.Full:
@@ -356,12 +376,21 @@ class FloatStreamer:
 
         t_dec_done = time.time()
 
-        # Idle chunks render back-to-back forever — keep them out of INFO so
-        # copied logs stay readable; speech chunks are the ones worth seeing.
-        (logger.info if is_speech else logger.debug)(
-            f"[STREAMER] Chunk rendered | speech={is_speech} | frames={frames_pushed} | "
-            f"ode={t_ode:.2f}s | encode={t_dec_done-t_dec:.2f}s | total={t_dec_done-t0:.2f}s"
-        )
+        if is_speech:
+            # Rendering slower than the video it produces means the frame buffer
+            # drains and video falls behind the audio.
+            video_s = frames_pushed / self.fps
+            took = t_dec_done - t0
+            slow = f" ⚠ SLOWER THAN REAL TIME ({took/video_s:.2f}x)" if video_s and took > video_s else ""
+            timeline.mark(
+                f"{tag} rendered {frames_pushed} frames ({video_s:.2f}s video) in {took:.2f}s "
+                f"[ode {t_ode:.2f}s, decode {t_dec_done-t_dec:.2f}s]{slow}", turn)
+        else:
+            # Idle chunks render back-to-back forever — keep them out of INFO.
+            logger.debug(
+                f"[STREAMER] Idle chunk rendered | frames={frames_pushed} | "
+                f"ode={t_ode:.2f}s | encode={t_dec_done-t_dec:.2f}s | total={t_dec_done-t0:.2f}s"
+            )
 
     def drain_buffer(self):
         while not self._frame_buffer.empty():
@@ -379,7 +408,7 @@ class FloatStreamer:
         self.drain_buffer()
         self._is_speaking = False
         self._idle_cfg_scale = self.opt.a_cfg_scale
-        logger.info("[STREAMER] Interrupted — speech dropped, transitioning to idle")
+        timeline.mark("streamer interrupted — queued speech dropped, back to idle")
 
     def signal_speech_end(self):
         """Tell the generation loop speech is over with no audio to inject — used

timeline.py

@@ -0,0 +1,41 @@
+"""
+Per-turn timeline logging. Every stage of one reply — LLM, TTS per chunk,
+feature encoding, rendering, WS delivery, client audio playback — logs as
+
+    [T3 +1.42s] c2/4 TTS done in 0.38s (2.9s audio)
+
+where T3 is the turn number and +1.42s is time since the user's message
+arrived, so a copied log reads top-to-bottom as an ordered breakdown of where
+the time went. Shared by app.py (request side) and float_streamer.py
+(render thread), hence its own module.
+"""
+import logging
+import time
+
+logger = logging.getLogger("timeline")
+
+_turn = 0
+_t0 = time.time()
+
+
+def begin(label: str) -> int:
+    """Start a new turn; later marks are timed relative to this moment."""
+    global _turn, _t0
+    _turn += 1
+    _t0 = time.time()
+    logger.info(f"[T{_turn} +  0.00s] ===== TURN START: {label}")
+    return _turn
+
+
+def current() -> int:
+    return _turn
+
+
+def mark(msg: str, turn: int = None):
+    """Log one stage of the current turn. A mark tagged with an older turn
+    (e.g. a late client report after barge-in) is labelled as such, since its
+    offset is measured against the newer turn's start."""
+    if turn is not None and turn != _turn:
+        logger.info(f"[T{turn} (late, now T{_turn})] {msg}")
+        return
+    logger.info(f"[T{_turn} +{time.time() - _t0:6.2f}s] {msg}")