diff --git a/README.md b/README.md index 2ded521..ad4c791 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,8 @@ The sample includes: - JWT validation for the WxCC data plane - A configuration-driven connector router - Local audio, AWS Lex, and Google CX Agent Studio connectors +- Immediate raw 8 kHz mu-law BYOVA `CHUNK` output for Google CX Agent Studio, + with one ordered `FINAL` per normal or terminal turn - gRPC and HTTP health checks - A development monitoring dashboard - Unit tests and local gRPC smoke-test utilities diff --git a/config/README.md b/config/README.md index e8ada5f..46fc5d2 100644 --- a/config/README.md +++ b/config/README.md @@ -30,7 +30,8 @@ buffered per stream while caller ingestion, ordered connector processing, and Wx delivery run independently. Both queue sizes must be greater than zero. The queues bound memory and apply backpressure; they are not production throughput targets. -`max_terminal_playback_seconds` is a safety ceiling for the buffered-WAV announcement gate. +`max_terminal_playback_seconds` is a safety ceiling for legacy complete-WAV +announcement responses. GECX raw CHUNK streaming does not use this delay. Maximum gRPC message sizes and the concurrent-stream option remain set in `main.py`. ## Voice Activity Detection @@ -106,7 +107,9 @@ and end-to-end sandbox test paths. ### GECX / CX Agent Studio Connector The GECX connector streams WxCC caller audio to Google CX Agent Studio through the CES -`BidiRunSession` API and returns each completed agent turn as a WxCC-compatible response. +`BidiRunSession` API. CES 8 kHz mu-law output frames are forwarded immediately +as raw BYOVA `CHUNK` responses, followed by exactly one normal or terminal +`FINAL`. ```yaml connectors: @@ -136,6 +139,10 @@ connectors: - "My GECX Agent" ``` +GECX CHUNK output currently requires `output_sample_rate_hertz: 8000` and +`output_audio_encoding: "MULAW"`. Unsupported output combinations fail during +connector initialization; broader output formats are not silently mislabeled. + See [`gecx_example.yaml`](gecx_example.yaml) for all options and the [GECX Setup Guide](../docs/guides/byova-gecx-setup.md) for IAM, deployment, and Webex Contact Center configuration. diff --git a/config/config.cloudrun.yaml b/config/config.cloudrun.yaml index 5f5c264..e9cf018 100644 --- a/config/config.cloudrun.yaml +++ b/config/config.cloudrun.yaml @@ -45,6 +45,7 @@ connectors: language_code: "en-US" input_sample_rate_hertz: 8000 input_audio_encoding: "MULAW" + # Required by the current raw BYOVA CHUNK output path. output_sample_rate_hertz: 8000 output_audio_encoding: "MULAW" initial_message: "Hello" diff --git a/config/config.yaml b/config/config.yaml index 37a82ae..7300203 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -11,7 +11,8 @@ gateway: # and response delivery run independently. request_queue_maxsize: 100 response_queue_maxsize: 100 - # Safety ceiling only; the actual terminal gate is derived from each CES WAV. + # Safety ceiling for legacy complete-WAV prompt responses. GECX raw CHUNK + # streaming preserves order without this playback delay. max_terminal_playback_seconds: 30 # Voice activity detection controls. The configured detector implementation is internal. diff --git a/config/gecx_example.yaml b/config/gecx_example.yaml index d6d2384..23183f1 100644 --- a/config/gecx_example.yaml +++ b/config/gecx_example.yaml @@ -29,7 +29,9 @@ gecx_connector: language_code: "en-US" - # WxCC telephony is typically 8 kHz MULAW in both directions + # Input can be normalized from WxCC metadata. The current raw BYOVA CHUNK + # output path requires exactly 8 kHz MULAW; other output combinations fail + # connector initialization until explicit format support is added. input_sample_rate_hertz: 8000 input_audio_encoding: "MULAW" output_sample_rate_hertz: 8000 diff --git a/docs/guides/byova-gecx-setup.md b/docs/guides/byova-gecx-setup.md index 0e163e3..8086d1d 100644 --- a/docs/guides/byova-gecx-setup.md +++ b/docs/guides/byova-gecx-setup.md @@ -8,9 +8,10 @@ This guide explains how to connect Webex Contact Center (WxCC) BYOVA to an agent Caller -> WxCC -> BYOVA Gateway (gRPC) -> GECXConnector -> CES BidiRunSession -> CX Agent Studio ``` -The connector sends WxCC caller audio to Google as it arrives. Its current -output mode buffers each CES agent turn and returns one WAV `FINAL` response; -BYOVA `CHUNK` output streaming is planned separately. +The connector sends WxCC caller audio to Google as it arrives and forwards each +CES 8 kHz mu-law output frame immediately as a BYOVA `CHUNK`. Each agent turn +ends with exactly one `FINAL`; terminal turns place `TRANSFER_TO_AGENT` or +`SESSION_END` on that final response. ## Prerequisites @@ -169,18 +170,16 @@ are worth understanding if you fork this connector. `GECXStreamingSession` runs a background thread per conversation that holds one CES `BidiRunSession` open. WxCC caller audio is pushed onto an inbound queue and -forwarded to CES; CES server messages (STT, agent text, TTS audio, barge-in, -end-of-session) are mapped to BYOVA responses on an outbound queue. The gateway -drains responses while caller audio is flowing and, after gateway speech-end -detection, waits for CES `turn_completed` before returning the final turn to -WxCC. This prevents a response that arrives after `END_OF_INPUT` from remaining -queued when WxCC stops sending caller audio. +forwarded to CES; CES server messages (STT, agent text, TTS audio, +interruption, and end-of-session) are mapped to BYOVA responses on an outbound +queue. After the gateway emits `END_OF_INPUT`, it consumes that queue +incrementally instead of materializing the complete turn. The first CES audio +frame can therefore reach WxCC before CES emits `turn_completed`. All terminal causes pass through one session-scoped decision guard. The first -decision rejects later caller input, finalizes agent audio received before the -decision, half-closes the CES request stream once, and emits at most one -terminal response. Duplicate `EndSession` messages and late CES output are -ignored. +decision rejects later caller input, preserves already-queued audio chunks, +half-closes the CES request stream once, and emits at most one terminal +`FINAL`. Duplicate `EndSession` messages and late CES output are ignored. | Terminal cause | WxCC outcome | |----------------|--------------| @@ -213,28 +212,36 @@ merges up to `input_pause_preroll_ms` of the resumed onset, and keeps one CES input turn. Otherwise it commits the boundary normally. Configure the observer under the top-level `voice_activity_detection` block in `config/config.yaml`. -### Audio format: WxCC expects a self-describing WAV clip +### Output audio: raw 8 kHz mu-law BYOVA chunks -This is the single most important detail. WxCC's `Prompt.audio_content` field -carries **no encoding metadata**, so the bytes must be a **self-describing WAV -file**. Telephony uses **8 kHz, 8-bit, mono mu-law** (`WAVE_FORMAT_MULAW`). +CES streams TTS output as small frames. The connector places every frame +directly in `Prompt.audio_content` with `response_type=CHUNK`, keeping +`is_barge_in_enabled=false`. It does not accumulate a full turn and does not +add a WAV header. -CES streams TTS output as many small raw frames per agent turn. If you forward -those raw frames straight to WxCC, the caller **hears nothing** (WxCC can't tell -what format the bytes are). The connector therefore: +The normal response sequence is: -1. **Buffers** all raw CES audio frames for an agent turn. -2. On turn completion (`turn_completed` / `end_session`), **wraps** the whole - buffer in a WAV header (via `wrap_output_audio` / `_build_wxcc_wav`). -3. Emits **one** complete WAV clip per turn as a single WxCC prompt — the same - shape the `local_audio_connector` produces. +```text +START_OF_INPUT (standalone FINAL event) +END_OF_INPUT (CHUNK) +audio (CHUNK) +audio (CHUNK) +... +turn complete (FINAL) +``` + +A terminal turn uses the same audio chunks, followed by one `FINAL` carrying +`TRANSFER_TO_AGENT` or `SESSION_END`. The initial greeting uses the same +`CHUNK`/`FINAL` pipeline. -Barge-in (`interruption_signal`) clears the buffer and any queued audio so the -agent stops talking when the caller interrupts. +The current CHUNK path intentionally supports only 8 kHz mu-law output. +`output_audio_encoding` must remain `MULAW` and +`output_sample_rate_hertz` must remain `8000`; unsupported combinations fail +configuration early. Broader output-format support requires explicit +conversion and validation. -> If you change `output_audio_encoding`/`output_sample_rate_hertz`, keep them -> consistent with the WAV header the connector writes. For WxCC telephony, leave -> them at `MULAW` / `8000`. +Barge-in remains disabled. CES `interruption_signal` cleanup is retained, but +caller-driven output cancellation is a separate implementation phase. ### Input audio @@ -312,19 +319,16 @@ Wire that branch to a queue that routes to human agents. (A normal `SESSION_END` ends the virtual-agent interaction without a transfer.) When CES includes its final spoken announcement with `EndSession`, the GECX -connector sends the CES announcement first and follows it with a prompt-free -terminal response. WxCC skips prompt audio when `TRANSFER_TO_AGENT` shares the -same response, so this ordering lets the full CES announcement play and then -transfers as soon as playback completes. The gateway calculates that gate from -the CES WAV byte rate and data length rather than applying a fixed termination -delay. - -The connector also keeps CES text with its matching CES audio until the turn is -complete. This prevents WxCC from synthesizing a separate text-only prompt ahead -of the provider audio, which would duplicate speech and delay later responses. -For GECX, the gateway also places `END_OF_INPUT` on that completed response -instead of sending it as a preceding standalone response; `START_OF_INPUT` -remains immediate. +connector streams the announcement as ordered `CHUNK` responses and follows it +with one prompt-free terminal `FINAL`. The GECX path no longer calculates a WAV +playback delay; the response stream itself carries the required order. + +The connector retains CES text as transcript/fallback state but leaves it off +audio chunks. This prevents WxCC from synthesizing a duplicate text prompt. +For GECX, `START_OF_INPUT` remains an immediate standalone `FINAL` event so +WxCC continues forwarding caller audio through a bounded natural pause. +`END_OF_INPUT` begins the output stream as a `CHUNK` event; audio chunks follow +it, and the output turn closes with one `FINAL`. Caller audio is buffered from a bounded pre-roll through the gateway's Silero speech-end boundary, then sent to CES as one contiguous turn. This prevents a @@ -345,7 +349,9 @@ window for an `EndSession` that follows the final TTS frames. | `api_endpoint` | No | CES endpoint; defaults to `ces..rep.googleapis.com` | | `service_account_key` | No | Path to SA JSON; omit to use ADC | | `initial_message` | No | Text sent when the CES stream opens (default: `Hello`) | -| `enable_partial_responses` | No | Map CES partial outputs to WxCC `PARTIAL` responses | +| `enable_partial_responses` | No | Request CES text streaming for logs, terminal-cue detection, and text-only fallback | +| `output_sample_rate_hertz` | No | Must be `8000` for the current raw CHUNK path | +| `output_audio_encoding` | No | Must be `MULAW` for the current raw CHUNK path | | `force_input_format` | No | `wxcc` forces 8 kHz MULAW when input metadata is unavailable | | `turn_response_timeout_seconds` | No | Maximum wait after gateway speech end for CES to complete the agent turn (default: `30`) | | `endpointing_silence_ms` | No | Codec-correct silence appended to each buffered caller turn for CES endpoint detection (default: `2000`; one second may leave a turn open until more audio arrives) | @@ -378,7 +384,7 @@ and grant the represented identity `roles/ces.client`. | Stream fails on start | `roles/ces.client`, API enabled, correct `location` | | `404` / `UNIMPLEMENTED` on BidiRunSession | Wrong endpoint — must be regional `ces..rep.googleapis.com` (auto-derived from `location`) | | `429 Resource exhausted` | CES per-app session quota; retry/backoff or request more quota | -| No audio to caller (silence) | WxCC needs a WAV-wrapped clip, not raw audio. Confirm `Audio out: NNNN bytes WAV` in logs and `output_audio_encoding: MULAW` / `output_sample_rate_hertz: 8000`. See [How it works](#audio-format-wxcc-expects-a-self-describing-wav-clip). | +| No audio to caller (silence) | Confirm `gecx_first_audio_chunk` appears, the next response is a BYOVA `CHUNK`, and output remains `MULAW` / `8000`. See [Output audio](#output-audio-raw-8-khz-mu-law-byova-chunks). | | Garbled speech | Confirm the gateway logs the declared WxCC encoding/sample rate; use `force_input_format: "wxcc"` only when the client omits metadata | | No response after `END_OF_INPUT` | Check for `turn_completed` or a turn-completion timeout in `[GECX]` logs; increase `turn_response_timeout_seconds` if the agent regularly needs more than 30 seconds | | `GoAway` from CES | The connector intentionally emits one `SESSION_END` and half-closes CES; it does not reconnect in the current implementation | @@ -391,8 +397,12 @@ Search gateway logs for `[GECX]`: - `Starting conversation` — session created - `STT` — recognition results from CES - `Agent` — text responses -- `Audio out: NNNN bytes WAV (MMMM raw)` — one WAV clip emitted per agent turn - (`NNNN` includes the WAV header; `MMMM` is the raw CES bytes buffered) +- `gecx_first_audio_chunk` — first raw CES frame queued for WxCC, including + first-frame latency +- `gecx_streamed_turn_complete` — one normal `FINAL` emitted after the logged + chunk and byte totals +- `gecx_terminal_decision` — one terminal `FINAL`, with chunk and byte totals + but without sensitive metadata values - `Barge-in` — interruption signal from CES - `gecx_terminal_decision` — the winning lifecycle decision, with `conversation_id`, CES `session`, `reason`, `outcome`, `source`, diff --git a/src/connectors/README.md b/src/connectors/README.md index dc245f3..c516eea 100644 --- a/src/connectors/README.md +++ b/src/connectors/README.md @@ -158,7 +158,8 @@ Experience) through the CES `BidiRunSession` API. **Features**: - Streams WxCC caller audio to Google as it arrives - Maps CES recognition, text, audio, interruption, and end-session messages -- Buffers each CES output turn into a WxCC-compatible 8 kHz mu-law WAV response +- Streams each raw 8 kHz mu-law CES output frame as a BYOVA `CHUNK` +- Emits exactly one normal or terminal `FINAL` after the ordered chunks - Maps CES escalation metadata to WxCC human-transfer events - Supports service-account credentials, ADC, OAuth, and short-lived access tokens diff --git a/src/connectors/gecx_connector.py b/src/connectors/gecx_connector.py index 68d9686..6aec156 100644 --- a/src/connectors/gecx_connector.py +++ b/src/connectors/gecx_connector.py @@ -18,7 +18,7 @@ import uuid from dataclasses import dataclass from enum import Enum -from typing import Any, Dict, Generator, Iterator, Optional, Tuple, Union +from typing import Any, Dict, Generator, Iterator, Optional, Tuple try: import audioop @@ -134,7 +134,6 @@ def __init__( self._stop_event = threading.Event() self._stream_started = threading.Event() self._turn_completed = threading.Event() - self._terminal_event = threading.Event() self._stream_error: Optional[str] = None self._thread: Optional[threading.Thread] = None self._lock = threading.Lock() @@ -142,12 +141,15 @@ def __init__( self._terminal_decision: Optional[GECXTerminalDecision] = None self._join_attempted = False self._started_at = time.monotonic() - # CES streams text and TTS output separately during an agent turn. WxCC - # expects one complete prompt, so keep both forms together and emit one - # response when the turn completes. Sending the text early as its own - # prompt makes WxCC synthesize it and blocks the later CES audio/event. + # CES streams text and TTS output separately during an agent turn. Text + # is retained only as a transcript/fallback; each audio frame is emitted + # immediately as raw telephony audio so WxCC can begin playback before + # CES marks the turn complete. self._text_buffer: list[str] = [] - self._audio_buffer = bytearray() + self._turn_audio_emitted = False + self._turn_audio_chunk_count = 0 + self._turn_audio_bytes = 0 + self._turn_started_at = time.monotonic() self._input_lock = threading.Lock() self._input_audio_buffer = bytearray() self._input_resume_buffer = bytearray() @@ -349,33 +351,136 @@ def begin_input_turn(self) -> bool: self._input_turn_active = True self._input_turn_paused = False self._input_resume_buffer.clear() + self._turn_started_at = time.monotonic() self._turn_completed.clear() return True - def wait_for_turn_responses( - self, timeout: float, *, terminate_on_timeout: bool = True - ) -> Tuple[bool, list[Dict[str, Any]]]: - """Wait for CES to complete the current turn, then drain its responses.""" - completed = self._turn_completed.wait(timeout=timeout) - if not completed: - self.logger.warning( - "[%s] [GECX] Timed out after %.1fs waiting for turn completion", - self.conversation_id, - timeout, - ) - if terminate_on_timeout: - self.terminate( - reason=GECXTerminalReason.TIMEOUT, - outcome=GECXTerminalOutcome.SESSION_END, - source="turn_response_timeout", - metadata={"timeout_seconds": timeout}, + def iter_turn_responses( + self, + timeout: float, + *, + terminal_grace_seconds: float = 0.0, + terminate_on_timeout: bool = True, + ) -> Iterator[Dict[str, Any]]: + """Yield CES audio frames immediately, followed by exactly one FINAL.""" + deadline = time.monotonic() + max(0.0, timeout) + terminal_grace_deadline: Optional[float] = None + + while True: + now = time.monotonic() + completed = self._turn_completed.is_set() + + if not completed and now >= deadline: + self.logger.warning( + "[%s] [GECX] Timed out after %.1fs waiting for turn completion", + self.conversation_id, + timeout, + ) + if terminate_on_timeout: + self.terminate( + reason=GECXTerminalReason.TIMEOUT, + outcome=GECXTerminalOutcome.SESSION_END, + source="turn_response_timeout", + metadata={"timeout_seconds": timeout}, + ) + continue + return + + try: + response = self.outbound_queue.get_nowait() + except queue.Empty: + response = None + if response is not None: + yield response + if response.get("message_type") in {"transfer", "session_end"}: + return + continue + + if completed and self.outbound_queue.empty(): + if self.is_terminal: + return + + buffered_text = self._active_text() + may_have_terminal = self.connector.may_have_delayed_terminal( + buffered_text ) - return completed, self.drain_responses() + if ( + may_have_terminal + and terminal_grace_seconds > 0 + and terminal_grace_deadline is None + ): + terminal_grace_deadline = ( + now + max(0.0, terminal_grace_seconds) + ) + self.logger.info( + "[%s] [GECX] Waiting up to %.1fs for a terminal event " + "after streamed announcement audio", + self.conversation_id, + terminal_grace_seconds, + ) + + if ( + terminal_grace_deadline is None + or now >= terminal_grace_deadline + ): + final_response = self._finish_normal_turn() + if final_response is not None: + yield final_response + return + + wait_deadline = deadline + if completed and terminal_grace_deadline is not None: + wait_deadline = terminal_grace_deadline + wait_seconds = min(0.1, max(0.0, wait_deadline - now)) + if wait_seconds <= 0: + continue + + try: + response = self.outbound_queue.get(timeout=wait_seconds) + except queue.Empty: + continue + + yield response + if response.get("message_type") in {"transfer", "session_end"}: + return - def wait_for_terminal_responses(self, timeout: float) -> list[Dict[str, Any]]: - """Wait briefly for an EndSession that CES emits after TTS completion.""" - self._terminal_event.wait(timeout=timeout) - return self.drain_responses() + def _active_text(self) -> str: + """Return the current CES text without exposing mutable session state.""" + with self._lock: + return "".join(self._text_buffer) + + def _finish_normal_turn(self) -> Optional[Dict[str, Any]]: + """Commit the sole non-terminal FINAL after all streamed audio chunks.""" + with self._lifecycle_lock: + if self._terminal_decision is not None: + return None + with self._lock: + buffered_text = "".join(self._text_buffer) + audio_emitted = self._turn_audio_emitted + chunk_count = self._turn_audio_chunk_count + audio_bytes = self._turn_audio_bytes + self._text_buffer = [] + self._turn_audio_emitted = False + self._turn_audio_chunk_count = 0 + self._turn_audio_bytes = 0 + + self.logger.info( + "gecx_streamed_turn_complete conversation_id=%s chunks=%d " + "audio_bytes=%d elapsed_seconds=%.3f terminal=false", + self.conversation_id, + chunk_count, + audio_bytes, + time.monotonic() - self._turn_started_at, + ) + return self.connector.create_response( + conversation_id=self.conversation_id, + message_type=( + "silence" if audio_emitted or not buffered_text else "agent_response" + ), + text="" if audio_emitted else buffered_text, + barge_in_enabled=False, + response_type="final", + ) def terminate( self, @@ -387,12 +492,10 @@ def terminate( ) -> bool: """Make the session's terminal decision exactly once. - The winning call rejects future input, preserves already-buffered agent - audio when the response channel is usable, half-closes the CES request - stream, wakes response waiters, and optionally queues one canonical - terminal response for the gateway. Preserved audio is attached to that - same response so WxCC receives the final prompt and terminal event - atomically. Later calls are no-ops. + The winning call rejects future input, preserves audio CHUNK responses + already queued for the gateway, half-closes the CES request stream, + wakes response waiters, and optionally queues one canonical terminal + FINAL after those chunks. Later calls are no-ops. """ terminal_metadata = dict(metadata or {}) decided_at = time.monotonic() @@ -433,40 +536,26 @@ def terminate( self._input_turn_paused = False with self._lock: buffered_text = "".join(self._text_buffer) - raw_audio = bytes(self._audio_buffer) + audio_emitted = self._turn_audio_emitted + chunk_count = self._turn_audio_chunk_count + audio_bytes = self._turn_audio_bytes self._text_buffer = [] - self._audio_buffer = bytearray() + self._turn_audio_emitted = False + self._turn_audio_chunk_count = 0 + self._turn_audio_bytes = 0 # Queue the stop sentinel exactly once. The request generator also # checks terminal state after dequeuing to close the final race where # it already pulled caller audio as termination was decided. self.inbound_queue.put(_STREAM_STOP) - # Preserve final CES output ahead of the terminal event. WxCC skips - # a prompt when TRANSFER_TO_AGENT shares its response, so the audio - # must be its own response. Gateway backpressure then advances to - # the terminal response as soon as prompt playback completes. - preserved_output = False - if outcome != GECXTerminalOutcome.SILENT: - wav_audio = ( - self.connector.wrap_output_audio(raw_audio) if raw_audio else b"" - ) - if buffered_text or wav_audio: - self.outbound_queue.put( - self.connector.create_response( - conversation_id=self.conversation_id, - message_type="audio" if wav_audio else "agent_response", - text=buffered_text or response_text, - audio_content=wav_audio, - barge_in_enabled=False, - response_type="final", - ) - ) - preserved_output = True - terminal_response = self._create_terminal_response( decision, - response_text="" if preserved_output else (response_text or None), + response_text=( + "" + if audio_emitted + else (buffered_text or response_text or None) + ), ) if terminal_response is not None: self.outbound_queue.put(terminal_response) @@ -474,18 +563,19 @@ def terminate( # Set completion only after the terminal response is visible so a # waiter cannot wake and drain the queue too early. self._turn_completed.set() - self._terminal_event.set() self.logger.info( "gecx_terminal_decision conversation_id=%s session=%s reason=%s " - "outcome=%s source=%s elapsed_seconds=%.3f metadata_keys=%s " - "end_session_metadata_keys=%s", + "outcome=%s source=%s elapsed_seconds=%.3f chunks=%d " + "audio_bytes=%d metadata_keys=%s end_session_metadata_keys=%s", self.conversation_id, self.session_path, reason.value, outcome.value, source, decided_at - self._started_at, + chunk_count, + audio_bytes, sorted(terminal_metadata), sorted( terminal_metadata.get("end_session", {}) @@ -667,7 +757,9 @@ def _handle_server_message(self, message: Any) -> None: if self._terminal_decision is None: with self._lock: self._text_buffer = [] - self._audio_buffer = bytearray() + self._turn_audio_emitted = False + self._turn_audio_chunk_count = 0 + self._turn_audio_bytes = 0 while True: try: self.outbound_queue.get_nowait() @@ -686,16 +778,11 @@ def _handle_server_message(self, message: Any) -> None: audio_bytes = self._decode_output_audio(output.audio) if audio_bytes: - self.logger.debug( - f"[{conversation_id}] [GECX] Buffered audio frame: " - f"{len(audio_bytes)} bytes" - ) - self._buffer_active_audio(audio_bytes) + self._emit_active_audio_chunk(audio_bytes) - # A completed turn means the full agent utterance has been streamed; - # emit it to WxCC as a single WAV clip. + # Audio frames are already queued as CHUNK responses. Completion + # only releases the response iterator to emit its sole FINAL. if output.turn_completed and not has_terminal_output: - self._flush_audio_buffer() turn_completed = True if has_terminal_output: @@ -728,13 +815,43 @@ def _enqueue_active_response(self, response: Dict[str, Any]) -> bool: self.outbound_queue.put(response) return True - def _buffer_active_audio(self, audio_bytes: bytes) -> bool: - """Buffer agent audio only while the session is active.""" + def _emit_active_audio_chunk(self, audio_bytes: bytes) -> bool: + """Queue one raw CES audio frame as a BYOVA CHUNK response.""" with self._lifecycle_lock: if self._terminal_decision is not None: return False with self._lock: - self._audio_buffer.extend(audio_bytes) + self._turn_audio_emitted = True + self._turn_audio_chunk_count += 1 + self._turn_audio_bytes += len(audio_bytes) + chunk_index = self._turn_audio_chunk_count + total_bytes = self._turn_audio_bytes + self.outbound_queue.put( + self.connector.create_response( + conversation_id=self.conversation_id, + message_type="audio", + audio_content=audio_bytes, + barge_in_enabled=False, + response_type="chunk", + ) + ) + + if chunk_index == 1: + self.logger.info( + "gecx_first_audio_chunk conversation_id=%s bytes=%d " + "elapsed_seconds=%.3f", + self.conversation_id, + len(audio_bytes), + time.monotonic() - self._turn_started_at, + ) + else: + self.logger.debug( + "[%s] [GECX] Audio chunk %d: %d bytes (%d total)", + self.conversation_id, + chunk_index, + len(audio_bytes), + total_bytes, + ) return True def _buffer_active_text(self, text: str) -> bool: @@ -746,48 +863,6 @@ def _buffer_active_text(self, text: str) -> bool: self._text_buffer.append(text) return True - def _flush_audio_buffer(self, response_type: str = "final") -> bool: - """Emit buffered CES text and audio as one WxCC prompt response. - - Returns True if a response was emitted. Safe to call repeatedly; it is - a no-op when both buffers are empty. - """ - with self._lifecycle_lock: - if self._terminal_decision is not None: - with self._lock: - self._text_buffer = [] - self._audio_buffer = bytearray() - return False - with self._lock: - if not self._text_buffer and not self._audio_buffer: - return False - buffered_text = "".join(self._text_buffer) - raw_audio = bytes(self._audio_buffer) - self._text_buffer = [] - self._audio_buffer = bytearray() - - wav_audio = ( - self.connector.wrap_output_audio(raw_audio) if raw_audio else b"" - ) - if not buffered_text and not wav_audio: - return False - - if wav_audio: - self.logger.info( - f"[{self.conversation_id}] [GECX] Audio out: " - f"{len(wav_audio)} bytes WAV ({len(raw_audio)} raw)" - ) - self.outbound_queue.put( - self.connector.create_response( - conversation_id=self.conversation_id, - message_type="audio" if wav_audio else "agent_response", - text=buffered_text, - audio_content=wav_audio, - response_type=response_type, - ) - ) - return True - @staticmethod def _metadata_to_dict(end_obj: Any) -> Dict[str, Any]: """Best-effort conversion of an EndSession.metadata Struct to a dict.""" @@ -997,6 +1072,20 @@ def __init__(self, config: Dict[str, Any]) -> None: self.output_sample_rate_hertz = config.get("output_sample_rate_hertz", 8000) self.input_audio_encoding = config.get("input_audio_encoding", "MULAW") self.output_audio_encoding = config.get("output_audio_encoding", "MULAW") + normalized_output_encoding = ( + str(self.output_audio_encoding) + .upper() + .replace("AUDIO_ENCODING_", "") + .replace("-", "_") + ) + if ( + normalized_output_encoding not in {"MULAW", "ULAW", "LINEAR_16_MULAW"} + or self.output_sample_rate_hertz != 8000 + ): + raise ValueError( + "GECX BYOVA CHUNK streaming currently requires " + "output_audio_encoding=MULAW and output_sample_rate_hertz=8000" + ) self.initial_message = config.get("initial_message", "Hello") self.enable_partial_responses = config.get("enable_partial_responses", True) self.force_input_format = config.get("force_input_format", "").lower() @@ -1186,60 +1275,6 @@ def create_error_response( error=error_message, ) - @staticmethod - def _build_wxcc_wav(audio_bytes: bytes, sample_rate: int, encoding: str) -> bytes: - """Wrap raw audio in a WAV container that WxCC can play. - - WxCC's ``Prompt.audio_content`` carries no encoding metadata, so the - payload must be a self-describing WAV file. Telephony uses 8 kHz, 8-bit - mono mu-law; PCM is supported as a fallback. - """ - if not audio_bytes: - return b"" - - enc = (encoding or "MULAW").upper().replace("AUDIO_ENCODING_", "") - if enc in ("MULAW", "ULAW", "LINEAR_16_MULAW"): - audio_format = 7 # WAVE_FORMAT_MULAW - bits_per_sample = 8 - elif enc == "ALAW": - audio_format = 6 # WAVE_FORMAT_ALAW - bits_per_sample = 8 - else: # LINEAR16 / PCM - audio_format = 1 # WAVE_FORMAT_PCM - bits_per_sample = 16 - - channels = 1 - bytes_per_sample = bits_per_sample // 8 - block_align = channels * bytes_per_sample - byte_rate = sample_rate * block_align - data_size = len(audio_bytes) - - header = struct.pack( - "<4sI4s4sIHHIIHH4sI", - b"RIFF", - 36 + data_size, - b"WAVE", - b"fmt ", - 16, - audio_format, - channels, - sample_rate, - byte_rate, - block_align, - bits_per_sample, - b"data", - data_size, - ) - return header + audio_bytes - - def wrap_output_audio(self, audio_bytes: bytes) -> bytes: - """Convert raw CES output audio into a WxCC-playable WAV clip.""" - return self._build_wxcc_wav( - audio_bytes, - self.output_sample_rate_hertz, - self.output_audio_encoding, - ) - def get_audio_delivery_mode(self) -> str: """GECX receives caller audio frames as they arrive from WxCC.""" return "streaming" @@ -1249,7 +1284,7 @@ def should_cleanup_on_client_stream_end(self) -> bool: return True def should_coalesce_speech_end_with_response(self) -> bool: - """Send END_OF_INPUT with the completed CES prompt and terminal event.""" + """Keep END_OF_INPUT ordered ahead of the CES response stream.""" return True def should_merge_speech_pauses(self) -> bool: @@ -1315,23 +1350,15 @@ def handle_speech_boundary( if not message_data.get("speech_turn_committed"): self.commit_speech_turn(conversation_id) - completed, responses = stream_session.wait_for_turn_responses( - timeout=self.turn_response_timeout_seconds + yield from stream_session.iter_turn_responses( + timeout=self.turn_response_timeout_seconds, + terminal_grace_seconds=self.terminal_response_grace_seconds, ) - yield from responses - if completed and self._may_have_delayed_terminal(responses): - yield from stream_session.wait_for_terminal_responses( - timeout=self.terminal_response_grace_seconds - ) @staticmethod - def _may_have_delayed_terminal(responses: list[Dict[str, Any]]) -> bool: - """Return whether a completed prompt likely precedes CES EndSession.""" - response_text = " ".join( - str(response.get("text", "")).lower() - for response in responses - if isinstance(response, dict) - ) + def may_have_delayed_terminal(response_text: str) -> bool: + """Return whether streamed announcement text likely precedes EndSession.""" + response_text = str(response_text or "").lower() terminal_cues = ( "goodbye", "great day", @@ -1349,7 +1376,7 @@ def get_available_agents(self) -> list: def start_conversation( self, conversation_id: str, request_data: Dict[str, Any] - ) -> Union[Dict[str, Any], list[Dict[str, Any]]]: + ) -> Generator[Dict[str, Any], None, None]: self.logger.info(f"[GECX] Starting conversation: {conversation_id}") try: session_id = _make_ces_session_id() @@ -1367,75 +1394,16 @@ def start_conversation( with self.sessions_lock: self.streaming_sessions[conversation_id] = stream_session - welcome_text = "Connected to GECX agent" - got_text = False - welcome_audio = b"" - ordered_greeting_responses: list[Dict[str, Any]] = [] - terminal_response: Optional[Dict[str, Any]] = None - _, greeting_responses = stream_session.wait_for_turn_responses( - timeout=8.0, terminate_on_timeout=False - ) - for response in greeting_responses: - if response.get("message_type") in {"transfer", "session_end"}: - terminal_response = response - else: - ordered_greeting_responses.append(response) - if response.get("text") and not got_text: - welcome_text = response["text"] - got_text = True - # Each audio response is already a complete WAV clip for one - # turn; keep the latest non-empty one for the greeting. - if response.get("audio_content"): - welcome_audio = response["audio_content"] - - # If the greeting turn had not completed within the wait window, the - # buffered frames were not yet flushed. Flush and pick them up so the - # caller still hears the greeting. - if not welcome_audio and stream_session._flush_audio_buffer(): - for response in stream_session.drain_responses(): - if response.get("message_type") in {"transfer", "session_end"}: - terminal_response = response - else: - ordered_greeting_responses.append(response) - if response.get("audio_content"): - welcome_audio = response["audio_content"] - - # CES may legitimately end or escalate during the initial turn. - # Preserve output and the terminal decision as ordered responses. - # WxCC skips prompt audio when TRANSFER_TO_AGENT shares its response. - if terminal_response is not None: - terminal_response = dict(terminal_response) - if terminal_response.get("text") or terminal_response.get( - "audio_content" - ): - terminal_audio = terminal_response.get("audio_content", b"") - ordered_greeting_responses.append( - self.create_response( - conversation_id=conversation_id, - message_type=( - "audio" if terminal_audio else "agent_response" - ), - text=terminal_response.get("text", ""), - audio_content=terminal_audio, - barge_in_enabled=False, - response_type="final", - ) - ) - terminal_response["text"] = "" - terminal_response["audio_content"] = b"" - return [*ordered_greeting_responses, terminal_response] - - return self.create_session_start_response( - conversation_id=conversation_id, - text=welcome_text, - audio_content=welcome_audio, + yield from stream_session.iter_turn_responses( + timeout=self.turn_response_timeout_seconds, + terminal_grace_seconds=self.terminal_response_grace_seconds, ) except Exception as exc: self.logger.error( f"[GECX] Error starting conversation {conversation_id}: {exc}", exc_info=True, ) - return self.create_error_response( + yield self.create_error_response( conversation_id=conversation_id, error_message=f"Failed to start GECX conversation: {exc}", ) @@ -1535,10 +1503,10 @@ def _handle_text_input( if not stream_session.enqueue_text(text): yield from stream_session.drain_responses() return - _, responses = stream_session.wait_for_turn_responses( - timeout=self.turn_response_timeout_seconds + yield from stream_session.iter_turn_responses( + timeout=self.turn_response_timeout_seconds, + terminal_grace_seconds=self.terminal_response_grace_seconds, ) - yield from responses def _handle_event_input( self, @@ -1557,10 +1525,10 @@ def _handle_event_input( if not stream_session.enqueue_event(event_name): yield from stream_session.drain_responses() return - _, responses = stream_session.wait_for_turn_responses( - timeout=self.turn_response_timeout_seconds + yield from stream_session.iter_turn_responses( + timeout=self.turn_response_timeout_seconds, + terminal_grace_seconds=self.terminal_response_grace_seconds, ) - yield from responses def end_conversation( self, conversation_id: str, message_data: Optional[Dict[str, Any]] = None diff --git a/src/core/README.md b/src/core/README.md index b7bd2fc..7248d21 100644 --- a/src/core/README.md +++ b/src/core/README.md @@ -43,6 +43,8 @@ The gRPC server implements the `VoiceVirtualAgentServicer` interface: **Key Features**: - Bidirectional streaming for real-time voice communication +- Independent bounded ingress/response queues and incremental connector + iterator consumption, including raw BYOVA `CHUNK` output - Session lifecycle tracking (start, message, end events) - Automatic session cleanup on stream termination - Error handling and logging diff --git a/src/core/wxcc_gateway_server.py b/src/core/wxcc_gateway_server.py index 6adc8f9..cf9ede5 100644 --- a/src/core/wxcc_gateway_server.py +++ b/src/core/wxcc_gateway_server.py @@ -305,7 +305,6 @@ def _start_conversation(self) -> Iterator[VoiceVAResponse]: if is_iterator: yield from self._iter_grpc_connector_responses( connector_response, - delay_terminal_after_audio=True, ) return @@ -454,10 +453,16 @@ def _process_speech_boundary( self.virtual_agent_id ) ) + if coalesce_speech_end: + # END_OF_INPUT begins the streamed response turn and therefore + # remains open through the following audio CHUNKs. START_OF_INPUT + # stays a standalone FINAL event so WxCC continues forwarding + # caller audio during the bounded speech-resume grace window. + boundary_connector_response["response_type"] = "chunk" if not coalesce_speech_end: response = self._convert_connector_response_to_grpc( - boundary_connector_response + boundary_connector_response, ) if response is not None: yield response @@ -483,39 +488,18 @@ def _process_speech_boundary( ) return - is_iterator = ( - hasattr(boundary_response, "__iter__") - and not isinstance(boundary_response, (dict, str, bytes)) - ) - if is_iterator: - boundary_responses = list(boundary_response) - elif boundary_response is None: - boundary_responses = [] - else: - boundary_responses = [boundary_response] - - has_terminal_response = any( - isinstance(response, dict) - and response.get("message_type") in {"transfer", "session_end"} - for response in boundary_responses - ) - response_kind = "terminal" if has_terminal_response else "normal" self.logger.info( - "Emitting END_OF_INPUT before %d %s connector response(s) for " + "Emitting END_OF_INPUT before the connector response stream for " "conversation %s", - len(boundary_responses), - response_kind, self.conversation_id, ) response = self._convert_connector_response_to_grpc( - boundary_connector_response + boundary_connector_response, + response_type=VoiceVAResponse.ResponseType.CHUNK, ) if response is not None: yield response - yield from self._iter_grpc_connector_responses( - boundary_responses, - delay_terminal_after_audio=has_terminal_response, - ) + yield from self._iter_grpc_connector_responses(boundary_response) def _iter_grpc_connector_responses( self, @@ -867,6 +851,11 @@ def _convert_connector_response_to_grpc( self.logger.debug( "Detected START_OF_INPUT event, setting minimal input_handling_config" ) + va_response.response_type = ( + response_type + if response_type is not None + else VoiceVAResponse.ResponseType.FINAL + ) # Set minimal input_handling_config for START_OF_INPUT events va_response.input_handling_config.CopyFrom( InputHandlingConfig( diff --git a/tests/README.md b/tests/README.md index a8ed2b5..859def1 100644 --- a/tests/README.md +++ b/tests/README.md @@ -163,7 +163,9 @@ The `test_gecx_connector.py` file covers: - CES session and deployment resource construction - `BidiRunSession` request ordering and initial text -- Agent text and WAV-wrapped audio response mapping +- Immediate raw mu-law `CHUNK` mapping before CES turn completion +- Exactly one normal or terminal `FINAL` after ordered audio chunks +- Initial-greeting, transfer-announcement, and session-end streaming order - WxCC encoding and sample-rate metadata handling - Normal `EndSession` and human-transfer metadata - CES stream half-close behavior after `EndSession` diff --git a/tests/test_gecx_connector.py b/tests/test_gecx_connector.py index a08486e..ec8bc46 100644 --- a/tests/test_gecx_connector.py +++ b/tests/test_gecx_connector.py @@ -84,6 +84,23 @@ def test_default_endpointing_silence_has_margin_above_one_second( assert connector.endpointing_silence_ms == 2000 + @pytest.mark.parametrize( + ("encoding", "sample_rate"), + [("LINEAR16", 8000), ("MULAW", 16000)], + ) + def test_rejects_output_format_that_wxcc_chunk_stream_cannot_play( + self, gecx_config, encoding, sample_rate + ): + gecx_config["output_audio_encoding"] = encoding + gecx_config["output_sample_rate_hertz"] = sample_rate + + with patch("src.connectors.gecx_connector.ces_v1.SessionServiceClient"): + with pytest.raises( + ValueError, + match="CHUNK streaming currently requires", + ): + GECXConnector(gecx_config) + class TestRequestGenerator: def test_first_message_is_session_config(self, connector): @@ -254,7 +271,7 @@ def test_committed_pause_preserves_resumed_onset_for_next_turn( class TestServerMessageMapping: - def test_session_output_maps_to_connector_responses(self, connector): + def test_session_output_streams_raw_chunk_then_one_final(self, connector): session = GECXStreamingSession( connector=connector, conversation_id="conv-1", @@ -277,17 +294,67 @@ def test_session_output_maps_to_connector_responses(self, connector): session.begin_input_turn() session._handle_server_message(message) - completed, responses = session.wait_for_turn_responses(timeout=0.1) + responses = list(session.iter_turn_responses(timeout=0.1)) - assert completed - assert len(responses) == 1 + assert session._turn_completed.is_set() + assert len(responses) == 2 assert responses[0]["message_type"] == "audio" - assert responses[0]["text"] == "Hi there" - assert responses[0]["response_type"] == "final" - # Text and audio are emitted atomically so WxCC does not synthesize a - # duplicate text-only prompt before playing the CES audio. - assert responses[0]["audio_content"].startswith(b"RIFF") - assert responses[0]["audio_content"].endswith(b"\x01\x02") + assert responses[0]["text"] == "" + assert responses[0]["response_type"] == "chunk" + assert responses[0]["audio_content"] == b"\x01\x02" + assert responses[1]["message_type"] == "silence" + assert responses[1]["response_type"] == "final" + assert responses[1]["audio_content"] == b"" + + def test_first_audio_chunk_is_available_before_ces_turn_completion( + self, connector + ): + session = GECXStreamingSession( + connector=connector, + conversation_id="conv-1", + session_path="projects/p/locations/us/apps/a/sessions/s1", + deployment_path=connector.deployment_path, + ) + session.begin_input_turn() + stream = session.iter_turn_responses(timeout=1.0) + + session._handle_server_message( + SimpleNamespace( + recognition_result=None, + interruption_signal=None, + end_session=None, + go_away=None, + session_output=SimpleNamespace( + text="", + audio=b"first frame", + turn_completed=False, + end_session=False, + ), + ) + ) + + assert session._turn_completed.is_set() is False + first_response = next(stream) + assert first_response["response_type"] == "chunk" + assert first_response["audio_content"] == b"first frame" + + session._handle_server_message( + SimpleNamespace( + recognition_result=None, + interruption_signal=None, + end_session=None, + go_away=None, + session_output=SimpleNamespace( + text="", + audio=b"", + turn_completed=True, + end_session=False, + ), + ) + ) + remaining = list(stream) + + assert [response["response_type"] for response in remaining] == ["final"] def test_end_session_emits_session_end_event(self, connector): session = GECXStreamingSession( @@ -397,13 +464,13 @@ def test_session_output_end_session_uses_nested_metadata(self, connector): "audio", "transfer", ] - assert responses[0]["text"] == ( - "Certainly. Let me connect you with a hotel specialist." - ) - assert responses[0]["audio_content"].endswith(b"transfer audio") + assert responses[0]["text"] == "" + assert responses[0]["response_type"] == "chunk" + assert responses[0]["audio_content"] == b"transfer audio" assert responses[0]["output_events"] == [] assert responses[1]["text"] == "" assert responses[1]["audio_content"] == b"" + assert responses[1]["response_type"] == "final" def test_configurable_escalation_alias_remains_supported(self, connector): connector.transfer_metadata_keys = ["custom_handoff"] @@ -479,36 +546,42 @@ def test_go_away_ends_session_without_reconnection(self, connector): assert session.terminal_decision.reason == GECXTerminalReason.GO_AWAY assert session.drain_responses()[0]["message_type"] == "session_end" - def test_initial_escalation_keeps_greeting_and_transfer_separate( + def test_initial_escalation_streams_greeting_before_transfer_final( self, connector ): - combined_terminal_response = connector.create_response( + greeting_chunk = connector.create_response( + conversation_id="conv-1", + message_type="audio", + audio_content=b"greeting", + barge_in_enabled=False, + response_type="chunk", + ) + terminal_response = connector.create_response( conversation_id="conv-1", message_type="transfer", - text="Let me connect you now.", - audio_content=b"RIFFgreeting", barge_in_enabled=False, response_type="final", ) stream_session = MagicMock() - stream_session.wait_for_turn_responses.return_value = ( - True, - [combined_terminal_response], + stream_session.iter_turn_responses.return_value = iter( + [greeting_chunk, terminal_response] ) with patch( "src.connectors.gecx_connector.GECXStreamingSession", return_value=stream_session, ): - responses = connector.start_conversation("conv-1", {}) + responses = list(connector.start_conversation("conv-1", {})) assert [response["message_type"] for response in responses] == [ "audio", "transfer", ] - assert responses[0]["audio_content"] == b"RIFFgreeting" + assert responses[0]["audio_content"] == b"greeting" + assert responses[0]["response_type"] == "chunk" assert responses[1]["audio_content"] == b"" assert responses[1]["text"] == "" + assert responses[1]["response_type"] == "final" class TestTerminalLifecycle: @@ -523,9 +596,8 @@ def _session(self, connector): def test_turn_timeout_decides_session_end(self, connector): session = self._session(connector) - completed, responses = session.wait_for_turn_responses(timeout=0) + responses = list(session.iter_turn_responses(timeout=0)) - assert not completed assert session.terminal_decision.reason == GECXTerminalReason.TIMEOUT assert [response["message_type"] for response in responses] == ["session_end"] @@ -621,15 +693,14 @@ def test_agent_output_before_terminal_is_preserved(self, connector): responses = session.drain_responses() assert [response["message_type"] for response in responses] == [ - "agent_response", - "session_end", + "session_end" ] assert responses[0]["text"] == "I can help with that" - assert responses[1]["text"] == "" + assert responses[0]["response_type"] == "final" - def test_buffered_agent_audio_before_terminal_is_preserved(self, connector): + def test_streamed_agent_audio_precedes_terminal_final(self, connector): session = self._session(connector) - session._buffer_active_audio(b"agent audio") + session._emit_active_audio_chunk(b"agent audio") session.terminate( GECXTerminalReason.GO_AWAY, @@ -642,7 +713,47 @@ def test_buffered_agent_audio_before_terminal_is_preserved(self, connector): "audio", "session_end", ] - assert responses[0]["audio_content"].endswith(b"agent audio") + assert responses[0]["audio_content"] == b"agent audio" + assert responses[0]["response_type"] == "chunk" + assert responses[1]["response_type"] == "final" + assert sum( + response["response_type"] == "final" for response in responses + ) == 1 + + def test_delayed_end_session_replaces_normal_final_after_streamed_audio( + self, connector + ): + session = self._session(connector) + session.begin_input_turn() + session._handle_server_message( + SimpleNamespace( + recognition_result=None, + interruption_signal=None, + end_session=None, + go_away=None, + session_output=SimpleNamespace( + text="Thank you. Have a great day.", + audio=b"goodbye audio", + turn_completed=True, + end_session=False, + ), + ) + ) + responses = session.iter_turn_responses( + timeout=1.0, + terminal_grace_seconds=1.0, + ) + + first_chunk = next(responses) + session._handle_end_session("conv-1", SimpleNamespace(metadata={})) + remaining = list(responses) + + assert first_chunk["response_type"] == "chunk" + assert first_chunk["audio_content"] == b"goodbye audio" + assert [response["message_type"] for response in remaining] == [ + "session_end" + ] + assert remaining[0]["response_type"] == "final" @pytest.mark.parametrize( ("gateway_reason", "terminal_reason"), @@ -766,17 +877,20 @@ def test_speech_started_resets_turn_completion(self, connector): assert responses == [] stream_session.begin_input_turn.assert_called_once_with() - def test_speech_ended_waits_for_and_yields_completed_turn(self, connector): + def test_speech_ended_yields_streamed_turn_responses(self, connector): stream_session = MagicMock() stream_session.is_terminal = False - expected = connector.create_response( + chunk = connector.create_response( conversation_id="conv-1", message_type="audio", - audio_content=b"RIFFaudio", + audio_content=b"raw audio", + response_type="chunk", + ) + final = connector.create_response( + conversation_id="conv-1", response_type="final", ) - stream_session.wait_for_turn_responses.return_value = (True, [expected]) - stream_session.wait_for_terminal_responses.return_value = [] + stream_session.iter_turn_responses.return_value = iter([chunk, final]) with patch.object(connector, "streaming_sessions", {"conv-1": stream_session}): responses = list( @@ -786,15 +900,17 @@ def test_speech_ended_waits_for_and_yields_completed_turn(self, connector): ) ) - assert responses == [expected] + assert responses == [chunk, final] stream_session.end_audio_turn.assert_called_once_with() - stream_session.wait_for_turn_responses.assert_called_once_with(timeout=30.0) - stream_session.wait_for_terminal_responses.assert_not_called() + stream_session.iter_turn_responses.assert_called_once_with( + timeout=30.0, + terminal_grace_seconds=3.0, + ) def test_precommitted_speech_end_only_waits_for_response(self, connector): stream_session = MagicMock() stream_session.is_terminal = False - stream_session.wait_for_turn_responses.return_value = (True, []) + stream_session.iter_turn_responses.return_value = iter([]) with patch.object(connector, "streaming_sessions", {"conv-1": stream_session}): responses = list( @@ -809,9 +925,12 @@ def test_precommitted_speech_end_only_waits_for_response(self, connector): assert responses == [] stream_session.end_audio_turn.assert_not_called() - stream_session.wait_for_turn_responses.assert_called_once_with(timeout=30.0) + stream_session.iter_turn_responses.assert_called_once_with( + timeout=30.0, + terminal_grace_seconds=3.0, + ) - def test_speech_ended_yields_delayed_terminal_after_agent_audio( + def test_speech_ended_yields_chunks_then_terminal_final( self, connector ): stream_session = MagicMock() @@ -819,17 +938,15 @@ def test_speech_ended_yields_delayed_terminal_after_agent_audio( audio = connector.create_response( conversation_id="conv-1", message_type="audio", - text="Alright. Have a great day.", - audio_content=b"RIFFaudio", - response_type="final", + audio_content=b"raw audio", + response_type="chunk", ) terminal = connector.create_response( conversation_id="conv-1", message_type="session_end", response_type="final", ) - stream_session.wait_for_turn_responses.return_value = (True, [audio]) - stream_session.wait_for_terminal_responses.return_value = [terminal] + stream_session.iter_turn_responses.return_value = iter([audio, terminal]) with patch.object(connector, "streaming_sessions", {"conv-1": stream_session}): responses = list( @@ -840,6 +957,10 @@ def test_speech_ended_yields_delayed_terminal_after_agent_audio( ) assert responses == [audio, terminal] - stream_session.wait_for_terminal_responses.assert_called_once_with( - timeout=3.0 + assert sum( + response["response_type"] == "final" for response in responses + ) == 1 + stream_session.iter_turn_responses.assert_called_once_with( + timeout=30.0, + terminal_grace_seconds=3.0, ) diff --git a/tests/test_wxcc_gateway_server.py b/tests/test_wxcc_gateway_server.py index aca8101..7f2b813 100644 --- a/tests/test_wxcc_gateway_server.py +++ b/tests/test_wxcc_gateway_server.py @@ -15,7 +15,11 @@ from src.core.wxcc_gateway_server import ConversationProcessor, WxCCGatewayServer from src.core.virtual_agent_router import VirtualAgentRouter from src.generated.byova_common_pb2 import EventInput -from src.generated.voicevirtualagent_pb2 import VoiceInput, VoiceVARequest +from src.generated.voicevirtualagent_pb2 import ( + VoiceInput, + VoiceVARequest, + VoiceVAResponse, +) from src.utils.silero_speech_boundary import SpeechBoundarySignal @@ -111,42 +115,37 @@ def test_process_audio_input_single_response(self, processor, mock_router, mock_ assert responses[0].prompts[0].text == "Hello, how can I help you?" assert responses[0].prompts[0].audio_content == b"audio_response_bytes" - def test_initial_escalation_plays_audio_before_transfer( + def test_initial_escalation_streams_chunk_before_transfer_final( self, processor, mock_router ): - wav_audio = bytearray(44 + 8000) - wav_audio[:4] = b"RIFF" - wav_audio[8:12] = b"WAVE" - wav_audio[28:32] = (8000).to_bytes(4, "little") - wav_audio[36:40] = b"data" - wav_audio[40:44] = (8000).to_bytes(4, "little") + raw_audio = b"\xff" * 800 audio_response = { "message_type": "audio", - "text": "Let me connect you now.", - "audio_content": bytes(wav_audio), + "text": "", + "audio_content": raw_audio, "barge_in_enabled": False, + "response_type": "chunk", } terminal_response = { "message_type": "transfer", "text": "", "audio_content": b"", "barge_in_enabled": False, + "response_type": "final", } mock_router.route_request.return_value = iter( [audio_response, terminal_response] ) - cancel_event = MagicMock() - cancel_event.wait.return_value = False - processor.set_stream_cancel_event(cancel_event) responses = list(processor._start_conversation()) assert len(responses) == 2 - assert responses[0].prompts[0].audio_content == bytes(wav_audio) + assert responses[0].response_type == VoiceVAResponse.ResponseType.CHUNK + assert responses[0].prompts[0].audio_content == raw_audio assert responses[0].output_events == [] + assert responses[1].response_type == VoiceVAResponse.ResponseType.FINAL assert responses[1].prompts == [] assert [event.event_type for event in responses[1].output_events] == [2] - cancel_event.wait.assert_called_once_with(1.0) def test_process_audio_input_uses_configured_rate_when_wxcc_omits_it(self, mock_router): processor = ConversationProcessor( @@ -170,17 +169,21 @@ def test_process_audio_input_uses_configured_rate_when_wxcc_omits_it(self, mock_ frame = processor.speech_boundary_observer.observe.call_args.args[0] assert frame.sample_rate_hertz == 16000 - def test_gateway_emits_speech_started_event(self, processor, mock_router, mock_audio_input): + def test_gateway_keeps_speech_started_as_standalone_final_event( + self, processor, mock_router, mock_audio_input + ): processor.speech_boundary_observer = MagicMock() processor.speech_boundary_observer.observe.return_value = [ SpeechBoundarySignal("speech_started", "test_conv_123", 8000) ] mock_router.route_request.return_value = None mock_router.should_observe_speech_boundaries.return_value = True + mock_router.should_coalesce_speech_end_with_response.return_value = True responses = list(processor._process_audio_input(mock_audio_input)) assert len(responses) == 1 + assert responses[0].response_type == VoiceVAResponse.ResponseType.FINAL assert responses[0].output_events[0].event_type == 4 assert responses[0].output_events[0].name == "" @@ -219,61 +222,112 @@ def test_gateway_flushes_lex_once_after_speech_ended( def test_gateway_sends_gecx_speech_end_before_normal_prompt( self, processor, mock_router, mock_audio_input ): - """Keep a normal GECX prompt separate from END_OF_INPUT.""" + """Keep END_OF_INPUT ahead of raw audio chunks and one normal FINAL.""" processor.speech_boundary_observer = MagicMock() processor.speech_boundary_observer.observe.return_value = [ SpeechBoundarySignal("speech_ended", "test_conv_123", 8000) ] - wav_audio = bytearray(44 + 8000) - wav_audio[:4] = b"RIFF" - wav_audio[8:12] = b"WAVE" - wav_audio[28:32] = (8000).to_bytes(4, "little") - wav_audio[36:40] = b"data" - wav_audio[40:44] = (8000).to_bytes(4, "little") - wav_audio = bytes(wav_audio) + raw_audio = b"\xff" * 800 audio_response = { "message_type": "audio", - "text": "What dates would you like to book?", - "audio_content": wav_audio, + "text": "", + "audio_content": raw_audio, + "barge_in_enabled": False, + "output_events": [], + "response_type": "chunk", + } + final_response = { + "message_type": "silence", + "text": "", + "audio_content": b"", "barge_in_enabled": False, "output_events": [], + "response_type": "final", } mock_router.route_request.side_effect = [ None, - iter([audio_response]), + iter([audio_response, final_response]), ] mock_router.should_observe_speech_boundaries.return_value = True mock_router.should_coalesce_speech_end_with_response.return_value = True responses = list(processor._process_audio_input(mock_audio_input)) - assert len(responses) == 2 + assert len(responses) == 3 assert responses[0].prompts == [] assert [event.event_type for event in responses[0].output_events] == [5] - assert responses[1].prompts[0].audio_content == wav_audio - assert responses[1].prompts[0].text == "What dates would you like to book?" + assert responses[0].response_type == VoiceVAResponse.ResponseType.CHUNK + assert responses[1].response_type == VoiceVAResponse.ResponseType.CHUNK + assert responses[1].prompts[0].audio_content == raw_audio + assert responses[1].prompts[0].text == "" assert responses[1].output_events == [] + assert responses[2].response_type == VoiceVAResponse.ResponseType.FINAL + assert responses[2].prompts == [] + assert sum( + response.response_type == VoiceVAResponse.ResponseType.FINAL + for response in responses + ) == 1 - def test_gateway_sends_gecx_speech_end_before_transfer_announcement( + def test_gateway_does_not_materialize_gecx_chunk_stream( + self, processor, mock_router + ): + raw_audio = b"\xff" * 800 + release_final = threading.Event() + + def connector_responses(): + yield { + "message_type": "audio", + "text": "", + "audio_content": raw_audio, + "barge_in_enabled": False, + "output_events": [], + "response_type": "chunk", + } + assert release_final.wait(1.0) + yield { + "message_type": "silence", + "text": "", + "audio_content": b"", + "barge_in_enabled": False, + "output_events": [], + "response_type": "final", + } + + mock_router.should_coalesce_speech_end_with_response.return_value = True + mock_router.route_request.return_value = connector_responses() + responses = processor._process_speech_boundary( + SpeechBoundarySignal("speech_ended", "test_conv_123", 8000) + ) + + end_of_input = next(responses) + first_chunk = next(responses) + + assert end_of_input.response_type == VoiceVAResponse.ResponseType.CHUNK + assert [event.event_type for event in end_of_input.output_events] == [5] + assert first_chunk.response_type == VoiceVAResponse.ResponseType.CHUNK + assert first_chunk.prompts[0].audio_content == raw_audio + + release_final.set() + final = next(responses) + assert final.response_type == VoiceVAResponse.ResponseType.FINAL + with pytest.raises(StopIteration): + next(responses) + + def test_gateway_streams_gecx_transfer_after_chunk_typed_speech_end( self, processor, mock_router, mock_audio_input ): processor.speech_boundary_observer = MagicMock() processor.speech_boundary_observer.observe.return_value = [ SpeechBoundarySignal("speech_ended", "test_conv_123", 8000) ] - wav_audio = bytearray(44 + 8000) - wav_audio[:4] = b"RIFF" - wav_audio[8:12] = b"WAVE" - wav_audio[28:32] = (8000).to_bytes(4, "little") - wav_audio[36:40] = b"data" - wav_audio[40:44] = (8000).to_bytes(4, "little") - wav_audio = bytes(wav_audio) + raw_audio = b"\xff" * 800 audio_response = { "message_type": "audio", - "text": "Let me connect you now.", - "audio_content": wav_audio, + "text": "", + "audio_content": raw_audio, "barge_in_enabled": False, "output_events": [], + "response_type": "chunk", } transfer_response = { "message_type": "transfer", @@ -281,6 +335,7 @@ def test_gateway_sends_gecx_speech_end_before_transfer_announcement( "audio_content": b"", "barge_in_enabled": False, "output_events": [], + "response_type": "final", } mock_router.route_request.side_effect = [ None, @@ -289,20 +344,19 @@ def test_gateway_sends_gecx_speech_end_before_transfer_announcement( mock_router.should_observe_speech_boundaries.return_value = True mock_router.should_coalesce_speech_end_with_response.return_value = True - cancel_event = MagicMock() - cancel_event.wait.return_value = False - processor.set_stream_cancel_event(cancel_event) responses = list(processor._process_audio_input(mock_audio_input)) assert len(responses) == 3 assert responses[0].prompts == [] assert [event.event_type for event in responses[0].output_events] == [5] - assert responses[1].prompts[0].audio_content == wav_audio - assert responses[1].prompts[0].text == "Let me connect you now." + assert responses[0].response_type == VoiceVAResponse.ResponseType.CHUNK + assert responses[1].response_type == VoiceVAResponse.ResponseType.CHUNK + assert responses[1].prompts[0].audio_content == raw_audio + assert responses[1].prompts[0].text == "" assert responses[1].output_events == [] + assert responses[2].response_type == VoiceVAResponse.ResponseType.FINAL assert responses[2].prompts == [] assert [event.event_type for event in responses[2].output_events] == [2] - cancel_event.wait.assert_called_once_with(1.0) def test_gateway_merges_speech_resumed_during_end_grace( self, processor, mock_router, mock_audio_input diff --git a/tools/byova_e2e/README.md b/tools/byova_e2e/README.md index 2e68cee..6674704 100644 --- a/tools/byova_e2e/README.md +++ b/tools/byova_e2e/README.md @@ -301,6 +301,10 @@ byova-e2e run --destination 9999 \ --config config/gecx-regression.spec.json \ --test normal-response +byova-e2e run --destination 9999 \ + --config config/gecx-regression.spec.json \ + --test long-response + byova-e2e run --destination 9999 \ --config config/gecx-regression.spec.json \ --test natural-pause @@ -325,13 +329,18 @@ byova-e2e run --destination 9999 \ Correlate each artifact window and config SHA-256 with gateway and CES evidence: - `normal-response`: one caller turn and no terminal output event. +- `long-response`: multiple ordered raw audio chunks, first audio before CES + turn completion, and exactly one normal `FINAL`. It uses a 2.5-second + remote-silence threshold so a natural pause inside streamed playback is not + mistaken for the end of the prompt. - `natural-pause`: one outward `START_OF_INPUT`, one merged resume, one outward `END_OF_INPUT`, and one complete CES transcript. - `multi-turn-response`: two caller injections, two complete GECX responses, - and no terminal output event. -- `speak-during-playback`: an audio-plane probe in Phase 2. The second caller + and no terminal output event. It uses the same 2.5-second streamed-prompt + completion threshold before injecting turn two. +- `speak-during-playback`: an audio-plane probe. The second caller injection begins after the first GECX response starts and the call remains - healthy through remote-audio completion. Buffered GECX prompts keep barge-in + healthy through remote-audio completion. Streamed GECX chunks keep barge-in disabled until Phase 4, so a green runner result does **not** prove CES received the second utterance; correlate the CES transcript and record that live semantic gap explicitly. diff --git a/tools/byova_e2e/config/gecx-regression.spec.json b/tools/byova_e2e/config/gecx-regression.spec.json index b17be6c..07807d0 100644 --- a/tools/byova_e2e/config/gecx-regression.spec.json +++ b/tools/byova_e2e/config/gecx-regression.spec.json @@ -32,6 +32,28 @@ } ] }, + { + "id": "long-response", + "title": "streams one detailed hotel response", + "description": "A detailed answer exercises multiple ordered BYOVA audio chunks before one final response.", + "use": { + "remoteSilenceMs": 2500 + }, + "steps": [ + { + "name": "Request a detailed hotel description", + "action": "speak", + "text": "Please give me a detailed description of all available hotel room types and amenities." + }, + { + "name": "Expect one complete streamed response", + "expect": { + "outcome": "response", + "responsePrompts": 1 + } + } + ] + }, { "id": "natural-pause", "title": "keeps a bounded natural pause in one caller turn", @@ -59,6 +81,9 @@ "id": "multi-turn-response", "title": "continues a hotel request across two caller turns", "description": "The gateway continues ingesting caller audio after delivering a complete GECX response.", + "use": { + "remoteSilenceMs": 2500 + }, "steps": [ { "name": "Start a hotel request", @@ -89,7 +114,7 @@ { "id": "speak-during-playback", "title": "probes caller injection while GECX audio is playing", - "description": "Audio-plane probe only: the browser injects during playback and verifies the call remains healthy. Because buffered GECX prompts keep barge-in disabled until Phase 4, a green runner result does not prove that CES received the second utterance; correlate the CES transcript.", + "description": "Audio-plane probe only: the browser injects during playback and verifies the call remains healthy. Because streamed GECX chunks keep barge-in disabled until Phase 4, a green runner result does not prove that CES received the second utterance; correlate the CES transcript.", "steps": [ { "name": "Request a detailed hotel description",