diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index ef08b05405a7..78e0dd7e25c3 100644 --- a/gateway/platforms/base.py +++ b/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. diff --git a/gateway/platforms/homeassistant.py b/gateway/platforms/homeassistant.py index 746465594cee..6bc9ae6eb613 100644 --- a/gateway/platforms/homeassistant.py +++ b/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( diff --git a/gateway/run.py b/gateway/run.py index 97f72121bb76..aadb067dcbe2 100644 --- a/gateway/run.py +++ b/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) diff --git a/gateway/session.py b/gateway/session.py index fcff336afa76..3129f7a325e8 100644 --- a/gateway/session.py +++ b/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() diff --git a/scripts/release.py b/scripts/release.py index 7bad9dd1a117..7c84fb9c038c 100755 --- a/scripts/release.py +++ b/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", diff --git a/tests/gateway/test_clean_shutdown_marker.py b/tests/gateway/test_clean_shutdown_marker.py index 1a476bc49a57..c6d3cab5c13a 100644 --- a/tests/gateway/test_clean_shutdown_marker.py +++ b/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.""" diff --git a/tests/gateway/test_restart_resume_pending.py b/tests/gateway/test_restart_resume_pending.py index 77c639d05f76..bda6c7a412f1 100644 --- a/tests/gateway/test_restart_resume_pending.py +++ b/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 # ---------------------------------------------------------------------------