diff --git a/backend/.env.example b/backend/.env.example index ad7f553..0847a89 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -45,3 +45,7 @@ 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 + +# Which voice agent backend /ws uses: "1" = end-to-end realtime model (above), +# "2" = LLM + ASR + TTS conversation pipeline. "2" is not wired in yet. +TIMEFLOW_VOICE_AGENT_MODE=1 diff --git a/backend/src/timeflow/data/schedule_unit_of_work.py b/backend/src/timeflow/data/schedule_unit_of_work.py index 90e0cad..bd0ba06 100644 --- a/backend/src/timeflow/data/schedule_unit_of_work.py +++ b/backend/src/timeflow/data/schedule_unit_of_work.py @@ -4,6 +4,7 @@ from sqlalchemy.orm import Session, sessionmaker +from timeflow.business.calendar.ports import ScheduleRepositoryPort from timeflow.data.repositories.schedule import ScheduleRepository @@ -13,7 +14,7 @@ class SqlAlchemyScheduleUnitOfWork: def __init__(self, session_factory: sessionmaker[Session]) -> None: self._session_factory = session_factory self._session: Session | None = None - self.schedules: ScheduleRepository + self.schedules: ScheduleRepositoryPort def __enter__(self) -> "SqlAlchemyScheduleUnitOfWork": self._session = self._session_factory() diff --git a/backend/src/timeflow/gateway/websocket/agent_ports.py b/backend/src/timeflow/gateway/websocket/agent_ports.py index 638a4b1..e70efa5 100644 --- a/backend/src/timeflow/gateway/websocket/agent_ports.py +++ b/backend/src/timeflow/gateway/websocket/agent_ports.py @@ -7,6 +7,11 @@ class StreamIdentity(Protocol): """Identifiers of the audio stream a result belongs to.""" + @property + def account_id(self) -> str: + """Account that owns this stream.""" + ... + @property def session_id(self) -> str: """Session the stream belongs to.""" @@ -114,8 +119,13 @@ def status(self) -> str: ... @property - def schedule(self) -> dict[str, Any]: - """Persisted schedule snapshot the command produced.""" + def schedule(self) -> dict[str, Any] | None: + """Persisted schedule snapshot a mutation produced, when there is one.""" + ... + + @property + def schedules(self) -> list[dict[str, Any]] | None: + """Matches a query found, when the command was a query.""" ... diff --git a/backend/src/timeflow/gateway/websocket/handlers/agent_audio.py b/backend/src/timeflow/gateway/websocket/handlers/agent_audio.py index 43803b3..4c4a670 100644 --- a/backend/src/timeflow/gateway/websocket/handlers/agent_audio.py +++ b/backend/src/timeflow/gateway/websocket/handlers/agent_audio.py @@ -11,6 +11,7 @@ class _StreamIdentity: """Identifiers lifted out of a stream context for the agent.""" + account_id: str session_id: str stream_id: str conversation_id: str @@ -32,6 +33,7 @@ async def consume(self, chunks: AsyncIterator[bytes], stream: StreamContext) -> def _identity_of(stream: StreamContext) -> _StreamIdentity: """Lift the identifiers the agent needs out of the transport's context.""" return _StreamIdentity( + account_id=stream.account_id, session_id=stream.session.session_id, stream_id=stream.stream_id, conversation_id=stream.conversation_id, diff --git a/backend/src/timeflow/gateway/websocket/handlers/agent_result.py b/backend/src/timeflow/gateway/websocket/handlers/agent_result.py index aa6e345..412a466 100644 --- a/backend/src/timeflow/gateway/websocket/handlers/agent_result.py +++ b/backend/src/timeflow/gateway/websocket/handlers/agent_result.py @@ -87,9 +87,10 @@ async def deliver_result(self, result: CommandOutcome, stream: StreamIdentity) - operation=result.operation, status=result.status, schedule=result.schedule, + schedules=result.schedules, ), ) - await self._send(stream.session_id, message.type, message.model_dump()) + await self._send(stream.session_id, message.type, message.model_dump(exclude_none=True)) async def deliver_question( self, question: DialogueQuestionInfo, stream: StreamIdentity diff --git a/backend/src/timeflow/gateway/websocket/messages/agent.py b/backend/src/timeflow/gateway/websocket/messages/agent.py index 6a6b765..c7ccd65 100644 --- a/backend/src/timeflow/gateway/websocket/messages/agent.py +++ b/backend/src/timeflow/gateway/websocket/messages/agent.py @@ -23,11 +23,12 @@ class VoiceAsrCompleted(BaseModel): class VoiceCommandResultPayload(BaseModel): - """The command that was carried out and the schedule it produced.""" + """The command that was carried out and the schedule(s) it produced.""" operation: str status: str - schedule: dict[str, Any] + schedule: dict[str, Any] | None = None + schedules: list[dict[str, Any]] | None = None class VoiceCommandResult(BaseModel): diff --git a/backend/src/timeflow/gateway/websocket/ports.py b/backend/src/timeflow/gateway/websocket/ports.py index e7dcb00..cfbc13f 100644 --- a/backend/src/timeflow/gateway/websocket/ports.py +++ b/backend/src/timeflow/gateway/websocket/ports.py @@ -33,6 +33,16 @@ class StreamContext: audio_config: AudioConfig request_id: str | None = None + @property + def account_id(self) -> str: + """Account that owns this stream.""" + return self.session.account_id + + @property + def session_id(self) -> str: + """Session this stream belongs to.""" + return self.session.session_id + class TokenVerifier(Protocol): """Verify an access token and resolve the owning account. diff --git a/backend/src/timeflow/infrastructure/settings.py b/backend/src/timeflow/infrastructure/settings.py index f2c69fd..2449766 100644 --- a/backend/src/timeflow/infrastructure/settings.py +++ b/backend/src/timeflow/infrastructure/settings.py @@ -45,6 +45,8 @@ class Settings: aliyun_audio_model: str = "qwen-audio-3.0-realtime-plus" aliyun_audio_region: str = "cn-beijing" aliyun_audio_voice: str = "longanqian" + # "1" = the end-to-end realtime model; "2" = the LLM+ASR+TTS conversation pipeline. + voice_agent_mode: str = "1" @classmethod def from_environment(cls, env_file: Path | str = ".env") -> "Settings": @@ -64,6 +66,7 @@ def from_environment(cls, env_file: Path | str = ".env") -> "Settings": openai_timeout_seconds = float(environ.get("TIMEFLOW_OPENAI_TIMEOUT_SECONDS", "30.0")) agent_max_tool_rounds = int(environ.get("TIMEFLOW_AGENT_MAX_TOOL_ROUNDS", "4")) + voice_agent_mode = environ.get("TIMEFLOW_VOICE_AGENT_MODE", "1") aliyun_tts_connect_timeout_seconds = float( environ.get("TIMEFLOW_ALIYUN_TTS_CONNECT_TIMEOUT_SECONDS", "10.0") ) @@ -83,6 +86,8 @@ def from_environment(cls, env_file: Path | str = ".env") -> "Settings": raise ValueError("TIMEFLOW_OPENAI_TIMEOUT_SECONDS must be greater than zero") if agent_max_tool_rounds <= 0: raise ValueError("TIMEFLOW_AGENT_MAX_TOOL_ROUNDS must be a positive integer") + if voice_agent_mode not in ("1", "2"): + raise ValueError("TIMEFLOW_VOICE_AGENT_MODE must be '1' or '2'") if aliyun_tts_connect_timeout_seconds <= 0 or aliyun_tts_task_timeout_seconds <= 0: raise ValueError("TTS timeouts must be greater than zero") @@ -133,6 +138,7 @@ def from_environment(cls, env_file: Path | str = ".env") -> "Settings": ), aliyun_audio_region=environ.get("TIMEFLOW_ALIYUN_AUDIO_REGION", "cn-beijing"), aliyun_audio_voice=environ.get("TIMEFLOW_ALIYUN_AUDIO_VOICE", "longanqian"), + voice_agent_mode=voice_agent_mode, ) def aliyun_audio_is_configured(self) -> bool: diff --git a/backend/src/timeflow/intelligence/ports.py b/backend/src/timeflow/intelligence/ports.py index e4b82e1..488eb44 100644 --- a/backend/src/timeflow/intelligence/ports.py +++ b/backend/src/timeflow/intelligence/ports.py @@ -8,6 +8,11 @@ class StreamInfo(Protocol): """Identifiers of the audio stream a result belongs to.""" + @property + def account_id(self) -> str: + """Account that owns this stream.""" + ... + @property def session_id(self) -> str: """Session the stream belongs to.""" @@ -76,7 +81,8 @@ class CommandResult: message_id: str operation: str status: str - schedule: dict[str, Any] + schedule: dict[str, Any] | None = None + schedules: list[dict[str, Any]] | None = None class ResultSink(Protocol): diff --git a/backend/src/timeflow/intelligence/realtime/agent.py b/backend/src/timeflow/intelligence/realtime/agent.py index 64f903c..e8fb9d5 100644 --- a/backend/src/timeflow/intelligence/realtime/agent.py +++ b/backend/src/timeflow/intelligence/realtime/agent.py @@ -1,15 +1,25 @@ -"""Agent that hands one turn to a realtime speech model and reports what it says.""" +"""Agent that hands turns to a realtime speech model and reports what it says.""" import asyncio import contextlib import logging +import time from collections.abc import AsyncIterator, Callable +from dataclasses import dataclass from typing import Any from uuid import uuid4 -from timeflow.intelligence.ports import AudioReply, ReplyText, ResultSink, StreamInfo +from timeflow.intelligence.ports import ( + AudioReply, + CommandResult, + DialogueQuestion, + ReplyText, + ResultSink, + StreamInfo, +) from timeflow.intelligence.ports import Transcript as HeardSpeech -from timeflow.intelligence.realtime.ports import RealtimeSessionFactory +from timeflow.intelligence.realtime.ports import RealtimeSession, RealtimeSessionFactory +from timeflow.intelligence.realtime.schedule_tools import ToolBox logger = logging.getLogger(__name__) @@ -17,8 +27,10 @@ REPLY_SAMPLE_RATE_HZ = 24_000 REPLY_AUDIO_FORMAT = "pcm" -# Only answers exist this round; the round that adds questions sets this per turn. +# An answer and a question sound alike on the wire, so the client is told which it is +# getting -- it may want to keep a microphone open for one of them. REPLY_PURPOSE = "command_result" +QUESTION_PURPOSE = "dialogue_question" # Language is reported as the model's own, not detected here. ASSUMED_LANGUAGE = "zh" @@ -26,6 +38,12 @@ # 16 kHz mono 16-bit: 32 bytes per millisecond. _INPUT_BYTES_PER_MS = 32 +# The vendor caps one session at 50 turns or 300 seconds and drops it on the spot. Both +# budgets are treated as smaller than they are, so a session is replaced while it still +# works rather than after a turn has already been lost to it. +SESSION_MAX_TURNS = 40 +SESSION_MAX_AGE_SECONDS = 240.0 + def new_audio_id() -> str: """Return a fresh identifier for one spoken reply.""" @@ -37,58 +55,173 @@ def new_reply_id() -> str: return f"reply_{uuid4().hex}" +def new_message_id() -> str: + """Return a fresh identifier for a result the client must acknowledge.""" + return f"msg_{uuid4().hex}" + + +def new_question_id() -> str: + """Return a fresh identifier for one question put to the user.""" + return f"question_{uuid4().hex}" + + +@dataclass(slots=True) +class _Held: + """A session kept open past its turn, so the next turn remembers this one.""" + + session: RealtimeSession + tools: ToolBox | None + opened_at: float + turns: int = 0 + + class RealtimeAgent: - """Feed one audio stream to a realtime model and push back what comes out.""" + """Feed audio streams to a realtime model, one conversation at a time.""" def __init__( self, sessions: RealtimeSessionFactory, result_sink: ResultSink, *, + tools_factory: Callable[[str], ToolBox] | None = None, instructions: Callable[[], str] | None = None, audio_id_factory: Callable[[], str] | None = None, reply_id_factory: Callable[[], str] | None = None, + message_id_factory: Callable[[], str] | None = None, + question_id_factory: Callable[[], str] | None = None, + clock: Callable[[], float] | None = None, ) -> None: - """Store the session source, the sink, and the id seams.""" + """Store the session source, the sink, the tools, and the id and clock seams.""" self._sessions = sessions self._result_sink = result_sink - # Called per turn, so a long-running server keeps saying what day it is now. + self._tools_factory = tools_factory + # Called per session, 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 + self._message_id_factory = message_id_factory or new_message_id + self._question_id_factory = question_id_factory or new_question_id + # Monotonic, not wall clock: a live session must not be thrown away for being + # hours old because the host's clock was corrected. + self._clock = clock or time.monotonic + self._held: dict[tuple[str, str], _Held] = {} + self._locks: dict[tuple[str, str], asyncio.Lock] = {} async def handle_audio(self, chunks: AsyncIterator[bytes], stream: StreamInfo) -> None: - """Run one turn: send the audio, then push the transcript, wording and speech.""" + """Run one turn on the conversation's session, keeping it for the next turn.""" + key = (stream.account_id, stream.conversation_id) + await self._close_spent() + self._forget_idle_locks() + + async with self._lock_for(key): + held = await self._session_for(key) + if held is None: + return + + turn = _Turn( + self._result_sink, + stream, + self._audio_id_factory(), + self._reply_id_factory(), + session=held.session, + tools=held.tools, + message_id_factory=self._message_id_factory, + question_id_factory=self._question_id_factory, + ) + pumping = asyncio.create_task(held.session.pump(turn)) + reusable = False + try: + sent_bytes = 0 + async for chunk in chunks: + await held.session.send_audio(chunk) + sent_bytes += len(chunk) + await held.session.finish_input() + turn.note_input(sent_bytes) + await pumping + reusable = turn.failure is None + 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() + # Kept only after a turn it survived. A failure, an exception, or a client + # that hung up mid-turn leaves state nobody has reasoned about, and + # reusing it carries that into the next turn. + if reusable: + held.turns += 1 + else: + await self._discard(key) + + async def _session_for(self, key: tuple[str, str]) -> _Held | None: + """Return the conversation's open session, opening one when there is none.""" + held = self._held.get(key) + if held is not None: + return held + + account_id, _ = key + tools = self._tools_factory(account_id) if self._tools_factory is not None else None + schemas = tools.tools() if tools is not None else [] try: - session = await self._sessions.open(self._instructions(), []) + session = await self._sessions.open(self._instructions(), schemas) except Exception: logger.exception("could not open a realtime session") - return + return None + fresh = _Held(session=session, tools=tools, opened_at=self._clock()) + self._held[key] = fresh + return fresh - turn = _Turn( - self._result_sink, - stream, - self._audio_id_factory(), - self._reply_id_factory(), - ) - pumping = asyncio.create_task(session.pump(turn)) - try: - sent_bytes = 0 - async for chunk in chunks: - await session.send_audio(chunk) - sent_bytes += len(chunk) - await session.finish_input() - turn.note_input(sent_bytes) - await pumping - except BaseException: - # Cancelled on every exit, not just the caller's: an orphaned pump sits in recv(). - pumping.cancel() - with contextlib.suppress(BaseException): - await pumping - raise - finally: - await turn.close() - await session.close() + async def _close_spent(self) -> None: + """Close sessions that used up a budget, whether or not they get used again. + + Swept here rather than on the conversation's next turn because a conversation may + have no next turn, and the session would hold a vendor connection until exit. + """ + spent = [ + key + for key, held in self._held.items() + if self._spent(held) and not self._lock_for(key).locked() + ] + for key in spent: + logger.info( + "replacing a realtime session that ran out of budget", + extra={"account_id": key[0], "conversation_id": key[1]}, + ) + await self._discard(key) + + def _spent(self, held: _Held) -> bool: + """Report whether a session has used up either of its budgets.""" + if held.turns >= SESSION_MAX_TURNS: + return True + return self._clock() - held.opened_at >= SESSION_MAX_AGE_SECONDS + + async def _discard(self, key: tuple[str, str]) -> None: + """Forget a conversation's session and close it.""" + held = self._held.pop(key, None) + if held is not None: + await held.session.close() + + def _lock_for(self, key: tuple[str, str]) -> asyncio.Lock: + """Return the lock that serializes turns within one conversation. + + A session is a sequence of append, commit, respond; two turns interleaving on one + would fold one burst of audio into the other's answer. Nothing is awaited between + the lookup and the acquire, so two turns cannot each create their own lock. + """ + lock = self._locks.get(key) + if lock is None: + lock = asyncio.Lock() + self._locks[key] = lock + return lock + + def _forget_idle_locks(self) -> None: + """Drop locks nobody holds, waits on, or has a session behind.""" + for key, lock in list(self._locks.items()): + if not lock.locked() and key not in self._held: + del self._locks[key] class _Turn: @@ -100,16 +233,27 @@ def __init__( stream: StreamInfo, audio_id: str, reply_id: str, + *, + session: RealtimeSession, + tools: ToolBox | None, + message_id_factory: Callable[[], str], + question_id_factory: Callable[[], 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._session = session + self._tools = tools + self._message_id_factory = message_id_factory + self._question_id_factory = question_id_factory self._spoken = "" + self._purpose = REPLY_PURPOSE self._input_bytes = 0 self._audio: asyncio.Queue[bytes | None] = asyncio.Queue() self._speaking: asyncio.Task[None] | None = None + self.failure: str | None = None def note_input(self, sent_bytes: int) -> None: """Record how much audio the user sent, for the transcript's duration.""" @@ -143,15 +287,52 @@ async def audio(self, data: bytes) -> None: await self._audio.put(data) async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: - """Note that a tool was asked for while none are offered, and answer nothing.""" - del arguments - logger.warning( - "realtime model asked for a tool while none are registered", - extra={"call_id": call_id, "tool": name}, + """Run the tool, tell the client what came of it, and let the model continue. + + The client needs the data to display and the model needs it to say anything true + about it, so both are answered from the one call. + """ + if self._tools is None: + logger.warning( + "realtime model asked for a tool while none are registered", + extra={"call_id": call_id, "tool": name}, + ) + return + + result = await self._tools.run(name, arguments) + if result.outcome is not None: + outcome = result.outcome + await self._result_sink.deliver_result( + CommandResult( + message_id=self._message_id_factory(), + operation=str(outcome["operation"]), + status=str(outcome["status"]), + schedule=outcome.get("schedule"), + schedules=outcome.get("schedules"), + ), + self._stream, + ) + if result.question is not None: + await self._ask(result.question) + await self._session.send_tool_result(call_id, result.output) + + async def _ask(self, question: dict[str, Any]) -> None: + """Push a question and mark the reply as one, so the audio says which it is.""" + self._purpose = QUESTION_PURPOSE + await self._result_sink.deliver_question( + DialogueQuestion( + question_id=self._question_id_factory(), + question_kind=str(question["question_kind"]), + speech_text=str(question["speech_text"]), + required_response=question["required_response"], + candidates=question["candidates"], + ), + self._stream, ) async def failed(self, message: str) -> None: """Record that the model could not finish this turn.""" + self.failure = message logger.warning("realtime session failed", extra={"reason": message}) async def close(self) -> None: @@ -172,7 +353,7 @@ async def _speak(self) -> None: audio_id=self._audio_id, audio_format=REPLY_AUDIO_FORMAT, sample_rate_hz=REPLY_SAMPLE_RATE_HZ, - purpose=REPLY_PURPOSE, + purpose=self._purpose, speech_text=self._spoken, ) await self._result_sink.deliver_audio(reply, self._drain(), self._stream) diff --git a/backend/src/timeflow/intelligence/realtime/instructions.py b/backend/src/timeflow/intelligence/realtime/instructions.py new file mode 100644 index 0000000..10ed30f --- /dev/null +++ b/backend/src/timeflow/intelligence/realtime/instructions.py @@ -0,0 +1,64 @@ +"""System instructions that make the realtime model behave as a schedule assistant.""" + +from collections.abc import Callable +from datetime import UTC, datetime +from zoneinfo import ZoneInfo + +LOCAL = ZoneInfo("Asia/Shanghai") + +_WEEKDAYS = ("星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日") + +_ROLE = """你是 TimeFlow 的日程助手，帮用户用说话的方式管理日程和提醒。 + +语言与口吻 +- 始终用中文回答，无论用户说什么语言。 +- 像朋友之间说话，简短自然，一两句话说完。不要客套，不要复述用户的原话。 + +输出格式 +- 只输出纯文本。不要 emoji，不要 Markdown，不要列表符号和标题。 +- 数字、时间、地点直接说出来，例如「明天下午三点，203」，不要写成「15:00」这种书面形式。 + +时间的理解 +- 用户会用相对说法：今天、明天、后天、下周三、这周末、下个月初。把它们理解成具体日期。 +- 「三点」这类没说上下午的，按最近的合理时间理解：白天说「三点」通常指下午三点。 +- 用户没说的信息不要自己编。 + +能做什么 +- schedule_query 查询日程，按时间范围、标题或地点筛选。 +- schedule_create 新建日程，schedule_update 修改，schedule_delete 删除。 +- request_user_input 向用户提问。 + +改动日程的规矩 +- 必要信息没齐就别建：时间型日程要有开始时间，地点型日程要有地点。缺什么先问。 +- 改和删都要先用 schedule_query 找到那条日程，拿它的 id 和 revision 去调用，不要凭印象编 id。 +- 删除之前先确认一次，question_kind 用 confirmation，把要删的那条说清楚。 +- 工具报 failed 就说没做成，说明原因。绝不要把没成功的说成已经办好了。 + +什么时候提问 +- 缺少必要信息时调用 request_user_input，question_kind 用 missing_field，required_response 写缺哪个字段。 +- 用户指代不明（「那个会」「上次那个」）时，**先用 schedule_query 查一遍**，再调用 request_user_input，question_kind 用 ambiguous_target，把查到的几条放进 candidates。不要空着 candidates 就问，客户端要靠它把选项列出来给用户点。 +- 一次只问一件事。缺日期又缺时间，先问日期。 +- 调用 request_user_input 之后，把 speech_text 的原话说出来，让用户听见问题。除此之外不要多说。 +- 能自己想明白的不要问。「明天」「下周三」这类你能算出来的，直接算，不要反问用户是哪天。 + +用户回答之后 +- 你记得上一轮问过什么。用户的回答是在补上一轮缺的那件事，不是一个新请求。 +- 补齐之后接着做原来那件事，不要从头再问一遍，也不要重复用户刚说的话。 +- 如果补上之后还缺别的，再问下一件。 + +查询之后怎么说 +- 先说有几条，再逐条说时间、标题、地点，一条一句。 +- 查不到就直接说没有，不要建议用户改条件重试。 +- 不要念日程的 id 或版本号。 +""" + + +def build_instructions(now: Callable[[], datetime] | None = None) -> str: + """Return the instructions with the current local time stated at the top.""" + clock = now or (lambda: datetime.now(UTC)) + local = clock().astimezone(LOCAL) + return ( + f"当前时间：{local.strftime('%Y年%m月%d日')} {_WEEKDAYS[local.weekday()]} " + f"{local.strftime('%H:%M')}（时区 Asia/Shanghai）。" + "用户说的今天、明天、这周都以此为基准。\n\n" + _ROLE + ) diff --git a/backend/src/timeflow/intelligence/realtime/schedule_tools.py b/backend/src/timeflow/intelligence/realtime/schedule_tools.py new file mode 100644 index 0000000..9b74221 --- /dev/null +++ b/backend/src/timeflow/intelligence/realtime/schedule_tools.py @@ -0,0 +1,447 @@ +"""The tools the realtime model may call, backed by ScheduleAgentService.""" + +import asyncio +import json +import logging +from collections.abc import Callable +from dataclasses import asdict, dataclass +from datetime import datetime +from enum import StrEnum +from functools import partial +from typing import Any +from zoneinfo import ZoneInfo + +from timeflow.business.calendar import ( + DeleteRecurringScheduleCommand, + RecurringDeleteScope, + ReminderStrength, + ReminderType, + ScheduleAgentService, + ScheduleBusinessError, + ScheduleKind, + ScheduleMutationResult, + ScheduleType, +) +from timeflow.intelligence.realtime.tool_mapping import ( + ToolInputError, + map_create_schedule_command, + map_delete_schedule_command, + map_find_schedules_query, + map_update_schedule_command, + normalize_datetime_args, +) + +logger = logging.getLogger(__name__) + +LOCAL = ZoneInfo("Asia/Shanghai") + +# Tool names from the conversation contract; names reach the client, so keep them. +SCHEDULE_CREATE = "schedule_create" +SCHEDULE_QUERY = "schedule_query" +SCHEDULE_UPDATE = "schedule_update" +SCHEDULE_DELETE = "schedule_delete" +REQUEST_USER_INPUT = "request_user_input" + +# Four question kinds registered; clients branch on this enum. +QUESTION_KINDS = ("missing_field", "ambiguous_target", "recurrence_scope", "confirmation") + + +@dataclass(frozen=True, slots=True) +class ToolResult: + """What a tool answered the model with, and what the client should be told.""" + + output: str + outcome: dict[str, Any] | None = None + question: dict[str, Any] | None = None + + +_DATETIME_SCHEMA = {"type": ["string", "null"], "format": "date-time"} +_NULLABLE_STRING_SCHEMA = {"type": ["string", "null"]} +_REMINDER_TYPE_SCHEMA = { + "type": ["string", "null"], + "enum": [ + ReminderType.AT_TIME.value, + ReminderType.BEFORE_START.value, + ReminderType.ARRIVE_LOCATION.value, + ReminderType.RETURN_TO_RECORDED_LOCATION.value, + None, + ], +} +_REMINDER_STRENGTH_SCHEMA = { + "type": ["string", "null"], + "enum": [ + ReminderStrength.LOW.value, + ReminderStrength.MEDIUM.value, + ReminderStrength.HIGH.value, + None, + ], +} +_EDITABLE_PROPERTIES: dict[str, Any] = { + "title": {"type": "string", "minLength": 1}, + "is_all_day": {"type": "boolean"}, + "start_time": _DATETIME_SCHEMA, + "end_time": _DATETIME_SCHEMA, + "timezone": {"type": "string", "minLength": 1}, + "recurrence_rule": _NULLABLE_STRING_SCHEMA, + "location_name": _NULLABLE_STRING_SCHEMA, + "latitude": {"type": ["number", "null"], "minimum": -90, "maximum": 90}, + "longitude": {"type": ["number", "null"], "minimum": -180, "maximum": 180}, + "reminder_type": _REMINDER_TYPE_SCHEMA, + "reminder_trigger_at": _DATETIME_SCHEMA, + "reminder_offset_minutes": {"type": ["integer", "null"], "minimum": 0}, + "reminder_strength": _REMINDER_STRENGTH_SCHEMA, +} + +TOOL_SCHEMAS: tuple[dict[str, Any], ...] = ( + { + "type": "function", + "function": { + "name": SCHEDULE_CREATE, + "description": "创建时间或地点日程，所有必要信息确认后调用。", + "parameters": { + "type": "object", + "properties": { + "schedule_type": { + "type": "string", + "enum": [ScheduleType.TIME.value, ScheduleType.LOCATION.value], + }, + "schedule_kind": { + "type": "string", + "enum": [ScheduleKind.ONCE.value, ScheduleKind.RECURRING.value], + }, + **_EDITABLE_PROPERTIES, + }, + "required": ["schedule_type", "schedule_kind", "title"], + "additionalProperties": False, + }, + }, + }, + { + "type": "function", + "function": { + "name": SCHEDULE_QUERY, + "description": "查询日程，用于列表、匹配、消歧、更新或删除前查找。", + "parameters": { + "type": "object", + "properties": { + "schedule_id": _NULLABLE_STRING_SCHEMA, + "title": _NULLABLE_STRING_SCHEMA, + "starts_at_or_after": _DATETIME_SCHEMA, + "starts_before": _DATETIME_SCHEMA, + "location_name": _NULLABLE_STRING_SCHEMA, + "include_deleted": {"type": "boolean"}, + }, + "additionalProperties": False, + }, + }, + }, + { + "type": "function", + "function": { + "name": SCHEDULE_UPDATE, + "description": "修改已确认的一条日程。", + "parameters": { + "type": "object", + "properties": { + "schedule_id": {"type": "string", "minLength": 1}, + "expected_revision": {"type": "integer", "minimum": 0}, + "changes": { + "type": "object", + "properties": _EDITABLE_PROPERTIES, + "minProperties": 1, + "additionalProperties": False, + }, + }, + "required": ["schedule_id", "expected_revision", "changes"], + "additionalProperties": False, + }, + }, + }, + { + "type": "function", + "function": { + "name": SCHEDULE_DELETE, + "description": "删除已确认的一条日程；周期日程需要指定 scope。", + "parameters": { + "type": "object", + "properties": { + "schedule_id": {"type": "string", "minLength": 1}, + "expected_revision": {"type": "integer", "minimum": 0}, + "schedule_kind": { + "type": "string", + "enum": [ScheduleKind.ONCE.value, ScheduleKind.RECURRING.value], + }, + "scope": { + "type": ["string", "null"], + "enum": [ + RecurringDeleteScope.THIS_OCCURRENCE.value, + RecurringDeleteScope.THIS_AND_FUTURE.value, + RecurringDeleteScope.ENTIRE_SERIES.value, + None, + ], + }, + }, + "required": ["schedule_id", "expected_revision", "schedule_kind"], + "additionalProperties": False, + }, + }, + }, + { + "type": "function", + "function": { + "name": REQUEST_USER_INPUT, + "description": ( + "信息不足时向用户提问，而不是自己猜。" + "缺少日期、时间、地点等必要信息，或者用户的说法匹配到多条日程时使用。" + "调用之后要把 speech_text 原话说出来，让用户听到问题。一次只问一件事。" + ), + "parameters": { + "type": "object", + "properties": { + "question_kind": { + "type": "string", + "enum": list(QUESTION_KINDS), + "description": ( + "missing_field 缺必要信息；ambiguous_target 匹配到多条日程；" + "recurrence_scope 周期日程范围不明；confirmation 需要用户确认" + ), + }, + "speech_text": { + "type": "string", + "description": "要问用户的话，一句口语，例如「这个会是哪天的？」", + }, + "required_response": { + "type": "string", + "description": "希望用户补充的字段名，例如 start_time、location", + }, + "candidates": { + "type": "array", + "items": {"type": "object"}, + "description": "匹配到多条时的候选日程，供客户端展示", + }, + }, + "required": ["question_kind", "speech_text"], + }, + }, + }, +) + + +class ToolBox: + """Run tools the model is allowed to call, and refuse the rest.""" + + def __init__( + self, + account_id: str, + service: ScheduleAgentService, + now: Callable[[], datetime] | None = None, + ) -> None: + """Store the account, the service, and the clock seam.""" + self._account_id = account_id + self._service = service + self._now = now or (lambda: datetime.now(LOCAL)) + + # The service is synchronous and reaches Postgres over a socket, so every call goes + # to a worker thread: awaiting it inline would stall the loop streaming this turn's + # audio. Each call opens its own session, so no session crosses a thread. + + def tools(self) -> list[dict[str, Any]]: + """Return the tool schemas to register on the session.""" + return [dict(tool) for tool in TOOL_SCHEMAS] + + async def run(self, name: str, arguments: dict[str, Any]) -> ToolResult: + """Run a tool and report what each side of the conversation should get.""" + if name == REQUEST_USER_INPUT: + return self._ask(arguments) + + # Normalize datetime fields before mapping + arguments = normalize_datetime_args(arguments) + + try: + if name == SCHEDULE_CREATE: + return await self._create(arguments) + if name == SCHEDULE_QUERY: + return await self._find(arguments) + if name == SCHEDULE_UPDATE: + return await self._update(arguments) + if name == SCHEDULE_DELETE: + return await self._delete(arguments) + except ToolInputError as exc: + return _refusal(str(exc)) + except ScheduleBusinessError as exc: + return _business_error(exc) + + logger.warning("realtime model asked for a tool that is not offered", extra={"tool": name}) + return _refusal(f"工具 {name} 不可用。") + + async def _create(self, arguments: dict[str, Any]) -> ToolResult: + command = map_create_schedule_command(arguments) + result = await asyncio.to_thread( + partial(self._service.create_schedule, account_id=self._account_id, command=command) + ) + return _mutation_result(result, "create_schedule") + + async def _find(self, arguments: dict[str, Any]) -> ToolResult: + query = map_find_schedules_query(arguments) + result = await asyncio.to_thread( + partial(self._service.find_schedules, account_id=self._account_id, query=query) + ) + schedules_with_local_time = [_for_model(s) for s in result.schedules] + return ToolResult( + output=json.dumps( + {"count": len(result.schedules), "schedules": schedules_with_local_time}, + ensure_ascii=False, + ), + outcome={ + "operation": "list_schedules", + "status": "applied", + "schedules": [_snapshot_for_client(s) for s in result.schedules], + }, + ) + + async def _update(self, arguments: dict[str, Any]) -> ToolResult: + command = map_update_schedule_command(arguments) + result = await asyncio.to_thread( + partial(self._service.update_schedule, account_id=self._account_id, command=command) + ) + return _mutation_result(result, "update_schedule") + + async def _delete(self, arguments: dict[str, Any]) -> ToolResult: + command = map_delete_schedule_command(arguments) + if isinstance(command, DeleteRecurringScheduleCommand): + result = await asyncio.to_thread( + partial( + self._service.delete_recurring_schedule, + account_id=self._account_id, + command=command, + ) + ) + else: + result = await asyncio.to_thread( + partial( + self._service.delete_once_schedule, + account_id=self._account_id, + command=command, + ) + ) + return _mutation_result(result, "delete_schedule") + + def _ask(self, arguments: dict[str, Any]) -> ToolResult: + """Turn a request to ask the user into a question, or refuse an unusable one.""" + kind = arguments.get("question_kind") + speech_text = arguments.get("speech_text") + if kind not in QUESTION_KINDS: + logger.warning("realtime model asked with an unknown kind", extra={"kind": kind}) + return _refusal(f"question_kind 必须是 {'、'.join(QUESTION_KINDS)} 之一。") + if not isinstance(speech_text, str) or not speech_text.strip(): + return _refusal("speech_text 不能为空，要写出问用户的原话。") + + candidates = _candidates(arguments.get("candidates")) + if kind == "ambiguous_target" and not candidates: + return _refusal("ambiguous_target 必须提供非空 candidates。先用 schedule_query 查询。") + + required = arguments.get("required_response") + return ToolResult( + output=json.dumps({"asked": True}, ensure_ascii=False), + question={ + "question_kind": kind, + "speech_text": speech_text.strip(), + "required_response": required if isinstance(required, str) and required else None, + "candidates": candidates, + }, + ) + + +def _refusal(reason: str) -> ToolResult: + """Tell the model why it got nothing, and tell the client nothing at all.""" + return ToolResult( + output=json.dumps({"status": "failed", "error": {"message": reason}}, ensure_ascii=False) + ) + + +def _business_error(error: ScheduleBusinessError) -> ToolResult: + """Tell the model a write was refused; nothing was committed, so the client hears none. + + voice.command.result reports a committed transaction (protocol §5.5). A refusal has + none to report, so only the model is told, and it says so out loud. + """ + return ToolResult( + output=json.dumps( + { + "status": "failed", + "error": { + "code": error.code.value, + "field": error.field, + "message": error.message, + "schedule_id": error.schedule_id, + }, + }, + ensure_ascii=False, + ) + ) + + +def _mutation_result(result: ScheduleMutationResult, operation: str) -> ToolResult: + """Convert a mutation result into model output and client outcome.""" + snapshot = result.schedules[0] if result.schedules else None + return ToolResult( + output=json.dumps( + {"status": "applied", "schedule": _for_model_dict(snapshot)}, ensure_ascii=False + ), + outcome={ + "operation": operation, + "status": "applied", + "schedule": _snapshot_for_client(snapshot) if snapshot else None, + }, + ) + + +def _for_model(schedule: Any) -> dict[str, Any]: + """Add a spoken-language local time beside the stored instant.""" + spoken = asdict(schedule) + spoken["starts_at_local"] = _local_text(spoken.get("start_time")) + result = _json_value(spoken) + assert isinstance(result, dict) + return result + + +def _for_model_dict(schedule: Any) -> dict[str, Any] | None: + if schedule is None: + return None + return _for_model(schedule) + + +def _snapshot_for_client(snapshot: Any) -> dict[str, Any]: + """Convert ScheduleSnapshot to client dict, filtering out audit fields.""" + d = asdict(snapshot) + return { + k: _json_value(v) + for k, v in d.items() + if k not in {"account_id", "created_at", "updated_at", "deleted_at"} + } + + +def _local_text(instant: datetime | None) -> str: + """Render a stored instant as local wall-clock text, empty for a schedule without one.""" + if instant is None: + return "" + return instant.astimezone(LOCAL).strftime("%Y-%m-%d %H:%M") + + +def _candidates(value: Any) -> tuple[dict[str, Any], ...]: + """Keep the choices that are actually objects, dropping anything else.""" + if not isinstance(value, list): + return () + return tuple(item for item in value if isinstance(item, dict)) + + +def _json_value(value: object) -> object: + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, StrEnum): + return value.value + if isinstance(value, dict): + return {key: _json_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_json_value(item) for item in value] + return value diff --git a/backend/src/timeflow/intelligence/realtime/tool_mapping.py b/backend/src/timeflow/intelligence/realtime/tool_mapping.py new file mode 100644 index 0000000..a43de93 --- /dev/null +++ b/backend/src/timeflow/intelligence/realtime/tool_mapping.py @@ -0,0 +1,280 @@ +"""Argument mapping and validation for schedule tools.""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime +from enum import StrEnum +from typing import TypeVar +from zoneinfo import ZoneInfo + +from timeflow.business.calendar import ( + CreateScheduleCommand, + DeleteOnceScheduleCommand, + DeleteRecurringScheduleCommand, + FindSchedulesQuery, + RecurringDeleteScope, + ReminderStrength, + ReminderType, + ScheduleKind, + ScheduleType, + ScheduleUpdatePatch, + UpdateScheduleCommand, +) + +_EnumT = TypeVar("_EnumT", bound=StrEnum) +LOCAL = ZoneInfo("Asia/Shanghai") + + +class ToolInputError(ValueError): + """A tool payload cannot be mapped to the business contract.""" + + +def normalize_datetime_args(arguments: dict[str, object]) -> dict[str, object]: + """Add local timezone offset to datetime strings that lack one. + + Mutates and returns the input dict for chaining. + """ + for key, value in arguments.items(): + if isinstance(value, str) and "T" in value: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + continue + # A "+" or "Z" check alone misses negative offsets like "-05:00", which + # fromisoformat parses as already aware -- reattaching LOCAL to those would + # silently shift the instant by the difference between the two zones. + if parsed.tzinfo is None or parsed.utcoffset() is None: + arguments[key] = parsed.replace(tzinfo=LOCAL).isoformat() + elif isinstance(value, dict): + normalize_datetime_args(value) + return arguments + + +def map_create_schedule_command(arguments: Mapping[str, object]) -> CreateScheduleCommand: + """Map model arguments into the stable create business command.""" + allowed = {"schedule_type", "schedule_kind", *ScheduleUpdatePatch.__optional_keys__} + _reject_unknown(arguments, allowed) + return CreateScheduleCommand( + schedule_type=_required_enum(arguments, "schedule_type", ScheduleType), + schedule_kind=_required_enum(arguments, "schedule_kind", ScheduleKind), + title=_required_string(arguments, "title"), + # Both default rather than being asked of the model: one deployment, one zone, and + # a model made to state them every call is a model that eventually invents them. + timezone=_optional_string(arguments, "timezone") or str(LOCAL.key), + is_all_day=_optional_bool(arguments, "is_all_day", default=False), + start_time=_optional_datetime(arguments, "start_time"), + end_time=_optional_datetime(arguments, "end_time"), + recurrence_rule=_optional_string(arguments, "recurrence_rule"), + location_name=_optional_string(arguments, "location_name"), + latitude=_optional_float(arguments, "latitude", minimum=-90, maximum=90), + longitude=_optional_float(arguments, "longitude", minimum=-180, maximum=180), + reminder_type=_optional_enum(arguments, "reminder_type", ReminderType), + reminder_trigger_at=_optional_datetime(arguments, "reminder_trigger_at"), + reminder_offset_minutes=_optional_int(arguments, "reminder_offset_minutes", minimum=0), + reminder_strength=_optional_enum(arguments, "reminder_strength", ReminderStrength), + ) + + +def map_find_schedules_query(arguments: Mapping[str, object]) -> FindSchedulesQuery: + """Map model arguments into the stable schedule query.""" + allowed = { + "schedule_id", + "title", + "starts_at_or_after", + "starts_before", + "location_name", + "include_deleted", + } + _reject_unknown(arguments, allowed) + return FindSchedulesQuery( + schedule_id=_optional_string(arguments, "schedule_id"), + title=_optional_string(arguments, "title"), + starts_at_or_after=_optional_datetime(arguments, "starts_at_or_after"), + starts_before=_optional_datetime(arguments, "starts_before"), + location_name=_optional_string(arguments, "location_name"), + include_deleted=_optional_bool(arguments, "include_deleted", default=False), + ) + + +def map_update_schedule_command(arguments: Mapping[str, object]) -> UpdateScheduleCommand: + """Map model arguments into the stable update business command.""" + _reject_unknown(arguments, {"schedule_id", "expected_revision", "changes"}) + raw_changes = arguments.get("changes") + if not isinstance(raw_changes, dict) or not raw_changes: + raise ToolInputError("changes must be a non-empty object") + from typing import cast + + changes = _map_update_patch(cast(Mapping[str, object], raw_changes)) + return UpdateScheduleCommand( + schedule_id=_required_string(arguments, "schedule_id"), + expected_revision=_required_int(arguments, "expected_revision", minimum=0), + changes=changes, + ) + + +def map_delete_schedule_command( + arguments: Mapping[str, object], +) -> DeleteOnceScheduleCommand | DeleteRecurringScheduleCommand: + """Map one delete tool into the appropriate stable business command.""" + _reject_unknown(arguments, {"schedule_id", "expected_revision", "schedule_kind", "scope"}) + schedule_id = _required_string(arguments, "schedule_id") + revision = _required_int(arguments, "expected_revision", minimum=0) + kind = _required_enum(arguments, "schedule_kind", ScheduleKind) + scope = _optional_enum(arguments, "scope", RecurringDeleteScope) + if kind is ScheduleKind.ONCE: + if scope is not None: + raise ToolInputError("scope is only valid for recurring schedules") + return DeleteOnceScheduleCommand(schedule_id=schedule_id, expected_revision=revision) + if scope is None: + raise ToolInputError("scope is required for recurring schedules") + return DeleteRecurringScheduleCommand( + schedule_id=schedule_id, + expected_revision=revision, + scope=scope, + ) + + +def _map_update_patch(arguments: Mapping[str, object]) -> ScheduleUpdatePatch: + _reject_unknown(arguments, set(ScheduleUpdatePatch.__optional_keys__)) + changes: ScheduleUpdatePatch = {} + if "title" in arguments: + changes["title"] = _required_string(arguments, "title") + if "is_all_day" in arguments: + changes["is_all_day"] = _required_bool(arguments, "is_all_day") + if "start_time" in arguments: + changes["start_time"] = _optional_datetime(arguments, "start_time") + if "end_time" in arguments: + changes["end_time"] = _optional_datetime(arguments, "end_time") + if "timezone" in arguments: + changes["timezone"] = _required_string(arguments, "timezone") + if "recurrence_rule" in arguments: + changes["recurrence_rule"] = _optional_string(arguments, "recurrence_rule") + if "location_name" in arguments: + changes["location_name"] = _optional_string(arguments, "location_name") + if "latitude" in arguments: + changes["latitude"] = _optional_float(arguments, "latitude", minimum=-90, maximum=90) + if "longitude" in arguments: + changes["longitude"] = _optional_float(arguments, "longitude", minimum=-180, maximum=180) + if "reminder_type" in arguments: + changes["reminder_type"] = _optional_enum(arguments, "reminder_type", ReminderType) + if "reminder_trigger_at" in arguments: + changes["reminder_trigger_at"] = _optional_datetime(arguments, "reminder_trigger_at") + if "reminder_offset_minutes" in arguments: + changes["reminder_offset_minutes"] = _optional_int( + arguments, "reminder_offset_minutes", minimum=0 + ) + if "reminder_strength" in arguments: + changes["reminder_strength"] = _optional_enum( + arguments, "reminder_strength", ReminderStrength + ) + return changes + + +def _reject_unknown(arguments: Mapping[str, object], allowed: set[str]) -> None: + unknown = set(arguments) - allowed + if unknown: + raise ToolInputError(f"Unexpected fields: {', '.join(sorted(unknown))}") + + +def _required_string(arguments: Mapping[str, object], field: str) -> str: + value = arguments.get(field) + if not isinstance(value, str) or not value.strip(): + raise ToolInputError(f"{field} must be a non-empty string") + return value + + +def _optional_string(arguments: Mapping[str, object], field: str) -> str | None: + value = arguments.get(field) + if value is None: + return None + if not isinstance(value, str): + raise ToolInputError(f"{field} must be a string or null") + return value + + +def _required_bool(arguments: Mapping[str, object], field: str) -> bool: + value = arguments.get(field) + if not isinstance(value, bool): + raise ToolInputError(f"{field} must be a boolean") + return value + + +def _optional_bool(arguments: Mapping[str, object], field: str, *, default: bool) -> bool: + value = arguments.get(field, default) + if not isinstance(value, bool): + raise ToolInputError(f"{field} must be a boolean") + return value + + +def _required_int(arguments: Mapping[str, object], field: str, *, minimum: int) -> int: + value = arguments.get(field) + if not isinstance(value, int) or isinstance(value, bool) or value < minimum: + raise ToolInputError(f"{field} must be an integer greater than or equal to {minimum}") + return value + + +def _optional_int(arguments: Mapping[str, object], field: str, *, minimum: int) -> int | None: + value = arguments.get(field) + if value is None: + return None + if not isinstance(value, int) or isinstance(value, bool) or value < minimum: + raise ToolInputError(f"{field} must be an integer greater than or equal to {minimum}") + return value + + +def _optional_float( + arguments: Mapping[str, object], + field: str, + *, + minimum: float, + maximum: float, +) -> float | None: + value = arguments.get(field) + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ToolInputError(f"{field} must be a number or null") + result = float(value) + if not minimum <= result <= maximum: + raise ToolInputError(f"{field} must be between {minimum} and {maximum}") + return result + + +def _optional_datetime(arguments: Mapping[str, object], field: str) -> datetime | None: + value = arguments.get(field) + if value is None: + return None + if not isinstance(value, str): + raise ToolInputError(f"{field} must be an ISO datetime string or null") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ToolInputError(f"{field} must be an ISO datetime string") from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise ToolInputError(f"{field} must include a timezone offset") + return parsed + + +def _required_enum(arguments: Mapping[str, object], field: str, enum_type: type[_EnumT]) -> _EnumT: + value = arguments.get(field) + if not isinstance(value, str): + raise ToolInputError(f"{field} must be a string") + try: + return enum_type(value) + except ValueError as exc: + raise ToolInputError(f"{field} has an unsupported value") from exc + + +def _optional_enum( + arguments: Mapping[str, object], field: str, enum_type: type[_EnumT] +) -> _EnumT | None: + value = arguments.get(field) + if value is None: + return None + if not isinstance(value, str): + raise ToolInputError(f"{field} must be a string or null") + try: + return enum_type(value) + except ValueError as exc: + raise ToolInputError(f"{field} has an unsupported value") from exc diff --git a/backend/src/timeflow/main.py b/backend/src/timeflow/main.py index 736a5c5..4a3343d 100644 --- a/backend/src/timeflow/main.py +++ b/backend/src/timeflow/main.py @@ -3,8 +3,12 @@ import logging from fastapi import FastAPI, WebSocket +from sqlalchemy.orm import Session, sessionmaker +from timeflow.business.calendar.service import ScheduleApplicationService from timeflow.business.health import HealthService +from timeflow.data.database import build_engine, build_session_factory +from timeflow.data.schedule_unit_of_work import SqlAlchemyScheduleUnitOfWork from timeflow.gateway.websocket.agent_ports import Agent from timeflow.gateway.websocket.connection_manager import ConnectionManager from timeflow.gateway.websocket.endpoint import ( @@ -26,6 +30,8 @@ from timeflow.infrastructure.settings import Settings, get_settings from timeflow.intelligence.fake_agent import FakeAgent from timeflow.intelligence.realtime.agent import RealtimeAgent +from timeflow.intelligence.realtime.instructions import build_instructions +from timeflow.intelligence.realtime.schedule_tools import ToolBox logger = logging.getLogger(__name__) @@ -40,6 +46,8 @@ def create_app( application = FastAPI(title=settings.app_name, version="0.1.0") health_service = HealthService() + session_factory = build_session_factory(build_engine(settings.database_url)) + if token_verifier is None: # Fail closed: the stand-in verifier accepts any non-empty token, so falling back # to it outside development would leave /ws effectively unauthenticated. @@ -56,7 +64,9 @@ def create_app( limiter = UnauthenticatedConnectionLimiter(settings.ws_max_unauthenticated_connections) if audio_sink is None: - audio_sink = AgentAudioSink(_build_agent(settings, WebSocketResultSink(connections))) + audio_sink = AgentAudioSink( + _build_agent(settings, WebSocketResultSink(connections), session_factory) + ) voice_streams = VoiceStreamHandlers( audio_sink, @@ -90,7 +100,29 @@ async def websocket_session(websocket: WebSocket) -> None: return application -def _build_agent(settings: Settings, result_sink: WebSocketResultSink) -> Agent: +def _build_agent( + settings: Settings, + result_sink: WebSocketResultSink, + session_factory: sessionmaker[Session], +) -> 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) + 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, + session_factory: sessionmaker[Session], +) -> 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 @@ -99,6 +131,14 @@ def _build_agent(settings: Settings, result_sink: WebSocketResultSink) -> Agent: """ if settings.aliyun_audio_is_configured(): logger.info("using the realtime model", extra={"model": settings.aliyun_audio_model}) + + schedule_service = ScheduleApplicationService( + lambda: SqlAlchemyScheduleUnitOfWork(session_factory) + ) + + def bind_account(account_id: str) -> ToolBox: + return ToolBox(account_id, schedule_service) + return RealtimeAgent( QwenAudioSessionFactory( QwenAudioConfig( @@ -110,6 +150,8 @@ def _build_agent(settings: Settings, result_sink: WebSocketResultSink) -> Agent: ) ), result_sink, + tools_factory=bind_account, + instructions=build_instructions, ) if settings.environment != "development": diff --git a/backend/tests/data/test_realtime_tools_postgres.py b/backend/tests/data/test_realtime_tools_postgres.py new file mode 100644 index 0000000..6f03126 --- /dev/null +++ b/backend/tests/data/test_realtime_tools_postgres.py @@ -0,0 +1,303 @@ +"""The realtime model's tools driving the real database, end to end.""" + +import asyncio +import json +import os +from collections.abc import Iterator +from datetime import UTC, datetime +from typing import Any + +import pytest +import sqlalchemy as sa +from sqlalchemy import Engine +from sqlalchemy.orm import Session, sessionmaker + +from timeflow.business.calendar.service import ScheduleApplicationService +from timeflow.data.database import build_session_factory +from timeflow.data.schedule_unit_of_work import SqlAlchemyScheduleUnitOfWork +from timeflow.intelligence.realtime.schedule_tools import ( + REQUEST_USER_INPUT, + SCHEDULE_CREATE, + SCHEDULE_DELETE, + SCHEDULE_QUERY, + SCHEDULE_UPDATE, + ToolBox, + ToolResult, +) + +ACCOUNT = "acct-tools-test" +OTHER_ACCOUNT = "acct-tools-other" +# A Monday, so "下周一" in a test reads the way a user would mean it. +NOW = datetime(2026, 9, 7, 9, 0) + + +def _service(factory: sessionmaker[Session]) -> ScheduleApplicationService: + """Build the application service the way the composition root does.""" + return ScheduleApplicationService(lambda: SqlAlchemyScheduleUnitOfWork(factory)) + + +@pytest.fixture(scope="module") +def engine() -> Iterator[Engine]: + """Connect only when an explicit disposable integration database is supplied.""" + database_url = os.getenv("TIMEFLOW_TEST_DATABASE_URL") + if database_url is None: + pytest.skip("TIMEFLOW_TEST_DATABASE_URL is not set") + built = sa.create_engine(database_url) + try: + yield built + finally: + built.dispose() + + +@pytest.fixture +def toolbox(engine: Engine) -> Iterator[ToolBox]: + """A toolbox bound to one account, on a fixed clock, over the real database.""" + factory = build_session_factory(engine) + created_at = datetime.now(UTC) + with factory() as session: + for account_id in (ACCOUNT, OTHER_ACCOUNT): + session.execute( + sa.text( + "INSERT INTO accounts (id, username, password_hash, created_at, updated_at) " + "VALUES (:id, :username, 'test-hash', :now, :now) " + "ON CONFLICT (id) DO NOTHING" + ), + {"id": account_id, "username": f"{account_id}-user", "now": created_at}, + ) + session.commit() + try: + yield ToolBox(ACCOUNT, _service(factory), now=lambda: NOW) + finally: + with factory() as session: + session.execute( + sa.text("DELETE FROM schedules WHERE account_id = ANY(:accounts)"), + {"accounts": [ACCOUNT, OTHER_ACCOUNT]}, + ) + session.execute( + sa.text("DELETE FROM accounts WHERE id = ANY(:accounts)"), + {"accounts": [ACCOUNT, OTHER_ACCOUNT]}, + ) + session.commit() + + +def call(toolbox: ToolBox, name: str, **arguments: Any) -> ToolResult: + """Run one tool the way the agent runs it, from outside a running loop.""" + return asyncio.run(toolbox.run(name, arguments)) + + +def output_of(result: ToolResult) -> dict[str, Any]: + """The JSON the model is handed back.""" + parsed: dict[str, Any] = json.loads(result.output) + return parsed + + +def test_creating_a_schedule_persists_it_and_tells_both_sides(toolbox: ToolBox) -> None: + """A create reaches the database and reports back to model and client alike.""" + result = call( + toolbox, + SCHEDULE_CREATE, + schedule_type="time", + schedule_kind="once", + title="明天下午三点开会", + start_time="2026-09-08T15:00:00", + end_time="2026-09-08T16:00:00", + reminder_type="before_start", + reminder_offset_minutes=15, + reminder_strength="medium", + ) + + assert result.question is None + assert output_of(result)["status"] == "applied" + assert result.outcome is not None + assert result.outcome["operation"] == "create_schedule" + assert result.outcome["status"] == "applied" + persisted = result.outcome["schedule"] + assert persisted["title"] == "明天下午三点开会" + assert persisted["revision"] == 1 + # The client is told about the schedule, never about who owns it or when the row moved. + assert "account_id" not in persisted + assert "created_at" not in persisted + + found = call(toolbox, SCHEDULE_QUERY, title="开会") + assert output_of(found)["count"] == 1 + + +def test_a_naive_local_time_is_stored_as_the_instant_it_names(toolbox: ToolBox) -> None: + """A time spoken without a zone means local time, not UTC.""" + call( + toolbox, + SCHEDULE_CREATE, + schedule_type="time", + schedule_kind="once", + title="七点晨跑", + start_time="2026-09-08T07:00:00", + ) + + found = call(toolbox, SCHEDULE_QUERY, title="晨跑") + + (schedule,) = output_of(found)["schedules"] + # Read back as the wall clock the user spoke, not shifted by the zone offset. + assert schedule["starts_at_local"] == "2026-09-08 07:00" + + +def test_a_query_only_ever_sees_the_bound_account(engine: Engine, toolbox: ToolBox) -> None: + """The toolbox is bound to one account and cannot be asked about another.""" + other = ToolBox( + OTHER_ACCOUNT, + _service(build_session_factory(engine)), + now=lambda: NOW, + ) + call( + other, + SCHEDULE_CREATE, + schedule_type="time", + schedule_kind="once", + title="别人的安排", + start_time="2026-09-08T15:00:00", + ) + + found = call(toolbox, SCHEDULE_QUERY, title="别人的安排") + + assert output_of(found)["count"] == 0 + + +def _created_id(toolbox: ToolBox, title: str = "去公司开会") -> tuple[str, int]: + """Create one schedule and return what an update needs to address it.""" + result = call( + toolbox, + SCHEDULE_CREATE, + schedule_type="time", + schedule_kind="once", + title=title, + start_time="2026-09-08T15:00:00", + ) + assert result.outcome is not None + created = result.outcome["schedule"] + schedule_id: str = created["id"] + revision: int = created["revision"] + return schedule_id, revision + + +def test_an_update_reaches_the_database_and_moves_the_revision(toolbox: ToolBox) -> None: + """The model can change a schedule it just created.""" + schedule_id, revision = _created_id(toolbox) + + result = call( + toolbox, + SCHEDULE_UPDATE, + schedule_id=schedule_id, + expected_revision=revision, + changes={"title": "改到下午四点开会", "start_time": "2026-09-08T16:00:00"}, + ) + + assert result.outcome is not None + updated = result.outcome["schedule"] + assert updated["title"] == "改到下午四点开会" + assert updated["revision"] == revision + 1 + + +def test_a_stale_revision_comes_back_as_a_conflict_the_model_can_read( + toolbox: ToolBox, +) -> None: + """A revision that moved on is reported as a conflict, not applied.""" + schedule_id, revision = _created_id(toolbox) + call( + toolbox, + SCHEDULE_UPDATE, + schedule_id=schedule_id, + expected_revision=revision, + changes={"title": "先改一次"}, + ) + + result = call( + toolbox, + SCHEDULE_UPDATE, + schedule_id=schedule_id, + expected_revision=revision, + changes={"title": "再改一次"}, + ) + + reported = output_of(result) + assert reported["status"] == "failed" + assert reported["error"]["code"] == "revision_conflict" + # Nothing was committed, so the client is sent no command result at all. + assert result.outcome is None + + +def test_a_delete_takes_the_schedule_out_of_what_the_model_can_find( + toolbox: ToolBox, +) -> None: + """Deleting removes it from later queries.""" + schedule_id, revision = _created_id(toolbox) + + result = call( + toolbox, + SCHEDULE_DELETE, + schedule_id=schedule_id, + expected_revision=revision, + schedule_kind="once", + ) + + assert output_of(result)["status"] == "applied" + assert output_of(call(toolbox, SCHEDULE_QUERY))["count"] == 0 + + +def test_a_schedule_the_database_will_not_accept_is_refused_not_half_written( + toolbox: ToolBox, +) -> None: + """A location schedule with no coordinates is refused and leaves nothing behind.""" + result = call( + toolbox, + SCHEDULE_CREATE, + schedule_type="location", + schedule_kind="once", + title="到了公司提醒我", + location_name="公司", + ) + + assert output_of(result)["status"] == "failed" + assert output_of(call(toolbox, SCHEDULE_QUERY))["count"] == 0 + + +def test_deleting_something_that_is_not_there_is_reported_not_raised( + toolbox: ToolBox, +) -> None: + """A missing schedule reads as not found, and the turn carries on.""" + result = call( + toolbox, + SCHEDULE_DELETE, + schedule_id="sch_nothing_here", + expected_revision=1, + schedule_kind="once", + ) + + assert output_of(result)["error"]["code"] == "schedule_not_found" + + +def test_asking_the_user_never_touches_the_database(toolbox: ToolBox) -> None: + """A question is asked, not applied: nothing is written.""" + result = call( + toolbox, + REQUEST_USER_INPUT, + question_kind="missing_field", + speech_text="是明天下午三点吗？", + required_response="start_time", + ) + + assert result.question is not None + assert result.question["speech_text"] == "是明天下午三点吗？" + assert result.outcome is None + assert output_of(call(toolbox, SCHEDULE_QUERY))["count"] == 0 + + +def test_disambiguating_without_candidates_is_refused(toolbox: ToolBox) -> None: + """Asking which one without offering any is a question the user cannot answer.""" + result = call( + toolbox, + REQUEST_USER_INPUT, + question_kind="ambiguous_target", + speech_text="你说的是哪一个？", + ) + + assert result.question is None + assert output_of(result)["status"] == "failed" diff --git a/backend/tests/data/test_voice_to_database.py b/backend/tests/data/test_voice_to_database.py new file mode 100644 index 0000000..f8c7645 --- /dev/null +++ b/backend/tests/data/test_voice_to_database.py @@ -0,0 +1,255 @@ +"""One spoken turn, from the socket to a committed row and back out.""" + +import json +import os +from collections.abc import AsyncIterator, Iterator +from datetime import UTC, datetime +from typing import Any + +import pytest +import sqlalchemy as sa +from fastapi import FastAPI, WebSocket +from fastapi.testclient import TestClient +from sqlalchemy import Engine + +from timeflow.business.calendar.service import ScheduleApplicationService +from timeflow.data.database import build_session_factory +from timeflow.data.schedule_unit_of_work import SqlAlchemyScheduleUnitOfWork +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 FAKE_ACCOUNT_ID, FakeTokenVerifier +from timeflow.intelligence.realtime.agent import RealtimeAgent +from timeflow.intelligence.realtime.instructions import build_instructions +from timeflow.intelligence.realtime.schedule_tools import SCHEDULE_CREATE, ToolBox + +HELLO: dict[str, Any] = { + "type": "session.hello", + "request_id": "req_001", + "payload": {"access_token": "token-abc", "device_id": "device_001"}, +} +START: dict[str, Any] = { + "type": "voice.stream.start", + "request_id": "req_002", + "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_002", "payload": {}} + + +class ToolCallingSession: + """A model that hears a request, calls one tool, then says what it did.""" + + def __init__(self, tool: str, arguments: dict[str, Any]) -> None: + """Store the single tool call this session will make.""" + self._tool = tool + self._arguments = arguments + self.tool_results: list[tuple[str, str]] = [] + self.closed = False + + async def send_audio(self, chunk: bytes) -> None: + """Accept the user's audio without inspecting it.""" + + async def finish_input(self) -> None: + """Accept the end of the user's turn.""" + + async def send_tool_result(self, call_id: str, output: str) -> None: + """Record what the tool answered, which is what the model would read.""" + self.tool_results.append((call_id, output)) + + async def close(self) -> None: + """Record release.""" + self.closed = True + + async def pump(self, observer: Any) -> None: + """Hear the user, call the tool, then speak a confirmation.""" + await observer.heard("明天下午三点开会") + await observer.tool_requested("call_001", self._tool, dict(self._arguments)) + await observer.spoke("好，记下了") + await observer.audio(b"pcm-reply") + + +class OneSessionFactory: + """Hand out the one scripted session, recording the tools registered on it.""" + + def __init__(self, session: ToolCallingSession) -> None: + """Store the session to hand out.""" + self._session = session + self.tools: list[dict[str, Any]] = [] + + async def open(self, instructions: str, tools: list[dict[str, Any]]) -> ToolCallingSession: + """Record the registered tools and return the scripted session.""" + self.tools = tools + return self._session + + +async def _one_chunk() -> AsyncIterator[bytes]: + """A turn's worth of audio.""" + yield b"pcm-user" + + +@pytest.fixture(scope="module") +def engine() -> Iterator[Engine]: + """Connect only when an explicit disposable integration database is supplied.""" + database_url = os.getenv("TIMEFLOW_TEST_DATABASE_URL") + if database_url is None: + pytest.skip("TIMEFLOW_TEST_DATABASE_URL is not set") + built = sa.create_engine(database_url) + try: + yield built + finally: + built.dispose() + + +@pytest.fixture +def account(engine: Engine) -> Iterator[str]: + """Give the stand-in verifier's account a row, since schedules reference one.""" + factory = build_session_factory(engine) + now = datetime.now(UTC) + with factory() as session: + session.execute( + sa.text( + "INSERT INTO accounts (id, username, password_hash, created_at, updated_at) " + "VALUES (:id, :username, 'test-hash', :now, :now) " + "ON CONFLICT (id) DO NOTHING" + ), + {"id": FAKE_ACCOUNT_ID, "username": f"{FAKE_ACCOUNT_ID}-user", "now": now}, + ) + session.commit() + try: + yield FAKE_ACCOUNT_ID + finally: + with factory() as session: + session.execute( + sa.text("DELETE FROM schedules WHERE account_id = :id"), {"id": FAKE_ACCOUNT_ID} + ) + session.execute(sa.text("DELETE FROM accounts WHERE id = :id"), {"id": FAKE_ACCOUNT_ID}) + session.commit() + + +def _build_app(engine: Engine, session: ToolCallingSession) -> tuple[FastAPI, OneSessionFactory]: + """Wire the real transport, agent, tools, and database behind one /ws route.""" + application = FastAPI() + connections = ConnectionManager() + factory = OneSessionFactory(session) + session_factory = build_session_factory(engine) + + schedule_service = ScheduleApplicationService( + lambda: SqlAlchemyScheduleUnitOfWork(session_factory) + ) + + def bind_account(account_id: str) -> ToolBox: + return ToolBox(account_id, schedule_service) + + agent = RealtimeAgent( + factory, + WebSocketResultSink(connections), + tools_factory=bind_account, + instructions=build_instructions, + ) + voice_streams = VoiceStreamHandlers( + AgentAudioSink(agent), max_audio_duration_ms=60_000, queue_max_chunks=64 + ) + 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 session the way the composition root does.""" + await run_websocket_session( + websocket, + SessionHandshake(FakeTokenVerifier()), + router, + connections, + UnauthenticatedConnectionLimiter(8), + handshake_timeout_seconds=5, + binary_handler=voice_streams.handle_binary, + disconnect_handler=voice_streams.handle_disconnect, + ) + + return application, factory + + +def _drain(client_socket: Any, wanted: str, limit: int = 12) -> dict[str, Any]: + """Read frames until the wanted type arrives, skipping binary audio.""" + for _ in range(limit): + frame = client_socket.receive() + if frame.get("type") == "websocket.send" and frame.get("text") is not None: + message: dict[str, Any] = json.loads(frame["text"]) + if message["type"] == wanted: + return message + raise AssertionError(f"never saw {wanted}") + + +def test_a_spoken_turn_commits_a_schedule_and_reports_it_to_the_client( + engine: Engine, account: str +) -> None: + """One turn: audio in, a tool call, a committed row, and a result on the wire.""" + session = ToolCallingSession( + SCHEDULE_CREATE, + { + "schedule_type": "time", + "schedule_kind": "once", + "title": "明天下午三点开会", + "start_time": "2026-09-08T15:00:00", + }, + ) + application, factory = _build_app(engine, session) + account_id: str | None = None + + with TestClient(application) as client, client.websocket_connect("/ws") as socket: + socket.send_json(HELLO) + ready = socket.receive_json() + assert ready["type"] == "session.ready" + socket.send_json(START) + started = socket.receive_json() + assert started["type"] == "voice.stream.started" + socket.send_bytes(b"pcm-user") + socket.send_json(END) + + transcript = _drain(socket, "voice.asr.completed") + assert transcript["payload"]["transcript"] == "明天下午三点开会" + result = _drain(socket, "voice.command.result") + + assert result["payload"]["operation"] == "create_schedule" + assert result["payload"]["status"] == "applied" + schedule_id = result["payload"]["schedule"]["schedule"]["id"] + + # The registered tools are the ones the model was actually offered. + assert SCHEDULE_CREATE in {tool["function"]["name"] for tool in factory.tools} + # The model was handed the tool's answer, so it can speak from what was committed. + assert session.tool_results and session.tool_results[0][0] == "call_001" + + with build_session_factory(engine)() as check: + row = check.execute( + sa.text( + "SELECT account_id, title, status, revision, start_time " + "FROM schedules WHERE id = :id" + ), + {"id": schedule_id}, + ).one() + account_id = row.account_id + assert account_id == account + assert row.title == "明天下午三点开会" + assert row.status == "active" + assert row.revision == 1 + # Stored as the instant 15:00 Asia/Shanghai names, which is 07:00 UTC. + assert row.start_time.astimezone(UTC).hour == 7 + check.execute(sa.text("DELETE FROM schedules WHERE id = :id"), {"id": schedule_id}) + check.commit() + + assert account_id is not None diff --git a/backend/tests/infrastructure/audio/test_null_sink.py b/backend/tests/infrastructure/audio/test_null_sink.py new file mode 100644 index 0000000..71ae80b --- /dev/null +++ b/backend/tests/infrastructure/audio/test_null_sink.py @@ -0,0 +1,41 @@ +"""The placeholder audio sink that drains a stream and keeps nothing.""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import AsyncIterator +from dataclasses import dataclass + +from timeflow.infrastructure.audio.null_sink import NullAudioSink + + +@dataclass +class _Stream: + stream_id: str = "stream-1" + + +async def _chunks(payloads: list[bytes]) -> AsyncIterator[bytes]: + for payload in payloads: + yield payload + + +def test_a_drained_stream_is_counted_and_discarded( + caplog: logging.LogCaptureFixture, +) -> None: + with caplog.at_level(logging.INFO): + asyncio.run(NullAudioSink().consume(_chunks([b"ab", b"cde"]), _Stream())) + + (record,) = [r for r in caplog.records if r.message == "audio stream drained"] + assert record.byte_count == 5 + assert record.stream_id == "stream-1" + + +def test_a_stream_that_carried_nothing_still_reports_zero( + caplog: logging.LogCaptureFixture, +) -> None: + with caplog.at_level(logging.INFO): + asyncio.run(NullAudioSink().consume(_chunks([]), _Stream())) + + (record,) = [r for r in caplog.records if r.message == "audio stream drained"] + assert record.byte_count == 0 diff --git a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py index e2be11f..f3bc836 100644 --- a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py +++ b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py @@ -500,3 +500,137 @@ async def scenario() -> None: async def _ready(transport: FakeTransport) -> FakeTransport: """Hand back an already-built transport, as a connect seam would.""" return transport + + +class BinaryTransport(FakeTransport): + """A transport that can also hand back a binary frame.""" + + def __init__(self, *inbound: Any) -> None: + """Queue frames that may be bytes as well as text.""" + super().__init__() + self._frames = list(inbound) + + async def recv(self) -> Any: + """Return the next frame, text or binary.""" + if not self._frames: + raise AssertionError("the pump read past the end of the scripted turn") + return self._frames.pop(0) + + +def test_a_binary_frame_is_skipped_and_the_turn_carries_on() -> None: + """The vendor sends JSON text; a binary frame is ignored rather than fatal.""" + + async def scenario() -> None: + transport = BinaryTransport(b"\x00\x01", _event("response.done")) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [] + + asyncio.run(scenario()) + + +def test_a_frame_that_is_not_json_fails_the_turn() -> None: + """Unparsable text ends the turn with a reason rather than raising.""" + + async def scenario() -> None: + transport = FakeTransport("not json at all") + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.kinds() == ["failed"] + assert "non-JSON" in observer.calls[0][1] + + asyncio.run(scenario()) + + +def test_a_json_frame_that_is_not_an_object_fails_the_turn() -> None: + """A bare array is valid JSON but not an event, so the turn ends.""" + + async def scenario() -> None: + transport = FakeTransport(json.dumps([1, 2, 3])) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.kinds() == ["failed"] + assert "non-object" in observer.calls[0][1] + + asyncio.run(scenario()) + + +def test_an_empty_audio_delta_reaches_nobody() -> None: + """A delta carrying no audio is dropped rather than pushed on as silence.""" + + async def scenario() -> None: + transport = FakeTransport( + _event("response.audio.delta", delta=""), + _event("response.audio.delta"), + _event("response.done"), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [] + + asyncio.run(scenario()) + + +def test_a_tool_call_with_unparsable_arguments_fails_the_turn() -> None: + """Arguments that are not JSON cannot be acted on, so the turn ends with a reason.""" + + async def scenario() -> None: + transport = FakeTransport( + _event( + "response.function_call_arguments.done", + call_id="call_1", + name="schedule_create", + arguments="{not json", + ) + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.kinds() == ["failed"] + + asyncio.run(scenario()) + + +def test_a_tool_call_whose_arguments_are_not_an_object_runs_with_none() -> None: + """Valid JSON that is not an object leaves the tool with no arguments, not a crash.""" + + async def scenario() -> None: + transport = FakeTransport( + _event( + "response.function_call_arguments.done", + call_id="call_1", + name="schedule_query", + arguments="[1, 2]", + ), + _event("response.done"), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls[0] == ("tool", ("call_1", "schedule_query", {})) + + asyncio.run(scenario()) + + +def test_an_error_event_with_no_message_still_reads_as_a_failure() -> None: + """A vendor error without a message gets a stand-in rather than an empty reason.""" + + async def scenario() -> None: + transport = FakeTransport(_event("error")) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG).pump(observer) + + assert observer.calls == [("failed", "realtime session reported an error")] + + asyncio.run(scenario()) diff --git a/backend/tests/intelligence/conversation/test_schedule_tools.py b/backend/tests/intelligence/conversation/test_schedule_tools.py index fab5bcb..f306468 100644 --- a/backend/tests/intelligence/conversation/test_schedule_tools.py +++ b/backend/tests/intelligence/conversation/test_schedule_tools.py @@ -310,3 +310,53 @@ async def test_location_search_remains_truthful_placeholder() -> None: def test_account_id_must_come_from_authenticated_context() -> None: with pytest.raises(ValueError, match="account_id"): build_agent_tool_registry(FakeScheduleService(), "") + + +@pytest.mark.parametrize( + ("tool_name", "arguments", "expected_call"), + [ + ("schedule_query", {"title": "项目同步"}, "query"), + ( + "schedule_update", + {"schedule_id": "schedule-1", "expected_revision": 1, "changes": {"title": "改过的"}}, + "update", + ), + ( + "schedule_delete", + {"schedule_id": "schedule-1", "expected_revision": 1, "schedule_kind": "once"}, + "delete_once", + ), + ( + "schedule_delete", + { + "schedule_id": "schedule-1", + "expected_revision": 1, + "schedule_kind": "recurring", + "scope": "entire_series", + }, + "delete_recurring", + ), + ], +) +@pytest.mark.asyncio +async def test_each_tool_reaches_the_service_call_that_matches_it( + tool_name: str, arguments: dict[str, object], expected_call: str +) -> None: + service = FakeScheduleService() + tool = build_agent_tool_registry(service, "account-1").get(tool_name) + + result = json.loads(await tool.execute(arguments)) + + assert [call for call, _, _ in service.calls] == [expected_call] + assert result["status"] == "ok" + + +@pytest.mark.asyncio +async def test_a_tool_whose_arguments_do_not_map_reports_the_reason() -> None: + service = FakeScheduleService() + tool = build_agent_tool_registry(service, "account-1").get("schedule_update") + + with pytest.raises(ScheduleToolInputError): + await tool.execute({"schedule_id": "schedule-1", "expected_revision": 1}) + + assert service.calls == [] diff --git a/backend/tests/intelligence/realtime/test_realtime_agent.py b/backend/tests/intelligence/realtime/test_realtime_agent.py index a1878ab..64912f7 100644 --- a/backend/tests/intelligence/realtime/test_realtime_agent.py +++ b/backend/tests/intelligence/realtime/test_realtime_agent.py @@ -19,6 +19,7 @@ class _Stream: """Identifiers of the audio stream a turn answers.""" + account_id: str = "acc_test" session_id: str = "ws_session_test" stream_id: str = "stream_test" conversation_id: str = "conversation_test" @@ -242,9 +243,9 @@ async def scenario() -> None: 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 + # Held open past the turn: a follow-up answering a question has to reach a model + # that still remembers asking it. + assert session.closed is False asyncio.run(scenario()) diff --git a/backend/tests/intelligence/realtime/test_realtime_sessions.py b/backend/tests/intelligence/realtime/test_realtime_sessions.py new file mode 100644 index 0000000..a888651 --- /dev/null +++ b/backend/tests/intelligence/realtime/test_realtime_sessions.py @@ -0,0 +1,348 @@ +"""Session reuse, budgets, and tool calls across turns of one conversation.""" + +from __future__ import annotations + +import asyncio +import json +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 ( + SESSION_MAX_AGE_SECONDS, + SESSION_MAX_TURNS, + RealtimeAgent, +) +from timeflow.intelligence.realtime.schedule_tools import ToolResult + + +@dataclass(frozen=True, slots=True) +class _Stream: + """Identifiers of the audio stream a turn answers.""" + + account_id: str = "acc_test" + 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.""" + + calls: list[tuple[str, Any]] = field(default_factory=list) + + async def deliver_transcript(self, transcript: Transcript, stream: Any) -> None: + self.calls.append(("transcript", transcript)) + + async def deliver_reply_text(self, reply: ReplyText, stream: Any) -> None: + self.calls.append(("done" if reply.done else "reply", reply.speech_text)) + + async def deliver_result(self, result: CommandResult, stream: Any) -> None: + self.calls.append(("result", result)) + + async def deliver_question(self, question: DialogueQuestion, stream: Any) -> None: + self.calls.append(("question", question)) + + async def deliver_audio( + self, reply: AudioReply, chunks: AsyncIterator[bytes], stream: Any + ) -> None: + 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 [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: + 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: + self.audio_sent.append(chunk) + + async def finish_input(self) -> None: + self.finished = True + + async def send_tool_result(self, call_id: str, output: str) -> None: + self.tool_results.append((call_id, output)) + + async def close(self) -> None: + self.closed = True + + async def pump(self, observer: Any) -> None: + for kind, payload in self._script: + await getattr(observer, kind)(*payload) + + +class CountingFactory: + """Hand out a fresh scripted session per open, counting the opens.""" + + def __init__(self, script: list[tuple[str, Any]] | None = None) -> None: + self.script = script or [] + self.opened: list[ScriptedSession] = [] + + async def open(self, instructions: str, tools: list[dict[str, Any]]) -> ScriptedSession: + self.opened.append(ScriptedSession(list(self.script))) + return self.opened[-1] + + +class StubToolBox: + """Return a scripted result, recording what the model asked for.""" + + def __init__(self, result: ToolResult) -> None: + self.result = result + self.calls: list[tuple[str, dict[str, Any]]] = [] + + def tools(self) -> list[dict[str, Any]]: + return [{"type": "function", "function": {"name": "schedule_query"}}] + + async def run(self, name: str, arguments: dict[str, Any]) -> ToolResult: + self.calls.append((name, arguments)) + return self.result + + +async def _chunks(*payloads: bytes) -> AsyncIterator[bytes]: + for payload in payloads: + yield payload + + +def test_a_second_turn_reuses_the_session_the_first_one_opened() -> None: + """One conversation gets one session: a follow-up reaches a model that remembers.""" + + async def scenario() -> None: + factory = CountingFactory([("spoke", ("好",))]) + agent = RealtimeAgent(factory, RecordingSink()) + + await agent.handle_audio(_chunks(b"a" * 3200), _Stream()) + await agent.handle_audio(_chunks(b"b" * 3200), _Stream()) + + assert len(factory.opened) == 1 + assert factory.opened[0].closed is False + + asyncio.run(scenario()) + + +def test_separate_conversations_do_not_share_a_session() -> None: + """Two conversations are two sessions, so neither hears the other's turn.""" + + async def scenario() -> None: + factory = CountingFactory([("spoke", ("好",))]) + agent = RealtimeAgent(factory, RecordingSink()) + + await agent.handle_audio(_chunks(b"a" * 3200), _Stream(conversation_id="conv_a")) + await agent.handle_audio(_chunks(b"b" * 3200), _Stream(conversation_id="conv_b")) + + assert len(factory.opened) == 2 + + asyncio.run(scenario()) + + +def test_a_session_that_used_up_its_turns_is_replaced() -> None: + """A session is swept once it has spent its turn budget, not left holding a connection.""" + + async def scenario() -> None: + factory = CountingFactory([("spoke", ("好",))]) + agent = RealtimeAgent(factory, RecordingSink()) + + for _ in range(SESSION_MAX_TURNS): + await agent.handle_audio(_chunks(b"a" * 3200), _Stream()) + assert len(factory.opened) == 1 + + # The sweep runs at the start of the next turn, so the spent one closes then. + await agent.handle_audio(_chunks(b"a" * 3200), _Stream()) + + assert len(factory.opened) == 2 + assert factory.opened[0].closed is True + + asyncio.run(scenario()) + + +def test_a_committed_tool_call_answers_the_client_and_the_model() -> None: + """One call feeds both sides: the client gets the data, the model gets its result.""" + + async def scenario() -> None: + tools = StubToolBox( + ToolResult( + output=json.dumps({"status": "applied"}), + outcome={"operation": "create_schedule", "status": "applied", "schedule": {}}, + ) + ) + sink = RecordingSink() + factory = CountingFactory( + [("tool_requested", ("call_1", "schedule_create", {"title": "开会"}))] + ) + + await RealtimeAgent(factory, sink, tools_factory=lambda _: tools).handle_audio( # type: ignore[arg-type] + _chunks(b"a" * 3200), _Stream() + ) + + assert tools.calls == [("schedule_create", {"title": "开会"})] + (result,) = [payload for kind, payload in sink.calls if kind == "result"] + assert result.operation == "create_schedule" + assert result.status == "applied" + assert result.schedule == {} + assert result.schedules is None + assert factory.opened[0].tool_results == [("call_1", json.dumps({"status": "applied"}))] + + asyncio.run(scenario()) + + +def test_a_mutation_result_reaches_the_client_flat_not_wrapped_in_the_outcome() -> None: + """The client reads payload.schedule.title, not payload.schedule.schedule.title.""" + + async def scenario() -> None: + snapshot = {"id": "sch_1", "title": "开会", "start_time": "2026-08-13T15:00:00+08:00"} + tools = StubToolBox( + ToolResult( + output=json.dumps({"status": "applied"}), + outcome={"operation": "create_schedule", "status": "applied", "schedule": snapshot}, + ) + ) + sink = RecordingSink() + factory = CountingFactory( + [("tool_requested", ("call_1", "schedule_create", {"title": "开会"}))] + ) + + await RealtimeAgent(factory, sink, tools_factory=lambda _: tools).handle_audio( # type: ignore[arg-type] + _chunks(b"a" * 3200), _Stream() + ) + + (result,) = [payload for kind, payload in sink.calls if kind == "result"] + assert result.schedule == snapshot + assert result.schedules is None + + asyncio.run(scenario()) + + +def test_a_query_result_carries_the_matches_as_schedules_not_schedule() -> None: + """A list_schedules outcome reaches the client as payload.schedules, per protocol §5.6.""" + + async def scenario() -> None: + matches = [{"id": "sch_1", "title": "开会"}, {"id": "sch_2", "title": "晨跑"}] + tools = StubToolBox( + ToolResult( + output=json.dumps({"count": 2}), + outcome={"operation": "list_schedules", "status": "applied", "schedules": matches}, + ) + ) + sink = RecordingSink() + factory = CountingFactory([("tool_requested", ("call_1", "schedule_query", {}))]) + + await RealtimeAgent(factory, sink, tools_factory=lambda _: tools).handle_audio( # type: ignore[arg-type] + _chunks(b"a" * 3200), _Stream() + ) + + (result,) = [payload for kind, payload in sink.calls if kind == "result"] + assert result.operation == "list_schedules" + assert result.schedules == matches + assert result.schedule is None + + asyncio.run(scenario()) + + +def test_a_refused_tool_call_reaches_the_model_and_not_the_client() -> None: + """A refusal has no committed transaction to report, so the client is told nothing.""" + + async def scenario() -> None: + tools = StubToolBox(ToolResult(output=json.dumps({"status": "failed"}))) + sink = RecordingSink() + factory = CountingFactory( + [("tool_requested", ("call_1", "schedule_create", {"title": "开会"}))] + ) + + await RealtimeAgent(factory, sink, tools_factory=lambda _: tools).handle_audio( # type: ignore[arg-type] + _chunks(b"a" * 3200), _Stream() + ) + + assert "result" not in sink.kinds() + assert factory.opened[0].tool_results == [("call_1", json.dumps({"status": "failed"}))] + + asyncio.run(scenario()) + + +def test_a_question_from_a_tool_is_pushed_to_the_client() -> None: + """A tool that needs more from the user turns into a question on the wire.""" + + async def scenario() -> None: + tools = StubToolBox( + ToolResult( + output=json.dumps({"asked": True}), + question={ + "question_kind": "missing_field", + "speech_text": "这个会是哪天的？", + "required_response": "start_time", + "candidates": (), + }, + ) + ) + sink = RecordingSink() + factory = CountingFactory( + [("tool_requested", ("call_1", "request_user_input", {})), ("audio", (b"pcm",))] + ) + + await RealtimeAgent(factory, sink, tools_factory=lambda _: tools).handle_audio( # type: ignore[arg-type] + _chunks(b"a" * 3200), _Stream() + ) + + (question,) = [payload for kind, payload in sink.calls if kind == "question"] + assert question.question_kind == "missing_field" + assert question.speech_text == "这个会是哪天的？" + assert question.required_response == "start_time" + # The audio that follows is marked as asking, not as reporting a change. + (reply,) = [payload for kind, payload in sink.calls if kind == "audio_start"] + assert reply.purpose == "dialogue_question" + + asyncio.run(scenario()) + + +def test_a_tool_call_with_no_tools_registered_is_ignored() -> None: + """With no toolbox bound there is nothing to run, and the turn carries on.""" + + async def scenario() -> None: + sink = RecordingSink() + factory = CountingFactory( + [("tool_requested", ("call_1", "schedule_create", {})), ("spoke", ("好",))] + ) + + await RealtimeAgent(factory, sink).handle_audio(_chunks(b"a" * 3200), _Stream()) + + assert factory.opened[0].tool_results == [] + assert "result" not in sink.kinds() + assert "done" in sink.kinds() + + asyncio.run(scenario()) + + +def test_a_session_that_grew_too_old_is_replaced() -> None: + """Age retires a session even when it has turns left.""" + + async def scenario() -> None: + now = 0.0 + factory = CountingFactory([("spoke", ("好",))]) + agent = RealtimeAgent(factory, RecordingSink(), clock=lambda: now) + + await agent.handle_audio(_chunks(b"a" * 3200), _Stream()) + now = SESSION_MAX_AGE_SECONDS + 1 + await agent.handle_audio(_chunks(b"a" * 3200), _Stream()) + + assert len(factory.opened) == 2 + assert factory.opened[0].closed is True + + asyncio.run(scenario()) diff --git a/backend/tests/intelligence/realtime/test_realtime_toolbox.py b/backend/tests/intelligence/realtime/test_realtime_toolbox.py new file mode 100644 index 0000000..d9a362d --- /dev/null +++ b/backend/tests/intelligence/realtime/test_realtime_toolbox.py @@ -0,0 +1,324 @@ +"""ToolBox routing, questions, and refusals, without a database behind them.""" + +from __future__ import annotations + +import asyncio +import json +from dataclasses import replace +from datetime import UTC, datetime +from typing import Any + +import pytest + +from timeflow.business.calendar import ( + ScheduleAgentService, + ScheduleBusinessError, + ScheduleErrorCode, + ScheduleKind, + ScheduleMutationResult, + ScheduleSearchResult, + ScheduleSnapshot, + ScheduleStatus, + ScheduleType, +) +from timeflow.intelligence.realtime.schedule_tools import ToolBox + +SNAPSHOT = ScheduleSnapshot( + id="sch_1", + account_id="acc_test", + schedule_type=ScheduleType.TIME, + schedule_kind=ScheduleKind.ONCE, + title="写周报", + is_all_day=False, + timezone="Asia/Shanghai", + status=ScheduleStatus.ACTIVE, + revision=1, + created_at=datetime(2026, 9, 7, 1, 0, tzinfo=UTC), + updated_at=datetime(2026, 9, 7, 1, 0, tzinfo=UTC), + # 07:00 UTC is the instant 15:00 Asia/Shanghai names. + start_time=datetime(2026, 9, 8, 7, 0, tzinfo=UTC), +) + + +class RecordingService(ScheduleAgentService): + """Accept every call, remembering which one the ToolBox chose.""" + + def __init__(self) -> None: + self.calls: list[str] = [] + + def create_schedule(self, *, account_id: str, command: Any) -> ScheduleMutationResult: + self.calls.append("create") + return ScheduleMutationResult(schedules=(SNAPSHOT,)) + + def find_schedules(self, *, account_id: str, query: Any) -> ScheduleSearchResult: + self.calls.append("find") + return ScheduleSearchResult(schedules=(SNAPSHOT,)) + + def update_schedule(self, *, account_id: str, command: Any) -> ScheduleMutationResult: + self.calls.append("update") + return ScheduleMutationResult(schedules=(SNAPSHOT,)) + + def delete_once_schedule(self, *, account_id: str, command: Any) -> ScheduleMutationResult: + self.calls.append("delete_once") + return ScheduleMutationResult(schedules=(SNAPSHOT,)) + + def delete_recurring_schedule(self, *, account_id: str, command: Any) -> ScheduleMutationResult: + self.calls.append("delete_recurring") + return ScheduleMutationResult(schedules=()) + + +class RefusingService(ScheduleAgentService): + """Refuse every call, the way the boundary does when a schedule is unacceptable.""" + + def __init__(self, error: ScheduleBusinessError) -> None: + self._error = error + + def create_schedule(self, *, account_id: str, command: Any) -> Any: + raise self._error + + def find_schedules(self, *, account_id: str, query: Any) -> Any: + raise self._error + + def update_schedule(self, *, account_id: str, command: Any) -> Any: + raise self._error + + def delete_once_schedule(self, *, account_id: str, command: Any) -> Any: + raise self._error + + def delete_recurring_schedule(self, *, account_id: str, command: Any) -> Any: + raise self._error + + +def refusing_toolbox() -> ToolBox: + return ToolBox( + "acc_test", + RefusingService( + ScheduleBusinessError( + code=ScheduleErrorCode.REVISION_CONFLICT, + message="那条日程已经变了。", + schedule_id="sch_1", + field="expected_revision", + ) + ), + ) + + +def run(name: str, arguments: dict[str, Any], box: ToolBox | None = None) -> Any: + return asyncio.run((box or refusing_toolbox()).run(name, arguments)) + + +def test_the_tool_schemas_are_handed_out_as_copies() -> None: + box = refusing_toolbox() + box.tools()[0]["type"] = "mutated" + assert all(tool["type"] == "function" for tool in box.tools()) + + +def test_every_registered_tool_has_a_name_and_parameters() -> None: + for tool in refusing_toolbox().tools(): + assert tool["type"] == "function" + assert tool["function"]["name"] + assert tool["function"]["parameters"]["type"] == "object" + + +def test_a_tool_that_is_not_offered_is_refused() -> None: + result = run("schedule_teleport", {}) + assert json.loads(result.output)["status"] == "failed" + assert result.outcome is None + + +def test_an_unmappable_argument_is_refused_before_the_service_is_reached() -> None: + result = run("schedule_create", {"schedule_type": "time", "schedule_kind": "once"}) + payload = json.loads(result.output) + assert payload["status"] == "failed" + assert "title" in payload["error"]["message"] + assert result.outcome is None + + +@pytest.mark.parametrize( + "name", + ["schedule_create", "schedule_query", "schedule_update", "schedule_delete"], +) +def test_a_refused_write_tells_the_model_and_not_the_client(name: str) -> None: + arguments: dict[str, Any] = { + "schedule_create": {"schedule_type": "time", "schedule_kind": "once", "title": "写周报"}, + "schedule_query": {}, + "schedule_update": { + "schedule_id": "sch_1", + "expected_revision": 1, + "changes": {"title": "改过的"}, + }, + "schedule_delete": { + "schedule_id": "sch_1", + "expected_revision": 1, + "schedule_kind": "once", + }, + }[name] + result = run(name, arguments) + payload = json.loads(result.output) + assert payload["status"] == "failed" + assert payload["error"]["code"] == "revision_conflict" + assert payload["error"]["schedule_id"] == "sch_1" + # No transaction committed, so there is no voice.command.result to send (protocol §5.5). + assert result.outcome is None + + +def test_a_question_reaches_the_client_and_not_the_calendar() -> None: + result = run( + "request_user_input", + { + "question_kind": "missing_field", + "speech_text": " 这个会是哪天的？ ", + "required_response": "start_time", + }, + ) + assert json.loads(result.output) == {"asked": True} + assert result.question is not None + assert result.question["speech_text"] == "这个会是哪天的？" + assert result.question["required_response"] == "start_time" + assert result.question["candidates"] == () + assert result.outcome is None + + +def test_an_ambiguous_target_carries_the_candidates_it_found() -> None: + result = run( + "request_user_input", + { + "question_kind": "ambiguous_target", + "speech_text": "是哪一个会？", + "candidates": [{"schedule_id": "sch_1"}, "not an object", {"schedule_id": "sch_2"}], + }, + ) + assert result.question is not None + # Anything that is not an object is dropped rather than passed to the client. + assert result.question["candidates"] == ({"schedule_id": "sch_1"}, {"schedule_id": "sch_2"}) + + +def test_an_ambiguous_target_with_nothing_to_choose_between_is_refused() -> None: + result = run( + "request_user_input", + {"question_kind": "ambiguous_target", "speech_text": "是哪一个会？"}, + ) + assert json.loads(result.output)["status"] == "failed" + assert result.question is None + + +def test_candidates_that_are_not_a_list_are_ignored() -> None: + result = run( + "request_user_input", + { + "question_kind": "missing_field", + "speech_text": "哪天？", + "candidates": "sch_1", + }, + ) + assert result.question is not None + assert result.question["candidates"] == () + + +@pytest.mark.parametrize( + "arguments", + [ + {"question_kind": "telepathy", "speech_text": "哪天？"}, + {"speech_text": "哪天？"}, + {"question_kind": "missing_field"}, + {"question_kind": "missing_field", "speech_text": " "}, + {"question_kind": "missing_field", "speech_text": 7}, + ], +) +def test_a_question_the_client_could_not_show_is_refused(arguments: dict[str, Any]) -> None: + result = run("request_user_input", arguments) + assert json.loads(result.output)["status"] == "failed" + assert result.question is None + assert result.outcome is None + + +@pytest.mark.parametrize( + ("arguments", "expected"), + [ + ({"schedule_kind": "once"}, "delete_once"), + ({"schedule_kind": "recurring", "scope": "entire_series"}, "delete_recurring"), + ], +) +def test_a_delete_reaches_the_call_that_matches_the_kind( + arguments: dict[str, Any], expected: str +) -> None: + service = RecordingService() + run( + "schedule_delete", + {"schedule_id": "sch_1", "expected_revision": 1, **arguments}, + ToolBox("acc_test", service), + ) + assert service.calls == [expected] + + +def test_a_delete_with_nothing_left_to_report_still_says_it_applied() -> None: + result = run( + "schedule_delete", + { + "schedule_id": "sch_1", + "expected_revision": 1, + "schedule_kind": "recurring", + "scope": "entire_series", + }, + ToolBox("acc_test", RecordingService()), + ) + assert json.loads(result.output) == {"status": "applied", "schedule": None} + assert result.outcome is not None + assert result.outcome["schedule"] is None + + +def test_a_committed_write_speaks_the_local_time_and_hides_the_audit_fields() -> None: + result = run( + "schedule_create", + {"schedule_type": "time", "schedule_kind": "once", "title": "写周报"}, + ToolBox("acc_test", RecordingService()), + ) + payload = json.loads(result.output) + assert payload["status"] == "applied" + assert payload["schedule"]["starts_at_local"] == "2026-09-08 15:00" + assert result.outcome is not None + assert result.outcome["operation"] == "create_schedule" + # The client is not told which account the row belongs to, nor when it was audited. + assert "account_id" not in result.outcome["schedule"] + assert "created_at" not in result.outcome["schedule"] + + +def test_a_schedule_without_a_start_time_speaks_no_local_time() -> None: + class LocationService(RecordingService): + def create_schedule(self, *, account_id: str, command: Any) -> ScheduleMutationResult: + return ScheduleMutationResult( + schedules=( + replace( + SNAPSHOT, + schedule_type=ScheduleType.LOCATION, + start_time=None, + location_name="公司", + ), + ) + ) + + result = run( + "schedule_create", + {"schedule_type": "location", "schedule_kind": "once", "title": "到公司"}, + ToolBox("acc_test", LocationService()), + ) + assert json.loads(result.output)["schedule"]["starts_at_local"] == "" + + +def test_a_query_reports_what_it_found_to_both_sides() -> None: + result = run("schedule_query", {}, ToolBox("acc_test", RecordingService())) + payload = json.loads(result.output) + assert payload["count"] == 1 + assert payload["schedules"][0]["starts_at_local"] == "2026-09-08 15:00" + assert result.outcome is not None + assert result.outcome["operation"] == "list_schedules" + assert len(result.outcome["schedules"]) == 1 + + +def test_a_blank_required_response_is_reported_as_absent() -> None: + result = run( + "request_user_input", + {"question_kind": "confirmation", "speech_text": "确认删除？", "required_response": ""}, + ) + assert result.question is not None + assert result.question["required_response"] is None diff --git a/backend/tests/intelligence/realtime/test_tool_mapping.py b/backend/tests/intelligence/realtime/test_tool_mapping.py new file mode 100644 index 0000000..a8588f8 --- /dev/null +++ b/backend/tests/intelligence/realtime/test_tool_mapping.py @@ -0,0 +1,316 @@ +"""Argument mapping and validation tests for the realtime schedule tools.""" + +from __future__ import annotations + +import pytest + +from timeflow.business.calendar import ( + DeleteOnceScheduleCommand, + DeleteRecurringScheduleCommand, + RecurringDeleteScope, + ReminderStrength, + ReminderType, + ScheduleKind, + ScheduleType, +) +from timeflow.intelligence.realtime.tool_mapping import ( + LOCAL, + ToolInputError, + map_create_schedule_command, + map_delete_schedule_command, + map_find_schedules_query, + map_update_schedule_command, + normalize_datetime_args, +) + +MINIMAL_CREATE = { + "schedule_type": "time", + "schedule_kind": "once", + "title": "写周报", +} + + +def test_a_bare_datetime_is_read_as_local_time() -> None: + arguments = normalize_datetime_args({"start_time": "2026-09-08T07:00:00"}) + assert arguments["start_time"] == "2026-09-08T07:00:00+08:00" + + +def test_an_offset_the_model_supplied_is_left_alone() -> None: + arguments = normalize_datetime_args({"start_time": "2026-09-08T07:00:00Z"}) + assert arguments["start_time"] == "2026-09-08T07:00:00Z" + + +def test_a_negative_offset_the_model_supplied_is_left_alone() -> None: + """A "+"/"Z" check alone would miss this and reattach LOCAL, shifting the instant.""" + arguments = normalize_datetime_args({"start_time": "2026-09-08T07:00:00-05:00"}) + assert arguments["start_time"] == "2026-09-08T07:00:00-05:00" + + +def test_nested_change_datetimes_are_normalized_too() -> None: + arguments = normalize_datetime_args({"changes": {"start_time": "2026-09-08T07:00:00"}}) + assert arguments["changes"] == {"start_time": "2026-09-08T07:00:00+08:00"} + + +def test_text_that_only_looks_like_a_datetime_is_left_alone() -> None: + arguments = normalize_datetime_args({"title": "Talk about T-shirts"}) + assert arguments["title"] == "Talk about T-shirts" + + +def test_a_create_without_a_timezone_gets_the_deployment_zone() -> None: + command = map_create_schedule_command(dict(MINIMAL_CREATE)) + assert command.timezone == str(LOCAL.key) + assert command.is_all_day is False + assert command.schedule_type is ScheduleType.TIME + assert command.schedule_kind is ScheduleKind.ONCE + + +def test_a_create_carries_every_optional_field_through() -> None: + command = map_create_schedule_command( + { + **MINIMAL_CREATE, + "schedule_type": "location", + "schedule_kind": "recurring", + "timezone": "UTC", + "is_all_day": True, + "start_time": "2026-09-08T07:00:00+08:00", + "end_time": "2026-09-08T08:00:00+08:00", + "recurrence_rule": "FREQ=WEEKLY", + "location_name": "公司", + "latitude": 31.2, + "longitude": 121.4, + "reminder_type": "at_time", + "reminder_trigger_at": "2026-09-08T06:30:00+08:00", + "reminder_offset_minutes": 30, + "reminder_strength": "high", + } + ) + assert command.timezone == "UTC" + assert command.is_all_day is True + assert command.recurrence_rule == "FREQ=WEEKLY" + assert command.location_name == "公司" + assert command.latitude == 31.2 + assert command.longitude == 121.4 + assert command.reminder_type is ReminderType.AT_TIME + assert command.reminder_offset_minutes == 30 + assert command.reminder_strength is ReminderStrength.HIGH + + +@pytest.mark.parametrize( + ("overrides", "message"), + [ + ({"nickname": "x"}, "Unexpected fields: nickname"), + ({"title": ""}, "title must be a non-empty string"), + ({"title": " "}, "title must be a non-empty string"), + ({"title": 7}, "title must be a non-empty string"), + ({"schedule_type": "telepathy"}, "schedule_type has an unsupported value"), + ({"schedule_type": 7}, "schedule_type must be a string"), + ({"timezone": 7}, "timezone must be a string or null"), + ({"is_all_day": "yes"}, "is_all_day must be a boolean"), + ({"start_time": 7}, "start_time must be an ISO datetime string or null"), + ({"start_time": "tomorrow"}, "start_time must be an ISO datetime string"), + ({"start_time": "2026-09-08T07:00:00"}, "start_time must include a timezone offset"), + ({"latitude": "north"}, "latitude must be a number or null"), + ({"latitude": True}, "latitude must be a number or null"), + ({"latitude": 91}, "latitude must be between -90 and 90"), + ({"longitude": 181}, "longitude must be between -180 and 180"), + ({"reminder_offset_minutes": -1}, "reminder_offset_minutes must be an integer"), + ({"reminder_offset_minutes": True}, "reminder_offset_minutes must be an integer"), + ({"reminder_type": 7}, "reminder_type must be a string or null"), + ({"reminder_type": "telepathy"}, "reminder_type has an unsupported value"), + ], +) +def test_a_create_the_contract_will_not_take_is_refused( + overrides: dict[str, object], message: str +) -> None: + with pytest.raises(ToolInputError, match=message): + map_create_schedule_command({**MINIMAL_CREATE, **overrides}) + + +def test_a_query_takes_every_filter() -> None: + query = map_find_schedules_query( + { + "schedule_id": "sch_1", + "title": "周报", + "starts_at_or_after": "2026-09-08T00:00:00+08:00", + "starts_before": "2026-09-09T00:00:00+08:00", + "location_name": "公司", + "include_deleted": True, + } + ) + assert query.schedule_id == "sch_1" + assert query.include_deleted is True + assert query.starts_at_or_after is not None + assert query.starts_before is not None + + +def test_a_query_with_no_filters_still_maps() -> None: + query = map_find_schedules_query({}) + assert query.schedule_id is None + assert query.include_deleted is False + + +def test_a_query_field_that_is_not_a_filter_is_refused() -> None: + with pytest.raises(ToolInputError, match="Unexpected fields: when"): + map_find_schedules_query({"when": "tomorrow"}) + + +def test_an_update_carries_every_patchable_field_through() -> None: + command = map_update_schedule_command( + { + "schedule_id": "sch_1", + "expected_revision": 3, + "changes": { + "title": "改过的标题", + "is_all_day": True, + "start_time": "2026-09-08T07:00:00+08:00", + "end_time": "2026-09-08T08:00:00+08:00", + "timezone": "UTC", + "recurrence_rule": "FREQ=DAILY", + "location_name": "家", + "latitude": 31.2, + "longitude": 121.4, + "reminder_type": "before_start", + "reminder_trigger_at": "2026-09-08T06:30:00+08:00", + "reminder_offset_minutes": 15, + "reminder_strength": "low", + }, + } + ) + assert command.schedule_id == "sch_1" + assert command.expected_revision == 3 + assert command.changes["title"] == "改过的标题" + assert command.changes["reminder_type"] is ReminderType.BEFORE_START + assert command.changes["reminder_strength"] is ReminderStrength.LOW + assert command.changes["reminder_offset_minutes"] == 15 + + +def test_an_update_can_clear_a_field_by_naming_it_null() -> None: + command = map_update_schedule_command( + { + "schedule_id": "sch_1", + "expected_revision": 1, + "changes": {"location_name": None, "recurrence_rule": None}, + } + ) + # Present-and-None differs from absent: one clears the column, the other leaves it. + assert command.changes["location_name"] is None + assert "start_time" not in command.changes + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + ({"schedule_id": "sch_1", "expected_revision": 1}, "changes must be a non-empty object"), + ( + {"schedule_id": "sch_1", "expected_revision": 1, "changes": {}}, + "changes must be a non-empty object", + ), + ( + {"schedule_id": "sch_1", "expected_revision": 1, "changes": "title"}, + "changes must be a non-empty object", + ), + ({"expected_revision": 1, "changes": {"title": "x"}}, "schedule_id must be a non-empty"), + ( + {"schedule_id": "sch_1", "expected_revision": -1, "changes": {"title": "x"}}, + "expected_revision must be an integer", + ), + ( + {"schedule_id": "sch_1", "expected_revision": True, "changes": {"title": "x"}}, + "expected_revision must be an integer", + ), + ( + {"schedule_id": "sch_1", "expected_revision": 1, "changes": {"colour": "red"}}, + "Unexpected fields: colour", + ), + ( + {"schedule_id": "sch_1", "expected_revision": 1, "changes": {"is_all_day": "yes"}}, + "is_all_day must be a boolean", + ), + ( + {"schedule_id": "sch_1", "expected_revision": 1, "changes": {"timezone": None}}, + "timezone must be a non-empty string", + ), + ({"schedule_id": "sch_1", "expected_revision": 1, "when": "x"}, "Unexpected fields: when"), + ], +) +def test_an_update_the_contract_will_not_take_is_refused( + payload: dict[str, object], message: str +) -> None: + with pytest.raises(ToolInputError, match=message): + map_update_schedule_command(payload) + + +def test_deleting_a_one_off_needs_no_scope() -> None: + command = map_delete_schedule_command( + {"schedule_id": "sch_1", "expected_revision": 2, "schedule_kind": "once"} + ) + assert isinstance(command, DeleteOnceScheduleCommand) + assert command.expected_revision == 2 + + +def test_deleting_a_series_carries_the_scope() -> None: + command = map_delete_schedule_command( + { + "schedule_id": "sch_1", + "expected_revision": 2, + "schedule_kind": "recurring", + "scope": "entire_series", + } + ) + assert isinstance(command, DeleteRecurringScheduleCommand) + assert command.scope is RecurringDeleteScope.ENTIRE_SERIES + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + ( + { + "schedule_id": "sch_1", + "expected_revision": 2, + "schedule_kind": "once", + "scope": "entire_series", + }, + "scope is only valid for recurring schedules", + ), + ( + {"schedule_id": "sch_1", "expected_revision": 2, "schedule_kind": "recurring"}, + "scope is required for recurring schedules", + ), + ( + {"schedule_id": "sch_1", "expected_revision": 2}, + "schedule_kind must be a string", + ), + ( + {"schedule_id": "sch_1", "expected_revision": 2, "schedule_kind": "sometimes"}, + "schedule_kind has an unsupported value", + ), + ( + { + "schedule_id": "sch_1", + "expected_revision": 2, + "schedule_kind": "recurring", + "scope": 7, + }, + "scope must be a string or null", + ), + ( + { + "schedule_id": "sch_1", + "expected_revision": 2, + "schedule_kind": "recurring", + "scope": "just_this_one_maybe", + }, + "scope has an unsupported value", + ), + ( + {"schedule_id": "sch_1", "expected_revision": 2, "schedule_kind": "once", "why": "x"}, + "Unexpected fields: why", + ), + ], +) +def test_a_delete_the_contract_will_not_take_is_refused( + payload: dict[str, object], message: str +) -> None: + with pytest.raises(ToolInputError, match=message): + map_delete_schedule_command(payload) diff --git a/backend/tests/test_agent_delivery.py b/backend/tests/test_agent_delivery.py index 9c199df..159f82d 100644 --- a/backend/tests/test_agent_delivery.py +++ b/backend/tests/test_agent_delivery.py @@ -102,6 +102,30 @@ async def scenario() -> None: assert frame["message_id"] == "msg_a" assert frame["conversation_id"] == "conversation_test" assert frame["payload"]["operation"] == "create_schedule" + # Flat, per protocol §5.5 -- not the whole outcome dict nested under "schedule". + assert frame["payload"]["schedule"] == {"id": "msg_a"} + assert "schedules" not in frame["payload"] + + asyncio.run(scenario()) + + +def test_deliver_result_for_a_query_sends_schedules_not_schedule() -> None: + """A list_schedules outcome carries payload.schedules, per protocol §5.6.""" + + async def scenario() -> None: + connections = ConnectionManager() + connection = RecordingConnection() + connections.register(SESSION_ID, connection) + matches = [{"id": "sch_1"}, {"id": "sch_2"}] + result = CommandResult( + message_id="msg_b", operation="list_schedules", status="applied", schedules=matches + ) + + await WebSocketResultSink(connections).deliver_result(result, _Identity()) + + frame = connection.frames[0] + assert frame["payload"]["schedules"] == matches + assert "schedule" not in frame["payload"] asyncio.run(scenario()) diff --git a/backend/tests/test_app_wiring.py b/backend/tests/test_app_wiring.py index 20eeeed..a121027 100644 --- a/backend/tests/test_app_wiring.py +++ b/backend/tests/test_app_wiring.py @@ -92,6 +92,24 @@ async def consume(self, chunks: object, stream: object) -> None: assert built is not None +def test_voice_agent_mode_two_fails_closed_until_the_conversation_agent_is_wired() -> None: + """The LLM+ASR+TTS pipeline is not an Agent yet, so selecting it must not build silently.""" + get_settings.cache_clear() + try: + with mock.patch.dict( + os.environ, + {"TIMEFLOW_ENVIRONMENT": "development", "TIMEFLOW_VOICE_AGENT_MODE": "2"}, + clear=False, + ): + create_app() + except RuntimeError as error: + assert "TIMEFLOW_VOICE_AGENT_MODE=2" in str(error) + return + finally: + get_settings.cache_clear() + raise AssertionError("expected create_app to refuse voice_agent_mode=2") + + 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. diff --git a/backend/tests/test_schedule_service_skeleton.py b/backend/tests/test_schedule_service_skeleton.py index d0fc19e..135e1cb 100644 --- a/backend/tests/test_schedule_service_skeleton.py +++ b/backend/tests/test_schedule_service_skeleton.py @@ -193,3 +193,36 @@ def test_business_error_codes_are_stable() -> None: "invalid_schedule_kind", "validation_failed", } + + +def test_the_boundary_itself_implements_none_of_the_five_operations() -> None: + """A subclass that defers to the port gets NotImplementedError, never a silent no-op.""" + + class Deferring(ScheduleAgentService): + """Call up to the abstract body for every operation.""" + + def create_schedule(self, *, account_id: str, command: object) -> object: + return super().create_schedule(account_id=account_id, command=command) # type: ignore[arg-type] + + def find_schedules(self, *, account_id: str, query: object) -> object: + return super().find_schedules(account_id=account_id, query=query) # type: ignore[arg-type] + + def update_schedule(self, *, account_id: str, command: object) -> object: + return super().update_schedule(account_id=account_id, command=command) # type: ignore[arg-type] + + def delete_once_schedule(self, *, account_id: str, command: object) -> object: + return super().delete_once_schedule(account_id=account_id, command=command) # type: ignore[arg-type] + + def delete_recurring_schedule(self, *, account_id: str, command: object) -> object: + return super().delete_recurring_schedule(account_id=account_id, command=command) # type: ignore[arg-type] + + deferring = Deferring() + for call in ( + lambda: deferring.create_schedule(account_id="acc", command=None), + lambda: deferring.find_schedules(account_id="acc", query=None), + lambda: deferring.update_schedule(account_id="acc", command=None), + lambda: deferring.delete_once_schedule(account_id="acc", command=None), + lambda: deferring.delete_recurring_schedule(account_id="acc", command=None), + ): + with pytest.raises(NotImplementedError): + call() diff --git a/backend/tests/test_settings.py b/backend/tests/test_settings.py index 52bb9e6..afa8256 100644 --- a/backend/tests/test_settings.py +++ b/backend/tests/test_settings.py @@ -23,6 +23,7 @@ "TIMEFLOW_OPENAI_MODEL", "TIMEFLOW_OPENAI_TIMEOUT_SECONDS", "TIMEFLOW_AGENT_MAX_TOOL_ROUNDS", + "TIMEFLOW_VOICE_AGENT_MODE", ) TTS_ENVIRONMENT_VARIABLES = ( "TIMEFLOW_ALIYUN_TTS_WS_URL", @@ -130,6 +131,7 @@ def test_settings_use_qwen_llm_defaults( assert settings.openai_model == "qwen-flash" assert settings.openai_timeout_seconds == 30.0 assert settings.agent_max_tool_rounds == 4 + assert settings.voice_agent_mode == "1" def test_settings_use_qwen_tts_defaults( @@ -178,6 +180,7 @@ def test_settings_convert_llm_environment_values(monkeypatch: MonkeyPatch) -> No monkeypatch.setenv("TIMEFLOW_OPENAI_MODEL", "custom-model") monkeypatch.setenv("TIMEFLOW_OPENAI_TIMEOUT_SECONDS", "12.5") monkeypatch.setenv("TIMEFLOW_AGENT_MAX_TOOL_ROUNDS", "6") + monkeypatch.setenv("TIMEFLOW_VOICE_AGENT_MODE", "2") settings = Settings.from_environment() @@ -186,6 +189,7 @@ def test_settings_convert_llm_environment_values(monkeypatch: MonkeyPatch) -> No assert settings.openai_model == "custom-model" assert settings.openai_timeout_seconds == 12.5 assert settings.agent_max_tool_rounds == 6 + assert settings.voice_agent_mode == "2" def test_settings_convert_tts_environment_values(monkeypatch: MonkeyPatch) -> None: @@ -250,6 +254,11 @@ def test_settings_convert_tts_environment_values(monkeypatch: MonkeyPatch) -> No "-1", "TIMEFLOW_AGENT_MAX_TOOL_ROUNDS must be a positive integer", ), + ( + "TIMEFLOW_VOICE_AGENT_MODE", + "3", + "TIMEFLOW_VOICE_AGENT_MODE must be '1' or '2'", + ), ( "TIMEFLOW_ALIYUN_TTS_CONNECT_TIMEOUT_SECONDS", "0",