fix(channels): fence restart during service shutdown (#6503)

* fix(channels): fence restart during service shutdown

* test(channels): fence restart during service stop

* test(channels): exercise restart after shutdown channel snapshot

* docs(channels): document service shutdown publication fence

* fix(channels): keep agent guidance within CI hard budget

* fix(channels): fence publication after shutdown completes

* test(channels): cover resume after completed service shutdown

* test(channels): mark direct-start fixtures as active services

* test(channels): mark direct-start fixtures as active services

* style(channels): format shutdown regression test

* fix(channels): drain channel startup completing after service stop

* test(channels): ensure late channel startup is drained after stop

* style(tests): format channel shutdown regression

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/channels/service.py

@@ -160,6 +160,7 @@ class ChannelService:
         self._channels: dict[str, Any] = {}  # name -> Channel instance
         self._config = config
         self._running = False
+        self._stopping = False
         self._readiness_locks: dict[str, asyncio.Lock] = {}
         self._config_epochs: dict[str, int] = {}
 
@@ -202,6 +203,8 @@ class ChannelService:
         """Start the manager and all enabled channels."""
         if self._running:
             return
+        if self._stopping:
+            raise RuntimeError("cannot start ChannelService while shutdown is incomplete")
 
         await self.manager.start()
         self._running = True
@@ -294,6 +297,7 @@ class ChannelService:
 
     async def stop(self) -> None:
         """Drain accepted messages while channels can still deliver replies."""
+        self._stopping = True
         self._running = False
         # Reject new provider work first. Existing workers keep draining during
         # manager.stop(), and channel transports remain alive until that drain
@@ -323,6 +327,7 @@ class ChannelService:
         if stop_errors:
             raise ExceptionGroup("one or more channels failed to stop", stop_errors)
 
+        self._stopping = False
         logger.info("ChannelService stopped")
 
     def _load_channel_config(self, name: str) -> dict[str, Any] | None:
@@ -484,6 +489,10 @@ class ChannelService:
 
     async def _start_channel(self, name: str, config: dict[str, Any]) -> bool:
         """Instantiate and start a single channel."""
+        if self._stopping or not self._running:
+            logger.warning("Refusing to start %s while ChannelService is stopped or stopping", name)
+            return False
+
         import_path = _CHANNEL_REGISTRY.get(name)
         if not import_path:
             logger.warning("Unknown channel type")
@@ -528,9 +537,21 @@ class ChannelService:
                 config["seen_event_store_path"] = await asyncio.to_thread(_default_seen_store_path)
             if self._connection_repo is not None:
                 config["connection_repo"] = self._connection_repo
+            # A stop may have started while this coroutine was suspended in
+            # pre-start I/O above. Fence publication again immediately before
+            # the new instance becomes service-owned.
+            if self._stopping or not self._running:
+                logger.warning("Refusing to publish %s channel while ChannelService is stopped or stopping", name)
+                return False
             channel = channel_cls(bus=self.bus, config=config)
             self._channels[name] = channel
             await channel.start()
+            # A concurrent service stop may finish while channel.start() is
+            # suspended, removing this instance before it subscribes. Drain
+            # the late-started transport rather than leaving an orphan.
+            if self._stopping or not self._running:
+                await self._stop_and_discard_channel(name, channel)
+                return False
             if not channel.is_running:
                 logger.error("Channel did not enter a running state after start()")
                 await self._stop_and_discard_channel(name, channel)

backend/tests/blocking_io/test_buzz_channel_seen_events.py

@@ -219,6 +219,7 @@ async def test_start_channel_resolves_seen_store_path_off_the_loop(tmp_path: Pat
     # path there. The test mirrors that boundary so only ``_start_channel``'s own
     # resolution is exercised on the loop.
     service = await asyncio.to_thread(ChannelService, channels_config={})
+    service._running = True  # Direct-start fixture models an active service.
     started = await service._start_channel("buzz", {"relay_url": "wss://buzz.example.com", "private_key": "unused-by-this-test"})
 
     # ``_start_channel`` keeps a blanket ``except Exception`` around the whole

backend/tests/test_channel_service_restart_stop_fence.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from app.channels import service as channel_service
+from app.channels.service import ChannelService
+from deerflow import reflection
+
+
+class _FakeChannel:
+    def __init__(self, *, bus: object, config: dict[str, object]) -> None:
+        self.is_running = False
+
+    async def start(self) -> None:
+        self.is_running = True
+
+    async def stop(self) -> None:
+        self.is_running = False
+
+
+@pytest.mark.asyncio
+async def test_restart_is_fenced_while_service_shutdown_is_in_progress(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    service = ChannelService(channels_config={"fake": {"enabled": True}})
+    service._running = True
+    stop_entered = asyncio.Event()
+    release_stop = asyncio.Event()
+
+    # Give stop() a real, already-owned transport to await. It snapshots
+    # _channels before awaiting this stop; a late restart must not publish a
+    # second transport that is absent from the snapshot.
+    existing = _FakeChannel(bus=service.bus, config={})
+    existing.is_running = True
+
+    async def blocking_channel_stop() -> None:
+        stop_entered.set()
+        await release_stop.wait()
+        existing.is_running = False
+
+    async def manager_stop() -> None:
+        pass
+
+    monkeypatch.setattr(existing, "stop", blocking_channel_stop)
+    monkeypatch.setattr(service.manager, "stop", manager_stop)
+    monkeypatch.setitem(channel_service._CHANNEL_REGISTRY, "fake", "tests.fake:FakeChannel")
+    monkeypatch.setattr(reflection, "resolve_class", lambda *_args, **_kwargs: _FakeChannel)
+    service._channels["existing"] = existing
+
+    stop_task = asyncio.create_task(service.stop())
+    await asyncio.wait_for(stop_entered.wait(), timeout=1)
+
+    try:
+        restarted = await service.restart_channel("fake", reload_config=False)
+    finally:
+        release_stop.set()
+        await asyncio.wait_for(stop_task, timeout=1)
+
+    assert restarted is False
+    assert service._channels == {}
+    assert service._running is False
+    assert service._stopping is False
+
+
+@pytest.mark.asyncio
+async def test_channel_cannot_publish_after_stop_finishes_during_prestart_io(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    service = ChannelService(channels_config={"buzz": {"enabled": True}})
+    service._running = True
+    entered_io = asyncio.Event()
+    resume_io = asyncio.Event()
+
+    async def held_to_thread(_function: object) -> str:
+        entered_io.set()
+        await resume_io.wait()
+        return "/tmp/buzz_seen_events.json"
+
+    async def manager_stop() -> None:
+        pass
+
+    monkeypatch.setattr(channel_service.asyncio, "to_thread", held_to_thread)
+    monkeypatch.setattr(service.manager, "stop", manager_stop)
+    monkeypatch.setitem(channel_service._CHANNEL_REGISTRY, "buzz", "tests.fake:FakeChannel")
+    monkeypatch.setattr(reflection, "resolve_class", lambda *_args, **_kwargs: _FakeChannel)
+
+    start_task = asyncio.create_task(service._start_channel("buzz", {"enabled": True}))
+    await asyncio.wait_for(entered_io.wait(), timeout=1)
+    await service.stop()
+    assert service._stopping is False
+    assert service._running is False
+
+    resume_io.set()
+    assert await asyncio.wait_for(start_task, timeout=1) is False
+    assert service._channels == {}
+
+
+@pytest.mark.asyncio
+async def test_channel_start_finishing_after_service_stop_is_drained(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    service = ChannelService(channels_config={"fake": {"enabled": True}})
+    service._running = True
+    start_entered = asyncio.Event()
+    release_start = asyncio.Event()
+    started = []
+
+    class LateChannel(_FakeChannel):
+        async def start(self) -> None:
+            start_entered.set()
+            await release_start.wait()
+            self.is_running = True
+            service.bus.subscribe_outbound(self.on_outbound)
+            started.append(self)
+
+        async def on_outbound(self, _message: object) -> None:
+            pass
+
+        async def stop(self) -> None:
+            service.bus.unsubscribe_outbound(self.on_outbound)
+            self.is_running = False
+
+    async def manager_stop() -> None:
+        pass
+
+    monkeypatch.setattr(service.manager, "stop", manager_stop)
+    monkeypatch.setitem(channel_service._CHANNEL_REGISTRY, "fake", "tests.fake:LateChannel")
+    monkeypatch.setattr(reflection, "resolve_class", lambda *_args, **_kwargs: LateChannel)
+
+    start_task = asyncio.create_task(service._start_channel("fake", {"enabled": True}))
+    await asyncio.wait_for(start_entered.wait(), timeout=1)
+    await service.stop()
+    assert service._channels == {}
+    release_start.set()
+
+    assert await asyncio.wait_for(start_task, timeout=1) is False
+    assert len(started) == 1
+    assert started[0].is_running is False
+    assert service._channels == {}
+    assert service.bus._outbound_listeners == []

backend/tests/test_qq_lifecycle.py

@@ -341,6 +341,7 @@ async def test_service_registers_starts_and_disposes_qq(tmp_path, monkeypatch):
         AsyncMock(return_value={"url": "wss://api.sgroup.qq.com/websocket/"}),
     )
     monkeypatch.setattr(qq.QQChannel, "_run_connection", connected)
+    service._running = True  # Direct-start fixture models an active service.
     try:
         assert await service._start_channel("qq", config)
         channel = service.get_channel("qq")