fix(config): guard the recovered stream cleanup delay like its heartbeat sibling (#5850)

backend/packages/harness/deerflow/config/stream_bridge_config.py

@@ -2,11 +2,12 @@
 
 from typing import Any, Literal
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, ValidationInfo, field_validator
 
 StreamBridgeType = Literal["memory", "redis"]
 DEFAULT_HEARTBEAT_INTERVAL_SECONDS = 15.0
 MAX_HEARTBEAT_INTERVAL_SECONDS = 86_400.0
+MAX_RECOVERED_STREAM_CLEANUP_DELAY_SECONDS = 86_400.0
 
 
 class StreamBridgeConfig(BaseModel):
@@ -54,15 +55,19 @@ class StreamBridgeConfig(BaseModel):
     recovered_stream_cleanup_delay_seconds: float = Field(
         default=60.0,
         ge=0,
-        description=("Seconds to wait after publishing an END marker for a recovered orphaned run before deleting the stream key. Gives reconnecting SSE clients time to drain the end signal. Only applies to the redis bridge."),
+        le=MAX_RECOVERED_STREAM_CLEANUP_DELAY_SECONDS,
+        allow_inf_nan=False,
+        description=(
+            "Seconds to wait after publishing an END marker for a recovered orphaned run before deleting the stream key (maximum 86400). Gives reconnecting SSE clients time to drain the end signal. Only applies to the redis bridge."
+        ),
     )
 
-    @field_validator("heartbeat_interval_seconds", mode="before")
+    @field_validator("heartbeat_interval_seconds", "recovered_stream_cleanup_delay_seconds", mode="before")
     @classmethod
-    def reject_boolean_heartbeat_interval(cls, value: Any) -> Any:
+    def reject_boolean_seconds(cls, value: Any, info: ValidationInfo) -> Any:
         """Reject booleans before Pydantic coerces them to floats."""
         if isinstance(value, bool):
-            raise ValueError("heartbeat_interval_seconds must be a number, not a boolean")
+            raise ValueError(f"{info.field_name} must be a number, not a boolean")
         return value
 
 

backend/tests/test_stream_bridge.py

@@ -10,7 +10,12 @@ import anyio
 import pytest
 from pydantic import ValidationError
 
-from deerflow.config.stream_bridge_config import MAX_HEARTBEAT_INTERVAL_SECONDS, StreamBridgeConfig, set_stream_bridge_config
+from deerflow.config.stream_bridge_config import (
+    MAX_HEARTBEAT_INTERVAL_SECONDS,
+    MAX_RECOVERED_STREAM_CLEANUP_DELAY_SECONDS,
+    StreamBridgeConfig,
+    set_stream_bridge_config,
+)
 from deerflow.runtime import END_SENTINEL, HEARTBEAT_SENTINEL, MemoryStreamBridge, StreamGap, make_stream_bridge
 
 # RedisStreamBridge is no longer re-exported from deerflow.runtime (redis is an
@@ -984,6 +989,31 @@ def test_stream_bridge_config_accepts_numeric_heartbeat_string():
     assert config.heartbeat_interval_seconds == 2.5
 
 
+@pytest.mark.parametrize(
+    "cleanup_delay",
+    [
+        True,
+        False,
+        -1,
+        float("inf"),
+        float("-inf"),
+        float("nan"),
+        MAX_RECOVERED_STREAM_CLEANUP_DELAY_SECONDS + 1,
+    ],
+)
+def test_stream_bridge_config_rejects_invalid_recovered_stream_cleanup_delay(cleanup_delay):
+    with pytest.raises(ValidationError, match="recovered_stream_cleanup_delay_seconds"):
+        StreamBridgeConfig(recovered_stream_cleanup_delay_seconds=cleanup_delay)
+
+
+@pytest.mark.parametrize("cleanup_delay", [0, 60, MAX_RECOVERED_STREAM_CLEANUP_DELAY_SECONDS])
+def test_stream_bridge_config_accepts_valid_recovered_stream_cleanup_delay(cleanup_delay):
+    """A delay of 0 is a valid 'delete as soon as END is published' setting."""
+    config = StreamBridgeConfig(recovered_stream_cleanup_delay_seconds=cleanup_delay)
+
+    assert config.recovered_stream_cleanup_delay_seconds == float(cleanup_delay)
+
+
 @pytest.mark.parametrize("heartbeat_interval", [True, MAX_HEARTBEAT_INTERVAL_SECONDS + 1])
 def test_memory_bridge_rejects_invalid_default_heartbeat(heartbeat_interval):
     with pytest.raises(ValueError, match="heartbeat_interval"):

config.example.yaml

@@ -2802,7 +2802,7 @@ run_ownership:
 #                                # to disable. This is not a run timeout.
 #   recovered_stream_cleanup_delay_seconds: 60  # seconds to wait after
 #                                # publishing END for a recovered orphan run
-#                                # before deleting the stream key.
+#                                # before deleting the stream key (max: 86400).
 #   max_connections: 100         # optional pool ceiling. Each live SSE client
 #                                # holds one connection blocked in XREAD ... BLOCK
 #                                # for up to the configured heartbeat interval, so