From 1bc1beaddf32b1649ecf6f02b35753b50a0c0008 Mon Sep 17 00:00:00 2001 From: yyy-router <1804384725@qq.com> Date: Thu, 20 Aug 2026 16:22:10 +0800 Subject: [PATCH 1/4] feat(voice): add composed composition root and gateway adapter --- backend/src/timeflow/composition.py | 80 ++++++++++++ .../websocket/handlers/composed_audio.py | 118 ++++++++++++++++++ 2 files changed, 198 insertions(+) create mode 100644 backend/src/timeflow/composition.py create mode 100644 backend/src/timeflow/gateway/websocket/handlers/composed_audio.py diff --git a/backend/src/timeflow/composition.py b/backend/src/timeflow/composition.py new file mode 100644 index 00000000..79bc2c5b --- /dev/null +++ b/backend/src/timeflow/composition.py @@ -0,0 +1,80 @@ +"""Composition-root factory for the injectable composed voice agent.""" + +from sqlalchemy.orm import Session, sessionmaker + +from timeflow.business.calendar import ScheduleApplicationService +from timeflow.data.database import build_engine, build_session_factory +from timeflow.data.schedule_unit_of_work import SqlAlchemyScheduleUnitOfWork +from timeflow.infrastructure.external.asr.qwen_realtime import QwenRealtimeAsr +from timeflow.infrastructure.external.llm.openai_compatible import OpenAICompatibleLlm +from timeflow.infrastructure.external.tts.qwen_audio_tts import QwenAudioTts +from timeflow.infrastructure.settings import Settings +from timeflow.intelligence.composed.agent import ComposedVoiceAgent +from timeflow.intelligence.conversation.agent import Agent +from timeflow.intelligence.conversation.schedule_tools import ScheduleResultObserver +from timeflow.intelligence.conversation.tools import build_agent_tool_registry +from timeflow.intelligence.location import ClientLocation, LocationSearchService +from timeflow.intelligence.ports import ResultSink + + +def build_composed_voice_agent( + settings: Settings, + result_sink: ResultSink, + *, + session_factory: sessionmaker[Session] | None = None, + location_service: LocationSearchService | None = None, +) -> ComposedVoiceAgent: + """Build complete composed dependencies for the mode-2 gateway adapter.""" + _validate_settings(settings) + if session_factory is None: + engine = build_engine(settings.database_url) + session_factory = build_session_factory(engine) + schedule_service = ScheduleApplicationService( + lambda: SqlAlchemyScheduleUnitOfWork(session_factory) + ) + llm = OpenAICompatibleLlm(settings) + + def agent_factory( + account_id: str, + observer: ScheduleResultObserver, + client_location: ClientLocation | None, + ) -> Agent: + return Agent( + llm, + build_agent_tool_registry( + schedule_service, + account_id, + observer, + location_service=location_service, + client_location=client_location, + ), + max_tool_rounds=settings.agent_max_tool_rounds, + ) + + return ComposedVoiceAgent( + QwenRealtimeAsr(settings), + agent_factory, + QwenAudioTts(settings), + result_sink, + location_service=location_service, + ) + + +def _validate_settings(settings: Settings) -> None: + missing: list[str] = [] + for name, value in ( + ("TIMEFLOW_DATABASE_URL", settings.database_url), + ("TIMEFLOW_ALIYUN_ASR_WS_URL", settings.aliyun_asr_ws_url), + ("TIMEFLOW_ALIYUN_ASR_API_KEY", settings.aliyun_asr_api_key), + ("TIMEFLOW_OPENAI_BASE_URL", settings.openai_base_url), + ("TIMEFLOW_OPENAI_API_KEY", settings.openai_api_key), + ("TIMEFLOW_ALIYUN_TTS_WS_URL", settings.aliyun_tts_ws_url), + ("TIMEFLOW_ALIYUN_TTS_API_KEY", settings.aliyun_tts_api_key), + ): + if not value: + missing.append(name) + if missing: + raise RuntimeError(f"Composed voice agent is not configured: {', '.join(missing)}") + + +__all__ = ["build_composed_voice_agent"] diff --git a/backend/src/timeflow/gateway/websocket/handlers/composed_audio.py b/backend/src/timeflow/gateway/websocket/handlers/composed_audio.py new file mode 100644 index 00000000..d26dac4f --- /dev/null +++ b/backend/src/timeflow/gateway/websocket/handlers/composed_audio.py @@ -0,0 +1,118 @@ +"""Gateway adapter dedicated to the composed voice backend.""" + +from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Protocol + +from timeflow.gateway.websocket.ports import StreamContext + + +class AudioStreamInfo(Protocol): + """Provider-neutral stream context required by the composed backend.""" + + @property + def session_id(self) -> str: ... + + @property + def account_id(self) -> str: ... + + @property + def timezone(self) -> str: ... + + @property + def voice_mode(self) -> str: ... + + @property + def latitude(self) -> float | None: ... + + @property + def longitude(self) -> float | None: ... + + @property + def coordinate_system(self) -> str | None: ... + + @property + def stream_id(self) -> str: ... + + @property + def conversation_id(self) -> str: ... + + @property + def request_id(self) -> str | None: ... + + @property + def audio_format(self) -> str: ... + + @property + def sample_rate_hz(self) -> int: ... + + @property + def channels(self) -> int: ... + + +@dataclass(frozen=True, slots=True) +class _ComposedAudioStream: + session_id: str + account_id: str + timezone: str + voice_mode: str + latitude: float | None + longitude: float | None + coordinate_system: str | None + stream_id: str + conversation_id: str + request_id: str | None + audio_format: str + sample_rate_hz: int + channels: int + + +class ComposedAudioAgent(Protocol): + """Composed backend capabilities required by this adapter.""" + + async def handle_audio( + self, + chunks: AsyncIterator[bytes], + stream: AudioStreamInfo, + ) -> None: ... + + async def interrupt(self, session_id: str, reason: str) -> None: ... + + async def close_session(self, session_id: str) -> None: ... + + +class ComposedAgentAudioSink: + """Translate Gateway stream context for the composed voice backend.""" + + def __init__(self, agent: ComposedAudioAgent) -> None: + self._agent = agent + + async def consume(self, chunks: AsyncIterator[bytes], stream: StreamContext) -> None: + await self._agent.handle_audio(chunks, _stream_info(stream)) + + async def interrupt(self, session_id: str, reason: str) -> None: + await self._agent.interrupt(session_id, reason) + + async def close_session(self, session_id: str) -> None: + await self._agent.close_session(session_id) + + +def _stream_info(stream: StreamContext) -> _ComposedAudioStream: + return _ComposedAudioStream( + session_id=stream.session.session_id, + account_id=stream.session.account_id, + timezone=stream.timezone, + voice_mode=stream.voice_mode, + latitude=stream.latitude, + longitude=stream.longitude, + coordinate_system=stream.coordinate_system, + stream_id=stream.stream_id, + conversation_id=stream.conversation_id, + request_id=stream.request_id, + audio_format=stream.audio_config.audio_format, + sample_rate_hz=stream.audio_config.sample_rate_hz, + channels=stream.audio_config.channels, + ) + + +__all__ = ["ComposedAgentAudioSink"] From 14536c8ff460ca6096a57e43ac2b1a18acf2c6de Mon Sep 17 00:00:00 2001 From: yyy-router <1804384725@qq.com> Date: Thu, 20 Aug 2026 16:22:10 +0800 Subject: [PATCH 2/4] feat(voice): add session lifecycle contract to the voice gateway --- .../gateway/websocket/handlers/voice_stream.py | 6 ++++++ backend/src/timeflow/gateway/websocket/ports.py | 15 ++++++++++++++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py b/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py index d93c1bff..dae47d34 100644 --- a/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py +++ b/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py @@ -18,6 +18,7 @@ ) from timeflow.gateway.websocket.ports import ( AudioConfig, + AudioSessionLifecycle, AudioSink, SessionContext, StreamContext, @@ -105,6 +106,9 @@ async def handle_start( if invalid is not None: return self._error(request_id, invalid) + if isinstance(self._audio_sink, AudioSessionLifecycle): + await self._audio_sink.interrupt(session.session_id, "new_audio_stream") + audio_config = AudioConfig( audio_format=payload.audio_format, sample_rate_hz=payload.sample_rate_hz, @@ -192,6 +196,8 @@ async def handle_disconnect(self, session: SessionContext) -> None: task.cancel() if tasks: await asyncio.gather(*tasks, return_exceptions=True) + if isinstance(self._audio_sink, AudioSessionLifecycle): + await self._audio_sink.close_session(session.session_id) async def _drain_to_sink(self, stream: _ActiveStream) -> None: """Feed queued chunks to the sink once the first frame has arrived.""" diff --git a/backend/src/timeflow/gateway/websocket/ports.py b/backend/src/timeflow/gateway/websocket/ports.py index 410d722a..aba0e9c2 100644 --- a/backend/src/timeflow/gateway/websocket/ports.py +++ b/backend/src/timeflow/gateway/websocket/ports.py @@ -2,7 +2,7 @@ from collections.abc import AsyncIterator from dataclasses import dataclass -from typing import Literal, Protocol +from typing import Literal, Protocol, runtime_checkable @dataclass(frozen=True, slots=True) @@ -83,3 +83,16 @@ class AudioSink(Protocol): async def consume(self, chunks: AsyncIterator[bytes], stream: StreamContext) -> None: """Drain the chunk stream; returning means this stream is finished.""" ... + + +@runtime_checkable +class AudioSessionLifecycle(Protocol): + """Optional session-level lifecycle supported by stateful audio sinks.""" + + async def interrupt(self, session_id: str, reason: str) -> None: + """Stop output and provider work for the current turn.""" + ... + + async def close_session(self, session_id: str) -> None: + """Release all session-scoped state after disconnect.""" + ... From f88a3bb6c5a20b165e75d83b5d6775688dc4535e Mon Sep 17 00:00:00 2001 From: yyy-router <1804384725@qq.com> Date: Thu, 20 Aug 2026 16:22:10 +0800 Subject: [PATCH 3/4] feat(voice): wire mode-2 agent and prewarm TTS in the app --- .../external/tts/qwen_audio_tts.py | 10 +++- .../src/timeflow/infrastructure/settings.py | 6 ++- .../timeflow/intelligence/speech/pipeline.py | 43 ++++++++++------- backend/src/timeflow/main.py | 48 +++++++------------ 4 files changed, 58 insertions(+), 49 deletions(-) diff --git a/backend/src/timeflow/infrastructure/external/tts/qwen_audio_tts.py b/backend/src/timeflow/infrastructure/external/tts/qwen_audio_tts.py index 382d7fa4..80136271 100644 --- a/backend/src/timeflow/infrastructure/external/tts/qwen_audio_tts.py +++ b/backend/src/timeflow/infrastructure/external/tts/qwen_audio_tts.py @@ -214,7 +214,15 @@ async def _default_connector( timeout: float, ) -> WebSocketConnection: """Open a provider connection using the websockets asyncio client.""" - return await connect(url, additional_headers=dict(headers), open_timeout=timeout) + return await connect( + url, + additional_headers=dict(headers), + open_timeout=timeout, + # The provider does not acknowledge a close frame until its idle timeout + # (~10s) expires. All audio has already been delivered by then, so cap the + # closing handshake instead of blocking the turn on it. + close_timeout=0.2, + ) class QwenAudioTts(TtsPort): diff --git a/backend/src/timeflow/infrastructure/settings.py b/backend/src/timeflow/infrastructure/settings.py index a0f4a2d4..f3965339 100644 --- a/backend/src/timeflow/infrastructure/settings.py +++ b/backend/src/timeflow/infrastructure/settings.py @@ -37,7 +37,9 @@ class Settings: aliyun_asr_api_key: str = "" aliyun_asr_model: str = "qwen3-asr-flash-realtime" aliyun_asr_language: str = "zh" - aliyun_asr_vad_threshold: float = 0.0 + # Tuned above the vendor default (0.2): in a room with background sound, a + # lower threshold lets the server VAD treat noise as speech and fire spurious turns. + aliyun_asr_vad_threshold: float = 0.6 aliyun_asr_vad_silence_duration_ms: int = 400 aliyun_asr_connect_timeout_seconds: float = 10.0 aliyun_asr_finish_timeout_seconds: float = 10.0 @@ -76,7 +78,7 @@ def from_environment(cls, env_file: Path | str = ".env") -> "Settings": """Load settings from TIMEFLOW-prefixed environment variables.""" load_dotenv(dotenv_path=env_file, override=False) - aliyun_asr_vad_threshold = float(environ.get("TIMEFLOW_ALIYUN_ASR_VAD_THRESHOLD", "0.0")) + aliyun_asr_vad_threshold = float(environ.get("TIMEFLOW_ALIYUN_ASR_VAD_THRESHOLD", "0.6")) aliyun_asr_vad_silence_duration_ms = int( environ.get("TIMEFLOW_ALIYUN_ASR_VAD_SILENCE_DURATION_MS", "400") ) diff --git a/backend/src/timeflow/intelligence/speech/pipeline.py b/backend/src/timeflow/intelligence/speech/pipeline.py index d0b67c6c..ffc3fce5 100644 --- a/backend/src/timeflow/intelligence/speech/pipeline.py +++ b/backend/src/timeflow/intelligence/speech/pipeline.py @@ -100,10 +100,14 @@ async def _stream( queue: asyncio.Queue[SpeechSegment | _EndOfSegments] = asyncio.Queue( maxsize=self._segment_queue_size ) - first_segment_ready = asyncio.Event() + # Fires as soon as the first non-empty text or question arrives, before the + # segmenter has necessarily produced a complete segment. The TTS stream is + # started on this signal so its WebSocket connect + handshake overlaps the + # remaining segmentation instead of serialising after the first segment. + speech_expected = asyncio.Event() metadata = _StreamMetadata() producer = asyncio.create_task( - self._produce_segments(events, queue, first_segment_ready, metadata) + self._produce_segments(events, queue, speech_expected, metadata) ) tts_stream: AsyncIterator[TtsAudioChunk | TtsCompleted] | None = None @@ -119,7 +123,7 @@ async def segment_stream() -> AsyncIterator[SpeechSegment]: queue.task_done() try: - await self._wait_for_first_segment(first_segment_ready, producer) + await self._wait_for_speech(speech_expected, producer) if metadata.purpose is None: await producer return @@ -171,7 +175,7 @@ async def _produce_segments( self, events: AsyncIterable[AgentEvent], queue: asyncio.Queue[SpeechSegment | _EndOfSegments], - first_segment_ready: asyncio.Event, + speech_expected: asyncio.Event, metadata: _StreamMetadata, ) -> None: segmenter = TextSegmenter( @@ -180,31 +184,38 @@ async def _produce_segments( ) next_index = 0 + def note_purpose(purpose: SpeechPurpose) -> None: + """Record the turn's purpose and signal that speech is coming.""" + if metadata.purpose is None: + metadata.purpose = purpose + speech_expected.set() + elif metadata.purpose != purpose: + raise ValueError("One speech turn cannot mix question and reply text") + async def submit(text: str, purpose: SpeechPurpose) -> None: nonlocal next_index normalized = text.strip() if not normalized: return - if metadata.purpose is None: - metadata.purpose = purpose - if purpose == "dialogue_question": - metadata.speech_text = normalized - elif metadata.purpose != purpose: - raise ValueError("One speech turn cannot mix question and reply text") await queue.put(SpeechSegment(next_index, normalized, purpose)) next_index += 1 - first_segment_ready.set() cancelled = False try: async for event in events: if isinstance(event, AgentTextDelta): + if event.text.strip(): + note_purpose("command_result") for segment in segmenter.push(event.text): await submit(segment, "command_result") elif isinstance(event, AgentQuestion): if metadata.purpose is not None or segmenter.flush() is not None: raise ValueError("One speech turn cannot mix question and reply text") - await submit(event.speech_text, "dialogue_question") + normalized = event.speech_text.strip() + if normalized: + note_purpose("dialogue_question") + metadata.speech_text = normalized + await submit(event.speech_text, "dialogue_question") return elif isinstance(event, AgentCompleted): if remainder := segmenter.flush(): @@ -221,14 +232,14 @@ async def submit(text: str, purpose: SpeechPurpose) -> None: finally: if not cancelled: await queue.put(_END_OF_SEGMENTS) - first_segment_ready.set() + speech_expected.set() @staticmethod - async def _wait_for_first_segment( - first_segment_ready: asyncio.Event, + async def _wait_for_speech( + speech_expected: asyncio.Event, producer: asyncio.Task[None], ) -> None: - waiter = asyncio.create_task(first_segment_ready.wait()) + waiter = asyncio.create_task(speech_expected.wait()) try: done, _ = await asyncio.wait( {waiter, producer}, diff --git a/backend/src/timeflow/main.py b/backend/src/timeflow/main.py index d367ec54..28c9df78 100644 --- a/backend/src/timeflow/main.py +++ b/backend/src/timeflow/main.py @@ -19,6 +19,7 @@ ) from timeflow.business.calendar.service import ScheduleApplicationService from timeflow.business.health import HealthService +from timeflow.composition import build_composed_voice_agent from timeflow.data.account_uow import SqlAlchemyAuthUnitOfWork from timeflow.data.database import build_engine, build_session_factory from timeflow.data.schedule_unit_of_work import SqlAlchemyScheduleUnitOfWork @@ -39,6 +40,7 @@ ) from timeflow.gateway.websocket.handlers.agent_audio import AgentAudioSink from timeflow.gateway.websocket.handlers.agent_result import WebSocketResultSink +from timeflow.gateway.websocket.handlers.composed_audio import ComposedAgentAudioSink from timeflow.gateway.websocket.handlers.message_ack import handle_message_ack from timeflow.gateway.websocket.handlers.session import SessionHandshake from timeflow.gateway.websocket.handlers.voice_stream import VoiceStreamHandlers @@ -115,15 +117,10 @@ def create_app( lambda: SqlAlchemyScheduleUnitOfWork(session_factory) ) - # Gated on mode "1": mode "2" never reaches a live turn (see _build_agent below), so - # building this client for it would just leak a connection nobody ever closes -- the - # mode-2 branch raises before create_app() returns, so lifespan's finally never runs. + # Both agent modes share the same owned Tencent HTTP client and location service; the + # client is closed by lifespan's finally once the application shuts down. location_service: LocationSearchService | None = None - if ( - audio_sink is None - and settings.voice_agent_mode == "1" - and settings.tencent_maps_is_configured() - ): + if audio_sink is None and settings.tencent_maps_is_configured(): owned_http_client = httpx.AsyncClient(timeout=settings.tencent_map_timeout_seconds) location_service = LocationSearchService( TencentMapsLocationPort( @@ -160,9 +157,19 @@ async def lifespan(_application: FastAPI) -> AsyncIterator[None]: if audio_sink is None: assert session_factory is not None result_sink = WebSocketResultSink(connections) - audio_sink = AgentAudioSink( - _build_agent(settings, result_sink, session_factory, location_service) - ) + if settings.voice_agent_mode == "1": + audio_sink = AgentAudioSink( + _build_realtime_agent(settings, result_sink, session_factory, location_service) + ) + else: + audio_sink = ComposedAgentAudioSink( + build_composed_voice_agent( + settings, + result_sink, + session_factory=session_factory, + location_service=location_service, + ) + ) voice_streams = VoiceStreamHandlers( audio_sink, @@ -221,25 +228,6 @@ def _build_access_token_service(settings: Settings) -> JwtAccessTokenService: ) -def _build_agent( - settings: Settings, - result_sink: WebSocketResultSink, - session_factory: sessionmaker[Session], - location_service: LocationSearchService | None, -) -> Agent: - """Dispatch on TIMEFLOW_VOICE_AGENT_MODE to the agent backend it selects.""" - if settings.voice_agent_mode == "1": - return _build_realtime_agent(settings, result_sink, session_factory, location_service) - if settings.voice_agent_mode == "2": - raise RuntimeError( - "TIMEFLOW_VOICE_AGENT_MODE=2 selects the LLM+ASR+TTS conversation agent, " - "which is not wired into the gateway yet (it does not implement the Agent " - "port). Set TIMEFLOW_VOICE_AGENT_MODE=1 to use the realtime agent." - ) - # Settings.from_environment already rejects anything but "1" or "2". - raise AssertionError(f"unreachable voice_agent_mode: {settings.voice_agent_mode!r}") - - def _build_realtime_agent( settings: Settings, result_sink: WebSocketResultSink, From 945e2e408a37d5fe3e9af2ea3106ab8bc2a1b52f Mon Sep 17 00:00:00 2001 From: yyy-router <1804384725@qq.com> Date: Thu, 20 Aug 2026 16:22:10 +0800 Subject: [PATCH 4/4] test(voice): cover composed gateway wiring and prewarm --- .../intelligence/speech/test_pipeline.py | 52 ++++++++++++++ backend/tests/test_agent_lifecycle.py | 71 +++++++++++++++++++ backend/tests/test_app_wiring.py | 49 ++++++++++--- backend/tests/test_composed_factory.py | 69 ++++++++++++++++++ backend/tests/test_settings.py | 2 +- backend/tests/test_ws_voice_stream.py | 34 +++++++++ 6 files changed, 265 insertions(+), 12 deletions(-) create mode 100644 backend/tests/test_agent_lifecycle.py create mode 100644 backend/tests/test_composed_factory.py diff --git a/backend/tests/intelligence/speech/test_pipeline.py b/backend/tests/intelligence/speech/test_pipeline.py index f31cc556..76c8676e 100644 --- a/backend/tests/intelligence/speech/test_pipeline.py +++ b/backend/tests/intelligence/speech/test_pipeline.py @@ -201,6 +201,58 @@ async def test_empty_turn_does_not_start_tts() -> None: assert tts.requests == [] +@pytest.mark.asyncio +async def test_tts_starts_before_first_segment_is_complete() -> None: + tts_started = asyncio.Event() + release_boundary = asyncio.Event() + + class PrewarmTts: + def stream(self, segments: AsyncIterable[SpeechSegment]) -> AsyncIterator[TtsEvent]: + return self._stream(segments) + + async def _stream(self, segments: AsyncIterable[SpeechSegment]) -> AsyncIterator[TtsEvent]: + tts_started.set() + iterator = aiter(segments) + _ = await anext(iterator) + yield TtsAudioChunk(b"audio") + with pytest.raises(StopAsyncIteration): + await anext(iterator) + yield TtsCompleted(2) + + async def events() -> AsyncIterator[object]: + yield AgentTextDelta("第一句") # no strong boundary yet, no segment + await release_boundary.wait() + yield AgentTextDelta("。") + yield AgentCompleted(None) + + pipeline = SpeechPipeline(PrewarmTts()) + stream = pipeline.stream(events()) + + output: list[object] = [] + + async def consume() -> None: + async for event in stream: + output.append(event) + + task = asyncio.create_task(consume()) + # The TTS stream must have started (its connect would be in flight) while the + # first sentence is still incomplete and no segment has been produced. + await asyncio.wait_for(tts_started.wait(), timeout=1) + assert release_boundary.is_set() is False + + release_boundary.set() + await asyncio.wait_for(task, timeout=1) + + started = output[0] + assert isinstance(started, SpeechAudioStarted) + assert started.purpose == "command_result" + assert started.speech_text == "" + assert output[1:] == [ + SpeechAudioChunk(started.audio_id, b"audio"), + SpeechAudioCompleted(started.audio_id, 2), + ] + + @pytest.mark.asyncio async def test_mixed_question_and_reply_is_rejected() -> None: pipeline = SpeechPipeline(FakeTts()) diff --git a/backend/tests/test_agent_lifecycle.py b/backend/tests/test_agent_lifecycle.py new file mode 100644 index 00000000..90fe7951 --- /dev/null +++ b/backend/tests/test_agent_lifecycle.py @@ -0,0 +1,71 @@ +"""Lifecycle and context translation for the composed Gateway adapter.""" + +import asyncio +from collections.abc import AsyncIterator +from dataclasses import dataclass, field +from typing import Any + +from timeflow.gateway.websocket.handlers.composed_audio import ComposedAgentAudioSink +from timeflow.gateway.websocket.ports import AudioConfig, SessionContext, StreamContext + + +@dataclass +class RecordingAgent: + """Record audio and lifecycle calls made by the composed adapter.""" + + calls: list[tuple[str, Any]] = field(default_factory=list) + + async def handle_audio(self, chunks: AsyncIterator[bytes], stream: Any) -> None: + """Record the lifted stream context and audio bytes.""" + audio = b"".join([chunk async for chunk in chunks]) + self.calls.append(("audio", (audio, stream))) + + async def interrupt(self, session_id: str, reason: str) -> None: + """Record an interruption.""" + self.calls.append(("interrupt", (session_id, reason))) + + async def close_session(self, session_id: str) -> None: + """Record session cleanup.""" + self.calls.append(("close", session_id)) + + +async def _chunks() -> AsyncIterator[bytes]: + yield b"first" + yield b"second" + + +def test_composed_audio_sink_forwards_context_and_lifecycle() -> None: + """The adapter carries composed metadata without changing the realtime adapter.""" + + async def scenario() -> None: + agent = RecordingAgent() + sink = ComposedAgentAudioSink(agent) + context = StreamContext( + stream_id="stream_1", + conversation_id="conversation_1", + session=SessionContext("session_1", "account_1", "device_1", "Asia/Shanghai"), + audio_config=AudioConfig("pcm_s16le", 16000, 1), + request_id="request_1", + ) + + await sink.consume(_chunks(), context) + await sink.interrupt("session_1", "user_interrupted") + await sink.close_session("session_1") + + _, (audio, stream) = agent.calls[0] + assert audio == b"firstsecond" + assert stream.session_id == "session_1" + assert stream.account_id == "account_1" + assert stream.timezone == "Asia/Shanghai" + assert stream.stream_id == "stream_1" + assert stream.conversation_id == "conversation_1" + assert stream.request_id == "request_1" + assert stream.audio_format == "pcm_s16le" + assert stream.sample_rate_hz == 16000 + assert stream.channels == 1 + assert agent.calls[1:] == [ + ("interrupt", ("session_1", "user_interrupted")), + ("close", "session_1"), + ] + + asyncio.run(scenario()) diff --git a/backend/tests/test_app_wiring.py b/backend/tests/test_app_wiring.py index 119d4351..36633dff 100644 --- a/backend/tests/test_app_wiring.py +++ b/backend/tests/test_app_wiring.py @@ -115,6 +115,18 @@ def _tencent_environment() -> dict[str, str]: } +def _composed_environment() -> dict[str, str]: + """模式 2 组合式 ASR/LLM/TTS 凭据就绪时的最小环境变量集合。""" + return { + "TIMEFLOW_ALIYUN_ASR_WS_URL": "wss://asr.example.test", + "TIMEFLOW_ALIYUN_ASR_API_KEY": "asr-key", + "TIMEFLOW_OPENAI_BASE_URL": "https://llm.example.test/v1", + "TIMEFLOW_OPENAI_API_KEY": "llm-key", + "TIMEFLOW_ALIYUN_TTS_WS_URL": "wss://tts.example.test", + "TIMEFLOW_ALIYUN_TTS_API_KEY": "tts-key", + } + + def _build_with_environment( environment: str, *, @@ -474,10 +486,19 @@ def capture_service(*args: Any, **kwargs: Any) -> ScheduleApplicationService: classifier_factory.assert_called_once_with(llm_factory.return_value) -def test_voice_agent_mode_two_fails_closed_until_conversation_agent_is_wired() -> None: - """未实现 Agent 端口前，不能静默选择 LLM+ASR+TTS 模式。""" - with pytest.raises(RuntimeError, match="TIMEFLOW_VOICE_AGENT_MODE=2"): - _build_with_environment("development", voice_agent_mode="2") +def test_voice_agent_mode_two_fails_closed_without_composed_credentials() -> None: + """模式 2 缺少组合式 ASR/LLM/TTS 凭据时快速失败，而不是静默降级。""" + missing = { + "TIMEFLOW_ALIYUN_ASR_WS_URL": "", + "TIMEFLOW_ALIYUN_ASR_API_KEY": "", + "TIMEFLOW_OPENAI_BASE_URL": "", + "TIMEFLOW_OPENAI_API_KEY": "", + "TIMEFLOW_ALIYUN_TTS_WS_URL": "", + "TIMEFLOW_ALIYUN_TTS_API_KEY": "", + } + with mock.patch.dict(os.environ, missing, clear=False): + with pytest.raises(RuntimeError, match="Composed voice agent is not configured"): + _build_with_environment("development", voice_agent_mode="2") def test_lifespan_disposes_the_database_engine_owned_by_the_application() -> None: @@ -544,13 +565,19 @@ def test_lifespan_closes_the_owned_tencent_http_client() -> None: assert client.closed is True -def test_voice_agent_mode_two_never_opens_a_tencent_http_client() -> None: - """mode=2 在能建出 Agent 之前就已失败，不该先泄漏一个没人关闭的 HTTP client。""" +def test_voice_agent_mode_two_opens_and_closes_a_tencent_http_client() -> None: + """模式 2 与模式 1 一样由组合根创建并释放腾讯地图 HTTP client。""" + client = _FakeAsyncClient(timeout=5.0) + environment = {**_tencent_environment(), **_composed_environment()} + with ( - mock.patch("timeflow.main.httpx.AsyncClient") as factory, - mock.patch.dict(os.environ, _tencent_environment(), clear=False), - pytest.raises(RuntimeError, match="TIMEFLOW_VOICE_AGENT_MODE=2"), + mock.patch("timeflow.main.httpx.AsyncClient", return_value=client) as factory, + mock.patch.dict(os.environ, environment, clear=False), ): - _build_with_environment("development", voice_agent_mode="2") + application = _build_with_environment("development", voice_agent_mode="2") + + factory.assert_called_once() + with TestClient(application): + assert client.closed is False - factory.assert_not_called() + assert client.closed is True diff --git a/backend/tests/test_composed_factory.py b/backend/tests/test_composed_factory.py new file mode 100644 index 00000000..19434e59 --- /dev/null +++ b/backend/tests/test_composed_factory.py @@ -0,0 +1,69 @@ +"""Composition-root validation for the injectable composed voice agent.""" + +from dataclasses import replace + +import pytest + +from timeflow.composition import build_composed_voice_agent +from timeflow.infrastructure.settings import Settings +from timeflow.intelligence.composed import ComposedVoiceAgent + + +class NullResultSink: + async def deliver_transcript(self, transcript: object, stream: object) -> None: ... + + async def deliver_reply_text(self, reply: object, stream: object) -> None: ... + + async def deliver_result(self, result: object, stream: object) -> None: ... + + async def deliver_question(self, question: object, stream: object) -> None: ... + + async def deliver_audio(self, reply: object, chunks: object, stream: object) -> None: ... + + async def deliver_canceled(self, canceled: object, stream: object) -> None: ... + + async def deliver_session_end(self, stream: object) -> None: ... + + +def settings() -> Settings: + return Settings( + app_name="TimeFlow", + environment="development", + database_url="sqlite+pysqlite:///:memory:", + ws_handshake_timeout_seconds=5, + ws_max_unauthenticated_connections=10, + ws_audio_queue_max_chunks=8, + ws_max_audio_duration_ms=120000, + aliyun_asr_ws_url="wss://asr.example.test", + aliyun_asr_api_key="asr-test-key", + openai_base_url="https://llm.example.test/v1", + openai_api_key="llm-test-key", + aliyun_tts_ws_url="wss://tts.example.test", + aliyun_tts_api_key="tts-test-key", + ) + + +def test_factory_builds_injectable_agent_without_changing_app_default() -> None: + agent = build_composed_voice_agent(settings(), NullResultSink()) + + assert isinstance(agent, ComposedVoiceAgent) + + +@pytest.mark.parametrize( + ("field", "setting_name"), + [ + ("database_url", "TIMEFLOW_DATABASE_URL"), + ("aliyun_asr_ws_url", "TIMEFLOW_ALIYUN_ASR_WS_URL"), + ("aliyun_asr_api_key", "TIMEFLOW_ALIYUN_ASR_API_KEY"), + ("openai_base_url", "TIMEFLOW_OPENAI_BASE_URL"), + ("openai_api_key", "TIMEFLOW_OPENAI_API_KEY"), + ("aliyun_tts_ws_url", "TIMEFLOW_ALIYUN_TTS_WS_URL"), + ("aliyun_tts_api_key", "TIMEFLOW_ALIYUN_TTS_API_KEY"), + ], +) +def test_factory_fails_fast_for_missing_composed_configuration( + field: str, + setting_name: str, +) -> None: + with pytest.raises(RuntimeError, match=setting_name): + build_composed_voice_agent(replace(settings(), **{field: ""}), NullResultSink()) diff --git a/backend/tests/test_settings.py b/backend/tests/test_settings.py index 122a6969..ea09e4a7 100644 --- a/backend/tests/test_settings.py +++ b/backend/tests/test_settings.py @@ -146,7 +146,7 @@ def test_settings_use_qwen_asr_defaults( assert settings.aliyun_asr_api_key == "" assert settings.aliyun_asr_model == "qwen3-asr-flash-realtime" assert settings.aliyun_asr_language == "zh" - assert settings.aliyun_asr_vad_threshold == 0.0 + assert settings.aliyun_asr_vad_threshold == 0.6 assert settings.aliyun_asr_vad_silence_duration_ms == 400 assert settings.aliyun_asr_connect_timeout_seconds == 10.0 assert settings.aliyun_asr_finish_timeout_seconds == 10.0 diff --git a/backend/tests/test_ws_voice_stream.py b/backend/tests/test_ws_voice_stream.py index 201f2423..9601adf6 100644 --- a/backend/tests/test_ws_voice_stream.py +++ b/backend/tests/test_ws_voice_stream.py @@ -60,6 +60,20 @@ def audio(self) -> bytes: return b"".join(self.chunks) +class LifecycleSink(CapturingSink): + """Record session lifecycle callbacks in addition to consumed audio.""" + + def __init__(self) -> None: + super().__init__() + self.lifecycle: list[tuple[str, str]] = [] + + async def interrupt(self, session_id: str, reason: str) -> None: + self.lifecycle.append(("interrupt", f"{session_id}:{reason}")) + + async def close_session(self, session_id: str) -> None: + self.lifecycle.append(("close", session_id)) + + class ExplodingSink: """A sink that fails the way a provider outage would.""" @@ -109,6 +123,26 @@ async def endpoint(websocket: WebSocket) -> None: return application +def test_stateful_sink_receives_start_interruption_and_disconnect_cleanup() -> None: + """A composed sink gets lifecycle hooks while stateless sinks remain supported.""" + sink = LifecycleSink() + client = TestClient(_build_app(sink)) + + with client.websocket_connect("/ws?device_id=device_001") as websocket: + websocket.send_json(VALID_HELLO) + websocket.receive_json() + websocket.send_json(START) + websocket.receive_json() + websocket.send_bytes(b"\x01\x02") + websocket.send_json({"type": "voice.stream.end", "payload": {"stream_id": "stream_test"}}) + assert sink.completed.wait(timeout=2) + + assert sink.lifecycle == [ + ("interrupt", "ws_session_test:new_audio_stream"), + ("close", "ws_session_test"), + ] + + def test_stream_start_assigns_stream_and_conversation_ids() -> None: """voice.stream.start is acknowledged with both identifiers.""" sink = CapturingSink()