fix(matrix): preserve genuine thread replies and fallback chains

.github/workflows/matrix-live.yml

@@ -0,0 +1,61 @@
+name: Matrix live client
+
+on:
+  pull_request:
+    paths:
+      - '.github/workflows/matrix-live.yml'
+      - '.github/actions/setup-pm/**'
+      - '.dockerignore'
+      - '*.py'
+      - 'agent/**'
+      - 'cron/**'
+      - 'gateway/**'
+      - 'hermes_cli/**'
+      - 'hermes_platform/**'
+      - 'locales/**'
+      - 'plugins/**'
+      - 'pm/**'
+      - 'providers/**'
+      - 'scripts/bundles/payload.py'
+      - 'scripts/ci/list_os_marked_tests.py'
+      - 'scripts/ci/setup_toolchain.py'
+      - 'scripts/run_tests.sh'
+      - 'scripts/run_tests_parallel.py'
+      - 'skills/**'
+      - 'tests/conftest.py'
+      - 'tests/fakes/fake_llm_provider.py'
+      - 'tests/integration/matrix_live/**'
+      - 'tools/**'
+      - 'tui_gateway/**'
+      - 'pyproject.toml'
+      - 'uv.lock'
+  push:
+    branches: [main]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  plain-exchange:
+    name: Separate client and Linux gateway
+    runs-on: ubuntu-latest-32-core
+    timeout-minutes: 45
+    env:
+      HERMES_TEST_MATRIX_GATEWAY_IMAGE: hermes-matrix-live:ci
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      - name: Set up locked Python and Matrix dependencies
+        uses: ./.github/actions/setup-pm
+        with:
+          extras: '["matrix"]'
+          test-environment: 'true'
+      # Build here so that the per-file test timeout does not have to cover a cold image build.
+      - name: Build the Linux gateway image
+        run: '"$HERMES_PYTHON" -m tests.integration.matrix_live.image_build "$HERMES_TEST_MATRIX_GATEWAY_IMAGE"'
+      - name: Run Matrix live client tests
+        run: scripts/run_tests.sh --include-integration tests/integration/matrix_live/
+        env:
+          HERMES_TEST_WORKERS: '1'
+          HERMES_TEST_FILE_TIMEOUT: '300'
+          HERMES_TEST_FILE_RETRIES: '0'

agent/context_compressor.py

@@ -35,6 +35,7 @@
 from agent.context_compressor_prellm import PreLlmSkipMixin
 from agent.context_compressor_summary import SummaryDispatchMixin
 from agent.error_classifier import FailoverReason, classify_api_error
