diff --git a/backend/.env.example b/backend/.env.example index e187c3e..975b109 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -27,3 +27,12 @@ TIMEFLOW_OPENAI_TIMEOUT_SECONDS=30 # Maximum serial Function Calling rounds allowed during one Agent turn. TIMEFLOW_AGENT_MAX_TOOL_ROUNDS=4 + +# Aliyun Qwen-Audio realtime (end-to-end: audio in, speech out) +# Endpoint is built as {WORKSPACE_ID}.{REGION}.maas.aliyuncs.com, so no URL to set. +# Leave API_KEY or WORKSPACE_ID empty and the stand-in agent is used instead. +TIMEFLOW_ALIYUN_AUDIO_API_KEY= +TIMEFLOW_ALIYUN_AUDIO_WORKSPACE_ID= +TIMEFLOW_ALIYUN_AUDIO_MODEL=qwen-audio-3.0-realtime-plus +TIMEFLOW_ALIYUN_AUDIO_REGION=cn-beijing +TIMEFLOW_ALIYUN_AUDIO_VOICE=longanqian diff --git a/backend/src/timeflow/infrastructure/external/realtime/__init__.py b/backend/src/timeflow/infrastructure/external/realtime/__init__.py new file mode 100644 index 0000000..1672cd1 --- /dev/null +++ b/backend/src/timeflow/infrastructure/external/realtime/__init__.py @@ -0,0 +1 @@ +"""Realtime speech model adapters.""" diff --git a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py new file mode 100644 index 0000000..32212db --- /dev/null +++ b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py @@ -0,0 +1,275 @@ +"""Aliyun Qwen-Audio realtime adapter: speaks the vendor's wire format, reports plainly.""" + +import asyncio +import base64 +import json +import logging +from dataclasses import dataclass +from typing import Any, Protocol + +logger = logging.getLogger(__name__) + +# Our protocol owns turn boundaries; the model must not also decide when a turn ended. +PUSH_TO_TALK: None = None + +# The model accepts 16 kHz mono PCM in and emits 24 kHz mono PCM out. +INPUT_SAMPLE_RATE_HZ = 16_000 +OUTPUT_SAMPLE_RATE_HZ = 24_000 + + +class Transport(Protocol): + """The subset of a WebSocket this adapter uses, so tests can supply their own.""" + + async def send(self, message: str) -> None: + """Send one text frame.""" + ... + + async def recv(self) -> str | bytes: + """Receive the next frame.""" + ... + + async def close(self) -> None: + """Close the connection.""" + ... + + +class Observer(Protocol): + """Where this adapter reports what the model says; restated, never imported.""" + + async def heard(self, text: str) -> None: + """The model reported what the user said.""" + ... + + async def spoke(self, text: str) -> None: + """The model reported the words it is saying.""" + ... + + async def audio(self, data: bytes) -> None: + """One chunk of the model's own speech, decoded to raw bytes.""" + ... + + async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: + """The model asked for a tool to run.""" + ... + + async def failed(self, message: str) -> None: + """The session cannot continue.""" + ... + + +@dataclass(frozen=True, slots=True) +class QwenAudioConfig: + """Where to reach the model and how to authenticate.""" + + api_key: str + workspace_id: str + model: str + region: str = "cn-beijing" + voice: str = "longanqian" + + def url(self) -> str: + """Build the region- and workspace-specific realtime endpoint.""" + host = f"{self.workspace_id}.{self.region}.maas.aliyuncs.com" + return f"wss://{host}/api-ws/v1/realtime?model={self.model}" + + def headers(self) -> dict[str, str]: + """Build the auth headers; the key never appears in logs or errors.""" + return {"Authorization": f"Bearer {self.api_key}"} + + +class QwenAudioSession: + """One turn's conversation with the model, translated to domain events.""" + + def __init__(self, transport: Transport, config: QwenAudioConfig) -> None: + """Store the open transport and the config it was opened with.""" + self._transport = transport + self._config = config + # A tool makes a turn two responses; the second one carries the audio. + self._open_responses = 0 + + async def configure(self, instructions: str, tools: list[dict[str, Any]]) -> None: + """Set the session up before any audio; turn_detection only takes effect here.""" + session: dict[str, Any] = { + "modalities": ["text", "audio"], + "voice": self._config.voice, + "input_audio_format": "pcm", + "output_audio_format": "pcm", + "turn_detection": PUSH_TO_TALK, + } + if instructions: + session["instructions"] = instructions + if tools: + session["tools"] = tools + await self._send({"type": "session.update", "session": session}) + + async def send_audio(self, chunk: bytes) -> None: + """Append one chunk of the user's speech, base64 encoded as the vendor expects.""" + await self._send( + { + "type": "input_audio_buffer.append", + "audio": base64.b64encode(chunk).decode("ascii"), + } + ) + + async def finish_input(self) -> None: + """Commit the buffered audio and ask for a reply.""" + await self._send({"type": "input_audio_buffer.commit"}) + await self._send({"type": "response.create"}) + self._open_responses += 1 + + async def send_tool_result(self, call_id: str, output: str) -> None: + """Write a tool's output back and let the model continue from it.""" + await self._send( + { + "type": "conversation.item.create", + "item": {"type": "function_call_output", "call_id": call_id, "output": output}, + } + ) + await self._send({"type": "response.create"}) + self._open_responses += 1 + + async def close(self) -> None: + """Close the underlying connection, ignoring an already-closed one.""" + try: + await self._transport.close() + except Exception: # noqa: BLE001 - closing must not mask the original outcome + logger.debug("closing a realtime session that was already gone") + + async def _send(self, event: dict[str, Any]) -> None: + """Serialize and send one client event.""" + await self._transport.send(json.dumps(event, ensure_ascii=False)) + + async def pump(self, observer: Observer) -> None: + """Report what the model says, decoded and renamed, until the turn ends or fails.""" + spoken = "" + while True: + try: + raw = await self._transport.recv() + except Exception as error: # noqa: BLE001 - any transport failure ends the turn + await observer.failed(f"realtime transport failed: {type(error).__name__}") + return + + if isinstance(raw, bytes): + # The vendor sends everything as JSON text; a binary frame is unexpected. + continue + try: + event = json.loads(raw) + except json.JSONDecodeError: + await observer.failed("realtime session sent a non-JSON frame") + return + if not isinstance(event, dict): + await observer.failed("realtime session sent a non-object frame") + return + + kind = event.get("type") + + if kind == "conversation.item.input_audio_transcription.completed": + await observer.heard(str(event.get("transcript", ""))) + elif kind == "response.audio_transcript.delta": + spoken += str(event.get("delta", "")) + await observer.spoke(spoken) + elif kind == "response.audio_transcript.done": + # Reported again in case the reply was short enough to skip increments. + final = str(event.get("transcript", "")) + if final and final != spoken: + spoken = final + await observer.spoke(spoken) + elif kind == "response.audio.delta": + decoded = _decode_audio(event.get("delta")) + if decoded: + await observer.audio(decoded) + elif kind == "response.function_call_arguments.done": + requested = _tool_request(event) + if requested is None: + await observer.failed("realtime session sent an unusable tool call") + return + await observer.tool_requested(**requested) + elif kind == "response.done": + # Not the turn's end if a tool ran: the next response is the one that speaks. + self._open_responses -= 1 + if self._open_responses <= 0: + return + elif kind == "error": + await observer.failed(_error_message(event)) + return + + +def _decode_audio(delta: Any) -> bytes: + """Decode one base64 audio delta, dropping a malformed one rather than failing the turn.""" + if not isinstance(delta, str) or not delta: + return b"" + try: + return base64.b64decode(delta, validate=True) + except (ValueError, TypeError): + logger.warning("dropped a malformed audio delta from the realtime session") + return b"" + + +def _tool_request(event: dict[str, Any]) -> dict[str, Any] | None: + """Lift a tool call out of a vendor event, or None when it cannot be acted on.""" + call_id = event.get("call_id") + name = event.get("name") + if not isinstance(call_id, str) or not isinstance(name, str): + return None + raw_arguments = event.get("arguments") + arguments: dict[str, Any] = {} + if isinstance(raw_arguments, str) and raw_arguments: + try: + parsed = json.loads(raw_arguments) + except json.JSONDecodeError: + logger.warning("realtime session sent unparsable tool arguments") + return None + if isinstance(parsed, dict): + arguments = parsed + return {"call_id": call_id, "name": name, "arguments": arguments} + + +def _error_message(event: dict[str, Any]) -> str: + """Extract a readable message from a vendor error event.""" + error = event.get("error") + if isinstance(error, dict): + message = error.get("message") + if isinstance(message, str) and message: + return message + return "realtime session reported an error" + + +class QwenAudioSessionFactory: + """Open one configured session per turn.""" + + def __init__( + self, + config: QwenAudioConfig, + *, + connect: Any = None, + open_timeout_seconds: float = 10.0, + ) -> None: + """Store the config plus the connect seam tests replace.""" + self._config = config + self._connect = connect + self._open_timeout_seconds = open_timeout_seconds + + async def open(self, instructions: str, tools: list[dict[str, Any]]) -> QwenAudioSession: + """Connect, configure, and return a session ready for audio. + Closes on failure: a socket the caller never receives is one nobody can close. + """ + connect = self._connect or _default_connect + async with asyncio.timeout(self._open_timeout_seconds): + transport = await connect(self._config) + session = QwenAudioSession(transport, self._config) + try: + await session.configure(instructions, tools) + except BaseException: + await session.close() + raise + return session + + +async def _default_connect(config: QwenAudioConfig) -> Transport: + """Open a real WebSocket to the vendor endpoint.""" + import websockets + + connection = await websockets.connect( + config.url(), additional_headers=config.headers(), max_size=None + ) + return connection diff --git a/backend/src/timeflow/infrastructure/settings.py b/backend/src/timeflow/infrastructure/settings.py index 3cd7ee1..c8f4b06 100644 --- a/backend/src/timeflow/infrastructure/settings.py +++ b/backend/src/timeflow/infrastructure/settings.py @@ -32,6 +32,13 @@ class Settings: openai_model: str = "qwen-flash" openai_timeout_seconds: float = 30.0 agent_max_tool_rounds: int = 4 + # Named beside aliyun_asr_* rather than realtime_*: both are Aliyun realtime services, + # so "realtime" alone does not say which. This one takes audio in and gives speech back. + aliyun_audio_api_key: str = "" + aliyun_audio_workspace_id: str = "" + aliyun_audio_model: str = "qwen-audio-3.0-realtime-plus" + aliyun_audio_region: str = "cn-beijing" + aliyun_audio_voice: str = "longanqian" @classmethod def from_environment(cls, env_file: Path | str = ".env") -> "Settings": @@ -98,8 +105,24 @@ def from_environment(cls, env_file: Path | str = ".env") -> "Settings": openai_model=environ.get("TIMEFLOW_OPENAI_MODEL", "qwen-flash"), openai_timeout_seconds=openai_timeout_seconds, agent_max_tool_rounds=agent_max_tool_rounds, + aliyun_audio_api_key=environ.get("TIMEFLOW_ALIYUN_AUDIO_API_KEY", ""), + aliyun_audio_workspace_id=environ.get("TIMEFLOW_ALIYUN_AUDIO_WORKSPACE_ID", ""), + aliyun_audio_model=environ.get( + "TIMEFLOW_ALIYUN_AUDIO_MODEL", + "qwen-audio-3.0-realtime-plus", + ), + aliyun_audio_region=environ.get("TIMEFLOW_ALIYUN_AUDIO_REGION", "cn-beijing"), + aliyun_audio_voice=environ.get("TIMEFLOW_ALIYUN_AUDIO_VOICE", "longanqian"), ) + def aliyun_audio_is_configured(self) -> bool: + """Report whether the end-to-end audio model can be reached. + + Only the two secrets are checked: the rest have working defaults, so a deployment + that sets just these gets a working model rather than a puzzling half-configured one. + """ + return bool(self.aliyun_audio_api_key and self.aliyun_audio_workspace_id) + @lru_cache def get_settings() -> Settings: diff --git a/backend/src/timeflow/intelligence/realtime/__init__.py b/backend/src/timeflow/intelligence/realtime/__init__.py new file mode 100644 index 0000000..c9fb23f --- /dev/null +++ b/backend/src/timeflow/intelligence/realtime/__init__.py @@ -0,0 +1 @@ +"""End-to-end realtime model: audio in, speech and tool calls out.""" diff --git a/backend/src/timeflow/intelligence/realtime/agent.py b/backend/src/timeflow/intelligence/realtime/agent.py new file mode 100644 index 0000000..64f903c --- /dev/null +++ b/backend/src/timeflow/intelligence/realtime/agent.py @@ -0,0 +1,186 @@ +"""Agent that hands one turn to a realtime speech model and reports what it says.""" + +import asyncio +import contextlib +import logging +from collections.abc import AsyncIterator, Callable +from typing import Any +from uuid import uuid4 + +from timeflow.intelligence.ports import AudioReply, ReplyText, ResultSink, StreamInfo +from timeflow.intelligence.ports import Transcript as HeardSpeech +from timeflow.intelligence.realtime.ports import RealtimeSessionFactory + +logger = logging.getLogger(__name__) + +# The model emits 24 kHz mono PCM; the client is told so in voice.tts.start. +REPLY_SAMPLE_RATE_HZ = 24_000 +REPLY_AUDIO_FORMAT = "pcm" + +# Only answers exist this round; the round that adds questions sets this per turn. +REPLY_PURPOSE = "command_result" + +# Language is reported as the model's own, not detected here. +ASSUMED_LANGUAGE = "zh" + +# 16 kHz mono 16-bit: 32 bytes per millisecond. +_INPUT_BYTES_PER_MS = 32 + + +def new_audio_id() -> str: + """Return a fresh identifier for one spoken reply.""" + return f"audio_{uuid4().hex}" + + +def new_reply_id() -> str: + """Return a fresh identifier tying one reply's wording updates together.""" + return f"reply_{uuid4().hex}" + + +class RealtimeAgent: + """Feed one audio stream to a realtime model and push back what comes out.""" + + def __init__( + self, + sessions: RealtimeSessionFactory, + result_sink: ResultSink, + *, + instructions: Callable[[], str] | None = None, + audio_id_factory: Callable[[], str] | None = None, + reply_id_factory: Callable[[], str] | None = None, + ) -> None: + """Store the session source, the sink, and the id seams.""" + self._sessions = sessions + self._result_sink = result_sink + # Called per turn, so a long-running server keeps saying what day it is now. + self._instructions = instructions or (lambda: "") + self._audio_id_factory = audio_id_factory or new_audio_id + self._reply_id_factory = reply_id_factory or new_reply_id + + async def handle_audio(self, chunks: AsyncIterator[bytes], stream: StreamInfo) -> None: + """Run one turn: send the audio, then push the transcript, wording and speech.""" + try: + session = await self._sessions.open(self._instructions(), []) + except Exception: + logger.exception("could not open a realtime session") + return + + turn = _Turn( + self._result_sink, + stream, + self._audio_id_factory(), + self._reply_id_factory(), + ) + pumping = asyncio.create_task(session.pump(turn)) + try: + sent_bytes = 0 + async for chunk in chunks: + await session.send_audio(chunk) + sent_bytes += len(chunk) + await session.finish_input() + turn.note_input(sent_bytes) + await pumping + except BaseException: + # Cancelled on every exit, not just the caller's: an orphaned pump sits in recv(). + pumping.cancel() + with contextlib.suppress(BaseException): + await pumping + raise + finally: + await turn.close() + await session.close() + + +class _Turn: + """One turn's model output, translated into ResultSink calls as it arrives.""" + + def __init__( + self, + result_sink: ResultSink, + stream: StreamInfo, + audio_id: str, + reply_id: str, + ) -> None: + """Start a turn that has heard nothing and said nothing.""" + self._result_sink = result_sink + self._stream = stream + self._audio_id = audio_id + self._reply_id = reply_id + self._spoken = "" + self._input_bytes = 0 + self._audio: asyncio.Queue[bytes | None] = asyncio.Queue() + self._speaking: asyncio.Task[None] | None = None + + def note_input(self, sent_bytes: int) -> None: + """Record how much audio the user sent, for the transcript's duration.""" + self._input_bytes = sent_bytes + + async def heard(self, text: str) -> None: + """Push what the user was heard to say.""" + if not text: + logger.info("realtime model returned an empty transcript") + return + await self._result_sink.deliver_transcript( + HeardSpeech( + text=text, + language=ASSUMED_LANGUAGE, + duration_ms=self._input_bytes // _INPUT_BYTES_PER_MS, + ), + self._stream, + ) + + async def spoke(self, text: str) -> None: + """Push the reply's wording so far, and keep it for the audio's opening message.""" + self._spoken = text + await self._result_sink.deliver_reply_text( + ReplyText(reply_id=self._reply_id, speech_text=text), self._stream + ) + + async def audio(self, data: bytes) -> None: + """Queue one chunk, starting the delivery on the first one.""" + if self._speaking is None: + self._speaking = asyncio.create_task(self._speak()) + await self._audio.put(data) + + async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: + """Note that a tool was asked for while none are offered, and answer nothing.""" + del arguments + logger.warning( + "realtime model asked for a tool while none are registered", + extra={"call_id": call_id, "tool": name}, + ) + + async def failed(self, message: str) -> None: + """Record that the model could not finish this turn.""" + logger.warning("realtime session failed", extra={"reason": message}) + + async def close(self) -> None: + """Settle the wording, then finish the audio -- in that order; see close's test.""" + if self._spoken: + await self._result_sink.deliver_reply_text( + ReplyText(reply_id=self._reply_id, speech_text=self._spoken, done=True), + self._stream, + ) + if self._speaking is None: + return + await self._audio.put(None) + await self._speaking + + async def _speak(self) -> None: + """Hand the queued audio to the sink as one continuous reply.""" + reply = AudioReply( + audio_id=self._audio_id, + audio_format=REPLY_AUDIO_FORMAT, + sample_rate_hz=REPLY_SAMPLE_RATE_HZ, + purpose=REPLY_PURPOSE, + speech_text=self._spoken, + ) + await self._result_sink.deliver_audio(reply, self._drain(), self._stream) + + async def _drain(self) -> AsyncIterator[bytes]: + """Yield queued chunks until the reply is closed.""" + while True: + chunk = await self._audio.get() + if chunk is None: + return + yield chunk diff --git a/backend/src/timeflow/intelligence/realtime/ports.py b/backend/src/timeflow/intelligence/realtime/ports.py new file mode 100644 index 0000000..9be716a --- /dev/null +++ b/backend/src/timeflow/intelligence/realtime/ports.py @@ -0,0 +1,59 @@ +"""What the dialogue layer needs from a realtime speech model, on its own terms.""" + +from typing import Any, Protocol + + +class TurnObserver(Protocol): + """What a realtime session reports while a turn runs, in this layer's own terms.""" + + async def heard(self, text: str) -> None: + """The model reported what the user said.""" + ... + + async def spoke(self, text: str) -> None: + """The model reported the words it is saying.""" + ... + + async def audio(self, data: bytes) -> None: + """One chunk of the model's own speech, already decoded to raw bytes.""" + ... + + async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: + """The model asked for a tool to run before it continues.""" + ... + + async def failed(self, message: str) -> None: + """The session cannot continue.""" + ... + + +class RealtimeSession(Protocol): + """One open conversation with a realtime speech model.""" + + async def send_audio(self, chunk: bytes) -> None: + """Hand one chunk of the user's speech to the model.""" + ... + + async def finish_input(self) -> None: + """Tell the model the user stopped talking and a reply is wanted.""" + ... + + async def send_tool_result(self, call_id: str, output: str) -> None: + """Return a tool's output and let the model continue from it.""" + ... + + async def pump(self, observer: TurnObserver) -> None: + """Report what the model says until the turn ends or the session fails.""" + ... + + async def close(self) -> None: + """Release the session.""" + ... + + +class RealtimeSessionFactory(Protocol): + """Open a session per turn, so one failure never poisons the next.""" + + async def open(self, instructions: str, tools: list[dict[str, Any]]) -> RealtimeSession: + """Connect and configure a session; raises when the model is unreachable.""" + ... diff --git a/backend/src/timeflow/main.py b/backend/src/timeflow/main.py index 75e00fa..736a5c5 100644 --- a/backend/src/timeflow/main.py +++ b/backend/src/timeflow/main.py @@ -1,8 +1,11 @@ """FastAPI application composition root.""" +import logging + from fastapi import FastAPI, WebSocket from timeflow.business.health import HealthService +from timeflow.gateway.websocket.agent_ports import Agent from timeflow.gateway.websocket.connection_manager import ConnectionManager from timeflow.gateway.websocket.endpoint import ( UnauthenticatedConnectionLimiter, @@ -15,9 +18,16 @@ from timeflow.gateway.websocket.handlers.voice_stream import VoiceStreamHandlers from timeflow.gateway.websocket.ports import AudioSink, TokenVerifier from timeflow.gateway.websocket.router import MessageRouter +from timeflow.infrastructure.external.realtime.qwen_audio import ( + QwenAudioConfig, + QwenAudioSessionFactory, +) from timeflow.infrastructure.security.token_verifier import FakeTokenVerifier -from timeflow.infrastructure.settings import get_settings +from timeflow.infrastructure.settings import Settings, get_settings from timeflow.intelligence.fake_agent import FakeAgent +from timeflow.intelligence.realtime.agent import RealtimeAgent + +logger = logging.getLogger(__name__) def create_app( @@ -46,16 +56,7 @@ def create_app( limiter = UnauthenticatedConnectionLimiter(settings.ws_max_unauthenticated_connections) if audio_sink is None: - # Fail closed like the verifier above: the stand-in agent reports commands as - # applied that were never carried out. - if settings.environment != "development": - raise RuntimeError( - "No AudioSink was injected and the stand-in agent is development-only; " - f"TIMEFLOW_ENVIRONMENT is {settings.environment!r}. " - "It reports commands as applied that were never carried out. " - "Inject a real sink before exposing /ws." - ) - audio_sink = AgentAudioSink(FakeAgent(WebSocketResultSink(connections))) + audio_sink = AgentAudioSink(_build_agent(settings, WebSocketResultSink(connections))) voice_streams = VoiceStreamHandlers( audio_sink, @@ -89,4 +90,40 @@ async def websocket_session(websocket: WebSocket) -> None: return application +def _build_agent(settings: Settings, result_sink: WebSocketResultSink) -> Agent: + """Return the realtime agent when it is configured, otherwise the stand-in. + + Fails closed on the stand-in rather than on the absence of credentials: a configured + deployment is a real one and should start, while falling back outside development + would leave a server reporting commands as applied that were never carried out. + """ + if settings.aliyun_audio_is_configured(): + logger.info("using the realtime model", extra={"model": settings.aliyun_audio_model}) + return RealtimeAgent( + QwenAudioSessionFactory( + QwenAudioConfig( + api_key=settings.aliyun_audio_api_key, + workspace_id=settings.aliyun_audio_workspace_id, + model=settings.aliyun_audio_model, + region=settings.aliyun_audio_region, + voice=settings.aliyun_audio_voice, + ) + ), + result_sink, + ) + + if settings.environment != "development": + raise RuntimeError( + "The realtime model is not configured and the stand-in agent is " + f"development-only; TIMEFLOW_ENVIRONMENT is {settings.environment!r}. " + "Set TIMEFLOW_ALIYUN_AUDIO_API_KEY and TIMEFLOW_ALIYUN_AUDIO_WORKSPACE_ID, " + "or inject an AudioSink, before exposing /ws." + ) + logger.info( + "realtime model not configured, using the stand-in agent", + extra={"needs": "TIMEFLOW_ALIYUN_AUDIO_API_KEY and TIMEFLOW_ALIYUN_AUDIO_WORKSPACE_ID"}, + ) + return FakeAgent(result_sink) + + app = create_app() diff --git a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py new file mode 100644 index 0000000..e2be11f --- /dev/null +++ b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py @@ -0,0 +1,502 @@ +"""Translating the vendor's realtime wire format, with a fake transport.""" + +import asyncio +import base64 +import json +from typing import Any + +from timeflow.infrastructure.external.realtime.qwen_audio import ( + Observer, + QwenAudioConfig, + QwenAudioSession, + QwenAudioSessionFactory, +) +from timeflow.intelligence.realtime.ports import ( + RealtimeSession, + RealtimeSessionFactory, + TurnObserver, +) + +CONFIG = QwenAudioConfig( + api_key="key-abc", workspace_id="ws_001", model="qwen-audio-3.0-realtime-plus" +) + + +class FakeTransport: + """A stand-in socket: records what was sent, replays a scripted server side.""" + + def __init__(self, *inbound: str) -> None: + """Queue the frames the server will send, in order.""" + self.sent: list[dict[str, Any]] = [] + self.closed = False + self._inbound = list(inbound) + + async def send(self, message: str) -> None: + """Record one client event.""" + self.sent.append(json.loads(message)) + + async def recv(self) -> str: + """Return the next scripted frame, raising once the script runs out. + + Raising rather than blocking: reading past a turn fails now, not on CI timeout. + """ + if not self._inbound: + raise AssertionError("the pump read past the end of the scripted turn") + return self._inbound.pop(0) + + async def close(self) -> None: + """Mark the connection closed.""" + self.closed = True + + def types(self) -> list[str]: + """Return the type of every client event sent, in order.""" + return [str(event["type"]) for event in self.sent] + + +class RecordingObserver: + """Collect what the session reports, in arrival order.""" + + def __init__(self) -> None: + """Start with nothing observed.""" + self.calls: list[tuple[str, Any]] = [] + + async def heard(self, text: str) -> None: + """Record the user's transcript.""" + self.calls.append(("heard", text)) + + async def spoke(self, text: str) -> None: + """Record the assistant's own words.""" + self.calls.append(("spoke", text)) + + async def audio(self, data: bytes) -> None: + """Record one decoded audio chunk.""" + self.calls.append(("audio", data)) + + async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: + """Record a tool call request.""" + self.calls.append(("tool", (call_id, name, arguments))) + + async def failed(self, message: str) -> None: + """Record a session failure.""" + self.calls.append(("failed", message)) + + def kinds(self) -> list[str]: + """Return just the kind of each observed call.""" + return [kind for kind, _ in self.calls] + + +def _event(kind: str, **fields: Any) -> str: + """Build one server frame.""" + return json.dumps({"type": kind, **fields}) + + +def test_the_endpoint_carries_the_workspace_and_model() -> None: + """The URL is built per workspace and region, and the key rides in a header.""" + assert CONFIG.url() == ( + "wss://ws_001.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime" + "?model=qwen-audio-3.0-realtime-plus" + ) + assert CONFIG.headers() == {"Authorization": "Bearer key-abc"} + + +def test_configure_puts_the_session_in_push_to_talk() -> None: + """turn_detection is null, because our own protocol owns turn boundaries. + + Letting the model decide when a turn ended would race voice.stream.end: it would + answer before the client said it had finished speaking. + """ + + async def scenario() -> None: + """Configure a session and read back what was sent.""" + transport = FakeTransport() + session = QwenAudioSession(transport, CONFIG) + + await session.configure("你是日程助手", [{"type": "function"}]) + + assert transport.types() == ["session.update"] + sent = transport.sent[0]["session"] + assert sent["turn_detection"] is None + assert sent["modalities"] == ["text", "audio"] + assert sent["input_audio_format"] == "pcm" + assert sent["output_audio_format"] == "pcm" + assert sent["instructions"] == "你是日程助手" + assert sent["tools"] == [{"type": "function"}] + + asyncio.run(scenario()) + + +def test_audio_is_base64_encoded_on_the_way_out() -> None: + """The vendor takes audio as base64 inside a JSON event, not as binary frames.""" + + async def scenario() -> None: + """Send one chunk and inspect the frame.""" + transport = FakeTransport() + session = QwenAudioSession(transport, CONFIG) + + await session.send_audio(b"\x01\x02\x03") + + assert transport.types() == ["input_audio_buffer.append"] + assert base64.b64decode(transport.sent[0]["audio"]) == b"\x01\x02\x03" + + asyncio.run(scenario()) + + +def test_finishing_input_commits_then_asks_for_a_reply() -> None: + """Both events are needed and in this order; commit alone produces no answer.""" + + async def scenario() -> None: + """Finish the input and read back what was sent.""" + transport = FakeTransport() + session = QwenAudioSession(transport, CONFIG) + + await session.finish_input() + + assert transport.types() == ["input_audio_buffer.commit", "response.create"] + + asyncio.run(scenario()) + + +def test_a_turn_is_reported_as_transcript_speech_audio_then_ends() -> None: + """Vendor event names and base64 stay inside the adapter.""" + + async def scenario() -> None: + """Replay a full turn and read what the observer saw.""" + transport = FakeTransport( + _event("conversation.item.input_audio_transcription.completed", transcript="明天开会"), + _event("response.audio_transcript.done", transcript="好，记下了"), + _event("response.audio.delta", delta=base64.b64encode(b"pcm-1").decode()), + _event("response.audio.delta", delta=base64.b64encode(b"pcm-2").decode()), + _event("response.done"), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [ + ("heard", "明天开会"), + ("spoke", "好，记下了"), + ("audio", b"pcm-1"), + ("audio", b"pcm-2"), + ] + + asyncio.run(scenario()) + + +def test_pump_returns_when_the_turn_is_done() -> None: + """response.done ends the loop rather than leaving it waiting on the socket.""" + + async def scenario() -> None: + """Replay a turn that only says it finished.""" + transport = FakeTransport(_event("response.done")) + + await asyncio.wait_for( + QwenAudioSession(transport, CONFIG).pump(RecordingObserver()), timeout=1.0 + ) + + asyncio.run(scenario()) + + +def test_a_tool_call_is_reported_with_parsed_arguments() -> None: + """Arguments arrive as a JSON string and are handed over already parsed.""" + + async def scenario() -> None: + """Replay a tool call request.""" + transport = FakeTransport( + _event( + "response.function_call_arguments.done", + call_id="call_1", + name="list_schedules", + arguments='{"range":"this_week"}', + ), + _event("response.done"), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [("tool", ("call_1", "list_schedules", {"range": "this_week"}))] + + asyncio.run(scenario()) + + +def test_a_tool_call_without_a_call_id_fails_the_turn() -> None: + """A tool call that cannot be answered ends the turn instead of being half-run. + + Without call_id there is no way to write the result back, so continuing would leave + the model waiting forever. + """ + + async def scenario() -> None: + """Replay a tool call missing its identifier.""" + transport = FakeTransport( + _event("response.function_call_arguments.done", name="list_schedules", arguments="{}") + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + # The reason matters, not just that it failed: reading past the end of the script + # also reports a failure, so a weaker assertion would pass even if this guard + # were removed. + assert observer.kinds() == ["failed"] + assert "tool call" in observer.calls[0][1] + + asyncio.run(scenario()) + + +def test_sending_a_tool_result_lets_the_model_continue() -> None: + """The output is written back as a conversation item, then a reply is requested.""" + + async def scenario() -> None: + """Send a tool result and read back what was sent.""" + transport = FakeTransport() + session = QwenAudioSession(transport, CONFIG) + + await session.send_tool_result("call_1", '{"count":2}') + + assert transport.types() == ["conversation.item.create", "response.create"] + item = transport.sent[0]["item"] + assert item == { + "type": "function_call_output", + "call_id": "call_1", + "output": '{"count":2}', + } + + asyncio.run(scenario()) + + +def test_a_malformed_audio_delta_is_dropped_not_fatal() -> None: + """One bad chunk does not end a turn that is otherwise fine.""" + + async def scenario() -> None: + """Replay a turn containing one undecodable delta.""" + transport = FakeTransport( + _event("response.audio.delta", delta="not-base64!!"), + _event("response.audio.delta", delta=base64.b64encode(b"good").decode()), + _event("response.done"), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [("audio", b"good")] + + asyncio.run(scenario()) + + +def test_a_vendor_error_event_fails_the_turn_with_its_message() -> None: + """The vendor's own message is surfaced rather than replaced by a generic one.""" + + async def scenario() -> None: + """Replay an error event.""" + transport = FakeTransport(_event("error", error={"message": "quota exceeded"})) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [("failed", "quota exceeded")] + + asyncio.run(scenario()) + + +def test_a_dropped_connection_fails_the_turn() -> None: + """A transport error ends the turn instead of propagating out of the pump.""" + + async def scenario() -> None: + """Replay a transport that raises on receive.""" + + class BrokenTransport(FakeTransport): + """A socket that fails on the first receive.""" + + async def recv(self) -> str: + """Fail as a dropped connection would.""" + raise ConnectionResetError + + observer = RecordingObserver() + + await QwenAudioSession(BrokenTransport(), CONFIG).pump(observer) + + assert observer.kinds() == ["failed"] + assert "ConnectionResetError" in observer.calls[0][1] + + asyncio.run(scenario()) + + +def test_unknown_vendor_events_are_ignored() -> None: + """Events this adapter does not act on do not stop the turn. + + The vendor emits many lifecycle events (session.created, response.created, speech + started/stopped); reacting to an unknown one would break on every API addition. + """ + + async def scenario() -> None: + """Replay lifecycle noise around one real event.""" + transport = FakeTransport( + _event("session.created"), + _event("response.created"), + _event("input_audio_buffer.speech_started"), + _event("conversation.item.input_audio_transcription.completed", transcript="喂"), + _event("response.output_item.added"), + _event("response.done"), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [("heard", "喂")] + + asyncio.run(scenario()) + + +def test_opening_a_session_connects_then_configures() -> None: + """The factory hands back a session that is already in push-to-talk.""" + + async def scenario() -> None: + """Open a session through the factory with a fake connect.""" + transport = FakeTransport() + + async def connect(config: QwenAudioConfig) -> FakeTransport: + """Return the fake transport instead of dialling out.""" + assert config is CONFIG + return transport + + factory = QwenAudioSessionFactory(CONFIG, connect=connect) + session = await factory.open("你是日程助手", []) + + assert isinstance(session, QwenAudioSession) + assert transport.types() == ["session.update"] + + asyncio.run(scenario()) + + +def test_closing_a_session_that_already_went_away_is_not_an_error() -> None: + """Cleanup must not raise, or a failed turn would fail twice.""" + + async def scenario() -> None: + """Close a transport that raises on close.""" + + class RefusesToClose(FakeTransport): + """A socket that fails when closed.""" + + async def close(self) -> None: + """Fail as an already-closed socket would.""" + raise ConnectionResetError + + await QwenAudioSession(RefusesToClose(), CONFIG).close() + + asyncio.run(scenario()) + + +def test_the_reply_text_is_reported_from_its_increments() -> None: + """spoke carries the text accumulated so far, so it is ready before the audio starts. + + Measured against the real model the increments finish well before the first audio + chunk, while the terminal event lands after it. Reporting only on the terminal event + would leave the first audio chunk with no text beside it. + """ + + async def scenario() -> None: + """Replay a reply whose text streams in three increments before any audio.""" + transport = FakeTransport( + _event("response.audio_transcript.delta", delta="好，"), + _event("response.audio_transcript.delta", delta="明天三点"), + _event("response.audio_transcript.delta", delta="记下了"), + _event("response.audio.delta", delta=base64.b64encode(b"pcm").decode()), + _event("response.audio_transcript.done", transcript="好，明天三点记下了"), + _event("response.done"), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [ + ("spoke", "好，"), + ("spoke", "好，明天三点"), + ("spoke", "好，明天三点记下了"), + ("audio", b"pcm"), + ] + + asyncio.run(scenario()) + + +def test_a_reply_with_no_increments_is_still_reported() -> None: + """A reply that only arrives as a terminal event is not lost.""" + + async def scenario() -> None: + """Replay a reply that skips increments entirely.""" + transport = FakeTransport( + _event("response.audio_transcript.done", transcript="好"), + _event("response.done"), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [("spoke", "好")] + + asyncio.run(scenario()) + + +def test_the_adapter_satisfies_the_dialogue_layer_s_ports() -> None: + """The two declarations of the seam are the same shape. + + Neither side imports the other, so nothing else would notice one of them drifting. + mypy rejects these assignments the moment they stop matching; no call site needed. + """ + session: RealtimeSession = QwenAudioSession(FakeTransport(), CONFIG) + factory: RealtimeSessionFactory = QwenAudioSessionFactory(CONFIG) + observer: Observer = _SeamObserver() + also_a_turn_observer: TurnObserver = _SeamObserver() + + assert (session, factory, observer, also_a_turn_observer) is not None + + +class _SeamObserver: + """An observer written once and checked against both declarations of the shape.""" + + async def heard(self, text: str) -> None: + """Ignore what the user said.""" + + async def spoke(self, text: str) -> None: + """Ignore what the model said.""" + + async def audio(self, data: bytes) -> None: + """Ignore the model's speech.""" + + async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: + """Ignore the tool request.""" + + async def failed(self, message: str) -> None: + """Ignore the failure.""" + + +def test_a_session_that_cannot_be_configured_closes_its_transport() -> None: + """A socket the caller never receives is a socket nobody can close.""" + + class RefusingTransport(FakeTransport): + """A transport that connects and then refuses the session update.""" + + async def send(self, message: str) -> None: + """Fail the way a rejected session.update would.""" + raise ConnectionResetError("session.update rejected") + + async def scenario() -> None: + """Open a session whose configuration fails.""" + transport = RefusingTransport() + factory = QwenAudioSessionFactory(CONFIG, connect=lambda config: _ready(transport)) + + try: + await factory.open("", []) + except ConnectionResetError: + pass + else: + raise AssertionError("expected the configuration failure to propagate") + + assert transport.closed is True + + asyncio.run(scenario()) + + +async def _ready(transport: FakeTransport) -> FakeTransport: + """Hand back an already-built transport, as a connect seam would.""" + return transport diff --git a/backend/tests/intelligence/realtime/test_realtime_agent.py b/backend/tests/intelligence/realtime/test_realtime_agent.py new file mode 100644 index 0000000..a1878ab --- /dev/null +++ b/backend/tests/intelligence/realtime/test_realtime_agent.py @@ -0,0 +1,410 @@ +"""Driving a realtime model through one turn, with a scripted fake session.""" + +import asyncio +from collections.abc import AsyncIterator +from dataclasses import dataclass, field +from typing import Any + +from timeflow.intelligence.ports import ( + AudioReply, + CommandResult, + DialogueQuestion, + ReplyText, + Transcript, +) +from timeflow.intelligence.realtime.agent import RealtimeAgent + + +@dataclass(frozen=True, slots=True) +class _Stream: + """Identifiers of the audio stream a turn answers.""" + + session_id: str = "ws_session_test" + stream_id: str = "stream_test" + conversation_id: str = "conversation_test" + request_id: str | None = "req_voice_001" + + +@dataclass +class RecordingSink: + """Record what the agent pushed, in order, resolving audio to bytes.""" + + calls: list[tuple[str, Any]] = field(default_factory=list) + + async def deliver_transcript(self, transcript: Transcript, stream: Any) -> None: + """Record what the user was heard to say.""" + self.calls.append(("transcript", transcript)) + + async def deliver_reply_text(self, reply: ReplyText, stream: Any) -> None: + """Record how much of the reply's wording had been settled.""" + self.calls.append(("done" if reply.done else "reply", reply.speech_text)) + + async def deliver_result(self, result: CommandResult, stream: Any) -> None: + """Record a command result.""" + self.calls.append(("result", result)) + + async def deliver_question(self, question: DialogueQuestion, stream: Any) -> None: + """Record a question put to the user.""" + self.calls.append(("question", question)) + + async def deliver_audio( + self, reply: AudioReply, chunks: AsyncIterator[bytes], stream: Any + ) -> None: + """Record the reply description, then drain its audio.""" + self.calls.append(("audio_start", reply)) + async for chunk in chunks: + self.calls.append(("audio", chunk)) + self.calls.append(("audio_end", reply.audio_id)) + + def kinds(self) -> list[str]: + """Return just the kind of each recorded call.""" + return [kind for kind, _ in self.calls] + + +class ScriptedSession: + """A session that replays a scripted sequence of observer calls.""" + + def __init__(self, script: list[tuple[str, Any]]) -> None: + """Store the script plus room to record what was sent.""" + self._script = script + self.audio_sent: list[bytes] = [] + self.finished = False + self.closed = False + self.tool_results: list[tuple[str, str]] = [] + + async def send_audio(self, chunk: bytes) -> None: + """Record one chunk of the user's speech.""" + self.audio_sent.append(chunk) + + async def finish_input(self) -> None: + """Record that the input was closed.""" + self.finished = True + + async def send_tool_result(self, call_id: str, output: str) -> None: + """Record a tool result written back.""" + self.tool_results.append((call_id, output)) + + async def close(self) -> None: + """Record that the session was released.""" + self.closed = True + + async def pump(self, observer: Any) -> None: + """Replay the script against the observer.""" + for kind, payload in self._script: + await getattr(observer, kind)(*payload) + + +class ScriptedFactory: + """Hand out one scripted session, recording how it was configured.""" + + def __init__(self, session: ScriptedSession) -> None: + """Store the session to hand out.""" + self._session = session + self.instructions: str | None = None + + async def open(self, instructions: str, tools: list[dict[str, Any]]) -> ScriptedSession: + """Record the configuration and return the scripted session.""" + self.instructions = instructions + return self._session + + +class FailingFactory: + """A factory that cannot reach the model.""" + + async def open(self, instructions: str, tools: list[dict[str, Any]]) -> ScriptedSession: + """Fail as an unreachable model would.""" + raise ConnectionRefusedError + + +async def _chunks(*payloads: bytes) -> AsyncIterator[bytes]: + """Yield the given chunks with no delay.""" + for payload in payloads: + yield payload + + +def test_a_turn_pushes_the_transcript_then_the_spoken_reply() -> None: + """The user's words go out first, then the reply's audio framed by its text.""" + + async def scenario() -> None: + """Replay a turn that hears, speaks, and sends two audio chunks.""" + session = ScriptedSession( + [ + ("heard", ("明天下午三点在203开会",)), + ("spoke", ("好，",)), + ("spoke", ("好，记下了",)), + ("audio", (b"pcm-1",)), + ("audio", (b"pcm-2",)), + ] + ) + sink = RecordingSink() + + await RealtimeAgent(ScriptedFactory(session), sink).handle_audio( + _chunks(b"a" * 3200), _Stream() + ) + + # The wording goes out as it forms, ahead of the audio for it, and is settled once + # at the end. Only the audio is framed by a start and an end. + assert sink.kinds() == [ + "transcript", + "reply", + "reply", + "audio_start", + "audio", + "audio", + "done", + "audio_end", + ] + assert [text for kind, text in sink.calls if kind in ("reply", "done")] == [ + "好，", + "好，记下了", + "好，记下了", + ] + heard = sink.calls[0][1] + assert heard.text == "明天下午三点在203开会" + assert heard.duration_ms == 100 # 3200 bytes at 16 kHz mono 16-bit + assert [chunk for kind, chunk in sink.calls if kind == "audio"] == [b"pcm-1", b"pcm-2"] + + asyncio.run(scenario()) + + +def test_the_reply_text_on_the_opening_message_is_the_text_known_by_then() -> None: + """voice.tts.start carries the reply's words, gathered before the audio began. + + The model streams its own text well before the first audio chunk but only confirms it + afterwards, so the opening message uses what has accumulated rather than waiting. + """ + + async def scenario() -> None: + """Replay a turn whose text streams fully before any audio.""" + session = ScriptedSession( + [ + ("spoke", ("好，",)), + ("spoke", ("好，明天三点",)), + ("spoke", ("好，明天三点记下了",)), + ("audio", (b"pcm",)), + ] + ) + sink = RecordingSink() + + await RealtimeAgent(ScriptedFactory(session), sink).handle_audio(_chunks(b"a"), _Stream()) + + reply = next(value for kind, value in sink.calls if kind == "audio_start") + assert reply.speech_text == "好，明天三点记下了" + assert reply.sample_rate_hz == 24000 + assert reply.audio_format == "pcm" + assert reply.purpose == "command_result" + + asyncio.run(scenario()) + + +def test_audio_goes_out_before_the_model_has_finished_speaking() -> None: + """Each chunk is handed on as it arrives, not collected and sent at the end. + + Buffering would erase the latency the realtime model exists for, which is the one + number the two candidate approaches are compared on. + """ + + async def scenario() -> None: + """Check what the sink has seen from inside the model's own reporting.""" + seen_midway: list[str] = [] + sink = RecordingSink() + + class WatchingSession(ScriptedSession): + """A session that inspects the sink between two audio chunks.""" + + async def pump(self, observer: Any) -> None: + """Report one chunk, look at the sink, then report another.""" + await observer.spoke("好") + await observer.audio(b"first") + await asyncio.sleep(0) + seen_midway.extend(sink.kinds()) + await observer.audio(b"second") + + await RealtimeAgent(ScriptedFactory(WatchingSession([])), sink).handle_audio( + _chunks(b"a"), _Stream() + ) + + assert seen_midway == ["reply", "audio_start", "audio"] + + asyncio.run(scenario()) + + +def test_the_user_audio_reaches_the_model_unchanged_then_input_is_closed() -> None: + """Chunks are forwarded byte for byte, and the model is told when to answer.""" + + async def scenario() -> None: + """Send three chunks and inspect what the session received.""" + session = ScriptedSession([]) + + await RealtimeAgent(ScriptedFactory(session), RecordingSink()).handle_audio( + _chunks(b"one", b"two", b"three"), _Stream() + ) + + assert session.audio_sent == [b"one", b"two", b"three"] + assert session.finished is True + # Closed at the end of the turn. Keeping it open so a follow-up can remember this + # turn is what the round adding questions needs, and it lands with them. + assert session.closed is True + + asyncio.run(scenario()) + + +def test_the_instructions_are_applied_when_the_session_opens() -> None: + """The assistant's role is set before any audio, as the vendor requires. + + Built per turn rather than stored: the instructions state the current date, and a + server running for days would otherwise keep telling the model it is still day one. + """ + + async def scenario() -> None: + """Open a turn with instructions and read them back.""" + factory = ScriptedFactory(ScriptedSession([])) + + await RealtimeAgent( + factory, RecordingSink(), instructions=lambda: "你是日程助手" + ).handle_audio(_chunks(b"a"), _Stream()) + + assert factory.instructions == "你是日程助手" + + asyncio.run(scenario()) + + +def test_an_empty_transcript_pushes_nothing() -> None: + """A turn the model could not transcribe does not send an empty transcript.""" + + async def scenario() -> None: + """Replay a turn that heard nothing.""" + sink = RecordingSink() + + await RealtimeAgent( + ScriptedFactory(ScriptedSession([("heard", ("",))])), sink + ).handle_audio(_chunks(b"a"), _Stream()) + + assert sink.calls == [] + + asyncio.run(scenario()) + + +def test_a_turn_with_no_audio_reply_sends_no_tts_messages() -> None: + """A text-only reply does not open an audio run that would never be filled. + + Its wording still goes out: that is the point of carrying text separately from the + audio, so a reply the model never spoke aloud still reaches the user. + """ + + async def scenario() -> None: + """Replay a turn that speaks but produces no audio.""" + sink = RecordingSink() + + await RealtimeAgent( + ScriptedFactory(ScriptedSession([("heard", ("在吗",)), ("spoke", ("在",))])), sink + ).handle_audio(_chunks(b"a"), _Stream()) + + assert sink.kinds() == ["transcript", "reply", "done"] + assert not [kind for kind in sink.kinds() if kind.startswith("audio")] + + asyncio.run(scenario()) + + +def test_an_unreachable_model_does_not_raise_into_the_transport() -> None: + """A model that cannot be opened ends the turn quietly; the session stays usable. + + The audio sink runs in a background task whose exceptions are only logged, so raising + here would lose the reason and leave the client waiting with no explanation either way. + """ + + async def scenario() -> None: + """Run a turn against a factory that refuses to connect.""" + sink = RecordingSink() + + await RealtimeAgent(FailingFactory(), sink).handle_audio(_chunks(b"a"), _Stream()) + + assert sink.calls == [] + + asyncio.run(scenario()) + + +def test_a_failing_session_still_closes_the_audio_it_started() -> None: + """A reply cut short is still closed, so the client is not left waiting.""" + + async def scenario() -> None: + """Replay a turn that sends one chunk and then fails.""" + session = ScriptedSession( + [("spoke", ("好",)), ("audio", (b"pcm",)), ("failed", ("quota exceeded",))] + ) + sink = RecordingSink() + + await RealtimeAgent(ScriptedFactory(session), sink).handle_audio(_chunks(b"a"), _Stream()) + + # Both runs are closed, not just the audio: a client showing the wording as it + # arrives needs to be told it is final, or a reply cut short reads as still coming. + # The settling update comes before audio_end, which is what ends the turn. + assert sink.kinds() == ["reply", "audio_start", "audio", "done", "audio_end"] + + asyncio.run(scenario()) + + +def test_the_wording_is_settled_before_the_audio_closes_the_turn() -> None: + """The settling update precedes voice.tts.end, because that message ends the turn. + + Found on the real model: with the order reversed, a client that stops reading once the + audio run closes -- the reasonable reading of the protocol -- never sees the wording + marked final, and goes on showing an answer as still arriving. + """ + + async def scenario() -> None: + """Replay a turn that speaks and sends audio, then inspect the tail.""" + sink = RecordingSink() + session = ScriptedSession([("spoke", ("在",)), ("audio", (b"pcm",))]) + + await RealtimeAgent(ScriptedFactory(session), sink).handle_audio(_chunks(b"a"), _Stream()) + + kinds = sink.kinds() + assert kinds[-1] == "audio_end" + assert kinds.index("done") < kinds.index("audio_end") + + asyncio.run(scenario()) + + +def test_a_failure_while_sending_audio_does_not_leave_the_pump_running() -> None: + """The pump is cancelled on any exit, not only on the caller being cancelled.""" + + class BlockingSession(ScriptedSession): + """A session whose pump waits forever unless it is cancelled.""" + + def __init__(self) -> None: + """Start with nothing scripted and no cancellation seen.""" + super().__init__([]) + self.pump_cancelled = False + + async def pump(self, observer: Any) -> None: + """Wait to be cancelled, recording that it was.""" + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + self.pump_cancelled = True + raise + + async def failing_chunks() -> AsyncIterator[bytes]: + """Yield one chunk, let the pump start, then fail as a broken stream would.""" + yield b"a" + await asyncio.sleep(0) + raise RuntimeError("the inbound stream broke") + + async def scenario() -> None: + """Run a turn whose audio source fails partway.""" + session = BlockingSession() + + try: + await RealtimeAgent(ScriptedFactory(session), RecordingSink()).handle_audio( + failing_chunks(), _Stream() + ) + except RuntimeError as error: + assert "inbound stream broke" in str(error) + else: + raise AssertionError("expected the stream failure to reach the caller") + + assert session.pump_cancelled is True + assert session.closed is True + + asyncio.run(scenario()) diff --git a/backend/tests/test_app_wiring.py b/backend/tests/test_app_wiring.py index da15389..20eeeed 100644 --- a/backend/tests/test_app_wiring.py +++ b/backend/tests/test_app_wiring.py @@ -15,11 +15,32 @@ async def verify(self, access_token: str) -> str | None: return None -def _build_with_environment(environment: str, **injected: object) -> object: - """Build the app with TIMEFLOW_ENVIRONMENT set, clearing the settings cache around it.""" +def _build_with_environment( + environment: str, *, audio_configured: bool = False, **injected: object +) -> object: + """Build the app with the environment set and the model's credentials pinned. + + The credentials are stated rather than inherited because a developer with a working + .env would otherwise get the configured path and see these guards not fire, while CI + with no credentials would see them fire -- the same test meaning two different things + depending on whose machine it runs on. + """ + credentials = ( + { + "TIMEFLOW_ALIYUN_AUDIO_API_KEY": "key-for-test", + "TIMEFLOW_ALIYUN_AUDIO_WORKSPACE_ID": "ws-for-test", + } + if audio_configured + else { + "TIMEFLOW_ALIYUN_AUDIO_API_KEY": "", + "TIMEFLOW_ALIYUN_AUDIO_WORKSPACE_ID": "", + } + ) get_settings.cache_clear() try: - with mock.patch.dict(os.environ, {"TIMEFLOW_ENVIRONMENT": environment}, clear=False): + with mock.patch.dict( + os.environ, {"TIMEFLOW_ENVIRONMENT": environment, **credentials}, clear=False + ): return create_app(**injected) # type: ignore[arg-type] finally: get_settings.cache_clear() @@ -69,3 +90,16 @@ async def consume(self, chunks: object, stream: object) -> None: ) assert built is not None + + +def test_a_configured_model_lets_a_deployment_build_without_an_injected_sink() -> None: + """The sink guard is about the stand-in, not about wiring a sink by hand. + + A deployment that gives the model its credentials has a real agent, so refusing to + start would only force every deployment to inject a sink it does not need to own. + """ + application = _build_with_environment( + "production", audio_configured=True, token_verifier=_RealVerifier() + ) + + assert application is not None