From ad9ec65c6f5b9ecfe069b5433fbcb0be595fc572 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?=
 <76432572+nankingjing@users.noreply.github.com>
Date: Sat, 11 Jul 2026 22:46:10 +0800
Subject: [PATCH] fix(stream_bridge): add stream_exists to MemoryStreamBridge,
 fixing SSE hang on reconnect after cleanup (#4071)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

MemoryStreamBridge is the DEFAULT stream backend. Its subscribe() creates a
fresh ended=False stream on every call via _get_or_create_stream(), but the
Gateway's reconnect guard _terminal_record_stream_missing only detects a
missing stream on bridges that expose stream_exists — and MemoryStreamBridge
did NOT define it (only RedisStreamBridge did). After worker cleanup pops the
stream (~60s post-run), a browser SSE reconnect or POST /wait hits
subscribe() -> creates a zombie stream -> yields heartbeat forever without
ever sending END_SENTINEL. The UI spinner never resolves and the coroutine
pins a server-side connection/request until external timeout.

Add the missing stream_exists method, mirroring RedisStreamBridge.
---
 .../deerflow/runtime/stream_bridge/memory.py    |  4 ++++
 backend/tests/test_stream_bridge.py             | 17 +++++++++++++++++
 2 files changed, 21 insertions(+)

diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py b/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py
index 0a03be8af..2690c8f4c 100644
--- a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py
+++ b/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py
@@ -86,6 +86,10 @@ class MemoryStreamBridge(StreamBridge):
             )
         return stream.start_offset
 
+    async def stream_exists(self, run_id: str) -> bool:
+        """Return whether the in-process event log still has data for *run_id*."""
+        return run_id in self._streams
+
     # -- StreamBridge API ------------------------------------------------------
 
     async def publish(self, run_id: str, event: str, data: Any) -> None:
diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py
index bf710b436..b5c413e53 100644
--- a/backend/tests/test_stream_bridge.py
+++ b/backend/tests/test_stream_bridge.py
@@ -176,6 +176,23 @@ async def test_cleanup(bridge: MemoryStreamBridge):
     assert run_id not in bridge._counters
 
 
+@pytest.mark.anyio
+async def test_stream_exists_reports_cleanup(bridge: MemoryStreamBridge):
+    """Callers can detect when the in-process event log has been cleaned up.
+
+    Before cleanup a completed run's retained history still exists; after
+    cleanup ``stream_exists`` reports False so a reconnecting subscriber does
+    not hang waiting on a stream whose data is already gone.
+    """
+    run_id = "run-post-cleanup"
+    await bridge.publish(run_id, "event-1", {"n": 1})
+    await bridge.publish_end(run_id)
+
+    assert await bridge.stream_exists(run_id) is True
+    await bridge.cleanup(run_id)
+    assert await bridge.stream_exists(run_id) is False
+
+
 @pytest.mark.anyio
 async def test_history_is_bounded():
     """Retained history should be bounded by queue_maxsize."""