feat(chatto): add chatto_send_ephemeral for self-deleting notices

A status update the agent posts today stays in the timeline forever, so it
competes with the answer for attention. This tool posts through
_post_chunks — inheriting the split and the partial-failure refusal rather
than becoming a third send implementation — and schedules a deletion after
a TTL (default 60 s, 5-3600) so the notice clears itself.

Deletion is deliberately best-effort: it runs in a task held by a
module-level set (asyncio keeps only weak references), a failure is logged
instead of raised because the notice was already delivered, and a gateway
restart before the TTL leaves the message behind rather than pretending
otherwise. The tool obeys the same room gates as chatto_send_room, so it
is not a way into a channel the plugin keeps read-only.

adapter.py

@@ -4404,6 +4404,12 @@ HISTORY_PAGE_MAX = 100
 # kept small enough that a model can digest the output without choking.
 PINNED_PAGE_SIZE = 20
 ATTACHMENT_PAGE_SIZE = 40
+# A self-deleting notice: long enough to read, short enough not to linger.
+# The floor keeps a TTL from deleting a message before anyone sees it; the
+# ceiling keeps a "transient" notice from becoming a day-long fixture.
+EPHEMERAL_TTL_DEFAULT = 60
+EPHEMERAL_TTL_MIN = 5
+EPHEMERAL_TTL_MAX = 3600
 
 
 CHATTO_LIST_ROOMS_SCHEMA = {
@@ -4707,6 +4713,52 @@ CHATTO_SEND_ROOM_SCHEMA = {
     },
 }
 
+CHATTO_SEND_EPHEMERAL_SCHEMA = {
+    "name": "chatto_send_ephemeral",
+    "description": (
+        "Send a text notice to a Chatto room that deletes itself after a TTL. "
+        "Use it for progress or status updates that should not stay in the "
+        "timeline. Same room rules as chatto_send_room: a channel the bot is "
+        "configured to answer in only. Returns the message_ids and the TTL, or "
+        "an error. Deletion is best-effort: a gateway restart before the TTL "
+        "expires leaves the notice in the room."
+    ),
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "room_id": {
+                "type": "string",
+                "description": (
+                    "The Chatto room ID of the destination. Exactly one of "
+                    "`room_id` or `room_name` is required."
+                ),
+            },
+            "room_name": {
+                "type": "string",
+                "description": (
+                    "The display name of the room (case-insensitive). Exactly one "
+                    "of `room_id` or `room_name` is required. If several rooms "
+                    "share the name, the call returns their IDs instead of guessing."
+                ),
+            },
+            "message": {
+                "type": "string",
+                "description": "The notice text (markdown supported).",
+            },
+            "ttl_seconds": {
+                "type": "integer",
+                "description": (
+                    f"Seconds until the notice is deleted (default "
+                    f"{EPHEMERAL_TTL_DEFAULT}, min {EPHEMERAL_TTL_MIN}, "
+                    f"max {EPHEMERAL_TTL_MAX})."
+                ),
+            },
+        },
+        "required": ["message"],
+        "additionalProperties": False,
+    },
+}
+
 CHATTO_MARK_THREAD_AS_READ_SCHEMA = {
     "name": "chatto_mark_thread_as_read",
     "description": (
@@ -5411,6 +5463,109 @@ async def _send_room(args: dict) -> str:
         )
 
 
+_EPHEMERAL_DELETION_TASKS: set[asyncio.Task[None]] = set()
+
+
+async def _send_ephemeral(args: dict) -> str:
+    """Post a notice that deletes itself. See _handle_chatto_send_ephemeral."""
+    room_id = args.get("room_id") or ""
+    room_name = args.get("room_name") or ""
+    message = args.get("message") or ""
+    ttl_seconds = _int_arg(
+        args,
+        "ttl_seconds",
+        EPHEMERAL_TTL_DEFAULT,
+        low=EPHEMERAL_TTL_MIN,
+        high=EPHEMERAL_TTL_MAX,
+    )
+
+    if not message:
+        raise _refuse("message is required")
+    if not room_id and not room_name:
+        raise _refuse("Exactly one of room_id or room_name is required")
+
+    cfg = _chatto_config_for_tools()
+    if cfg is None:
+        raise _refuse("Chatto configuration unavailable")
+    async with _tool_client(cfg) as client:
+        resolved = await _resolve_room(client, room_id, room_name)
+        room = resolved.room
+
+        # Same gates as chatto_send_room: this tool posts, so it must not be a
+        # way into a room the plugin keeps read-only.
+        if not _room_is_writable(cfg, room):
+            raise _read_only_room_refusal(room.name or room.id)
+        if not resolved.is_member:
+            raise _not_joined_refusal(room)
+
+        message_ids, _root = await _post_chunks(client, room.id, message)
+
+    _schedule_ephemeral_deletion(cfg, room.id, message_ids, ttl_seconds)
+    return json.dumps(
+        {
+            "success": True,
+            "room_id": room.id,
+            "message_ids": message_ids,
+            "chunks_sent": len(message_ids),
+            "expires_in_seconds": ttl_seconds,
+        }
+    )
+
+
+def _schedule_ephemeral_deletion(
+    cfg: ChattoConfiguration,
+    room_id: str,
+    event_ids: list[str],
+    ttl_seconds: int,
+) -> None:
+    """Delete a delivered notice once its TTL has passed, best-effort.
+
+    The task is held in a module-level set because asyncio keeps only a weak
+    reference: an unreferenced task can be collected before it ever runs.
+    Failure to delete is logged, never raised — the message was delivered, and
+    a process restart simply leaves it behind.
+    """
+    task = asyncio.create_task(
+        _delete_after_ttl(cfg, room_id, event_ids, ttl_seconds)
+    )
+    _EPHEMERAL_DELETION_TASKS.add(task)
+    task.add_done_callback(_EPHEMERAL_DELETION_TASKS.discard)
+
+
+async def _delete_after_ttl(
+    cfg: ChattoConfiguration,
+    room_id: str,
+    event_ids: list[str],
+    ttl_seconds: int,
+) -> None:
+    """Sleep out the TTL, then delete each chunk it covered."""
+    await asyncio.sleep(ttl_seconds)
+    try:
+        async with _tool_client(cfg) as client:
+            for event_id in event_ids:
+                try:
+                    await client.delete_message(room_id, event_id)
+                except ChattoError as exc:
+                    logger.debug(
+                        "Chatto tools: ephemeral message %s in %s not deleted: %s",
+                        event_id,
+                        room_id,
+                        exc,
+                    )
+    except Exception as exc:
+        # No caller to raise to: the notice was delivered and this task exists
+        # only to make it disappear again. A transport oddity that chattolib
+        # did not map to ChattoError must not surface as an unretrieved task
+        # exception, so it is logged like any other cleanup failure.
+        logger.debug(
+            "Chatto tools: ephemeral cleanup of %s in %s failed: %s",
+            event_ids,
+            room_id,
+            exc,
+            exc_info=True,
+        )
+
+
 async def _list_rooms(args: dict) -> str:
     """List the rooms visible to the bot. See _handle_chatto_list_rooms."""
     name_filter = (args.get("name") or "").strip().casefold()
@@ -5883,6 +6038,24 @@ async def _handle_chatto_send_room(args: dict) -> str:
         return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})
 
 
