fix(streaming): make Redis stream END publication idempotent (#7572)

* fix(streaming): make Redis stream END publication idempotent

* fix(streaming): align Redis test fake with Lua END checks

Remove the unreachable owned missing-stream branch, mirror tail-only data checks in both fake script paths, and pin legacy END and owned missing-buffer behavior against real Redis.

* fix(streaming): gate legacy END scans after writer rollout

* fix(helm): synchronize embedded config version with schema

* test(runtime): use real stream bridge config in close coverage

Keep new configuration defaults in lifecycle fixtures and surface owner-task failures instead of masking them as event timeouts. Drain every helper task and cover context entry/close failures.

---------

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

backend/app/gateway/AGENTS.md

@@ -211,7 +211,7 @@ or `deployment.multi_instance: true`.
 - `DELETE /api/threads/{id}` holds a durable `delete` reservation for the whole cleanup and removes thread filesystem data (off-loop, drained on cancel), checkpoints, owner-scoped historical runs, run events, feedback and thread metadata best-effort. Historical-run cleanup removes only `operation_kind="run"` rows, so internal thread-operation rows — including the reservation protecting that request — stay durable until `RunManager.reserve_thread_operation()` exits. This cleans persisted rows; preventing an already-admitted write from re-creating state for a deleted thread is a separate thread-incarnation contract, not part of this cleanup.
 - Creator SSE and `/wait` apply `on_disconnect` only to confirmed request disconnect or cancellation; bridge failures propagate without cancelling the run. `/wait` consumes bridge END; errors return SDK `__error__`, never stale state.
 - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy.
-- Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. Terminal `orphan_recovered` and `scheduled_task_orphan_recovered` rows permit heartbeat fallback because their producer is gone; ordinary terminal rows still await END. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection.
+- Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on data and first-END writes) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. Terminal `orphan_recovered` and `scheduled_task_orphan_recovered` rows permit heartbeat fallback because their producer is gone; ordinary terminal rows still await END. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection.
 - Thread-scoped create, stream and wait share an optional owner/thread-scoped `Idempotency-Key` (internal callers: owner-only, cross-thread reuse 409); never pass the raw external key to `RunManager`'s process-wide index. Stores resolve an existing key before `interrupt`/`rollback`, so a retry reuses rather than interrupts its run. Reuse requires matching input, assistant and conversation references; keyed resume stores a versioned strict typed-JSON SHA-256 (`true`, `1` and `1.0` differ) in the private `runs.idempotency_request_json` column, never public `kwargs`; ordinary input needs no extra identity. New readers accept the exact old-writer resume identity shape from `kwargs`, every response strips it; identity-less legacy resume fails closed. Validate strict resume JSON with or without a key; persist the identity only for keyed resume admission. Missing headers stay non-idempotent; stateless `/api/runs/*` is outside this contract (it creates a thread before admission).
 - For a rolling upgrade, route every keyed resume submission/retry on create/stream/wait only to upgraded workers; otherwise pause that traffic or replace old workers first. Old workers ignore the private identity and can reuse `approve` for `deny` with null input. Old-reader schema compatibility does not protect admission. Upgraded workers reject identity-less legacy resumes with 409. Rollout details: `README.md`, `backend/docs/API.md`.
 - Reused `/wait` returns durable status/error, never the latest thread checkpoint; `task is None` is not completion, and store-backed fields must refresh after waiting. Capture reuse before awaiting because the cached record's `idempotency_reused` flag is sticky. A creating `/runs/stream` retry whose terminal stream is gone opts into `gap`/`stream_replay_gap` via `emit_gap_on_missing_stream`, while observer joins and default/stateless consumers emit `end`; never infer this from disconnect policy or the sticky flag. A reused running `store_only` record on a process-local bridge returns 409 without `Retry-After`.

backend/docs/CONFIGURATION.md

@@ -763,6 +763,7 @@ Notes:
 - Set `multi_instance: true` (or export `DEER_FLOW_MULTI_INSTANCE=1`, which lets deploy tooling such as a Helm chart set it from its replica count) on every instance that shares one database. Startup then enforces the same prerequisites as `GATEWAY_WORKERS > 1`: `database.backend: postgres`, `run_events.backend: db`, `run_ownership.heartbeat_enabled: true`, and a Redis stream bridge (`stream_bridge.type: redis` or `DEER_FLOW_STREAM_BRIDGE_REDIS_URL`). It refuses an explicit `sandbox.ownership.type: memory`, process-local browser tools, and `scheduler.enabled: true` without `scheduler.multi_instance: true`.
 - `DEER_FLOW_MULTI_INSTANCE` treats blank, `0`, `false`, `no` and `off` as "not declared"; any other value declares a multi-instance deployment, so a typo fails closed.
 - `GET /health/ready` pings the Redis stream bridge on every probe and answers 503 (`stream_bridge: unreachable`) while Redis is down, so the orchestrator drains that instance instead of routing it runs it cannot publish or stream; the memory bridge reports `not_configured`. When `sandbox.provisioner_url` is set, the body also carries the provisioner's `/health` verdict (`provisioner: ok|unreachable`), which never changes the status code because every instance shares one provisioner. That probe follows the sandbox clients' proxy policy: loopback, private, link-local and cluster-local provisioner addresses bypass `HTTP_PROXY`, external hosts keep the environment's proxy settings.
