feat: Add post-tool response timeout and enhance error handling in channel metrics

.env.example

@@ -33,6 +33,7 @@ LANGFUSE_BASE_URL=  # Optional alternative to LANGFUSE_HOST
 LANGFUSE_PUBLIC_KEY=
 LANGFUSE_SECRET_KEY=
 LANGFUSE_TRACE_FINALIZE_TIMEOUT_MS=8000
+LANGFUSE_POST_TOOL_RESPONSE_TIMEOUT_MS=30000
 LANGFUSE_MAX_PENDING_TRACE_TASKS=200
 LANGFUSE_TRACE_FLUSH_TIMEOUT_MS=1000
 

src/agent/models/llm_runtime.py

@@ -11,7 +11,7 @@ from src.agent.prompts.runtime import MCP_STARTUP_GREETING
 from src.core.logger import logger
 
 NVIDIA_OPENAI_BASE_URL = "https://integrate.api.nvidia.com/v1"
-MCP_STARTUP_GREETING_TIMEOUT_SEC = 4.0
+MCP_STARTUP_GREETING_TIMEOUT_SEC = 8.0
 MCP_GENERATE_REPLY_BLOCK_MESSAGE = (
     "Manual generate_reply is disabled in MCP mode; use session.say(...) instead."
 )
@@ -176,7 +176,7 @@ def run_startup_greeting(
             return session.say(
                 MCP_STARTUP_GREETING,
                 allow_interruptions=True,
-                add_to_chat_ctx=True,
+                add_to_chat_ctx=False,
             )
         except Exception as exc:
             logger.warning(f"MCP startup greeting could not start: {exc}")

src/agent/runtime/session.py

@@ -23,7 +23,7 @@ from src.agent.models.stt_factory import create_stt
 from src.agent.runtime.assistant import Assistant
 from src.agent.runtime.tasks import (
     cancel_task_for_shutdown,
-    schedule_llm_warmup_task,
+    run_llm_warmup,
     schedule_startup_greeting_task,
 )
 from src.agent.tools.feedback import ToolFeedbackController
@@ -63,7 +63,6 @@ async def session_handler(ctx: agents.JobContext) -> None:
     )
     trace_provider = setup_langfuse_tracer()
     startup_greeting_task: asyncio.Task[Any] | None = None
-    llm_warmup_task: asyncio.Task[Any] | None = None
     tool_feedback = ToolFeedbackController(enabled=False)
 
     if trace_provider:
@@ -84,14 +83,6 @@ async def session_handler(ctx: agents.JobContext) -> None:
 
     ctx.add_shutdown_callback(cancel_startup_greeting)
 
-    async def cancel_llm_warmup(_: str) -> None:
-        await cancel_task_for_shutdown(
-            llm_warmup_task,
-            task_name="llm warm-up",
-        )
-
-    ctx.add_shutdown_callback(cancel_llm_warmup)
-
     async def close_tool_feedback(_: str) -> None:
         await tool_feedback.aclose()
 
@@ -160,7 +151,12 @@ async def session_handler(ctx: agents.JobContext) -> None:
     )
     mcp_runtime_active = llm_runtime.mcp_runtime_active
     tool_feedback = ToolFeedbackController(enabled=mcp_runtime_active)
