Add fallback session ID support in MetricsCollector: Introduce a mechanism to generate a console-prefixed fallback session ID when metadata is absent. Update session handling logic to utilize the fallback ID and enhance test coverage for fallback scenarios.

src/agent/agent.py

@@ -1,6 +1,7 @@
 import asyncio
 import base64
 import json
+import sys
 
 from livekit import agents, rtc
 from livekit.agents import AgentServer, AgentSession, Agent, room_io
@@ -35,6 +36,13 @@ def _normalize_langfuse_host() -> str | None:
     return host.rstrip("/")
 
 
+def _fallback_session_prefix() -> str | None:
+    """Use console-prefixed fallback session id when running `... console`."""
+    if any(arg == "console" for arg in sys.argv[1:]):
+        return "console"
+    return None
+
+
 def setup_langfuse_tracer() -> TracerProvider | None:
     """Configure LiveKit telemetry tracer to export traces to Langfuse."""
     global _langfuse_tracer_provider
@@ -160,6 +168,7 @@ async def session_handler(ctx: agents.JobContext) -> None:
         room_name=ctx.room.name,
         room_id=initial_room_id,
         participant_id=initial_participant_id,
+        fallback_session_prefix=_fallback_session_prefix(),
         langfuse_enabled=trace_provider is not None,
     )
 

src/agent/metrics_collector.py

@@ -179,6 +179,7 @@ class MetricsCollector:
         room_name: str,
         room_id: Optional[str] = None,
         participant_id: Optional[str] = None,
+        fallback_session_prefix: Optional[str] = None,
         langfuse_enabled: bool = False,
     ) -> None:
         """Initialize metrics collector.
@@ -189,6 +190,8 @@ class MetricsCollector:
             room_name: LiveKit room name
             room_id: LiveKit room id (sid) when available
             participant_id: LiveKit participant identity when available
+            fallback_session_prefix: Prefix used for generated fallback session id
+                (e.g. "console" -> "console_<uuid>") when no metadata session id exists
             langfuse_enabled: Enable one-trace-per-turn Langfuse traces
         """
         self._room = room
@@ -201,7 +204,10 @@ class MetricsCollector:
 
         self._room_name = room_name or self.UNKNOWN_ROOM_ID
         self._room_id = room_id or room_name or self.UNKNOWN_ROOM_ID
-        self._session_id = self.UNKNOWN_SESSION_ID
+        self._fallback_session_id = self._build_fallback_session_id(
+            fallback_session_prefix
+        )
+        self._session_id = self._fallback_session_id or self.UNKNOWN_SESSION_ID
         self._participant_id = participant_id or self.UNKNOWN_PARTICIPANT_ID
         self._langfuse_enabled = langfuse_enabled
         self._pending_trace_turns: deque[TraceTurn] = deque()
@@ -259,8 +265,9 @@ class MetricsCollector:
 
             for turn in self._pending_trace_turns:
                 if (
-                    turn.session_id == self.UNKNOWN_SESSION_ID
-                    and self._session_id != self.UNKNOWN_SESSION_ID
+                    turn.session_id in {self.UNKNOWN_SESSION_ID, self._fallback_session_id}
+                    and self._session_id
+                    not in {self.UNKNOWN_SESSION_ID, self._fallback_session_id}
                 ):
                     turn.session_id = self._session_id
                 if (
@@ -1248,6 +1255,12 @@ class MetricsCollector:
         normalized = value.strip()
         return normalized or None
 
+    def _build_fallback_session_id(self, prefix: Optional[str]) -> Optional[str]:
+        normalized_prefix = self._normalize_optional_text(prefix)
+        if not normalized_prefix:
+            return None
+        return f"{normalized_prefix}_{uuid.uuid4()}"
+
     async def _resolve_room_id(self) -> str:
         if self._room_id and self._room_id != self._room_name:
             return self._room_id

tests/test_langfuse_turn_tracing.py

@@ -494,3 +494,75 @@ def test_trace_finalize_timeout_for_missing_assistant_text(
     root = turn_spans[0]
     assert root.attributes["langfuse.trace.metadata.assistant_text_missing"] is True
     assert root.attributes["langfuse.trace.output"] == "[assistant text unavailable]"
+
+
+def test_fallback_console_session_id_is_used_when_metadata_absent(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import src.agent.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="console-user",
+        fallback_session_prefix="console",
+        langfuse_enabled=True,
+    )
+
+    async def _run() -> None:
+        await collector.on_user_input_transcribed("console test", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-console"))
+        await collector.on_conversation_item_added(role="assistant", content="console reply")
+        await collector.on_metrics_collected(_make_tts_metrics("speech-console"))
+        await collector.wait_for_pending_trace_tasks()
+
+    asyncio.run(_run())
+
+    turn_spans = [span for span in fake_tracer.spans if span.name == "turn"]
+    assert len(turn_spans) == 1
+    session_id = turn_spans[0].attributes["session_id"]
+    assert session_id.startswith("console_")
+    assert session_id != "unknown-session"
+
+
+def test_real_session_metadata_overrides_fallback_for_pending_turns(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import src.agent.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="console-user",
+        fallback_session_prefix="console",
+        langfuse_enabled=True,
+    )
+
+    async def _run() -> None:
+        await collector.on_user_input_transcribed("override test", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-override"))
+        await collector.on_metrics_collected(_make_tts_metrics("speech-override"))
+        await collector.on_session_metadata(
+            session_id="session-real",
+            participant_id="web-override",
+        )
+        await collector.on_conversation_item_added(role="assistant", content="reply")
+        await collector.wait_for_pending_trace_tasks()
+
+    asyncio.run(_run())
+
+    turn_spans = [span for span in fake_tracer.spans if span.name == "turn"]
+    assert len(turn_spans) == 1
+    assert turn_spans[0].attributes["session_id"] == "session-real"