feat(voice): wire composed voice into gateway and prewarm TTS

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"]

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"]

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."""

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."""
+        ...

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):

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")
         )

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},