Thread attachment sends like text sends

_post_attachment_message turned every reply_to into a thread root,
ignoring auto_thread and the DM check: with auto_thread=false a text
reply landed in the room while its image/document sibling quietly
opened a thread. Both paths now resolve their thread through one
_resolve_outbound_thread helper — metadata thread_id wins, reply_to
anchors only with auto_thread enabled, DMs never thread. Attachments
still never open an auto-thread of their own; only send()'s chunked
text path does, because only there do further chunks follow.

adapter.py

@@ -1686,6 +1686,30 @@ class ChattoAdapter(BasePlatformAdapter):
     # Sending (ConnectRPC — unchanged from polling version)
     # ------------------------------------------------------------------ #
 
+    def _resolve_outbound_thread(
+        self,
+        chat_id: str,
+        reply_to: str | None,
+        metadata: dict[str, Any] | None,
+    ) -> str | None:
+        """The thread an outbound message belongs in, or ``None`` for the room.
+
+        One definition for every outbound path (text and attachments):
+        ``metadata["thread_id"]`` wins — it names the thread root to stay in;
+        otherwise ``reply_to`` anchors a thread under the incoming message,
+        but only with ``auto_thread`` enabled. DMs never thread.
+
+        Deliberately does NOT open an auto-thread of its own — only the
+        chunked text path in :meth:`send` does that, because only there do
+        further chunks follow into the freshly created thread.
+        """
+        thread_id = (metadata or {}).get("thread_id")
+        if reply_to and self.chatto_config.auto_thread.value and not thread_id:
+            thread_id = reply_to
+        if self._room_kinds.get(chat_id) == RoomKind.DM:
+            return None
+        return str(thread_id) if thread_id else None
+
     async def send(
         self,
         chat_id: str,
@@ -1713,21 +1737,9 @@ class ChattoAdapter(BasePlatformAdapter):
         formatted = self.format_message(content)
         chunks = self.truncate_message(formatted, ChattoConstants.MAX_MESSAGE_LENGTH)
 
-        # Thread support — resolve thread_id once
-        # DM rooms don't support threads, so skip threading for DMs
-        thread_id = (metadata or {}).get("thread_id")
-        # Only use reply_to as thread_id if auto_thread is enabled.
-        # When auto_thread=false, responses go directly in the room
-        # without threading under the incoming message.
-        # If we already have thread_id from metadata, keep it (it's the
-        # thread root); only use reply_to when we don't already have one.
-        if reply_to and self.chatto_config.auto_thread.value and not thread_id:
-            thread_id = reply_to
-        # Check if this is a DM room — DMs don't support threads
+        thread_id = self._resolve_outbound_thread(chat_id, reply_to, metadata)
         room_kind = self._room_kinds.get(chat_id)
         is_dm = room_kind == RoomKind.DM
-        if is_dm:
-            thread_id = None
 
         # Auto-thread: by default, Chatto creates a thread for replies to room
         # messages (not DMs, not already in a thread). This keeps conversations
@@ -2332,10 +2344,14 @@ class ChattoAdapter(BasePlatformAdapter):
         reply_to: str | None,
         metadata: dict[str, Any] | None,
     ) -> SendResult:
-        """Post one message carrying already-uploaded assets."""
-        thread_id = (metadata or {}).get("thread_id")
-        if reply_to:
-            thread_id = reply_to
+        """Post one message carrying already-uploaded assets.
+
+        Threading follows the same rules as a text send
+        (:meth:`_resolve_outbound_thread`), so with ``auto_thread`` disabled an
+        attachment reply lands in the room like its text counterpart instead of
+        quietly opening a thread.
+        """
+        thread_id = self._resolve_outbound_thread(chat_id, reply_to, metadata)
 
         client = await self._get_chatto_client()
         if client is None:
@@ -2347,7 +2363,7 @@ class ChattoAdapter(BasePlatformAdapter):
                 room_id=str(chat_id),
                 body=self.format_message(caption) if caption else "",
                 attachment_asset_ids=asset_ids,
-                thread_root_event_id=str(thread_id) if thread_id else "",
+                thread_root_event_id=thread_id or "",
             )
             self._mark_seen(msg.id)
             return SendResult(success=True, message_id=msg.id)

test_adapter.py

@@ -761,6 +761,36 @@ class TestNativeSends:
         assert sent_text.startswith("a clip\n")
         assert "/tmp/clip.mp4" not in sent_text
 
+    async def test_attachment_reply_threads_only_with_auto_thread(self, adapter):
+        """auto_thread=false must keep attachment replies in the room, exactly
+        like text replies — no quiet thread under the incoming message."""
+        adapter.chatto_config.auto_thread.value = False
+        await adapter.send_document("room-1", "/tmp/thing.bin", reply_to="incoming-1")
+        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
+        assert call_kwargs["thread_root_event_id"] == ""
+
+    async def test_attachment_reply_threads_under_the_incoming_message(self, adapter):
+        await adapter.send_document("room-1", "/tmp/thing.bin", reply_to="incoming-1")
+        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
+        assert call_kwargs["thread_root_event_id"] == "incoming-1"
+
+    async def test_attachment_metadata_thread_id_wins_over_reply_to(self, adapter):
+        await adapter.send_document(
+            "room-1",
+            "/tmp/thing.bin",
+            reply_to="incoming-1",
+            metadata={"thread_id": "root-9"},
+        )
+        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
+        assert call_kwargs["thread_root_event_id"] == "root-9"
+
+    async def test_attachment_replies_to_dms_never_thread(self, adapter):
+        """DMs don't support threads — not even with auto_thread enabled."""
+        adapter._room_kinds["dm-1"] = RoomKind.DM
+        await adapter.send_document("dm-1", "/tmp/thing.bin", reply_to="incoming-1")
+        call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
+        assert call_kwargs["thread_root_event_id"] == ""
+
 
 # -- Batched image delivery --