Give the send tool and the adapter one room gate and one chunk loop

_room_is_writable re-implemented _room_policy + _answers_in_room, and
_post_chunks was a third chunked-post loop beside send() — two copies of rules
that must agree, which is how they drift until the tools stop honouring the
platform's own promises.

_room_answers_in(cfg, room_id, kind) is now both gates: a DM answers, anything
else needs a mention list, optional beating require. _room_policy stays because
the /join hint and the joined-rooms log need the enum, not a bool.

_deliver_chunks owns the split, the routing and the stop-at-first-failure and
returns a _ChunkDelivery; send() shapes that into a SendResult, _post_chunks
into a refusal. Those result contracts genuinely differ; the sending did not.

Both extractions verified by mutation. Dropping create_thread from the shared
loop fails the split-chunk and thread-follow tests. Dropping on_posted —
_chatto's echo filter — passed the whole suite, which showed nothing pinned it:
added two tests that the sent message (single and split) lands in the dedup
ring, and both now fail without on_posted.

AGENTS.md

@@ -408,6 +408,29 @@ one job are one too many, because they drift silently until behaviour
 differs (an uncapped downloader next to a capped one shipped here for
 months).
 
+**The two send paths, one implementation.** A user cannot tell whether the agent
+replied on its own or through a tool, so the two must not behave differently.
+Two rules already exist once each, and each answers for both callers:
+
+- **`_room_answers_in`** — whether the bot takes part in a room. It is both the
+  *inbound* gate (`ChattoAdapter._answers_in_room`) and the *outbound* one
+  (`_room_is_writable` on the send tools). Two copies of "a DM answers, a
+  channel needs a mention list" would drift until the send tool became a way
+  around the read-only promise, and the tool's copy is the security-relevant
+  one. `_room_policy` still exists because callers that must *name* the policy
+  (`/join` hint, the joined-rooms log) need the enum, not a bool.
+- **`_deliver_chunks`** — the split at `SPLIT_THRESHOLD`, the routing of a
+  split answer as one answer (chunk 1 opens or joins a thread, later chunks
+  carry that root) and the stop-at-first-failure. `send()` shapes the result
+  into a `SendResult` and `_post_chunks` into a refusal, because those result
+  contracts genuinely differ; the *sending* is the same job and must not be.
+  `on_posted` is how the adapter marks its own output seen — without it the
+  realtime echo of its own message would come back as input.
+
+The standalone cron sender keeps its own copy on purpose (out of process, no
+adapter — see *Why `vendor/` exists*), so agreement there is on the shared
+constants, not on shared code.
+
 **State questions: check what exists before adding.** The sweep above catches
 dead state after the fact; the cheaper move is to not create parallel state
 in the first place. Before introducing a new instance field, look through the

adapter.py

@@ -32,13 +32,13 @@ import os
 import re
 import tempfile
 from collections import deque
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Callable
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from difflib import SequenceMatcher
 from enum import StrEnum
-from typing import Any
+from typing import Any, NamedTuple
 from urllib.parse import unquote, urlsplit
 
 import httpx
@@ -1076,7 +1076,9 @@ class ChattoAdapter(BasePlatformAdapter):
         The two lists are mutually exclusive (enforced by
         ``hermes_validate_config``), so membership decides: optional beats
         require in the face of contradictory runtime config, and a room on
-        neither list stays silent.
+        neither list stays silent. Callers that only need yes/no ask
+        ``_answers_in_room``, which delegates to the shared ``_room_answers_in``
+        the send tools use.
         """
         if room_id in self.chatto_config.optional_mention_rooms.value:
             return RoomPolicy.OPEN
@@ -1087,16 +1089,15 @@ class ChattoAdapter(BasePlatformAdapter):
     def _answers_in_room(self, room_id: str) -> bool:
         """Whether inbound messages from this room reach the agent pipeline.
 
