feat: consolidated upstream PRs (session TTL, registry, matrix improvements, telemetry)

agent/restart_awareness.py

@@ -0,0 +1,120 @@
+"""
+Restart awareness: persistent activity tracker for gateway restarts.
+
+Writes and reads a JSON activity file so the agent can recover context
+after an unplanned or intentional gateway restart.
+"""
+
+import json
+import os
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from hermes_constants import get_hermes_home
+
+
+def _activity_path() -> Path:
+    return get_hermes_home() / "state" / "current_activity.json"
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def update_activity(
+    current_task: str,
+    files_modified: Optional[List[str]] = None,
+    last_action: Optional[str] = None,
+    next_expected_step: Optional[str] = None,
+    mode: str = "simple",
+) -> None:
+    """Write current activity state to disk."""
+    path = _activity_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    data = {
+        "current_task": current_task,
+        "files_modified": files_modified or [],
+        "last_action": last_action or "",
+        "next_expected_step": next_expected_step or "",
+        "mode": mode,
+        "updated_at": _now(),
+    }
+    path.write_text(json.dumps(data, indent=2))
+
+
+def read_activity() -> Optional[Dict]:
+    """Read the last known activity state, or None if none exists."""
+    path = _activity_path()
+    if not path.exists():
+        return None
+    try:
+        return json.loads(path.read_text())
+    except (json.JSONDecodeError, OSError):
+        return None
+
+
+def clear_activity() -> None:
+    """Clear the activity file after a clean handoff."""
+    path = _activity_path()
+    if path.exists():
+        path.unlink()
+
+
+def _compute_staleness(updated_at_str: Optional[str]) -> tuple[bool, str]:
+    """Return (is_stale, age_text) given an ISO timestamp string."""
+    if not updated_at_str:
+        return False, ""
+    try:
+        updated_at = datetime.fromisoformat(updated_at_str.replace("Z", "+00:00"))
+        if updated_at.tzinfo is None:
+            updated_at = updated_at.replace(tzinfo=timezone.utc)
+        age = datetime.now(timezone.utc) - updated_at
+        is_stale = age > timedelta(minutes=30)
+        total_minutes = age.total_seconds() // 60
+        if total_minutes >= 1440:
+            age_text = f"{int(total_minutes // 1440)}d {int((total_minutes % 1440) // 60)}h old"
+        elif total_minutes >= 60:
+            age_text = f"{int(total_minutes // 60)}h {int(total_minutes % 60)}m old"
+        else:
+            age_text = f"{int(total_minutes)}m old"
+        return is_stale, age_text
+    except Exception:
+        return False, ""
+
+
+def build_handoff(activity: Dict) -> str:
+    """Build the handoff text to inject on first message after restart."""
+    mode = activity.get("mode", "simple")
+    updated_at_str = activity.get("updated_at")
+    is_stale, age_text = _compute_staleness(updated_at_str)
+
+    handoff_lines = []
+    if mode == "verbose":
+        if is_stale:
+            handoff_lines.append(
+                f"[Restart handoff \u2014 STALE ({age_text})] Activity below is from {updated_at_str or 'unknown time'}. "
+                "Do NOT auto-execute the next step. Summarize the task and ask the user whether to continue."
+            )
+        else:
+            handoff_lines.append(
+                "[Restart handoff \u2014 FRESH] Agent restarted while working on the task below. Resume immediately."
+            )
+        handoff_lines.append(f"Task: {activity.get('current_task', '?')}")
+        if activity.get("files_modified"):
+            handoff_lines.append(f"Files touched: {', '.join(activity['files_modified'])}")
+        if activity.get("last_action"):
+            handoff_lines.append(f"Last action: {activity['last_action']}")
+        if activity.get("next_expected_step"):
+            handoff_lines.append(f"Next step: {activity['next_expected_step']}")
+    else:
+        if is_stale:
+            handoff_lines.append(
+                f"[Restart handoff \u2014 STALE ({age_text})] "
+                "Back after gateway restart. Last recorded activity is old \u2014 ask the user what to do next."
+            )
+        else:
+            handoff_lines.append(
+                "[Restart handoff \u2014 FRESH] Back after gateway restart."
+            )
+    return "\n".join(handoff_lines)