+async def _handle_chatto_send_ephemeral(args: dict) -> str:
+    """Tool handler for chatto_send_ephemeral.
+
+    Returns a JSON string: Hermes' tool contract accepts no other type, and a
+    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
+    type". Every early exit arrives here as a _ToolRefusal and leaves as that
+    refusal's own JSON, so no path escapes as an unhandled error.
+    """
+    try:
+        return await _send_ephemeral(args)
+    except _ToolRefusal as refusal:
+        return refusal.result_json
+    except ChattoAuthError as exc:
+        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
+    except ChattoError as exc:
+        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})
+
+
 async def _handle_chatto_list_rooms(args: dict) -> str:
     """Tool handler for chatto_list_rooms.
 
@@ -6227,6 +6400,11 @@ def _register_bundled_skills(ctx) -> None:
 _PLUGIN_TOOLS = (
     ("chatto_send_dm", CHATTO_SEND_DM_SCHEMA, _handle_chatto_send_dm),
     ("chatto_send_room", CHATTO_SEND_ROOM_SCHEMA, _handle_chatto_send_room),
+    (
+        "chatto_send_ephemeral",
+        CHATTO_SEND_EPHEMERAL_SCHEMA,
+        _handle_chatto_send_ephemeral,
+    ),
     ("chatto_list_rooms", CHATTO_LIST_ROOMS_SCHEMA, _handle_chatto_list_rooms),
     ("chatto_get_message", CHATTO_GET_MESSAGE_SCHEMA, _handle_chatto_get_message),
     ("chatto_get_history", CHATTO_GET_HISTORY_SCHEMA, _handle_chatto_get_history),

plugin.yaml

@@ -13,6 +13,7 @@ author: Paul Klumpp
 provides_tools:
   - chatto_send_dm
   - chatto_send_room
+  - chatto_send_ephemeral
   - chatto_list_rooms
   - chatto_get_message
   - chatto_get_history

test_adapter.py

@@ -101,6 +101,7 @@ from adapter import (
     _handle_chatto_mark_thread_as_read,
     _handle_chatto_pin_message,
     _handle_chatto_send_dm,
+    _handle_chatto_send_ephemeral,
     _handle_chatto_send_room,
     _handle_chatto_set_custom_status,
     _Mentions,
@@ -4637,6 +4638,7 @@ class TestPluginToolSchemas:
         assert [tool["name"] for tool in ctx.registered_tools] == [
             "chatto_send_dm",
             "chatto_send_room",
+            "chatto_send_ephemeral",
             "chatto_list_rooms",
             "chatto_get_message",
             "chatto_get_history",
@@ -5905,6 +5907,77 @@ class TestPluginToolHandlers:
         assert result["unpinned"] is True
         client.unpin_message.assert_awaited_once_with("room-1", "msg-1")
 
+    # -- chatto_send_ephemeral --
+
+    async def test_send_ephemeral_posts_then_schedules_the_ttl(self):
+        """The handler answers before the notice is gone: deletion is a
+        scheduled task, not something the tool call waits out."""
+        room = _make_room("room-1", "Ops", RoomKind.CHANNEL)
+        client = self._stub_client()
+        client.get_room = AsyncMock(return_value=_make_room_state(room))
+        schedule = MagicMock()
+        with patch("adapter._schedule_ephemeral_deletion", schedule):
+            result, client, _ = await self._run(
+                _handle_chatto_send_ephemeral,
+                {"room_id": "room-1", "message": "working", "ttl_seconds": 30},
+                client=client,
+                optional_mention_rooms="room-1",
+            )
+
+        assert result["success"] is True
+        assert result["expires_in_seconds"] == 30
+        client.post_message.assert_awaited_once()
+        schedule.assert_called_once()
+        _cfg, room_id, event_ids, ttl = schedule.call_args.args
+        assert (room_id, event_ids, ttl) == ("room-1", ["msg-1"], 30)
+
+    async def test_send_ephemeral_refuses_a_read_only_room(self):
+        """It posts, so an unlisted channel stays closed to it too."""
+        room = _make_room("room-1", "Ops", RoomKind.CHANNEL)
+        client = self._stub_client()
+        client.get_room = AsyncMock(return_value=_make_room_state(room))
+        schedule = MagicMock()
+        with patch("adapter._schedule_ephemeral_deletion", schedule):
+            result, _, _ = await self._run(
+                _handle_chatto_send_ephemeral,
+                {"room_id": "room-1", "message": "working"},
+                client=client,
+            )
+
+        assert result["success"] is False
+        assert "read-only" in result["error"]
+        client.post_message.assert_not_awaited()
+        schedule.assert_not_called()
+
+    async def test_delete_after_ttl_removes_every_chunk(self):
+        """A split notice is several messages; the TTL covers all of them."""
+        from adapter import _delete_after_ttl
+
+        client = self._stub_client()
+        client.delete_message = AsyncMock()
+        cfg = ChattoConfiguration(pconfig=_make_config(token="test-token"))
+        with patch("adapter.ChattoClient", return_value=client):
+            await _delete_after_ttl(cfg, "room-1", ["msg-1", "msg-2"], 0)
+
+        assert [call.args for call in client.delete_message.await_args_list] == [
+            ("room-1", "msg-1"),
+            ("room-1", "msg-2"),
+        ]
+        client.close.assert_awaited_once()
+
+    async def test_a_failed_delete_is_swallowed_and_the_rest_still_go(self):
+        """The notice was already delivered; a delete that fails is not an
+        error the caller could act on, and one chunk must not strand the rest."""
+        from adapter import _delete_after_ttl
+
+        client = self._stub_client()
+        client.delete_message = AsyncMock(side_effect=[ChattoError("gone"), None])
+        cfg = ChattoConfiguration(pconfig=_make_config(token="test-token"))
+        with patch("adapter.ChattoClient", return_value=client):
+            await _delete_after_ttl(cfg, "room-1", ["msg-1", "msg-2"], 0)
+
+        assert client.delete_message.await_count == 2
+
     # -- custom status --
 
     async def test_set_custom_status_passes_emoji_and_text(self):