feat: support shared-workspace file flow for matrix

sdk/agent_api_wrapper.py

@@ -86,6 +86,55 @@ class AgentApiWrapper(AgentApi):
             **self._init_kwargs,
         )
 
+    @staticmethod
+    def _event_kind(event: object) -> str:
+        raw_kind = getattr(event, "type", None)
+        if hasattr(raw_kind, "value"):
+            raw_kind = raw_kind.value
+        if raw_kind is None:
+            raw_kind = event.__class__.__name__
+
+        kind = str(raw_kind).replace("-", "_")
+        if "_" in kind:
+            return kind.upper()
+
+        normalized = []
+        for index, char in enumerate(kind):
+            if index and char.isupper() and not kind[index - 1].isupper():
+                normalized.append("_")
+            normalized.append(char)
+        return "".join(normalized).upper()
+
+    @classmethod
+    def _is_kind(cls, event: object, *needles: str) -> bool:
+        kind = cls._event_kind(event)
+        return any(needle in kind for needle in needles)
+
+    @classmethod
+    def _is_text_event(cls, event: object) -> bool:
+        return hasattr(event, "text") or cls._is_kind(event, "TEXT_CHUNK")
+
+    @classmethod
+    def _is_end_event(cls, event: object) -> bool:
+        kind = cls._event_kind(event)
+        return kind == "END" or kind.endswith("_END")
+
+    @classmethod
+    def _is_send_file_event(cls, event: object) -> bool:
+        return "SEND_FILE" in cls._event_kind(event)
+
+    async def _publish_event(self, event: object, *, queue_event: object | None = None) -> None:
+        if self.callback:
+            self.callback(event)
+        if self._current_queue:
+            await self._current_queue.put(queue_event if queue_event is not None else event)
+
+    async def _publish_error(self, event: object) -> None:
+        if self.callback:
+            self.callback(event)
+        if self._current_queue and hasattr(event, "code") and hasattr(event, "details"):
+            await self._current_queue.put(AgentException(getattr(event, "code"), getattr(event, "details")))
+
     async def _listen(self):
         try:
             async for msg in self._ws:
@@ -93,7 +142,7 @@ class AgentApiWrapper(AgentApi):
                     try:
                         outgoing_msg = ServerMessage.validate_json(msg.data)
 
-                        if isinstance(outgoing_msg, MsgEventTextChunk):
+                        if self._is_text_event(outgoing_msg):
                             if self._current_queue:
                                 await self._current_queue.put(outgoing_msg)
                             elif self.callback:
@@ -101,29 +150,22 @@ class AgentApiWrapper(AgentApi):
                             else:
                                 logger.warning("[%s] AgentEvent without active request", self.id)
 
-                        elif isinstance(outgoing_msg, MsgEventEnd):
+                        elif self._is_end_event(outgoing_msg):
                             self.last_tokens_used = outgoing_msg.tokens_used
-                            if self._current_queue:
-                                await self._current_queue.put(outgoing_msg)
+                            await self._publish_event(outgoing_msg)
 
-                        elif isinstance(outgoing_msg, MsgError):
-                            if self.callback:
-                                self.callback(outgoing_msg)
+                        elif self._is_kind(outgoing_msg, "ERROR"):
                             error = AgentException(outgoing_msg.code, outgoing_msg.details)
                             logger.error("[%s] Agent error: %s", self.id, error)
-                            if self._current_queue:
-                                await self._current_queue.put(error)
+                            await self._publish_error(outgoing_msg)
 
-                        elif isinstance(outgoing_msg, MsgGracefulDisconnect):
-                            if self.callback:
-                                self.callback(outgoing_msg)
+                        elif self._is_kind(outgoing_msg, "GRACEFUL_DISCONNECT"):
+                            await self._publish_event(outgoing_msg)
                             logger.info("[%s] Gracefully disconnecting", self.id)
                             break
 
                         else:
-                            logger.warning("[%s] Unknown message type: %s", self.id, outgoing_msg.type)
-                            if self.callback:
-                                self.callback(outgoing_msg)
+                            await self._publish_event(outgoing_msg)
 
                     except Exception as exc:
                         logger.error("[%s] Failed to deserialize message: %s", self.id, exc)

