Implement edit_message, delete_message, create_handoff_thread, format_message

The stream consumer drives streaming replies through edit_message
(gateway/stream_consumer.py:412, run.py:4530 and friends). The base class
reports 'Not supported', so every incremental update was arriving as a NEW
message and long answers came out as a cascade. chattolib has had
update_message/delete_message all along.

edit_message refuses content beyond the per-message limit rather than
truncating it: the caller then falls back to send(), which splits properly.
finalize is a no-op here — Chatto has no in-progress card state to close out,
so no REQUIRES_EDIT_FINALIZE.

delete_message backs the stream consumer's fresh-final cleanup and the
ephemeral-reply TTL.

create_handoff_thread anchors a handoff on a seed message (Chatto threads
hang off a message, not off the room) and returns its id; DMs get None
because they cannot carry threads.

format_message stays deliberately thin — Chatto renders Markdown natively,
so it only normalises CRLF and collapses runs of blank lines. Its call site
in send() was guarded by a hasattr() that could never be false.

adapter.py

@@ -850,7 +850,7 @@ class ChattoAdapter(BasePlatformAdapter):
         if not content:
             return SendResult(success=False, error="Empty message")
 
-        formatted = self.format_message(content) if hasattr(self, "format_message") else content
+        formatted = self.format_message(content)
         chunks = self.truncate_message(formatted, ChattoConstants.MAX_MESSAGE_LENGTH)
 
         # Thread support — resolve thread_id once
@@ -945,6 +945,165 @@ class ChattoAdapter(BasePlatformAdapter):
 
         return SendResult(success=True, message_id=first_id, raw_response=last_resp)
 