gateway/config.py

@@ -230,6 +230,7 @@ class SessionResetPolicy:
     mode: str = "both"  # "daily", "idle", "both", or "none"
     at_hour: int = 4  # Hour for daily reset (0-23, local time)
     idle_minutes: int = 1440  # Minutes of inactivity before reset (24 hours)
+    max_duration_minutes: Optional[int] = None  # Hard max session age, regardless of mode
     notify: bool = True  # Send a notification to the user when auto-reset occurs
     notify_exclude_platforms: tuple = ("api_server", "webhook")  # Platforms that don't get reset notifications
 
@@ -238,6 +239,7 @@ def to_dict(self) -> Dict[str, Any]:
             "mode": self.mode,
             "at_hour": self.at_hour,
             "idle_minutes": self.idle_minutes,
+            "max_duration_minutes": self.max_duration_minutes,
             "notify": self.notify,
             "notify_exclude_platforms": list(self.notify_exclude_platforms),
         }
@@ -248,12 +250,14 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionResetPolicy":
         mode = data.get("mode")
         at_hour = data.get("at_hour")
         idle_minutes = data.get("idle_minutes")
+        max_duration_minutes = data.get("max_duration_minutes")
         notify = data.get("notify")
         exclude = data.get("notify_exclude_platforms")
         return cls(
             mode=mode if mode is not None else "both",
             at_hour=at_hour if at_hour is not None else 4,
             idle_minutes=idle_minutes if idle_minutes is not None else 1440,
+            max_duration_minutes=max_duration_minutes if max_duration_minutes is not None else None,
             notify=_coerce_bool(notify, True),
             notify_exclude_platforms=tuple(exclude) if exclude is not None else ("api_server", "webhook"),
         )

gateway/platforms/base.py

@@ -1305,6 +1305,20 @@ def _mark_connected(self) -> None:
         self._fatal_error_message = None
         self._fatal_error_retryable = True
         self._write_runtime_status_safe("connected", platform_state="connected", error_code=None, error_message=None)
+        # Restart awareness: read persisted activity state if present.
+        try:
+            from agent.restart_awareness import read_activity, build_handoff
+            activity = read_activity()
+            if activity:
+                self._pending_restart_handoff = build_handoff(activity)
+                logger.info(
+                    "[%s] restart handoff \u2014 task=%s mode=%s",
+                    self.name,
+                    activity.get("current_task", "?"),
+                    activity.get("mode", "simple"),
+                )
+        except Exception:
+            pass
 
     def _mark_disconnected(self) -> None:
         self._running = False

gateway/platforms/matrix.py

@@ -384,12 +384,11 @@ def __init__(self, config: PlatformConfig):
             "MATRIX_REACTIONS", "true"
         ).lower() not in ("false", "0", "no")
         self._pending_reactions: dict[tuple[str, str], str] = {}
-        # Delay before redacting reactions so Matrix homeservers have time to
-        # deliver the final message event without tripping "missing event"
-        # errors in some clients.  5s is empirically safe; not user-tunable —
-        # if that changes, add a config.yaml entry rather than an env var.
-        self._reaction_redaction_delay_seconds = 5.0
-        self._reaction_redaction_tasks: Set[asyncio.Task] = set()
+
+        # Presence state tracking: map actual activity to Matrix presence.
+        # online = actively processing, unavailable = idle/connected, offline = disconnected.
+        self._presence_active_count: int = 0
+        self._presence_current_state: str = "offline"
 
         # Proxy support — resolve once at init, reuse for all HTTP traffic.
         self._proxy_url: str | None = resolve_proxy_url(platform_env_var="MATRIX_PROXY")