sdk/interface.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 from datetime import datetime
 from typing import Any, AsyncIterator, Literal, Protocol
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 
 class User(BaseModel):
@@ -17,10 +17,11 @@ class User(BaseModel):
 
 
 class Attachment(BaseModel):
-    url: str
-    mime_type: str
+    url: str | None = None
+    mime_type: str | None = None
     size: int | None = None
     filename: str | None = None
+    workspace_path: str | None = None
 
 
 class MessageResponse(BaseModel):
@@ -28,6 +29,7 @@ class MessageResponse(BaseModel):
     response: str
     tokens_used: int
     finished: bool
+    attachments: list[Attachment] = Field(default_factory=list)
 
 
 class MessageChunk(BaseModel):

sdk/real.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
+from pathlib import Path
 from typing import AsyncIterator
 
 from sdk.agent_api_wrapper import AgentApiWrapper
@@ -71,21 +73,43 @@ class RealPlatformClient(PlatformClient):
     ) -> MessageResponse:
         response_parts: list[str] = []
         tokens_used = 0
+        sent_attachments: list[Attachment] = []
         message_id = user_id
+        saw_end_event = False
 
-        async for chunk in self.stream_message(user_id, chat_id, text, attachments=attachments):
-            message_id = chunk.message_id
-            if chunk.delta:
-                response_parts.append(chunk.delta)
-            if chunk.finished:
-                tokens_used = chunk.tokens_used
+        lock = self._get_chat_send_lock(chat_id)
+        async with lock:
+            chat_api = await self._get_chat_api(chat_id)
+            if hasattr(chat_api, "last_tokens_used"):
+                chat_api.last_tokens_used = 0
 