+from agent.message_metadata import keep_newest_channel_state
 from agent.micro_compaction import MicroCompactionMixin
 from agent.prompt_builder import STEER_DISPLAY_KIND
 from agent.model_metadata import (
@@ -5745,14 +5746,16 @@ def _assemble_compressed(
         summary_role, merge_into_tail, force_user_leading, first_tail_visible_idx = (
             self._summary_placement(compressed, tail_messages, compress_start)
         )
+        summary_row = None
         if not merge_into_tail:
             # End marker stops weak models treating the quoted summary as fresh input (#11475) or
             # regurgitating it (#33256).
-            compressed.append({
+            summary_row = {
                 "role": summary_role, "content": summary + "\n\n" + _SUMMARY_END_MARKER,
                 COMPRESSED_SUMMARY_METADATA_KEY: True,
                 COMPRESSED_SUMMARY_HAS_USER_TURN_KEY: bool(self._summary_has_user_turn),
-            })
+            }
+            compressed.append(summary_row)
         # Default carrier is tail[0]: an exempt row absorbs the summary invisibly. The forced repair
         # path needs a non-empty role=user row, so it targets the template-visible row.
         merge_target_idx = first_tail_visible_idx if force_user_leading and first_tail_visible_idx is not None else 0
@@ -5763,7 +5766,10 @@ def _assemble_compressed(
                 msg[_COMPACTION_TAIL_MARKER] = True
             if merge_into_tail and tail_idx == merge_target_idx:
                 self._merge_summary_into_tail_row(msg, summary, summary_role, force_user_leading)
+                summary_row = msg
             compressed.append(msg)
+        if summary_row is not None:
+            keep_newest_channel_state(messages, compressed, summary_row)
         return compressed
 
 

agent/message_metadata.py

@@ -113,6 +113,35 @@ def message_identity(msg: MutableMapping[str, Any], *, with_tool_uids: bool = Fa
     return identity
 
 
+# The ``display_metadata`` entry in which the gateway saves a snapshot of the chat (its name, topic and
+# similar state) on a user row. The newest row with one is the snapshot that the conversation has
+# acknowledged, so a rewrite that removes that row must record the snapshot on a row that it keeps.
+CHANNEL_STATE_METADATA_KEY = "channel_state"
+
+
+def _channel_state(msg: Any) -> Optional[dict]:
+    metadata = msg.get("display_metadata") if isinstance(msg, Mapping) else None
+    state = metadata.get(CHANNEL_STATE_METADATA_KEY) if isinstance(metadata, Mapping) else None
+    return state if isinstance(state, dict) else None
+
+
+def newest_channel_state(messages: List[Any]) -> Optional[dict]:
+    """The channel state saved on the newest row of *messages* that has one."""
+    return next((state for msg in reversed(messages) if (state := _channel_state(msg)) is not None), None)
+
+
+def keep_newest_channel_state(
+    original: List[Any], rewritten: List[Any], carrier: MutableMapping[str, Any],
+) -> None:
+    """Record the newest channel state of *original* on *carrier* when *rewritten* no longer has it as
+    its newest. *carrier* is the row of *rewritten* that replaces the removed rows (a compaction summary
+    or a merge survivor), and every row after it in *rewritten* must be a row of *original*."""
+    state = newest_channel_state(original)
+    if state is None or newest_channel_state(rewritten) == state:
+        return
+    carrier["display_metadata"] = {**(carrier.get("display_metadata") or {}), CHANNEL_STATE_METADATA_KEY: state}
+
+
 def record_absorbed_message(
     survivor: MutableMapping[str, Any], dropped: Mapping[str, Any], *, dropped_leads: bool = False,
 ) -> None:
@@ -123,8 +152,9 @@ def record_absorbed_message(
     text leads; with *dropped_leads* the dropped dict's text was put first (the real user anchor folded
     into a scaffolding turn), so its uid becomes the survivor's and the survivor's former uid is recorded.
     A dict without a uid (unflushed, scaffolding, engine-authored) contributes nothing; an empty result
-    leaves the survivor untouched.
+    leaves the survivor untouched. The survivor also keeps the newest channel state of the two constituents.
     """
+    keep_newest_channel_state([dropped, survivor] if dropped_leads else [survivor, dropped], [survivor], survivor)
     survivor_uid = message_uid_or_none(survivor)
     dropped_uid = message_uid_or_none(dropped)
     own = uid_list(survivor.get(ABSORBED_MESSAGE_UIDS))

gateway/platforms/base.py

@@ -428,7 +428,7 @@ def is_host_excluded_by_no_proxy(hostname: str, no_proxy_value: str | None = Non
 from gateway.platforms.base_exec_approval import (
     approval_timeout_seconds, ea_action_labels, ea_default_reason_text, ea_header_text,
     ea_reason_label_text, ea_smart_deny_line_text, format_approval_deadline_line)
-from gateway.platforms.event import MessageEvent, MessageType, ProcessingOutcome
+from gateway.platforms.event import MessageEvent, MessageType, ProcessingOutcome, TurnContextUpdate
 from gateway.warning_notifications import diagnostic_wake_muted
 from hermes_cli.observability.shared_metrics_gateway import records_delivery, stop_reply_clock
 from gateway.session import SessionSource, build_session_key
@@ -1829,6 +1829,8 @@ def _padded_inline_flags(msg: MessageEvent) -> list[Optional[bool]]:
                 existing.media_text_inlined.extend(incoming_inline_flags)
             if event.text:
                 existing.text = BasePlatformAdapter._merge_caption(existing.text, event.text)
+            existing.absorb_message_ids(event)
+            existing.absorb_reply_context(event)
             existing.absorb_reply_expected(event)
             if existing_is_photo or incoming_is_photo:
                 existing.message_type = MessageType.PHOTO
@@ -1844,6 +1846,8 @@ def _padded_inline_flags(msg: MessageEvent) -> list[Optional[bool]]:
         if merge_text and both_text:
             if event.text:
                 existing.text = _append_text(existing.text, event.text)
+            existing.absorb_message_ids(event)
+            existing.absorb_reply_context(event)
             existing.absorb_reply_expected(event)
             return
     pending_messages[session_key] = event
@@ -2584,6 +2588,8 @@ def _enqueue_text_event(self, event: "MessageEvent") -> None:
             if event.media_urls:
                 existing.media_urls.extend(event.media_urls)
                 existing.media_types.extend(event.media_types)
+            existing.absorb_message_ids(event)
+            existing.absorb_reply_context(event)
             existing.absorb_reply_expected(event)
         existing._last_chunk_len = len(event.text or "")  # type: ignore[attr-defined]
         prior_task = self._pending_text_batch_tasks.get(key)
@@ -3528,6 +3534,25 @@ def pop_post_delivery_callback(
     _OK_EMOJI: Optional[str] = None
     _FAIL_EMOJI: Optional[str] = None
 
+    async def prepare_turn_context(
+        self, event: MessageEvent, *, origin: Optional[SessionSource],
+        acknowledged_state: Optional[Dict[str, Any]], first_turn: bool,
+    ) -> Optional[TurnContextUpdate]:
+        """Report context for this turn: changes to the chat since the conversation last
+        acknowledged its state, and earlier messages that a new session has not seen.
+
+        The gateway calls this while it prepares every inbound turn. ``origin`` is the session's
+        origin source, or ``None`` before the session exists. ``acknowledged_state`` is the
+        ``channel_state`` saved with the most recent transcript row that has one.
+        ``first_turn`` is true when the session transcript is empty. Return ``None`` to add no note
+        and leave the saved state unchanged.
+
+        For an adapter that overrides this hook, the session-context prompt keeps the chat name,
+        topic and user name from the session origin, so a rename does not rewrite the system prompt
+        of a running conversation. An override must therefore report name and topic changes in its
+        note."""
+        return None
+
     async def on_processing_start(self, event: MessageEvent) -> None:
         """Hook called when background processing begins."""
 
@@ -3823,6 +3848,13 @@ def _is_queue_text_debounce_candidate(self, event: MessageEvent) -> bool:
         return result
 
     def _can_merge_text_debounce_events(self, existing: MessageEvent, event: MessageEvent) -> bool:
+        """Return True when two text debounce events came from the same sender and do not reply
+        to different messages."""
+        return (self._same_text_debounce_sender(existing, event)
+                and not existing.reply_context_conflicts(event))
+
+    @staticmethod
+    def _same_text_debounce_sender(existing: MessageEvent, event: MessageEvent) -> bool:
         """Return True when two text debounce events came from the same sender."""
 
         def _identity(candidate: MessageEvent) -> tuple[str, ...] | None:
@@ -3861,21 +3893,39 @@ async def _queue_text_debounce(self, session_key: str, event: MessageEvent) -> N
                 existing_pending = self._pending_messages.get(session_key)
                 if existing_pending is not None and self._can_merge_text_debounce_events(existing_pending, event):
                     merge_pending_message_event(self._pending_messages, session_key, event, merge_text=True)
-                return
+                    return
+                if not self._same_text_debounce_sender(state.event, event):
+                    return
+                logger.debug("[%s] Busy text for %s replies to a third message; merging it into the "
+                             "debounce buffer, which keeps its own reply context", self.name, session_key)
         now = time.monotonic()
         if state is None:
             state = TextDebounceState(event=event, task=None, first_ts=now, last_ts=now)
             store[session_key] = state
         else:
             if event.text:
                 state.event.text = _append_text(state.event.text, event.text)
+            if event.media_urls:
+                state.event.media_text_inlined.extend(
+                    [None] * (len(state.event.media_urls) - len(state.event.media_text_inlined))
+                )
+                state.event.media_urls.extend(event.media_urls)
+                state.event.media_types.extend(event.media_types)
+                state.event.media_text_inlined.extend(event.media_text_inlined)
+                state.event.media_text_inlined.extend(
+                    [None] * (len(state.event.media_urls) - len(state.event.media_text_inlined))
+                )
+            state.event.absorb_reply_context(event)
             state.event.absorb_reply_expected(event)
             latest_message_id = getattr(event, "message_id", None)
-            latest_anchor = latest_message_id or getattr(event, "reply_to_message_id", None)
             if latest_message_id is not None:
+                state.event.merged_message_ids.extend(
+                    message_id for message_id in (state.event.message_id, *event.merged_message_ids)
+                    if message_id
+                )
                 state.event.message_id = str(latest_message_id)
-            if latest_anchor is not None and hasattr(state.event, "reply_to_message_id"):
-                state.event.reply_to_message_id = str(latest_anchor)
+            else:
+                state.event.absorb_message_ids(event)
             state.last_ts = now
         state.cancel_timer()
         delay = self._text_debounce_delay(session_key)

gateway/platforms/event.py

@@ -41,6 +41,18 @@ class ProcessingOutcome(Enum):
     CANCELLED = "cancelled"
 
 
+@dataclass(frozen=True)
+class TurnContextUpdate:
+    """What ``BasePlatformAdapter.prepare_turn_context`` reports for one turn.
+
+    ``note`` is prepended to the user message. ``channel_state`` is saved with the user transcript
+    row, so the change is acknowledged only when the turn that reported it is saved. It is ``None``
+    when the adapter could not read the chat state, and the saved state then stays unchanged.
+    """
+    note: Optional[str]
+    channel_state: Optional[Dict[str, Any]]
+
+
 @dataclass
 class MessageEvent:
     """Incoming message from a platform — the normalized shape all adapters produce."""
@@ -101,6 +113,14 @@ class MessageEvent:
     # knows the message was meant for someone else); None means unknown and keeps the visible
     # fallback, like True.
     reply_expected: Optional[bool] = None
+    # Snapshot from ``BasePlatformAdapter.prepare_turn_context``. The user transcript row saves it,
+    # and the saved snapshot is the baseline for the adapter's next comparison.
+    channel_state: Optional[Dict[str, Any]] = None
+    # Whether the quoted author passed the adapter's authorisation check; None when the adapter
+    # did not check. The reply pointer identifies the author only when this is set.
+    reply_to_author_authorized: Optional[bool] = None
+    # IDs of other events represented in this turn.
+    merged_message_ids: List[str] = field(default_factory=list)
 
     # Process-local admission receipt, never routing metadata or execution acknowledgement.
     _gateway_accepted: bool = field(default=False, init=False, repr=False, compare=False)
@@ -112,6 +132,11 @@ def absorb_reply_expected(self, other: "MessageEvent") -> None:
         if self.reply_expected is not True and other.reply_expected is not False:
             self.reply_expected = other.reply_expected
 
+    def absorb_message_ids(self, other: "MessageEvent") -> None:
+        self.merged_message_ids.extend(
+            message_id for message_id in (other.message_id, *other.merged_message_ids) if message_id
+        )
+
     def _command_text(self) -> str:
         """Return the message text with leading Desktop attachment refs stripped.
 
@@ -122,6 +147,27 @@ def _command_text(self) -> str:
         """
         return _ATTACHMENT_REF_RE.sub("", (self.text or "").lstrip()).lstrip()
 
+    def _replies_to_message(self) -> bool:
+        return bool(self.reply_to_message_id or self.reply_to_text)
+
+    def _reply_context(self) -> tuple:
+        return (self.reply_to_message_id, self.reply_to_text, self.reply_to_author_id,
+                self.reply_to_author_name, bool(self.reply_to_is_own_message), self.reply_to_author_authorized)
+
+    def reply_context_conflicts(self, other: "MessageEvent") -> bool:
+        """True when both events reply to a message and their reply contexts differ. A merged
+        event has room for only one reply context."""
+        return (self._replies_to_message() and other._replies_to_message()
+                and self._reply_context() != other._reply_context())
+
+    def absorb_reply_context(self, other: "MessageEvent") -> None:
+        """One turn now answers *other* too: take its reply context if this event has none."""
+        if self._replies_to_message() or not other._replies_to_message():
+            return
+        (self.reply_to_message_id, self.reply_to_text, self.reply_to_author_id,
+         self.reply_to_author_name, self.reply_to_is_own_message,
+         self.reply_to_author_authorized) = other._reply_context()
+
     def is_command(self) -> bool:
         """Check if this is a command message (e.g., /new, /reset)."""
         return self.allow_gateway_control and self._command_text().startswith("/")

gateway/run_agent_cache.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import importlib
 import logging
 import threading
@@ -14,7 +15,8 @@
 
 from agent.interrupt_compat import _accepts_keyword
 from gateway.config import Platform
-from gateway.session import SessionSource, build_session_context_prompt
+from gateway.run_inbound_turn_context import reports_turn_context
+from gateway.session import SessionSource, build_session_context_prompt, with_chat_metadata_from
 from gateway.session_prompt_pin import PROMPT_PIN_VERSION, sanitize_prompt_pin
 from gateway.run_shutdown import _log_suppressed
 from hermes_cli.config import DEFAULT_CONFIG, cfg_get
@@ -703,6 +705,16 @@ async def _persist_prompt_pins(self, session_key: Optional[str], expected_sessio
             # Durability protects cache continuity; a store outage must not block the user turn.
             logger.debug("Failed to persist prompt pin for %s", session_key, exc_info=True)
 
+    def _prompt_session_context(self: "GatewayRunner", context, session_entry):
+        """*context* for the session-context prompt. When the receiving adapter reports chat changes
+        through ``prepare_turn_context``, the prompt keeps the names and topic from the session
+        origin, so a rename does not rewrite the system prompt. Tools read *context* itself and see
+        the current names."""
+        origin = session_entry.origin if session_entry else None
+        if origin is None or not reports_turn_context(self._intake_adapter_for(context.source)):
+            return context
+        return dataclasses.replace(context, source=with_chat_metadata_from(context.source, origin))
+
     def _pinned_session_context_prompt(
         self, context, redact_pii: bool, session_key: Optional[str], *, internal: bool = False,
     ) -> str:

gateway/run_busy.py

@@ -1053,6 +1053,7 @@ async def _busy_queue_command(self, event: MessageEvent, quick_key: str, source)
                 reply_to_message_id=event.reply_to_message_id, reply_to_text=event.reply_to_text,
                 reply_to_author_id=event.reply_to_author_id,
                 reply_to_author_name=event.reply_to_author_name,
+                reply_to_author_authorized=event.reply_to_author_authorized,
                 reply_to_is_own_message=event.reply_to_is_own_message, auto_skill=event.auto_skill,
                 channel_prompt=event.channel_prompt, channel_context=event.channel_context,
                 internal=event.internal, timestamp=event.timestamp,