feat(realtime): schedule mutations, real database, and command.result fix

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

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

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

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,

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

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

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.

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:

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