-    llm_warmup_task = schedule_llm_warmup_task(
+    logger.info(
+        "Running LLM warm-up before session start: provider=%s model=%s",
+        llm_runtime.provider,
+        llm_runtime.model,
+    )
+    await run_llm_warmup(
         llm_client=llm_runtime.llm,
         conn_options=llm_conn_options,
         provider=llm_runtime.provider,

src/agent/traces/channel_metrics.py

@@ -90,7 +90,13 @@ class ChannelPublisher:
                 reliable=True,
             )
         except Exception as e:
-            logger.error(f"Failed to publish live metrics update: {e}")
+            if _is_preconnect_publish_error(e):
+                logger.debug(
+                    "Skipping live metrics update before room connect: %s",
+                    e,
+                )
+            else:
+                logger.error(f"Failed to publish live metrics update: {e}")
 
     async def publish_conversation_turn(self, turn_metrics: TurnMetrics) -> None:
         """Publish completed turn metrics to LiveKit data channel."""
@@ -140,6 +146,10 @@ def _stt_display_duration(stt_metrics: STTMetrics) -> float:
     return stt_metrics.audio_duration
 
 
+def _is_preconnect_publish_error(exc: Exception) -> bool:
+    return "cannot access local participant before connecting" in str(exc).lower()
+
+
 def _build_partial_latencies(
     *,
     turn_metrics: Optional[TurnMetrics],

src/agent/traces/metrics_collector.py

@@ -250,6 +250,7 @@ class MetricsCollector:
         self._llm_stall_tasks: dict[str, asyncio.Task[None]] = {}
         self._latest_vad_metrics: Optional[VADMetrics] = None
         self._latest_vad_metric_attributes: Optional[dict[str, Any]] = None
+        self._first_final_user_turn_logged = False
         self._llm_stall_timeout_sec = max(
             float(
                 getattr(
@@ -281,6 +282,14 @@ class MetricsCollector:
     def _trace_finalize_timeout_sec(self, value: float) -> None:
         self._tracer._trace_finalize_timeout_sec = value
 
+    @property
+    def _trace_post_tool_response_timeout_sec(self) -> float:
+        return self._tracer._trace_post_tool_response_timeout_sec
+
+    @_trace_post_tool_response_timeout_sec.setter
+    def _trace_post_tool_response_timeout_sec(self, value: float) -> None:
+        self._tracer._trace_post_tool_response_timeout_sec = value
+
     # ------------------------------------------------------------------
     # Public event handlers
     # ------------------------------------------------------------------
@@ -310,6 +319,14 @@ class MetricsCollector:
         if not normalized:
             return
         self._pending_transcripts.append(normalized)
+        if not self._first_final_user_turn_logged:
+            self._first_final_user_turn_logged = True
+            logger.info(
+                "First finalized user transcript received: room=%s chars=%s preview=%r",
+                self._room_name,
+                len(normalized),
+                normalized[:80],
+            )
         self._start_llm_stall_watchdog(transcript=normalized)
         room_id = await self._resolve_room_id()
         await self._tracer.create_turn(user_transcript=normalized, room_id=room_id)

src/agent/traces/turn_tracer.py

@@ -150,6 +150,7 @@ class ToolExecutionBlock:
 
 
 _DEFAULT_TRACE_FINALIZE_TIMEOUT_MS = 8000.0
+_DEFAULT_POST_TOOL_RESPONSE_TIMEOUT_MS = 30000.0
 _DEFAULT_MAX_PENDING_TRACE_TASKS = 200
 _DEFAULT_TRACE_FLUSH_TIMEOUT_SEC = 1.0
 
@@ -186,6 +187,7 @@ class TurnTracer:
         self._trace_lock = asyncio.Lock()
         self._trace_emit_tasks: set[asyncio.Task[None]] = set()
         self._trace_finalize_tasks: dict[str, asyncio.Task[None]] = {}
+        self._trace_finalize_task_versions: dict[str, int] = {}
 
         self._trace_finalize_timeout_sec = (
             max(
@@ -198,6 +200,17 @@ class TurnTracer:
             )
             / 1000.0
         )
+        self._trace_post_tool_response_timeout_sec = (
+            max(
+                getattr(
+                    settings.langfuse,
+                    "LANGFUSE_POST_TOOL_RESPONSE_TIMEOUT_MS",
+                    _DEFAULT_POST_TOOL_RESPONSE_TIMEOUT_MS,
+                ),
+                0.0,
+            )
+            / 1000.0
+        )
         self._trace_max_pending_tasks = max(
             int(
                 getattr(
@@ -563,18 +576,21 @@ class TurnTracer:
             return
 
         completed_turn: Optional[TraceTurn] = None
-        schedule_timeout_for_turn: Optional[str] = None
+        schedule_timeout_for_turn: Optional[tuple[str, float]] = None
         async with self._trace_lock:
             if trace_turn not in self._pending_trace_turns:
                 return
             if not self._is_complete(trace_turn):
                 if self._should_schedule_finalize_timeout(trace_turn):
-                    schedule_timeout_for_turn = trace_turn.turn_id
+                    schedule_timeout_for_turn = (
+                        trace_turn.turn_id,
+                        self._resolve_finalize_timeout_sec(trace_turn),
+                    )
             else:
                 completed_turn = self._finalize_locked(trace_turn)
 
         if schedule_timeout_for_turn:
-            self._schedule_finalize_timeout(schedule_timeout_for_turn)
+            self._schedule_finalize_timeout(*schedule_timeout_for_turn)
         if completed_turn:
             self._schedule_trace_emit(completed_turn)
 
@@ -682,22 +698,31 @@ class TurnTracer:
             and turn.tts_calls
             and not self._is_complete(turn)
             and not (turn.tool_phase_open and turn.last_tool_event_at is None)
-            and turn.turn_id not in self._trace_finalize_tasks
-            and self._trace_finalize_timeout_sec > 0.0
+            and self._resolve_finalize_timeout_sec(turn) > 0.0
         )
 
+    def _resolve_finalize_timeout_sec(self, turn: TraceTurn) -> float:
+        if self._requires_post_tool_response(turn):
+            return self._trace_post_tool_response_timeout_sec
+        return self._trace_finalize_timeout_sec
+
     def _requires_post_tool_response(self, turn: TraceTurn) -> bool:
         if not turn.tool_step_announced and turn.last_tool_event_order is None:
             return False
         return not self._post_tool_response_observed(turn)
 
-    def _post_tool_response_observed(self, turn: TraceTurn) -> bool:
+    def _post_tool_assistant_observed(self, turn: TraceTurn) -> bool:
         if turn.last_tool_event_order is None:
             return False
-        assistant_seen = bool(
+        return bool(
             turn.assistant_text_updated_order is not None
             and turn.assistant_text_updated_order > turn.last_tool_event_order
         )
+
+    def _post_tool_response_observed(self, turn: TraceTurn) -> bool:
+        if turn.last_tool_event_order is None:
+            return False
+        assistant_seen = self._post_tool_assistant_observed(turn)
         tts_seen = bool(
             turn.tts_updated_order is not None
             and turn.tts_updated_order > turn.last_tool_event_order
@@ -717,7 +742,12 @@ class TurnTracer:
         *,
         missing_assistant_fallback: bool = False,
         tool_post_response_missing: bool = False,
+        drop_assistant_text: bool = False,
     ) -> TraceTurn:
+        if drop_assistant_text:
+            turn.assistant_text = ""
+            turn.response_text = ""
+
         if not turn.prompt_text:
             turn.prompt_text = turn.user_transcript
         if not turn.response_text and turn.assistant_text:
@@ -726,7 +756,13 @@ class TurnTracer:
             turn.assistant_text = turn.response_text
 
         if missing_assistant_fallback and not turn.assistant_text:
-            fallback = self._best_available_assistant_text(turn)
+            fallback = self._best_available_assistant_text(
+                turn,
+                min_observed_order=(
+                    turn.last_tool_event_order if tool_post_response_missing else None
+                ),
+                include_pending_agent_transcripts=not tool_post_response_missing,
+            )
             if fallback:
                 turn.assistant_text = fallback
                 if not turn.response_text:
@@ -746,15 +782,40 @@ class TurnTracer:
         self._cancel_finalize_timeout(turn.turn_id)
         return turn
 
-    def _best_available_assistant_text(self, turn: TraceTurn) -> str:
+    def _best_available_assistant_text(
+        self,
+        turn: TraceTurn,
+        *,
+        min_observed_order: Optional[int] = None,
+        include_pending_agent_transcripts: bool = True,
+    ) -> str:
         if turn.assistant_text.strip():
-            return turn.assistant_text.strip()
+            if (
+                min_observed_order is None
+                or (
+                    turn.assistant_text_updated_order is not None
+                    and turn.assistant_text_updated_order > min_observed_order
+                )
+            ):
+                return turn.assistant_text.strip()
         if turn.response_text.strip():
-            return turn.response_text.strip()
+            if (
+                min_observed_order is None
+                or (
+                    turn.assistant_text_updated_order is not None
+                    and turn.assistant_text_updated_order > min_observed_order
+                )
+            ):
+                return turn.response_text.strip()
         for tts_call in reversed(turn.tts_calls):
+            if (
+                min_observed_order is not None
+                and tts_call.observed_order <= min_observed_order
+            ):
+                continue
             if tts_call.assistant_text.strip():
                 return tts_call.assistant_text.strip()
-        if self._pending_agent_transcripts:
+        if include_pending_agent_transcripts and self._pending_agent_transcripts:
             return self._pending_agent_transcripts.popleft().strip()
         return ""
 
@@ -762,32 +823,66 @@ class TurnTracer:
     # Timeout scheduling
     # ------------------------------------------------------------------
 
-    def _schedule_finalize_timeout(self, turn_id: str) -> None:
-        if turn_id in self._trace_finalize_tasks:
+    def _schedule_finalize_timeout(self, turn_id: str, timeout_sec: float) -> None:
+        if timeout_sec <= 0.0:
             return
-        task = asyncio.create_task(self._finalize_after_timeout(turn_id))
+
+        version = self._trace_finalize_task_versions.get(turn_id, 0) + 1
+        self._trace_finalize_task_versions[turn_id] = version
+
+        existing_task = self._trace_finalize_tasks.get(turn_id)
+        current = asyncio.current_task()
+        if existing_task and not existing_task.done() and existing_task is not current:
+            existing_task.cancel()
+
+        task = asyncio.create_task(
+            self._finalize_after_timeout(
+                turn_id=turn_id,
+                version=version,
+                timeout_sec=timeout_sec,
+            )
+        )
         self._trace_finalize_tasks[turn_id] = task
         task.add_done_callback(
-            lambda _: self._trace_finalize_tasks.pop(turn_id, None)
+            lambda _task, tid=turn_id, v=version: self._on_finalize_timeout_task_done(
+                turn_id=tid,
+                version=v,
+            )
         )
 
+    def _on_finalize_timeout_task_done(self, *, turn_id: str, version: int) -> None:
+        if self._trace_finalize_task_versions.get(turn_id) != version:
+            return
+        self._trace_finalize_tasks.pop(turn_id, None)
+
     def _cancel_finalize_timeout(self, turn_id: str) -> None:
+        self._trace_finalize_task_versions.pop(turn_id, None)
         task = self._trace_finalize_tasks.pop(turn_id, None)
         current = asyncio.current_task()
         if task and not task.done() and task is not current:
             task.cancel()
 
-    async def _finalize_after_timeout(self, turn_id: str) -> None:
-        await asyncio.sleep(self._trace_finalize_timeout_sec)
+    async def _finalize_after_timeout(
+        self,
+        *,
+        turn_id: str,
+        version: int,
+        timeout_sec: float,
+    ) -> None:
+        await asyncio.sleep(timeout_sec)
 
         completed_turn: Optional[TraceTurn] = None
         async with self._trace_lock:
+            if self._trace_finalize_task_versions.get(turn_id) != version:
+                return
+
             pending_turn = next(
                 (t for t in self._pending_trace_turns if t.turn_id == turn_id),
                 None,
             )
             if not pending_turn:
                 return
+
             if self._is_complete(pending_turn):
                 completed_turn = self._finalize_locked(pending_turn)
             elif (
@@ -798,15 +893,24 @@ class TurnTracer:
                 requires_post_tool_response = self._requires_post_tool_response(
                     pending_turn
                 )
+                missing_post_tool_assistant = bool(
+                    requires_post_tool_response
+                    and not self._post_tool_assistant_observed(pending_turn)
+                )
                 completed_turn = self._finalize_locked(
                     pending_turn,
-                    missing_assistant_fallback=not bool(pending_turn.assistant_text),
+                    missing_assistant_fallback=(
+                        missing_post_tool_assistant
+                        or not bool(pending_turn.assistant_text)
+                    ),
                     tool_post_response_missing=requires_post_tool_response,
+                    drop_assistant_text=missing_post_tool_assistant,
                 )
 
         if completed_turn:
             self._schedule_trace_emit(completed_turn)
 
+
     # ------------------------------------------------------------------
     # Trace emission
     # ------------------------------------------------------------------

src/core/settings.py

@@ -188,7 +188,7 @@ class LLMSettings(CoreSettings):
 
     # NVIDIA settings
     NVIDIA_API_KEY: Optional[str] = Field(default=None)
-    NVIDIA_MODEL: str = Field(default="qwen/qwen3-next-80b-a3b-instruct")
+    NVIDIA_MODEL: str = Field(default="meta/llama-3.1-8b-instruct") #"qwen/qwen3-next-80b-a3b-instruct"
 
     # Ollama settings
     OLLAMA_BASE_URL: str = Field(
@@ -268,6 +268,15 @@ class LangfuseSettings(CoreSettings):
         le=10000.0,
         description="Timeout to wait for assistant text before force-finalizing trace",
     )
+    LANGFUSE_POST_TOOL_RESPONSE_TIMEOUT_MS: float = Field(
+        default=30000.0,
+        ge=0.0,
+        le=120000.0,
+        description=(
+            "Timeout to wait for post-tool assistant response before force-finalizing trace; "
+            "telemetry only, does not affect live audio latency"
+        ),
+    )
     LANGFUSE_MAX_PENDING_TRACE_TASKS: int = Field(
         default=200,
         ge=1,

tests/test_agent_mcp_runtime.py

@@ -245,7 +245,7 @@ def testrun_startup_greeting_uses_say_in_mcp_mode() -> None:
             "text": MCP_STARTUP_GREETING,
             "kwargs": {
                 "allow_interruptions": True,
-                "add_to_chat_ctx": True,
+                "add_to_chat_ctx": False,
             },
         }
     ]
@@ -280,7 +280,7 @@ def testrun_startup_greeting_swallows_say_exception() -> None:
     handle = run_startup_greeting(session, mcp_runtime_active=True)  # type: ignore[arg-type]
 
     assert handle is None
-    assert session.say_calls == [{"text": MCP_STARTUP_GREETING, "kwargs": {"allow_interruptions": True, "add_to_chat_ctx": True}}]
+    assert session.say_calls == [{"text": MCP_STARTUP_GREETING, "kwargs": {"allow_interruptions": True, "add_to_chat_ctx": False}}]
     assert session.generate_reply_calls == []
 
 

tests/test_channel_metrics.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+
+from src.agent.traces.channel_metrics import ChannelPublisher
+
+
+class _FailingLocalParticipant:
+    def __init__(self, exc: Exception) -> None:
+        self._exc = exc
+
+    async def publish_data(
+        self,
+        *,
+        payload: bytes,
+        topic: str,
+        reliable: bool,
+    ) -> None:
+        _ = (payload, topic, reliable)
+        raise self._exc
+
+
+class _FakeRoom:
+    def __init__(self, exc: Exception) -> None:
+        self.local_participant = _FailingLocalParticipant(exc)
+
+
+def test_publish_live_update_downgrades_preconnect_publish_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import src.agent.traces.channel_metrics as channel_metrics_module
+
+    debug_logs: list[tuple[Any, ...]] = []
+    error_logs: list[tuple[Any, ...]] = []
+    monkeypatch.setattr(
+        channel_metrics_module.logger,
+        "debug",
+        lambda *args, **kwargs: debug_logs.append((args, kwargs)),
+    )
+    monkeypatch.setattr(
+        channel_metrics_module.logger,
+        "error",
+        lambda *args, **kwargs: error_logs.append((args, kwargs)),
+    )
+
+    room = _FakeRoom(
+        RuntimeError("cannot access local participant before connecting")
+    )
+    publisher = ChannelPublisher(room)  # type: ignore[arg-type]
+
+    asyncio.run(
+        publisher.publish_live_update(
+            speech_id=None,
+            stage="llm",
+            role=None,
+            turn_metrics=None,
+        )
+    )
+
+    assert debug_logs
+    assert not error_logs
+
+
+def test_publish_live_update_keeps_error_logging_for_other_failures(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import src.agent.traces.channel_metrics as channel_metrics_module
+
+    debug_logs: list[tuple[Any, ...]] = []
+    error_logs: list[tuple[Any, ...]] = []
+    monkeypatch.setattr(
+        channel_metrics_module.logger,
+        "debug",
+        lambda *args, **kwargs: debug_logs.append((args, kwargs)),
+    )
+    monkeypatch.setattr(
+        channel_metrics_module.logger,
+        "error",
+        lambda *args, **kwargs: error_logs.append((args, kwargs)),
+    )
+
+    room = _FakeRoom(RuntimeError("unexpected publish failure"))
+    publisher = ChannelPublisher(room)  # type: ignore[arg-type]
+
+    asyncio.run(
+        publisher.publish_live_update(
+            speech_id=None,
+            stage="llm",
+            role=None,
+            turn_metrics=None,
+        )
+    )
+
+    assert error_logs
+    assert not debug_logs

tests/test_langfuse_turn_tracing.py

@@ -868,6 +868,7 @@ def test_timeout_finalizes_tool_turn_with_missing_post_tool_response(
         langfuse_enabled=True,
     )
     collector._trace_finalize_timeout_sec = 0.01
+    collector._trace_post_tool_response_timeout_sec = 0.01
 
     async def _run() -> None:
         await collector.on_session_metadata(
@@ -907,6 +908,80 @@ def test_timeout_finalizes_tool_turn_with_missing_post_tool_response(
     root = fake_tracer.spans[0]
     assert root.attributes["tool.phase_announced"] is True
     assert root.attributes["tool.post_response_missing"] is True
+    assert root.attributes["langfuse.trace.output"] == "[assistant text unavailable]"
+
+
+def test_post_tool_timeout_prevents_early_finalize_of_pre_tool_leadin(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import src.agent.traces.metrics_collector as metrics_collector_module
+
+    fake_tracer = _FakeTracer()
+    monkeypatch.setattr(metrics_collector_module, "tracer", fake_tracer)
+
+    room = _FakeRoom()
+    collector = MetricsCollector(
+        room=room,  # type: ignore[arg-type]
+        model_name="moonshine",
+        room_name=room.name,
+        room_id="RM123",
+        participant_id="web-123",
+        langfuse_enabled=True,
+    )
+    collector._trace_finalize_timeout_sec = 0.01
+    collector._trace_post_tool_response_timeout_sec = 0.08
+
+    async def _run() -> None:
+        await collector.on_session_metadata(
+            session_id="session-post-tool-timeout-window",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("find me the best paper", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-post-tool-timeout-window"))
+        await collector.on_tool_step_started()
+        await collector.on_conversation_item_added(role="assistant", content="I'll look that up.")
+        await collector.on_metrics_collected(_make_tts_metrics("speech-post-tool-timeout-window"))
+        await collector.on_function_tools_executed(
+            function_calls=[
+                _FakeFunctionCall(
+                    name="paper_search",
+                    call_id="call-post-tool-timeout-window",
+                    arguments='{"query":"mps cubic phases"}',
+                    created_at=400.0,
+                )
+            ],
+            function_call_outputs=[
+                _FakeFunctionCallOutput(
+                    output='{"results":[{"title":"A key paper"}]}',
+                    is_error=False,
+                    created_at=400.2,
+                )
+            ],
+            created_at=400.2,
+        )
+
+        # The base finalize timeout has elapsed, but post-tool timeout should keep the turn pending.
+        await asyncio.sleep(0.03)
+        await collector.wait_for_pending_trace_tasks()
+        assert not fake_tracer.spans
+
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content="The most cited paper is Attention Is All You Need.",
+        )
+        await collector.on_metrics_collected(
+            _make_tts_metrics("speech-post-tool-timeout-window")
+        )
+        await collector.wait_for_pending_trace_tasks()
+
+    asyncio.run(_run())
+
+    root = fake_tracer.spans[0]
+    assert (
+        root.attributes["langfuse.trace.output"]
+        == "The most cited paper is Attention Is All You Need."
+    )
+    assert root.attributes["tool.post_response_missing"] is False
 
 
 def test_tool_event_without_matching_turn_is_ignored(