-        return MessageResponse(
-            message_id=message_id,
-            response="".join(response_parts),
-            tokens_used=tokens_used,
-            finished=True,
-        )
+            async for event in self._stream_agent_events(chat_api, text, attachments=attachments):
+                message_id = user_id
+                if self._is_text_event(event):
+                    chunk_text = getattr(event, "text", "")
+                    if chunk_text:
+                        response_parts.append(chunk_text)
+                elif self._is_end_event(event):
+                    tokens_used = getattr(event, "tokens_used", tokens_used)
+                    saw_end_event = True
+                elif self._is_send_file_event(event):
+                    attachment = self._attachment_from_send_file_event(event)
+                    if attachment is not None:
+                        sent_attachments.append(attachment)
+
+            if not saw_end_event:
+                tokens_used = getattr(chat_api, "last_tokens_used", tokens_used)
+            await self._prototype_state.set_last_tokens_used(str(chat_id), tokens_used)
+
+        response_kwargs = {
+            "message_id": message_id,
+            "response": "".join(response_parts),
+            "tokens_used": tokens_used,
+            "finished": True,
+        }
+        if self._message_response_accepts_attachments():
+            response_kwargs["attachments"] = sent_attachments
+        return MessageResponse(**response_kwargs)
 
     async def stream_message(
         self,
@@ -99,20 +123,37 @@ class RealPlatformClient(PlatformClient):
             chat_api = await self._get_chat_api(chat_id)
             if hasattr(chat_api, "last_tokens_used"):
                 chat_api.last_tokens_used = 0
-            async for event in chat_api.send_message(text):
+            saw_end_event = False
+            async for event in self._stream_agent_events(chat_api, text, attachments=attachments):
+                if self._is_text_event(event):
+                    yield MessageChunk(
+                        message_id=user_id,
+                        delta=getattr(event, "text", ""),
+                        finished=False,
+                    )
+                elif self._is_end_event(event):
+                    tokens_used = getattr(event, "tokens_used", 0)
+                    saw_end_event = True
+                    await self._prototype_state.set_last_tokens_used(str(chat_id), tokens_used)
+                    yield MessageChunk(
+                        message_id=user_id,
+                        delta="",
+                        finished=True,
+                        tokens_used=tokens_used,
+                    )
+                elif self._is_send_file_event(event):
+                    continue
+                else:
+                    continue
+            if not saw_end_event:
+                tokens_used = getattr(chat_api, "last_tokens_used", 0)
+                await self._prototype_state.set_last_tokens_used(str(chat_id), tokens_used)
                 yield MessageChunk(
                     message_id=user_id,
-                    delta=event.text,
-                    finished=False,
+                    delta="",
+                    finished=True,
+                    tokens_used=tokens_used,
                 )
-            tokens_used = getattr(chat_api, "last_tokens_used", 0)
-            await self._prototype_state.set_last_tokens_used(str(chat_id), tokens_used)
-            yield MessageChunk(
-                message_id=user_id,
-                delta="",
-                finished=True,
-                tokens_used=tokens_used,
-            )
 
     async def get_settings(self, user_id: str) -> UserSettings:
         return await self._prototype_state.get_settings(user_id)
@@ -140,3 +181,107 @@ class RealPlatformClient(PlatformClient):
             close = getattr(self._agent_api, "close", None)
             if callable(close):
                 await close()
+
+    async def _stream_agent_events(
+        self,
+        chat_api,
+        text: str,
+        attachments: list[Attachment] | None = None,
+    ) -> AsyncIterator[object]:
+        send_message = chat_api.send_message
+        attachment_paths = self._attachment_paths(attachments)
+        if attachment_paths and self._send_message_accepts_attachments(send_message):
+            event_stream = send_message(text, attachments=attachment_paths)
+        else:
+            event_stream = send_message(text)
+        async for event in event_stream:
+            yield event
+
+    @staticmethod
+    def _attachment_paths(attachments: list[Attachment] | None) -> list[str]:
+        if not attachments:
+            return []
+        paths = []
+        for attachment in attachments:
+            if attachment.workspace_path:
+                paths.append(attachment.workspace_path)
+        return paths
+
+    @staticmethod
+    def _send_message_accepts_attachments(send_message) -> bool:
+        try:
+            parameters = inspect.signature(send_message).parameters
+        except (TypeError, ValueError):
+            return False
+        return "attachments" in parameters or any(
+            parameter.kind == inspect.Parameter.VAR_KEYWORD for parameter in parameters.values()
+        )
+
+    @staticmethod
+    def _event_kind(event: object) -> str:
+        raw_kind = getattr(event, "type", None)
+        if hasattr(raw_kind, "value"):
+            raw_kind = raw_kind.value
+        if raw_kind is None:
+            raw_kind = event.__class__.__name__
+
+        kind = str(raw_kind).replace("-", "_")
+        if "_" in kind:
+            return kind.upper()
+        normalized = []
+        for index, char in enumerate(kind):
+            if index and char.isupper() and not kind[index - 1].isupper():
+                normalized.append("_")
+            normalized.append(char)
+        return "".join(normalized).upper()
+
+    @classmethod
+    def _is_text_event(cls, event: object) -> bool:
+        return hasattr(event, "text") or "TEXT_CHUNK" in cls._event_kind(event)
+
+    @classmethod
+    def _is_end_event(cls, event: object) -> bool:
+        kind = cls._event_kind(event)
+        return kind == "END" or kind.endswith("_END")
+
+    @classmethod
+    def _is_send_file_event(cls, event: object) -> bool:
+        kind = cls._event_kind(event)
+        return "SEND_FILE" in kind
+
+    @staticmethod
+    def _attachment_from_send_file_event(event: object) -> Attachment | None:
+        location = None
+        for attr in ("url", "workspace_path", "path", "file_path", "uri"):
+            value = getattr(event, attr, None)
+            if value:
+                location = str(value)
+                break
+        if location is None:
+            return None
+
+        mime_type = getattr(event, "mime_type", None) or "application/octet-stream"
+        filename = getattr(event, "filename", None) or Path(location).name or None
+        size = getattr(event, "size", None)
+        workspace_path = location
+        if workspace_path.startswith("/workspace/"):
+            workspace_path = workspace_path[len("/workspace/"):]
+        elif workspace_path == "/workspace":
+            workspace_path = ""
+        return Attachment(
+            url=location,
+            mime_type=mime_type,
+            size=size,
+            filename=filename,
+            workspace_path=workspace_path or None,
+        )
+
+    @staticmethod
+    def _message_response_accepts_attachments() -> bool:
+        fields = getattr(MessageResponse, "model_fields", None)
+        if isinstance(fields, dict):
+            return "attachments" in fields
+        try:
+            return "attachments" in inspect.signature(MessageResponse).parameters
+        except (TypeError, ValueError):
+            return False