diff --git a/configs/caller_phrases.yaml b/configs/caller_phrases.yaml new file mode 100644 index 000000000..aaef84bad --- /dev/null +++ b/configs/caller_phrases.yaml @@ -0,0 +1,69 @@ +de: + backchannels: + - mhm + - aha + - ja + barge_in_openers: + - Moment— + - Entschuldigung— + - Warte— + - Also— +en: + backchannels: + - uh-huh + - mm-hmm + barge_in_openers: + - Wait— + - Sorry— + - Hold on— + - Actually— +es: + backchannels: + - ajá + - mmm + - ya + barge_in_openers: + - Espera + - Perdona + - Un momento + - Bueno +fr: + backchannels: + - hum hum + - oui + - d’accord + barge_in_openers: + - Attendez + - Pardon + - Juste + - En fait +fr-CA: + backchannels: + - hum hum + - ouais + - OK + barge_in_openers: + - Attends + - Excuse + - Juste + - En fait +hi: + backchannels: + - हूँ + - अच्छा + - हाँ + barge_in_openers: + - रुकिए + - सुनिए + - माफ़ कीजिए + - असल में +ko: + backchannels: + - 네 + - 음 + - 아 + barge_in_openers: + - 잠깐만요 + - 죄송한데 + - 아니요 + - 근데 diff --git a/configs/prompts/simulation.yaml b/configs/prompts/simulation.yaml index 798c93287..5e6ebdcae 100644 --- a/configs/prompts/simulation.yaml +++ b/configs/prompts/simulation.yaml @@ -507,3 +507,146 @@ user_simulator: For other languages, spell out characters using the standard phonetic alphabet or character names used in that language. IMPORTANT: Before ending the conversation, confirm with the agent that there are no outstanding actions. The end_call tool should only be called in a turn that is a brief goodbye — never in the same turn where you are providing the agent with data, an identifier, a request to transfer to a live agent, an approval to proceed, or any kind of additional information. + + interruption_decision: | + You are analyzing a conversation to decide if the user should interrupt the agent. + + This is what the user called to accomplish, so you can judge what they still have left + to do. The fields mean: + + - GOAL: what the user is trying to achieve overall. + - MUST HAVE: non-negotiable requirements. The user will never accept an outcome that + fails any of these, and will not hang up until all of them are met. + - NICE TO HAVE: things the user wants but will give up if necessary. + - HOW THEY EVALUATE OPTIONS: the steps the user follows when the agent presents choices. + - RESOLVED WHEN: the success condition. Once this is met the user says a brief goodbye + and ends the call, so there is nothing left worth interrupting for. + - FAILED WHEN: the failure condition. This also ends the call. + - ESCALATION: how the user handles being transferred to a live agent. + + + {user_goal} + + + Conversation history (most recent at bottom): + + + {conversation_history} + + + The agent is CURRENTLY speaking (you can see their ongoing speech in the conversation above). + + Based on the conversation so far, should the user interrupt the agent NOW? + + Consider: + - Has the user heard enough to understand what the agent is asking or saying? + - Has the user heard enough to have a response, question, or correction ready? + - Did the agent just complete the sentence which has all the pertinent information the user was looking for? + - Do NOT repeatedly interrupt the agent if it has spoken only a few words (roughly fewer than 5 words in English, or the equivalent short fragment in whatever language the conversation is in — word counts are not comparable across languages). + - Are the user's request(s) basically accomplished and the user is likely to hang up on the next turn (if so it should not interrupt)? + - Is this a logical point in the conversation to interrupt? + + Respond with ONLY "YES" if the user should interrupt now, or "NO" if they should keep listening. + + backchannel_decision: | + You simulate a natural listener who occasionally makes a brief continuer sound to show they're following along. + + The conversation may be in any language. Judge the FUNCTION of what is said, not the + presence of any particular word: every language has these sounds, and the caller draws + its own from a vocabulary configured for the conversation's language. The English + examples below illustrate the judgement, not the words to look for. + + + {conversation_history} + + + The agent is still speaking [CURRENTLY SPEAKING, INCOMPLETE]. Ignore the trailing incomplete word/phrase — focus only on the COMPLETE sentences delivered so far in the agent's current turn. + + Continuers (English: "uh-huh", "mm-hmm", "yeah"; every language has equivalents) are brief + sounds that mean "I'm listening, keep going." They: + - Happen naturally during extended speech + - Show engagement without interrupting + - Are NOT responses to specific content — just signals of attention + + Say YES if: + - The agent has completed at least 2 full, substantive sentences in their current turn + (Short phrases like "Thanks for your patience" or "Let me check on that" don't count as substantive) + - The user hasn't spoken or backchanneled recently (check the last 3 exchanges for ANY brief + acknowledgement sound from the user, in whatever language the conversation is in) + - It would feel natural to briefly signal "I'm still here" + + Say NO if: + - The agent just started speaking (fewer than 2 substantive sentences) + - The user spoke OR backchanneled within the last 2-3 exchanges + - The agent's current turn contains or ends with a question + - The agent is wrapping up or about to finish their thought + + Frequency guidance: + - Continuers are occasional, not constant + - Even when conditions seem right, real listeners only backchannel sometimes + - Aim for roughly 1 continuer per 4-6 sentences of extended agent speech + - When in doubt, say NO — silence is also natural + - Too few continuers is better than too many + + Examples (English, for illustration — apply the same judgement in any language): + + AGENT: "Hi there! How can I hel [CURRENTLY SPEAKING, INCOMPLETE]" + → NO (just started) + + AGENT: "Thanks for your patience. [CURRENTLY SPEAKING, INCOMPLETE]" + → NO (only 1 short sentence, not substantive enough) + + AGENT: "Sure, I can help with that. First I'll need to verify your account. Could you provide your email or your name and zi [CURRENTLY SPEAKING, INCOMPLETE]" + → NO (agent is asking a question) + + AGENT: "No problem. We can use your name and zip code instead. Let me look that up for you. I'll check our system now and see if I can fin [CURRENTLY SPEAKING, INCOMPLETE]" + → YES (3 substantive sentences, agent explaining process) + + AGENT: "I found your order. It includes a keyboard, thermostat, and headphones. The order was delivered last Tuesday. Now for the exchange, we have a few opti [CURRENTLY SPEAKING, INCOMPLETE]" + → YES (extended explanation with specific details) + + [If the user made a brief acknowledgement sound 2 exchanges ago] + AGENT: "...and those are the available options. Now I'll need your input on which [CURRENTLY SPEAKING, INCOMPLETE]" + → NO (user backchanneled recently, don't do it again so soon) + + Respond with ONLY "YES" or "NO". + + cascade_next_interruption: | + You just said this line to the agent: + + + {utterance} + + + Write the single line you would say to cut the agent off mid-reply, if its + answer turns out to need correcting or clarifying. + + Rules: + - Write it as a natural interruption, not a full turn. Short. + - It must follow from your own goal, not from any specific thing the agent + might say — you have not heard the reply yet. + - Write it in the same language as the line above; the caller does not switch + languages mid-call. + - Reply with ONLY the spoken line, nothing else. No quotes, no labels. + - If you would have no reason to interrupt whatever the agent says next, + reply with exactly NONE. + + cascade_relevance_gate: | + The caller prepared this line a moment ago, before hearing the rest of what + the agent is currently saying: + + + {candidate} + + + Here is what the agent has said so far in its current turn: + + + {heard} + + + Would saying the prepared line right now still make sense as a natural + interruption? Answer NO if it has become irrelevant, already been addressed, + or would read as a non-sequitur. + + Respond with ONLY "YES" or "NO". diff --git a/docs/assistant_server_contract.md b/docs/assistant_server_contract.md index 6b3da3b1d..4f680d9ac 100644 --- a/docs/assistant_server_contract.md +++ b/docs/assistant_server_contract.md @@ -125,6 +125,7 @@ and decoding. | `start` | Stream opened, contains `streamSid` | Parse `msg["start"]["streamSid"]` | | `media` | Audio chunk (8 kHz mulaw, base64) | `parse_twilio_media_message(raw)` → `bytes` | | `stop` | Stream closed | Break receive loop | +| `truncate` | Cascade tick-driven barge-in; discard audio after `audio_end_ms` | OpenAI Realtime truncates its active audio item | | `user_speech_start` | User began speaking (wall-clock timestamp) | `msg["timestamp_ms"]` | | `user_speech_stop` | User stopped speaking (wall-clock timestamp) | `msg["timestamp_ms"]` | diff --git a/docs/cascade_user_simulator.md b/docs/cascade_user_simulator.md index 7263b54fb..8c70a8e62 100644 --- a/docs/cascade_user_simulator.md +++ b/docs/cascade_user_simulator.md @@ -57,3 +57,37 @@ The caller ends with `goodbye` when its LLM calls `end_call`, or `timeout` when conversation budget expires. Sustained assistant silence ends with `inactivity_timeout`. A stalled tick-driven provider ends with the distinct `provider_stalled` reason and preserves partial artifacts for diagnosis. + +## Experimental out-of-turn behaviors + +These behaviors are opt-in and are still being tested. Enable them independently +in the cascade configuration: + +```dotenv +EVA_USER_SIMULATOR__DECISION_LLM=user-llm +EVA_USER_SIMULATOR__ENABLE_BACKCHANNEL=true +EVA_USER_SIMULATOR__ENABLE_INTERRUPTIONS=true +EVA_USER_SIMULATOR__SPECULATIVE_GENERATION=true +``` + +All three behavior flags default to `false`. `decision_llm` selects the deployment +for listener and relevance checks; its default is `user-llm`. + +Backchannels play cached continuers while the assistant speaks without taking an +ordinary caller turn. Interruptions generate goal-aware barge-ins, limited to one +per detected assistant turn. Speculative generation pre-renders candidate +interruptions and checks their relevance before use; it is useful when +interruptions are also enabled. + +The tick-driven adapter reports the played audio position when a barge-in reaches +the wire, discards unheard buffered audio, and requests OpenAI Realtime item +truncation. The real-time adapter retains its normal streaming behavior. + +`configs/caller_phrases.yaml` supplies localized continuers and interruption +openers. `scripts/add_culture_data.py` generates these phrase sets alongside other +language data. Phrase audio is cached across conversations. + +The decision trace adds listener verdicts, skipped-check context, interruption +outcomes, and speculative-candidate decisions. Use it alongside the event log and +audio to evaluate actual timing; passing unit tests alone does not establish +behavioral quality. diff --git a/scripts/add_culture_data.py b/scripts/add_culture_data.py index 4a0b8211f..e86f7c9cb 100644 --- a/scripts/add_culture_data.py +++ b/scripts/add_culture_data.py @@ -103,6 +103,7 @@ REPO_ROOT = Path(__file__).resolve().parent.parent DATA_DIR = REPO_ROOT / "data" INITIAL_MESSAGES_PATH = REPO_ROOT / "configs" / "agents" / "initial_messages.yaml" +CALLER_PHRASES_PATH = REPO_ROOT / "configs" / "caller_phrases.yaml" WER_CONFIGS_DIR = REPO_ROOT / "src" / "eva" / "utils" / "wer_normalization" / "configs" DEFAULT_MODEL = "gpt-5.5-2026-04-23" @@ -353,6 +354,63 @@ def _update_initial_messages(language: str, message: str) -> None: INITIAL_MESSAGES_PATH.write_text(yaml.safe_dump(existing, allow_unicode=True, sort_keys=True), encoding="utf-8") +async def _translate_caller_phrases( + language: str, language_name: str, llm: LLMClient, overwrite: bool = False +) -> dict[str, list[str]] | None: + """Generate the simulated caller's out-of-turn vocabulary for ``language``. + + These are the continuers and barge-in openers the caller speaks *as audio*, so + they must be things a native speaker actually says, not translations of the + English ones — "uh-huh" has no word-for-word equivalent in most languages. + + Returns None when the language already has phrases and ``overwrite`` is unset. + """ + existing: dict[str, Any] = {} + if CALLER_PHRASES_PATH.exists(): + existing = yaml.safe_load(CALLER_PHRASES_PATH.read_text(encoding="utf-8")) or {} + if language in existing and not overwrite: + return None + + english = existing.get("en", {}) + prompt = ( + f"You are localising a simulated phone caller for {language_name}.\n\n" + "Produce two short lists of things the caller says out loud:\n\n" + "1. backchannels — brief continuer sounds meaning 'I am listening, keep going', " + "said while the other person is still talking. Give the sounds a native speaker " + "actually makes, not translations of English ones.\n" + "2. barge_in_openers — the very first word or two of an interruption, said just " + "before the interrupting sentence. Keep them to one or two words so they sound " + "like a real cut-in.\n\n" + "These are spoken aloud by a text-to-speech voice, so use ordinary spelling with " + "no stage directions, no parentheses and no transliteration hints. Give 2-3 " + "backchannels and 3-4 openers.\n\n" + f"For reference, the English set is: {json.dumps(english, ensure_ascii=False)}\n\n" + 'Return JSON: {"backchannels": ["..."], "barge_in_openers": ["..."]}' + ) + text, _ = await llm.generate_text( + [{"role": "user", "content": prompt}], + response_format={"type": "json_object"}, + ) + data = extract_and_load_json(text) + backchannels = [str(p).strip() for p in (data.get("backchannels") or []) if str(p).strip()] + openers = [str(p).strip() for p in (data.get("barge_in_openers") or []) if str(p).strip()] + if not backchannels or not openers: + raise ValueError(f"Caller phrase generation returned an incomplete result: {data!r}") + return {"backchannels": backchannels, "barge_in_openers": openers} + + +def _update_caller_phrases(language: str, phrases: dict[str, list[str]]) -> None: + """Merge one language's phrase set into configs/caller_phrases.yaml.""" + existing: dict[str, Any] = {} + if CALLER_PHRASES_PATH.exists(): + existing = yaml.safe_load(CALLER_PHRASES_PATH.read_text(encoding="utf-8")) or {} + if existing.get(language) == phrases: + return + existing[language] = phrases + CALLER_PHRASES_PATH.parent.mkdir(parents=True, exist_ok=True) + CALLER_PHRASES_PATH.write_text(yaml.safe_dump(existing, allow_unicode=True, sort_keys=True), encoding="utf-8") + + async def _translate_aliases( name_to_base: dict[str, list[str]], language_name: str, @@ -680,6 +738,16 @@ async def amain(args: argparse.Namespace) -> int: _update_initial_messages(args.language, initial_message) logger.info(f"Updated {INITIAL_MESSAGES_PATH}") + logger.info(f"Generating caller out-of-turn phrases for {args.language_name}") + caller_phrases = await _translate_caller_phrases(args.language, args.language_name, llm, args.overwrite_all) + if caller_phrases is None: + logger.info(f"Caller phrases for {args.language} already present — skipping") + elif args.dry_run: + logger.info(f"[dry-run] would write caller phrases for {args.language}: {caller_phrases}") + else: + _update_caller_phrases(args.language, caller_phrases) + logger.info(f"Updated {CALLER_PHRASES_PATH}: {caller_phrases}") + for domain in domains: logger.info(f"=== Domain: {domain} ===") if domain == "airline": diff --git a/src/eva/__init__.py b/src/eva/__init__.py index 575a18a3d..385df2d4d 100644 --- a/src/eva/__init__.py +++ b/src/eva/__init__.py @@ -7,7 +7,7 @@ # Bump simulation_version when changes affect benchmark outputs (agent code, # user simulator, orchestrator, simulation prompts, agent configs, tool mocks). -simulation_version = "2.0.34" +simulation_version = "2.0.35" # Bump metrics_version when changes affect metric computation (metrics code, # judge prompts, pricing tables, postprocessor). diff --git a/src/eva/assistant/openai_realtime_server.py b/src/eva/assistant/openai_realtime_server.py index 5987d9b49..5582b2579 100644 --- a/src/eva/assistant/openai_realtime_server.py +++ b/src/eva/assistant/openai_realtime_server.py @@ -69,6 +69,8 @@ class _AssistantResponseState: first_audio_wall_ms: str | None = None responding: bool = False has_function_calls: bool = False + active_item_id: str | None = None + """Item currently producing audio; the truncation target on a caller barge-in.""" class OpenAIRealtimeAssistantServer(AbstractAssistantServer): @@ -389,6 +391,10 @@ async def _forward_user_audio(self, websocket: WebSocket, conn: Any) -> None: logger.debug("Twilio stream stopped") break + if event_type == "truncate": + await self._truncate_response(conn, int(data.get("audio_end_ms", 0))) + continue + if event_type == "user_speech_start": # Timestamp from audio_interface when user audio actually started self._audio_interface_speech_start_ts = data.get("timestamp_ms") @@ -514,6 +520,23 @@ async def _handle_openai_event( case _: logger.debug(f"Unhandled {self._service_name} event: {event_type}") + async def _truncate_response(self, conn: Any, audio_end_ms: int) -> None: + """Discard generated audio past the position the caller actually heard. + + Only a tick-driven caller sends this: with pacing off the provider runs + ahead of the caller's playout, so without truncation it would believe the + caller heard seconds of audio that were never released. + """ + item_id = self._assistant_state.active_item_id + if not item_id: + return + try: + await conn.conversation.item.truncate(item_id=item_id, content_index=0, audio_end_ms=audio_end_ms) + except Exception as exc: + logger.warning(f"Truncating item {item_id} at {audio_end_ms}ms failed: {exc}") + return + logger.info(f"Truncated item {item_id} at {audio_end_ms}ms on caller barge-in") + # ── Event handlers ──────────────────────────────────────────────── async def _on_speech_started(self, event: Any) -> None: @@ -600,6 +623,12 @@ async def _on_audio_delta(self, event: Any, audio_output_queue: asyncio.Queue[by return pcm16_bytes = base64.b64decode(delta_b64) + # Read off the delta rather than response.output_item.added: it names the same + # item and is already dispatched here, so no extra event handler is needed. + item_id = getattr(event, "item_id", None) + if item_id: + self._assistant_state.active_item_id = item_id + if self._assistant_state.first_audio_wall_ms is None: self._assistant_state.first_audio_wall_ms = _wall_ms() self._assistant_state.responding = True diff --git a/src/eva/models/config.py b/src/eva/models/config.py index c59611e86..a747b4381 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -505,6 +505,22 @@ class CascadeSimulatorConfig(BaseModel): description="Provider-native keyword arguments passed through to the TTS client.", ) + decision_llm: str = Field( + "user-llm", + description=( + "Model for the YES/NO interrupt, backchannel, and relevance checks. Defaults to the " + "caller's own deployment so it resolves through the same EVA_MODEL_LIST router; point " + "it at a cheaper deployment to cut the cost of the per-tick checks." + ), + ) + + enable_backchannel: bool = Field(False, description="Caller emits continuers while the assistant speaks.") + enable_interruptions: bool = Field(False, description="Caller may barge in reacting to the assistant mid-turn.") + speculative_generation: bool = Field( + False, + description="Pre-render a candidate interruption on the turn call, gated by a relevance check before firing.", + ) + UserSimulatorConfig = Annotated[ ElevenLabsSimulatorConfig | OpenAIRealtimeSimulatorConfig | CascadeSimulatorConfig, diff --git a/src/eva/user_simulator/cascade/adapter/base.py b/src/eva/user_simulator/cascade/adapter/base.py index 3e90ae09c..c81914c66 100644 --- a/src/eva/user_simulator/cascade/adapter/base.py +++ b/src/eva/user_simulator/cascade/adapter/base.py @@ -20,8 +20,12 @@ async def start(self) -> None: """Establish the connection. Must return once ready to exchange audio.""" @abstractmethod - async def run_tick(self, tick_number: int, outgoing_audio: bytes | None) -> TickResult: - """Send this tick's caller audio (None means silence) and collect what arrived.""" + async def run_tick(self, tick_number: int, outgoing_audio: bytes | None, *, barge_in: bool = False) -> TickResult: + """Send this tick's caller audio (None means silence) and collect what arrived. + + ``barge_in`` signals that this tick begins an interruption. Adapters that + cannot truncate provider-side audio ignore it. + """ @abstractmethod async def stop(self) -> None: diff --git a/src/eva/user_simulator/cascade/adapter/realtime_ws.py b/src/eva/user_simulator/cascade/adapter/realtime_ws.py index b59aaa255..4dd3e817e 100644 --- a/src/eva/user_simulator/cascade/adapter/realtime_ws.py +++ b/src/eva/user_simulator/cascade/adapter/realtime_ws.py @@ -69,8 +69,12 @@ async def start(self) -> None: await self._ws.send(json.dumps({"event": event, "conversation_id": self._conversation_id})) self._receive_task = asyncio.create_task(self._receive_loop()) - async def run_tick(self, tick_number: int, outgoing_audio: bytes | None) -> TickResult: - """Send one tick of caller audio at wire cadence and collect one tick of assistant audio.""" + async def run_tick(self, tick_number: int, outgoing_audio: bytes | None, *, barge_in: bool = False) -> TickResult: + """Send one tick of caller audio at wire cadence and collect one tick of assistant audio. + + ``barge_in`` is accepted and ignored: this assistant paces its own output, so + it has generated nothing past what the caller already heard to discard. + """ tick_start = asyncio.get_event_loop().time() if self._error is not None: raise RuntimeError("RealtimeWSAdapter receive loop failed") from self._error diff --git a/src/eva/user_simulator/cascade/adapter/tick_driven.py b/src/eva/user_simulator/cascade/adapter/tick_driven.py index 89f4d2adb..7c0944dea 100644 --- a/src/eva/user_simulator/cascade/adapter/tick_driven.py +++ b/src/eva/user_simulator/cascade/adapter/tick_driven.py @@ -33,7 +33,7 @@ from eva.user_simulator.cascade.adapter.realtime_ws import FRAMES_PER_TICK, RealtimeWSAdapter from eva.user_simulator.cascade.constants import BYTES_PER_TICK, SILENCE_BYTE, TICK_DURATION_MS -from eva.user_simulator.cascade.tick_result import TickResult, split_tick_audio +from eva.user_simulator.cascade.tick_result import TickResult, played_audio_ms, split_tick_audio from eva.utils.logging import get_logger logger = get_logger(__name__) @@ -69,14 +69,39 @@ def __init__( bytes_per_tick=bytes_per_tick, perturbator=perturbator, ) + self._ticks_released = 0 self._last_inbound_monotonic = time.monotonic() self._audio_arrived = asyncio.Event() - async def run_tick(self, tick_number: int, outgoing_audio: bytes | None) -> TickResult: - """Send this tick's caller audio unpaced and release exactly one tick of assistant audio.""" + @property + def played_ms(self) -> int: + """Assistant audio released into the conversation so far, in simulated ms.""" + return played_audio_ms(ticks_released=self._ticks_released) + + async def run_tick(self, tick_number: int, outgoing_audio: bytes | None, *, barge_in: bool = False) -> TickResult: + """Send this tick's caller audio unpaced and release exactly one tick of assistant audio. + + When ``barge_in`` is set, first tell the assistant to discard the audio it + generated past the position the caller has actually heard. + """ if self._error is not None: raise RuntimeError("TickDrivenAdapter receive loop failed") from self._error + interruption_start: int | None = None + if barge_in: + interruption_start = self.played_ms + await self._ws.send( + json.dumps( + { + "event": "truncate", + "conversation_id": self._conversation_id, + "audio_end_ms": interruption_start, + } + ) + ) + # Everything already buffered is audio the caller never heard. + self._inbound.clear() + is_speaking = bool(outgoing_audio) if is_speaking and not self._caller_speaking: await self._send_speech_event("user_speech_start") @@ -96,6 +121,8 @@ async def run_tick(self, tick_number: int, outgoing_audio: bytes | None) -> Tick raw = bytes(self._inbound[: self._bytes_per_tick]) del self._inbound[: len(raw)] chunk, _ = split_tick_audio(raw, self._bytes_per_tick) + if raw: + self._ticks_released += 1 stalled = self._provider_has_stalled(bool(raw)) @@ -107,6 +134,7 @@ async def run_tick(self, tick_number: int, outgoing_audio: bytes | None) -> Tick assistant_audio=chunk, assistant_audio_raw_bytes=len(raw), wall_clock_ms=int(time.time() * 1000), + interruption_audio_start_ms=interruption_start, provider_stalled=stalled, ) diff --git a/src/eva/user_simulator/cascade/constants.py b/src/eva/user_simulator/cascade/constants.py index 9adb448b7..94e5c930f 100644 --- a/src/eva/user_simulator/cascade/constants.py +++ b/src/eva/user_simulator/cascade/constants.py @@ -35,6 +35,9 @@ SILENCE_BYTE = b"\x00" """PCM16 silence, used to pad partial ticks.""" +LISTENER_CHECK_INTERVAL_MS = 2000 +"""How often the interrupt and backchannel checks run while the assistant speaks.""" + def ms_to_ticks(milliseconds: int) -> int: """Convert milliseconds to whole ticks, flooring.""" diff --git a/src/eva/user_simulator/cascade/decision_log.py b/src/eva/user_simulator/cascade/decision_log.py index 3d75cd3db..f9cc29a73 100644 --- a/src/eva/user_simulator/cascade/decision_log.py +++ b/src/eva/user_simulator/cascade/decision_log.py @@ -1,4 +1,4 @@ -"""Per-tick diagnostic trace for the cascade caller.""" +"""Per-tick diagnostic trace for the cascade caller's out-of-turn decisions.""" from __future__ import annotations @@ -12,7 +12,17 @@ class DecisionLog: - """Write and flush simulator diagnostics as they occur.""" + """Records every listener check, including the ones that declined or never ran. + + `user_simulator_events.jsonl` only carries actions the caller took, so a check that + ran and said NO is indistinguishable there from a check that never fired. This trace + separates the two, which is the difference between "the model does not want to + interrupt" and "the interrupt path is unreachable". + + Rows are written and flushed as they happen rather than buffered to the end: a + conversation that dies mid-run is exactly the one worth having a trace for, and the + event log's save-at-exit is why failed attempts currently leave no diagnostics at all. + """ def __init__(self, output_path: Path) -> None: self.output_path = output_path diff --git a/src/eva/user_simulator/cascade/decisions.py b/src/eva/user_simulator/cascade/decisions.py new file mode 100644 index 000000000..6afedf3d7 --- /dev/null +++ b/src/eva/user_simulator/cascade/decisions.py @@ -0,0 +1,95 @@ +"""Listener-reaction checks: should the caller interrupt or backchannel right now.""" + +from __future__ import annotations + +import asyncio +import time +from dataclasses import dataclass, field +from typing import Protocol + +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + + +class DecisionLLM(Protocol): + """Minimal interface the checks need from a model client.""" + + async def decide(self, prompt: str) -> str: + """Return the model's raw reply to a YES/NO question.""" + ... + + +@dataclass(frozen=True) +class CheckTrace: + """What one YES/NO check actually did, for the diagnostic trace.""" + + ran: bool + raw: str = "" + latency_ms: int = 0 + error: str = "" + + +@dataclass(frozen=True) +class ListenerVerdict: + """Outcome of one check tick.""" + + should_interrupt: bool + should_backchannel: bool + interrupt_trace: CheckTrace = field(default_factory=lambda: CheckTrace(ran=False)) + backchannel_trace: CheckTrace = field(default_factory=lambda: CheckTrace(ran=False)) + + +def parse_yes_no(raw: str) -> bool: + """Read a bare YES/NO reply. Anything unrecognized counts as NO.""" + return raw.strip().upper() == "YES" + + +class ListenerDecisions: + """Runs the interrupt and backchannel checks concurrently against the partial transcript. + + Both fail closed: an exception yields "don't act", so a provider hiccup can + never inject a barge-in that the caller never actually decided to make. + """ + + def __init__( + self, llm: DecisionLLM, *, interrupt_prompt: str, backchannel_prompt: str, user_goal: str = "" + ) -> None: + self._llm = llm + self._interrupt_prompt = interrupt_prompt + self._backchannel_prompt = backchannel_prompt + self._user_goal = user_goal + + async def evaluate( + self, conversation_history: str, *, allow_interrupt: bool, allow_backchannel: bool + ) -> ListenerVerdict: + """Run whichever checks are enabled. Interrupt wins ties (tau: streaming.py:2549).""" + interrupt_trace, backchannel_trace = await asyncio.gather( + self._check(self._interrupt_prompt, conversation_history, enabled=allow_interrupt), + self._check(self._backchannel_prompt, conversation_history, enabled=allow_backchannel), + ) + interrupt = interrupt_trace.ran and parse_yes_no(interrupt_trace.raw) + backchannel = backchannel_trace.ran and parse_yes_no(backchannel_trace.raw) + return ListenerVerdict( + should_interrupt=interrupt, + should_backchannel=backchannel and not interrupt, + interrupt_trace=interrupt_trace, + backchannel_trace=backchannel_trace, + ) + + async def _check(self, template: str, conversation_history: str, *, enabled: bool) -> CheckTrace: + """Ask the model one YES/NO question, reporting what happened rather than just the answer. + + Both templates are filled with the same arguments; `str.format` ignores the ones a + given prompt does not use, so the backchannel prompt needs no goal slot. + """ + if not enabled: + return CheckTrace(ran=False) + started = time.monotonic() + try: + filled = template.format(conversation_history=conversation_history, user_goal=self._user_goal) + reply = await self._llm.decide(filled) + except Exception as exc: + logger.warning(f"Listener check failed, defaulting to no action: {exc}") + return CheckTrace(ran=True, latency_ms=int((time.monotonic() - started) * 1000), error=str(exc)) + return CheckTrace(ran=True, raw=reply, latency_ms=int((time.monotonic() - started) * 1000)) diff --git a/src/eva/user_simulator/cascade/phrase_cache.py b/src/eva/user_simulator/cascade/phrase_cache.py new file mode 100644 index 000000000..69d7e9668 --- /dev/null +++ b/src/eva/user_simulator/cascade/phrase_cache.py @@ -0,0 +1,84 @@ +"""Pre-rendered audio for the caller's fixed phrase vocabularies.""" + +from __future__ import annotations + +import asyncio +import random +from typing import ClassVar, Protocol + +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + + +class SpeechSynthesizer(Protocol): + """Minimal interface the cache needs from a TTS client.""" + + async def synthesize(self, text: str, *, voice_id: str) -> bytes: + """Render text to PCM16 audio.""" + ... + + +class PhraseCache: + """Renders a fixed vocabulary once so it can be voiced at zero latency. + + Backchannels and barge-in openers are short, fixed, and voice-stable, which + makes them cacheable — and caching is the only way a 300ms "mm-hmm" reliably + lands on the tick the decision chose. + + The audio is held **per process, keyed by (voice_id, phrase)**, not per + simulator. The vocabulary depends only on the voice, so a run of N records + against two voices needs two renders of each phrase rather than 2N: every + conversation after the first finds the audio already there and starts without + waiting on TTS at all. Only the RNG is per-instance, so each conversation + still chooses its phrases independently and reproducibly. + """ + + _audio: ClassVar[dict[tuple[str, str], bytes]] = {} + _render_lock: ClassVar[asyncio.Lock | None] = None + + def __init__(self, tts: SpeechSynthesizer, *, voice_id: str, seed: int = 0) -> None: + self._tts = tts + self._voice_id = voice_id + self._rng = random.Random(seed) + + @classmethod + def _lock(cls) -> asyncio.Lock: + """Lazily create the shared render lock, on whichever loop is running.""" + if cls._render_lock is None: + cls._render_lock = asyncio.Lock() + return cls._render_lock + + @classmethod + def clear(cls) -> None: + """Drop all cached audio. For tests, and for a voice set changing mid-process.""" + cls._audio.clear() + + async def prerender(self, phrases: list[str]) -> None: + """Synthesize any phrase this voice has not rendered yet. + + Serialized across conversations: concurrent records share one voice, so + without the lock each would discover the same empty cache and render the + same phrases. The lock is held only for genuine misses. + """ + async with self._lock(): + missing = [p for p in phrases if (self._voice_id, p) not in self._audio] + if not missing: + logger.debug(f"All {len(phrases)} caller phrases already rendered for {self._voice_id}") + return + rendered = await asyncio.gather(*(self._tts.synthesize(p, voice_id=self._voice_id) for p in missing)) + self._audio.update( + {(self._voice_id, phrase): audio for phrase, audio in zip(missing, rendered, strict=True)} + ) + logger.info(f"Pre-rendered {len(missing)} caller phrases for voice {self._voice_id}") + + def get(self, phrase: str) -> bytes: + """Return cached audio for a phrase in this cache's voice.""" + key = (self._voice_id, phrase) + if key not in self._audio: + raise KeyError(f"Phrase not pre-rendered for voice {self._voice_id}: {phrase!r}") + return self._audio[key] + + def choose(self, phrases: list[str]) -> str: + """Pick a phrase using the cache's seeded RNG, so runs stay reproducible.""" + return self._rng.choice(phrases) diff --git a/src/eva/user_simulator/cascade/phrases.py b/src/eva/user_simulator/cascade/phrases.py new file mode 100644 index 000000000..6031d92d9 --- /dev/null +++ b/src/eva/user_simulator/cascade/phrases.py @@ -0,0 +1,102 @@ +"""Per-language phrase vocabularies for the caller's out-of-turn behavior. + +Data rather than constants because these are *words*, and words are language +specific. Timing thresholds stay in `constants.py` — varying those across runs +makes metrics incomparable, whereas speaking English continuers into a French +conversation is simply wrong. + +Generated per language by `scripts/add_culture_data.py`, alongside the initial +message it already translates, so a run only ever reads this file. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import yaml + +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + +PHRASES_PATH = Path(__file__).resolve().parents[4] / "configs" / "caller_phrases.yaml" + +FALLBACK_LANGUAGE = "en" + +_cache: dict[str, CallerPhrases] = {} + + +@dataclass(frozen=True) +class CallerPhrases: + """The fixed things the caller can say without taking a turn.""" + + backchannels: list[str] + """Continuers: "I'm listening, keep going".""" + + barge_in_openers: list[str] + """Opening fragment of an interruption, voiced ahead of the content.""" + + @property + def vocabulary(self) -> list[str]: + """Every phrase that needs pre-rendering, in a stable order.""" + return [*self.backchannels, *self.barge_in_openers] + + +def _read_file() -> dict[str, Any]: + """Load the phrase file, or return empty when it is missing or unreadable.""" + try: + data = yaml.safe_load(PHRASES_PATH.read_text(encoding="utf-8")) + except OSError as exc: + logger.warning(f"Caller phrase file unreadable ({exc}); out-of-turn behavior has no vocabulary") + return {} + return data or {} + + +def candidate_languages(language: str) -> list[str]: + """Language tags to try, most specific first: 'fr-CA' -> 'fr-CA', 'fr', 'en'. + + A regional variant nearly always shares its continuers with the base language, + so falling back to it is far better than falling back to English. + """ + candidates = [language] + if "-" in language: + candidates.append(language.split("-", 1)[0]) + if FALLBACK_LANGUAGE not in candidates: + candidates.append(FALLBACK_LANGUAGE) + return candidates + + +def load_phrases(language: str) -> CallerPhrases: + """Return the caller's phrase vocabulary for a language. + + Falls back to the base language and then to English rather than raising: a + missing translation should degrade the realism of the behavior, not abort the + run. The fallback is logged because English continuers in a non-English call + are a defect in the data, not an acceptable outcome. + """ + if language in _cache: + return _cache[language] + + data = _read_file() + for candidate in candidate_languages(language): + entry = data.get(candidate) + if not entry: + continue + if candidate != language: + logger.warning( + f"No caller phrases for {language!r}; falling back to {candidate!r}. " + f"Run scripts/add_culture_data.py --language {language} to generate them." + ) + phrases = CallerPhrases( + backchannels=list(entry.get("backchannels") or []), + barge_in_openers=list(entry.get("barge_in_openers") or []), + ) + _cache[language] = phrases + return phrases + + logger.error(f"No caller phrases for {language!r} and no {FALLBACK_LANGUAGE!r} fallback in {PHRASES_PATH}") + empty = CallerPhrases(backchannels=[], barge_in_openers=[]) + _cache[language] = empty + return empty diff --git a/src/eva/user_simulator/cascade/scheduler.py b/src/eva/user_simulator/cascade/scheduler.py index bc3ca57b5..8d96d57f9 100644 --- a/src/eva/user_simulator/cascade/scheduler.py +++ b/src/eva/user_simulator/cascade/scheduler.py @@ -5,6 +5,7 @@ from eva.user_simulator.cascade.adapter.base import Adapter from eva.user_simulator.cascade.constants import ( BYTES_PER_TICK, + LISTENER_CHECK_INTERVAL_MS, WAIT_TO_RESPOND_OTHER_MS, WAIT_TO_RESPOND_SELF_MS, ms_to_ticks, @@ -34,6 +35,8 @@ def __init__(self, adapter: Adapter, *, bytes_per_tick: int = BYTES_PER_TICK) -> self._assistant_has_spoken = False self._awaiting_reply = False self._caller_spoke_this_tick = False + self._backchannel_bytes = 0 + self._barge_in_armed = False @property def tick(self) -> int: @@ -49,6 +52,25 @@ def enqueue_utterance(self, audio: bytes) -> None: """ self._playout.extend(audio) + def enqueue_backchannel(self, audio: bytes) -> None: + """Append a continuer, which sounds but does not take the caller's turn. + + A backchannel earns no reply, so counting it as a turn leaves the caller + waiting for one that never comes and the call dies at the inactivity + timeout instead of reaching a goodbye. + """ + self._backchannel_bytes += len(audio) + self._playout.extend(audio) + + def arm_barge_in(self) -> None: + """Mark the next tick that puts caller audio on the wire as an interruption. + + Armed rather than passed directly because an interruption is enqueued as + audio and only reaches the wire on a later tick; the truncation must carry + the played position as of *that* tick, not as of the decision. + """ + self._barge_in_armed = True + @property def caller_is_speaking(self) -> bool: """Whether caller audio is still queued for playout.""" @@ -74,6 +96,12 @@ def assistant_is_speaking(self) -> bool: """Whether the assistant produced audio on the most recent tick.""" return self._ticks_since_assistant_speech == 0 + def is_check_tick(self) -> bool: + """Whether the listener-reaction checks should run now (tau: streaming.py:2514-2521).""" + if not self.assistant_is_speaking or self.caller_is_speaking: + return False + return self.tick % ms_to_ticks(LISTENER_CHECK_INTERVAL_MS) == 0 + def may_take_turn(self) -> bool: """Whether both silence thresholds are satisfied (tau: streaming.py:2590-2606). @@ -100,8 +128,16 @@ async def run_tick(self) -> TickResult: raised exception leaves the queue and tick count exactly as they were. """ outgoing, consumed = self._peek_chunk() - result = await self._adapter.run_tick(self._tick, outgoing) + barge_in = self._barge_in_armed and outgoing is not None + result = await self._adapter.run_tick(self._tick, outgoing, barge_in=barge_in) del self._playout[:consumed] + if barge_in: + self._barge_in_armed = False + + # Backchannel bytes sit at the head of the queue, so this tick is a continuer + # only while they remain. Anything past them is real speech and takes the turn. + was_backchannel = consumed > 0 and self._backchannel_bytes > 0 + self._backchannel_bytes = max(0, self._backchannel_bytes - consumed) self._caller_spoke_this_tick = outgoing is not None self._ticks_since_caller_speech = 0 if outgoing else self._ticks_since_caller_speech + 1 @@ -111,7 +147,7 @@ async def run_tick(self) -> TickResult: self._assistant_has_spoken = self._assistant_has_spoken or result.has_assistant_speech if result.has_assistant_speech: self._awaiting_reply = False - if outgoing: + if outgoing and not was_backchannel: self._awaiting_reply = True self._tick += 1 diff --git a/src/eva/user_simulator/cascade/simulator.py b/src/eva/user_simulator/cascade/simulator.py index 852e9bb23..b650f8ef5 100644 --- a/src/eva/user_simulator/cascade/simulator.py +++ b/src/eva/user_simulator/cascade/simulator.py @@ -4,6 +4,7 @@ import audioop import re +import time from pathlib import Path import websockets @@ -19,9 +20,13 @@ INACTIVITY_TIMEOUT_MS, TICK_DURATION_MS, TRANSCRIPT_WAIT_MS, + WAIT_TO_RESPOND_OTHER_MS, ms_to_ticks, ) from eva.user_simulator.cascade.decision_log import DecisionLog +from eva.user_simulator.cascade.decisions import ListenerDecisions, parse_yes_no +from eva.user_simulator.cascade.phrase_cache import PhraseCache +from eva.user_simulator.cascade.phrases import load_phrases from eva.user_simulator.cascade.scheduler import TickScheduler from eva.user_simulator.cascade.stt_livekit import LiveKitStreamingSTT from eva.user_simulator.cascade.tick_result import TickResult @@ -30,6 +35,7 @@ # Shared with the OpenAI Realtime provider so both simulators hang up on the same rules. from eva.user_simulator.openai_realtime import END_CALL_DESCRIPTION from eva.utils.logging import get_logger +from eva.utils.prompt_manager import PromptManager logger = get_logger(__name__) @@ -69,6 +75,120 @@ def extract_turn(message: object) -> tuple[str, bool]: return parse_turn_response(content), end_call +def extract_optional_line(message: object) -> str: + """Read a bare spoken line from a dedicated call, empty when the model declined. + + A bare line rather than a JSON field: demanding JSON on a caller call suppressed + the end_call tool call entirely (Plan 1). + """ + content = message if isinstance(message, str) else (getattr(message, "content", None) or "") + line = _FENCE.sub("", content).strip() + return "" if line.upper() == "NONE" else line + + +def interrupt_slip_ms(*, elapsed_s: float) -> int: + """How far past its intended moment a barge-in actually landed. + + Measured on the wall clock, not the tick counter. `run_tick` is only pumped by + the `_run` loop, so while `_play_interruption` awaits its generation no tick can + advance — a tick-delta slip is structurally always zero and the staleness check + built on it never engages. + """ + return max(0, int(elapsed_s * 1000)) + + +def summarize_goal(goal: dict) -> str: + """State what the caller wants and what would end the call, for the listener checks. + + Field meanings live in the prompt template rather than here, so the explanations stay + static across the many calls this decision makes rather than being rebuilt per call. + `edge_cases` and `information_required` are deliberately omitted: they are long and + describe how to answer questions, not whether the goal is finished. + """ + tree = goal.get("decision_tree", {}) or {} + sections = [ + ("GOAL", goal.get("high_level_user_goal")), + ("MUST HAVE", tree.get("must_have_criteria")), + ("NICE TO HAVE", tree.get("nice_to_have_criteria")), + ("HOW THEY EVALUATE OPTIONS", tree.get("negotiation_behavior")), + ("RESOLVED WHEN", tree.get("resolution_condition")), + ("FAILED WHEN", tree.get("failure_condition")), + ("ESCALATION", tree.get("escalation_behavior")), + ] + lines: list[str] = [] + for label, value in sections: + if not value: + continue + if isinstance(value, list): + lines.append(f"{label}:") + lines += [f"- {item}" for item in value] + else: + lines.append(f"{label}: {value}") + return "\n".join(lines) + + +def is_new_assistant_turn(*, ticks_silent_before: int) -> bool: + """Whether assistant audio arriving now starts a new turn or resumes the current one. + + A pause shorter than the turn-end threshold is a gap *inside* one turn — between + sentences, or a hole in the audio stream — not a new turn. Treating any single quiet + tick as a boundary re-armed the interruption cap mid-utterance, so one assistant turn + could collect several barge-ins. + """ + return ticks_silent_before >= ms_to_ticks(WAIT_TO_RESPOND_OTHER_MS) + + +def should_drop_interrupt(*, assistant_still_speaking: bool, same_assistant_turn: bool) -> bool: + """Whether a barge-in has gone stale and should be abandoned. + + Staleness is a fact about the assistant, not about how long generation took: a line is + still a real interruption whenever the assistant is mid-utterance, however many wall-clock + seconds elapsed. Both conditions are needed — "still speaking" alone is satisfied by a + *later* turn, which would land the line as a non-sequitur against speech it never heard. + """ + return not (assistant_still_speaking and same_assistant_turn) + + +async def candidate_is_relevant(llm, *, candidate: str, heard: str) -> bool: + """Whether a pre-generated interruption still fits what the assistant is saying. + + Fails closed: an unusable answer means we fall back to generating fresh + content rather than firing something that may have gone stale. + """ + prompt = PromptManager().get_prompt("user_simulator.cascade_relevance_gate", candidate=candidate, heard=heard) + try: + reply = await llm.decide(prompt) + except Exception as exc: + logger.warning(f"Relevance gate failed, discarding candidate: {exc}") + return False + return parse_yes_no(reply) + + +def play_backchannel(scheduler, cache, phrases: list[str]) -> str: + """Queue a cached continuer and return the phrase that was chosen. + + Queued as a backchannel, not an utterance: it must not consume the caller's + turn, or the caller waits for a reply the continuer never earns. + """ + phrase = cache.choose(phrases) + scheduler.enqueue_backchannel(cache.get(phrase)) + return phrase + + +class _DecisionClient: + """Adapts LiteLLMClient to the single-prompt interface the checks expect.""" + + def __init__(self, client: LiteLLMClient) -> None: + self._client = client + + async def decide(self, prompt: str) -> str: + """Ask one YES/NO question and return the raw reply.""" + message, _stats = await self._client.complete(messages=[{"role": "user", "content": prompt}]) + if isinstance(message, str): + return message + return getattr(message, "content", None) or "" + + TICK_DRIVEN_FRAMEWORKS = frozenset({"openai_realtime"}) """Frameworks whose clock the caller can own. Others keep real-time streaming.""" @@ -89,6 +209,10 @@ class CascadeUserSimulator(AbstractUserSimulator): _ticks_awaiting_transcript = 0 _ticks_assistant_silent = 0 + _ticks_since_assistant_started = 0 + _may_interrupt_this_turn = False + _assistant_turn_index = 0 + _last_checked_text = "" def __init__( self, @@ -123,7 +247,15 @@ def __init__( self._tts = CartesiaTTS(simulator_config.tts_params, language=language) self._llm = LiteLLMClient(model=simulator_config.llm) self._voice_id = self._tts.voice_for_persona(persona_config) + # The caller's out-of-turn vocabulary is language data, not code. + self._phrases = load_phrases(language) self._history: list[dict[str, str]] = [] + # Shared by the listener checks and the relevance gate so both cost one client. + self._decision_client = _DecisionClient(LiteLLMClient(model=simulator_config.decision_llm)) + self._phrase_cache: PhraseCache | None = None + self._decisions: ListenerDecisions | None = None + self._candidate_text = "" + self._candidate_audio = b"" self._decision_log = DecisionLog(self.output_dir / "user_simulator_decisions.jsonl") async def run_conversation(self) -> str: @@ -154,6 +286,7 @@ async def _run(self) -> None: await adapter.start() await self._stt.start() + await self._prepare_listener_behaviors() self.event_logger.log_connection_state("connected", {"server_url": self.server_url}) max_ticks = self.timeout * 1000 // TICK_DURATION_MS @@ -187,6 +320,8 @@ async def _run(self) -> None: ) self._on_conversation_end("provider_stalled") break + # Captured before the inactivity check, which clears it on a speech tick. + silent_before = self._ticks_assistant_silent if self._assistant_is_inactive(scheduler, result): logger.warning( f"tick {scheduler.tick}: assistant silent for " @@ -194,9 +329,31 @@ async def _run(self) -> None: ) self._on_conversation_end("inactivity_timeout") break - self._log_tick_state(scheduler, result) if result.has_assistant_speech: + if is_new_assistant_turn(ticks_silent_before=silent_before): + self._ticks_since_assistant_started = 0 + # One roll per assistant turn, so a turn carries at most one barge-in. + self._may_interrupt_this_turn = self._config.enable_interruptions + self._assistant_turn_index += 1 + # The transcript is consumed between turns, so it can shrink back + # toward the in-flight partial and collide with a value already + # judged. Clearing here keeps "unchanged" meaning "unchanged + # within this turn", which is the only span it is asked about. + self._last_checked_text = "" + self._decision_log.log( + "assistant_turn_start", + tick=scheduler.tick, + turn_index=self._assistant_turn_index, + ticks_silent_before=silent_before, + armed_interrupt=self._may_interrupt_this_turn, + ) + self._ticks_since_assistant_started += 1 + is_check = scheduler.is_check_tick() + self._log_tick_state(scheduler, result, is_check_tick=is_check) + if is_check and await self._run_checks(scheduler): + break continue + self._log_tick_state(scheduler, result, is_check_tick=False) if scheduler.caller_is_speaking or not scheduler.may_take_turn(): continue heard, waiting = self._collect_heard_text(scheduler) @@ -212,6 +369,200 @@ async def _run(self) -> None: await adapter.stop() self.event_logger.log_connection_state("session_ended", {"reason": self._end_reason}) + async def _prepare_listener_behaviors(self) -> None: + """Pre-render the fixed vocabularies and build the checks, when any is enabled. + + Rendering happens once at connect time rather than on demand: a cached phrase + is the only way a 300ms reaction lands on the tick its check chose. + """ + vocabulary: list[str] = [] + if self._config.enable_backchannel: + vocabulary += self._phrases.backchannels + if self._config.enable_interruptions: + vocabulary += self._phrases.barge_in_openers + if not vocabulary: + return + + self._phrase_cache = PhraseCache(self._tts, voice_id=self._voice_id) + await self._phrase_cache.prerender(vocabulary) + prompts = PromptManager() + self._decisions = ListenerDecisions( + self._decision_client, + interrupt_prompt=prompts.get_template("user_simulator.interruption_decision"), + backchannel_prompt=prompts.get_template("user_simulator.backchannel_decision"), + user_goal=summarize_goal(self.goal), + ) + + async def _run_checks(self, scheduler: TickScheduler) -> bool: + """Run the listener-reaction checks and act on the verdict. True means hang up. + + Gated on there being something new to judge. The checks fire on a timer while + the assistant speaks — ~101 times in one measured call, two concurrent LLM calls + each — but their only input is the transcript, so a tick where the transcript has + not moved re-asks a question already answered. Skipping those is free: identical + input, identical verdict. + """ + if self._decisions is None or self._phrase_cache is None: + return False + + history = self._stt.buffer.current_text() + # Nothing transcribed yet is not a question worth asking, and a transcript that + # has not moved re-asks one already answered. + if not history or history == self._last_checked_text: + return False + self._last_checked_text = history + + # Backchannelling stays available after a barge-in: a listener who cuts in and + # later hums along is ordinary, and the one-barge-in-per-turn cap is about not + # talking over the assistant repeatedly, not about going quiet afterwards. + # `allow_interrupt` alone already skips the interrupt call without reaching the + # model once the cap is spent. + verdict = await self._decisions.evaluate( + history, + allow_interrupt=self._may_interrupt_this_turn, + allow_backchannel=self._config.enable_backchannel, + ) + self._decision_log.log( + "listener_check", + tick=scheduler.tick, + allow_interrupt=self._may_interrupt_this_turn, + allow_backchannel=self._config.enable_backchannel, + heard_chars=len(history), + heard=history, + interrupt_ran=verdict.interrupt_trace.ran, + interrupt_raw=verdict.interrupt_trace.raw, + interrupt_latency_ms=verdict.interrupt_trace.latency_ms, + interrupt_error=verdict.interrupt_trace.error, + backchannel_ran=verdict.backchannel_trace.ran, + backchannel_raw=verdict.backchannel_trace.raw, + backchannel_error=verdict.backchannel_trace.error, + should_interrupt=verdict.should_interrupt, + should_backchannel=verdict.should_backchannel, + ) + if verdict.should_interrupt: + self._may_interrupt_this_turn = False + return await self._play_interruption(scheduler) + if verdict.should_backchannel: + phrase = play_backchannel(scheduler, self._phrase_cache, self._phrases.backchannels) + # Recorded too, or the saved clean track diverges from what went on the wire. + self._record_audio("user_clean", self._phrase_cache.get(phrase)) + self.event_logger.log_event("backchannel", {"text": phrase, "tick_index": scheduler.tick}) + return False + + async def _play_interruption(self, scheduler: TickScheduler) -> bool: + """Decide what to say, then voice the opener and the content together. + + Returns True when the caller decided to hang up. + + The content is generated *before* anything reaches the wire. Emitting the opener + first hid its ~1s latency, but committed the caller to barging in before knowing + whether it had anything to say: a hang-up or a dropped-as-stale line then left an + orphaned "Actually—" hanging with nothing behind it, which is worse than either a + late line or silence. Generating first means "say nothing" is actually available. + """ + if self._phrase_cache is None: + return False + intended_tick = scheduler.tick + intended_turn = self._assistant_turn_index + started_at = time.monotonic() + opener = self._phrase_cache.choose(self._phrases.barge_in_openers) + opener_audio = self._phrase_cache.get(opener) + + if self._config.speculative_generation and self._candidate_audio: + candidate, audio = self._candidate_text, self._candidate_audio + self._candidate_text, self._candidate_audio = "", b"" + relevant = await candidate_is_relevant( + self._decision_client, candidate=candidate, heard=self._stt.buffer.current_text() + ) + self._decision_log.log("relevance_gate", tick=intended_tick, candidate=candidate, relevant=relevant) + if relevant: + # Tell the adapter the next tick that reaches the wire cuts the + # assistant off, so a tick-driven transport can truncate the audio + # the caller never heard. Ignored on the real-time path. + scheduler.arm_barge_in() + scheduler.enqueue_utterance(opener_audio) + self._record_audio("user_clean", opener_audio) + scheduler.enqueue_utterance(audio) + self._record_audio("user_clean", audio) + self._history.append({"role": "user", "content": candidate}) + self._on_user_speaks(candidate) + self.event_logger.log_event( + "interruption", + { + "text": candidate, + "opener": opener, + "intended_tick": intended_tick, + "actual_tick": scheduler.tick, + "slip_ms": interrupt_slip_ms(elapsed_s=time.monotonic() - started_at), + "speculative": True, + "dropped": False, + }, + ) + self._decision_log.log( + "interruption", tick=intended_tick, outcome="spoken", speculative=True, text=candidate + ) + return False + self.event_logger.log_event("interruption_candidate_rejected", {"text": candidate}) + self._decision_log.log("interruption", tick=intended_tick, outcome="candidate_rejected", text=candidate) + + # Consumed, not peeked: leaving it in the buffer would re-append the same + # assistant prefix at the next ordinary turn and duplicate it in the history. + heard = self._stt.buffer.heard_text() + self._stt.buffer.take_committed() + self._stt.buffer.in_flight = "" + if heard: + self._history.append({"role": "assistant", "content": heard}) + self._on_assistant_speaks(heard) + message, _stats = await self._llm.complete(messages=self._messages(), tools=[END_CALL_TOOL]) + utterance, end_call = extract_turn(message) + + slip = interrupt_slip_ms(elapsed_s=time.monotonic() - started_at) + # A hang-up is never stale: the caller has decided the call is over, and + # dropping it here is what left conversations looping until the timeout. + dropped = not end_call and should_drop_interrupt( + assistant_still_speaking=scheduler.assistant_is_speaking, + same_assistant_turn=self._assistant_turn_index == intended_turn, + ) + self.event_logger.log_event( + "interruption", + { + "text": utterance, + "opener": opener, + "intended_tick": intended_tick, + "actual_tick": scheduler.tick, + "slip_ms": slip, + "dropped": dropped, + "end_call": end_call, + }, + ) + self._decision_log.log( + "interruption", + tick=intended_tick, + outcome="dropped" if dropped else ("end_call" if end_call else "spoken"), + text=utterance, + slip_ms=slip, + assistant_still_speaking=scheduler.assistant_is_speaking, + intended_turn=intended_turn, + actual_turn=self._assistant_turn_index, + ) + # Nothing has reached the wire yet, so a stale line or a hang-up costs no audio. + if dropped: + return False + + if utterance: + scheduler.arm_barge_in() + scheduler.enqueue_utterance(opener_audio) + self._record_audio("user_clean", opener_audio) + self._history.append({"role": "user", "content": utterance}) + self._on_user_speaks(utterance) + async for chunk in self._tts.stream(utterance, voice_id=self._voice_id): + self._record_audio("user_clean", chunk) + scheduler.enqueue_utterance(chunk) + + if end_call: + self._on_conversation_end("goodbye") + return end_call + def _assistant_is_inactive(self, scheduler: TickScheduler, result: TickResult) -> bool: """Whether the assistant has produced no audio for INACTIVITY_TIMEOUT_MS *contiguously*. @@ -255,9 +606,20 @@ def _log_audio_boundaries( elif not result.has_assistant_speech and assistant_was_speaking: self.event_logger.log_audio_end("assistant", seconds) - def _log_tick_state(self, scheduler: TickScheduler, result: TickResult) -> None: - """Trace speech state and audio levels for each simulator tick.""" + def _log_tick_state(self, scheduler: TickScheduler, result: TickResult, *, is_check_tick: bool) -> None: + """Trace one tick's speech state, so a check that never ran can be traced to its gate. + + `rms` is why this exists: `has_assistant_speech` is true for any non-zero bytes, + digital silence included, so a transport that pads with silence reads as continuous + speech. Recording both lets that be measured rather than inferred. + + `transcript_moved` serves the same purpose for the other gate. A check tick only + reaches the judges when the transcript has changed since the last one judged, so + without this a check tick with no following `listener_check` is ambiguous between + "nothing new was heard" and "the check ran and something went wrong". + """ raw = result.assistant_audio[: result.assistant_audio_raw_bytes] + history = self._stt.buffer.current_text() self._decision_log.log( "tick", tick=scheduler.tick, @@ -267,6 +629,11 @@ def _log_tick_state(self, scheduler: TickScheduler, result: TickResult) -> None: caller_is_speaking=scheduler.caller_is_speaking, caller_spoke_this_tick=scheduler.caller_spoke_this_tick, ticks_assistant_silent=self._ticks_assistant_silent, + ticks_since_assistant_started=self._ticks_since_assistant_started, + may_interrupt_this_turn=self._may_interrupt_this_turn, + is_check_tick=is_check_tick, + transcript_moved=bool(history) and history != self._last_checked_text, + has_candidate=bool(self._candidate_audio), ) def _collect_heard_text(self, scheduler: TickScheduler) -> tuple[str, bool]: @@ -321,8 +688,51 @@ async def _take_turn(self, scheduler: TickScheduler, heard: str) -> bool: self._on_conversation_end("goodbye") return True + # After the audio is queued: the caller is now speaking, so this generation + # runs behind its own outgoing audio and costs no conversational latency. + if utterance: + await self._prerender_candidate(utterance) return False + async def _prerender_candidate(self, utterance: str) -> None: + """Pre-generate and pre-render the line the caller would barge in with. + + Done on the caller's own turn, where the latency is already hidden behind + its outgoing audio, so a later barge-in can fire without waiting on + generation. The relevance gate is what keeps this from degrading into a + scripted interruption that lands as a non-sequitur. + """ + self._candidate_text, self._candidate_audio = "", b"" + # A candidate only ever gets spoken by a barge-in, so rendering one with + # interruptions off buys an LLM call and a synthesis per turn for nothing. + if not self._config.speculative_generation or not self._config.enable_interruptions: + return + started = time.monotonic() + prompt = PromptManager().get_prompt("user_simulator.cascade_next_interruption", utterance=utterance) + try: + message, _stats = await self._llm.complete( + messages=[*self._messages(), {"role": "user", "content": prompt}] + ) + except Exception as exc: + logger.warning(f"Speculative interruption generation failed: {exc}") + self._decision_log.log("candidate_generation", ok=False, error=str(exc)) + return + candidate = extract_optional_line(message) + if not candidate: + raw = message if isinstance(message, str) else (getattr(message, "content", None) or "") + self._decision_log.log("candidate_generation", ok=False, declined=True, after=utterance, raw=raw[:400]) + return + self._candidate_text = candidate + self._candidate_audio = await self._tts.synthesize(candidate, voice_id=self._voice_id) + self._decision_log.log( + "candidate_generation", + ok=True, + after=utterance, + candidate=candidate, + audio_bytes=len(self._candidate_audio), + latency_ms=int((time.monotonic() - started) * 1000), + ) + def _messages(self) -> list[dict[str, str]]: """Build the message list: the shared per-domain caller prompt plus flipped history. diff --git a/src/eva/user_simulator/cascade/stt.py b/src/eva/user_simulator/cascade/stt.py index 677df3951..750136fe6 100644 --- a/src/eva/user_simulator/cascade/stt.py +++ b/src/eva/user_simulator/cascade/stt.py @@ -19,6 +19,17 @@ def commit(self, text: str) -> None: self.committed = f"{self.committed} {text}".strip() if self.committed else text self.in_flight = "" + def current_text(self) -> str: + """Return everything heard so far, marking the partial as still in progress. + + The marker is load-bearing: the backchannel prompt's few-shot examples all + end in it, and it is what tells the check to judge only the complete + sentences rather than the truncated trailing word. + """ + if not self.in_flight: + return self.committed + return f"{self.committed} {self.in_flight} [CURRENTLY SPEAKING, INCOMPLETE]".strip() + def heard_text(self) -> str: """Return everything heard so far without the in-progress marker, for transcript use.""" return f"{self.committed} {self.in_flight}".strip() diff --git a/src/eva/user_simulator/cascade/tick_result.py b/src/eva/user_simulator/cascade/tick_result.py index 265efb48f..f2343459d 100644 --- a/src/eva/user_simulator/cascade/tick_result.py +++ b/src/eva/user_simulator/cascade/tick_result.py @@ -5,7 +5,7 @@ from dataclasses import dataclass from typing import NamedTuple -from eva.user_simulator.cascade.constants import BYTES_PER_TICK, SILENCE_BYTE +from eva.user_simulator.cascade.constants import BYTES_PER_TICK, SILENCE_BYTE, TICK_DURATION_MS @dataclass(frozen=True) @@ -24,6 +24,12 @@ class TickResult: wall_clock_ms: int """Unix ms at the tick's I/O boundary. For latency metrics only, never for ordering.""" + skip_item_id: str | None = None + """Provider item whose remaining audio must be discarded after a barge-in.""" + + interruption_audio_start_ms: int | None = None + """Played position where the caller cut in, in simulated ms.""" + provider_stalled: bool = False """The assistant has produced nothing for so long that the run is not usable. @@ -39,6 +45,16 @@ def has_assistant_speech(self) -> bool: return self.assistant_audio_raw_bytes > 0 +def played_audio_ms(*, ticks_released: int) -> int: + """Assistant audio actually released into the conversation, in simulated ms. + + This is deliberately not "bytes received": a realtime provider generates + faster than real time, so the provider's idea of playback position runs ahead + of what the caller has heard. Truncation must use this number. + """ + return ticks_released * TICK_DURATION_MS + + class TickAudioSplit(NamedTuple): """Result of splitting audio at a tick boundary.""" diff --git a/tests/unit/assistant/test_openai_realtime_pacing.py b/tests/unit/assistant/test_openai_realtime_pacing.py index e1de6a103..1c29b0467 100644 --- a/tests/unit/assistant/test_openai_realtime_pacing.py +++ b/tests/unit/assistant/test_openai_realtime_pacing.py @@ -106,3 +106,38 @@ class _Conversation: item = _Item() self.conversation = _Conversation() + + +@pytest.mark.asyncio +async def test_truncate_targets_the_item_currently_producing_audio(tmp_path): + server = _server(paced=False, tmp_path=tmp_path) + server._assistant_state.active_item_id = "item_42" + conn = FakeConn() + + await server._truncate_response(conn, 600) + + assert conn.truncated == [{"item_id": "item_42", "content_index": 0, "audio_end_ms": 600}] + + +@pytest.mark.asyncio +async def test_truncate_is_a_no_op_when_no_item_is_active(tmp_path): + server = _server(paced=False, tmp_path=tmp_path) + conn = FakeConn() + + await server._truncate_response(conn, 600) + + assert conn.truncated == [] + + +@pytest.mark.asyncio +async def test_audio_delta_records_the_active_item(tmp_path): + import base64 + from types import SimpleNamespace + + server = _server(paced=False, tmp_path=tmp_path) + queue: asyncio.Queue[bytes] = asyncio.Queue() + event = SimpleNamespace(delta=base64.b64encode(b"\x00" * 480).decode(), item_id="item_7") + + await server._on_audio_delta(event, queue) + + assert server._assistant_state.active_item_id == "item_7" diff --git a/tests/unit/models/test_config_models.py b/tests/unit/models/test_config_models.py index f9ec45964..b1c820461 100644 --- a/tests/unit/models/test_config_models.py +++ b/tests/unit/models/test_config_models.py @@ -1270,3 +1270,30 @@ def test_user_simulator_union_discriminates_cascade(): parsed = TypeAdapter(UserSimulatorConfig).validate_python({"provider": "cascade"}) assert isinstance(parsed, CascadeSimulatorConfig) + + +def test_cascade_behaviors_default_off(): + from eva.models.config import CascadeSimulatorConfig + + config = CascadeSimulatorConfig() + + assert config.enable_backchannel is False + assert config.enable_interruptions is False + assert config.speculative_generation is False + + +def test_cascade_behaviors_can_be_enabled_independently(): + from eva.models.config import CascadeSimulatorConfig + + config = CascadeSimulatorConfig(enable_backchannel=True) + + assert config.enable_backchannel is True + assert config.enable_interruptions is False + + +def test_cascade_decision_llm_defaults_to_the_caller_llm(): + from eva.models.config import CascadeSimulatorConfig + + config = CascadeSimulatorConfig() + + assert config.decision_llm == config.llm diff --git a/tests/unit/user_simulator/cascade/conftest.py b/tests/unit/user_simulator/cascade/conftest.py new file mode 100644 index 000000000..038040e32 --- /dev/null +++ b/tests/unit/user_simulator/cascade/conftest.py @@ -0,0 +1,17 @@ +import pytest + +from eva.user_simulator.cascade.phrase_cache import PhraseCache + + +@pytest.fixture(autouse=True) +def _isolate_phrase_cache(): + """Clear the process-global phrase audio between tests. + + The cache is shared across conversations on purpose — that is what stops every + record re-rendering the same "mm-hmm" — but tests are conversations too, so + without this one test's renders satisfy the next one's prerender and call + counts come out wrong. + """ + PhraseCache.clear() + yield + PhraseCache.clear() diff --git a/tests/unit/user_simulator/cascade/test_adapter_base.py b/tests/unit/user_simulator/cascade/test_adapter_base.py index 1f6cfd83d..bac379088 100644 --- a/tests/unit/user_simulator/cascade/test_adapter_base.py +++ b/tests/unit/user_simulator/cascade/test_adapter_base.py @@ -15,7 +15,9 @@ class StubAdapter(Adapter): async def start(self) -> None: pass - async def run_tick(self, tick_number: int, outgoing_audio: bytes | None) -> TickResult: + async def run_tick( + self, tick_number: int, outgoing_audio: bytes | None, *, barge_in: bool = False + ) -> TickResult: return TickResult( tick_number=tick_number, assistant_audio=b"\x00" * 4, diff --git a/tests/unit/user_simulator/cascade/test_constants.py b/tests/unit/user_simulator/cascade/test_constants.py index cec8808b2..1aeacec03 100644 --- a/tests/unit/user_simulator/cascade/test_constants.py +++ b/tests/unit/user_simulator/cascade/test_constants.py @@ -25,3 +25,19 @@ def test_threshold_constants_are_exact_multiples_of_tick_duration(): def test_ms_to_ticks_converts_and_floors_sub_tick_remainder(): assert ms_to_ticks(WAIT_TO_RESPOND_OTHER_MS) == 5 assert ms_to_ticks(150) == 0 + + +def test_listener_check_interval_is_two_seconds_in_ticks(): + from eva.user_simulator.cascade.constants import LISTENER_CHECK_INTERVAL_MS, ms_to_ticks + + assert LISTENER_CHECK_INTERVAL_MS == 2000 + assert ms_to_ticks(LISTENER_CHECK_INTERVAL_MS) == 10 + + +def test_the_vocabularies_are_no_longer_constants(): + # They moved to configs/caller_phrases.yaml because they are language data, not + # timing. Timing constants staying here is the whole distinction. + from eva.user_simulator.cascade import constants + + assert not hasattr(constants, "BACKCHANNEL_PHRASES") + assert not hasattr(constants, "BARGE_IN_OPENERS") diff --git a/tests/unit/user_simulator/cascade/test_decision_log.py b/tests/unit/user_simulator/cascade/test_decision_log.py index 8bcb6bed4..7ca3311b6 100644 --- a/tests/unit/user_simulator/cascade/test_decision_log.py +++ b/tests/unit/user_simulator/cascade/test_decision_log.py @@ -1,4 +1,4 @@ -"""Tests for the caller's tick trace.""" +"""Tests for the caller's out-of-turn decision trace.""" import json @@ -8,12 +8,12 @@ def test_rows_are_written_one_json_object_per_line(tmp_path): log = DecisionLog(tmp_path / "trace.jsonl") log.log("tick", tick=1, has_assistant_speech=True) - log.log("session_ended", tick=10, reason="goodbye") + log.log("listener_check", tick=10, should_interrupt=False) log.save() rows = [json.loads(line) for line in (tmp_path / "trace.jsonl").read_text().splitlines()] - assert [r["kind"] for r in rows] == ["tick", "session_ended"] + assert [r["kind"] for r in rows] == ["tick", "listener_check"] assert rows[0]["has_assistant_speech"] is True @@ -43,6 +43,36 @@ def test_summary_counts_rows_by_kind(tmp_path): log = DecisionLog(tmp_path / "trace.jsonl") for _ in range(3): log.log("tick") - log.log("session_ended") + log.log("listener_check") - assert log.summary() == {"tick": 3, "session_ended": 1} + assert log.summary() == {"tick": 3, "listener_check": 1} + + +async def test_a_declined_check_is_still_recorded_with_its_raw_reply(): + # The whole point: a NO must be distinguishable from a check that never ran. + from eva.user_simulator.cascade.decisions import ListenerDecisions + from tests.unit.user_simulator.cascade.test_decisions import FakeLLM + + decisions = ListenerDecisions( + FakeLLM(["NO", "NO"]), interrupt_prompt="{conversation_history}", backchannel_prompt="{conversation_history}" + ) + + verdict = await decisions.evaluate("agent talking", allow_interrupt=True, allow_backchannel=True) + + assert verdict.should_interrupt is False + assert verdict.interrupt_trace.ran is True + assert verdict.interrupt_trace.raw == "NO" + + +async def test_a_check_that_was_not_allowed_reports_that_it_never_ran(): + from eva.user_simulator.cascade.decisions import ListenerDecisions + from tests.unit.user_simulator.cascade.test_decisions import FakeLLM + + decisions = ListenerDecisions( + FakeLLM([]), interrupt_prompt="{conversation_history}", backchannel_prompt="{conversation_history}" + ) + + verdict = await decisions.evaluate("agent talking", allow_interrupt=False, allow_backchannel=False) + + assert verdict.interrupt_trace.ran is False + assert verdict.interrupt_trace.raw == "" diff --git a/tests/unit/user_simulator/cascade/test_decisions.py b/tests/unit/user_simulator/cascade/test_decisions.py new file mode 100644 index 000000000..7145c06a1 --- /dev/null +++ b/tests/unit/user_simulator/cascade/test_decisions.py @@ -0,0 +1,155 @@ +from eva.user_simulator.cascade.decisions import ListenerDecisions, parse_yes_no + + +def test_parse_yes_no_accepts_plain_yes(): + assert parse_yes_no("YES") is True + + +def test_parse_yes_no_is_case_and_whitespace_insensitive(): + assert parse_yes_no(" yes\n") is True + + +def test_parse_yes_no_treats_anything_else_as_no(): + assert parse_yes_no("NO") is False + assert parse_yes_no("maybe") is False + assert parse_yes_no("") is False + + +class FakeLLM: + """Returns scripted replies, or raises when configured to.""" + + def __init__(self, replies: list[str] | None = None, error: Exception | None = None) -> None: + self.replies = replies or [] + self.error = error + self.calls = 0 + + async def decide(self, prompt: str) -> str: + self.calls += 1 + if self.error is not None: + raise self.error + return self.replies.pop(0) if self.replies else "NO" + + +async def test_both_checks_run_and_interrupt_wins_ties(): + llm = FakeLLM(["YES", "YES"]) + decisions = ListenerDecisions( + llm, interrupt_prompt="i {conversation_history}", backchannel_prompt="b {conversation_history}" + ) + + verdict = await decisions.evaluate("AGENT: hello", allow_interrupt=True, allow_backchannel=True) + + assert verdict.should_interrupt is True + assert verdict.should_backchannel is False + assert llm.calls == 2 + + +async def test_backchannel_alone_when_interrupt_declines(): + llm = FakeLLM(["NO", "YES"]) + decisions = ListenerDecisions( + llm, interrupt_prompt="i {conversation_history}", backchannel_prompt="b {conversation_history}" + ) + + verdict = await decisions.evaluate("AGENT: hello", allow_interrupt=True, allow_backchannel=True) + + assert verdict.should_interrupt is False + assert verdict.should_backchannel is True + + +async def test_disabled_behaviors_are_not_called_at_all(): + llm = FakeLLM(["YES"]) + decisions = ListenerDecisions( + llm, interrupt_prompt="i {conversation_history}", backchannel_prompt="b {conversation_history}" + ) + + verdict = await decisions.evaluate("AGENT: hello", allow_interrupt=False, allow_backchannel=True) + + assert verdict.should_interrupt is False + assert llm.calls == 1 + + +async def test_a_failing_check_fails_closed(): + llm = FakeLLM(error=RuntimeError("provider down")) + decisions = ListenerDecisions( + llm, interrupt_prompt="i {conversation_history}", backchannel_prompt="b {conversation_history}" + ) + + verdict = await decisions.evaluate("AGENT: hello", allow_interrupt=True, allow_backchannel=True) + + assert verdict.should_interrupt is False + assert verdict.should_backchannel is False + + +async def test_the_goal_is_substituted_into_the_interrupt_prompt(): + llm = FakeLLM(["NO", "NO"]) + seen: list[str] = [] + + class _Recorder(FakeLLM): + async def decide(self, prompt: str) -> str: + seen.append(prompt) + return await super().decide(prompt) + + decisions = ListenerDecisions( + _Recorder(["NO", "NO"]), + interrupt_prompt="goal={user_goal} history={conversation_history}", + backchannel_prompt="b {conversation_history}", + user_goal="Unlock my account.", + ) + + await decisions.evaluate("AGENT: hello", allow_interrupt=True, allow_backchannel=True) + + assert "goal=Unlock my account." in seen[0] + assert llm.calls == 0 + + +async def test_a_backchannel_prompt_without_a_goal_slot_still_works(): + # str.format ignores unused keyword arguments, so one signature serves both prompts. + decisions = ListenerDecisions( + FakeLLM(["NO", "YES"]), + interrupt_prompt="i {conversation_history}", + backchannel_prompt="b {conversation_history}", + user_goal="Unlock my account.", + ) + + verdict = await decisions.evaluate("AGENT: hello", allow_interrupt=True, allow_backchannel=True) + + assert verdict.should_backchannel is True + + +def test_goal_summary_is_compact_and_names_what_is_left(): + from eva.user_simulator.cascade.simulator import summarize_goal + + summary = summarize_goal( + { + "high_level_user_goal": "Unlock my AD account.", + "decision_tree": { + "must_have_criteria": ["account unlocked"], + "nice_to_have_criteria": ["a case number"], + "negotiation_behavior": "take the fastest fix offered", + "resolution_condition": "user can sign in", + "failure_condition": "agent cannot unlock it", + "escalation_behavior": "do not ask for a live agent", + "edge_cases": ["a very long irrelevant edge case " * 40], + }, + } + ) + + for expected in ( + "Unlock my AD account.", + "account unlocked", + "a case number", + "take the fastest fix offered", + "user can sign in", + "agent cannot unlock it", + "do not ask for a live agent", + ): + assert expected in summary, expected + # edge_cases is long and describes how to answer questions, not whether the goal is done. + assert "irrelevant" not in summary + + +def test_goal_summary_omits_absent_fields_without_blank_labels(): + from eva.user_simulator.cascade.simulator import summarize_goal + + summary = summarize_goal({"high_level_user_goal": "Unlock my account.", "decision_tree": {}}) + + assert summary == "GOAL: Unlock my account." diff --git a/tests/unit/user_simulator/cascade/test_phrase_cache.py b/tests/unit/user_simulator/cascade/test_phrase_cache.py new file mode 100644 index 000000000..b951976ec --- /dev/null +++ b/tests/unit/user_simulator/cascade/test_phrase_cache.py @@ -0,0 +1,101 @@ +import pytest + +from eva.user_simulator.cascade.phrase_cache import PhraseCache + + +class FakeTTS: + """Counts synthesis calls and returns deterministic audio per phrase.""" + + def __init__(self) -> None: + self.calls: list[str] = [] + + async def synthesize(self, text: str, *, voice_id: str) -> bytes: + self.calls.append(text) + return text.encode() + + +async def test_prerender_synthesizes_every_phrase_once(): + tts = FakeTTS() + cache = PhraseCache(tts, voice_id="voice-f") + + await cache.prerender(["uh-huh", "mm-hmm"]) + + assert sorted(tts.calls) == ["mm-hmm", "uh-huh"] + + +async def test_cached_audio_is_returned_without_further_synthesis(): + tts = FakeTTS() + cache = PhraseCache(tts, voice_id="voice-f") + await cache.prerender(["uh-huh"]) + + audio = cache.get("uh-huh") + + assert audio == b"uh-huh" + assert len(tts.calls) == 1 + + +async def test_choose_returns_a_phrase_from_the_cache_deterministically(): + tts = FakeTTS() + cache = PhraseCache(tts, voice_id="voice-f", seed=7) + await cache.prerender(["uh-huh", "mm-hmm"]) + + first = cache.choose(["uh-huh", "mm-hmm"]) + replay = PhraseCache(FakeTTS(), voice_id="voice-f", seed=7) + await replay.prerender(["uh-huh", "mm-hmm"]) + + assert first == replay.choose(["uh-huh", "mm-hmm"]) + + +async def test_requesting_an_unrendered_phrase_raises(): + cache = PhraseCache(FakeTTS(), voice_id="voice-f") + + with pytest.raises(KeyError, match="not pre-rendered"): + cache.get("never-rendered") + + +async def test_a_second_conversation_on_the_same_voice_renders_nothing(): + # This is the point of the global cache: a run of N records against one voice + # renders each phrase once, not N times. + tts = FakeTTS() + first = PhraseCache(tts, voice_id="voice-f") + await first.prerender(["uh-huh", "mm-hmm"]) + + second = PhraseCache(tts, voice_id="voice-f") + await second.prerender(["uh-huh", "mm-hmm"]) + + assert len(tts.calls) == 2 + assert second.get("uh-huh") == b"uh-huh" + + +async def test_only_the_phrases_a_voice_is_missing_are_rendered(): + tts = FakeTTS() + await PhraseCache(tts, voice_id="voice-f").prerender(["uh-huh"]) + tts.calls.clear() + + await PhraseCache(tts, voice_id="voice-f").prerender(["uh-huh", "mm-hmm"]) + + assert tts.calls == ["mm-hmm"] + + +async def test_each_voice_keeps_its_own_audio(): + # Two genders means two voices; one must never be served the other's audio. + tts = FakeTTS() + female = PhraseCache(tts, voice_id="voice-f") + male = PhraseCache(tts, voice_id="voice-m") + await female.prerender(["uh-huh"]) + await male.prerender(["uh-huh"]) + + assert len(tts.calls) == 2 + assert female.get("uh-huh") == b"uh-huh" + assert male.get("uh-huh") == b"uh-huh" + + +async def test_concurrent_conversations_do_not_double_render(): + import asyncio + + tts = FakeTTS() + caches = [PhraseCache(tts, voice_id="voice-f") for _ in range(4)] + + await asyncio.gather(*(c.prerender(["uh-huh", "mm-hmm"]) for c in caches)) + + assert sorted(tts.calls) == ["mm-hmm", "uh-huh"] diff --git a/tests/unit/user_simulator/cascade/test_phrases.py b/tests/unit/user_simulator/cascade/test_phrases.py new file mode 100644 index 000000000..60ca00f36 --- /dev/null +++ b/tests/unit/user_simulator/cascade/test_phrases.py @@ -0,0 +1,88 @@ +import pytest +import yaml + +from eva.user_simulator.cascade import phrases as phrases_module +from eva.user_simulator.cascade.phrases import ( + PHRASES_PATH, + CallerPhrases, + candidate_languages, + load_phrases, +) + + +@pytest.fixture(autouse=True) +def _clear_phrase_cache(): + phrases_module._cache.clear() + yield + phrases_module._cache.clear() + + +@pytest.fixture +def phrase_file(tmp_path, monkeypatch): + """Point the loader at a temporary phrase file.""" + + def write(data): + path = tmp_path / "caller_phrases.yaml" + path.write_text(yaml.safe_dump(data, allow_unicode=True), encoding="utf-8") + monkeypatch.setattr(phrases_module, "PHRASES_PATH", path) + return path + + return write + + +def test_the_shipped_file_has_english(): + # English is the fallback for every language, so it is the one entry that must exist. + data = yaml.safe_load(PHRASES_PATH.read_text(encoding="utf-8")) + + assert data["en"]["backchannels"] + assert data["en"]["barge_in_openers"] + + +def test_a_language_with_its_own_entry_uses_it(phrase_file): + phrase_file( + { + "en": {"backchannels": ["uh-huh"], "barge_in_openers": ["Wait—"]}, + "fr": {"backchannels": ["hmm", "ouais"], "barge_in_openers": ["Attendez—"]}, + } + ) + + assert load_phrases("fr") == CallerPhrases(backchannels=["hmm", "ouais"], barge_in_openers=["Attendez—"]) + + +def test_a_regional_variant_falls_back_to_its_base_language(phrase_file): + # fr-CA shares its continuers with fr; falling all the way to English would be worse. + phrase_file( + { + "en": {"backchannels": ["uh-huh"], "barge_in_openers": ["Wait—"]}, + "fr": {"backchannels": ["hmm"], "barge_in_openers": ["Attendez—"]}, + } + ) + + assert load_phrases("fr-CA").backchannels == ["hmm"] + + +def test_an_unknown_language_falls_back_to_english_rather_than_failing(phrase_file): + # A missing translation should degrade the behavior's realism, not abort the run. + phrase_file({"en": {"backchannels": ["uh-huh"], "barge_in_openers": ["Wait—"]}}) + + assert load_phrases("ja").backchannels == ["uh-huh"] + + +def test_a_missing_file_yields_an_empty_vocabulary(phrase_file, tmp_path, monkeypatch): + monkeypatch.setattr(phrases_module, "PHRASES_PATH", tmp_path / "absent.yaml") + + result = load_phrases("en") + + assert result.vocabulary == [] + + +def test_candidate_order_is_most_specific_first(): + assert candidate_languages("fr-CA") == ["fr-CA", "fr", "en"] + assert candidate_languages("fr") == ["fr", "en"] + assert candidate_languages("en") == ["en"] + + +def test_vocabulary_is_everything_that_needs_rendering(phrase_file): + phrase_file({"en": {"backchannels": ["uh-huh", "mm-hmm"], "barge_in_openers": ["Wait—"]}}) + + assert load_phrases("en").vocabulary == ["uh-huh", "mm-hmm", "Wait—"] diff --git a/tests/unit/user_simulator/cascade/test_prompt.py b/tests/unit/user_simulator/cascade/test_prompt.py index e3c0a91c9..f3ad979bb 100644 --- a/tests/unit/user_simulator/cascade/test_prompt.py +++ b/tests/unit/user_simulator/cascade/test_prompt.py @@ -1,3 +1,6 @@ +from eva.utils.prompt_manager import PromptManager + + def test_cascade_reuses_the_shared_end_call_description(): # A cascade-specific copy would drift from the other providers' hang-up rules. from eva.user_simulator.cascade.simulator import END_CALL_DESCRIPTION as cascade_description @@ -9,6 +12,8 @@ def test_cascade_reuses_the_shared_end_call_description(): def test_the_turn_call_carries_no_cascade_specific_contract(): # The per-domain user_simulator prompt already carries persona, goal and end_call rules; # layering a cascade-only contract on top is what suppressed the end_call tool call. + # Out-of-turn behavior prompts exist, but they are only ever used in their own + # standalone calls — never appended to the system prompt of the turn call. from eva.user_simulator.cascade.simulator import CascadeUserSimulator sim = object.__new__(CascadeUserSimulator) @@ -16,3 +21,33 @@ def test_the_turn_call_carries_no_cascade_specific_contract(): sim._history = [] assert sim._messages()[0]["content"] == "SYSTEM PROMPT" + + +def test_interruption_decision_prompt_has_a_history_slot_and_binary_contract(): + prompt = PromptManager().get_prompt( + "user_simulator.interruption_decision", conversation_history="AGENT: hello", user_goal="Unlock my account." + ) + + assert "AGENT: hello" in prompt + assert "YES" in prompt + assert "NO" in prompt + + +def test_backchannel_decision_prompt_has_a_history_slot_and_frequency_guidance(): + prompt = PromptManager().get_prompt("user_simulator.backchannel_decision", conversation_history="AGENT: hello") + + assert "AGENT: hello" in prompt + assert "CURRENTLY SPEAKING, INCOMPLETE" in prompt + assert "When in doubt, say NO" in prompt + + +def test_interruption_decision_prompt_carries_the_user_goal(): + prompt = PromptManager().get_prompt( + "user_simulator.interruption_decision", + conversation_history="AGENT: hello", + user_goal="Get my account unlocked.", + ) + + assert "Get my account unlocked." in prompt + # The goodbye case the caller kept barging in on. + assert "likely to hang up" in prompt diff --git a/tests/unit/user_simulator/cascade/test_scheduler.py b/tests/unit/user_simulator/cascade/test_scheduler.py index 9e7943d5a..b59d509b0 100644 --- a/tests/unit/user_simulator/cascade/test_scheduler.py +++ b/tests/unit/user_simulator/cascade/test_scheduler.py @@ -12,13 +12,16 @@ def __init__(self, speech_ticks: list[bool]) -> None: self.speech_ticks = speech_ticks self.sent: list[bytes | None] = [] self.received_ticks: list[int] = [] + self.barge_in_ticks: list[int] = [] async def start(self) -> None: pass - async def run_tick(self, tick_number: int, outgoing_audio: bytes | None) -> TickResult: + async def run_tick(self, tick_number: int, outgoing_audio: bytes | None, *, barge_in: bool = False) -> TickResult: self.sent.append(outgoing_audio) self.received_ticks.append(tick_number) + if barge_in: + self.barge_in_ticks.append(tick_number) speaking = self.speech_ticks[tick_number] if tick_number < len(self.speech_ticks) else False return TickResult( tick_number=tick_number, @@ -195,7 +198,7 @@ def __init__(self, fail_on_tick: int) -> None: async def start(self) -> None: pass - async def run_tick(self, tick_number: int, outgoing_audio: bytes | None) -> TickResult: + async def run_tick(self, tick_number: int, outgoing_audio: bytes | None, *, barge_in: bool = False) -> TickResult: if tick_number == self.fail_on_tick: raise RuntimeError("adapter failure") return TickResult( @@ -222,3 +225,103 @@ async def test_failed_adapter_call_leaves_queue_and_tick_unadvanced(): assert scheduler.tick == 0 assert bytes(scheduler._playout) == utterance + + +async def test_check_tick_only_while_assistant_speaks_and_caller_is_silent(): + scheduler = _scheduler([True] * 40) + + # Tick 0..9 consumed; tick index 10 is the first multiple of the interval. + for _ in range(10): + await scheduler.run_tick() + + assert scheduler.is_check_tick() is True + + +async def test_not_a_check_tick_between_intervals(): + scheduler = _scheduler([True] * 40) + + for _ in range(11): + await scheduler.run_tick() + + assert scheduler.is_check_tick() is False + + +async def test_not_a_check_tick_when_the_assistant_is_silent(): + scheduler = _scheduler([True] + [False] * 40) + + for _ in range(10): + await scheduler.run_tick() + + assert scheduler.is_check_tick() is False + + +async def test_not_a_check_tick_while_the_caller_is_speaking(): + scheduler = _scheduler([True] * 40) + for _ in range(9): + await scheduler.run_tick() + scheduler.enqueue_utterance(b"\x02" * (BYTES_PER_TICK * 5)) + await scheduler.run_tick() + + assert scheduler.is_check_tick() is False + + +async def test_a_backchannel_does_not_consume_the_callers_turn(): + # A continuer earns no reply, so treating it as a turn deadlocks may_take_turn() + # until the inactivity timeout — 7/8 live conversations died this way. + scheduler = _scheduler([True] + [False] * 60) + await scheduler.run_tick() # tick 0: assistant greets + scheduler.enqueue_backchannel(b"\x02" * BYTES_PER_TICK) + await scheduler.run_tick() # caller says "mm-hmm" + + for _ in range(40): + await scheduler.run_tick() + + assert scheduler.may_take_turn() is True + + +async def test_a_real_utterance_still_consumes_the_turn(): + scheduler = _scheduler([True] + [False] * 60) + await scheduler.run_tick() + scheduler.enqueue_utterance(b"\x02" * BYTES_PER_TICK) + await scheduler.run_tick() + + for _ in range(40): + await scheduler.run_tick() + + assert scheduler.may_take_turn() is False + + +async def test_an_utterance_queued_after_a_backchannel_still_consumes_the_turn(): + scheduler = _scheduler([True] + [False] * 60) + await scheduler.run_tick() + scheduler.enqueue_backchannel(b"\x02" * BYTES_PER_TICK) + scheduler.enqueue_utterance(b"\x03" * BYTES_PER_TICK) + + for _ in range(40): + await scheduler.run_tick() + + assert scheduler.may_take_turn() is False + + +async def test_armed_barge_in_fires_on_the_tick_audio_reaches_the_wire(): + scheduler = _scheduler([True, True, True]) + adapter = scheduler._adapter + + await scheduler.run_tick() # silent tick: nothing on the wire yet + scheduler.arm_barge_in() + scheduler.enqueue_utterance(b"\x01" * BYTES_PER_TICK * 2) + await scheduler.run_tick() + await scheduler.run_tick() + + # Exactly the first tick that carried caller audio, and only that one. + assert adapter.barge_in_ticks == [1] + + +async def test_arming_a_barge_in_with_nothing_queued_does_not_fire_on_silence(): + scheduler = _scheduler([True, True]) + adapter = scheduler._adapter + + scheduler.arm_barge_in() + await scheduler.run_tick() + + assert adapter.barge_in_ticks == [] diff --git a/tests/unit/user_simulator/cascade/test_simulator.py b/tests/unit/user_simulator/cascade/test_simulator.py index f495f1195..a7b824caa 100644 --- a/tests/unit/user_simulator/cascade/test_simulator.py +++ b/tests/unit/user_simulator/cascade/test_simulator.py @@ -1,3 +1,4 @@ +from eva.user_simulator.cascade.phrases import load_phrases from eva.user_simulator.cascade.simulator import CascadeUserSimulator, extract_turn, parse_turn_response @@ -285,6 +286,309 @@ def test_tick_counters_exist_without_manual_setup(): assert sim._ticks_awaiting_transcript == 0 +class RecordingScheduler: + """Captures what the simulator queues for playout.""" + + def __init__(self) -> None: + self.queued: list[bytes] = [] + self.backchannels: list[bytes] = [] + self.tick = 0 + + def enqueue_utterance(self, audio: bytes) -> None: + self.queued.append(audio) + + def enqueue_backchannel(self, audio: bytes) -> None: + self.backchannels.append(audio) + + +class StubCache: + """Phrase cache stand-in that always picks the first phrase.""" + + def choose(self, phrases): + return phrases[0] + + def get(self, phrase): + return b"CACHED" + + +async def test_backchannel_queues_cached_audio_without_synthesis(): + from eva.user_simulator.cascade import simulator as module + + scheduler = RecordingScheduler() + played = module.play_backchannel(scheduler, StubCache(), ["uh-huh", "mm-hmm"]) + + assert played == "uh-huh" + # Queued as a backchannel so it does not consume the caller's turn. + assert scheduler.backchannels == [b"CACHED"] + assert scheduler.queued == [] + + +def test_verdict_with_no_action_queues_nothing(): + from eva.user_simulator.cascade.decisions import ListenerVerdict + + verdict = ListenerVerdict(should_interrupt=False, should_backchannel=False) + + assert verdict.should_interrupt is False + assert verdict.should_backchannel is False + + +def test_a_barge_in_is_kept_while_the_assistant_is_mid_utterance(): + from eva.user_simulator.cascade.simulator import should_drop_interrupt + + assert should_drop_interrupt(assistant_still_speaking=True, same_assistant_turn=True) is False + + +def test_a_slow_barge_in_is_no_longer_dropped_for_being_slow(): + # Wall-clock slip is not staleness: the assistant is still talking, so the line still lands. + from eva.user_simulator.cascade.simulator import should_drop_interrupt + + assert should_drop_interrupt(assistant_still_speaking=True, same_assistant_turn=True) is False + + +def test_a_barge_in_is_dropped_once_the_assistant_stopped(): + # No longer an interruption — it would land as an ordinary reply. + from eva.user_simulator.cascade.simulator import should_drop_interrupt + + assert should_drop_interrupt(assistant_still_speaking=False, same_assistant_turn=True) is True + + +def test_a_barge_in_is_dropped_when_the_assistant_moved_to_a_later_turn(): + # "Still speaking" is satisfied by a *different* turn, which would land the line as a + # non-sequitur against speech the caller never reacted to. + from eva.user_simulator.cascade.simulator import should_drop_interrupt + + assert should_drop_interrupt(assistant_still_speaking=True, same_assistant_turn=False) is True + + +def test_extract_optional_line_reads_a_plain_line(): + from eva.user_simulator.cascade.simulator import extract_optional_line + + assert extract_optional_line("I wanted Friday, not Thursday.") == "I wanted Friday, not Thursday." + + +def test_extract_optional_line_strips_a_code_fence(): + from eva.user_simulator.cascade.simulator import extract_optional_line + + assert extract_optional_line("```\nI meant Friday.\n```") == "I meant Friday." + + +def test_extract_optional_line_is_empty_for_an_empty_reply(): + from eva.user_simulator.cascade.simulator import extract_optional_line + + assert extract_optional_line("") == "" + + +def test_extract_optional_line_reads_through_a_message_object(): + from eva.user_simulator.cascade.simulator import extract_optional_line + + class _Message: + content = "I meant Friday." + + assert extract_optional_line(_Message()) == "I meant Friday." + + +def test_extract_optional_line_rejects_a_refusal_style_non_answer(): + # The prompt allows the model to decline by returning NONE. + from eva.user_simulator.cascade.simulator import extract_optional_line + + assert extract_optional_line("NONE") == "" + + +async def test_relevance_gate_allows_a_still_relevant_candidate(): + from eva.user_simulator.cascade.simulator import candidate_is_relevant + from tests.unit.user_simulator.cascade.test_decisions import FakeLLM + + llm = FakeLLM(["YES"]) + + assert await candidate_is_relevant(llm, candidate="I wanted Friday.", heard="Booking Thursday now") is True + + +async def test_relevance_gate_rejects_a_stale_candidate(): + from eva.user_simulator.cascade.simulator import candidate_is_relevant + from tests.unit.user_simulator.cascade.test_decisions import FakeLLM + + llm = FakeLLM(["NO"]) + + assert await candidate_is_relevant(llm, candidate="I wanted Friday.", heard="What is your name?") is False + + +async def test_relevance_gate_fails_closed(): + from eva.user_simulator.cascade.simulator import candidate_is_relevant + from tests.unit.user_simulator.cascade.test_decisions import FakeLLM + + llm = FakeLLM(error=RuntimeError("down")) + + assert await candidate_is_relevant(llm, candidate="x", heard="y") is False + + +def test_slip_is_measured_on_the_wall_clock_not_the_tick_counter(): + # The tick counter cannot advance during _play_interruption's await: run_tick is + # only pumped by _run, so a tick-delta slip is structurally always zero. + from eva.user_simulator.cascade.simulator import interrupt_slip_ms + + assert interrupt_slip_ms(elapsed_s=1.0) == 1000 + assert interrupt_slip_ms(elapsed_s=0.0) == 0 + assert interrupt_slip_ms(elapsed_s=2.4) == 2400 + + +def test_slip_never_reports_negative_for_a_clock_hiccup(): + from eva.user_simulator.cascade.simulator import interrupt_slip_ms + + assert interrupt_slip_ms(elapsed_s=-0.5) == 0 + + +class _EndCallMessage: + """LLM reply that hangs up via the tool and says nothing.""" + + content = "" + + class _Fn: + name = "end_call" + + class _Call: + function = None + + def __init__(self) -> None: + call = self._Call() + call.function = self._Fn() + self.tool_calls = [call] + + +def _interrupting_simulator(message, framework="elevenlabs"): + """Bare simulator wired for _play_interruption only.""" + from eva.models.config import CascadeSimulatorConfig + from eva.user_simulator.cascade.stt import TranscriptBuffer + + sim = CascadeUserSimulator.__new__(CascadeUserSimulator) + sim._decision_log = _trace_sink() + sim._framework = framework + sim._config = CascadeSimulatorConfig(enable_interruptions=True) + sim._phrases = load_phrases("en") + sim._history = [] + sim._voice_id = "voice-f" + sim._build_prompt = lambda: "SYSTEM PROMPT" + sim.event_logger = _FakeEventLogger() + sim._phrase_cache = StubCache() + sim._record_audio = lambda *a, **k: None + sim._on_user_speaks = lambda *a, **k: None + sim._on_assistant_speaks = lambda *a, **k: None + sim.ended = [] + sim._on_conversation_end = sim.ended.append + buffer = TranscriptBuffer() + buffer.committed = "Your account is unlocked." + sim._stt = type("_Stt", (), {"buffer": buffer})() + + class _Llm: + async def complete(self, messages, tools=None): + return message, {} + + class _Tts: + async def stream(self, text, *, voice_id): + yield text.encode() + + sim._llm, sim._tts = _Llm(), _Tts() + return sim + + +class _InterruptScheduler: + tick = 40 + assistant_is_speaking = True + + def __init__(self) -> None: + self.queued: list[bytes] = [] + self.barge_in_armed = False + + def arm_barge_in(self) -> None: + self.barge_in_armed = True + + def enqueue_utterance(self, audio: bytes) -> None: + self.queued.append(audio) + + +async def test_a_hangup_during_an_interruption_ends_the_call(): + # Discarding end_call here stranded the caller: it emitted nothing and the + # conversation ran on to the inactivity timeout, looping the assistant. + sim = _interrupting_simulator(_EndCallMessage()) + + hung_up = await sim._play_interruption(_InterruptScheduler()) + + assert hung_up is True + assert sim.ended == ["goodbye"] + + +async def test_a_hangup_is_never_dropped_as_stale(): + sim = _interrupting_simulator(_EndCallMessage()) + + class _StoppedScheduler(_InterruptScheduler): + assistant_is_speaking = False # would normally drop the interruption + + assert await sim._play_interruption(_StoppedScheduler()) is True + assert sim.ended == ["goodbye"] + + +async def test_an_ordinary_interruption_does_not_end_the_call(): + sim = _interrupting_simulator("It says my account is locked out.") + + assert await sim._play_interruption(_InterruptScheduler()) is False + assert sim.ended == [] + + +async def test_only_one_interruption_fires_per_assistant_turn(): + # The eligibility flag is cleared on firing, so a second check in the same + # assistant turn is offered allow_interrupt=False and cannot barge in again. + from eva.models.config import CascadeSimulatorConfig + from eva.user_simulator.cascade.decisions import ListenerVerdict + + sim = CascadeUserSimulator.__new__(CascadeUserSimulator) + sim._decision_log = _trace_sink() + sim._config = CascadeSimulatorConfig(enable_interruptions=True) + sim._phrase_cache = StubCache() + sim._may_interrupt_this_turn = True + sim.plays = 0 + offered = [] + + class _Decisions: + async def evaluate(self, text, *, allow_interrupt, allow_backchannel): + offered.append(allow_interrupt) + return ListenerVerdict(should_interrupt=allow_interrupt, should_backchannel=False) + + async def _play(scheduler): + sim.plays += 1 + return False + + sim._decisions = _Decisions() + sim._play_interruption = _play + # The transcript has to grow between the two checks, or the unchanged-transcript + # gate skips the second one before eligibility is ever consulted. + from eva.user_simulator.cascade.stt import TranscriptBuffer + + buffer = TranscriptBuffer() + sim._stt = type("_Stt", (), {"buffer": buffer})() + + buffer.committed = "Let me pull up" + await sim._run_checks(_InterruptScheduler()) + buffer.committed = "Let me pull up your account." + await sim._run_checks(_InterruptScheduler()) + + assert offered == [True, False] + assert sim.plays == 1 + + +def test_a_short_pause_inside_one_assistant_turn_is_not_a_new_turn(): + # A single quiet tick used to re-arm the interruption cap mid-utterance. + from eva.user_simulator.cascade.simulator import is_new_assistant_turn + + assert is_new_assistant_turn(ticks_silent_before=1) is False + assert is_new_assistant_turn(ticks_silent_before=4) is False + + +def test_a_sustained_gap_starts_a_new_assistant_turn(): + from eva.user_simulator.cascade.constants import WAIT_TO_RESPOND_OTHER_MS, ms_to_ticks + from eva.user_simulator.cascade.simulator import is_new_assistant_turn + + assert is_new_assistant_turn(ticks_silent_before=ms_to_ticks(WAIT_TO_RESPOND_OTHER_MS)) is True + + def test_inactivity_measures_contiguous_silence_not_cumulative(): # The reset was unreachable in the live loop, so scattered quiet ticks accumulated # and killed healthy calls once they happened to total two minutes. @@ -305,6 +609,39 @@ def test_inactivity_measures_contiguous_silence_not_cumulative(): assert sim._assistant_is_inactive(_SilenceScheduler(), _tick(False)) is True +async def test_a_dropped_interruption_emits_no_audio_at_all(): + # Emitting the opener up front meant a stale drop left it orphaned on the wire. + sim = _interrupting_simulator("Active Directory.") + + class _StaleScheduler(_InterruptScheduler): + assistant_is_speaking = False # forces should_drop_interrupt + + scheduler = _StaleScheduler() + assert await sim._play_interruption(scheduler) is False + assert scheduler.queued == [] + + +async def test_a_hangup_during_an_interruption_emits_no_opener(): + sim = _interrupting_simulator(_EndCallMessage()) + scheduler = _InterruptScheduler() + + assert await sim._play_interruption(scheduler) is True + assert scheduler.queued == [] + assert sim.ended == ["goodbye"] + + +async def test_a_kept_interruption_speaks_the_opener_then_the_content(): + sim = _interrupting_simulator("Active Directory.") + scheduler = _InterruptScheduler() + + assert await sim._play_interruption(scheduler) is False + + assert scheduler.queued[0] == b"CACHED" + assert b"Active Directory." in b"".join(scheduler.queued[1:]) + # A tick-driven transport needs to know this audio cuts the assistant off. + assert scheduler.barge_in_armed is True + + def test_openai_realtime_uses_the_tick_driven_adapter(): from eva.user_simulator.cascade.adapter.tick_driven import TickDrivenAdapter from eva.user_simulator.cascade.simulator import adapter_class_for_framework @@ -334,6 +671,52 @@ def test_unported_frameworks_default_to_the_real_time_adapter(): assert adapter_class_for_framework("gemini_live") is RealtimeWSAdapter +async def test_the_incomplete_marker_never_reaches_the_conversation_history(): + sim = _interrupting_simulator("Active Directory.") + sim._stt.buffer.apply_partial("and I still nee") + + await sim._play_interruption(_InterruptScheduler()) + + heard = [m["content"] for m in sim._history if m["role"] == "assistant"] + assert heard == ["Your account is unlocked. and I still nee"] + + +async def test_a_slow_barge_in_survives_on_every_transport(): + # Slip is now reported, not enforced: the assistant is mid-utterance, so the line lands. + from eva.user_simulator.cascade import simulator as module + + for framework in ("openai_realtime", "pipecat"): + sim = _interrupting_simulator("Active Directory.", framework=framework) + scheduler = _InterruptScheduler() + + original = module.interrupt_slip_ms + module.interrupt_slip_ms = lambda *, elapsed_s: 2400 + try: + assert await sim._play_interruption(scheduler) is False + finally: + module.interrupt_slip_ms = original + + assert scheduler.queued != [], framework + + +async def test_a_barge_in_is_abandoned_when_a_new_assistant_turn_started_meanwhile(): + # The assistant finished the utterance we reacted to and began another one while the + # line was being generated; firing now would answer speech the caller never heard. + sim = _interrupting_simulator("Active Directory.", framework="pipecat") + scheduler = _InterruptScheduler() + + original_complete = sim._llm.complete + + async def _complete_then_new_turn(messages, tools=None): + sim._assistant_turn_index += 1 + return await original_complete(messages, tools) + + sim._llm.complete = _complete_then_new_turn + + assert await sim._play_interruption(scheduler) is False + assert scheduler.queued == [] + + def test_provider_stall_is_a_distinct_terminal_reason_from_inactivity(): # inactivity_timeout is a legitimate end the metrics treat as definitive when the # user spoke last; a stalled peer is an invalid record the runner should retry. The @@ -353,3 +736,158 @@ def test_provider_stall_is_a_distinct_terminal_reason_from_inactivity(): TickResult(tick_number=9, assistant_audio=b"", assistant_audio_raw_bytes=0, wall_clock_ms=0).provider_stalled is False ) + + +class _CountingDecisions: + """ListenerDecisions stand-in that records every evaluate() it is asked to run.""" + + def __init__(self) -> None: + self.calls: list[str] = [] + + async def evaluate(self, heard, *, allow_interrupt, allow_backchannel): + from eva.user_simulator.cascade.decisions import ListenerVerdict + + self.calls.append(heard) + self.last_allow_backchannel = allow_backchannel + return ListenerVerdict(should_interrupt=False, should_backchannel=False) + + +def _checking_simulator(): + from eva.models.config import CascadeSimulatorConfig + from eva.user_simulator.cascade.stt import TranscriptBuffer + + sim = CascadeUserSimulator.__new__(CascadeUserSimulator) + sim._decision_log = _trace_sink() + sim._config = CascadeSimulatorConfig(enable_interruptions=True, enable_backchannel=True) + sim._phrases = load_phrases("en") + sim._decisions = _CountingDecisions() + sim._phrase_cache = StubCache() + sim._may_interrupt_this_turn = True + sim._last_checked_text = "" + sim.event_logger = _FakeEventLogger() + sim._record_audio = lambda *a, **k: None + buffer = TranscriptBuffer() + sim._stt = type("_Stt", (), {"buffer": buffer})() + return sim, buffer + + +async def test_an_unchanged_transcript_does_not_re_ask_the_judges(): + # The checks fire on a timer but read only the transcript, so a tick where it has + # not moved re-asks a question already answered — ~101 checks x 2 calls per call. + sim, buffer = _checking_simulator() + buffer.committed = "Let me pull up your account." + + await sim._run_checks(_FakeScheduler()) + await sim._run_checks(_FakeScheduler()) + await sim._run_checks(_FakeScheduler()) + + assert sim._decisions.calls == ["Let me pull up your account."] + + +async def test_a_grown_transcript_is_judged_again(): + sim, buffer = _checking_simulator() + buffer.committed = "Let me pull up" + + await sim._run_checks(_FakeScheduler()) + buffer.committed = "Let me pull up your account." + await sim._run_checks(_FakeScheduler()) + + assert len(sim._decisions.calls) == 2 + + +async def test_an_empty_transcript_is_never_judged(): + sim, _buffer = _checking_simulator() + + await sim._run_checks(_FakeScheduler()) + + assert sim._decisions.calls == [] + + +async def test_backchannelling_survives_a_barge_in_in_the_same_turn(): + # Interrupting and later humming along are not mutually exclusive for a real + # listener; the per-turn cap is about not talking over the assistant twice, not + # about going silent for the rest of the turn. + sim, buffer = _checking_simulator() + sim._may_interrupt_this_turn = False # this turn has already barged in + buffer.committed = "Let me pull up your account." + + await sim._run_checks(_FakeScheduler()) + + assert sim._decisions.last_allow_backchannel is True + + +async def test_backchannelling_is_off_only_when_the_config_disables_it(): + from eva.models.config import CascadeSimulatorConfig + + sim, buffer = _checking_simulator() + sim._config = CascadeSimulatorConfig(enable_interruptions=True, enable_backchannel=False) + buffer.committed = "Let me pull up your account." + + await sim._run_checks(_FakeScheduler()) + + assert sim._decisions.last_allow_backchannel is False + + +async def test_a_transcript_already_judged_in_an_earlier_turn_is_judged_again(): + # take_committed() runs between turns, so current_text() can return a string this + # gate has already seen. Without clearing on a turn boundary that later turn's + # check would be skipped as "unchanged" and the barge-in never considered. + sim, buffer = _checking_simulator() + buffer.committed = "Anything else I can help with?" + await sim._run_checks(_FakeScheduler()) + assert len(sim._decisions.calls) == 1 + + # The ordinary turn path consumes the transcript, then the assistant says the same + # thing again a turn later — a real pattern for closing questions. + buffer.take_committed() + sim._last_checked_text = "" # what the new-turn branch in _run does + buffer.committed = "Anything else I can help with?" + + await sim._run_checks(_FakeScheduler()) + + assert len(sim._decisions.calls) == 2 + + +async def test_a_skipped_check_writes_no_listener_check_row(tmp_path): + # The trace exists to separate "the model said NO" from "the check never ran". A row + # for a check that was gated out before reaching the judges would blur exactly that. + from eva.user_simulator.cascade.decision_log import DecisionLog + + sim, buffer = _checking_simulator() + sim._decision_log = DecisionLog(tmp_path / "trace.jsonl") + buffer.committed = "Let me pull up your account." + + await sim._run_checks(_FakeScheduler()) + await sim._run_checks(_FakeScheduler()) + await sim._run_checks(_FakeScheduler()) + + assert sim._decisions.calls == ["Let me pull up your account."] + assert sim._decision_log._counts.get("listener_check") == 1 + + +async def test_the_tick_trace_records_whether_the_transcript_moved(tmp_path): + # A check tick with no listener_check row is otherwise ambiguous between "nothing + # new was heard" and "the check ran and failed". + from eva.user_simulator.cascade.decision_log import DecisionLog + + sim, buffer = _checking_simulator() + rows = [] + sim._decision_log = DecisionLog(tmp_path / "trace.jsonl") + sim._decision_log.log = lambda kind, **fields: rows.append((kind, fields)) + sim._ticks_assistant_silent = 0 + sim._ticks_since_assistant_started = 3 + sim._candidate_audio = b"" + buffer.committed = "Let me pull up your account." + + sim._log_tick_state(_TickTraceScheduler(), _tick(True), is_check_tick=True) + sim._last_checked_text = buffer.current_text() + sim._log_tick_state(_TickTraceScheduler(), _tick(True), is_check_tick=True) + + assert rows[0][1]["transcript_moved"] is True + assert rows[1][1]["transcript_moved"] is False + + +class _TickTraceScheduler: + tick = 12 + caller_is_speaking = False + caller_spoke_this_tick = False diff --git a/tests/unit/user_simulator/cascade/test_stt.py b/tests/unit/user_simulator/cascade/test_stt.py index e4e82c17d..357b0fa23 100644 --- a/tests/unit/user_simulator/cascade/test_stt.py +++ b/tests/unit/user_simulator/cascade/test_stt.py @@ -38,6 +38,30 @@ def test_take_committed_drains_the_buffer(): assert buffer.committed == "" +def test_current_text_marks_the_in_flight_partial_as_incomplete(): + buffer = TranscriptBuffer() + buffer.commit("I found your order.") + buffer.apply_partial("It includes a keyboa") + + assert buffer.current_text() == "I found your order. It includes a keyboa [CURRENTLY SPEAKING, INCOMPLETE]" + + +def test_current_text_omits_the_marker_when_nothing_is_in_flight(): + buffer = TranscriptBuffer() + buffer.commit("I found your order.") + + assert buffer.current_text() == "I found your order." + + +def test_current_text_does_not_consume_the_committed_text(): + buffer = TranscriptBuffer() + buffer.commit("hello") + + buffer.current_text() + + assert buffer.take_committed() == "hello" + + def test_heard_text_never_carries_the_prompt_marker_into_the_transcript(): buffer = TranscriptBuffer() buffer.commit("I found your order.") diff --git a/tests/unit/user_simulator/cascade/test_tick_driven_adapter.py b/tests/unit/user_simulator/cascade/test_tick_driven_adapter.py index e67c88aed..8e9a43ac4 100644 --- a/tests/unit/user_simulator/cascade/test_tick_driven_adapter.py +++ b/tests/unit/user_simulator/cascade/test_tick_driven_adapter.py @@ -69,6 +69,32 @@ async def test_a_silent_tick_still_puts_a_full_tick_of_silence_on_the_wire(): await adapter.stop() +async def test_played_position_tracks_released_ticks(): + ws = FakeWebSocket() + adapter = TickDrivenAdapter(websocket=ws, conversation_id="c1", bytes_per_tick=BYTES_PER_TICK) + await adapter.start() + await ws.inbound.put(_media_frame(b"\xff" * 8000)) + await _settle() + + for tick in range(3): + await adapter.run_tick(tick, None) + + assert adapter.played_ms == 600 + await adapter.stop() + + +async def test_a_tick_with_no_assistant_audio_does_not_advance_the_played_position(): + ws = FakeWebSocket() + adapter = TickDrivenAdapter(websocket=ws, conversation_id="c1", bytes_per_tick=BYTES_PER_TICK) + await adapter.start() + + for tick in range(3): + await adapter.run_tick(tick, None) + + assert adapter.played_ms == 0 + await adapter.stop() + + async def test_ticks_with_audio_already_buffered_do_not_wait_out_the_tick_duration(): ws = FakeWebSocket() adapter = TickDrivenAdapter(websocket=ws, conversation_id="c1", bytes_per_tick=BYTES_PER_TICK) @@ -119,6 +145,39 @@ async def _deliver_late() -> None: await adapter.stop() +async def test_barge_in_reports_the_played_position_not_the_received_position(): + ws = FakeWebSocket() + adapter = TickDrivenAdapter(websocket=ws, conversation_id="c1", bytes_per_tick=BYTES_PER_TICK) + await adapter.start() + # 1s of audio arrives at once but only 3 ticks (600ms) get released. + await ws.inbound.put(_media_frame(b"\xff" * 8000)) + await _settle() + for tick in range(3): + await adapter.run_tick(tick, None) + + result = await adapter.run_tick(3, b"\x00" * BYTES_PER_TICK, barge_in=True) + + assert result.interruption_audio_start_ms == 600 + truncate = [json.loads(m) for m in ws.sent if json.loads(m).get("event") == "truncate"] + assert truncate and truncate[0]["audio_end_ms"] == 600 + await adapter.stop() + + +async def test_barge_in_discards_audio_the_caller_never_heard(): + ws = FakeWebSocket() + adapter = TickDrivenAdapter(websocket=ws, conversation_id="c1", bytes_per_tick=BYTES_PER_TICK) + await adapter.start() + await ws.inbound.put(_media_frame(b"\xff" * 8000)) + await _settle() + + result = await adapter.run_tick(0, b"\x00" * BYTES_PER_TICK, barge_in=True) + + # The buffered second of assistant audio is audio the caller cut off. + assert result.assistant_audio_raw_bytes == 0 + assert adapter.played_ms == 0 + await adapter.stop() + + async def test_a_stalled_provider_is_reported_not_raised(): # Raising aborted the tick loop from under the simulator, so the record ended on the # generic "error" reason and the terminal state the runner reads was never written. diff --git a/tests/unit/user_simulator/cascade/test_tick_result.py b/tests/unit/user_simulator/cascade/test_tick_result.py index f1696f656..57a338ed7 100644 --- a/tests/unit/user_simulator/cascade/test_tick_result.py +++ b/tests/unit/user_simulator/cascade/test_tick_result.py @@ -1,6 +1,6 @@ import pytest -from eva.user_simulator.cascade.tick_result import TickResult, split_tick_audio +from eva.user_simulator.cascade.tick_result import TickResult, played_audio_ms, split_tick_audio @pytest.mark.parametrize("input_length", [0, 1, 7, 8, 9, 17]) @@ -42,3 +42,33 @@ def test_has_assistant_speech_is_true_when_real_audio_arrived(): result = TickResult(tick_number=3, assistant_audio=b"\x01" * 8, assistant_audio_raw_bytes=8, wall_clock_ms=1) assert result.has_assistant_speech is True + + +def test_played_position_reflects_released_ticks_not_received_bytes(): + # 12 ticks released at 200ms each, regardless of how much arrived early. + assert played_audio_ms(ticks_released=12) == 2400 + + +def test_played_position_is_zero_before_anything_is_released(): + assert played_audio_ms(ticks_released=0) == 0 + + +def test_truncation_defaults_are_inert(): + result = TickResult(tick_number=1, assistant_audio=b"\x00" * 8, assistant_audio_raw_bytes=0, wall_clock_ms=0) + + assert result.skip_item_id is None + assert result.interruption_audio_start_ms is None + + +def test_truncation_fields_carry_the_played_position(): + result = TickResult( + tick_number=1, + assistant_audio=b"\x00" * 8, + assistant_audio_raw_bytes=0, + wall_clock_ms=0, + skip_item_id="item_42", + interruption_audio_start_ms=2400, + ) + + assert result.skip_item_id == "item_42" + assert result.interruption_audio_start_ms == 2400