-        Chatto knows only DMs and channels, so the split is ``kind == DM``:
-        a DM always answers (``/join`` must stay reachable), every other
-        room — channel-kind or an unknown kind, which is how servers that
-        never set ``kind`` show up — opts in through the mention lists. A
-        room whose policy is SILENT stays read-only (marked as read, never
-        seeded or answered).
+        Delegates to the shared ``_room_answers_in``, which the send tools gate
+        on too — see there for the rule and why it exists only once. A room
+        that answers is never read-only (its policy is not SILENT), and an
+        unknown kind falls through to the mention lists rather than being
+        mistaken for a DM.
         """
-        if self._room_kinds.get(room_id) == RoomKind.DM:
-            return True
-        return self._room_policy(room_id) != RoomPolicy.SILENT
+        return _room_answers_in(
+            self.chatto_config, room_id, self._room_kinds.get(room_id)
+        )
 
     async def _room_roster_context(
         self, client: ChattoClient, room_id: str
@@ -2288,9 +2289,6 @@ class ChattoAdapter(BasePlatformAdapter):
         if not content:
             return SendResult(success=False, error="Empty message")
 
-        formatted = _normalise_outbound_text(content)
-        chunks = self.truncate_message(formatted, ChattoConstants.SPLIT_THRESHOLD)
-
         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
@@ -2302,45 +2300,39 @@ class ChattoAdapter(BasePlatformAdapter):
             self.chatto_config.auto_thread.value and not thread_id and not is_dm
         )
 
-        message_ids: list[str] = []
-        last_error: str | None = None
-        retryable = False
-
         client = await self._get_chatto_client()
         if client is None:
             return SendResult(
                 success=False, error="Chatto client not available", retryable=True
             )
 
-        for i, chunk in enumerate(chunks):
-            try:
-                msg_obj = await client.post_message(
-                    room_id=chat_id,
-                    body=chunk,
-                    thread_root_event_id=str(thread_id) if thread_id else "",
-                )
-            except Exception as e:
-                # ChattoError included: both read as "this chunk did not go
-                # out" and stop the batch — the SendResult carries the reason.
-                # Only server-side failures count as retryable; a bug in our
-                # own code must not read as a transient network blip.
-                last_error = str(e)
-                retryable = isinstance(e, ChattoError)
-                break
-
-            self._mark_seen(msg_obj.id)
-            message_ids.append(msg_obj.id)
-            # Auto-thread: first chunk becomes the thread root,
-            # subsequent chunks go in the thread
-            if use_auto_thread and i == 0 and not thread_id:
-                thread_id = msg_obj.id
+        # The split, the routing and the stop-at-first-failure all live in
+        # _deliver_chunks, shared with the plugin tools: a long answer must
+        # reach the room the same way whoever sent it.
+        delivery = await _deliver_chunks(
+            client,
+            chat_id,
+            content,
+            thread_root_event_id=str(thread_id) if thread_id else "",
+            create_thread=use_auto_thread,
+            on_posted=self._mark_seen,
+        )
+        message_ids = delivery.message_ids
+        if use_auto_thread and message_ids and not thread_id:
+            thread_id = delivery.root_event_id
 
         # Nothing got through at all — report the failure instead of a phantom success.
         if not message_ids:
+            # Only server-side failures count as retryable; a bug in our own
+            # code must not read as a transient network blip.
             return SendResult(
                 success=False,
-                error=last_error or "Chatto: message could not be sent",
-                retryable=retryable,
+                error=(
+                    str(delivery.error)
+                    if delivery.error
+                    else "Chatto: message could not be sent"
+                ),
+                retryable=isinstance(delivery.error, ChattoError),
             )
 
         first_id = message_ids[0]
@@ -2360,13 +2352,13 @@ class ChattoAdapter(BasePlatformAdapter):
                 )
 
         # A later chunk failed after earlier ones went out: partial delivery.
-        if last_error:
+        if delivery.error is not None:
             logger.warning(
                 "Chatto: sent %d/%d chunk(s) to %s before failing: %s",
                 len(message_ids),
-                len(chunks),
+                delivery.chunks_total,
                 chat_id,
-                last_error,
+                delivery.error,
             )
 
         # raw_response stays unset (dict-shaped per the SendResult contract):
@@ -3743,6 +3735,38 @@ def _chatto_config_for_tools() -> ChattoConfiguration | None:
     return resolved
 
 
+def _room_answers_in(
+    cfg: ChattoConfiguration | None, room_id: str, kind: RoomKind | None
+) -> bool:
+    """Whether the bot takes part in this room: answers there, and may post.
+
+    The single definition of that question, on purpose. It answers for two
+    callers whose disagreement would be a bug the user cannot see:
+
+    - ``ChattoAdapter._answers_in_room``, the *inbound* gate — whether a
+      message here reaches the agent at all
+    - ``_room_is_writable``, the *outbound* gate on the send tools, so a tool
+      cannot post into a room that would never have answered
+
+    Chatto knows only DMs and channels, so the split is ``kind == DM``: a DM
+    always answers (``/join`` must stay reachable), and every other kind —
+    channel, an unknown kind (which is how servers that never set ``kind`` show
+    up), or a room the adapter has not resolved yet, so ``kind`` is None —
+    opts in through one of the two mention lists. The lists are mutually
+    exclusive by validation; should contradictory config reach a running adapter
+    anyway, optional beats require, because answering a room is safer than
+    muting it. A ``cfg`` of None means no configuration could be resolved at
+    all, and then nothing but a DM qualifies.
+    """
+    if kind == RoomKind.DM:
+        return True
+    if cfg is None:
+        return False
+    if room_id in cfg.optional_mention_rooms.value:
+        return True
+    return room_id in cfg.require_mention_rooms.value
+
+
 def _pconfig_or_empty(pconfig: PlatformConfig | None) -> PlatformConfig:
     """The Chatto PlatformConfig from config.yaml, or an empty stand-in."""
     if pconfig is not None:
@@ -3757,19 +3781,12 @@ def _room_is_writable(
 ) -> bool:
     """Whether the bot may post into this room on its own initiative.
 
