Seed room history and resolve reply targets on fresh room turns

A reply posted in a room without a thread opened a session the agent had
never seen: every fresh channel message anchors a session of its own
(thread_id = message.id), so the turn arrived with no room context and no
idea what the reply pointed at. The bot's own output was unreachable too
— its realtime echo is filtered as a self-event, so no session ever held
what it had answered before.

Two fixes, one contract ("a session that starts here knows what it is
about"):

- The reply pointer: resolve reply_to_text/reply_to_is_own_message via
  get_message on the target, which the gateway renders as
  "[Replying to ...]" — the same context injection Signal does for
  quotes. Best-effort: an unreadable or deleted target costs the
  pointer, never the turn.

- The room block: fetch get_room_events_around and render an
  "[Earlier in this room]" block through the same renderer as the
  thread seed (_earlier_context_block — one renderer, so thread and
  room cannot drift apart), filtered to messages older than the
  trigger. No cursor bookkeeping, unlike the thread seed: every fresh
  message roots a NEW session, so a cursor written for one session's
  frontier would describe another session's past. The one re-dispatch
  path that hits an existing session (an edit restarting a turn) is
  covered by the _session_exists guard, which fails open.

Filtered by timestamp rather than page position because "around" serves
both directions of the anchor and a session start is the past, not the
future.

AGENTS.md

@@ -282,6 +282,26 @@ changes must keep them, or consciously renegotiate the docs.**
   is no session-reset path on this platform — `/join` and `/leave` are the
   only commands and they ride DMs — so an emptied session cannot strand a
   thread the way it would elsewhere; do not assume one exists.
+- **A fresh room turn opens with the room behind it.** Every fresh channel
+  message anchors a session of its own (`thread_id = message.id` when no
+  inbound thread root — unconditional, not gated on `auto_thread`, which only
+  rules whether the *outgoing* reply opens a server-side thread), so a
+  non-thread turn would otherwise start empty even when the room talked for
+  hours. The turn therefore carries an `[Earlier in this room]` block built
+  from `get_room_events_around`, filtered to messages older than the trigger
+  (`around` serves both directions; a session start is the past, not the
+  future), rendered by the same shared renderer as the thread block
+  (`_earlier_context_block`, same caps and labelling — two renderers would
+  drift until thread and room quoted differently). No cursor and no
+  `_seeded` bookkeeping: every fresh message roots a *new* session, so a
+  cursor written for one session's frontier would describe another session's
+  past; the one re-dispatch path that hits an existing session (an edit
+  restarting a turn) is covered by the `_session_exists` guard, which fails
+  open so a first turn always seeds. DMs never seed (one session per DM).
+  A reply also names what it answers: `reply_to_text`/`reply_to_is_own_message`
+  are resolved via `get_message` on the target — the gateway renders
+  `[Replying to …]` from them — best-effort, an unreadable target costs the
+  pointer, never the turn.
 - **Room roster rides on every channel turn.** Threads hold isolated sessions
   and the agent only sees `channel_context` per dispatch, so each channel turn
   carries the current member roster — repetition is the point, deduplicating

CONCEPTS.md

@@ -137,6 +137,65 @@ multiplexed gateway into `agent:main:` and names a session nobody is running.
 - `[Earlier in this thread]` is reference text, not trusted input: only the
   trigger passes the auth gate.
 
+## Room sessions open blind, and why they do not carry a cursor
+
+### The second reported case
+
+The thread seed solved "reply to a cron message opens a thread the agent has
+never seen". The room case is the same failure one level up: a human replies in
+a channel *without* opening a thread, pointing at a message the bot posted. The
+reply roots a session of its own — the anchoring gives every fresh channel
+message `thread_id = message.id`, unconditionally — so the turn is answered by
+an agent that has never seen the room's conversation, nor the bot's own earlier
+answer, nor what the reply points at.
+
+### Two half-fixes, one contract
+
+The fix has two halves, and the user-visible contract is "a session that starts
+here knows what it is about":
+
+1. **The reply pointer.** `MessageEvent.reply_to_text` and
+   `reply_to_is_own_message` are resolved by fetching the target with
+   `get_message`. The gateway renders `[Replying to …]` from them — Signal does
+   the same for quotes — and without them the turn reads as a comment from
+   nowhere even when everything else is in place. One RPC per reply, skipped
+   for non-replies; a deleted or unreadable target costs the pointer, never
+   the turn.
+
+2. **The room block.** `[Earlier in this room]`, built from
+   `get_room_events_around`, filtered to messages older than the trigger.
+   `around` serves both directions of the anchor, and a session start is
+   answered with the past, not the future — a message someone typed while the
+   agent was thinking is not "earlier", and handing it over would answer a
+   question the room may not have asked yet.
+
+### Why no cursor, when the thread seed needs one
+
+The thread seed's whole machinery — `_seeded_threads`, the session copy under
+`chatto.thread_seed_cursor`, the blind pass — exists because a thread session
+*survives across turns*: turn two must know what turn one already delivered, or
+it repeats and skips. A room session does not survive across turns in the same
+sense. Every fresh channel message roots a *new* session, so there is no
+"session frontier" to track: the question each turn asks is "what did the room
+say before this message", and the trigger answers it on its own.
+
+A per-room cursor would be actively wrong here. It would record one session's
+frontier and hand the *next* message's session a block written for the first
+one's — delivering messages the second conversation never saw and skipping
+others. The one dispatch path where the same message *does* hit a session that
+already exists is an edit restarting a turn, and that is covered by the
+`_session_exists` guard (fail-open when no store is wired, so a first turn
+always seeds).
+
+### What this costs
+
+- One extra RPC per first turn into a room: `get_room_events_around`, capped at
+  20 events. The roster fetch already dominates the per-turn budget for a
+  channel turn, and the lookup is skipped entirely for DMs.
+- `[Earlier in this room]` is reference text, not trusted input: only the
+  trigger passes the auth gate, authors off the allowlist are labelled
+  `(unverified)` by the shared renderer, own output `(this bot)`.
+
 ## Realtime resume
 
 ### The gap

README.md

@@ -16,6 +16,7 @@ Before setup, here's the part most people want to know: how Hermes behaves once
 | **DMs** | For such a user, Hermes answers every DM without an `@mention`. Each DM has its own session. |
 | **Rooms** | For such a user, opt-in per room: a room listed in `CHATTO_REQUIRE_MENTION_ROOMS` gets answers only when Hermes is addressed (`@mention`, `@all`, `@here`); one listed in `CHATTO_OPTIONAL_MENTION_ROOMS` gets an answer for every message. A room on neither list stays silent — read-only. |
 | **Threads** | If you reply in a thread, Hermes keeps the thread context isolated from the parent room. The first turn into a thread also carries the messages that came before it — including ones the bot itself posted, such as a cron report nobody replied to — so the agent never answers a reply whose thread it cannot read. The bot auto-follows threads it participates in. |
+| **Fresh room messages** | A message posted in a room without a thread opens its own session, so it carries what the room said before it (an `[Earlier in this room]` block, same budget as the thread block), and a reply names what it answers (`[Replying to "…"]`). The agent never answers a room message as if the room had been silent. |
 | **Room awareness** | Every channel turn comes with the member roster attached (who is in the room, with live presence), so Hermes always knows his audience — in every thread, every turn. The roster is fetched once per room and kept fresh from presence events without extra directory lookups; the first 20 members are listed, larger rooms get an "and N more" note. DMs never carry one. |
 | **Processing indicators** | Hermes adds a 👀 reaction when it starts processing a message, and replaces it with ✅ on success, ❌ on failure, or 🚫 when processing was cancelled. |
 | **Editing your messages** | An edit within 5 minutes of posting counts as a correction: while Hermes is still working on the original, he restarts with the edited text (🚫 → 👀); if he had ignored or not yet answered the message, the gates are re-checked against the new text — adding a forgotten `@mention` this way works. Edits to messages Hermes has already answered change nothing. Disable with `CHATTO_EDIT_DISPATCH=false`. |
@@ -229,6 +230,12 @@ Seeding happens once per session, not per message and not per restart: later tur
 How the plugin keeps track of that is a developer question:
 [CONCEPTS.md](CONCEPTS.md#thread-sessions-and-the-frontier).
 
+#### Room history on the first turn
+
+Every fresh channel message anchors a session of its own (the same anchoring that keeps thread sessions isolated), so a turn that is not inside a thread would otherwise start from nothing — even when the room has been talking for hours and even when the message replies to something the bot itself posted. Every such turn therefore carries what the room said before it, as an `[Earlier in this room]` block with the same shape and the same 20-message / 2000-character budget as the thread block, including messages the bot itself posted, which never enter a session any other way. Direct messages share one session per DM and never seed; messages inside a thread use the thread block instead.
+
+A message that is a reply to another one also names what it answers: the reply pointer `[Replying to "…"]` — `[Replying to your previous message: "…"]` when the answered message is the bot's own — is rendered ahead of the message text. An unreadable or deleted target costs the pointer, never the reply itself.
+
 ### Interactive Prompts
 
 The plugin replaces the default text-based clarify and exec-approval
@@ -429,7 +436,7 @@ Credentials never come from the model: no tool takes a server URL or a token as
 
 Long messages are split at the same limit as every other send (see [Message Splitting](#message-splitting)). If a chunk fails after earlier ones were delivered, the tool reports what it already sent plus a `partial: true` marker, so a retry can repeat only the missing part.
 
-A message the tool posts is the bot's own, so Chatto echoes it back and the adapter drops it as a self-event — it never becomes a turn of its own. In a thread, the next human message still reaches the agent with what it said: each turn seeds the thread with whatever landed since the last turn, so a `chatto_send_room` reply is in the session the next time someone speaks there, and a gateway restart in between does not change that.
+A message the tool posts is the bot's own, so Chatto echoes it back and the adapter drops it as a self-event — it never becomes a turn of its own. In a thread, the next human message still reaches the agent with what it said: each turn seeds the thread with whatever landed since the last turn, so a `chatto_send_room` reply is in the session the next time someone speaks there, and a gateway restart in between does not change that. In a room without a thread, the same holds through the room block: the next message's `[Earlier in this room]` fetch reaches back over the tool's messages too, because they are part of what the room said before that message.
 
 ## Troubleshooting
 

adapter.py

@@ -1247,7 +1247,9 @@ class ChattoAdapter(BasePlatformAdapter):
         room_id = str(getattr(source, "chat_id", "") or "")
         if not room_id:
             return None
-        by_room, _malformed = _parse_room_toolsets(self.chatto_config.room_toolsets.value)
+        by_room, _malformed = _parse_room_toolsets(
+            self.chatto_config.room_toolsets.value
+        )
         return by_room.get(room_id) or None
 
     async def _room_roster_context(
@@ -1703,6 +1705,29 @@ class ChattoAdapter(BasePlatformAdapter):
             return default
         return value if value is not None else default
 
+    async def _reply_target(
+        self, client: ChattoClient, room_id: str, in_reply_to: str
+    ) -> Message | None:
+        """The message a reply points at, or ``None`` when it cannot be read.
+
+        Best-effort on purpose: a deleted or unreadable target costs the
+        context line, not the turn — an unanswered reply is worse than one
+        answered without its quote.
+        """
+        try:
+            target = await client.get_message(room_id=room_id, event_id=in_reply_to)
+        except Exception:
+            logger.debug(
+                "Chatto: reply target %s/%s could not be read",
+                room_id,
+                in_reply_to,
+                exc_info=True,
+            )
+            return None
+        if target is None or target.deleted_at:
+            return None
+        return target
+
     def _is_own(self, message: Message) -> bool:
         """Whether this message is one the bot itself posted."""
         return self.me is not None and message.actor_id == self.me.id
@@ -1723,10 +1748,85 @@ class ChattoAdapter(BasePlatformAdapter):
     def _thread_session_exists(self, source: Any) -> bool:
         """Whether the gateway already has a session for this source.
 
-        The gateway's own routing index is the authority on "does this thread's
+        Kept as the thread path's own name: it answers the same question as
+        ``_session_exists`` and delegates to it, so the two paths cannot
+        disagree about what "a session exists" means.
+        """
+        return self._session_exists(source)
+
+    async def _room_context_for(
+        self,
+        client: ChattoClient,
+        *,
+        room_id: str,
+        trigger: Message,
+        source: Any,
+    ) -> str:
+        """The room's earlier messages, delivered when a room session starts.
+
+        A fresh room message roots its own session (see _admit_and_build), so
+        the first turn into a room is answered with nothing but that message —
+        even when the room has been talking for hours and even when the bot
+        itself posted there. The reported case: a human replies to the bot's
+        message without opening a thread, and the agent answers a conversation
+        it has never seen.
+
+        The block is bounded by the same caps as the thread seed, and it is
+        skipped as soon as the session holds a turn: the goal is to open a room
+        conversation with its recent context, not to re-deliver the room's
+        history on every message.
+
+        Seeding is anchored on the trigger, not on a per-room cursor: every
+        fresh room message opens a new session, so a cursor written for one
+        session's frontier would hand the next message's session a block
+        written for the first one's. The trigger-anchored fetch makes the
+        block "what the room said before this message", which is the question
+        a session start actually asks.
+        """
+        if self._session_exists(source):
+            return ""
+
+        try:
+            page, _anchor_index = await client.get_room_events_around(
+                room_id, trigger.id, limit=ChattoConstants.THREAD_CONTEXT_MESSAGE_LIMIT
+            )
+        except Exception:
+            logger.debug(
+                "Chatto: room context lookup failed for %s/%s",
+                room_id,
+                trigger.id,
+                exc_info=True,
+            )
+            return ""
+        messages = [ev.message for ev in page.events if ev.message is not None]
+        # The page carries its own user directory, which answers the author
+        # names in the block without a member-list call — the same hydration
+        # the read tools use for a fetched timeline.
+        for user_id, user in page.users_by_id.items():
+            self._user_cache.setdefault(user_id, user)
+        # ``around`` serves both directions of the trigger, and a session
+        # start is answered with the past, not the future: newer messages are
+        # not "earlier", and including one would hand the agent a message the
+        # room may still be typing around. Age is compared by time rather
+        # than by position, because the page's order is display order, not a
+        # promise about either side of the anchor.
+        trigger_time = trigger.created_at or datetime.min.replace(tzinfo=UTC)
+        earlier = [
+            m
+            for m in _oldest_first(messages)
+            if m.id != trigger.id
+            and not m.deleted_at
+            and (m.created_at or datetime.min.replace(tzinfo=UTC)) <= trigger_time
+        ]
+        return self._earlier_context_block(earlier, "[Earlier in this room]")
+
+    def _session_exists(self, source: Any) -> bool:
+        """Whether the gateway already has a session for this source.
+
+        The gateway's own routing index is the authority on "does this
         conversation already exist" — it is persistent (state.db), so unlike
-        the in-memory deque it answers correctly after a restart. Returns False
-        when no store is wired (tests, or a startup path before
+        the in-memory dedup state it answers correctly after a restart.
+        Returns False when no store is wired (tests, or a startup path before
         ``set_session_store`` runs) so we fail open and seed.
         """
         store = getattr(self, "_session_store", None)
@@ -1742,6 +1842,17 @@ class ChattoAdapter(BasePlatformAdapter):
         messages go first: the reply's immediate predecessors are what the
         turn is about.
         """
+        return self._earlier_context_block(messages, "[Earlier in this thread]")
+
+    def _earlier_context_block(self, messages: list[Message], header: str) -> str:
+        """Render earlier messages as one reference block under ``header``.
+
+        Shared by the thread and the room seed: both answer the same question
+        ("what did this conversation say before the trigger"), so both get the
+        same rendering, the same caps and the same own/verified labelling — two
+        copies of one renderer would drift until a thread and a room disagreed
+        about what a quoted message looks like.
+        """
         entries: list[str] = []
         for message in messages:
             entry = self._thread_message_entry(message)
@@ -1761,10 +1872,10 @@ class ChattoAdapter(BasePlatformAdapter):
             return ""
         body = "\n".join(entries)
         omitted = f"… {dropped} earlier message(s) omitted\n" if dropped else ""
-        return f"{omitted}[Earlier in this thread]\n{body}"
+        return f"{omitted}{header}\n{body}"
 
     def _thread_message_entry(self, message: Message) -> str:
-        """One rendered thread message, or ``""`` for one with no body.
+        """One rendered message for an earlier-context block, or ``""``.
 
         Attachment-only messages render empty: the block is text context, and
         inventing a placeholder for a picture the agent cannot see here would
@@ -2382,6 +2493,37 @@ class ChattoAdapter(BasePlatformAdapter):
             if thread_context:
                 context_blocks.append(thread_context)
 
+            # A fresh room message roots its own session (the anchoring above),
+            # so it starts empty even when the room talked for hours. The reply
+            # pointer above carries what this message answers; this block adds
+            # what the room said before it — including the bot's own output,
+            # whose echo is filtered as a self-event and reaches no session
+            # any other way.
+            if not thread_root_event_id:
+                room_context = await self._room_context_for(
+                    client,
+                    room_id=room_id,
+                    trigger=message,
+                    source=source,
+                )
+                if room_context:
+                    context_blocks.append(room_context)
+
+        # A reply must not read as a fresh conversation: the gateway renders
+        # ``reply_to_text`` into the turn as "[Replying to …]", but only when the
+        # adapter resolved it. Signal does this for quotes; a Chatto reply's
+        # target is one get_message away, so a session that starts here (every
+        # fresh room message roots its own — see the anchoring above) at least
+        # knows what the reply answers. DMs get the same treatment: a reply is
+        # the one pointer a session start can carry there too.
+        reply_to_text: str | None = None
+        reply_to_is_own = False
+        if message.in_reply_to:
+            target = await self._reply_target(client, room_id, message.in_reply_to)
+            if target is not None:
+                reply_to_text = target.body or ""
+                reply_to_is_own = self._is_own(target)
+
         # prepare a MessageEvent
         message_event = MessageEvent(
             text=message_body,
@@ -2390,6 +2532,8 @@ class ChattoAdapter(BasePlatformAdapter):
             timestamp=message.created_at or datetime.now(UTC),
             raw_message=message,
             reply_to_message_id=message.in_reply_to,
+            reply_to_text=reply_to_text,
+            reply_to_is_own_message=reply_to_is_own,
             channel_context="\n\n".join(context_blocks) or None,
         )
         if message_event.is_command():
@@ -4218,6 +4362,7 @@ def _persist_realtime_cursor(cfg: ChattoConfiguration, cursor: str | None) -> No
 # unknown-key check reads it as a likely typo or renamed field.
 _NEAR_MISS_RATIO = 0.75
 
+
 def _warn_on_unknown_extra_keys(config: PlatformConfig) -> None:
     """Warn about ``extra`` keys that look like misspelled config fields.
 
@@ -4281,9 +4426,7 @@ def hermes_validate_config(config: PlatformConfig) -> bool:
         )
         return False
 
-    _by_room, malformed_toolsets = _parse_room_toolsets(
-        resolved.room_toolsets.value
-    )
+    _by_room, malformed_toolsets = _parse_room_toolsets(resolved.room_toolsets.value)
     if malformed_toolsets:
         logger.error(
             "Chatto: 'room_toolsets' entries must look like 'room_id:toolset' — "
@@ -4668,9 +4811,7 @@ CHATTO_PIN_MESSAGE_SCHEMA = {
             },
             "pin": {
                 "type": "boolean",
-                "description": (
-                    "True to pin, false to unpin the message."
-                ),
+                "description": ("True to pin, false to unpin the message."),
             },
         },
         "required": ["event_id", "pin"],
@@ -4692,8 +4833,7 @@ CHATTO_SET_CUSTOM_STATUS_SCHEMA = {
             "emoji": {
                 "type": "string",
                 "description": (
-                    "A single emoji to show next to the status text "
-                    "(e.g. 🛠️, ⏳, 🔴)."
+                    "A single emoji to show next to the status text (e.g. 🛠️, ⏳, 🔴)."
                 ),
             },
             "text": {
@@ -4717,9 +4857,7 @@ CHATTO_SET_CUSTOM_STATUS_SCHEMA = {
 
 CHATTO_DELETE_CUSTOM_STATUS_SCHEMA = {
     "name": "chatto_delete_custom_status",
-    "description": (
-        "Clear the bot's custom status, reverting to no status display."
-    ),
+    "description": ("Clear the bot's custom status, reverting to no status display."),
     "parameters": {
         "type": "object",
         "properties": {},
@@ -4909,9 +5047,7 @@ CHATTO_MARK_THREAD_AS_READ_SCHEMA = {
             },
             "thread_root_event_id": {
                 "type": "string",
-                "description": (
-                    "The event ID of the thread root message."
-                ),
+                "description": ("The event ID of the thread root message."),
             },
         },
         "required": ["thread_root_event_id"],
@@ -5680,9 +5816,7 @@ def _schedule_ephemeral_deletion(
     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)
-    )
+    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)
 
@@ -6289,9 +6423,7 @@ async def _set_custom_status(args: dict) -> str:
                 "success": True,
                 "emoji": emoji,
                 "text": text,
-                "expires_at": (
-                    _iso_or_none(expires_at) if expires_at else None
-                ),
+                "expires_at": (_iso_or_none(expires_at) if expires_at else None),
             }
         )
 
@@ -6339,9 +6471,7 @@ async def _mark_thread_as_read(args: dict) -> str:
         if not room_id:
             resolved = await _resolve_room(client, room_id, room_name)
             room_id = resolved.room.id
-        prev_last_read = await client.mark_thread_as_read(
-            room_id, thread_root_event_id
-        )
+        prev_last_read = await client.mark_thread_as_read(room_id, thread_root_event_id)
         return json.dumps(
             {
                 "success": True,

test_adapter.py

@@ -708,9 +708,7 @@ class TestRoomToolsets:
         """A thread's source carries the room as ``chat_id`` and the thread
         separately, so the room's restriction has to reach its threads."""
         adapter = _make_adapter(room_toolsets=["room-1:web"])
-        assert adapter.toolsets_for_source(self._source(chat_type="thread")) == [
-            "web"
-        ]
+        assert adapter.toolsets_for_source(self._source(chat_type="thread")) == ["web"]
 
     def test_a_duplicate_toolset_is_listed_once(self):
         adapter = _make_adapter(room_toolsets=["room-1:web", "room-1:web"])
@@ -1421,9 +1419,7 @@ class TestRealtimeResume:
 
     async def test_reconnect_resumes_from_the_last_event_cursor(self):
         adapter = _make_adapter()
-        resume_cursors = await self._run_loop(
-            adapter, [[self._event("cur-1")], []]
-        )
+        resume_cursors = await self._run_loop(adapter, [[self._event("cur-1")], []])
         assert resume_cursors == [None, "cur-1"]
 
     async def test_an_unreplayable_event_keeps_the_previous_cursor(self):
@@ -1457,9 +1453,7 @@ class TestPersistedRealtimeCursor:
         _persist_realtime_cursor(self._cfg(), "cur-42")
         assert _restore_realtime_cursor(self._cfg()) == "cur-42"
 
-    def test_a_cursor_older_than_the_window_is_discarded(
-        self, monkeypatch, tmp_path
-    ):
+    def test_a_cursor_older_than_the_window_is_discarded(self, monkeypatch, tmp_path):
         """Replaying an outage of hours is worse than losing it: the window is
         what keeps a restart bounded."""
         from adapter import _profile_scope_key, _restore_realtime_cursor
@@ -2813,6 +2807,20 @@ def _serve_thread(adapter, messages, limit=None):
     )
 
 
+def _serve_room_timeline(adapter, messages):
+    """Point ``get_room_events_around`` at a room's timeline, oldest first.
+
+    Modelled on the server's own handler for ``GetRoomEventsAround``: it serves
+    the whole room timeline around the anchor, and the adapter's room seed
+    filters both ends (newer messages and the trigger itself) from it.
+    """
+
+    async def around(room_id, event_id, *, limit=None):
+        return TimelinePage(events=[_thread_event(m) for m in messages]), 0
+
+    adapter._chatto_client.get_room_events_around = AsyncMock(side_effect=around)
+
+
 class _FakeSessionStore:
     """An in-memory session store with the real store's shape.
 
@@ -2849,6 +2857,276 @@ class _FakeSessionStore:
         return True
 
 
+class TestRoomContext:
+    """A fresh room message roots its own session, so the first turn into a
+    room starts empty even when the room has been talking for hours.
+
+    The reported case: a human replies to the bot's message in a channel,
+    without opening a thread. The agent answers with no idea what the reply
+    refers to — the reply pointer names the message, but nothing carried what
+    that message said.
+    """
+
+    def _adapter(self):
+        adapter = _make_adapter()
+        adapter.chatto_config.allow_all_users.value = True
+        adapter.me = _make_user("bot-user-id", "hermes_bot")
+        adapter._user_cache["user-1"] = _roster_user("user-1", "alice")
+        # The roster rides on every channel turn; an AsyncMock without a
+        # roster shape would answer a list of IDs, not the (ids, Page) pair.
+        adapter._chatto_client.list_room_members = AsyncMock(
+            return_value=(["user-1"], Page())
+        )
+        adapter._chatto_client.batch_get_room_members = AsyncMock(
+            return_value=[DirectoryMember(user=_roster_user("user-1", "alice"))]
+        )
+        adapter.handle_message = AsyncMock()
+        return adapter
+
+    def _seed_channel(self, adapter, room_id="room-1"):
+        adapter.chatto_config.require_mention_rooms.value = [room_id]
+        adapter._room_kinds[room_id] = RoomKind.CHANNEL
+
+    async def _dispatch(
+        self,
+        adapter,
+        body,
+        *,
+        room_id="room-1",
+        message_id="msg-9",
+        created_at=None,
+        reply_to="",
+        reply_target=None,
+    ):
+        payload = _make_posted_payload(room_id=room_id, message_event_id=message_id)
+        message = _make_message(
+            body=body,
+            room_id=room_id,
+            message_id=message_id,
+            created_at=created_at,
+        )
+        message.in_reply_to = reply_to
+        if reply_target is not None:
+            # One get_message per call: the trigger first, then the reply
+            # target — the order the dispatch path asks in.
+            adapter._chatto_client.get_message = AsyncMock(
+                side_effect=[message, reply_target]
+            )
+        else:
+            _with_get_message(adapter, payload, message)
+        await adapter._dispatch_message_posted(payload, payload.message_event_id)
+
+    def _dispatched_event(self, adapter):
+        return adapter.handle_message.await_args.args[0]
+
+    async def test_a_room_reply_opens_with_the_room_behind_it(self):
+        """The reported bug: the first turn into a room must not be answered
+        as a conversation the agent has never seen."""
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        # A conversation the room had before this turn, ending in an answer
+        # the bot itself posted — whose echo never reaches a session.
+        past = _make_message(
+            body="Deploy finished at noon",
+            message_id="msg-1",
+            actor_id="bot-user-id",
+            created_at=datetime(2026, 10, 22, 12, 0, tzinfo=UTC),
+        )
+        trigger = _make_message(
+            body="@hermes_bot which build?",
+            message_id="msg-9",
+            created_at=datetime(2026, 10, 22, 12, 5, tzinfo=UTC),
+        )
+        _serve_room_timeline(adapter, [past, trigger])
+
+        await self._dispatch(adapter, trigger.body, created_at=trigger.created_at)
+
+        context = self._dispatched_event(adapter).channel_context or ""
+        assert "Deploy finished at noon" in context
+
+    async def test_the_block_names_the_room(self):
+        """The renderer is shared with the thread seed, so the header is the
+        one difference — a room block must say room, not thread."""
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        past = _make_message(
+            body="Deploy finished at noon",
+            message_id="msg-1",
+            created_at=datetime(2026, 10, 22, 12, 0, tzinfo=UTC),
+        )
+        trigger = _make_message(
+            body="@hermes_bot which build?",
+            message_id="msg-9",
+            created_at=datetime(2026, 10, 22, 12, 5, tzinfo=UTC),
+        )
+        _serve_room_timeline(adapter, [past, trigger])
+
+        await self._dispatch(adapter, trigger.body, created_at=trigger.created_at)
+
+        context = self._dispatched_event(adapter).channel_context or ""
+        assert "[Earlier in this room]" in context
+        assert "[Earlier in this thread]" not in context
+
+    async def test_a_message_newer_than_the_trigger_is_not_earlier(self):
+        """``around`` serves both directions; the block is the room's past,
+        not its future."""
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        trigger = _make_message(
+            body="@hermes_bot which build?",
+            message_id="msg-9",
+            created_at=datetime(2026, 10, 22, 12, 0, tzinfo=UTC),
+        )
+        newer = _make_message(
+            body="Typed while the agent was thinking",
+            message_id="msg-10",
+            created_at=datetime(2026, 10, 22, 12, 6, tzinfo=UTC),
+        )
+        _serve_room_timeline(adapter, [trigger, newer])
+
+        await self._dispatch(adapter, trigger.body, created_at=trigger.created_at)
+
+        context = self._dispatched_event(adapter).channel_context or ""
+        assert "Typed while the agent was thinking" not in context
+
+    async def test_an_edit_restart_does_not_reseed(self):
+        """The session guard protects the one path where the same message id
+        is dispatched twice: an edit that restarts a turn already running,
+        whose session now exists. The restart must not carry the room's
+        history a second time — the session holds the original turn."""
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        adapter.set_session_store(_FakeSessionStore())
+        store = adapter._session_store
+        past = _make_message(
+            body="Deploy finished at noon",
+            message_id="msg-1",
+            created_at=datetime(2026, 10, 22, 12, 0, tzinfo=UTC),
+        )
+        trigger = _make_message(
+            body="@hermes_bot which build?",
+            message_id="msg-9",
+            created_at=datetime(2026, 10, 22, 12, 5, tzinfo=UTC),
+        )
+        _serve_room_timeline(adapter, [past, trigger])
+
+        await self._dispatch(adapter, trigger.body, created_at=trigger.created_at)
+        assert "Deploy finished at noon" in (
+            self._dispatched_event(adapter).channel_context or ""
+        )
+
+        # The gateway has run the turn now, so the session exists — the same
+        # message dispatched again (what an edit-restart does) must not carry
+        # the room's past into a session that already holds it.
+        session_key = adapter._session_key_for(self._dispatched_event(adapter).source)
+        store.create_session(session_key)
+        await self._dispatch(adapter, trigger.body, created_at=trigger.created_at)
+
+        context = self._dispatched_event(adapter).channel_context or ""
+        assert "Deploy finished at noon" not in context
+
+    async def test_a_failed_lookup_leaves_the_turn_running(self):
+        """Best-effort, like the thread seed: the block is context, and its
+        absence must not cost the turn itself."""
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        adapter._chatto_client.get_room_events_around = AsyncMock(
+            side_effect=ChattoError("boom")
+        )
+
+        await self._dispatch(adapter, "@hermes_bot help")
+
+        assert adapter.handle_message.await_count == 1
+
+    async def test_a_reply_names_what_it_answers(self):
+        """The other half of the report: a reply to the bot's message must
+        reach the agent as a reply, not as a comment from nowhere. The
+        gateway renders the pointer from ``reply_to_text`` — unfilled, the
+        turn reads as a fresh conversation about a quote that is not there."""
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        target = _make_message(
+            body="Build 4711 is green, logs attached",
+            message_id="msg-1",
+            actor_id="bot-user-id",
+        )
+        trigger = _make_message(
+            body="@hermes_bot and the artifacts?",
+            message_id="msg-9",
+            created_at=datetime(2026, 10, 22, 12, 5, tzinfo=UTC),
+        )
+        _serve_room_timeline(adapter, [target, trigger])
+
+        await self._dispatch(
+            adapter,
+            trigger.body,
+            created_at=trigger.created_at,
+            reply_to="msg-1",
+            reply_target=target,
+        )
+
+        event = self._dispatched_event(adapter)
+        assert event.reply_to_message_id == "msg-1"
+        assert event.reply_to_text == "Build 4711 is green, logs attached"
+        assert event.reply_to_is_own_message is True
+
+    async def test_a_reply_to_someone_else_is_not_own(self):
+        """``reply_to_is_own_message`` is what makes the gateway say "your
+        previous message" — claiming a colleague's message as the bot's
+        would be worse than saying nothing."""
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        target = _make_message(
+            body="I pushed a fix for that",
+            message_id="msg-1",
+            actor_id="user-2",
+        )
+        trigger = _make_message(
+            body="@hermes_bot can you review?",
+            message_id="msg-9",
+            created_at=datetime(2026, 10, 22, 12, 5, tzinfo=UTC),
+        )
+        _serve_room_timeline(adapter, [target, trigger])
+
+        await self._dispatch(
+            adapter,
+            trigger.body,
+            created_at=trigger.created_at,
+            reply_to="msg-1",
+            reply_target=target,
+        )
+
+        event = self._dispatched_event(adapter)
+        assert event.reply_to_text == "I pushed a fix for that"
+        assert event.reply_to_is_own_message is False
+
+    async def test_an_unreadable_reply_target_costs_the_pointer_not_the_turn(self):
+        """Best-effort like the room block: a deleted or unreadable target
+        leaves the pointer empty, and the reply is still dispatched."""
+        adapter = self._adapter()
+        self._seed_channel(adapter)
+        _serve_room_timeline(adapter, [])
+        deleted = _make_message(
+            body="Build 4711 is green, logs attached",
+            message_id="msg-gone",
+            created_at=datetime(2026, 10, 22, 12, 0, tzinfo=UTC),
+        )
+        deleted.deleted_at = datetime(2026, 10, 22, 12, 2, tzinfo=UTC)
+
+        await self._dispatch(
+            adapter,
+            "@hermes_bot hi",
+            reply_to="msg-gone",
+            reply_target=deleted,
+        )
+
+        event = self._dispatched_event(adapter)
+        assert event.reply_to_message_id == "msg-gone"
+        assert event.reply_to_text is None
+        assert event.reply_to_is_own_message is False
+        assert adapter.handle_message.await_count == 1
+
+
 class TestThreadContext:
     """A thread session starts empty, so the first turn into a Chatto thread
     seeds the messages that came before it.
@@ -5992,8 +6270,12 @@ class TestPluginToolHandlers:
         client.list_pinned_messages = AsyncMock(
             return_value=PinnedMessagesPage(
                 pinned_messages=[
-                    PinnedMessage(message=_make_message(message_id="msg-1", actor_id="user-1")),
-                    PinnedMessage(message=_make_message(message_id="msg-2", actor_id="user-2")),
+                    PinnedMessage(
+                        message=_make_message(message_id="msg-1", actor_id="user-1")
+                    ),
+                    PinnedMessage(
+                        message=_make_message(message_id="msg-2", actor_id="user-2")
+                    ),
                 ],
                 page=Page(total_count=2, has_more=False),
             )
@@ -6010,7 +6292,10 @@ class TestPluginToolHandlers:
             client=client,
         )
         assert [p["event_id"] for p in result["pinned_messages"]] == ["msg-1", "msg-2"]
-        assert [p["author_login"] for p in result["pinned_messages"]] == ["alice", "bob"]
+        assert [p["author_login"] for p in result["pinned_messages"]] == [
+            "alice",
+            "bob",
+        ]
         assert client.batch_get_room_members.await_count == 1
 
     # -- chatto_pin_message --
@@ -6217,9 +6502,7 @@ class TestPluginToolHandlers:
                     name="Team",
                     description="Internal rooms",
                     rooms=[
-                        _make_room_state(
-                            _make_room("room-1", "Ops", RoomKind.CHANNEL)
-                        )
+                        _make_room_state(_make_room("room-1", "Ops", RoomKind.CHANNEL))
                     ],
                 )
             ]