feat(voice): add public add_conversation_item method to AgentSession

[wardandr]

Summary

Adds AgentSession.add_conversation_item(item: ChatMessage) -> bool — a public, async method to insert a ChatMessage into the session history and the active agent's chat context, emitting conversation_item_added the same way the internal audio pipeline does. Works uniformly across waterfall (STT → LLM → TTS) and realtime pipelines.

• Inserts into session.history (session-level ChatContext) and, when an agent is running, into its chat context
• Emits conversation_item_added so event listeners (transcript capture, analytics, etc.) observe the item normally
• Waterfall: mirrors the internal two-step pattern (agent ctx insert + session history + event)
• Realtime: push-then-commit — the item is pushed to the live model context via _rt_session.update_chat_ctx, and only a successful push commits locally. A failed push raises RealtimeError and leaves the session untouched, so retrying with the same item id works
• Realtime models without a mutable chat context raise RealtimeError before any mutation
• Idempotent by message ID — duplicate IDs are detected via ChatContext.get_by_id and silently skipped (returns False)
• Items added before session.start() reach session history and the event only; they are never backfilled into the agent's context once it starts

Use cases

• Late STT finals — final transcript arrives after turn commitment (Turn commits with end_of_turn_probability: null when STT transcript arrives after turn flush, splitting one utterance into two turns #6504)
• DTMF-derived input — keypad-collected dates/account numbers that should appear as user input
• Out-of-band text — SIP INFO or backend caller data that should become part of the conversation record

Current workaround this replaces

# Private API — mirrors agent_activity.py internals
self._agent._chat_ctx.items.append(user_message)
self._session._conversation_item_added(user_message)

New public API

msg = ChatMessage(role="user", content=["collected via DTMF"])
added = await session.add_conversation_item(msg)  # True if added, False if deduped

Closes #7085

Test plan

• test_add_conversation_item_appears_in_history — message retrievable from session.history
• test_add_conversation_item_emits_event — conversation_item_added event fires with correct payload
• test_add_conversation_item_visible_to_agent — message appears in agent.chat_ctx
• test_add_conversation_item_dedup_by_id — second call with same ID returns False; no duplicate event or agent-ctx entry
• test_add_conversation_item_before_start — history + event fire pre-start; pins that the item is never backfilled into the agent's context
• test_add_conversation_item_realtime_pushes_to_rt_session — item reaches the live realtime session context, local history, agent ctx, and event
• test_add_conversation_item_realtime_non_mutable_raises — raises RealtimeError with zero side effects
• test_add_conversation_item_realtime_push_failure_leaves_state_clean — failed push mutates nothing; retry with same id succeeds
• test_add_conversation_item_realtime_dedup_by_id — dedup applies on the realtime path
• Full unit suite: no regressions vs main; lint, format, and type-check (livekit.agents) pass

[CLAassistant]

All committers have signed the CLA.

[devin-ai-integration]

Devin Review found 3 new potential issues.

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

[devin-ai-integration]

🟡 Concurrent transcript duplicates conversation item

When a matching final transcript lands during update_chat_ctx, add_conversation_item commits the same ID again. History and listeners receive duplicates.

Prompt for agents

Make the realtime commit phase re-check session history after update_chat_ctx returns. If another pipeline path committed the same ID during the await, avoid inserting or emitting it again and return the idempotent result. Keep the agent context, session history, and documented return value coherent under this race.

---

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

[wardandr]

This race is already prevented. The method holds _activity_lock across the entire transaction (dedup check → provider push → local commit). The pipeline's _user_turn_completed_task path acquires the same lock before committing a transcript, so the two paths are serialized — a matching final transcript cannot land and commit while add_conversation_item is mid-flight.

[devin-ai-integration]

🔴 Fallback swaps drop added items

During a fallback swap, update_chat_ctx reports success without sending the item. add_conversation_item commits locally, but the replacement model never receives it.

Prompt for agents

Coordinate add_conversation_item with _FallbackRealtimeSession swaps. The fallback adapter currently drops update_chat_ctx calls while _swapping because ordinary callers already changed the agent context, but add_conversation_item uses push-then-commit and has not changed it yet. Ensure the item is included in the swap replay or make the update wait/retry against the replacement child before committing local state.

---

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

[wardandr]

This is a pre-existing limitation of _FallbackRealtimeSession's update_chat_ctx contract — it silently drops calls during a swap. That affects every caller of update_chat_ctx, not just add_conversation_item. Fixing the fallback adapter's swap semantics is out of scope for this PR; it would require changes to the _FallbackRealtimeSession internals.

[devin-ai-integration]

🟡 Rejected updates appear successfully added

When update_chat_ctx absorbs a provider rejection, add_conversation_item returns True and commits locally. The live model never sees the item.

Prompt for agents

Define a provider-level acknowledgement contract that lets add_conversation_item determine whether its specific item was accepted. OpenAI update_chat_ctx currently logs per-item rejections and returns normally. Do not commit or return True unless the requested item reached the provider; otherwise raise RealtimeError and retain retry semantics.

---

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

[wardandr]

Same as above — this is a pre-existing contract issue with the RealtimeSession interface. OpenAI's update_chat_ctx absorbs per-item rejections and returns normally, which affects all callers. Defining a provider-level acknowledgement contract would require changes to the RealtimeSession base class, which is out of scope here. add_conversation_item follows the same contract as every other internal caller of update_chat_ctx.