diff --git a/backend/src/timeflow/gateway/websocket/agent_ports.py b/backend/src/timeflow/gateway/websocket/agent_ports.py new file mode 100644 index 00000000..b95a4dc6 --- /dev/null +++ b/backend/src/timeflow/gateway/websocket/agent_ports.py @@ -0,0 +1,108 @@ +"""What the transport needs from a dialogue agent, stated on the transport's own terms.""" + +from collections.abc import AsyncIterator +from typing import Any, Protocol + + +class StreamIdentity(Protocol): + """Identifiers of the audio stream a result belongs to.""" + + @property + def session_id(self) -> str: + """Session the stream belongs to.""" + ... + + @property + def stream_id(self) -> str: + """The stream itself.""" + ... + + @property + def conversation_id(self) -> str: + """Conversation the stream continues.""" + ... + + @property + def request_id(self) -> str | None: + """Request that opened the stream, when the client supplied one.""" + ... + + +class TranscriptResult(Protocol): + """What the user was heard to say.""" + + @property + def text(self) -> str: + """The transcribed words.""" + ... + + @property + def language(self) -> str: + """Language the words were recognized as.""" + ... + + @property + def duration_ms(self) -> int: + """How long the audio ran.""" + ... + + +class CommandOutcome(Protocol): + """A command that was carried out, ready to be sent to the client.""" + + @property + def message_id(self) -> str: + """Identifier the client echoes back in message.ack.""" + ... + + @property + def operation(self) -> str: + """Command that was carried out.""" + ... + + @property + def status(self) -> str: + """Outcome of that command.""" + ... + + @property + def schedule(self) -> dict[str, Any]: + """Persisted schedule snapshot the command produced.""" + ... + + +class AudioReplyInfo(Protocol): + """Format and purpose of a spoken reply, announced before its audio.""" + + @property + def audio_id(self) -> str: + """Identifier distinguishing this reply from the next.""" + ... + + @property + def audio_format(self) -> str: + """Encoding of the audio frames that follow.""" + ... + + @property + def sample_rate_hz(self) -> int: + """Sample rate the frames were produced at.""" + ... + + @property + def purpose(self) -> str: + """Why the reply is being spoken.""" + ... + + @property + def speech_text(self) -> str: + """The words the audio says.""" + ... + + +class Agent(Protocol): + """Take one audio stream and act on what it contains.""" + + async def handle_audio(self, chunks: AsyncIterator[bytes], stream: StreamIdentity) -> None: + """Consume the audio; returning only confirms receipt, not a result.""" + ... diff --git a/backend/src/timeflow/gateway/websocket/handlers/agent_audio.py b/backend/src/timeflow/gateway/websocket/handlers/agent_audio.py new file mode 100644 index 00000000..43803b3e --- /dev/null +++ b/backend/src/timeflow/gateway/websocket/handlers/agent_audio.py @@ -0,0 +1,39 @@ +"""Audio sink that forwards an inbound stream to the dialogue agent unchanged.""" + +from collections.abc import AsyncIterator +from dataclasses import dataclass + +from timeflow.gateway.websocket.agent_ports import Agent +from timeflow.gateway.websocket.ports import StreamContext + + +@dataclass(frozen=True, slots=True) +class _StreamIdentity: + """Identifiers lifted out of a stream context for the agent.""" + + session_id: str + stream_id: str + conversation_id: str + request_id: str | None + + +class AgentAudioSink: + """Hand each inbound stream to the agent as raw audio.""" + + def __init__(self, agent: Agent) -> None: + """Store the agent that receives the audio.""" + self._agent = agent + + async def consume(self, chunks: AsyncIterator[bytes], stream: StreamContext) -> None: + """Forward the stream unchanged, then return once the agent has taken it.""" + await self._agent.handle_audio(chunks, _identity_of(stream)) + + +def _identity_of(stream: StreamContext) -> _StreamIdentity: + """Lift the identifiers the agent needs out of the transport's context.""" + return _StreamIdentity( + session_id=stream.session.session_id, + stream_id=stream.stream_id, + conversation_id=stream.conversation_id, + request_id=stream.request_id, + ) diff --git a/backend/src/timeflow/gateway/websocket/handlers/agent_result.py b/backend/src/timeflow/gateway/websocket/handlers/agent_result.py new file mode 100644 index 00000000..6aa19c87 --- /dev/null +++ b/backend/src/timeflow/gateway/websocket/handlers/agent_result.py @@ -0,0 +1,95 @@ +"""Result sink that pushes transcripts, command results and spoken replies to the client.""" + +import logging +from collections.abc import AsyncIterator + +from timeflow.gateway.websocket.agent_ports import ( + AudioReplyInfo, + CommandOutcome, + StreamIdentity, + TranscriptResult, +) +from timeflow.gateway.websocket.connection_manager import ConnectionManager +from timeflow.gateway.websocket.messages.agent import ( + VoiceAsrCompleted, + VoiceAsrCompletedPayload, + VoiceCommandResult, + VoiceCommandResultPayload, +) +from timeflow.gateway.websocket.messages.tts import ( + VoiceTtsEnd, + VoiceTtsStart, + VoiceTtsStartPayload, +) + +logger = logging.getLogger(__name__) + + +class WebSocketResultSink: + """Translate results into wire messages and push them to the session.""" + + def __init__(self, connections: ConnectionManager) -> None: + """Store the registry used to reach the session.""" + self._connections = connections + + async def deliver_transcript( + self, transcript: TranscriptResult, stream: StreamIdentity + ) -> None: + """Push what the user was heard to say.""" + message = VoiceAsrCompleted( + request_id=stream.request_id, + conversation_id=stream.conversation_id, + payload=VoiceAsrCompletedPayload( + transcript=transcript.text, + language=transcript.language, + duration_ms=transcript.duration_ms, + ), + ) + await self._send(stream.session_id, message.type, message.model_dump()) + + async def deliver_result(self, result: CommandOutcome, stream: StreamIdentity) -> None: + """Push the outcome of a command the agent carried out.""" + message = VoiceCommandResult( + message_id=result.message_id, + request_id=stream.request_id, + conversation_id=stream.conversation_id, + payload=VoiceCommandResultPayload( + operation=result.operation, + status=result.status, + schedule=result.schedule, + ), + ) + await self._send(stream.session_id, message.type, message.model_dump()) + + async def deliver_audio( + self, reply: AudioReplyInfo, chunks: AsyncIterator[bytes], stream: StreamIdentity + ) -> None: + """Speak a reply, putting each chunk on the wire as it is produced.""" + start = VoiceTtsStart( + conversation_id=stream.conversation_id, + audio_id=reply.audio_id, + payload=VoiceTtsStartPayload( + format=reply.audio_format, + sample_rate_hz=reply.sample_rate_hz, + purpose=reply.purpose, + speech_text=reply.speech_text, + ), + ) + end = VoiceTtsEnd(conversation_id=stream.conversation_id, audio_id=reply.audio_id) + + delivered = await self._connections.stream_audio( + stream.session_id, start.model_dump(), chunks, end.model_dump() + ) + if not delivered: + logger.info( + "stopped speaking to a session that had gone", + extra={"session_id": stream.session_id, "audio_id": reply.audio_id}, + ) + + async def _send(self, session_id: str, message_type: str, envelope: dict[str, object]) -> None: + """Send one message, logging rather than raising when the session has gone.""" + if not await self._connections.send(session_id, envelope): + logger.info( + "dropped a result for a session that had gone", + extra={"session_id": session_id, "message_type": message_type}, + ) diff --git a/backend/src/timeflow/gateway/websocket/handlers/message_ack.py b/backend/src/timeflow/gateway/websocket/handlers/message_ack.py new file mode 100644 index 00000000..860552ec --- /dev/null +++ b/backend/src/timeflow/gateway/websocket/handlers/message_ack.py @@ -0,0 +1,34 @@ +"""The message.ack confirmation, recorded and never answered.""" + +import logging +from typing import Any + +from pydantic import ValidationError + +from timeflow.gateway.websocket.envelope import ERROR_MALFORMED_MESSAGE, build_error_envelope +from timeflow.gateway.websocket.messages.agent import MessageAck +from timeflow.gateway.websocket.ports import SessionContext + +logger = logging.getLogger(__name__) + + +async def handle_message_ack( + raw_message: dict[str, Any], session: SessionContext +) -> dict[str, Any] | None: + """Record an applied command result and reply with nothing.""" + try: + ack = MessageAck.model_validate(raw_message) + except ValidationError: + return build_error_envelope( + "message.ack", None, ERROR_MALFORMED_MESSAGE, "message.ack payload is invalid" + ) + + logger.info( + "client acknowledged a command result", + extra={ + "session_id": session.session_id, + "message_id": ack.message_id, + "status": ack.status, + }, + ) + return None diff --git a/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py b/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py index eeb22af6..87fd0f3f 100644 --- a/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py +++ b/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py @@ -108,6 +108,7 @@ async def handle_start( conversation_id=payload.conversation_id or self._conversation_id_factory(), session=session, audio_config=audio_config, + request_id=request_id, ) stream = _ActiveStream( context=context, diff --git a/backend/src/timeflow/gateway/websocket/messages/agent.py b/backend/src/timeflow/gateway/websocket/messages/agent.py new file mode 100644 index 00000000..6a6b7650 --- /dev/null +++ b/backend/src/timeflow/gateway/websocket/messages/agent.py @@ -0,0 +1,48 @@ +"""Transcript and command result messages pushed after a stream ends.""" + +from typing import Any, Literal + +from pydantic import BaseModel + + +class VoiceAsrCompletedPayload(BaseModel): + """What the user was heard to say.""" + + transcript: str + language: str + duration_ms: int + + +class VoiceAsrCompleted(BaseModel): + """Server message carrying the final transcript of a stream.""" + + type: Literal["voice.asr.completed"] = "voice.asr.completed" + request_id: str | None = None + conversation_id: str + payload: VoiceAsrCompletedPayload + + +class VoiceCommandResultPayload(BaseModel): + """The command that was carried out and the schedule it produced.""" + + operation: str + status: str + schedule: dict[str, Any] + + +class VoiceCommandResult(BaseModel): + """Server message carrying a committed command result.""" + + type: Literal["voice.command.result"] = "voice.command.result" + message_id: str + request_id: str | None = None + conversation_id: str + payload: VoiceCommandResultPayload + + +class MessageAck(BaseModel): + """Client confirmation that a command result was applied locally.""" + + type: Literal["message.ack"] + message_id: str + status: str diff --git a/backend/src/timeflow/gateway/websocket/ports.py b/backend/src/timeflow/gateway/websocket/ports.py index 439ae218..e7dcb00e 100644 --- a/backend/src/timeflow/gateway/websocket/ports.py +++ b/backend/src/timeflow/gateway/websocket/ports.py @@ -31,6 +31,7 @@ class StreamContext: conversation_id: str session: SessionContext audio_config: AudioConfig + request_id: str | None = None class TokenVerifier(Protocol): diff --git a/backend/src/timeflow/intelligence/fake_agent.py b/backend/src/timeflow/intelligence/fake_agent.py new file mode 100644 index 00000000..6634ec3e --- /dev/null +++ b/backend/src/timeflow/intelligence/fake_agent.py @@ -0,0 +1,94 @@ +"""Agent that answers every stream with the same fixed result, until a real one exists.""" + +import logging +from collections.abc import AsyncIterator, Callable +from typing import Any +from uuid import uuid4 + +from timeflow.intelligence.ports import CommandResult, ResultSink, StreamInfo, Transcript + +logger = logging.getLogger(__name__) + +FAKE_TRANSCRIPT = "明天下午三点在203开会" +FAKE_LANGUAGE = "zh" +FAKE_OPERATION = "create_schedule" +FAKE_STATUS = "applied" + + +def new_message_id() -> str: + """Return a fresh identifier for a result the client must acknowledge.""" + return f"msg_{uuid4().hex}" + + +def fake_schedule(schedule_id: str) -> dict[str, Any]: + """Return a persisted-looking schedule snapshot.""" + return { + "id": schedule_id, + "schedule_type": "time", + "schedule_kind": "once", + "title": "开会", + "is_all_day": False, + "start_time": "2026-08-07T07:00:00Z", + "end_time": None, + "timezone": "Asia/Shanghai", + "recurrence_rule": None, + "location_name": "203", + "latitude": None, + "longitude": None, + "status": "active", + "reminder_type": "before_start", + "reminder_trigger_at": None, + "reminder_offset_minutes": 15, + "reminder_strength": "medium", + "reminder_disposition_state": None, + "revision": 1, + "updated_at": "2026-08-07T06:45:00Z", + } + + +class FakeAgent: + """Drain a stream, then deliver one fixed successful result.""" + + def __init__( + self, + result_sink: ResultSink, + *, + message_id_factory: Callable[[], str] | None = None, + schedule_id: str = "schedule_fake_001", + ) -> None: + """Store the sink plus the id seams.""" + self._result_sink = result_sink + self._message_id_factory = message_id_factory or new_message_id + self._schedule_id = schedule_id + + async def handle_audio(self, chunks: AsyncIterator[bytes], stream: StreamInfo) -> None: + """Read the audio to its end, then deliver the fixed transcript and result.""" + byte_count = 0 + async for chunk in chunks: + byte_count += len(chunk) + + logger.info( + "audio received by agent", + extra={"stream_id": stream.stream_id, "byte_count": byte_count}, + ) + + transcript = Transcript( + text=FAKE_TRANSCRIPT, + language=FAKE_LANGUAGE, + duration_ms=_duration_ms_of(byte_count), + ) + await self._result_sink.deliver_transcript(transcript, stream) + + result = CommandResult( + message_id=self._message_id_factory(), + operation=FAKE_OPERATION, + status=FAKE_STATUS, + schedule=fake_schedule(self._schedule_id), + ) + await self._result_sink.deliver_result(result, stream) + + +def _duration_ms_of(byte_count: int) -> int: + """Convert 16 kHz mono 16-bit byte count into milliseconds.""" + bytes_per_ms = 16000 * 2 // 1000 + return byte_count // bytes_per_ms diff --git a/backend/src/timeflow/intelligence/ports.py b/backend/src/timeflow/intelligence/ports.py new file mode 100644 index 00000000..d5e71730 --- /dev/null +++ b/backend/src/timeflow/intelligence/ports.py @@ -0,0 +1,102 @@ +"""Ports the dialogue layer exposes, plus the values passed across them.""" + +from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Any, Protocol + + +class StreamInfo(Protocol): + """Identifiers of the audio stream a result belongs to. + + Stated as a shape rather than a class because these values travel inward: the caller + already holds them, so it supplies the object and this layer only reads it. + """ + + @property + def session_id(self) -> str: + """Session the stream belongs to.""" + ... + + @property + def stream_id(self) -> str: + """The stream itself.""" + ... + + @property + def conversation_id(self) -> str: + """Conversation the stream continues.""" + ... + + @property + def request_id(self) -> str | None: + """Request that opened the stream, when the client supplied one.""" + ... + + +@dataclass(frozen=True, slots=True) +class Transcript: + """What the user was heard to say.""" + + text: str + language: str + duration_ms: int + + +@dataclass(frozen=True, slots=True) +class AudioReply: + """What a spoken reply says and how it is encoded, sent ahead of the audio itself. + + Describes the audio without holding it; the bytes arrive separately as a stream. + """ + + audio_id: str + audio_format: str + sample_rate_hz: int + purpose: str + speech_text: str + + +@dataclass(frozen=True, slots=True) +class CommandResult: + """A command that was carried out, and the schedule state it left behind. + + Holds no stream identifiers: the caller passes those separately, so the same result + is never coupled to one particular stream. + """ + + message_id: str + operation: str + status: str + schedule: dict[str, Any] + + +class ResultSink(Protocol): + """Push results back to the client, one method per protocol message.""" + + async def deliver_transcript(self, transcript: Transcript, stream: StreamInfo) -> None: + """Send what the user was heard to say; call as soon as it is known.""" + ... + + async def deliver_result(self, result: CommandResult, stream: StreamInfo) -> None: + """Send the outcome of a command; call once it has actually been carried out.""" + ... + + async def deliver_audio( + self, reply: AudioReply, chunks: AsyncIterator[bytes], stream: StreamInfo + ) -> None: + """Speak a reply, forwarding chunks as they are produced. + + Cancelling stops it early and still closes the run. + """ + ... + + +class AgentPort(Protocol): + """Take one audio stream and act on what it contains.""" + + async def handle_audio(self, chunks: AsyncIterator[bytes], stream: StreamInfo) -> None: + """Consume the audio; returning only confirms receipt, not a result. + + Results reach the client through ResultSink, not through this return value. + """ + ... diff --git a/backend/src/timeflow/main.py b/backend/src/timeflow/main.py index b3a0aea1..055aa17c 100644 --- a/backend/src/timeflow/main.py +++ b/backend/src/timeflow/main.py @@ -8,13 +8,16 @@ UnauthenticatedConnectionLimiter, run_websocket_session, ) +from timeflow.gateway.websocket.handlers.agent_audio import AgentAudioSink +from timeflow.gateway.websocket.handlers.agent_result import WebSocketResultSink +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 from timeflow.gateway.websocket.ports import AudioSink, TokenVerifier from timeflow.gateway.websocket.router import MessageRouter -from timeflow.infrastructure.audio.null_sink import NullAudioSink from timeflow.infrastructure.security.token_verifier import FakeTokenVerifier from timeflow.infrastructure.settings import get_settings +from timeflow.intelligence.fake_agent import FakeAgent def create_app( @@ -42,13 +45,14 @@ def create_app( connections = ConnectionManager() limiter = UnauthenticatedConnectionLimiter(settings.ws_max_unauthenticated_connections) voice_streams = VoiceStreamHandlers( - audio_sink or NullAudioSink(), + audio_sink or AgentAudioSink(FakeAgent(WebSocketResultSink(connections))), max_audio_duration_ms=settings.ws_max_audio_duration_ms, queue_max_chunks=settings.ws_audio_queue_max_chunks, ) router = MessageRouter() router.register("voice.stream.start", voice_streams.handle_start) router.register("voice.stream.end", voice_streams.handle_end) + router.register("message.ack", handle_message_ack) @application.get("/api/v1/health") def health() -> dict[str, str]: diff --git a/backend/tests/test_agent_delivery.py b/backend/tests/test_agent_delivery.py new file mode 100644 index 00000000..d0d090b3 --- /dev/null +++ b/backend/tests/test_agent_delivery.py @@ -0,0 +1,307 @@ +"""How the two result messages reach the wire, when things go wrong or happen at once.""" + +import asyncio +from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Any + +from timeflow.gateway.websocket.connection_manager import ConnectionManager +from timeflow.gateway.websocket.handlers.agent_result import WebSocketResultSink +from timeflow.intelligence.ports import AudioReply, CommandResult, Transcript + +SESSION_ID = "ws_session_test" + + +@dataclass(frozen=True, slots=True) +class _Identity: + """Identifiers of the stream a result answers.""" + + session_id: str = SESSION_ID + stream_id: str = "stream_test" + conversation_id: str = "conversation_test" + request_id: str | None = "req_voice_001" + + +class RecordingConnection: + """A stand-in socket that records the frames written to it.""" + + def __init__(self) -> None: + """Start with nothing sent.""" + self.frames: list[dict[str, Any]] = [] + + async def send_json(self, data: Any) -> None: + """Record one JSON frame.""" + self.frames.append(data) + + async def send_bytes(self, data: bytes) -> None: + """Record one binary frame as a marker, so audio shows up in the frame order.""" + self.frames.append({"type": "audio"}) + + +def _transcript(tag: str) -> Transcript: + """Build a transcript tagged so its message can be told apart.""" + return Transcript(text=tag, language="zh", duration_ms=10) + + +def _result(tag: str) -> CommandResult: + """Build a command result tagged so its message can be told apart.""" + return CommandResult( + message_id=tag, + operation="create_schedule", + status="applied", + schedule={"id": tag}, + ) + + +def test_deliver_transcript_sends_voice_asr_completed() -> None: + """The transcript goes out on its own, carrying the stream's identifiers. + + Uses real Chinese speech so the assertion also covers non-ASCII surviving the + pydantic model and the JSON encoding, which every real transcript will need. + """ + + async def scenario() -> None: + """Deliver only a transcript to a connected session.""" + connections = ConnectionManager() + connection = RecordingConnection() + connections.register(SESSION_ID, connection) + + await WebSocketResultSink(connections).deliver_transcript( + _transcript("明天下午三点在203开会"), _Identity() + ) + + assert len(connection.frames) == 1 + frame = connection.frames[0] + assert frame["type"] == "voice.asr.completed" + assert frame["request_id"] == "req_voice_001" + assert frame["conversation_id"] == "conversation_test" + assert frame["payload"]["transcript"] == "明天下午三点在203开会" + + asyncio.run(scenario()) + + +def test_deliver_result_sends_voice_command_result() -> None: + """The command result goes out on its own, without needing a transcript first.""" + + async def scenario() -> None: + """Deliver only a command result to a connected session.""" + connections = ConnectionManager() + connection = RecordingConnection() + connections.register(SESSION_ID, connection) + + await WebSocketResultSink(connections).deliver_result(_result("msg_a"), _Identity()) + + assert len(connection.frames) == 1 + frame = connection.frames[0] + assert frame["type"] == "voice.command.result" + assert frame["message_id"] == "msg_a" + assert frame["conversation_id"] == "conversation_test" + assert frame["payload"]["operation"] == "create_schedule" + + asyncio.run(scenario()) + + +def test_the_two_messages_are_independent_calls() -> None: + """Each message is one call, so the caller decides when each goes out. + + Pins the split: the sink no longer bundles a turn, so a transcript can be sent long + before its command result is known. + """ + + async def scenario() -> None: + """Send the transcript, then the result, as two separate calls.""" + connections = ConnectionManager() + connection = RecordingConnection() + connections.register(SESSION_ID, connection) + sink = WebSocketResultSink(connections) + + await sink.deliver_transcript(_transcript("msg_a"), _Identity()) + assert [frame["type"] for frame in connection.frames] == ["voice.asr.completed"] + + await sink.deliver_result(_result("msg_a"), _Identity()) + assert [frame["type"] for frame in connection.frames] == [ + "voice.asr.completed", + "voice.command.result", + ] + + asyncio.run(scenario()) + + +def test_delivering_to_a_gone_session_sends_nothing() -> None: + """A result for a session that has closed is dropped rather than raising.""" + + async def scenario() -> None: + """Deliver both messages for a session nobody registered.""" + connections = ConnectionManager() + sink = WebSocketResultSink(connections) + + await sink.deliver_transcript(_transcript("msg_a"), _Identity()) + await sink.deliver_result(_result("msg_a"), _Identity()) + + asyncio.run(scenario()) + + +def test_a_session_that_dies_after_the_transcript_gets_no_result() -> None: + """A connection that leaves after the first message is never written to again. + + Pins the observable half-delivered turn: whatever delivery does internally, a socket + that has left the registry receives nothing further. + """ + + async def scenario() -> None: + """Let the connection unregister itself once the first frame lands.""" + connections = ConnectionManager() + + class DiesAfterFirstFrame(RecordingConnection): + """A socket that drops out of the registry after one frame.""" + + async def send_json(self, data: Any) -> None: + """Record the frame, then leave the registry.""" + await super().send_json(data) + connections.unregister(SESSION_ID, self) + + connection = DiesAfterFirstFrame() + connections.register(SESSION_ID, connection) + sink = WebSocketResultSink(connections) + + await sink.deliver_transcript(_transcript("msg_a"), _Identity()) + await sink.deliver_result(_result("msg_a"), _Identity()) + + assert [frame["type"] for frame in connection.frames] == ["voice.asr.completed"] + + asyncio.run(scenario()) + + +def test_a_result_follows_a_reconnect_that_lands_before_it_is_written() -> None: + """A result addressed to a replaced connection is not written to the stale socket. + + The reconnect is made to land while delivery is already waiting for the session's + write lock, which is the only moment the stale socket could still be written to. + """ + + async def scenario() -> None: + """Hold the write lock, start delivery, reconnect, then let delivery proceed.""" + connections = ConnectionManager() + stale = RecordingConnection() + live = RecordingConnection() + connections.register(SESSION_ID, stale) + + async with connections.lock_for(SESSION_ID): + delivery = asyncio.create_task( + WebSocketResultSink(connections).deliver_result(_result("msg_a"), _Identity()) + ) + await asyncio.sleep(0) + connections.register(SESSION_ID, live) + + await delivery + + assert stale.frames == [] + + asyncio.run(scenario()) + + +def test_results_for_different_sessions_stay_apart() -> None: + """Two sessions each receive only their own result.""" + + async def scenario() -> None: + """Deliver one result per session, concurrently.""" + connections = ConnectionManager() + first = RecordingConnection() + second = RecordingConnection() + connections.register("ws_session_first", first) + connections.register("ws_session_second", second) + sink = WebSocketResultSink(connections) + + await asyncio.gather( + sink.deliver_result(_result("msg_first"), _Identity(session_id="ws_session_first")), + sink.deliver_result(_result("msg_second"), _Identity(session_id="ws_session_second")), + ) + + assert [frame["message_id"] for frame in first.frames] == ["msg_first"] + assert [frame["message_id"] for frame in second.frames] == ["msg_second"] + + asyncio.run(scenario()) + + +def _reply() -> AudioReply: + """Build a spoken reply matching what the realtime model emits.""" + return AudioReply( + audio_id="audio_001", + audio_format="pcm", + sample_rate_hz=24000, + purpose="command_result", + speech_text="好，明天下午三点在203的会已经记下了", + ) + + +async def _chunks(*payloads: bytes) -> AsyncIterator[bytes]: + """Yield the given chunks with no delay.""" + for payload in payloads: + yield payload + + +def test_deliver_audio_translates_the_reply_into_voice_tts_start() -> None: + """The reply description becomes a voice.tts.start carrying the stream's identifiers. + + Only the translation is checked here; the framing and cancellation behaviour of the + burst itself belongs to the transport and is covered there. + """ + + async def scenario() -> None: + """Speak a one-chunk reply to a connected session.""" + connections = ConnectionManager() + connection = RecordingConnection() + connections.register(SESSION_ID, connection) + + await WebSocketResultSink(connections).deliver_audio(_reply(), _chunks(b"aa"), _Identity()) + + start = connection.frames[0] + assert start["type"] == "voice.tts.start" + assert start["conversation_id"] == "conversation_test" + assert start["audio_id"] == "audio_001" + assert start["payload"] == { + "format": "pcm", + "sample_rate_hz": 24000, + "purpose": "command_result", + "speech_text": "好，明天下午三点在203的会已经记下了", + "schedule_id": None, + "audio_version": None, + } + assert "ok" not in start + + asyncio.run(scenario()) + + +def test_the_reply_text_arrives_before_any_audio() -> None: + """speech_text rides on the start message, so the words are known before the sound.""" + + async def scenario() -> None: + """Speak a reply and read the frame order.""" + connections = ConnectionManager() + connection = RecordingConnection() + connections.register(SESSION_ID, connection) + + await WebSocketResultSink(connections).deliver_audio( + _reply(), _chunks(b"aa", b"bb"), _Identity() + ) + + assert [frame["type"] for frame in connection.frames] == [ + "voice.tts.start", + "audio", + "audio", + "voice.tts.end", + ] + + asyncio.run(scenario()) + + +def test_speaking_to_a_gone_session_sends_nothing() -> None: + """A reply for a session that has closed is dropped rather than raising.""" + + async def scenario() -> None: + """Speak to a session nobody registered.""" + connections = ConnectionManager() + + await WebSocketResultSink(connections).deliver_audio(_reply(), _chunks(b"aa"), _Identity()) + + asyncio.run(scenario()) diff --git a/backend/tests/test_agent_flow.py b/backend/tests/test_agent_flow.py new file mode 100644 index 00000000..921daabb --- /dev/null +++ b/backend/tests/test_agent_flow.py @@ -0,0 +1,354 @@ +"""The turn from inbound audio to a delivered result.""" + +import threading +from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Any + +from fastapi import FastAPI, WebSocket +from fastapi.testclient import TestClient + +from timeflow.gateway.websocket.agent_ports import Agent, StreamIdentity +from timeflow.gateway.websocket.connection_manager import ConnectionManager +from timeflow.gateway.websocket.endpoint import ( + UnauthenticatedConnectionLimiter, + run_websocket_session, +) +from timeflow.gateway.websocket.handlers.agent_audio import AgentAudioSink +from timeflow.gateway.websocket.handlers.agent_result import WebSocketResultSink +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 +from timeflow.gateway.websocket.router import MessageRouter +from timeflow.infrastructure.security.token_verifier import FakeTokenVerifier +from timeflow.intelligence.fake_agent import FAKE_TRANSCRIPT, FakeAgent + +VALID_HELLO: dict[str, Any] = { + "type": "session.hello", + "request_id": "req_session_001", + "payload": {"access_token": "token-abc", "device_id": "device_001"}, +} +START: dict[str, Any] = { + "type": "voice.stream.start", + "request_id": "req_voice_001", + "payload": { + "conversation_id": None, + "audio_format": "pcm_s16le", + "sample_rate_hz": 16000, + "channels": 1, + }, +} +END: dict[str, Any] = { + "type": "voice.stream.end", + "request_id": "req_voice_001", + "payload": {"stream_id": "stream_test"}, +} + + +@dataclass(frozen=True, slots=True) +class _Identity: + """Stand-in for the identifiers the transport hands to an agent.""" + + session_id: str + stream_id: str + conversation_id: str + request_id: str | None + + +class CapturingAgent: + """Record the audio an agent receives, and signal when the stream ended.""" + + def __init__(self) -> None: + """Start with nothing captured.""" + self.chunks: list[bytes] = [] + self.streams: list[StreamIdentity] = [] + self.completed = threading.Event() + + async def handle_audio(self, chunks: AsyncIterator[bytes], stream: StreamIdentity) -> None: + """Collect the stream's chunks in arrival order.""" + self.streams.append(stream) + async for chunk in chunks: + self.chunks.append(chunk) + self.completed.set() + + def audio(self) -> bytes: + """Return everything received, concatenated.""" + return b"".join(self.chunks) + + +def _build_app(agent: Agent | None = None, *, max_audio_duration_ms: int = 120_000) -> FastAPI: + """Build an app wiring the transport to an agent, with fixed identifiers.""" + application = FastAPI() + connections = ConnectionManager() + handshake = SessionHandshake(FakeTokenVerifier(), session_id_factory=lambda: "ws_session_test") + limiter = UnauthenticatedConnectionLimiter(100) + voice_streams = VoiceStreamHandlers( + AgentAudioSink( + agent + or FakeAgent( + WebSocketResultSink(connections), + message_id_factory=lambda: "msg_test", + ) + ), + max_audio_duration_ms=max_audio_duration_ms, + stream_id_factory=lambda: "stream_test", + conversation_id_factory=lambda: "conversation_test", + ) + router = MessageRouter() + router.register("voice.stream.start", voice_streams.handle_start) + router.register("voice.stream.end", voice_streams.handle_end) + router.register("message.ack", handle_message_ack) + + @application.websocket("/ws") + async def endpoint(websocket: WebSocket) -> None: + """Serve one transport session.""" + await run_websocket_session( + websocket, + handshake, + router, + connections, + limiter, + handshake_timeout_seconds=5.0, + binary_handler=voice_streams.handle_binary, + disconnect_handler=voice_streams.handle_disconnect, + ) + + return application + + +def test_a_finished_stream_yields_a_transcript_then_a_command_result() -> None: + """Closing a stream pushes the transcript first and the command result second.""" + client = TestClient(_build_app()) + + 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" * 160) + websocket.send_json(END) + first = websocket.receive_json() + second = websocket.receive_json() + + assert first["type"] == "voice.asr.completed" + assert second["type"] == "voice.command.result" + + +def test_the_transcript_message_matches_the_documented_shape() -> None: + """voice.asr.completed carries its identifiers beside payload and omits ok.""" + client = TestClient(_build_app()) + + 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" * 160) + websocket.send_json(END) + transcript = websocket.receive_json() + + assert transcript["request_id"] == "req_voice_001" + assert transcript["conversation_id"] == "conversation_test" + assert "ok" not in transcript + assert transcript["payload"]["transcript"] == FAKE_TRANSCRIPT + assert transcript["payload"]["language"] == "zh" + assert transcript["payload"]["duration_ms"] == 10 + + +def test_the_command_result_message_matches_the_documented_shape() -> None: + """voice.command.result carries message_id at the top level and omits ok.""" + client = TestClient(_build_app()) + + 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" * 160) + websocket.send_json(END) + websocket.receive_json() + result = websocket.receive_json() + + assert result["message_id"] == "msg_test" + assert result["request_id"] == "req_voice_001" + assert result["conversation_id"] == "conversation_test" + assert "ok" not in result + assert result["payload"]["operation"] == "create_schedule" + assert result["payload"]["status"] == "applied" + assert result["payload"]["schedule"]["id"] == "schedule_fake_001" + + +def test_the_agent_receives_the_audio_unchanged() -> None: + """Every byte reaches the agent in send order, with no transcription in between.""" + agent = CapturingAgent() + client = TestClient(_build_app(agent)) + frames = [b"\x01\x02" * 8, b"\x03\x04" * 8, b"\x05\x06" * 8] + + 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() + for frame in frames: + websocket.send_bytes(frame) + websocket.send_json(END) + assert agent.completed.wait(timeout=2) + + assert agent.chunks == frames + assert agent.audio() == b"".join(frames) + + +def test_the_agent_receives_the_stream_identifiers() -> None: + """The agent is told which session, stream, conversation and request it is serving.""" + agent = CapturingAgent() + client = TestClient(_build_app(agent)) + + 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(END) + assert agent.completed.wait(timeout=2) + + assert len(agent.streams) == 1 + stream = agent.streams[0] + assert stream.session_id == "ws_session_test" + assert stream.stream_id == "stream_test" + assert stream.conversation_id == "conversation_test" + assert stream.request_id == "req_voice_001" + + +def test_an_acknowledgement_is_accepted_silently() -> None: + """message.ack draws no reply, so the next reply belongs to the message after it.""" + client = TestClient(_build_app()) + + with client.websocket_connect("/ws?device_id=device_001") as websocket: + websocket.send_json(VALID_HELLO) + websocket.receive_json() + websocket.send_json({"type": "message.ack", "message_id": "msg_test", "status": "applied"}) + websocket.send_json({"type": "unknown.probe"}) + reply = websocket.receive_json() + + assert reply["error"]["code"] == "UNKNOWN_MESSAGE_TYPE" + + +def test_acknowledging_an_unknown_message_is_not_an_error() -> None: + """An ack for a message the server never sent is accepted as already done.""" + client = TestClient(_build_app()) + + with client.websocket_connect("/ws?device_id=device_001") as websocket: + websocket.send_json(VALID_HELLO) + websocket.receive_json() + websocket.send_json({"type": "message.ack", "message_id": "msg_never", "status": "applied"}) + websocket.send_json({"type": "message.ack", "message_id": "msg_never", "status": "applied"}) + websocket.send_json({"type": "unknown.probe"}) + reply = websocket.receive_json() + + assert reply["error"]["code"] == "UNKNOWN_MESSAGE_TYPE" + + +def test_a_malformed_acknowledgement_is_refused() -> None: + """An ack without message_id is reported rather than silently accepted.""" + client = TestClient(_build_app()) + + with client.websocket_connect("/ws?device_id=device_001") as websocket: + websocket.send_json(VALID_HELLO) + websocket.receive_json() + websocket.send_json({"type": "message.ack", "status": "applied"}) + reply = websocket.receive_json() + + assert reply["ok"] is False + assert reply["error"]["code"] == "MALFORMED_MESSAGE" + + +def test_disconnecting_before_the_stream_ends_delivers_nothing() -> None: + """Dropping the connection mid-stream leaves the turn unfinished and unsent.""" + agent = CapturingAgent() + client = TestClient(_build_app(agent)) + + 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") + + assert not agent.completed.is_set() + + +def test_audio_beyond_the_budget_delivers_no_result() -> None: + """A stream cut off for exceeding its budget produces no turn at all. + + The agent has already begun reading when the limit trips, so this pins down that + the abort leaves nothing half-delivered. + """ + agent = CapturingAgent() + # 10 ms of 16 kHz mono 16-bit audio is 320 bytes. + client = TestClient(_build_app(agent, max_audio_duration_ms=10)) + + 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"\x00" * 200) + websocket.send_bytes(b"\x00" * 200) + refusal = websocket.receive_json() + + assert refusal["error"]["code"] == "AUDIO_INVALID" + assert not agent.completed.is_set() + + +def test_audio_survives_a_stream_longer_than_the_queue() -> None: + """Every byte reaches the agent even when the stream far exceeds the queue depth. + + The queue between the receive loop and the agent is bounded, so a long stream makes + it block and resume repeatedly; this pins down that nothing is dropped in between. + """ + agent = CapturingAgent() + client = TestClient(_build_app(agent)) + frames = [bytes([index % 256]) * 320 for index in range(200)] + + 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() + for frame in frames: + websocket.send_bytes(frame) + websocket.send_json(END) + assert agent.completed.wait(timeout=10) + + assert agent.chunks == frames + assert agent.audio() == b"".join(frames) + + +def test_a_later_turn_keeps_the_conversation_it_was_given() -> None: + """A second turn carrying the first turn's conversation id stays in that conversation.""" + client = TestClient(_build_app()) + + 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" * 160) + websocket.send_json(END) + first_transcript = websocket.receive_json() + websocket.receive_json() + + continued = { + **START, + "payload": {**START["payload"], "conversation_id": first_transcript["conversation_id"]}, + } + websocket.send_json(continued) + websocket.receive_json() + websocket.send_bytes(b"\x03\x04" * 160) + websocket.send_json(END) + second_transcript = websocket.receive_json() + second_result = websocket.receive_json() + + assert second_transcript["conversation_id"] == first_transcript["conversation_id"] + assert second_result["conversation_id"] == first_transcript["conversation_id"] diff --git a/backend/tests/test_architecture.py b/backend/tests/test_architecture.py index c62425e3..f8b3f5fe 100644 --- a/backend/tests/test_architecture.py +++ b/backend/tests/test_architecture.py @@ -37,7 +37,8 @@ "timeflow.intelligence", } ), - # A.3: protocol only; must not reach the database or adapt vendor services. + # A.3: protocol only; must not reach the database, adapt vendor services, or depend + # on the dialogue layer -- the agent seam is structural, declared in agent_ports.py. "gateway": frozenset( { "sqlalchemy", @@ -45,6 +46,7 @@ "dashscope", "timeflow.data", "timeflow.infrastructure.external", + "timeflow.intelligence", } ), # A.4: runtime capability only; must not depend on product layers.