@@ -746,6 +745,34 @@ async def connect(self) -> bool:
                     try:
                         await olm.verify_with_recovery_key(recovery_key)
                         logger.info("Matrix: cross-signing verified via recovery key")
+
+                        # Attempt to self-sign our own device with the recovered SSK.
+                        # This is required for Element to show the green shield
+                        # (device cross-signed by owner).
+                        try:
+                            if client.device_id:
+                                own_device = await olm.get_or_fetch_device(
+                                    client.mxid, client.device_id
+                                )
+                                if own_device:
+                                    await olm.sign_own_device(own_device)
+                                    logger.info(
+                                        "Matrix: successfully self-signed device %s with cross-signing key",
+                                        client.device_id,
+                                    )
+                                else:
+                                    logger.warning(
+                                        "Matrix: could not retrieve own device identity for self-signing"
+                                    )
+                            else:
+                                logger.warning(
+                                    "Matrix: no device_id set, cannot self-sign device"
+                                )
+                        except Exception as exc:
+                            logger.warning(
+                                "Matrix: self-signing device with cross-signing key failed: %s",
+                                exc,
+                            )
                     except Exception as exc:
                         logger.warning(
                             "Matrix: recovery key verification failed: %s", exc
@@ -864,6 +891,7 @@ async def connect(self) -> bool:
         # Start the sync loop.
         self._sync_task = asyncio.create_task(self._sync_loop())
         self._mark_connected()
+        await self.set_presence("online")
         return True
 
     async def disconnect(self) -> None:
@@ -877,14 +905,6 @@ async def disconnect(self) -> None:
             except (asyncio.CancelledError, Exception):
                 pass
 
-        redaction_tasks = list(self._reaction_redaction_tasks)
-        for task in redaction_tasks:
-            if not task.done():
-                task.cancel()
-        if redaction_tasks:
-            await asyncio.gather(*redaction_tasks, return_exceptions=True)
-        self._reaction_redaction_tasks.clear()
-
         # Close the SQLite crypto store database.
         if hasattr(self, "_crypto_db") and self._crypto_db:
             try:
@@ -893,6 +913,12 @@ async def disconnect(self) -> None:
                 logger.debug("Matrix: could not close crypto DB on disconnect: %s", exc)
 
         if self._client:
+            try:
+                self._presence_active_count = 0
+                self._presence_current_state = "offline"
+                await self.set_presence("offline")
+            except Exception:
+                pass
             try:
                 await self._client.api.session.close()
             except Exception:
@@ -1975,73 +2001,45 @@ async def _redact_reaction(
         """Remove a reaction by redacting its event."""
         return await self.redact_message(room_id, reaction_event_id, reason)
 
-    def _schedule_reaction_redaction(
-        self,
-        room_id: str,
-        reaction_event_id: str,
-        reason: str = "",
-    ) -> None:
-        """Redact a reaction after a short delay so message delivery settles."""
-
-        async def _redact_later() -> None:
-            try:
-                if self._reaction_redaction_delay_seconds:
-                    await asyncio.sleep(self._reaction_redaction_delay_seconds)
-                if not await self._redact_reaction(room_id, reaction_event_id, reason):
-                    logger.debug(
-                        "Matrix: failed to redact reaction %s", reaction_event_id
-                    )
-            except asyncio.CancelledError:
-                raise
-            except Exception as exc:
-                logger.debug(
-                    "Matrix: delayed reaction redaction failed for %s: %s",
-                    reaction_event_id,
-                    exc,
-                )
-
-        task = asyncio.create_task(_redact_later())
-        self._reaction_redaction_tasks.add(task)
-        task.add_done_callback(self._reaction_redaction_tasks.discard)
-
     async def on_processing_start(self, event: MessageEvent) -> None:
-        """Add eyes reaction when the agent starts processing a message."""
-        if not self._reactions_enabled:
-            return
-        msg_id = event.message_id
-        room_id = event.source.chat_id
-        if msg_id and room_id:
-            reaction_event_id = await self._send_reaction(room_id, msg_id, "\U0001f440")
-            if reaction_event_id:
-                self._pending_reactions[(room_id, msg_id)] = reaction_event_id
+        """Add eyes reaction and set presence to unavailable when actively working."""
+        if self._reactions_enabled:
+            msg_id = event.message_id
+            room_id = event.source.chat_id
+            if msg_id and room_id:
+                reaction_event_id = await self._send_reaction(room_id, msg_id, "\U0001f440")
+                if reaction_event_id:
+                    self._pending_reactions[(room_id, msg_id)] = reaction_event_id
+        # Track active processing count and transition presence.
+        self._presence_active_count += 1
+        if self._presence_active_count == 1 and self._presence_current_state != "unavailable":
+            await self.set_presence("unavailable", status_msg="working on Hermes")
 
     async def on_processing_complete(
         self,
         event: MessageEvent,
         outcome: ProcessingOutcome,
     ) -> None:
-        """Replace eyes with checkmark (success) or cross (failure)."""
-        if not self._reactions_enabled:
-            return
-        msg_id = event.message_id
-        room_id = event.source.chat_id
-        if not msg_id or not room_id:
-            return
-        if outcome == ProcessingOutcome.CANCELLED:
-            return
-        reaction_key = (room_id, msg_id)
-        if reaction_key in self._pending_reactions:
-            eyes_event_id = self._pending_reactions.pop(reaction_key)
-            self._schedule_reaction_redaction(
-                room_id,
-                eyes_event_id,
-                "processing complete",
-            )
-        await self._send_reaction(
-            room_id,
-            msg_id,
-            "\u2705" if outcome == ProcessingOutcome.SUCCESS else "\u274c",
-        )
+        """Replace eyes with checkmark (success) or cross (failure), and transition presence back to idle."""
+        if self._reactions_enabled:
+            msg_id = event.message_id
+            room_id = event.source.chat_id
+            if msg_id and room_id:
+                if outcome != ProcessingOutcome.CANCELLED:
+                    reaction_key = (room_id, msg_id)
+                    if reaction_key in self._pending_reactions:
+                        eyes_event_id = self._pending_reactions.pop(reaction_key)
+                        if not await self._redact_reaction(room_id, eyes_event_id):
+                            logger.debug("Matrix: failed to redact eyes reaction %s", eyes_event_id)
+                    await self._send_reaction(
+                        room_id,
+                        msg_id,
+                        "\u2705" if outcome == ProcessingOutcome.SUCCESS else "\u274c",
+                    )
+        # Track active processing count and transition presence.
+        self._presence_active_count = max(0, self._presence_active_count - 1)
+        if self._presence_active_count == 0 and self._presence_current_state != "online":
+            await self.set_presence("online")
 
     async def _on_reaction(self, event: Any) -> None:
         """Handle incoming reaction events."""
@@ -2115,8 +2113,11 @@ async def _redact_bot_approval_reactions(
     ) -> None:
         """Redact the bot's seed ✅/❎ reactions, leaving only the user's reaction."""
         for emoji, evt_id in prompt.bot_reaction_events.items():
-            self._schedule_reaction_redaction(room_id, evt_id, "approval resolved")
-            logger.debug("Matrix: scheduled bot reaction redaction %s (%s)", emoji, evt_id)
+            try:
+                await self.redact_message(room_id, evt_id, "approval resolved")
+                logger.debug("Matrix: redacted bot reaction %s (%s)", emoji, evt_id)
+            except Exception as exc:
+                logger.debug("Matrix: failed to redact bot reaction %s: %s", emoji, exc)
 
     # ------------------------------------------------------------------
     # Text message aggregation (handles Matrix client-side splits)