+- `stream_bridge.legacy_end_scan_enabled` defaults to `true` for Redis: when the tail is not END, END publication scans all retained entries, including a normal run's first END. Cost grows with the retained entry count and total payload bytes inside atomic Lua, blocking other Redis clients; `queue_maxsize` bounds entries, not bytes. Stop every old writer and expire or remove every legacy replay buffer before setting this to `false` on all Gateways and restarting them. With `stream_ttl_seconds: 0`, remove those buffers explicitly. Disabled mode detects END only at the tail, so both data and END checks have constant cost relative to retained history; it cannot deduplicate an old mid-stream END. No performance ceiling is implied by the default entry cap.
 - `stream_bridge.producer_lease_recovery_enabled` is a restart-required activation gate for recovering a terminal Redis stream whose producer died before publishing END. Keep it `false` during the binary rollout: upgraded writers already publish producer markers, while old writers do not. After every old Gateway has stopped and its runs have drained, set it to `true` on every instance and restart them together. Enabling it while an old worker may still finalize a run can make a new consumer mistake that untracked legacy producer for completed history.
 - The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, the `auth.local.throttle_storage: memory` warning, and the WeChat QR-login guard, which otherwise only look at the worker count.
 - Login lockouts are shared through the database: `auth.local.throttle_storage` (startup-only) defaults to `auto`, which keeps the per-IP failed-login counters for `POST /api/v1/auth/login/local` in the `login_throttle` table whenever `database.backend` is `sqlite` or `postgres`, so every instance enforces one `max_login_attempts` limit per client IP and a lockout on one instance holds on all of them. `memory` keeps the historical per-process counter (N instances give an attacker N × `max_login_attempts` guesses; a declared multi-instance deployment logs a warning), and `db` forces the table (it falls back to memory with a warning when the database backend is `memory`, and refuses to start when the configured database's engine is unavailable; `auto` falls back with a warning in that case). `max_login_attempts` and `lockout_seconds` stay live-read.

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 会在每次 `publish()` / `publish_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 继续等待，避免截断尾帧。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` 条数据事件；游标早于保留水位线时返回 `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/config/stream_bridge_config.py

@@ -54,6 +54,14 @@ class StreamBridgeConfig(BaseModel):
             "even if cleanup never runs. Set to 0 to disable. Only applies to the redis bridge."
         ),
     )
+    legacy_end_scan_enabled: bool = Field(
+        default=True,
+        description=(
+            "Scan retained Redis history when END is not the tail, including first END publication. "
+            "Disable only after all old writers stop and all legacy replay buffers expire or are removed. "
+            "Disabling makes END detection tail-only. Restart every Gateway after changing; Redis only."
+        ),
+    )
     producer_lease_recovery_enabled: bool = Field(
         default=False,
         description=("Allow terminal Redis consumers to recover a missing END from an expired producer lease. Enable only after every Gateway writer has been upgraded and pre-upgrade runs have drained."),

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

@@ -42,6 +42,21 @@ Memory/Redis bridges keep startup-only `stream_bridge.heartbeat_interval_seconds
 
 Reconnect cursor parsing requires ASCII digits and must treat integer-conversion failures as malformed input. Redis cursors additionally require unsigned 64-bit components; use the same parser before sending a cursor to `XREAD` and when detecting replay gaps. Preserve malformed-cursor policy: memory replays retained events, Redis tails active streams and replays ended streams through `end`. `tests/test_stream_bridge.py` directly tests both parsers offline, including oversized sequences, Redis component overflow, non-ASCII digits, and valid uint64 boundaries; integration tests also exercise malformed cursors against real Redis.
 
+### Redis Stream END Publication
+
+Keep the producer-token check first, then END detection, then `XADD` in one Lua
+script. Ordinary event publication checks only the stream tail (`XREVRANGE ...
+COUNT 1`), so token streaming remains O(1) as retained history grows. END
+publication checks the tail first. With `legacy_end_scan_enabled` (default true),
+a non-END tail triggers a full retained-window scan even on first END; its
+payload-byte cost blocks Redis. Disable only after old writers stop and legacy
+buffers expire or are removed; restart all Gateways together. Disabled mode is
+tail-only and cannot deduplicate an old mid-stream END. Do not add an unbounded boolean state key for
+this fast path; cleanup, expiry, and a new producer generation would then need a
+separate lifecycle contract. An unowned recovery END for a missing stream is a
+successful no-op; a live owner's write retains its existing missing-buffer
+`XADD` behavior. Test fakes must mirror the scan switch and both tail fast paths. Coverage: `tests/test_stream_bridge.py`.
+
 ### Memory Stream Replay
 
 Memory keeps the exact last evicted event ID as bounded per-run metadata. A

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

@@ -85,6 +85,7 @@ async def make_stream_bridge(app_config: AppConfig | None = None) -> AsyncIterat
             heartbeat_interval=config.heartbeat_interval_seconds,
             max_connections=config.max_connections,
             stream_ttl_seconds=config.stream_ttl_seconds,
+            legacy_end_scan_enabled=config.legacy_end_scan_enabled,
             producer_lease_recovery_enabled=getattr(
                 config,
                 "producer_lease_recovery_enabled",

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

@@ -40,6 +40,10 @@ _KIND_EVENT = "event"
 _KIND_END = "end"
 _REDIS_STREAM_ID_RE = re.compile(r"[0-9]+(-[0-9]+)?")
 _MAX_STREAM_ID_COMPONENT = 2**64 - 1
+# Lua XADD scripts return one of these integers before any event ID.
+_XADD_BLOCKED = 0
+_XADD_ALREADY_ENDED = 1
+_XADD_STREAM_MISSING = 2
 
 # Batch size for ``XREAD``. Reading more than one entry per round-trip collapses
 # a large ``Last-Event-ID`` replay into far fewer calls; live tailing still
@@ -71,9 +75,44 @@ redis.call("expire", KEYS[1], ARGV[2])
 if tonumber(ARGV[3]) > 0 then redis.call("expire", KEYS[2], ARGV[3]) end
 return 1
 """
-_COMPARE_XADD_SCRIPT = """
-if redis.call("get", KEYS[1]) ~= ARGV[1] then return false end
-if redis.call("get", KEYS[2]) ~= ARGV[1] then return false end
+_END_MARKER_CHECK_LUA = """
+local function entry_is_end(fields)
+  for index = 1, #fields, 2 do
+    if fields[index] == "kind" and fields[index + 1] == "end" then
+      return true
+    end
+  end
+  return false
+end
+
+local function tail_is_end(stream_key)
+  local entries = redis.call("xrevrange", stream_key, "+", "-", "COUNT", 1)
+  return #entries > 0 and entry_is_end(entries[1][2])
+end
+
+local function has_end_marker(stream_key, legacy_scan_enabled)
+  if tail_is_end(stream_key) then return true end
+  if legacy_scan_enabled ~= "1" then return false end
+  -- Compatibility path for retained streams written by the old buggy writer:
+  -- an END may precede later entries. The first END on a normal stream also
+  -- reaches this scan; operators may disable it only after legacy buffers drain.
+  local entries = redis.call("xrange", stream_key, "-", "+")
+  for _, entry in ipairs(entries) do
+    if entry_is_end(entry[2]) then return true end
+  end
+  return false
+end
+"""
+_COMPARE_XADD_SCRIPT = (
+    _END_MARKER_CHECK_LUA
+    + """
+if redis.call("get", KEYS[1]) ~= ARGV[1] then return 0 end
+if redis.call("get", KEYS[2]) ~= ARGV[1] then return 0 end
+if ARGV[4] == "end" then
+  if has_end_marker(KEYS[3], ARGV[7]) then return 1 end
+elseif tail_is_end(KEYS[3]) then
+  return 1
+end
 local event_id = redis.call(
   "xadd", KEYS[3], "MAXLEN", "=", ARGV[2], "*",
   "kind", ARGV[4], "event", ARGV[5], "data", ARGV[6]
@@ -81,8 +120,17 @@ local event_id = redis.call(
 if tonumber(ARGV[3]) > 0 then redis.call("expire", KEYS[3], ARGV[3]) end
 return event_id
 """
-_XADD_IF_NO_ACTIVE_PRODUCER_SCRIPT = """
-if redis.call("exists", KEYS[1]) == 1 then return false end
+)
+_XADD_IF_NO_ACTIVE_PRODUCER_SCRIPT = (
+    _END_MARKER_CHECK_LUA
+    + """
+if redis.call("exists", KEYS[1]) == 1 then return 0 end
+if ARGV[3] == "end" and redis.call("exists", KEYS[2]) == 0 then return 2 end
+if ARGV[3] == "end" then
+  if has_end_marker(KEYS[2], ARGV[6]) then return 1 end
+elseif tail_is_end(KEYS[2]) then
+  return 1
+end
 local event_id = redis.call(
   "xadd", KEYS[2], "MAXLEN", "=", ARGV[1], "*",
   "kind", ARGV[3], "event", ARGV[4], "data", ARGV[5]
@@ -90,6 +138,7 @@ local event_id = redis.call(
 if tonumber(ARGV[2]) > 0 then redis.call("expire", KEYS[2], ARGV[2]) end
 return event_id
 """
+)
 _CLEANUP_IF_TOKEN_SCRIPT = """
 local tracked = redis.call("get", KEYS[2])
 local active = redis.call("get", KEYS[3])
@@ -139,11 +188,13 @@ class RedisStreamBridge(StreamBridge):
         max_connections: int | None = None,
         stream_ttl_seconds: int | None = 86400,
         producer_lease_recovery_enabled: bool = False,
+        legacy_end_scan_enabled: bool = True,
         client: Redis | None = None,
     ) -> None:
         super().__init__(heartbeat_interval=heartbeat_interval)
         self._redis_url = redis_url
         self._maxsize = max(1, queue_maxsize)
+        self._legacy_end_scan_enabled = legacy_end_scan_enabled
         self._key_prefix = key_prefix.rstrip(":")
         if stream_ttl_seconds is not None and stream_ttl_seconds > 0:
             self._stream_ttl_seconds = stream_ttl_seconds
@@ -442,11 +493,17 @@ class RedisStreamBridge(StreamBridge):
             fields.get("kind", _KIND_EVENT),
             fields.get("event", ""),
             fields.get("data", ""),
+            int(self._legacy_end_scan_enabled),
         )
-        if result:
-            return
-        self._cancel_lost_producer(run_id, lease)
-        raise RuntimeError(f"Redis stream producer lease was lost for run {run_id}")
+        if result == _XADD_BLOCKED:
+            self._cancel_lost_producer(run_id, lease)
+            raise RuntimeError(f"Redis stream producer lease was lost for run {run_id}")
+        if result == _XADD_ALREADY_ENDED:
+            if fields.get("kind") == _KIND_END:
+                return
+            raise RuntimeError(f"Redis stream for run {run_id} has already ended")
+        # Every remaining successful Lua reply is an XADD stream ID.
+        assert isinstance(result, (str, bytes)), f"Unexpected Redis stream append result: {result!r}"
 
     async def _xadd_without_active_producer(
         self,
@@ -465,10 +522,21 @@ class RedisStreamBridge(StreamBridge):
             fields.get("kind", _KIND_EVENT),
             fields.get("event", ""),
             fields.get("data", ""),
+            int(self._legacy_end_scan_enabled),
         )
-        if result:
-            return
-        raise RuntimeError(f"Cannot publish stream data for run {run_id} because another active producer exists")
+        if result == _XADD_BLOCKED:
+            raise RuntimeError(f"Cannot publish stream data for run {run_id} because another active producer exists")
+        if result == _XADD_ALREADY_ENDED:
+            if fields.get("kind") == _KIND_END:
+                return
+            raise RuntimeError(f"Redis stream for run {run_id} has already ended")
+        if result == _XADD_STREAM_MISSING:
+            if fields.get("kind") == _KIND_END:
+                logger.debug("Skipping stream END for %s: retained stream no longer exists", run_id)
+                return
+            raise RuntimeError(f"Redis stream for run {run_id} no longer exists")
+        # Every remaining successful Lua reply is an XADD stream ID.
+        assert isinstance(result, (str, bytes)), f"Unexpected Redis stream append result: {result!r}"
 
     async def _stop_producer_lease(
         self,

backend/tests/AGENTS.md

@@ -18,6 +18,10 @@ ownership and capture cleanup. Retain the real XLSX conversion coverage.
 
 Backend tests must preserve the runtime invariants they exercise without changing production execution topology.
 
+Provider cleanup tests use concrete configuration models and event gates.
+Surface owner-task failures while awaiting lifecycle events; join owners and
+event waiters during teardown.
+
 `test_bench_sandbox_prewarm.py` controls observation timestamps through the
 benchmark module's clock binding. Keep asyncio's clock real, initialize
 `Trace.started` from the controlled clock, and advance ticks at event gates.

backend/tests/test_runtime_provider_close_cancellation.py

@@ -1,8 +1,10 @@
 import asyncio
+from contextlib import asynccontextmanager
 from types import SimpleNamespace
 
 import pytest
 
+from deerflow.config.stream_bridge_config import StreamBridgeConfig
 from deerflow.runtime.checkpoint_cache import provider as checkpoint_provider
 from deerflow.runtime.checkpoint_cache import redis as checkpoint_redis
 from deerflow.runtime.stream_bridge import async_provider as stream_provider
@@ -30,6 +32,20 @@ class _BlockingStreamBridge:
         await self.allow_close.wait()
 
 
+async def _wait_for_owner_event(task: asyncio.Task[None], event: asyncio.Event) -> None:
+    waiter = asyncio.create_task(event.wait())
+    try:
+        done, _ = await asyncio.wait((task, waiter), timeout=1, return_when=asyncio.FIRST_COMPLETED)
+        if task in done:
+            await task
+            raise AssertionError("provider owner finished before its lifecycle event")
+        if waiter not in done:
+            raise TimeoutError("provider did not reach its lifecycle event")
+    finally:
+        waiter.cancel()
+        await asyncio.gather(waiter, return_exceptions=True)
+
+
 async def _assert_close_is_drained(cm, close_started: asyncio.Event, allow_close: asyncio.Event) -> None:
     entered = asyncio.Event()
     leave = asyncio.Event()
@@ -42,9 +58,9 @@ async def _assert_close_is_drained(cm, close_started: asyncio.Event, allow_close
     task: asyncio.Task[None] | None = None
     try:
         task = asyncio.create_task(owner())
-        await asyncio.wait_for(entered.wait(), timeout=1)
+        await _wait_for_owner_event(task, entered)
         leave.set()
-        await asyncio.wait_for(close_started.wait(), timeout=1)
+        await _wait_for_owner_event(task, close_started)
 
         task.cancel()
         for _ in range(5):
@@ -61,8 +77,9 @@ async def _assert_close_is_drained(cm, close_started: asyncio.Event, allow_close
             await task
     finally:
         allow_close.set()
-        if task is not None and not task.done():
-            task.cancel()
+        if task is not None:
+            if not task.done():
+                task.cancel()
             await asyncio.gather(task, return_exceptions=True)
 
 
@@ -107,7 +124,7 @@ async def test_checkpoint_cache_redis_context_drains_close_across_repeated_cance
 async def test_stream_bridge_context_drains_close_across_repeated_cancellation(monkeypatch: pytest.MonkeyPatch) -> None:
     bridge = _BlockingStreamBridge()
     monkeypatch.setattr(stream_memory, "MemoryStreamBridge", lambda **_kwargs: bridge)
-    config = SimpleNamespace(stream_bridge=SimpleNamespace(type="memory", queue_maxsize=8, heartbeat_interval_seconds=1.0))
+    config = SimpleNamespace(stream_bridge=StreamBridgeConfig(type="memory", queue_maxsize=8, heartbeat_interval_seconds=1.0))
 
     await _assert_close_is_drained(
         stream_provider.make_stream_bridge(config),
@@ -123,7 +140,7 @@ async def test_stream_bridge_redis_context_drains_close_across_repeated_cancella
     bridge = _BlockingStreamBridge()
     monkeypatch.setattr(stream_redis, "RedisStreamBridge", lambda *_args, **_kwargs: bridge)
     config = SimpleNamespace(
-        stream_bridge=SimpleNamespace(
+        stream_bridge=StreamBridgeConfig(
             type="redis",
             redis_url="redis://localhost:6379/0",
             queue_maxsize=8,
@@ -138,3 +155,21 @@ async def test_stream_bridge_redis_context_drains_close_across_repeated_cancella
         bridge.close_started,
         bridge.allow_close,
     )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure_stage", ["enter", "close"])
+async def test_close_drain_helper_propagates_owner_failure(failure_stage: str) -> None:
+    error = RuntimeError(f"provider {failure_stage} failed")
+
+    @asynccontextmanager
+    async def failing_context():
+        if failure_stage == "enter":
+            raise error
+        yield
+        raise error
+
+    with pytest.raises(RuntimeError, match=f"provider {failure_stage} failed") as raised:
+        await _assert_close_is_drained(failing_context(), asyncio.Event(), asyncio.Event())
+
+    assert raised.value is error

backend/tests/test_stream_bridge.py

@@ -5,7 +5,7 @@ import os
 import re
 import uuid
 from collections import defaultdict
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, Mock
 
 import anyio
 import pytest
@@ -50,6 +50,7 @@ class _FakeRedis:
         self.closed = False
         self.ping_calls = 0
         self.ping_error: Exception | None = None
+        self._eval_lock = asyncio.Lock()
 
     async def ping(self):
         self.ping_calls += 1
@@ -107,7 +108,20 @@ class _FakeRedis:
     async def mget(self, *names):
         return [self.values.get(name) for name in names]
 
+    def _tail_is_end(self, stream_key):
+        entries = self.streams.get(stream_key, [])
+        return bool(entries and entries[-1][1].get("kind") == "end")
+
+    def _has_end_marker(self, stream_key):
+        return any(fields.get("kind") == "end" for _event_id, fields in self.streams.get(stream_key, []))
+
     async def eval(self, script, numkeys, *args):
+        # Redis executes a Lua script atomically. Keep the fake's assertion
+        # surface faithful even though individual emulated commands yield.
+        async with self._eval_lock:
+            return await self._eval(script, numkeys, *args)
+
+    async def _eval(self, script, numkeys, *args):
         keys = args[:numkeys]
         values = args[numkeys:]
         if numkeys == 2 and not values:
@@ -133,11 +147,18 @@ class _FakeRedis:
                 return 0
             await self.delete(stream_key, tracking_key, active_key)
             return 1
-        if numkeys == 2 and len(values) == 5:
+        if numkeys == 2 and len(values) == 6:
             active_key, stream_key = keys
-            maxlen, stream_ttl, kind, event, data = values
+            maxlen, stream_ttl, kind, event, data, legacy_scan_enabled = values
             if active_key in self.values:
-                return None
+                return 0
+            if kind == "end" and stream_key not in self.streams:
+                return 2
+            if kind == "end":
+                if self._tail_is_end(stream_key) or (int(legacy_scan_enabled) and self._has_end_marker(stream_key)):
+                    return 1
+            elif self._tail_is_end(stream_key):
+                return 1
             event_id = await self.xadd(
                 stream_key,
                 {"kind": kind, "event": event, "data": data},
@@ -166,9 +187,14 @@ class _FakeRedis:
             return 1
         if numkeys == 3:
             active_key, tracking_key, stream_key = keys
-            token, maxlen, stream_ttl, kind, event, data = values
+            token, maxlen, stream_ttl, kind, event, data, legacy_scan_enabled = values
             if self.values.get(active_key) != token or self.values.get(tracking_key) != token:
-                return None
+                return 0
+            if kind == "end":
+                if self._tail_is_end(stream_key) or (int(legacy_scan_enabled) and self._has_end_marker(stream_key)):
+                    return 1
+            elif self._tail_is_end(stream_key):
+                return 1
             event_id = await self.xadd(
                 stream_key,
                 {"kind": kind, "event": event, "data": data},
@@ -227,6 +253,24 @@ class _DelayedBlockingReadRedis:
         return response
 
 
+class _LostAckRedis:
+    """Delegate to Redis but drop one successful EVAL response."""
+
+    def __init__(self, redis) -> None:
+        self._delegate = redis
+        self.drop_next_successful_eval = False
+
+    def __getattr__(self, name):
+        return getattr(self._delegate, name)
+
+    async def eval(self, script, numkeys, *args):
+        result = await self._delegate.eval(script, numkeys, *args)
+        if self.drop_next_successful_eval and result:
+            self.drop_next_successful_eval = False
+            raise TimeoutError("simulated lost Redis EVAL acknowledgement")
+        return result
+
+
 class _FakeRedisPipeline:
     def __init__(self, redis: _FakeRedis) -> None:
         self.redis = redis
@@ -704,6 +748,148 @@ async def test_redis_publish_subscribe(redis_bridge: RedisStreamBridge):
     assert received[-1] is END_SENTINEL
 
 
+@pytest.mark.anyio
+async def test_redis_publish_end_is_idempotent(redis_bridge: RedisStreamBridge):
+    """Concurrent and repeated END publication must append at most one marker."""
+    fake = redis_bridge._redis
+    run_id = "redis-run-idempotent-end"
+    key = redis_bridge._stream_key(run_id)
+
+    await redis_bridge.start_producer_lease(run_id)
+    try:
+        await redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+        await asyncio.gather(*(redis_bridge.publish_end(run_id) for _ in range(8)))
+        await redis_bridge.publish_end(run_id)
+
+        entries = fake.streams[key]
+        assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"]
+
+        received = [entry async for entry in redis_bridge.subscribe(run_id, heartbeat_interval=1.0)]
+        assert [entry.event for entry in received[:-1]] == ["metadata"]
+        assert received[-1] is END_SENTINEL
+    finally:
+        await redis_bridge.finish_producer_lease(run_id)
+
+
+@pytest.mark.anyio
+async def test_redis_publish_end_recognizes_legacy_end_before_late_events(redis_bridge: RedisStreamBridge):
+    """An END written by the old buggy writer still prevents a duplicate END."""
+    fake = redis_bridge._redis
+    run_id = "redis-run-legacy-end-not-tail"
+    key = redis_bridge._stream_key(run_id)
+    fake.streams[key] = [
+        ("1-0", {"kind": "event", "event": "metadata", "data": "{}"}),
+        ("2-0", {"kind": "end"}),
+        ("3-0", {"kind": "event", "event": "late", "data": "{}"}),
+    ]
+
+    await redis_bridge.publish_end(run_id)
+
+    assert [fields["kind"] for _event_id, fields in fake.streams[key]].count("end") == 1
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("owned", [False, True], ids=["recovery", "owner"])
+async def test_redis_legacy_midstream_end_allows_data_but_deduplicates_end(owned: bool):
+    """Data checks only the tail; END retries still inspect the retained window."""
+    fake = _FakeRedis()
+    bridge = RedisStreamBridge(redis_url="redis://fake", queue_maxsize=8, client=fake)
+    run_id = "redis-run-legacy-data-after-end"
+    key = bridge._stream_key(run_id)
+    if owned:
+        await bridge.start_producer_lease(run_id)
+    try:
+        await fake.xadd(key, {"kind": "end"})
+        await fake.xadd(key, {"kind": "event", "event": "legacy-late", "data": "{}"})
+        before = list(fake.streams[key])
+
+        await bridge.publish(run_id, "current-data", {"step": 2})
+
+        entries = list(fake.streams[key])
+        assert entries[:2] == before
+        assert len(entries) == 3
+        assert entries[-1][1]["event"] == "current-data"
+        await bridge.publish_end(run_id)
+        assert fake.streams[key] == entries
+    finally:
+        if owned:
+            await bridge.finish_producer_lease(run_id)
+
+
+@pytest.mark.anyio
+async def test_redis_owned_end_recreates_missing_buffer_with_live_lease():
+    """Only unowned recovery END is a no-op on a missing buffer."""
+    fake = _FakeRedis()
+    bridge = RedisStreamBridge(redis_url="redis://fake", client=fake)
+    run_id = "redis-run-owned-missing-buffer"
+    key = bridge._stream_key(run_id)
+    await bridge.start_producer_lease(run_id)
+    try:
+        await bridge.publish(run_id, "metadata", {})
+        await fake.delete(key)
+
+        await bridge.publish_end(run_id)
+
+        assert [fields["kind"] for _event_id, fields in fake.streams[key]] == ["end"]
+    finally:
+        await bridge.finish_producer_lease(run_id)
+
+
+@pytest.mark.anyio
+async def test_redis_publish_after_end_is_rejected_without_appending(redis_bridge: RedisStreamBridge):
+    """Once END is retained, later event publication cannot corrupt terminal order."""
+    fake = redis_bridge._redis
+    run_id = "redis-run-publish-after-end"
+    key = redis_bridge._stream_key(run_id)
+
+    await redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+    await redis_bridge.publish_end(run_id)
+    before = list(fake.streams[key])
+
+    with pytest.raises(RuntimeError, match="already ended"):
+        await redis_bridge.publish(run_id, "late", {"step": 2})
+
+    assert fake.streams[key] == before
+
+
+@pytest.mark.anyio
+async def test_redis_event_end_race_never_appends_after_end(redis_bridge: RedisStreamBridge):
+    """The atomic terminal check keeps END last even when an event races it."""
+    fake = redis_bridge._redis
+    run_id = "redis-run-event-end-race"
+    key = redis_bridge._stream_key(run_id)
+
+    await redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+    event_result, end_result = await asyncio.gather(
+        redis_bridge.publish(run_id, "racing-event", {"step": 2}),
+        redis_bridge.publish_end(run_id),
+        return_exceptions=True,
+    )
+
+    assert end_result is None
+    assert event_result is None or "already ended" in str(event_result)
+    kinds = [fields["kind"] for _event_id, fields in fake.streams[key]]
+    assert kinds.count("end") == 1
+    assert kinds[-1] == "end"
+
+
+@pytest.mark.anyio
+async def test_redis_recovery_end_after_cleanup_does_not_resurrect_stream(redis_bridge: RedisStreamBridge):
+    """A delayed recovery END is a no-op when cleanup already removed the stream."""
+    fake = redis_bridge._redis
+    run_id = "redis-run-cleaned-end-retry"
+    key = redis_bridge._stream_key(run_id)
+
+    await redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+    await redis_bridge.publish_end(run_id)
+    await redis_bridge.cleanup(run_id)
+    assert key not in fake.streams
+
+    await redis_bridge.publish_end(run_id)
+
+    assert key not in fake.streams
+
+
 @pytest.mark.anyio
 async def test_redis_replays_after_last_event_id(redis_bridge: RedisStreamBridge):
     """Redis XREAD should resume after Last-Event-ID."""
@@ -1098,6 +1284,30 @@ async def test_redis_finish_does_not_delete_a_successor_token():
     assert fake.values[active_key] == "successor-token"
 
 
+@pytest.mark.anyio
+async def test_redis_lost_owner_does_not_get_idempotent_end_success():
+    """Lease validation precedes retained-END detection for a stale owner."""
+    fake = _FakeRedis()
+    bridge = RedisStreamBridge(redis_url="redis://fake", client=fake)
+    run_id = "redis-run-lost-owner-existing-end"
+    active_key = bridge._producer_active_key(run_id)
+    key = bridge._stream_key(run_id)
+    replacement_owner = asyncio.create_task(asyncio.Event().wait())
+
+    await bridge.start_producer_lease(run_id)
+    bridge.transfer_producer_lease(run_id, replacement_owner)
+    await bridge.publish_end(run_id)
+    fake.values[active_key] = "successor-token"
+    try:
+        with pytest.raises(RuntimeError, match="producer lease was lost"):
+            await bridge.publish_end(run_id)
+        assert [fields["kind"] for _event_id, fields in fake.streams[key]].count("end") == 1
+    finally:
+        await bridge.finish_producer_lease(run_id)
+        replacement_owner.cancel()
+        await asyncio.gather(replacement_owner, return_exceptions=True)
+
+
 @pytest.mark.anyio
 async def test_redis_publish_is_atomically_fenced_by_producer_token():
     fake = _FakeRedis()
@@ -2103,6 +2313,7 @@ async def test_make_stream_bridge_passes_redis_options(monkeypatch):
             max_connections=50,
             stream_ttl_seconds=42,
             producer_lease_recovery_enabled=True,
+            legacy_end_scan_enabled=False,
         )
     )
     try:
@@ -2111,6 +2322,7 @@ async def test_make_stream_bridge_passes_redis_options(monkeypatch):
             assert bridge.heartbeat_interval == 2.5
             assert bridge._stream_ttl_seconds == 42
             assert bridge.producer_lease_recovery_enabled is True
+            assert bridge._legacy_end_scan_enabled is False
         assert captured["max_connections"] == 50
         assert captured["decode_responses"] is True
     finally:
@@ -2188,6 +2400,189 @@ async def test_redis_integration_publish_subscribe_and_id_format(real_redis_brid
     assert re.match(r"^\d+-\d+$", received[0].id), received[0].id
 
 
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+async def test_redis_integration_end_is_idempotent_across_independent_clients():
+    """Concurrent END writers must serialize through Redis, not local process state."""
+    from redis.asyncio import Redis
+
+    clients = [
+        Redis.from_url(REDIS_TEST_URL, decode_responses=True),
+        Redis.from_url(REDIS_TEST_URL, decode_responses=True),
+    ]
+    key_prefix = f"deerflow:test:{uuid.uuid4().hex}"
+    bridges = [
+        RedisStreamBridge(
+            redis_url=REDIS_TEST_URL,
+            queue_maxsize=2,
+            key_prefix=key_prefix,
+            client=client,
+        )
+        for client in clients
+    ]
+    run_id = "integ-end-idempotent"
+    key = bridges[0]._stream_key(run_id)
+    try:
+        await bridges[0].publish(run_id, "metadata", {"run_id": run_id})
+        await asyncio.gather(*(bridge.publish_end(run_id) for bridge in bridges))
+        await asyncio.gather(*(bridge.publish_end(run_id) for bridge in bridges))
+
+        entries = await clients[0].xrange(key)
+        assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"]
+
+        first_event_id = entries[0][0]
+        replayed = [
+            entry
+            async for entry in bridges[1].subscribe(
+                run_id,
+                last_event_id=first_event_id,
+                heartbeat_interval=1.0,
+            )
+        ]
+        assert replayed == [END_SENTINEL]
+    finally:
+        async for existing_key in clients[0].scan_iter(f"{key_prefix}:*"):
+            await clients[0].delete(existing_key)
+        await asyncio.gather(*(client.aclose() for client in clients))
+
+
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+async def test_redis_integration_recovery_end_cannot_race_active_owner(real_redis_bridge):
+    """Recovery may append END only after the valid producer's lease is released."""
+    from redis.asyncio import Redis
+
+    recovery_client = Redis.from_url(REDIS_TEST_URL, decode_responses=True)
+    recovery = RedisStreamBridge(
+        redis_url=REDIS_TEST_URL,
+        queue_maxsize=2,
+        key_prefix=real_redis_bridge._key_prefix,
+        client=recovery_client,
+    )
+    run_id = "integ-end-owner-recovery-race"
+    key = real_redis_bridge._stream_key(run_id)
+    try:
+        await real_redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+        await real_redis_bridge.start_producer_lease(run_id)
+
+        owner_result, recovery_result = await asyncio.gather(
+            real_redis_bridge.publish_end(run_id),
+            recovery.publish_end(run_id),
+            return_exceptions=True,
+        )
+        assert owner_result is None
+        assert isinstance(recovery_result, RuntimeError)
+        assert "active producer" in str(recovery_result)
+
+        entries = await recovery_client.xrange(key)
+        assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"]
+
+        await real_redis_bridge.finish_producer_lease(run_id)
+        await recovery.publish_end(run_id)
+        entries = await recovery_client.xrange(key)
+        assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"]
+    finally:
+        await real_redis_bridge.finish_producer_lease(run_id)
+        await recovery_client.aclose()
+
+
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+async def test_redis_integration_end_retry_after_lost_ack_is_idempotent(real_redis_bridge):
+    """A retry confirms the retained END without appending a duplicate."""
+    run_id = "integ-end-lost-ack"
+    key = real_redis_bridge._stream_key(run_id)
+
+    await real_redis_bridge.start_producer_lease(run_id)
+    try:
+        await real_redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+        lossy = _LostAckRedis(real_redis_bridge._redis)
+        real_redis_bridge._redis = lossy
+        lossy.drop_next_successful_eval = True
+
+        with pytest.raises(TimeoutError, match="lost Redis EVAL acknowledgement"):
+            await real_redis_bridge.publish_end(run_id)
+
+        await real_redis_bridge.publish_end(run_id)
+
+        entries = await lossy.xrange(key)
+        assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"]
+    finally:
+        await real_redis_bridge.finish_producer_lease(run_id)
+
+
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+async def test_redis_integration_cleanup_prevents_late_end_from_recreating_stream(real_redis_bridge):
+    """An END retry after cleanup is acknowledged as a no-op, not a resurrection."""
+    run_id = "integ-end-after-cleanup"
+    key = real_redis_bridge._stream_key(run_id)
+
+    await real_redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+    await real_redis_bridge.publish_end(run_id)
+    await real_redis_bridge.cleanup(run_id)
+    assert await real_redis_bridge._redis.exists(key) == 0
+
+    await real_redis_bridge.publish_end(run_id)
+
+    assert await real_redis_bridge._redis.exists(key) == 0
+
+
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+@pytest.mark.parametrize("owned", [False, True], ids=["recovery", "owner"])
+async def test_redis_integration_legacy_midstream_end_allows_data_but_deduplicates_end(real_redis_bridge, owned: bool):
+    """Pin the fake's legacy-stream decisions against the real Lua scripts."""
+    client = real_redis_bridge._redis
+    bridge = RedisStreamBridge(redis_url=REDIS_TEST_URL, queue_maxsize=8, key_prefix=real_redis_bridge._key_prefix, client=client)
+    run_id = "integ-legacy-data-after-end"
+    key = bridge._stream_key(run_id)
+    if owned:
+        await bridge.start_producer_lease(run_id)
+    try:
+        await client.xadd(key, {"kind": "end"})
+        await client.xadd(key, {"kind": "event", "event": "legacy-late", "data": "{}"})
+        before = await client.xrange(key)
+
+        await bridge.publish(run_id, "current-data", {"step": 2})
+
+        entries = await client.xrange(key)
+        assert entries[:2] == before
+        assert len(entries) == 3
+        assert entries[-1][1]["event"] == "current-data"
+        await bridge.publish_end(run_id)
+        assert await client.xrange(key) == entries
+    finally:
+        if owned:
+            await bridge.finish_producer_lease(run_id)
+
+
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+async def test_redis_integration_owned_end_recreates_missing_buffer_with_live_lease(real_redis_bridge):
+    """A live owner retains the existing XADD behavior after buffer removal."""
+    bridge = real_redis_bridge
+    client = bridge._redis
+    run_id = "integ-owned-missing-buffer"
+    key = bridge._stream_key(run_id)
+    await bridge.start_producer_lease(run_id)
+    try:
+        await bridge.publish(run_id, "metadata", {})
+        await client.delete(key)
+
+        await bridge.publish_end(run_id)
+
+        assert [fields["kind"] for _event_id, fields in await client.xrange(key)] == ["end"]
+    finally:
+        await bridge.finish_producer_lease(run_id)
+
+
 @pytest.mark.integration
 @requires_redis
 @pytest.mark.anyio
@@ -2385,3 +2780,80 @@ async def test_redis_integration_stream_ttl_reclaims_key():
         async for existing_key in client.scan_iter(f"{key_prefix}:*"):
             await client.delete(existing_key)
         await client.aclose()
+
+
+@pytest.mark.parametrize("enabled", [True, False])
+def test_stream_bridge_config_legacy_end_scan_switch(enabled):
+    assert StreamBridgeConfig().legacy_end_scan_enabled is True
+    assert StreamBridgeConfig(legacy_end_scan_enabled=enabled).legacy_end_scan_enabled is enabled
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("owned", [False, True], ids=["recovery", "owner"])
+@pytest.mark.parametrize("enabled", [True, False], ids=["compatibility-scan", "tail-only"])
+async def test_redis_end_scan_switch_and_tail_fast_path(owned, enabled):
+    fake = _FakeRedis()
+    fake._has_end_marker = Mock(wraps=fake._has_end_marker)
+    bridge = RedisStreamBridge(redis_url="redis://fake", queue_maxsize=8, legacy_end_scan_enabled=enabled, client=fake)
+    run_id = "scan-switch"
+    if owned:
+        await bridge.start_producer_lease(run_id)
+    try:
+        await bridge.publish(run_id, "values", {"content": "x" * 4096})
+        assert fake._has_end_marker.call_count == 0
+        await bridge.publish_end(run_id)
+        assert fake._has_end_marker.call_count == int(enabled)
+        await bridge.publish_end(run_id)
+        with pytest.raises(RuntimeError, match="already ended"):
+            await bridge.publish(run_id, "late", {})
+        assert fake._has_end_marker.call_count == int(enabled)
+        assert [fields["kind"] for _, fields in fake.streams[bridge._stream_key(run_id)]] == ["event", "end"]
+    finally:
+        if owned:
+            await bridge.finish_producer_lease(run_id)
+        await bridge.close()
+
+
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+@pytest.mark.parametrize("owned", [False, True], ids=["recovery", "owner"])
+@pytest.mark.parametrize("enabled", [True, False], ids=["compatibility-scan", "tail-only"])
+async def test_redis_integration_end_scan_switch_with_xrange_denied(real_redis_bridge, owned, enabled):
+    """Deny XRANGE so real Lua proves that tail-only END never scans history."""
+    from redis.asyncio import ConnectionPool, Redis
+    from redis.exceptions import ResponseError
+
+    admin = real_redis_bridge._redis
+    server = await admin.info("server")
+    if int(server["redis_version"].split(".", 1)[0]) < 6:
+        pytest.skip("This scan-path probe uses Redis 6+ ACLs")
+    username = "end-scan-" + uuid.uuid4().hex
+    prefix = real_redis_bridge._key_prefix
+    await admin.acl_setuser(username, enabled=True, reset=True, passwords=["+fixture-password"], keys=[f"{prefix}:*"], categories=["+all"], commands=["-xrange"])
+    pool = ConnectionPool.from_url(REDIS_TEST_URL, decode_responses=True)
+    pool.connection_kwargs.update(username=username, password="fixture-password")
+    client = Redis(connection_pool=pool)
+    bridge = RedisStreamBridge(redis_url=REDIS_TEST_URL, queue_maxsize=8, key_prefix=prefix, legacy_end_scan_enabled=enabled, client=client)
+    run_id = "scan-switch-acl"
+    try:
+        if owned:
+            await bridge.start_producer_lease(run_id)
+        await bridge.publish(run_id, "values", {"content": "x" * 4096})
+        if enabled:
+            with pytest.raises(ResponseError, match="(?i)acl|permission|user executing|allowed"):
+                await bridge.publish_end(run_id)
+            assert await admin.xlen(bridge._stream_key(run_id)) == 1
+        else:
+            await bridge.publish_end(run_id)
+            await bridge.publish_end(run_id)
+            with pytest.raises(RuntimeError, match="already ended"):
+                await bridge.publish(run_id, "late", {})
+            assert [fields["kind"] for _, fields in await admin.xrange(bridge._stream_key(run_id))] == ["event", "end"]
+    finally:
+        try:
+            await bridge.close()
+        finally:
+            await client.aclose()
+            await pool.disconnect()
+            await admin.acl_deluser(username)

config.example.yaml

@@ -20,7 +20,7 @@
 # ============================================================================
 # Bump this number when the config schema changes.
 # Run `make config-upgrade` to merge new fields into your local config.yaml.
-config_version: 62
+config_version: 63
 
 # Optional operator-owned literal prompt extensions (see CONFIGURATION.md):
 # lead_prompt_overlay:
@@ -2990,6 +2990,13 @@ run_ownership:
 #   stream_ttl_seconds: 86400    # rolling TTL for retained stream buffers.
 #                                # Refreshed on each publish/publish_end; set 0
 #                                # to disable. This is not a run timeout.
+#   legacy_end_scan_enabled: true  # restart-required Redis compatibility scan.
+#                                # First END scans all retained payloads while true.
+#                                # Set false only after all old writers stop and
+#                                # every legacy buffer expires or is removed.
+#                                # With TTL disabled, remove old buffers explicitly.
+#                                # Restart all Gateways together; false uses only
+#                                # the tail for O(1) END detection.
 #   producer_lease_recovery_enabled: false  # restart-required activation gate.
 #                                # First roll this binary to every Gateway and
 #                                # drain pre-upgrade runs; then enable on every

deploy/helm/deer-flow/README.md

@@ -135,7 +135,7 @@ they resolve from the `secrets` map):
 
 ```yaml
 config: |
-  config_version: 62
+  config_version: 63
   models:
     - name: gpt-4
       use: langchain_openai:ChatOpenAI

deploy/helm/deer-flow/values.yaml

@@ -367,7 +367,7 @@ ingress:
 # -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
 #    inline literal secret values here. The default enables provisioner sandbox.
 config: |
-  config_version: 62
+  config_version: 63
   log_level: info
   recursion_limit: 100
   max_recursion_limit: 1000