fix(streaming): isolate in-memory replay payloads (#7644)

* fix(streaming): isolate in-memory replay payloads

* docs(streaming): explain sentinel identity during replay

* docs(streaming): document replay payload ownership

---------

Co-authored-by: yinx65588-boop <yinx65588-boop@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -722,6 +722,10 @@ DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser
 > normal `end` signal. A new reader whose initial attachment loses the race with
 > cleanup receives a gap so it can recover the saved result.
 >
+> In-memory stream events retain the payload as it was published. Reusing a
+> producer's data object or editing a received event does not change another
+> subscriber's output or reconnect replay.
+>
 > During SSE replay-gap recovery, the Web UI preserves a reconnect pointer claimed by another run while durable state is loading. An older recovery cannot overwrite that pointer when it rejoins its own stream. Even a stale foreign pointer is preserved, so the next reload may first check that run before active-run recovery rejoins, adding a reconnect cycle.
 >
 > Store-only SSE and `/wait` consumers can finish after a heartbeat when run-manager

backend/docs/STREAMING.md

@@ -143,7 +143,7 @@ sequenceDiagram
 关键组件：
 
 - `runtime/runs/worker.py::run_agent` — 在 `asyncio.Task` 里跑 `agent.astream()`，把每个 chunk 通过 `serialize(chunk, mode=mode)` 转成 JSON，再 `bridge.publish()`。
-- `runtime/stream_bridge` — 抽象 Queue。`publish/subscribe` 解耦生产者和消费者，支持 `Last-Event-ID` 重连、心跳、多订阅者 fan-out。`stream_bridge.heartbeat_interval_seconds`（默认 15 秒）是 bridge 实例的默认心跳周期，统一作用于 Gateway SSE、`/wait` 和内部 channel watcher；显式传给 `subscribe()` 的值仍可覆盖单次订阅。Memory 和 Redis 都只保留 `queue_maxsize` 条数据事件；游标早于保留水位线时返回 `StreamGap`，不会从当前最早事件静默部分重放。Redis backend 会在写入数据或首个 END 时刷新 retained stream key TTL；启动恢复与基于 worker lease 的周期恢复共用 Gateway stream terminalization 路径：`RunManager` 先将 orphan run 持久化为 `error` 并写入显式的 `stop_reason=orphan_recovered`，随后 Gateway 发布 `END_SENTINEL` 并安排 stream cleanup。周期扫描、逐行状态写入和 Gateway callback 作为一个受监督的 single-flight 后台 task 执行；慢任务不会堆积，也不会阻塞唯一的 lease heartbeat。shutdown 优先收敛活跃 run，再处理恢复 task；尚未执行的延迟 stream cleanup 会改为立即删除。只有 runtime `yield` 前、无并发请求的启动恢复会把最新受影响 thread 标记为 error；周期恢复不做非原子的 thread 投影。普通 durable terminal status 本身仍不足以证明尾帧已发布；Redis producer 在 durable admission 前取得短租约，经 request task、worker、finalizer 转移并保留到 END。只有显式 `orphan_recovered` 或 durable terminal 加已失效的 tracked producer lease 才能在 heartbeat 上触发 END fallback；lease loss 会阻止后续 stream 写入，active 或 untracked producer 继续等待，避免截断尾帧。Redis END 通过同一个 Lua 原子完成 lease 校验、既有 END 检查和必要写入，因此重复调用、owner/recovery 竞争和 ACK 丢失重试至多追加一个 END；当 END 已位于 stream 尾部时，后续事件写入会被拒绝；旧 stream 中 END 后已有迟到数据时，仍允许继续写入数据。cleanup 后无 owner 的迟到 END 重试是成功 no-op，不会复活已删除的 stream。普通事件只做 O(1) 尾部 END 检查。`stream_bridge.legacy_end_scan_enabled` 默认 `true`：尾部不是 END 时扫描整个保留窗口，包括正常 run 的首次 END；原子 Lua 的耗时随保留条目数量和 payload 总字节数增长，期间会阻塞其他 Redis 客户端。只有全部旧 writer 停止且旧 replay buffer 已过期或清除后，才可设为 `false` 并重启所有 Gateway，使 END 检查也只看尾部。TTL 为 0 时必须显式清除旧 buffer。TTL 仍是 Redis 内存和故障安全网，不是正常的 subscriber 终止机制。
+- `runtime/stream_bridge` — 抽象 Queue。`publish/subscribe` 解耦生产者和消费者，支持 `Last-Event-ID` 重连、心跳、多订阅者 fan-out。`stream_bridge.heartbeat_interval_seconds`（默认 15 秒）是 bridge 实例的默认心跳周期，统一作用于 Gateway SSE、`/wait` 和内部 channel watcher；显式传给 `subscribe()` 的值仍可覆盖单次订阅。Memory 和 Redis 都只保留 `queue_maxsize` 条数据事件。`MemoryStreamBridge` 在分配事件 ID 前深拷贝发布的 payload，并在 stream Condition 外为每个订阅者返回独立深拷贝，因此生产者复用 payload 或订阅者修改数据都不会影响保留帧及重连重放；心跳与 END sentinel 保持对象身份。游标早于保留水位线时返回 `StreamGap`，不会从当前最早事件静默部分重放。Redis backend 会在写入数据或首个 END 时刷新 retained stream key TTL；启动恢复与基于 worker lease 的周期恢复共用 Gateway stream terminalization 路径：`RunManager` 先将 orphan run 持久化为 `error` 并写入显式的 `stop_reason=orphan_recovered`，随后 Gateway 发布 `END_SENTINEL` 并安排 stream cleanup。周期扫描、逐行状态写入和 Gateway callback 作为一个受监督的 single-flight 后台 task 执行；慢任务不会堆积，也不会阻塞唯一的 lease heartbeat。shutdown 优先收敛活跃 run，再处理恢复 task；尚未执行的延迟 stream cleanup 会改为立即删除。只有 runtime `yield` 前、无并发请求的启动恢复会把最新受影响 thread 标记为 error；周期恢复不做非原子的 thread 投影。普通 durable terminal status 本身仍不足以证明尾帧已发布；Redis producer 在 durable admission 前取得短租约，经 request task、worker、finalizer 转移并保留到 END。只有显式 `orphan_recovered` 或 durable terminal 加已失效的 tracked producer lease 才能在 heartbeat 上触发 END fallback；lease loss 会阻止后续 stream 写入，active 或 untracked producer 继续等待，避免截断尾帧。Redis END 通过同一个 Lua 原子完成 lease 校验、既有 END 检查和必要写入，因此重复调用、owner/recovery 竞争和 ACK 丢失重试至多追加一个 END；当 END 已位于 stream 尾部时，后续事件写入会被拒绝；旧 stream 中 END 后已有迟到数据时，仍允许继续写入数据。cleanup 后无 owner 的迟到 END 重试是成功 no-op，不会复活已删除的 stream。普通事件只做 O(1) 尾部 END 检查。`stream_bridge.legacy_end_scan_enabled` 默认 `true`：尾部不是 END 时扫描整个保留窗口，包括正常 run 的首次 END；原子 Lua 的耗时随保留条目数量和 payload 总字节数增长，期间会阻塞其他 Redis 客户端。只有全部旧 writer 停止且旧 replay buffer 已过期或清除后，才可设为 `false` 并重启所有 Gateway，使 END 检查也只看尾部。TTL 为 0 时必须显式清除旧 buffer。TTL 仍是 Redis 内存和故障安全网，不是正常的 subscriber 终止机制。
 - `stream_bridge.producer_lease_recovery_enabled` 是上述 producer-lease fallback 的 restart-required activate gate。升级时先保持 `false` 完成所有 writer 的滚动发布并排空旧 run，再在所有 Gateway 上同时设为 `true` 并重启；mixed-version 阶段不得启用。
 - `app/gateway/services.py::sse_consumer` — 从 bridge 订阅，格式化为 SSE wire 帧。
 - `runtime/serialization.py::serialize` — mode-aware 序列化；`messages` mode 下 `serialize_messages_tuple` 把 `(chunk, metadata)` 转成 `[chunk.model_dump(), metadata]`。

backend/packages/harness/deerflow/runtime/AGENTS.md

@@ -54,6 +54,12 @@ successful no-op; a live owner's write retains its existing missing-buffer
 
 ### Memory Stream Replay
 
+Memory publication snapshots the payload before allocating an event ID. Each
+subscriber receives its own deep copy outside the stream Condition, so producer
+reuse and reader edits cannot rewrite retained frames. Heartbeat/END sentinels
+keep identity. `tests/test_memory_stream_payload_isolation.py` covers nested
+payloads and Worker-to-Gateway SSE/reconnect metadata, including subgraphs.
+
 Memory keeps the exact last evicted event ID as bounded per-run metadata. A
 reconnect from that known predecessor replays all retained successors and END,
 matching a live subscriber at the same position. Older or unverifiable evicted

backend/packages/harness/deerflow/runtime/stream_bridge/memory.py

@@ -7,6 +7,7 @@ import logging
 import re
 import time
 from collections.abc import AsyncIterator
+from copy import deepcopy
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -116,6 +117,7 @@ class MemoryStreamBridge(StreamBridge):
     # -- StreamBridge API ------------------------------------------------------
 
     async def publish(self, run_id: str, event: str, data: Any) -> None:
+        data = deepcopy(data)
         stream = self._get_or_create_stream(run_id)
         entry = StreamEvent(id=self._next_id(run_id), event=event, data=data)
         async with stream.condition:
@@ -191,7 +193,8 @@ class MemoryStreamBridge(StreamBridge):
             if entry is END_SENTINEL:
                 yield END_SENTINEL
                 return
-            yield entry
+            # Consumers identify sentinels with `is`; never deep-copy them.
+            yield deepcopy(entry) if isinstance(entry, StreamEvent) and entry is not HEARTBEAT_SENTINEL else entry
             if should_stop:
                 return
 

backend/tests/test_memory_stream_payload_isolation.py

@@ -0,0 +1,104 @@
+from copy import deepcopy
+
+import httpx
+import pytest
+from fastapi import FastAPI, Request
+from langgraph_sdk.sse import SSEDecoder, aiter_lines_raw
+from starlette.responses import StreamingResponse
+
+from deerflow.runtime.stream_bridge import END_SENTINEL, MemoryStreamBridge
+
+
+@pytest.mark.asyncio
+async def test_publish_captures_each_nested_payload_before_producer_reuses_it():
+    bridge = MemoryStreamBridge()
+    payload = {"messages": [{"content": [{"text": "first"}]}]}
+    first = deepcopy(payload)
+    try:
+        await bridge.publish("run", "values", payload)
+        payload["messages"][0]["content"][0]["text"] = "second"
+        second = deepcopy(payload)
+        await bridge.publish("run", "values", payload)
+        payload["messages"].clear()
+        await bridge.publish_end("run")
+
+        events = [entry async for entry in bridge.subscribe("run")]
+        assert [entry.data for entry in events[:-1]] == [first, second]
+        assert events[-1] is END_SENTINEL
+    finally:
+        await bridge.close()
+
+
+@pytest.mark.asyncio
+async def test_subscriber_mutation_does_not_reach_another_reader_or_replay():
+    bridge = MemoryStreamBridge()
+    expected = {"tool_calls": [{"args": {"paths": ["original.txt"]}}]}
+    try:
+        await bridge.publish("run", "metadata", {"run_id": "run"})
+        await bridge.publish("run", "updates", deepcopy(expected))
+        await bridge.publish_end("run")
+        initial = [entry async for entry in bridge.subscribe("run")]
+        initial[1].data["tool_calls"][0]["args"]["paths"].append("reader-edit.txt")
+
+        other = [entry async for entry in bridge.subscribe("run")]
+        assert other[1].data == expected
+        replay = [entry async for entry in bridge.subscribe("run", last_event_id=initial[0].id)]
+        assert replay[0].id == initial[1].id
+        assert replay[0].event == "updates"
+        assert replay[0].data == expected
+        assert other[-1] is replay[-1] is END_SENTINEL
+        assert replay[0].data is not other[1].data
+    finally:
+        await bridge.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("namespace", [(), ("child:task",)], ids=["root", "subgraph"])
+async def test_worker_metadata_stays_at_publication_time_in_gateway_sse(namespace):
+    from langchain_core.messages import AIMessageChunk
+
+    from app.gateway.services import sse_consumer
+    from deerflow.runtime import RunManager, RunStatus
+    from deerflow.runtime.runs.worker import _publish_stream_item
+    from deerflow.runtime.serialization import serialize
+
+    bridge = MemoryStreamBridge()
+    manager = RunManager()
+    record = await manager.create("snapshot-sse")
+    app = FastAPI()
+
+    @app.get("/stream")
+    async def stream(request: Request):
+        return StreamingResponse(sse_consumer(bridge, record, request, manager, apply_on_disconnect=False), media_type="text/event-stream")
+
+    try:
+        expected = []
+        metadata = {"langgraph_node": "model", "tags": [], "langgraph_step": 1}
+        for step in (1, 2):
+            metadata["langgraph_step"] = step
+            metadata["tags"].append(f"step-{step}")
+            chunk = (AIMessageChunk(content=f"part-{step}", id="answer"), metadata)
+            expected.append(deepcopy(serialize(chunk, mode="messages")))
+            await _publish_stream_item(bridge=bridge, run_id=record.run_id, mode="messages", chunk=chunk, namespace=namespace, file_tool_chunk_batcher=None, subagent_events=None)
+        metadata["tags"].clear()
+        await manager.set_status(record.run_id, RunStatus.success)
+        await bridge.publish_end(record.run_id)
+
+        async def decode(response):
+            assert response.status_code == 200
+            decoder = SSEDecoder()
+            frames = []
+            async for line in aiter_lines_raw(response):
+                if (frame := decoder.decode(line)) is not None:
+                    frames.append(frame)
+            return frames
+
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            frames = await decode(await client.get("/stream"))
+            event_name = "messages" + ("|" + "|".join(namespace) if namespace else "")
+            assert [(frame.event, frame.data) for frame in frames] == [(event_name, data) for data in expected] + [("end", None)]
+            replay = await decode(await client.get("/stream", headers={"Last-Event-ID": frames[0].id}))
+            assert [(frame.id, frame.event, frame.data) for frame in replay] == [(frame.id, frame.event, frame.data) for frame in frames[1:]]
+    finally:
+        await manager.shutdown()
+        await bridge.close()