feat(telemetry): one agent_turn span per speech handle

[davidzhao]

What

A reply that calls a tool runs two generations (LLM steps) in two tasks. They were two agent_turn spans linked only by lk.parent_generation_id, so one response rendered as two turns. A speech handle is now exactly one agent_turn.

How

• The speech handle owns the span. The first reply task (pipeline, realtime, or say) opens agent_turn under agent_session; the follow-up generation after a tool call runs in a new task but continues the open span instead of opening a second one. It ends with the speech in SpeechHandle._mark_done, whichever step that was on, recording the speech's error (redaction-aware) and the gen_ai.invoke_agent.duration metric for the whole turn.
• Each generation is a generation event on the span with its lk.generation_id and lk.parent_generation_id. On the span itself lk.generation_id names the latest generation and the new lk.generation_count how many there were (more than one means tools ran before the final reply). lk.speech_id is set at creation.
• Every step's llm_node, function_tool, tts_node, realtime_inference and agent_speaking nest under the one turn.
• A preemptive generation discarded for a successor answering the same user turn (a newer attempt, or the real reply after on_user_turn_completed changed the transcript) hands its open agent_turn to that successor: the wasted generation stays visible under the one turn with a preemptive_generation_discarded event, and lk.speech_id follows the speech that answered. An attempt cancelled with no successor yet (the user resumed) still ends as its own turn. The queue-wait and interruption helpers tolerate a span already ended with the speech.

agent_turn  (lk.speech_id, lk.generation_count=2; events: generation ×2)
├─ llm_node          generation 1: the tool call
├─ function_tool
├─ llm_node          generation 2: the reply
├─ tts_node
└─ agent_speaking

Tests

tests/test_agent_turn_span.py: a tool-calling reply through the fake session yields one agent_turn with two generation events, both llm_nodes, the tool and the speech inside it, and the turn outlasting the last child; a plain reply is one generation. Existing coverage (test_coverage_spans, test_eou_wait_span, test_agent_session) passes unchanged.

Stacked on #7137.

🤖 Generated with Claude Code

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

Devin Review found 1 new potential issue.

1 flag not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)

[devin-ai-integration]

🟡 Pipeline failures leave turns successful

When pipeline inference stores a failure in SpeechHandle._error, _mark_done() receives no error. The overall turn then closes successfully despite the failed reply.

Suggested change

	if isinstance(error, Exception):
	trace_utils.record_exception(span, error)
	effective_error = error or self._error
	if isinstance(effective_error, Exception):
	trace_utils.record_exception(span, effective_error)

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

Devin Review found 2 new potential issues.

1 flag not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)

[devin-ai-integration]

🟥 Turn failures can expose customer content

Failed reply tasks call _mark_done without their exception. The original traceback enters logs instead of the span's redaction-aware exception path.

(Refers to this code)

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

🟡 Rapid preemptive retries duplicate turns

When replacement precedes the canceled reply task's first run, _continue_discarded_turn transfers nothing. Both tasks later create separate spans for one turn.

Prompt for agents

Handle preemptive replacements whose discarded SpeechHandle has not opened its agent_turn span yet. In livekit-agents/livekit/agents/voice/agent_activity.py, _continue_discarded_turn currently transfers only an existing span. A newly created task may not have run before another synchronous preemptive callback replaces it, so _take_agent_turn returns None. The canceled task can later enter _agent_turn and create one span while the successor creates another. Preserve the logical turn association across this pre-start state, or prevent the canceled handle from opening an independent span. Add coverage where replacement occurs before the discarded task first enters _agent_turn.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[chenghao-mou]

This seems valid.

[devin-ai-integration]

Devin Review found 1 new potential issue.

1 flag not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)

[devin-ai-integration]

🟡 Trace sampling disables duration metrics

When tracing drops an agent_turn, _end_agent_turn returns before recording its invoke-agent duration metric. Trace sampling therefore removes latency observations.

Suggested change

	span, self._agent_turn_span = self._agent_turn_span, None
	if span is None or not span.is_recording():
	return
	from ..telemetry import otel_metrics, utils as trace_utils
	if isinstance(error, Exception):
	trace_utils.record_exception(span, error)
	if self._agent_turn_started_at is not None and self._agent_turn_agent_name is not None:
	otel_metrics.record_invoke_agent_duration(
	time.perf_counter() - self._agent_turn_started_at,
	agent_name=self._agent_turn_agent_name,
	)
	span.end()
	span, self._agent_turn_span = self._agent_turn_span, None
	if span is None:
	return
	from ..telemetry import otel_metrics, utils as trace_utils
	if self._agent_turn_started_at is not None and self._agent_turn_agent_name is not None:
	otel_metrics.record_invoke_agent_duration(
	time.perf_counter() - self._agent_turn_started_at,
	agent_name=self._agent_turn_agent_name,
	)
	if not span.is_recording():
	return
	if isinstance(error, Exception):
	trace_utils.record_exception(span, error)
	span.end()

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[chenghao-mou]

lgtm. One of the Devin comment seems valid when the task is only cancelled later:

`SpeechHandle._cancel()` does not immediately cancel the task. It:

1. Sets `_interrupt_fut`.
2. Leaves the task runnable.
3. Calls `task.cancel()` only after the five-second interruption timeout.