feat(matrix): expose unread counts and scoped mark-read actions

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

@@ -24,6 +24,7 @@
 from typing import Any, Callable, Dict, List, NamedTuple, Optional, Tuple, TYPE_CHECKING, Union
 from urllib.parse import urlparse, parse_qs, urlunparse
 
+from agent.auxiliary_client_lifecycle import _close_cached_client
 from agent.error_classifier import (
     _BILLING_PATTERNS,
     _OVERLOADED_PATTERNS,
@@ -5899,73 +5900,6 @@ def neuter_async_httpx_del() -> None:
         pass  # Graceful degradation if the SDK changes its internals
 
 
-def _force_close_async_httpx(client: Any) -> None:
-    """Mark the httpx AsyncClient inside an AsyncOpenAI client as closed so ``__del__`` won't
-    schedule ``aclose()`` on a dead loop. Skips the full async close — the OS drops connections."""
-    with contextlib.suppress(Exception):
-        from httpx._client import ClientState
-        inner = getattr(client, "_client", None)
-        if inner is not None and not getattr(inner, "is_closed", True):
-            inner._state = ClientState.CLOSED
-
-
-def _schedule_async_close(close_result: Any, client: Any) -> None:
-    """Finish an async close without leaking an unawaited coroutine."""
-    async def _await_close() -> None:
-        try:
-            await close_result
-        except Exception:
-            pass
-        finally:
-            _force_close_async_httpx(client)
-    runner = _await_close()
-    try:
-        import asyncio as _aio
-        try:
-            loop = _aio.get_running_loop()
-        except RuntimeError:
-            _aio.run(runner)
-        else:
-            task = loop.create_task(runner)
-
-            def _consume(completed_task) -> None:
-                with contextlib.suppress(BaseException):
-                    completed_task.exception()
-            task.add_done_callback(_consume)
-            runner = None
-    except Exception:
-        if runner is not None:
-            with contextlib.suppress(Exception):
-                runner.close()
-        _force_close_async_httpx(client)
-
-
-def _close_cached_client(client: Any, *, close_async: bool = False) -> None:
-    """Close one cached client, awaiting async transports only when safe."""
-    if client is None:
-        return
-    close_fn = getattr(client, "close", None)
-    if not callable(close_fn):
-        _force_close_async_httpx(client)
-        return
-    try:
-        close_result = close_fn()
-    except Exception:
-        _force_close_async_httpx(client)
-        return
-    if inspect.isawaitable(close_result):
-        if close_async:
-            _schedule_async_close(close_result, client)
-        else:
-            # Never await a client owned by another live loop; close the coroutine (no
-            # unawaited warning) and neuter the transport.
-            with contextlib.suppress(Exception):
-                close_result.close()
-            _force_close_async_httpx(client)
-        return
-    _force_close_async_httpx(client)
-
-
 def shutdown_cached_clients() -> None:
     """Close all cached clients; call at CLI shutdown *before* the loop closes.
 

agent/auxiliary_client_lifecycle.py

@@ -0,0 +1,103 @@
+"""Teardown of cached auxiliary-client transports."""
+
+import contextlib
+import inspect
+from typing import Any, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from asyncio import Task
+
+
+def _force_close_async_httpx(client: Any) -> None:
+    """Mark the httpx AsyncClient inside an AsyncOpenAI client as closed so ``__del__`` won't
+    schedule ``aclose()`` on a dead loop. Skips the full async close — the OS drops connections."""
+    with contextlib.suppress(Exception):
+        from httpx._client import ClientState
+
+        inner = getattr(client, "_client", None)
+        if inner is not None and not getattr(inner, "is_closed", True):
+            inner._state = ClientState.CLOSED
+
+
+def _schedule_async_close(close_result: Any, client: Any) -> Optional["Task[None]"]:
+    """Finish an async close without leaking an unawaited coroutine."""
+
+    async def _await_close() -> None:
+        try:
+            await close_result
+        except Exception:
+            pass
+        finally:
+            _force_close_async_httpx(client)
+
+    runner = _await_close()
+    try:
+        import asyncio as _aio
+
+        try:
+            loop = _aio.get_running_loop()
+        except RuntimeError:
+            _aio.run(runner)
+        else:
+            task = loop.create_task(runner)
+
+            def _consume(completed_task) -> None:
+                with contextlib.suppress(BaseException):
+                    completed_task.exception()
+
+            task.add_done_callback(_consume)
+            runner = None
+            return task
+    except Exception:
+        if runner is not None:
+            with contextlib.suppress(Exception):
+                runner.close()
+        _force_close_async_httpx(client)
+
+
+def _close_cached_client(
+    client: Any, *, close_async: bool = False
+) -> Optional["Task[None]"]:
+    """Close one cached client, awaiting async transports only when safe."""
+    if client is None:
+        return
+    close_fn = getattr(client, "close", None)
+    if not callable(close_fn):
+        _force_close_async_httpx(client)
+        return
+    try:
+        close_result = close_fn()
+    except Exception:
+        _force_close_async_httpx(client)
+        return
+    if inspect.isawaitable(close_result):
+        if close_async:
+            return _schedule_async_close(close_result, client)
+        else:
+            # Never await a client owned by another live loop; close the coroutine (no
+            # unawaited warning) and neuter the transport.
+            with contextlib.suppress(Exception):
+                close_result.close()
+            _force_close_async_httpx(client)
+        return
+    _force_close_async_httpx(client)
+
+
+async def shutdown_cached_clients_for_current_loop() -> None:
+    """Close cached transports owned by the current event loop."""
+    import asyncio
+    from agent import auxiliary_client
+
+    loop = asyncio.get_running_loop()
+    with auxiliary_client._client_cache_lock:
+        keys = [
+            key
+            for key, entry in auxiliary_client._client_cache.items()
+            if entry[2] is loop
+        ]
+        clients = [auxiliary_client._client_cache.pop(key)[0] for key in keys]
+
+    tasks = [_close_cached_client(client, close_async=True) for client in clients]
+    pending = [task for task in tasks if task is not None]
+    if pending:
+        await asyncio.gather(*pending)

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,
+    *,
+    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))

apps/desktop/src/lib/desktop-toolsets.ts

@@ -13,6 +13,7 @@ const DESKTOP_HIDDEN_TOOLSETS = new Set([
   // adapter; `hermes tools` restricts these off the CLI too.
   'discord',
   'discord_admin',
+  'matrix_admin',
   'yuanbao',
   // Internal plumbing, not a user capability toggle.
   'context_engine',