chore(discord): best-effort flush thread mappings on stop() + restart regression tests (#5461)

* fix(discord): flush thread mappings on shutdown to survive restart (#2897)

The Discord channel already persists channel->thread mappings, but a mapping created right before a hard shutdown (process killed between thread creation and its background persistence write) could still be lost. Flush in-memory mappings in stop() as a best-effort safety net.

Add regression tests covering the persist/load round-trip across a simulated restart and the stop() flush path.

Refs #2897

* fix(discord): gate stop() flush on _thread_store_loaded (#5461 review)

Address review feedback:
- Only flush thread mappings from stop() once _load_active_threads() has
  run, so a stop() taken before the load (start() bailed on a missing
  bot_token / discord import error) cannot overwrite the store with {}.
- Fix isort order in the test module (ruff I001).
- Make the restart test a plain sync test (it never awaits).
- Add a regression test proving stop() does not clobber the store before load.

* fix(discord): gate stop() flush on _thread_store_loaded (#5461 review)

Address review feedback:
- Only flush thread mappings from stop() once _load_active_threads() has
  run, so a stop() taken before the load (start() bailed on a missing
  bot_token / discord import error) cannot overwrite the store with {}.
- Fix isort order in the test module (ruff I001).
- Make the restart test a plain sync test (it never awaits).
- Add a regression test proving stop() does not clobber the store before load.

---------

Co-authored-by: wcy12378 <wcy12378@users.noreply.github.com>

backend/app/channels/discord.py

@@ -71,6 +71,12 @@ class DiscordChannel(Channel):
         # Lock protecting _active_threads and the JSON file from concurrent access.
         # _run_client (Discord loop thread) and the main thread both read/write.
         self._thread_store_lock = threading.Lock()
+        # Set once _load_active_threads() has run: the in-memory map is then a
+        # faithful view of the file, so stop() may flush it. It stays False when
+        # start() bails before the load (missing bot_token, discord import
+        # error) — ChannelService then stops the instance, and an ungated flush
+        # would overwrite the persisted mappings with an empty snapshot (#2897).
+        self._thread_store_loaded = False
         store = config.get("channel_store")
         if store is not None:
             self._thread_store_path = store._path.parent / "discord_threads.json"
@@ -140,17 +146,24 @@ class DiscordChannel(Channel):
             try:
                 if not self._thread_store_path.exists():
                     logger.debug("[Discord] no thread mappings file at %s", self._thread_store_path)
-                    return
-                data = json.loads(self._thread_store_path.read_text())
-                self._active_threads.clear()
-                self._active_thread_ids.clear()
-                for channel_id, thread_id in data.items():
-                    self._active_threads[channel_id] = thread_id
-                    self._active_thread_ids.add(thread_id)
-                if self._active_threads:
-                    logger.info("[Discord] restored %d thread mappings from %s", len(self._active_threads), self._thread_store_path)
+                else:
+                    data = json.loads(self._thread_store_path.read_text())
+                    self._active_threads.clear()
+                    self._active_thread_ids.clear()
+                    for channel_id, thread_id in data.items():
+                        self._active_threads[channel_id] = thread_id
+                        self._active_thread_ids.add(thread_id)
+                    if self._active_threads:
+                        logger.info("[Discord] restored %d thread mappings from %s", len(self._active_threads), self._thread_store_path)
             except Exception:
                 logger.exception("[Discord] failed to load thread mappings")
+                return
+            # The in-memory map is now a superset of the on-disk store (empty
+            # when the file is absent), so stop() may safely flush it. Until
+            # this point — a start() that bailed on a missing bot_token or a
+            # discord import error, which ChannelService then stops — the map is
+            # empty and flushing would clobber the file with {}.
+            self._thread_store_loaded = True
 
     def _record_thread_mapping(self, channel_id: str, thread_id: str) -> None:
         """Synchronously update the in-memory channel->thread mapping and its reverse-lookup set.
@@ -196,6 +209,20 @@ class DiscordChannel(Channel):
         self._running = False
         self.bus.unsubscribe_outbound(self._on_outbound)
 
+        # Best-effort durability: flush in-memory thread mappings so the most
+        # recent channel->thread mapping survives a hard shutdown (process
+        # killed between a thread creation and its background persistence
+        # write). The create path already persists off the event loop after
+        # each new thread, so this is a safety net, not the primary write path.
+        # Gated on _thread_store_loaded: a stop() taken before the initial load
+        # has an empty in-memory map, and flushing it would overwrite the file
+        # with {} — the #2897 data loss this PR exists to prevent.
+        if self._thread_store_loaded:
+            try:
+                await asyncio.to_thread(self._persist_thread_mappings)
+            except Exception:
+                logger.warning("[Discord] failed to flush thread mappings during shutdown")
+
         discord_loop = self._discord_loop
         current_loop = asyncio.get_running_loop()
         if discord_loop is None or discord_loop is current_loop:

backend/tests/test_discord_channel.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import builtins
 import gc
+import json
 import threading
 import weakref
 from types import SimpleNamespace
@@ -33,6 +34,87 @@ def test_discord_channel_init() -> None:
     assert channel.name == "discord"
 
 
+# ---------------------------------------------------------------------------
+# thread-mapping persistence across restart (#2897)
+# ---------------------------------------------------------------------------
+
+
+def test_discord_thread_mapping_persists_across_restart(tmp_path) -> None:
+    """A channel->thread mapping written before shutdown is restored on a
+    subsequent start, so conversations are not lost across restarts (#2897)."""
+    store_path = tmp_path / "discord_threads.json"
+
+    # First process lifetime: record and persist a mapping.
+    first = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
+    first._thread_store_path = store_path
+    first._record_thread_mapping("chan-1", "thread-1")
+    first._persist_thread_mappings()
+    assert store_path.exists()
+    assert json.loads(store_path.read_text()) == {"chan-1": "thread-1"}
+
+    # Restart: a brand-new channel instance reads the same file.
+    second = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
+    second._thread_store_path = store_path
+    second._active_threads.clear()
+    second._active_thread_ids.clear()
+    second._load_active_threads()
+
+    assert second._active_threads == {"chan-1": "thread-1"}
+    assert "thread-1" in second._active_thread_ids
+
+
+@pytest.mark.asyncio
+async def test_discord_stop_flushes_thread_mappings(tmp_path) -> None:
+    """stop() best-effort flushes in-memory thread mappings to disk so the
+    most recent mapping survives a hard shutdown."""
+    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
+    channel._thread_store_path = tmp_path / "discord_threads.json"
+    # stop() only flushes once the load has marked the in-memory map
+    # authoritative (see DiscordChannel._thread_store_loaded); simulate the
+    # normal post-start() state so this exercises the flush, not the guard.
+    channel._thread_store_loaded = True
+    # Minimal shutdown context: no live client/loop/thread to tear down.
+    channel._discord_loop = None
+    channel._client = None
+    channel._thread = None
+    channel._cancel_ephemeral_tasks = AsyncMock()
+
+    channel._record_thread_mapping("chan-2", "thread-2")
+    await channel.stop()
+
+    assert channel._thread_store_path.exists()
+    assert json.loads(channel._thread_store_path.read_text()) == {"chan-2": "thread-2"}
+
+
+@pytest.mark.asyncio
+async def test_discord_stop_does_not_clobber_store_before_load(tmp_path) -> None:
+    """stop() before the initial load must not overwrite the persisted file.
+
+    ``ChannelService`` deliberately stops a channel whose ``start()`` bailed
+    before ``_load_active_threads()`` ran (missing bot_token / discord import
+    error), when the in-memory map is still empty. An ungated flush would write
+    ``{}`` over the persisted mappings — the #2897 data loss this PR exists to
+    prevent.
+    """
+    store_path = tmp_path / "discord_threads.json"
+    store_path.write_text(json.dumps({"chan-9": "thread-9"}))
+
+    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
+    channel._thread_store_path = store_path
+    # start() bailed before the load, so the flag is still False.
+    assert channel._thread_store_loaded is False
+    # Minimal shutdown context: no live client/loop/thread to tear down.
+    channel._discord_loop = None
+    channel._client = None
+    channel._thread = None
+    channel._cancel_ephemeral_tasks = AsyncMock()
+
+    await channel.stop()
+
+    # The pre-existing mapping survives intact: nothing was flushed over it.
+    assert json.loads(store_path.read_text()) == {"chan-9": "thread-9"}
+
+
 def _make_discord_message(text: str):
     return SimpleNamespace(
         id=111,