fix(gateway): batch critical fixes — session resume, /new race, HA WebSocket scheme

gateway/platforms/base.py

@@ -2489,15 +2489,20 @@ async def _dispatch_active_session_command(
 
         try:
             response = await self._message_handler(event)
-            # Old adapter task (if any) is cancelled AFTER the runner has
-            # fully handled the command — keeps ordering deterministic.
-            await self.cancel_session_processing(
-                session_key,
-                release_guard=False,
-                discard_pending=False,
-            )
             _text, _eph_ttl = self._unwrap_ephemeral(response)
+            # Send the response BEFORE cancelling the old task so the send
+            # cannot be affected by task-cancellation side effects (race
+            # condition fix — issue #18912).  Previously the send happened
+            # after cancel_session_processing, which could silently drop the
+            # "/new" confirmation when an agent was actively running.
             if _text:
+                logger.info(
+                    "[%s] Sending command '/%s' response (%d chars) to %s",
+                    self.name,
+                    cmd,
+                    len(_text),
+                    event.source.chat_id,
+                )
                 _r = await self._send_with_retry(
                     chat_id=event.source.chat_id,
                     content=_text,
@@ -2510,6 +2515,13 @@ async def _dispatch_active_session_command(
                         message_id=_r.message_id,
                         ttl_seconds=_eph_ttl,
                     )
+            # Old adapter task (if any) is cancelled AFTER the response has
+            # been sent — keeps ordering deterministic and avoids the race.
+            await self.cancel_session_processing(
+                session_key,
+                release_guard=False,
+                discard_pending=False,
+            )
         except Exception:
             # On failure, restore the original guard if one still exists so
             # we don't leave the session in a half-reset state.

gateway/platforms/homeassistant.py

@@ -139,7 +139,7 @@ async def connect(self) -> bool:
 
     async def _ws_connect(self) -> bool:
         """Establish WebSocket connection and authenticate."""
-        ws_url = self._hass_url.replace("http://", "ws://").replace("https://", "wss://")
+        ws_url = self._hass_url.replace("https://", "wss://").replace("http://", "ws://")
         ws_url = f"{ws_url}/api/websocket"
 
         self._session = aiohttp.ClientSession(

gateway/run.py

@@ -2755,7 +2755,7 @@ async def start(self) -> bool:
             try:
                 suspended = self.session_store.suspend_recently_active()
                 if suspended:
-                    logger.info("Suspended %d in-flight session(s) from previous run", suspended)
+                    logger.info("Marked %d in-flight session(s) as resumable from previous run", suspended)
             except Exception as e:
                 logger.warning("Session suspension on startup failed: %s", e)
 

gateway/session.py

@@ -1086,19 +1086,22 @@ def prune_old_entries(self, max_age_days: int) -> int:
         return len(removed_keys)
 
     def suspend_recently_active(self, max_age_seconds: int = 120) -> int:
-        """Mark recently-active sessions as suspended.
-
-        Called on gateway startup to prevent sessions that were likely
-        in-flight when the gateway last exited from being blindly resumed
-        (#7536).  Only suspends sessions updated within *max_age_seconds*
-        to avoid resetting long-idle sessions that are harmless to resume.
-        Returns the number of sessions that were suspended.
-
-        Entries flagged ``resume_pending=True`` are skipped — those were
-        marked intentionally by the drain-timeout path as recoverable.
-        Terminal escalation for genuinely stuck ``resume_pending`` sessions
-        is handled by the existing ``.restart_failure_counts`` stuck-loop
-        counter, which runs after this method on startup.
+        """Mark recently-active sessions as resumable after an unexpected exit.
+
+        Called on gateway startup after a crash or fast restart to preserve
+        in-flight sessions instead of destroying their conversation history
+        (#7536).  Only marks sessions updated within *max_age_seconds* to
+        avoid touching long-idle sessions.  Sets ``resume_pending=True`` so
+        the next incoming message on the same session_key auto-resumes from
+        the existing transcript.
+
+        Entries already flagged ``resume_pending=True`` are skipped.  Entries
+        explicitly ``suspended=True`` (from /stop or stuck-loop escalation)
+        are also skipped.  Terminal escalation for genuinely stuck sessions
+        is still handled by the existing ``.restart_failure_counts`` counter
+        (threshold 3), which runs after this method and sets ``suspended=True``.
+
+        Returns the number of sessions marked resumable.
         """
         from datetime import timedelta
 
@@ -1110,7 +1113,9 @@ def suspend_recently_active(self, max_age_seconds: int = 120) -> int:
                 if entry.resume_pending:
                     continue
                 if not entry.suspended and entry.updated_at >= cutoff:
-                    entry.suspended = True
+                    entry.resume_pending = True
+                    entry.resume_reason = "restart_interrupted"
+                    entry.last_resume_marked_at = _now()
                     count += 1
             if count:
                 self._save()

scripts/release.py

@@ -507,6 +507,8 @@
     "jz.pentest@gmail.com": "0xyg3n",
     "7093928+0xyg3n@users.noreply.github.com": "0xyg3n",
     "nftpoetrist@gmail.com": "nftpoetrist",  # PR #18982
+    "millerc79@users.noreply.github.com": "millerc79",  # PR #19033
+    "hermes@example.com": "shellybotmoyer",  # PR #18915 (bot-committed)
     "hypnosis.mda@gmail.com": "Hypn0sis",
     "ywt000818@gmail.com": "OwenYWT",
     "dhandhalyabhavik@gmail.com": "v1k22",

tests/gateway/test_clean_shutdown_marker.py

@@ -49,9 +49,10 @@ def test_suspends_recently_active_sessions(self, tmp_path):
         count = store.suspend_recently_active()
         assert count == 1
 
-        # Re-fetch — should be suspended now
+        # Re-fetch — should be resume_pending (preserved, not wiped)
         refreshed = store.get_or_create_session(source)
-        assert refreshed.was_auto_reset
+        assert refreshed.resume_pending
+        assert refreshed.session_id == entry.session_id  # same session preserved
 
     def test_does_not_suspend_old_sessions(self, tmp_path):
         store = _make_store(tmp_path)
@@ -66,21 +67,22 @@ def test_does_not_suspend_old_sessions(self, tmp_path):
         count = store.suspend_recently_active(max_age_seconds=120)
         assert count == 0
 
-    def test_already_suspended_not_double_counted(self, tmp_path):
+    def test_already_resume_pending_not_double_counted(self, tmp_path):
         store = _make_store(tmp_path)
         source = _make_source()
         entry = store.get_or_create_session(source)
 
-        # Suspend once
+        # Mark resume_pending once
         count1 = store.suspend_recently_active()
         assert count1 == 1
 
-        # Create a new session (the old one got reset on next access)
+        # Re-fetch returns the SAME session (preserved, not reset)
         entry2 = store.get_or_create_session(source)
+        assert entry2.session_id == entry.session_id
 
-        # Suspend again — the new session is recent but not yet suspended
+        # Second call skips already-resume_pending entries
         count2 = store.suspend_recently_active()
-        assert count2 == 1
+        assert count2 == 0
 
 
 # ---------------------------------------------------------------------------
@@ -180,11 +182,11 @@ def test_no_marker_triggers_suspension(self, tmp_path, monkeypatch):
         else:
             store.suspend_recently_active()
 
-        # Session SHOULD be suspended (crash recovery)
+        # Session SHOULD be resume_pending (crash recovery preserves history)
         with store._lock:
             store._ensure_loaded_locked()
-            suspended_count = sum(1 for e in store._entries.values() if e.suspended)
-        assert suspended_count == 1, "Session should be suspended after crash (no marker)"
+            resume_count = sum(1 for e in store._entries.values() if e.resume_pending)
+        assert resume_count == 1, "Session should be resume_pending after crash (no marker)"
 
     def test_marker_written_on_restart_stop(self, tmp_path, monkeypatch):
         """stop(restart=True) should also write the marker."""

tests/gateway/test_restart_resume_pending.py

@@ -376,8 +376,8 @@ def test_resume_pending_entries_not_suspended(self, tmp_path):
         assert e.suspended is False
         assert e.resume_pending is True
 
-    def test_non_resume_pending_still_suspended(self, tmp_path):
-        """Non-resume sessions still get the old crash-recovery suspension."""
+    def test_non_resume_pending_gets_resume_pending(self, tmp_path):
+        """Non-resume sessions are now marked resume_pending (not suspended)."""
         store = _make_store(tmp_path)
         source_a = _make_source(chat_id="a")
         source_b = _make_source(chat_id="b")
@@ -386,9 +386,11 @@ def test_non_resume_pending_still_suspended(self, tmp_path):
         store.mark_resume_pending(entry_a.session_key)
 
         count = store.suspend_recently_active()
+        # entry_a is already resume_pending → skipped. entry_b gets marked.
         assert count == 1
         assert store._entries[entry_a.session_key].suspended is False
-        assert store._entries[entry_b.session_key].suspended is True
+        assert store._entries[entry_b.session_key].resume_pending is True
+        assert store._entries[entry_b.session_key].suspended is False
 
 
 # ---------------------------------------------------------------------------