From 23bc97378d946d881fe54a321baa8654c3c86067 Mon Sep 17 00:00:00 2001 From: Jithendranath Gupta Yenduri Date: Fri, 25 Sep 2026 11:20:07 -0700 Subject: [PATCH 1/2] Listen for pushed LiveKit nudges and deliver each nudge once [ENG-3604] DeepTrust now pushes nudges into a LiveKit room with RoomService.SendData on topic deeptrust.nudge. Nothing in the adapter listened for them, and attach delivered every nudge on every analyze response, so a standing finding re-nudged the agent each turn. With interrupt on, the same text also went in twice: once as a system message and again as reply instructions. - listen(room, session) is the cloud way. It registers only the room data_received handler, never calls the DeepTrust API and needs no key. It returns a stop function and stops on its own when the session closes. - attach(..., room=ctx.room) is the SDK way with pushed nudges as well. When the AgentSession closes it ends the DeepTrust call. - A per-session dedupe keyed by nudge id, and by rendered text for older APIs that send no id, delivers a nudge at most once whether it arrives pushed or on an analyze response. - Delivery adds a system message to the chat context and, when interrupting, calls generate_reply() with no instructions, so the text reaches the model once and stays for the rest of the call. - Packets sent by a participant are ignored. The server API sends with no participant, and accepting participant packets would let the caller's own client write into the agent's context. - Nudge gains an optional id, parsed from analyze findings. - examples/livekit gains cloud.py for listen; main.py passes room. - Version 0.0.2. Co-Authored-By: Claude Opus 5.5 --- README.md | 45 ++++- examples/README.md | 2 +- examples/livekit/README.md | 27 ++- examples/livekit/cloud.py | 54 +++++ examples/livekit/main.py | 15 +- pyproject.toml | 2 +- src/deeptrust/_version.py | 2 +- src/deeptrust/agents/_session.py | 1 + src/deeptrust/agents/livekit.py | 220 +++++++++++++++++--- src/deeptrust/types.py | 8 +- tests/test_adapters.py | 332 ++++++++++++++++++++++++++++--- 11 files changed, 635 insertions(+), 73 deletions(-) create mode 100644 examples/livekit/cloud.py diff --git a/README.md b/README.md index 5eacb94..4b3022e 100644 --- a/README.md +++ b/README.md @@ -102,17 +102,49 @@ await call.analyze() # None. nothing new was said pip install "deeptrust-ai[livekit]" ``` +There are two ways in, and both are one line in the worker. + +**The cloud way.** DeepTrust reads the call's transcript itself and pushes each +nudge into the room, so the worker only has to listen and needs no DeepTrust +key. + +1. Connect the LiveKit project in the DeepTrust dashboard (Settings, Voice + Agents). +2. In LiveKit Cloud, add the webhook URL the dashboard gives you, so DeepTrust + hears when a room starts and ends. +3. Add `listen` to the worker: + +```python +from deeptrust.agents.livekit import listen + +listen(ctx.room, session, agent=agent) +``` + +`listen` never calls the DeepTrust API. It returns a function that stops +listening, and stops on its own when the session closes. + +**The SDK way.** The worker sends the transcript from its own process: + ```python from deeptrust.agents import DeepTrust, User from deeptrust.agents.livekit import attach -attach(session, DeepTrust(), external_id=ctx.room.name, user=caller) +attach(session, DeepTrust(), external_id=ctx.room.name, room=ctx.room, user=caller) ``` That subscribes to the session's conversation items, runs a job when the caller -says something new, and delivers the nudge. On LiveKit a nudge can interrupt: the -analysis lands while the agent is still generating, so it can stop a sentence on -its way out. Pass `interrupt=False` to shape the next turn instead. +says something new, and delivers the nudge. With `room` it also delivers nudges +DeepTrust pushes into the room. When the session closes it ends the DeepTrust +call, so post-call processing starts at once. + +Either way a nudge reaches the agent once. The same nudge arriving on a push and +on an analyze response, or reported again on a later turn, is delivered the +first time only. It goes into the agent's chat context as a system message, so +it shapes the rest of the call, not just the next reply. + +On LiveKit a nudge can interrupt: it lands while the agent is still generating, +so it can stop a sentence on its way out. Pass `interrupt=False` to shape the +next turn instead. ## ElevenLabs @@ -292,8 +324,9 @@ Point a client at it with `DEEPTRUST_BASE_URL`. ## Status -`0.0.1`, the first release. `analyze`, `end`, `watch` and both adapters work -against the hosted API. `check` is defined and raises `NotImplementedError`. +`0.0.2`. `analyze`, `end`, `watch` and the adapters work against the hosted +API. On LiveKit, `listen` receives nudges DeepTrust pushes into the room, and +`attach` delivers each nudge once instead of on every turn. `check` is defined and raises `NotImplementedError`. The shapes in `deeptrust.types` are the part most likely to move. The key travels in `X-DeepTrust-Api-Key`; the bearer form is still sent and diff --git a/examples/README.md b/examples/README.md index af21b75..ce73a58 100644 --- a/examples/README.md +++ b/examples/README.md @@ -5,7 +5,7 @@ README showing real output from a real run. | | | |---|---| -| [`livekit/`](livekit) | An agent with DeepTrust attached in one line. Nudges can interrupt a reply in progress. | +| [`livekit/`](livekit) | An agent with DeepTrust attached in one line, the SDK way or the cloud way. Nudges can interrupt a reply in progress. | | [`elevenlabs/`](elevenlabs) | An agent it provisions for you, watched from outside. No code inside the agent at all. | | [`elevenlabs-webhook/`](elevenlabs-webhook) | The call-start webhook you already own, turned into a watched call. Your process holds the socket. | diff --git a/examples/livekit/README.md b/examples/livekit/README.md index 74f7622..78af926 100644 --- a/examples/livekit/README.md +++ b/examples/livekit/README.md @@ -1,28 +1,45 @@ # LiveKit -An IT service desk agent on LiveKit with DeepTrust attached. One line of -DeepTrust code, and everything else is an ordinary agent. +An IT service desk agent on LiveKit with DeepTrust attached. There are two +workers here, one per way in, and each has one line of DeepTrust code. + +**The SDK way, `main.py`.** The worker sends the transcript from its own +process: ```python attach( session, DeepTrust(), external_id=ctx.room.name, + room=ctx.room, agent=agent, user=User(id=participant.identity, role="MEMBER"), ) ``` That wires both directions. Caller turns go out for analysis, and any nudge -that comes back is added to the agent's context and interrupts the reply in -progress. +that comes back, or that DeepTrust pushes into the room, is added to the +agent's context once and interrupts the reply in progress. When the session +closes, the DeepTrust call is ended. + +**The cloud way, `cloud.py`.** DeepTrust reads the transcript itself and +pushes nudges into the room, so the worker only listens and needs no DeepTrust +key: + +```python +listen(ctx.room, session, agent=agent) +``` + +For that, connect the LiveKit project in the DeepTrust dashboard (Settings, +Voice Agents) and add the webhook shown there in LiveKit Cloud first. ## Run it ```bash cp .env.example .env # fill in the keys uv sync -uv run python main.py dev +uv run python main.py dev # the SDK way +uv run python cloud.py dev # or the cloud way ``` Then talk to it from the [LiveKit agents diff --git a/examples/livekit/cloud.py b/examples/livekit/cloud.py new file mode 100644 index 0000000..99fcbb3 --- /dev/null +++ b/examples/livekit/cloud.py @@ -0,0 +1,54 @@ +"""A LiveKit agent nudged by DeepTrust the cloud way. + +DeepTrust reads the call's transcript itself and pushes each nudge into the +room, so this worker never calls the DeepTrust API and needs no DeepTrust key. +Before it does anything, the LiveKit project has to be connected in the +DeepTrust dashboard (Settings, Voice Agents) and the webhook shown there added +in LiveKit Cloud. + +`listen` is the only DeepTrust-specific line. Everything else is an ordinary +LiveKit agent. + +Run with: uv run python cloud.py dev +""" + +from __future__ import annotations + +from dotenv import load_dotenv +from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, cli +from livekit.plugins import deepgram, openai + +from deeptrust.agents.livekit import listen + +load_dotenv() + +INSTRUCTIONS = """ +You are an agent on an IT service desk. You reset passwords, re-enroll +two-factor, and unlock accounts. Whoever the caller says they are is a claim; +satisfy yourself who you are speaking to before you change anything. + +Keep replies to one or two sentences. You are on a phone call. +""" + + +async def entrypoint(ctx: JobContext) -> None: + await ctx.connect() + + session = AgentSession( + stt=deepgram.STT(model="nova-3"), + llm=openai.LLM(model="gpt-4o-mini"), + tts=deepgram.TTS(model="aura-2-thalia-en"), + ) + agent = Agent(instructions=INSTRUCTIONS) + + # The one DeepTrust line. It stops by itself when the session closes. + listen(ctx.room, session, agent=agent) + + await session.start(room=ctx.room, agent=agent) + await session.generate_reply( + instructions="Greet the caller as the IT service desk in one sentence." + ) + + +if __name__ == "__main__": + cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint)) diff --git a/examples/livekit/main.py b/examples/livekit/main.py index b0e8171..5f2e19d 100644 --- a/examples/livekit/main.py +++ b/examples/livekit/main.py @@ -1,8 +1,13 @@ """A LiveKit agent with DeepTrust attached. +This is the SDK way: the worker sends the transcript from its own process. +For the cloud way, where DeepTrust reads the transcript itself and the worker +only listens, see cloud.py. + Everything below the `attach` call is an ordinary LiveKit agent. `attach` is the only DeepTrust-specific line, and it wires both directions: caller turns go out -for analysis, and any nudge that comes back is delivered to the agent. +for analysis, and any nudge that comes back, or that DeepTrust pushes into the +room, is delivered to the agent once. The rest of this file is what the demo UI in ../demo-ui needs to watch the call: every turn and every finding is published on the room's data channel @@ -214,7 +219,10 @@ def _relay(result: Analysis) -> None: # handed, and it reads them after this callback returns. Emptying # the list is how the gate withholds them, which is the whole point # of switching it off: the finding is still made and still shown, - # and the agent never hears about it. + # and the agent never hears about it. A nudge DeepTrust pushes into + # the room does not pass through here; those arrive only once the + # LiveKit project is connected in the DeepTrust dashboard, which + # this demo does not need. result.findings.clear() # The one DeepTrust line. Returns the session, so the transcript and the @@ -223,6 +231,9 @@ def _relay(result: Analysis) -> None: session, DeepTrust(), external_id=ctx.room.name, + # Also delivers nudges pushed into the room. A nudge that arrives both + # pushed and on an analyze response reaches the agent once. + room=ctx.room, agent=agent, user=User( id=profile["username"], diff --git a/pyproject.toml b/pyproject.toml index c540130..dd17ffe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "deeptrust-ai" -version = "0.0.1" +version = "0.0.2" description = "QA and runtime nudges for voice agents, while the call is still happening." readme = "README.md" requires-python = ">=3.11" diff --git a/src/deeptrust/_version.py b/src/deeptrust/_version.py index f102a9c..3b93d0b 100644 --- a/src/deeptrust/_version.py +++ b/src/deeptrust/_version.py @@ -1 +1 @@ -__version__ = "0.0.1" +__version__ = "0.0.2" diff --git a/src/deeptrust/agents/_session.py b/src/deeptrust/agents/_session.py index e6903c2..79d9eaa 100644 --- a/src/deeptrust/agents/_session.py +++ b/src/deeptrust/agents/_session.py @@ -32,6 +32,7 @@ def _nudge(d: dict[str, Any] | None) -> Nudge | None: title=str(d.get("title") or ""), description=str(d.get("description") or ""), details=str(d.get("details") or ""), + id=str(d["id"]) if d.get("id") else None, ) diff --git a/src/deeptrust/agents/livekit.py b/src/deeptrust/agents/livekit.py index 7b7c4ae..8eafb49 100644 --- a/src/deeptrust/agents/livekit.py +++ b/src/deeptrust/agents/livekit.py @@ -1,19 +1,33 @@ """LiveKit adapter. +There are two ways to run DeepTrust on LiveKit, and each is one call. + +The cloud way. Connect the LiveKit project in the DeepTrust dashboard +(Settings, Voice Agents) and add the webhook it gives you in LiveKit Cloud. +DeepTrust then reads the call's transcript itself and pushes nudges into the +room. The worker only has to listen for them, and needs no DeepTrust key: + + from deeptrust.agents.livekit import listen + + listen(ctx.room, session, agent=agent) + +The SDK way. The worker sends the transcript from this process and delivers the +nudges that come back: + from deeptrust.agents import DeepTrust, User from deeptrust.agents.livekit import attach - dt = DeepTrust() attach( session, - dt, + DeepTrust(), external_id=ctx.room.name, + room=ctx.room, user=User(id=account_id, role="MEMBER"), ) -`attach` subscribes to the session's conversation items, analyzes the -transcript when the user has said something new, and delivers any nudge to the -agent. +`attach` analyzes the transcript whenever the caller says something new, and +with `room` it also listens for pushed nudges. Either way a nudge reaches the +agent once, however many times and by whichever route it arrives. A nudge is added to the agent's chat context, and by default also interrupts: because the agent runs in this process, a nudge can arrive while it is still @@ -26,18 +40,162 @@ from __future__ import annotations import asyncio +import contextlib +import json from collections.abc import Callable from typing import Any +from ..errors import DeepTrustError from ..types import Nudge, User from . import DeepTrust +# The data channel topic DeepTrust publishes nudges on, and the `type` inside +# the payload. Both are checked: the topic routes, the type confirms. +NUDGE_TOPIC = "deeptrust.nudge" + + +def _spawner() -> Callable[[Any], None]: + # asyncio holds only a weak reference to a task, so a task nobody keeps can + # be garbage collected mid-flight. Hence the set. + tasks: set[asyncio.Task[None]] = set() + + def spawn(coro: Any) -> None: + t = asyncio.create_task(coro) + tasks.add(t) + t.add_done_callback(tasks.discard) + + return spawn + + +def _read_packet(packet: Any) -> Nudge | None: + """The nudge in a data packet, or None if the packet is not one. + + Only packets sent from the server side are accepted. DeepTrust sends with + the server API, which LiveKit delivers with no participant. Anything a + participant publishes, including the caller's own client, could otherwise + put words into the agent's context by using the same topic. + """ + if getattr(packet, "topic", None) != NUDGE_TOPIC: + return None + if getattr(packet, "participant", None) is not None: + return None + try: + d = json.loads(bytes(packet.data).decode("utf-8")) + except (AttributeError, TypeError, ValueError): + return None + if not isinstance(d, dict) or d.get("type") != NUDGE_TOPIC: + return None + + def _str(key: str) -> str: + v = d.get(key) + return v if isinstance(v, str) else "" + + nudge = Nudge( + title=_str("title"), + description=_str("description"), + details=_str("details"), + id=_str("id") or None, + ) + if not nudge.render() and _str("text"): + # `text` is the rendered form. A payload that carries only that is still + # something to say, so it goes in as the description. + nudge = Nudge( + title=nudge.title, description=_str("text"), details="", id=nudge.id + ) + return nudge if nudge.render() else None + + +class _Delivery: + """Delivers nudges to one agent session, each one at most once.""" + + def __init__(self, agent_session: Any, agent: Any, interrupt: bool) -> None: + self._session = agent_session + self._agent = agent + self._interrupt = interrupt + self._seen: set[str] = set() + + async def deliver(self, nudge: Nudge) -> None: + text = nudge.render() + if not text: + return + # The push and the analyze response carry the same id for the same + # nudge. An older API sends no id, and then the text is the only thing + # the two have in common, so both are recorded and either one matches. + keys = {f"text:{text}"} + if nudge.id: + keys.add(f"id:{nudge.id}") + if keys & self._seen: + return + target = self._agent or getattr(self._session, "current_agent", None) + if target is None: + return + # Marked before the first await, so a push and a response that arrive + # together cannot both get past the check. + self._seen |= keys + + # Delivered once, as a system message in the chat context. A message + # there stays for the rest of the call and shapes every later reply, + # where `instructions` apply to one reply only. The reply generated to + # interrupt then reads it from the context, so passing the text again as + # instructions would only say the same thing twice. + chat = target.chat_ctx.copy() + chat.add_message(role="system", content=text) + await target.update_chat_ctx(chat) + if self._interrupt: + self._session.interrupt() + self._session.generate_reply() + + def listen(self, room: Any, spawn: Callable[[Any], None]) -> Callable[[], None]: + """Deliver nudges pushed into `room`. Returns a function that stops.""" + + def _on_data(packet: Any) -> None: + nudge = _read_packet(packet) + if nudge is not None: + spawn(self.deliver(nudge)) + + room.on("data_received", _on_data) + stopped = False + + def stop() -> None: + nonlocal stopped + if not stopped: + stopped = True + room.off("data_received", _on_data) + + return stop + + +def listen( + room: Any, + agent_session: Any, + *, + agent: Any = None, + interrupt: bool = True, +) -> Callable[[], None]: + """Deliver the nudges DeepTrust pushes into `room` to the agent. + + For the cloud way, where DeepTrust reads the transcript itself: this never + calls the DeepTrust API and needs no key. `room` is the `livekit.rtc.Room` + the agent is in, usually `ctx.room`. + + Pass `agent` if it is available. Otherwise it is read from + `agent_session.current_agent`. + + Returns a function that stops listening. It stops by itself when the + session closes. + """ + delivery = _Delivery(agent_session, agent, interrupt) + stop = delivery.listen(room, _spawner()) + agent_session.on("close", lambda _ev: stop()) + return stop + def attach( agent_session: Any, dt: DeepTrust, *, external_id: str, + room: Any = None, agent: Any = None, user: User | None = None, interrupt: bool = True, @@ -48,33 +206,22 @@ def attach( Analysis runs on caller turns only. Feeding an agent's own replies back in doubles the work and lets its answers reclassify the call. + Pass `room`, usually `ctx.room`, to also deliver nudges DeepTrust pushes + into the room. A nudge that arrives both ways is delivered once. + Pass `agent` if it is available. Otherwise it is read from `agent_session.current_agent`. + When the session closes, the DeepTrust call is ended so post-call + processing starts at once. + Returns the DeepTrust session, so the transcript and findings remain reachable. """ call = dt.session(external_id=external_id, user=user, platform="livekit") - tasks: set[asyncio.Task[None]] = set() - - def _spawn(coro: Any) -> None: - # asyncio holds only a weak reference to a task, so a task nobody keeps - # can be garbage collected mid-flight. Hence the set. - t = asyncio.create_task(coro) - tasks.add(t) - t.add_done_callback(tasks.discard) - - async def _deliver(nudge: Nudge) -> None: - text = nudge.render() - target = agent or getattr(agent_session, "current_agent", None) - if target is None: - return - chat = target.chat_ctx.copy() - chat.add_message(role="system", content=text) - await target.update_chat_ctx(chat) - if interrupt: - agent_session.interrupt() - agent_session.generate_reply(instructions=text) + spawn = _spawner() + delivery = _Delivery(agent_session, agent, interrupt) + stop = delivery.listen(room, spawn) if room is not None else None async def _run(role: str, text: str) -> None: call.append(role, text) @@ -86,12 +233,11 @@ async def _run(role: str, text: str) -> None: if on_analysis: on_analysis(result) for nudge in result.nudges: - await _deliver(nudge) + await delivery.deliver(nudge) # The last turn taken, so a repeat of it can be recognised. LiveKit emits a # conversation item more than once for the same speech in practice, and a - # transcript that carries the same sentence twice is analysed twice: the - # same finding is reported again and the same nudge delivered again. + # transcript that carries the same sentence twice is analysed twice. last: dict[str, str] = {} def _on_item(ev: Any) -> None: @@ -108,10 +254,20 @@ def _on_item(ev: Any) -> None: return last[role] = text - _spawn(_run(role, text)) + spawn(_run(role, text)) + + async def _end() -> None: + # The server ends an idle call on its own a few minutes later, so a + # failed end costs time, not the record. + with contextlib.suppress(DeepTrustError): + await call.end() + + def _on_close(_ev: Any) -> None: + if stop is not None: + stop() + spawn(_end()) - # Registered by call rather than by decoration: AgentSession.on is untyped - # on LiveKit's side, and decorating with it erases our own signature. - agent_session.on("conversation_item_added")(_on_item) + agent_session.on("conversation_item_added", _on_item) + agent_session.on("close", _on_close) return call diff --git a/src/deeptrust/types.py b/src/deeptrust/types.py index 52839da..608147c 100644 --- a/src/deeptrust/types.py +++ b/src/deeptrust/types.py @@ -105,11 +105,17 @@ def to_wire(self) -> dict[str, Any]: @dataclass(frozen=True) class Nudge: - """Something to tell the agent mid-call, with the action to take.""" + """Something to tell the agent mid-call, with the action to take. + + `id` identifies the nudge across the ways it can reach an agent: the same + nudge carries the same id on an analyze response and when DeepTrust pushes + it into a call. Older API versions do not send one, so it may be None. + """ title: str description: str details: str + id: str | None = None def render(self) -> str: """`description` and `details` joined, for platforms that accept a diff --git a/tests/test_adapters.py b/tests/test_adapters.py index d33b351..1a00430 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py @@ -6,7 +6,9 @@ from __future__ import annotations +import asyncio import json +from dataclasses import dataclass from typing import Any import httpx @@ -15,7 +17,7 @@ from deeptrust.agents import DeepTrust from deeptrust.agents.elevenlabs import Monitor, _read_turn, contextual_update_command -from deeptrust.agents.livekit import attach +from deeptrust.agents.livekit import attach, listen from deeptrust.agents.vapi import ( SECRET_HEADER, Bridge, @@ -56,6 +58,7 @@ def __init__(self) -> None: async def update_chat_ctx(self, chat: Any) -> None: self.updated.append(chat) + self.chat_ctx = chat class FakeChat: @@ -71,32 +74,106 @@ def add_message(self, role: str, content: str) -> None: self.messages.append((role, content)) -class FakeSession: - """Stands in for a LiveKit AgentSession.""" +class FakeEmitter: + """Shaped like `livekit.rtc.EventEmitter`, which both `AgentSession` and + `rtc.Room` are: `on(event, callback)` registers, and `on(event)` alone + returns a decorator; `off(event, callback)` removes.""" def __init__(self) -> None: - self.current_agent = FakeAgent() - self.handlers: dict[str, Any] = {} - self.interrupted = 0 - self.replies: list[str] = [] + self.handlers: dict[str, set[Any]] = {} + + def on(self, event: str, callback: Any = None) -> Any: + if callback is not None: + self.handlers.setdefault(event, set()).add(callback) + return callback - def on(self, event: str) -> Any: def deco(fn: Any) -> Any: - self.handlers[event] = fn + self.handlers.setdefault(event, set()).add(fn) return fn return deco + def off(self, event: str, callback: Any) -> None: + self.handlers.get(event, set()).discard(callback) + + def emit(self, event: str, arg: Any) -> None: + for fn in list(self.handlers.get(event, set())): + fn(arg) + + +class FakeSession(FakeEmitter): + """Stands in for a LiveKit AgentSession.""" + + def __init__(self) -> None: + super().__init__() + self.current_agent = FakeAgent() + self.interrupted = 0 + # The keyword arguments of each generate_reply call. + self.replies: list[dict[str, Any]] = [] + def interrupt(self) -> None: self.interrupted += 1 - def generate_reply(self, instructions: str) -> None: - self.replies.append(instructions) + def generate_reply(self, **kw: Any) -> None: + self.replies.append(kw) async def say(self, role: str, text: str) -> None: item = type("Item", (), {"text_content": text, "role": role})() ev = type("Ev", (), {"item": item})() - self.handlers["conversation_item_added"](ev) + self.emit("conversation_item_added", ev) + + def close(self) -> None: + self.emit("close", type("CloseEvent", (), {"reason": "user_initiated"})()) + + @property + def system_messages(self) -> list[str]: + return [c for r, c in self.current_agent.chat_ctx.messages if r == "system"] + + +@dataclass +class FakePacket: + """Shaped like `livekit.rtc.DataPacket`. `participant` is None when the + packet was sent with the server API, which is how DeepTrust sends.""" + + data: bytes + kind: int = 1 # DataPacketKind.KIND_RELIABLE + participant: Any = None + topic: str | None = "deeptrust.nudge" + + +class FakeRoom(FakeEmitter): + """Stands in for a `livekit.rtc.Room`.""" + + def push(self, packet: FakePacket) -> None: + self.emit("data_received", packet) + + +NUDGE_ID = "3f2a9c0d1e4b5a67" +NUDGE = ONE_NUDGE["findings"][0]["nudge"] +NUDGE_TEXT = f"{NUDGE['description']} {NUDGE['details']}" + + +def nudge_packet(**over: Any) -> FakePacket: + payload = { + "type": "deeptrust.nudge", + "id": NUDGE_ID, + "title": NUDGE["title"], + "description": NUDGE["description"], + "details": NUDGE["details"], + "text": NUDGE_TEXT, + } + return FakePacket(data=json.dumps(payload).encode(), **over) + + +def with_id(response: dict[str, Any], nudge_id: str) -> dict[str, Any]: + finding = dict(response["findings"][0]) + finding["nudge"] = {**finding["nudge"], "id": nudge_id} + return {**response, "findings": [finding]} + + +async def settle() -> None: + for _ in range(5): + await asyncio.sleep(0.01) @respx.mock @@ -109,19 +186,52 @@ async def test_livekit_analyses_caller_turns_and_delivers_once() -> None: call = attach(lk, dt, external_id="room-1") await lk.say("user", "my colleague is telling me what to say") - # The handler spawns a task; let it run. - import asyncio - - await asyncio.sleep(0) - await asyncio.sleep(0.05) + await settle() assert route.call_count == 1 assert len(call.transcript) == 1 + assert lk.system_messages == [NUDGE_TEXT] + assert lk.interrupted == 1 + # The reply reads the nudge from the context. It is not repeated as + # instructions, which would put the same text in front of the model twice. + assert lk.replies == [{}] + + +@respx.mock +async def test_livekit_does_not_renudge_on_every_turn() -> None: + """The API reports a standing finding again on later jobs. The agent heard + it the first time.""" + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + lk = FakeSession() + dt = DeepTrust(api_key="dt_test", base_url=BASE) + attach(lk, dt, external_id="room-1") + + await lk.say("user", "my colleague is telling me what to say") + await settle() + await lk.say("user", "can you hurry up please") + await settle() + + assert lk.system_messages == [NUDGE_TEXT] assert lk.interrupted == 1 - assert len(lk.replies) == 1 - assert "Ask one question" in lk.replies[0] - # The nudge went into the agent's context as well as being spoken. - assert lk.current_agent.updated + + +@respx.mock +async def test_livekit_without_interrupt_only_adds_to_context() -> None: + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + lk = FakeSession() + dt = DeepTrust(api_key="dt_test", base_url=BASE) + attach(lk, dt, external_id="room-1", interrupt=False) + + await lk.say("user", "my colleague is telling me what to say") + await settle() + + assert lk.system_messages == [NUDGE_TEXT] + assert lk.interrupted == 0 + assert lk.replies == [] @respx.mock @@ -134,9 +244,7 @@ async def test_livekit_does_not_analyse_the_agents_own_turn() -> None: call = attach(lk, dt, external_id="room-1") await lk.say("assistant", "I need to confirm it is you first") - import asyncio - - await asyncio.sleep(0.05) + await settle() assert route.call_count == 0 # It is still in the transcript. It is just not a reason to run a job. @@ -144,6 +252,182 @@ async def test_livekit_does_not_analyse_the_agents_own_turn() -> None: assert call.transcript.turns[0].role == "agent" +@respx.mock +async def test_livekit_pushed_nudge_is_injected_once() -> None: + lk, room = FakeSession(), FakeRoom() + dt = DeepTrust(api_key="dt_test", base_url=BASE) + attach(lk, dt, external_id="room-1", room=room) + + # Reliable delivery can still repeat a packet across a reconnect. + room.push(nudge_packet()) + room.push(nudge_packet()) + await settle() + + assert lk.system_messages == [NUDGE_TEXT] + assert lk.interrupted == 1 + assert lk.replies == [{}] + + +@respx.mock +async def test_livekit_analyze_response_with_a_pushed_id_is_ignored() -> None: + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=with_id(ONE_NUDGE, NUDGE_ID)) + ) + lk, room = FakeSession(), FakeRoom() + dt = DeepTrust(api_key="dt_test", base_url=BASE) + attach(lk, dt, external_id="room-1", room=room) + + room.push(nudge_packet()) + await settle() + await lk.say("user", "my colleague is telling me what to say") + await settle() + + assert lk.system_messages == [NUDGE_TEXT] + assert lk.interrupted == 1 + + +@respx.mock +async def test_livekit_dedupes_by_text_when_the_response_has_no_id() -> None: + """An older API sends no id on the analyze response.""" + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + lk, room = FakeSession(), FakeRoom() + dt = DeepTrust(api_key="dt_test", base_url=BASE) + attach(lk, dt, external_id="room-1", room=room) + + await lk.say("user", "my colleague is telling me what to say") + await settle() + room.push(nudge_packet()) + await settle() + + assert lk.system_messages == [NUDGE_TEXT] + + +@respx.mock +async def test_livekit_distinct_nudges_are_each_delivered() -> None: + lk, room = FakeSession(), FakeRoom() + listen(room, lk) + + room.push(nudge_packet()) + other = { + "type": "deeptrust.nudge", + "id": "0000000000000001", + "title": "Unverified", + "description": "The caller has not been verified.", + "details": None, + "text": "The caller has not been verified.", + } + room.push(FakePacket(data=json.dumps(other).encode())) + await settle() + + assert lk.system_messages == [NUDGE_TEXT, "The caller has not been verified."] + + +@pytest.mark.parametrize( + "packet", + [ + pytest.param(nudge_packet(topic="ca"), id="other-topic"), + pytest.param(nudge_packet(topic=None), id="no-topic"), + pytest.param(FakePacket(data=b"not json"), id="not-json"), + pytest.param(FakePacket(data=b"\xff\xfe"), id="not-utf8"), + pytest.param(FakePacket(data=b"[1, 2]"), id="not-an-object"), + pytest.param( + FakePacket(data=json.dumps({"type": "other", "text": "x"}).encode()), + id="other-type", + ), + pytest.param( + FakePacket(data=json.dumps({"type": "deeptrust.nudge"}).encode()), + id="empty-nudge", + ), + # A participant, the caller included, can publish on any topic. Only the + # server API sends with no participant. + pytest.param(nudge_packet(participant=object()), id="from-a-participant"), + ], +) +async def test_livekit_ignores_packets_that_are_not_deeptrust_nudges( + packet: FakePacket, +) -> None: + lk, room = FakeSession(), FakeRoom() + listen(room, lk) + + room.push(packet) + await settle() + + assert lk.system_messages == [] + assert lk.interrupted == 0 + assert lk.replies == [] + + +async def test_livekit_listen_never_touches_http() -> None: + with respx.mock(assert_all_mocked=True, assert_all_called=False) as mock: + lk, room = FakeSession(), FakeRoom() + listen(room, lk) + + await lk.say("user", "my colleague is telling me what to say") + room.push(nudge_packet()) + await settle() + lk.close() + await settle() + + assert mock.calls.call_count == 0 + assert lk.system_messages == [NUDGE_TEXT] + + +async def test_livekit_listen_stops() -> None: + lk, room = FakeSession(), FakeRoom() + stop = listen(room, lk) + stop() + stop() + + room.push(nudge_packet()) + await settle() + + assert lk.system_messages == [] + assert not room.handlers["data_received"] + + +@respx.mock +async def test_livekit_close_ends_the_call_and_stops_listening() -> None: + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + end = respx.post(f"{BASE}/agents/sessions/sess_1/end").mock( + return_value=httpx.Response(200, json={"ended": True}) + ) + lk, room = FakeSession(), FakeRoom() + dt = DeepTrust(api_key="dt_test", base_url=BASE) + attach(lk, dt, external_id="room-1", room=room) + + await lk.say("user", "my colleague is telling me what to say") + await settle() + lk.close() + await settle() + + assert end.call_count == 1 + assert not room.handlers["data_received"] + + +@respx.mock +async def test_livekit_close_survives_a_failed_end() -> None: + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + end = respx.post(f"{BASE}/agents/sessions/sess_1/end").mock( + return_value=httpx.Response(500, json={"detail": "boom"}) + ) + lk = FakeSession() + dt = DeepTrust(api_key="dt_test", base_url=BASE) + attach(lk, dt, external_id="room-1") + + await lk.say("user", "my colleague is telling me what to say") + await settle() + lk.close() + await settle() + + assert end.call_count >= 1 + + def test_elevenlabs_event_reader() -> None: assert _read_turn( { From 013395abe2bc2951d014a377b5dceac2b717e500 Mon Sep 17 00:00:00 2001 From: Jithendranath Gupta Yenduri Date: Fri, 25 Sep 2026 12:37:42 -0700 Subject: [PATCH 2/2] fix(livekit): send the agent's last reply before ending the call [ENG-3604] Found on a real LiveKit phone call: turns are sent to DeepTrust when the caller speaks, so an agent reply after the caller's last line was never sent, and close ended the call without it. On that call the lost line was the agent confirming it had reset the password. On close, attach now waits for any send in flight, sends the unsent turns, then ends the call. Co-Authored-By: Claude Opus 5.5 --- src/deeptrust/agents/livekit.py | 17 ++++++++++++++++- tests/test_adapters.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/src/deeptrust/agents/livekit.py b/src/deeptrust/agents/livekit.py index 8eafb49..91f08c7 100644 --- a/src/deeptrust/agents/livekit.py +++ b/src/deeptrust/agents/livekit.py @@ -223,6 +223,10 @@ def attach( delivery = _Delivery(agent_session, agent, interrupt) stop = delivery.listen(room, spawn) if room is not None else None + # Turns being sent right now, so the close can wait for them before it + # sends what is left and ends the call. + running: set[asyncio.Task[None]] = set() + async def _run(role: str, text: str) -> None: call.append(role, text) if role != "user": @@ -254,9 +258,20 @@ def _on_item(ev: Any) -> None: return last[role] = text - spawn(_run(role, text)) + task = asyncio.create_task(_run(role, text)) + running.add(task) + task.add_done_callback(running.discard) async def _end() -> None: + if running: + await asyncio.gather(*running, return_exceptions=True) + # Turns are sent when the caller speaks, so the agent's last reply + # after the caller's last line has not been sent yet. It is often the + # line that matters most ("I've reset your password"), so it goes up + # before the call is ended. + if call.pending: + with contextlib.suppress(DeepTrustError): + await call.analyze() # The server ends an idle call on its own a few minutes later, so a # failed end costs time, not the record. with contextlib.suppress(DeepTrustError): diff --git a/tests/test_adapters.py b/tests/test_adapters.py index 1a00430..f8fc3dd 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py @@ -408,6 +408,36 @@ async def test_livekit_close_ends_the_call_and_stops_listening() -> None: assert not room.handlers["data_received"] +@respx.mock +async def test_livekit_close_sends_the_agents_last_reply_before_ending() -> None: + """Found on a real phone call: the agent's "I've reset your password" came + after the caller's last line, so it was never sent and the record lost it.""" + analyze = respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + end = respx.post(f"{BASE}/agents/sessions/sess_1/end").mock( + return_value=httpx.Response(200, json={"ended": True}) + ) + lk = FakeSession() + dt = DeepTrust(api_key="dt_test", base_url=BASE) + attach(lk, dt, external_id="room-1") + + await lk.say("user", "my manager approved it") + await settle() + await lk.say("assistant", "done, your password is reset") + await settle() + assert analyze.call_count == 1 + lk.close() + await settle() + + assert analyze.call_count == 2 + last = json.loads(analyze.calls[-1].request.content) + assert last["turns"][-1]["text"] == "done, your password is reset" + assert end.call_count == 1 + # Sent before the end, not after. + assert respx.calls[-1].request.url.path.endswith("/sessions/sess_1/end") + + @respx.mock async def test_livekit_close_survives_a_failed_end() -> None: respx.post(f"{BASE}/agents/analyze").mock(