+    def format_message(self, content: str) -> str:
+        """Normalise outgoing text for Chatto.
+
+        Chatto renders Markdown natively, so there is nothing to escape or
+        translate — the only transformations here are the ones that measurably
+        render wrong: CRLF line endings (which show up as stray blank lines)
+        and runs of more than two blank lines.
+
+        BasePlatformAdapter override
+        """
+        if not content:
+            return content
+        normalised = content.replace("\r\n", "\n").replace("\r", "\n")
+        while "\n\n\n\n" in normalised:
+            normalised = normalised.replace("\n\n\n\n", "\n\n\n")
+        return normalised
+
+    async def edit_message(
+        self,
+        chat_id: str,
+        message_id: str,
+        content: str,
+        *,
+        finalize: bool = False,
+    ) -> SendResult:
+        """Edit a message we previously sent, via MessageService/UpdateMessage.
+
+        The stream consumer drives streaming replies through this: without the
+        override the base class reports "Not supported" and every incremental
+        update arrives as a *new* message.
+
+        ``finalize`` is a no-op for Chatto — an edit is an edit here, there is
+        no in-progress card state to close out (hence no
+        ``REQUIRES_EDIT_FINALIZE``).
+
+        Content that exceeds the per-message limit is refused rather than
+        silently truncated, so the caller falls back to ``send()``, which
+        splits across messages.
+
+        BasePlatformAdapter override
+        """
+        if not message_id:
+            return SendResult(success=False, error="Chatto: no message id to edit")
+        if not content:
+            return SendResult(success=False, error="Empty message")
+
+        formatted = self.format_message(content)
+        if len(formatted) > ChattoConstants.MAX_MESSAGE_LENGTH:
+            # Refuse instead of truncating: the caller's fallback path splits.
+            return SendResult(
+                success=False,
+                error=(
+                    f"Chatto: edit exceeds {ChattoConstants.MAX_MESSAGE_LENGTH} "
+                    f"chars ({len(formatted)})"
+                ),
+            )
+
+        try:
+            client = await self._require_client()
+        except RuntimeError:
+            return SendResult(success=False, error="Chatto client not available", retryable=True)
+
+        try:
+            msg = await client.update_message(
+                room_id=str(chat_id),
+                event_id=str(message_id),
+                body=formatted,
+            )
+        except ChattoError as e:
+            logger.warning("Chatto: UpdateMessage failed for %s: %s", message_id, e)
+            return SendResult(success=False, error=str(e), retryable=True)
+        except Exception as e:
+            logger.warning("Chatto: UpdateMessage error for %s: %s", message_id, e)
+            return SendResult(success=False, error=str(e), retryable=False)
+
+        # Our own edit comes back as a message_edited event; mark it seen so it
+        # is never mistaken for inbound traffic.
+        edited_id = getattr(msg, "id", "") or str(message_id)
+        self._mark_seen(edited_id)
+        return SendResult(success=True, message_id=edited_id, raw_response=msg)
+
+    async def delete_message(self, chat_id: str, message_id: str) -> bool:
+        """Delete a message via MessageService/DeleteMessage.
+
+        Used by the stream consumer's fresh-final cleanup (removing a preview
+        message once the completed reply has been sent) and by the ephemeral
+        reply TTL.
+
+        BasePlatformAdapter override
+        """
+        if not chat_id or not message_id:
+            return False
+        try:
+            client = await self._require_client()
+        except RuntimeError:
+            logger.warning("Chatto: DeleteMessage — client unavailable")
+            return False
+        try:
+            return bool(await client.delete_message(
+                room_id=str(chat_id), event_id=str(message_id),
+            ))
+        except ChattoError as e:
+            logger.warning("Chatto: DeleteMessage failed for %s: %s", message_id, e)
+            return False
+        except Exception as e:
+            logger.warning("Chatto: DeleteMessage error for %s: %s", message_id, e)
+            return False
+
+    async def create_handoff_thread(
+        self, parent_chat_id: str, name: str,
+    ) -> Optional[str]:
+        """Anchor a session handoff in a fresh thread under *parent_chat_id*.
+
+        Chatto threads hang off a message, not off the room, so we post a seed
+        message and hand its ID back as the thread root — the same shape the
+        Slack adapter uses.  DMs don't support threads, so they get ``None``
+        and the watcher keeps delivering into the DM itself.
+
+        BasePlatformAdapter override
+        """
+        if not parent_chat_id:
+            return None
+        if self._room_kinds.get(parent_chat_id) == RoomKind.DM:
+            logger.debug("Chatto: handoff thread skipped — %s is a DM", parent_chat_id)
+            return None
+
+        try:
+            client = await self._require_client()
+        except RuntimeError:
+            logger.warning("Chatto: handoff thread — client unavailable")
+            return None
+
+        seed_text = f"🧵 Hermes handoff — **{(name or 'session').strip()[:80]}**"
+        try:
+            msg = await client.post_message(room_id=str(parent_chat_id), body=seed_text)
+        except Exception as e:
+            logger.warning(
+                "Chatto: handoff thread seed-post failed for room %s: %s",
+                parent_chat_id, e,
+            )
+            return None
+
+        seed_id = getattr(msg, "id", "") or ""
+        if not seed_id:
+            logger.warning("Chatto: handoff thread seed-post returned no message id")
+            return None
+
+        self._mark_seen(seed_id)
+        self._our_message_ids.add(seed_id)
+        self._our_thread_roots.add(seed_id)
+        try:
+            await client.follow_thread(str(parent_chat_id), seed_id)
+        except Exception:
+            logger.debug(
+                "Chatto: follow_thread failed for handoff %s/%s",
+                parent_chat_id, seed_id, exc_info=True,
+            )
+        return seed_id
+
     # Overridden from BaseAdapter:
     async def send_typing(self, chat_id: str, metadata=None) -> None:
         """Start a persistent typing indicator for a room.

test_adapter.py

@@ -370,28 +370,123 @@ class TestMessageEditing:
         adapter._token = "test-token"
         return adapter
 
-    @pytest.mark.xfail(
-        strict=True,
-        reason="ChattoAdapter does not override edit_message yet, so the base "
-               "class reports 'Not supported' and callers send a new message "
-               "instead of editing. chattolib.update_message() exists — drop "
-               "this marker once the override lands.",
-    )
     async def test_edit_message(self, adapter):
         result = await adapter.edit_message("room-1", "msg-1", "New content")
         assert result.success is True
         adapter._chatto_client.update_message.assert_called_once()
+        call_kwargs = adapter._chatto_client.update_message.call_args.kwargs
+        assert call_kwargs["room_id"] == "room-1"
+        assert call_kwargs["event_id"] == "msg-1"
+        assert call_kwargs["body"] == "New content"
+
+    async def test_edit_message_marks_own_edit_seen(self, adapter):
+        """The edit echoes back as message_edited — it must not look inbound."""
+        mock_msg = MagicMock()
+        mock_msg.id = "msg-1"
+        adapter._chatto_client.update_message.return_value = mock_msg
+        await adapter.edit_message("room-1", "msg-1", "New content")
+        assert adapter._is_seen("msg-1") is True
+
+    async def test_edit_message_too_long_refuses(self, adapter):
+        """Overlong content must fall back to send() (which splits), not be
+        silently truncated into a lossy edit."""
+        result = await adapter.edit_message(
+            "room-1", "msg-1", "x" * (_MAX_MESSAGE_LENGTH + 1),
+        )
+        assert result.success is False
+        adapter._chatto_client.update_message.assert_not_called()
+
+    async def test_edit_message_empty_content(self, adapter):
+        result = await adapter.edit_message("room-1", "msg-1", "")
+        assert result.success is False
+        adapter._chatto_client.update_message.assert_not_called()
+
+    async def test_edit_message_error_is_retryable(self, adapter):
+        from chattolib.exceptions import ChattoError
+
+        adapter._chatto_client.update_message.side_effect = ChattoError("boom")
+        result = await adapter.edit_message("room-1", "msg-1", "New content")
+        assert result.success is False
+        assert result.retryable is True
 
-    @pytest.mark.xfail(
-        strict=True,
-        reason="ChattoAdapter does not override delete_message yet, so the base "
-               "class returns False. chattolib.delete_message() exists — drop "
-               "this marker once the override lands.",
-    )
     async def test_delete_message(self, adapter):
         result = await adapter.delete_message("room-1", "msg-1")
         assert result is True
         adapter._chatto_client.delete_message.assert_called_once()
+        call_kwargs = adapter._chatto_client.delete_message.call_args.kwargs
+        assert call_kwargs["room_id"] == "room-1"
+        assert call_kwargs["event_id"] == "msg-1"
+
+    async def test_delete_message_missing_ids(self, adapter):
+        assert await adapter.delete_message("", "msg-1") is False
+        assert await adapter.delete_message("room-1", "") is False
+        adapter._chatto_client.delete_message.assert_not_called()
+
+    async def test_delete_message_error_returns_false(self, adapter):
+        from chattolib.exceptions import ChattoError
+
+        adapter._chatto_client.delete_message.side_effect = ChattoError("nope")
+        assert await adapter.delete_message("room-1", "msg-1") is False
+
+
+# -- Outgoing text formatting --
+
+class TestFormatMessage:
+    """format_message() only fixes what renders wrong in Chatto."""
+
+    def test_normalises_crlf(self):
+        adapter = _make_adapter()
+        assert adapter.format_message("a\r\nb\rc") == "a\nb\nc"
+
+    def test_collapses_excess_blank_lines(self):
+        adapter = _make_adapter()
+        assert adapter.format_message("a\n\n\n\n\n\nb") == "a\n\n\nb"
+
+    def test_leaves_markdown_untouched(self):
+        adapter = _make_adapter()
+        text = "**bold** `code`\n\n```py\nx = 1\n```\n- item"
+        assert adapter.format_message(text) == text
+
+    def test_empty_content(self):
+        adapter = _make_adapter()
+        assert adapter.format_message("") == ""
+
+
+# -- Handoff threads --
+
+class TestHandoffThread:
+    """create_handoff_thread() anchors a handoff on a seed message."""
+
+    @pytest_asyncio.fixture
+    def adapter(self):
+        adapter = _make_adapter()
+        adapter._chatto_client.post_message = AsyncMock()
+        adapter._chatto_client.follow_thread = AsyncMock()
+        return adapter
+
+    async def test_returns_seed_message_id(self, adapter):
+        mock_msg = MagicMock()
+        mock_msg.id = "seed-1"
+        adapter._chatto_client.post_message.return_value = mock_msg
+        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
+
+        result = await adapter.create_handoff_thread("room-1", "Refactor run")
+
+        assert result == "seed-1"
+        assert adapter._chatto_client.post_message.call_args.kwargs["room_id"] == "room-1"
+        adapter._chatto_client.follow_thread.assert_called_once_with("room-1", "seed-1")
+        # Our own seed must not come back in as inbound traffic.
+        assert adapter._is_seen("seed-1") is True
+
+    async def test_dm_has_no_threads(self, adapter):
+        adapter._room_kinds["dm-1"] = RoomKind.DM
+        assert await adapter.create_handoff_thread("dm-1", "x") is None
+        adapter._chatto_client.post_message.assert_not_called()
+
+    async def test_seed_post_failure(self, adapter):
+        adapter._chatto_client.post_message.side_effect = RuntimeError("down")
+        adapter._room_kinds["room-1"] = RoomKind.CHANNEL
+        assert await adapter.create_handoff_thread("room-1", "x") is None
 
 
 # -- Reaction event forwarding --