From 01fe9157b266b605250d155bdd80409aa733a063 Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Tue, 11 Aug 2026 15:47:51 +0800 Subject: [PATCH 1/5] feat(infrastructure): adapt the Qwen-Audio realtime model Audio goes in and speech comes back out, with tool calls in between. The adapter keeps the vendor's event names and its base64 inside itself: what the dialogue layer receives is already decoded and named in its own terms. The seam is declared twice on purpose. Infrastructure must not depend on a product layer, so the adapter states the shape it reports through as its own Observer while the dialogue layer states the same shape as TurnObserver. Nothing imports the other, which also means nothing would notice one side drifting -- so a test assigns across both declarations and mypy rejects it the moment they stop matching. Two behaviours are worth naming. Turn boundaries stay with our own protocol (turn_detection is off), because voice.stream.start and end already draw them and a model deciding as well would fight them. And a turn is one response unless a tool runs: answering the tool asks the model to respond again, and that second response is the one carrying the audio, so the pump counts them instead of stopping at the first response.done. --- backend/.env.example | 9 + .../external/realtime/__init__.py | 1 + .../external/realtime/qwen_audio.py | 295 +++++++++++ .../src/timeflow/infrastructure/settings.py | 23 + .../intelligence/realtime/__init__.py | 1 + .../timeflow/intelligence/realtime/ports.py | 64 +++ .../external/realtime/test_qwen_audio.py | 476 ++++++++++++++++++ 7 files changed, 869 insertions(+) create mode 100644 backend/src/timeflow/infrastructure/external/realtime/__init__.py create mode 100644 backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py create mode 100644 backend/src/timeflow/intelligence/realtime/__init__.py create mode 100644 backend/src/timeflow/intelligence/realtime/ports.py create mode 100644 backend/tests/infrastructure/external/realtime/test_qwen_audio.py 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..c2fcc30 --- /dev/null +++ b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py @@ -0,0 +1,295 @@ +"""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__) + +# The client controls turn boundaries: our own protocol already opens and closes a stream +# with voice.stream.start / end, so 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. + + The dialogue layer declares the same shape for its own use. Restating it here rather + than importing it keeps the dependency direction outward-only: infrastructure never + reaches into a product layer. + """ + + 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 turn is one response unless a tool runs: answering a tool asks the model to + # respond again, and that second response is the one that 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 is sent. + + turn_detection is only settable before the first audio frame, so this must run + first; push-to-talk is required because our own protocol owns turn boundaries. + """ + 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 until the turn ends or the session fails. + + Vendor event names and base64 stay inside this method: whatever the observer + receives is already decoded and named in the dialogue layer's terms. + + The reply's own text is reported from its increments rather than from the vendor's + terminal event. Measured against the real model, the increments finish about 600 ms + before the first audio chunk arrives while the terminal event lands 500 ms after it, + so waiting for that event would mean either sending audio with no text beside it or + holding the audio back and losing the latency the model exists for. + """ + 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": + # Terminal marker only; the text was already reported from its increments. + # It still gets reported once more, because a reply short enough to arrive + # without increments would otherwise never be reported at all. + 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": + # A response that only asked for a tool is not the end of the turn: the + # answer to that tool triggers another response, and that is the one that + # speaks. Returning here would leave its audio unread. + 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.""" + connect = self._connect or _default_connect + async with asyncio.timeout(self._open_timeout_seconds): + transport = await connect(self._config) + session = QwenAudioSession(transport, self._config) + await session.configure(instructions, tools) + 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/ports.py b/backend/src/timeflow/intelligence/realtime/ports.py new file mode 100644 index 0000000..d25d382 --- /dev/null +++ b/backend/src/timeflow/intelligence/realtime/ports.py @@ -0,0 +1,64 @@ +"""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. + + Named in the dialogue layer's terms rather than a vendor's event names, so swapping + providers never reaches orchestration. The adapter calls these; it never imports this + module -- it declares the same shape on its own side. + """ + + 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/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py new file mode 100644 index 0000000..6629e4b --- /dev/null +++ b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py @@ -0,0 +1,476 @@ +"""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 is deliberate: if a change makes the pump read past + the end of a turn, the test fails immediately instead of hanging until CI times + out, and the reason is visible in the failure. + """ + 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 sides of the seam are the same shape, and this is what checks it. + + The dialogue layer declares RealtimeSession and TurnObserver; the adapter declares + Transport and Observer. Neither imports the other, because infrastructure must not + depend on a product layer -- which also means nothing at all notices when one side + grows a parameter the other does not have. These assignments are that notice: mypy + rejects them the moment the shapes drift apart, and no call site is needed to provoke + it. + """ + 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.""" From 7d1b6c3e7ca16a50ae65c03af9af06ccceffb78a Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Tue, 11 Aug 2026 16:50:32 +0800 Subject: [PATCH 2/5] feat(intelligence): let the realtime model answer a turn out loud The transport had a stand-in behind it that reported the same fixed result to every stream. This puts the real model there: audio goes to it, and what comes back reaches the client through three of the outlets the boundary already has -- the user's words, the reply's wording as it forms, and the speech itself. The wording is forwarded as the model produces it rather than held for the audio's opening message. Measured against the real model it finishes about 600 ms before the first audio chunk exists, and voice.tts.start cannot be sent until there is audio to announce, so binding them costs the client that gap. It is settled with a final update after the audio ends, not before, so a client can use it to stop showing that an answer is still coming. Composition fails closed on the stand-in rather than on missing credentials: a deployment that configures the model is a real one and should start, while falling back outside development would leave a server calling commands applied that were never carried out. --- .../external/realtime/qwen_audio.py | 1 - .../timeflow/intelligence/realtime/agent.py | 211 +++++++++++ backend/src/timeflow/main.py | 59 ++- .../realtime/test_realtime_agent.py | 343 ++++++++++++++++++ 4 files changed, 602 insertions(+), 12 deletions(-) create mode 100644 backend/src/timeflow/intelligence/realtime/agent.py create mode 100644 backend/tests/intelligence/realtime/test_realtime_agent.py diff --git a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py index c2fcc30..2f17846 100644 --- a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py +++ b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py @@ -63,7 +63,6 @@ async def failed(self, message: str) -> None: ... - @dataclass(frozen=True, slots=True) class QwenAudioConfig: """Where to reach the model and how to authenticate.""" diff --git a/backend/src/timeflow/intelligence/realtime/agent.py b/backend/src/timeflow/intelligence/realtime/agent.py new file mode 100644 index 0000000..e8b6984 --- /dev/null +++ b/backend/src/timeflow/intelligence/realtime/agent.py @@ -0,0 +1,211 @@ +"""Agent that hands one turn to a realtime speech model and reports what it says.""" + +import asyncio +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" + +# Why the reply is being spoken. Only answers exist this round; a question sounds the same +# on the wire, so the round that adds them will set this per turn instead. +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 rather than stored, so a server running for days keeps telling + # the model what day it is now rather than what day it started. + 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: + # Returning rather than raising: the transport only asked us to take the audio, + # and a model we cannot reach is not a transport fault to report as one. + 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 asyncio.CancelledError: + pumping.cancel() + raise + finally: + await turn.close() + await session.close() + + +class _Turn: + """One turn's worth of model output, translated into ResultSink calls. + + Exists because the two sides disagree about who drives. The session reports by + calling methods; ResultSink.deliver_audio wants to pull an iterator. A queue in + between lets audio go out chunk by chunk as it arrives, instead of being collected + and sent once the reply is over. + """ + + 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, and keep it for the audio's opening message. + + The session already reports accumulations rather than fragments, which is the shape + ReplyText wants, so this forwards them as they come. Measured against the real + model the wording finishes about 600 ms before the first audio chunk arrives, so + holding it back until the audio starts would waste that whole gap. + """ + 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. + + Delivery starts here rather than at the top of the turn because the opening message + carries the reply's text, and that is only known once the model has begun speaking. + """ + 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. + + No tools are registered this round, so the model should not be asking. It is + logged rather than answered: writing an invented result back would have it speak + about schedules that were never looked up. + """ + 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: + """Finish the audio if any started, then settle the reply's wording. + + The settling update goes last so it marks the end of the whole turn rather than + just the end of the text: a client that hides its "still answering" state on it + would otherwise hide it while audio was still arriving. + """ + if self._speaking is not None: + await self._audio.put(None) + await self._speaking + if self._spoken: + await self._result_sink.deliver_reply_text( + ReplyText(reply_id=self._reply_id, speech_text=self._spoken, done=True), + self._stream, + ) + + 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/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/intelligence/realtime/test_realtime_agent.py b/backend/tests/intelligence/realtime/test_realtime_agent.py new file mode 100644 index 0000000..4ec6118 --- /dev/null +++ b/backend/tests/intelligence/realtime/test_realtime_agent.py @@ -0,0 +1,343 @@ +"""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", + "audio_end", + "done", + ] + 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. + assert sink.kinds() == ["reply", "audio_start", "audio", "audio_end", "done"] + + asyncio.run(scenario()) From e2d07df1c5bb7b7cbdbd9b7b96b68a7aca9a42af Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Tue, 11 Aug 2026 16:54:24 +0800 Subject: [PATCH 3/5] fix(intelligence): settle the reply's wording before the audio closes the turn Found by running it: voice.tts.end is what ends a turn, so a client that stops reading there -- the reasonable reading -- never saw the update marking the wording final, and would go on showing an answer as still arriving. Measured on the real model the wording is final around 800 ms before the audio ends, so the turn's end is merely the earliest moment this side can be certain, not the moment it becomes true. The wiring test that covers the stand-in guard was inheriting whatever credentials the developer's .env held, which made it mean one thing in CI and another locally. It now pins them, and a new case covers the path this change opens: a deployment that configures the model builds without injecting a sink. --- .../timeflow/intelligence/realtime/agent.py | 17 ++++---- .../realtime/test_realtime_agent.py | 27 ++++++++++++- backend/tests/test_app_wiring.py | 40 +++++++++++++++++-- 3 files changed, 72 insertions(+), 12 deletions(-) diff --git a/backend/src/timeflow/intelligence/realtime/agent.py b/backend/src/timeflow/intelligence/realtime/agent.py index e8b6984..3177904 100644 --- a/backend/src/timeflow/intelligence/realtime/agent.py +++ b/backend/src/timeflow/intelligence/realtime/agent.py @@ -176,20 +176,23 @@ async def failed(self, message: str) -> None: logger.warning("realtime session failed", extra={"reason": message}) async def close(self) -> None: - """Finish the audio if any started, then settle the reply's wording. + """Settle the reply's wording, then finish the audio if any started. - The settling update goes last so it marks the end of the whole turn rather than - just the end of the text: a client that hides its "still answering" state on it - would otherwise hide it while audio was still arriving. + The settling update has to precede the audio's closing message, because that + message is what ends a turn: a client that stops reading on voice.tts.end -- the + reasonable thing to do -- would never see an update sent after it. The wording is + in fact final well before this point; the session reports no completion for it, so + the end of the turn is the earliest moment this side can be sure. """ - if self._speaking is not None: - await self._audio.put(None) - await self._speaking 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.""" diff --git a/backend/tests/intelligence/realtime/test_realtime_agent.py b/backend/tests/intelligence/realtime/test_realtime_agent.py index 4ec6118..bd63f72 100644 --- a/backend/tests/intelligence/realtime/test_realtime_agent.py +++ b/backend/tests/intelligence/realtime/test_realtime_agent.py @@ -151,8 +151,8 @@ async def scenario() -> None: "audio_start", "audio", "audio", - "audio_end", "done", + "audio_end", ] assert [text for kind, text in sink.calls if kind in ("reply", "done")] == [ "好，", @@ -338,6 +338,29 @@ async def scenario() -> None: # 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. - assert sink.kinds() == ["reply", "audio_start", "audio", "audio_end", "done"] + # 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()) 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 From 1c329764655103b86ad922ca5d96b605920540e4 Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Tue, 11 Aug 2026 17:22:05 +0800 Subject: [PATCH 4/5] style(intelligence): one-line docstrings; the reasoning lives in the pull request Only the facts a maintainer would break by not knowing them are kept, one line each: that our protocol owns turn boundaries, that a tool makes a turn two responses, and that the wording has to settle before the audio closes. The last of those is guarded by a test, which is a stronger place for it than prose. --- .../external/realtime/qwen_audio.py | 39 +++------------- .../timeflow/intelligence/realtime/agent.py | 46 +++---------------- .../timeflow/intelligence/realtime/ports.py | 7 +-- .../external/realtime/test_qwen_audio.py | 16 ++----- 4 files changed, 20 insertions(+), 88 deletions(-) diff --git a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py index 2f17846..fb811f9 100644 --- a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py +++ b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py @@ -9,8 +9,7 @@ logger = logging.getLogger(__name__) -# The client controls turn boundaries: our own protocol already opens and closes a stream -# with voice.stream.start / end, so the model must not also decide when a turn ended. +# 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. @@ -35,12 +34,7 @@ async def close(self) -> None: class Observer(Protocol): - """Where this adapter reports what the model says. - - The dialogue layer declares the same shape for its own use. Restating it here rather - than importing it keeps the dependency direction outward-only: infrastructure never - reaches into a product layer. - """ + """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.""" @@ -90,16 +84,11 @@ 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 turn is one response unless a tool runs: answering a tool asks the model to - # respond again, and that second response is the one that carries the audio. + # 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 is sent. - - turn_detection is only settable before the first audio frame, so this must run - first; push-to-talk is required because our own protocol owns turn boundaries. - """ + """Set the session up before any audio; turn_detection only takes effect here.""" session: dict[str, Any] = { "modalities": ["text", "audio"], "voice": self._config.voice, @@ -151,17 +140,7 @@ async def _send(self, event: dict[str, Any]) -> None: await self._transport.send(json.dumps(event, ensure_ascii=False)) async def pump(self, observer: Observer) -> None: - """Report what the model says until the turn ends or the session fails. - - Vendor event names and base64 stay inside this method: whatever the observer - receives is already decoded and named in the dialogue layer's terms. - - The reply's own text is reported from its increments rather than from the vendor's - terminal event. Measured against the real model, the increments finish about 600 ms - before the first audio chunk arrives while the terminal event lands 500 ms after it, - so waiting for that event would mean either sending audio with no text beside it or - holding the audio back and losing the latency the model exists for. - """ + """Report what the model says, decoded and renamed, until the turn ends or fails.""" spoken = "" while True: try: @@ -190,9 +169,7 @@ async def pump(self, observer: Observer) -> None: spoken += str(event.get("delta", "")) await observer.spoke(spoken) elif kind == "response.audio_transcript.done": - # Terminal marker only; the text was already reported from its increments. - # It still gets reported once more, because a reply short enough to arrive - # without increments would otherwise never be reported at all. + # Reported again in case the reply was short enough to skip increments. final = str(event.get("transcript", "")) if final and final != spoken: spoken = final @@ -208,9 +185,7 @@ async def pump(self, observer: Observer) -> None: return await observer.tool_requested(**requested) elif kind == "response.done": - # A response that only asked for a tool is not the end of the turn: the - # answer to that tool triggers another response, and that is the one that - # speaks. Returning here would leave its audio unread. + # 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 diff --git a/backend/src/timeflow/intelligence/realtime/agent.py b/backend/src/timeflow/intelligence/realtime/agent.py index 3177904..f4df201 100644 --- a/backend/src/timeflow/intelligence/realtime/agent.py +++ b/backend/src/timeflow/intelligence/realtime/agent.py @@ -16,8 +16,7 @@ REPLY_SAMPLE_RATE_HZ = 24_000 REPLY_AUDIO_FORMAT = "pcm" -# Why the reply is being spoken. Only answers exist this round; a question sounds the same -# on the wire, so the round that adds them will set this per turn instead. +# 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. @@ -52,8 +51,7 @@ def __init__( """Store the session source, the sink, and the id seams.""" self._sessions = sessions self._result_sink = result_sink - # Called per turn rather than stored, so a server running for days keeps telling - # the model what day it is now rather than what day it started. + # 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 @@ -63,8 +61,6 @@ async def handle_audio(self, chunks: AsyncIterator[bytes], stream: StreamInfo) - try: session = await self._sessions.open(self._instructions(), []) except Exception: - # Returning rather than raising: the transport only asked us to take the audio, - # and a model we cannot reach is not a transport fault to report as one. logger.exception("could not open a realtime session") return @@ -92,13 +88,7 @@ async def handle_audio(self, chunks: AsyncIterator[bytes], stream: StreamInfo) - class _Turn: - """One turn's worth of model output, translated into ResultSink calls. - - Exists because the two sides disagree about who drives. The session reports by - calling methods; ResultSink.deliver_audio wants to pull an iterator. A queue in - between lets audio go out chunk by chunk as it arrives, instead of being collected - and sent once the reply is over. - """ + """One turn's model output, translated into ResultSink calls as it arrives.""" def __init__( self, @@ -136,35 +126,20 @@ async def heard(self, text: str) -> None: ) async def spoke(self, text: str) -> None: - """Push the reply's wording, and keep it for the audio's opening message. - - The session already reports accumulations rather than fragments, which is the shape - ReplyText wants, so this forwards them as they come. Measured against the real - model the wording finishes about 600 ms before the first audio chunk arrives, so - holding it back until the audio starts would waste that whole gap. - """ + """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. - - Delivery starts here rather than at the top of the turn because the opening message - carries the reply's text, and that is only known once the model has begun speaking. - """ + """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. - - No tools are registered this round, so the model should not be asking. It is - logged rather than answered: writing an invented result back would have it speak - about schedules that were never looked up. - """ + """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", @@ -176,14 +151,7 @@ async def failed(self, message: str) -> None: logger.warning("realtime session failed", extra={"reason": message}) async def close(self) -> None: - """Settle the reply's wording, then finish the audio if any started. - - The settling update has to precede the audio's closing message, because that - message is what ends a turn: a client that stops reading on voice.tts.end -- the - reasonable thing to do -- would never see an update sent after it. The wording is - in fact final well before this point; the session reports no completion for it, so - the end of the turn is the earliest moment this side can be sure. - """ + """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), diff --git a/backend/src/timeflow/intelligence/realtime/ports.py b/backend/src/timeflow/intelligence/realtime/ports.py index d25d382..9be716a 100644 --- a/backend/src/timeflow/intelligence/realtime/ports.py +++ b/backend/src/timeflow/intelligence/realtime/ports.py @@ -4,12 +4,7 @@ class TurnObserver(Protocol): - """What a realtime session reports while a turn runs. - - Named in the dialogue layer's terms rather than a vendor's event names, so swapping - providers never reaches orchestration. The adapter calls these; it never imports this - module -- it declares the same shape on its own side. - """ + """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.""" diff --git a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py index 6629e4b..95ec39c 100644 --- a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py +++ b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py @@ -38,9 +38,7 @@ async def send(self, message: str) -> None: async def recv(self) -> str: """Return the next scripted frame, raising once the script runs out. - Raising rather than blocking is deliberate: if a change makes the pump read past - the end of a turn, the test fails immediately instead of hanging until CI times - out, and the reason is visible in the failure. + 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") @@ -440,14 +438,10 @@ async def scenario() -> None: def test_the_adapter_satisfies_the_dialogue_layer_s_ports() -> None: - """The two sides of the seam are the same shape, and this is what checks it. - - The dialogue layer declares RealtimeSession and TurnObserver; the adapter declares - Transport and Observer. Neither imports the other, because infrastructure must not - depend on a product layer -- which also means nothing at all notices when one side - grows a parameter the other does not have. These assignments are that notice: mypy - rejects them the moment the shapes drift apart, and no call site is needed to provoke - it. + """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) From a308b9b9b60887d1b157a8933c2d6107d21a636f Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Tue, 11 Aug 2026 17:49:42 +0800 Subject: [PATCH 5/5] fix(realtime): close what a failed turn would otherwise leave open Two ways a turn could end while still holding something open, both found in review of this branch: A session whose `configure()` failed left its WebSocket open. The caller never receives the session, so nobody else can close it. The factory now closes the transport itself before re-raising, and `configure()` moved inside the open timeout -- it had none before. An exception while sending audio left the pump task sitting in `recv()`. Only `CancelledError` was cancelling it, so any other failure -- and the send path is where transport failures show up -- orphaned the task. Cancellation now happens on every exceptional exit. One test each: a transport that refuses to send, and a chunk iterator that raises mid-stream. --- .../external/realtime/qwen_audio.py | 12 +++-- .../timeflow/intelligence/realtime/agent.py | 6 ++- .../external/realtime/test_qwen_audio.py | 32 ++++++++++++++ .../realtime/test_realtime_agent.py | 44 +++++++++++++++++++ 4 files changed, 90 insertions(+), 4 deletions(-) diff --git a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py index fb811f9..32212db 100644 --- a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py +++ b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py @@ -250,12 +250,18 @@ def __init__( 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.""" + """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) - await session.configure(instructions, tools) + session = QwenAudioSession(transport, self._config) + try: + await session.configure(instructions, tools) + except BaseException: + await session.close() + raise return session diff --git a/backend/src/timeflow/intelligence/realtime/agent.py b/backend/src/timeflow/intelligence/realtime/agent.py index f4df201..64f903c 100644 --- a/backend/src/timeflow/intelligence/realtime/agent.py +++ b/backend/src/timeflow/intelligence/realtime/agent.py @@ -1,6 +1,7 @@ """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 @@ -79,8 +80,11 @@ async def handle_audio(self, chunks: AsyncIterator[bytes], stream: StreamInfo) - await session.finish_input() turn.note_input(sent_bytes) await pumping - except asyncio.CancelledError: + 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() diff --git a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py index 95ec39c..e2be11f 100644 --- a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py +++ b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py @@ -468,3 +468,35 @@ async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any 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 index bd63f72..a1878ab 100644 --- a/backend/tests/intelligence/realtime/test_realtime_agent.py +++ b/backend/tests/intelligence/realtime/test_realtime_agent.py @@ -364,3 +364,47 @@ async def scenario() -> None: 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())