-    Mirrors ``ChattoAdapter._room_policy``: a DM always qualifies (its
-    conversations are not gated), while a channel has to be opted in through one
-    of the two mention lists. A channel on neither list stays read-only, and a
-    send tool must not become the way around that.
+    Delegates to ``_room_answers_in`` so the outbound gate is the *same*
+    function as the inbound one: a room that would never be answered must not
+    be postable, and two copies of that rule would drift apart until the send
+    tool became a way around the read-only promise.
     """
-    if room.kind == RoomKind.DM:
-        return True
-    listed = (
-        (cfg.require_mention_rooms.value + cfg.optional_mention_rooms.value)
-        if cfg
-        else []
-    )
-    return room_id in listed
+    return _room_answers_in(cfg, room_id, room.kind)
 
 
 def _read_only_room_refusal(room_label: str) -> _ToolRefusal:
@@ -3787,23 +3804,44 @@ def _read_only_room_refusal(room_label: str) -> _ToolRefusal:
     )
 
 
-async def _post_chunks(
+class _ChunkDelivery(NamedTuple):
+    """What a split send actually put into a room.
+
+    ``error`` is the exception that stopped the batch, if any: the chunks in
+    ``message_ids`` did go out and must not be resent. Each caller shapes this
+    into its own result contract — a ``SendResult`` for the adapter, a refusal
+    for the tools — because those are genuinely different contracts; the
+    *sending* is not. ``chunks_total`` counts what the split produced, so a
+    caller can report "got 1 of 3 out" without re-deriving the split.
+    """
+
+    message_ids: list[str]
+    root_event_id: str
+    error: Exception | None
+    chunks_total: int
+
+
+async def _deliver_chunks(
     client: ChattoClient,
     room_id: str,
     message: str,
     *,
     thread_root_event_id: str = "",
     create_thread: bool = False,
-) -> tuple[list[str], str]:
-    """Post a message, split at the server limit. ``(message_ids, root_event_id)``.
-
-    The root is the thread the chunks landed in: the one passed in, or the first
-    chunk's own ID when ``create_thread`` asked for a new thread — the chunks
-    after it then land inside the thread that chunk opened.
-
-    A chunk that fails leaves the earlier ones delivered, so this refuses with
-    exactly what went out: calling a half-delivered send a plain failure
-    invites an agent to retry and duplicate what the room already has.
+    on_posted: Callable[[str], None] | None = None,
+) -> _ChunkDelivery:
+    """Post a message split at the server limit, stopping at the first failure.
+
+    One implementation for both send paths, because the decisions are the same
+    and a copy of each drifts: split at ``SPLIT_THRESHOLD`` like every other
+    send, and route a split answer as *one* answer — chunk 1 opens a thread
+    (``create_thread``) or joins the one passed in, and every later chunk
+    carries that root, so a long reply does not scatter across the timeline.
+
+    ``on_posted`` is the adapter's ``_mark_seen``: the tools have no dedup state
+    to keep, the adapter must not treat its own output as unread input. The
+    first failure stops the batch and is returned rather than raised — the
+    caller knows better than this function what a partial send means to it.
     """
     normalised = _normalise_outbound_text(message)
     chunks = BasePlatformAdapter.truncate_message(
@@ -3811,26 +3849,54 @@ async def _post_chunks(
     )
     message_ids: list[str] = []
     root_id = thread_root_event_id
-    for index, chunk in enumerate(chunks):
+    for chunk in chunks:
         try:
             posted = await client.post_message(
-                room_id, chunk, thread_root_event_id=root_id
+                room_id=room_id, body=chunk, thread_root_event_id=root_id
             )
-        except (ChattoError, RuntimeError) as exc:
-            raise _refuse(
-                f"Chatto API error on chunk {index + 1}/{len(chunks)}: {exc}",
-                # partial marks a half-delivered send; the caller must not
-                # resend it from the top without knowing about these chunks.
-                partial=bool(message_ids),
-                room_id=room_id,
-                message_ids=message_ids,
-                chunks_sent=len(message_ids),
-                chunks_total=len(chunks),
-            ) from exc
+        except Exception as exc:
+            return _ChunkDelivery(message_ids, root_id, exc, len(chunks))
         message_ids.append(posted.id)
+        if on_posted is not None:
+            on_posted(posted.id)
         if create_thread and not root_id:
             root_id = posted.id
-    return message_ids, root_id
+    return _ChunkDelivery(message_ids, root_id, None, len(chunks))
+
+
+async def _post_chunks(
+    client: ChattoClient,
+    room_id: str,
+    message: str,
+    *,
+    thread_root_event_id: str = "",
+    create_thread: bool = False,
+) -> tuple[list[str], str]:
+    """Post a message through ``_deliver_chunks`` as a tool result. ``(ids, root)``.
+
+    A chunk that fails leaves the earlier ones delivered, so this refuses with
+    exactly what went out: calling a half-delivered send a plain failure
+    invites an agent to retry and duplicate what the room already has.
+    """
+    delivery = await _deliver_chunks(
+        client,
+        room_id,
+        message,
+        thread_root_event_id=thread_root_event_id,
+        create_thread=create_thread,
+    )
+    if delivery.error is not None:
+        raise _refuse(
+            f"Chatto API error: {delivery.error}",
+            # partial marks a half-delivered send; the caller must not
+            # resend it from the top without knowing about these chunks.
+            partial=bool(delivery.message_ids),
+            room_id=room_id,
+            message_ids=delivery.message_ids,
+            chunks_sent=len(delivery.message_ids),
+            chunks_total=delivery.chunks_total,
+        ) from delivery.error
+    return delivery.message_ids, delivery.root_event_id
 
 
 async def _find_user_by_login(client: ChattoClient, user_login: str) -> tuple[str, str]:

test_adapter.py

@@ -81,6 +81,7 @@ from adapter import (
     _handle_chatto_list_rooms,
     _handle_chatto_send_dm,
     _handle_chatto_send_room,
+    _room_answers_in,
     chat_type_for_room_kind,
     hermes_env_enablement_fn,
     hermes_standalone_sender_fn,
@@ -565,6 +566,28 @@ class TestSend:
         call_kwargs = adapter._chatto_client.post_message.call_args.kwargs
         assert call_kwargs["thread_root_event_id"] == "thread-123"
 
+    async def test_send_marks_its_own_post_as_seen(self, adapter):
+        """Chatto echoes a posted message back over the realtime channel. The
+        dedup ring is one of the two things that must stop that echo becoming
+        input (the actor_id self-filter is the other), so a sent message has to
+        land in it — the bot must never answer itself.
+        """
+        adapter._chatto_client.post_message.return_value = _make_message(
+            body="Hello world", message_id="msg-echo"
+        )
+        await adapter.send("room-1", "Hello world")
+        assert adapter._is_seen("msg-echo") is True
+
+    async def test_send_marks_every_chunk_of_a_split_as_seen(self, adapter):
+        adapter._chatto_client.post_message.side_effect = [
+            _make_message(body="a", message_id="chunk-1"),
+            _make_message(body="b", message_id="chunk-2"),
+        ]
+        result = await adapter.send("room-1", "x" * 12000)
+        assert result.success is True
+        assert adapter._is_seen("chunk-1") is True
+        assert adapter._is_seen("chunk-2") is True
+
     async def test_send_raw_response_stays_dict_shaped(self, adapter):
         # The cron scheduler calls .get() on SendResult.raw_response when a job
         # targets a thread; a chattolib Message there crashes delivery
@@ -3659,6 +3682,77 @@ class TestPluginToolHandlers:
         client.start_dm.assert_not_awaited()
         client.post_message.assert_not_awaited()
 
+    # -- the two room gates must not drift --
+
+    @pytest.mark.parametrize(
+        ("kind", "require", "optional"),
+        [
+            (RoomKind.CHANNEL, [], []),
+            (RoomKind.CHANNEL, ["room-1"], []),
+            (RoomKind.CHANNEL, [], ["room-1"]),
+            (RoomKind.CHANNEL, ["room-1"], ["room-1"]),
+            (RoomKind.UNSPECIFIED, [], []),
+            (RoomKind.UNSPECIFIED, [], ["room-1"]),
+            (RoomKind.DM, [], []),
+            (RoomKind.DM, ["room-1"], []),
+        ],
+        ids=[
+            "unlisted-channel",
+            "require-only",
+            "optional-only",
+            "on-both-lists",
+            "unknown-kind-unlisted",
+            "unknown-kind-optional",
+            "dm-unlisted",
+            "dm-on-require-list",
+        ],
+    )
+    async def test_send_gate_agrees_with_the_inbound_gate(
+        self, kind, require, optional
+    ):
+        """The send tool must gate exactly what the inbound path gates. A DM
+        answers regardless; any other kind answers only from a mention list,
+        with optional beating require. The tool's refusal and the adapter's
+        answer are asserted separately, so neither can pass by being wrong in
+        the same direction — and a room that stays silent must also refuse to
+        be posted into.
+        """
+        answers = _room_answers_in(
+            ChattoConfiguration(
+                pconfig=_make_config(
+                    token="test-token",
+                    require_mention_rooms=require,
+                    optional_mention_rooms=optional,
+                )
+            ),
+            "room-1",
+            kind,
+        )
+
+        room = _make_room("room-1", "Ops", kind)
+        client = self._stub_client()
+        client.get_room = AsyncMock(return_value=_make_room_state(room))
+        result, _, _ = await self._run(
+            _handle_chatto_send_room,
+            {"room_id": "room-1", "message": "hi"},
+            client=client,
+            require_mention_rooms=require,
+            optional_mention_rooms=optional,
+        )
+        assert result["success"] is answers, (
+            f"send gate for {kind}: tool="
+            f"{'accepted' if result['success'] else 'refused'}, "
+            f"rule says={'answers' if answers else 'silent'}"
+        )
+
+        adapter = _make_adapter(
+            require_mention_rooms=require, optional_mention_rooms=optional
+        )
+        adapter._room_kinds["room-1"] = kind
+        assert adapter._answers_in_room("room-1") is answers, (
+            f"inbound gate for {kind} disagrees with the shared rule"
+        )
+
     # -- room policy --
 
     async def test_send_room_refuses_a_read_only_room(self):
@@ -3820,6 +3914,27 @@ class TestPluginToolHandlers:
 
     # -- chunked delivery --
 
+    async def test_tool_send_keeps_split_chunks_in_the_thread_chunk_one_opened(self):
+        """A split message is one answer, so it must not scatter across the
+        timeline: chunk 1 posts to the room and roots a thread, and every later
+        chunk carries that root. The adapter's send() made this choice first;
+        the tool path has to make the same one."""
+        room = _make_room("room-1", "Ops", RoomKind.CHANNEL)
+        client = self._stub_client()
+        client.get_room = AsyncMock(return_value=_make_room_state(room))
+        result, _, _ = await self._run(
+            _handle_chatto_send_room,
+            {"room_id": "room-1", "message": "x" * 12000, "create_thread": True},
+            client=client,
+            optional_mention_rooms="room-1",
+        )
+        assert result["success"] is True
+        roots = [
+            call.kwargs["thread_root_event_id"]
+            for call in client.post_message.await_args_list
+        ]
+        assert roots == ["", "msg-1"]
+
     async def test_send_room_reports_a_partial_delivery(self):
         """Chunks already sent are out in the world; reporting total failure
         invites a retry that sends them twice."""