feat(matrix): create threads through the live session

.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/image_routing.py

@@ -534,9 +534,12 @@ def build_native_content_parts(
     user_text: str,
     image_paths: list[str],
     image_urls: Optional[list[str]] = None,
-) -> tuple[list[dict[str, Any]], list[str]]:
+    *,
+    revalidate: Optional[Callable[[list[str]], tuple[str, list[str]]]] = None,
+) -> tuple[List[dict[str, Any]], list[str]]:
     """Build an OpenAI-style ``content`` list for a user turn.
 
+    ``revalidate`` refreshes text and selects paths after local file conversion.
     Local paths become base64 ``data:`` URLs; remote URLs pass through verbatim.
     When any image attaches, one text part combines the caption (or a neutral
     default) with a ``[Image attached at: <path>]`` / ``[Image attached: <url>]``
@@ -545,20 +548,33 @@ def build_native_content_parts(
     ``skipped`` holds unreadable local paths (URLs are never skipped).
     """
     skipped: list[str] = []
+    converted_paths: list[str] = []
     attached: list[tuple[str, str]] = []  # (url, hint)
     for raw_path in image_paths:
         p = Path(raw_path)
         data_url = _file_to_data_url(p) if p.exists() and p.is_file() else None
         if data_url:
+            converted_paths.append(raw_path)
             attached.append((data_url, f"[Image attached at: {raw_path}]"))
         else:
             skipped.append(str(raw_path))
-    attached += [(u, f"[Image attached: {u}]") for u in ((u or "").strip() for u in image_urls or []) if u]
+    if revalidate is not None:
+        user_text, retained = revalidate(converted_paths)
+        attached = [
+            part for path, part in zip(converted_paths, attached) if path in retained
+        ]
+    attached += [
+        (u, f"[Image attached: {u}]")
+        for u in ((u or "").strip() for u in image_urls or [])
+        if u
+    ]
 
     text = (user_text or "").strip()
     if not attached:
         return ([{"type": "text", "text": text}] if text else []), skipped
-    combined_text = f"{text or 'What do you see in this image?'}\n\n" + "\n".join(h for _, h in attached)
+    combined_text = f"{text or 'What do you see in this image?'}\n\n" + "\n".join(
+        h for _, h in attached
+    )
     image_parts = [{"type": "image_url", "image_url": {"url": u}} for u, _ in attached]
     return [{"type": "text", "text": combined_text}, *image_parts], skipped
 

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/inbound_context.py

@@ -0,0 +1,122 @@
+"""Refreshable external context for a new gateway input."""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass, field
+from typing import Any, Callable, Protocol, TypeVar
+
+from gateway.platforms.event import MessageEvent, TurnContextUpdate
+
+_T = TypeVar("_T")
+
+_LOOP_CALL_TIMEOUT_SECONDS = 10.0
+
+
+class InboundContextSnapshot(Protocol):
+    def use_turn_context(self, update: TurnContextUpdate | None) -> None: ...
+
+    async def refresh(self) -> None: ...
+
+    def prepend_turn_context(self, text: str) -> str: ...
+
+    def reply_event(self, event: MessageEvent) -> MessageEvent: ...
+
+    def reply_image_paths(self) -> list[str]: ...
+
+
+@dataclass(frozen=True)
+class QuotedImageEnrichment:
+    path: str
+    text: str
+
+
+@dataclass
+class PreparedInboundMessage:
+    """A new input whose external context is rendered again before model use.
+
+    The snapshot reads platform state that belongs to the event loop that
+    prepared the input. Calls from any other thread run on that loop.
+    """
+
+    snapshot: InboundContextSnapshot
+    event: MessageEvent
+    text: str
+    redact_pii: bool = False
+    quoted_images: tuple[QuotedImageEnrichment, ...] = ()
+    message_text: str | None = None
+    persist_user_message: str | None = None
+    persist_user_timestamp: float | None = None
+    loop: asyncio.AbstractEventLoop = field(
+        default_factory=asyncio.get_running_loop, repr=False, compare=False
+    )
+
+    def retained_image_paths(self, paths: list[str]) -> list[str]:
+        return self._on_loop(self._retained_image_paths, paths)
+
+    def revalidate_native_input(
+        self, runner: Any, message: str, paths: list[str]
+    ) -> tuple[str, list[str]]:
+        return self._on_loop(self._revalidate_native_input, runner, message, paths)
+
+    def render(self, runner: Any, *, timestamps: bool = False) -> str:
+        return self._on_loop(self._render, runner, timestamps)
+
+    def _on_loop(self, call: Callable[..., _T], *args: Any) -> _T:
+        try:
+            running = asyncio.get_running_loop()
+        except RuntimeError:
+            running = None
+        if running is self.loop or not self.loop.is_running():
+            return call(*args)
+
+        async def run() -> _T:
+            return call(*args)
+
+        future = asyncio.run_coroutine_threadsafe(run(), self.loop)
+        try:
+            return future.result(timeout=_LOOP_CALL_TIMEOUT_SECONDS)
+        except TimeoutError:
+            future.cancel()
+            raise
+
+    def _retained_image_paths(self, paths: list[str]) -> list[str]:
+        current = self.snapshot.reply_image_paths()
+        authored = self.event.authored_media().media_urls
+        quoted = {image.path for image in self.quoted_images}
+        return [
+            path
+            for path in paths
+            if path not in quoted or path in current or path in authored
+        ]
+
+    def _revalidate_native_input(
+        self, runner: Any, message: str, paths: list[str]
+    ) -> tuple[str, list[str]]:
+        previous = self.message_text
+        current = self._render(runner, True)
+        if previous is not None and previous in message:
+            current = message.replace(previous, current, 1)
+        return current, self._retained_image_paths(paths)
+
+    def _render(self, runner: Any, timestamps: bool) -> str:
+        text = self.text
+        current = self.snapshot.reply_image_paths()
+        descriptions = [
+            image.text
+            for image in self.quoted_images
+            if image.path in current and image.text
+        ]
+        if descriptions:
+            text = "\n\n".join([*descriptions, text])
+        reply = self.snapshot.reply_event(self.event)
+        text = runner._prepend_inbound_reply_context(
+            reply, self.event.source, text, redact_pii=self.redact_pii,
+        )
+        text = self.snapshot.prepend_turn_context(text)
+        if timestamps:
+            text, self.persist_user_message, self.persist_user_timestamp = (
+                runner._hmwa_apply_message_timestamp(self.event, text)
+            )
+        self.message_text = text
+        return text

gateway/platform_registry.py

@@ -83,6 +83,8 @@ class PlatformEntry:
     allow_all_env: str = ""  # truthy "allow everyone" switch
     max_message_length: int = 0  # smart-chunking cap; 0 = no limit
     pii_safe: bool = False  # session descriptions redact PII (phone numbers, etc.)
+    # The adapter reads ``metadata["non_conversational"]``, so the gateway sets it on status sends.
+    reads_non_conversational_mark: bool = False
     emoji: str = "🔌"  # CLI/gateway display
     allow_update_command: bool = True  # /update may be issued from this platform
     platform_hint: str = ""  # injected into the system prompt; empty = none