fix(stream_bridge): add stream_exists to MemoryStreamBridge, fixing SSE hang on reconnect after cleanup (#4071)

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.

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:

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."""