diff --git a/README.md b/README.md index 6a2ff036b5..c134626344 100644 --- a/README.md +++ b/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 diff --git a/backend/docs/STREAMING.md b/backend/docs/STREAMING.md index 9355d6686c..a53e250cc6 100644 --- a/backend/docs/STREAMING.md +++ b/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]`。 diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index e1945b7d81..515ae37172 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py b/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py index c11d31ee28..c784e85861 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py +++ b/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 diff --git a/backend/tests/test_memory_stream_payload_isolation.py b/backend/tests/test_memory_stream_payload_isolation.py new file mode 100644 index 0000000000..b07e5618fc --- /dev/null +++ b/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()