fix(gateway): retry lifecycle notifications

gateway/run.py

@@ -13760,8 +13760,8 @@ async def _watch_update_progress(
             # Fall back to old behavior: wait for exit code and send final notification
             while (pending_path.exists() or claimed_path.exists()) and loop.time() < deadline:
                 if exit_code_path.exists():
-                    await self._send_update_notification()
-                    return
+                    if await self._send_update_notification():
+                        return
                 await asyncio.sleep(poll_interval)
             if (pending_path.exists() or claimed_path.exists()) and not exit_code_path.exists():
                 exit_code_path.write_text("124")
@@ -13825,6 +13825,8 @@ async def _flush_buffer() -> None:
                     logger.info("Update finished (exit=%s), notified %s", exit_code, session_key)
                 except Exception as e:
                     logger.warning("Update final notification failed: %s", e)
+                    await asyncio.sleep(poll_interval)
+                    continue
 
                 # Cleanup
                 for p in (pending_path, claimed_path, output_path,
@@ -13899,19 +13901,33 @@ async def _flush_buffer() -> None:
 
             await asyncio.sleep(poll_interval)
 
+        if exit_code_path.exists():
+            logger.warning(
+                "Update watcher stopped before final notification was delivered; "
+                "leaving update marker files for the next gateway startup"
+            )
+            return
+
         # Timeout
         if not exit_code_path.exists():
             logger.warning("Update watcher timed out after %.0fs", timeout)
             exit_code_path.write_text("124")
             await _flush_buffer()
+            timeout_notified = False
             try:
                 await adapter.send(
                     chat_id,
                     "❌ Hermes update timed out after 30 minutes.",
                     metadata=metadata,
                 )
-            except Exception:
-                pass
+                timeout_notified = True
+            except Exception as e:
+                logger.warning("Update timeout notification failed: %s", e)
+            if not timeout_notified:
+                logger.warning(
+                    "Preserving timed-out update marker files so notification can retry"
+                )
+                return
             for p in (pending_path, claimed_path, output_path,
                       exit_code_path, prompt_path):
                 p.unlink(missing_ok=True)
@@ -13921,8 +13937,10 @@ async def _flush_buffer() -> None:
     async def _send_update_notification(self) -> bool:
         """If an update finished, notify the user.
 
-        Returns False when the update is still running so a caller can retry
-        later. Returns True after a definitive send/skip decision.
+        Returns False when the update is still running or the completion
+        notification could not be delivered yet. Marker files are preserved
+        in that case so a later gateway startup can retry.
+        Returns True after a definitive send/skip decision.
 
         This is the legacy notification path used when the streaming watcher
         cannot resolve the adapter (e.g. after a gateway restart where the
@@ -13937,7 +13955,6 @@ async def _send_update_notification(self) -> bool:
             return False
 
         cleanup = True
-        active_pending_path = claimed_path
         try:
             if pending_path.exists():
                 try:
@@ -13948,16 +13965,19 @@ async def _send_update_notification(self) -> bool:
             elif not claimed_path.exists():
                 return True
 
-            pending = json.loads(claimed_path.read_text())
+            try:
+                pending = json.loads(claimed_path.read_text())
+            except Exception as e:
+                logger.warning("Post-update notification marker is invalid: %s", e)
+                return True
+
             platform_str = pending.get("platform")
             chat_id = pending.get("chat_id")
             thread_id = pending.get("thread_id")
 
             if not exit_code_path.exists():
                 logger.info("Update notification deferred: update still running")
                 cleanup = False
-                active_pending_path = pending_path
-                claimed_path.replace(pending_path)
                 return False
 
             exit_code_raw = exit_code_path.read_text().strip() or "1"
@@ -13972,36 +13992,50 @@ async def _send_update_notification(self) -> bool:
             platform = Platform(platform_str)
             adapter = self.adapters.get(platform)
 
-            if adapter and chat_id:
-                metadata = {"thread_id": thread_id} if thread_id else None
-                # Strip ANSI escape codes for clean display
-                output = re.sub(r'\x1b\[[0-9;]*m', '', output).strip()
-                if output:
-                    if len(output) > 3500:
-                        output = "…" + output[-3500:]
-                    if exit_code == 0:
-                        msg = f"✅ Hermes update finished.\n\n```\n{output}\n```"
-                    else:
-                        msg = f"❌ Hermes update failed.\n\n```\n{output}\n```"
-                elif exit_code == 0:
-                    msg = "✅ Hermes update finished successfully."
-                else:
-                    msg = "❌ Hermes update failed. Check the gateway logs or run `hermes update` manually for details."
-                await adapter.send(chat_id, msg, metadata=metadata)
+            if not adapter or not chat_id:
                 logger.info(
-                    "Sent post-update notification to %s:%s (exit=%s)",
+                    "Update notification deferred: %s adapter/chat not ready",
                     platform_str,
-                    chat_id,
-                    exit_code,
                 )
+                cleanup = False
+                return False
+
+            metadata = {"thread_id": thread_id} if thread_id else None
+            # Strip ANSI escape codes for clean display
+            output = re.sub(r'\x1b\[[0-9;]*m', '', output).strip()
+            if output:
+                if len(output) > 3500:
+                    output = "…" + output[-3500:]
+                if exit_code == 0:
+                    msg = f"✅ Hermes update finished.\n\n```\n{output}\n```"
+                else:
+                    msg = f"❌ Hermes update failed.\n\n```\n{output}\n```"
+            elif exit_code == 0:
+                msg = "✅ Hermes update finished successfully."
+            else:
+                msg = "❌ Hermes update failed. Check the gateway logs or run `hermes update` manually for details."
+            await adapter.send(chat_id, msg, metadata=metadata)
+            logger.info(
+                "Sent post-update notification to %s:%s (exit=%s)",
+                platform_str,
+                chat_id,
+                exit_code,
+            )
         except Exception as e:
             logger.warning("Post-update notification failed: %s", e)
+            cleanup = False
+            return False
         finally:
             if cleanup:
-                active_pending_path.unlink(missing_ok=True)
+                pending_path.unlink(missing_ok=True)
                 claimed_path.unlink(missing_ok=True)
                 output_path.unlink(missing_ok=True)
                 exit_code_path.unlink(missing_ok=True)
+            elif claimed_path.exists():
+                try:
+                    claimed_path.replace(pending_path)
+                except OSError as e:
+                    logger.warning("Could not restore update pending marker for retry: %s", e)
 
         return True
 
@@ -14011,8 +14045,14 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
         if not notify_path.exists():
             return None
 
+        cleanup = True
         try:
-            data = json.loads(notify_path.read_text())
+            try:
+                data = json.loads(notify_path.read_text())
+            except Exception as e:
+                logger.warning("Restart notification marker is invalid: %s", e)
+                return None
+
             platform_str = data.get("platform")
             chat_id = data.get("chat_id")
             thread_id = data.get("thread_id")
@@ -14024,9 +14064,10 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
             adapter = self.adapters.get(platform)
             if not adapter:
                 logger.debug(
-                    "Restart notification skipped: %s adapter not connected",
+                    "Restart notification deferred: %s adapter not connected",
                     platform_str,
                 )
+                cleanup = False
                 return None
 
             platform_cfg = self.config.platforms.get(platform)
@@ -14054,6 +14095,7 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
                     chat_id,
                     getattr(result, "error", "send returned success=False"),
                 )
+                cleanup = False
                 return None
 
             logger.info(
@@ -14064,9 +14106,11 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
             return str(platform_str), str(chat_id), str(thread_id) if thread_id else None
         except Exception as e:
             logger.warning("Restart notification failed: %s", e)
+            cleanup = False
             return None
         finally:
-            notify_path.unlink(missing_ok=True)
+            if cleanup:
+                notify_path.unlink(missing_ok=True)
 
     async def _send_home_channel_startup_notifications(
         self,

tests/gateway/test_restart_notification.py

@@ -403,7 +403,7 @@ async def test_send_restart_notification_noop_when_no_file(tmp_path, monkeypatch
 
 @pytest.mark.asyncio
 async def test_send_restart_notification_skips_when_adapter_missing(tmp_path, monkeypatch):
-    """If the requester's platform isn't connected, clean up without crashing."""
+    """If the requester's platform isn't connected, keep marker for retry."""
     monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
 
     notify_path = tmp_path / ".restart_notify.json"
@@ -416,15 +416,14 @@ async def test_send_restart_notification_skips_when_adapter_missing(tmp_path, mo
 
     await runner._send_restart_notification()
 
-    # File cleaned up even though we couldn't send
-    assert not notify_path.exists()
+    assert notify_path.exists()
 
 
 @pytest.mark.asyncio
-async def test_send_restart_notification_cleans_up_on_send_failure(
+async def test_send_restart_notification_preserves_marker_on_send_failure(
     tmp_path, monkeypatch
 ):
-    """If the adapter.send() raises, the file is still cleaned up."""
+    """If adapter.send() raises, keep marker for a later retry."""
     monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
 
     notify_path = tmp_path / ".restart_notify.json"
@@ -438,9 +437,8 @@ async def test_send_restart_notification_cleans_up_on_send_failure(
 
     delivered_target = await runner._send_restart_notification()
 
-    # File cleaned up even though send raised.
     assert delivered_target is None
-    assert not notify_path.exists()
+    assert notify_path.exists()
 
 
 @pytest.mark.asyncio
@@ -492,8 +490,7 @@ async def test_send_restart_notification_logs_warning_on_sendresult_failure(
         "Expected a WARNING line mentioning the failure; "
         f"got records: {[(r.levelname, r.getMessage()) for r in caplog.records]}"
     )
-    # Still cleans up.
-    assert not notify_path.exists()
+    assert notify_path.exists()
 
 
 @pytest.mark.asyncio

tests/gateway/test_update_command.py

@@ -605,8 +605,8 @@ async def test_cleans_up_files_after_notification(self, tmp_path):
         assert not exit_code_path.exists()
 
     @pytest.mark.asyncio
-    async def test_cleans_up_on_error(self, tmp_path):
-        """Files are cleaned up even if notification fails."""
+    async def test_preserves_files_on_send_error(self, tmp_path):
+        """Files are preserved if notification delivery fails."""
         runner = _make_runner()
         hermes_home = tmp_path / "hermes"
         hermes_home.mkdir()
@@ -626,12 +626,13 @@ async def test_cleans_up_on_error(self, tmp_path):
         runner.adapters = {Platform.TELEGRAM: mock_adapter}
 
         with patch("gateway.run._hermes_home", hermes_home):
-            await runner._send_update_notification()
+            result = await runner._send_update_notification()
 
-        # Files should still be cleaned up (finally block)
-        assert not pending_path.exists()
-        assert not output_path.exists()
-        assert not exit_code_path.exists()
+        assert result is False
+        assert pending_path.exists()
+        assert output_path.exists()
+        assert exit_code_path.exists()
+        assert not (hermes_home / ".update_pending.claimed.json").exists()
 
     @pytest.mark.asyncio
     async def test_handles_corrupt_pending_file(self, tmp_path):
@@ -674,9 +675,10 @@ async def test_no_adapter_for_platform(self, tmp_path):
 
         # send should not have been called (wrong platform)
         mock_adapter.send.assert_not_called()
-        # Files should still be cleaned up
-        assert not pending_path.exists()
-        assert not exit_code_path.exists()
+        # Files should be kept so a later adapter reconnect/startup can retry.
+        assert pending_path.exists()
+        assert exit_code_path.exists()
+        assert not (hermes_home / ".update_pending.claimed.json").exists()
 
 
 # ---------------------------------------------------------------------------

tests/gateway/test_update_streaming.py

@@ -413,6 +413,38 @@ async def test_cleans_up_on_completion(self, tmp_path):
         assert not output_path.exists()
         assert not exit_code_path.exists()
 
+    @pytest.mark.asyncio
+    async def test_preserves_markers_when_final_notification_fails(self, tmp_path):
+        """Final delivery failures leave marker files for a later retry."""
+        runner = _make_runner()
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+
+        pending = {"platform": "telegram", "chat_id": "111", "user_id": "222",
+                   "session_key": "agent:main:telegram:dm:111"}
+        pending_path = hermes_home / ".update_pending.json"
+        output_path = hermes_home / ".update_output.txt"
+        exit_code_path = hermes_home / ".update_exit_code"
+        pending_path.write_text(json.dumps(pending))
+        output_path.write_text("done\n")
+        exit_code_path.write_text("0")
+
+        mock_adapter = AsyncMock()
+        mock_adapter.send.side_effect = RuntimeError("network error")
+        runner.adapters = {Platform.TELEGRAM: mock_adapter}
+
+        with patch("gateway.run._hermes_home", hermes_home):
+            await runner._watch_update_progress(
+                poll_interval=0.05,
+                stream_interval=0.05,
+                timeout=0.15,
+            )
+
+        assert pending_path.exists()
+        assert output_path.exists()
+        assert exit_code_path.exists()
+        assert not (hermes_home / ".update_pending.claimed.json").exists()
+
     @pytest.mark.asyncio
     async def test_failure_exit_code(self, tmp_path):
         """Non-zero exit code sends failure message."""