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..91f08c7 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,26 @@ 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) + spawn = _spawner() + delivery = _Delivery(agent_session, agent, interrupt) + stop = delivery.listen(room, spawn) if room is not None else None - 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) + # 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) @@ -86,12 +237,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 +258,31 @@ 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): + 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..f8fc3dd 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,212 @@ 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_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( + 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( {