diff --git a/backend/packages/harness/deerflow/config/stream_bridge_config.py b/backend/packages/harness/deerflow/config/stream_bridge_config.py index d2e5cfa7a9..fb2124dc3d 100644 --- a/backend/packages/harness/deerflow/config/stream_bridge_config.py +++ b/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 diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py index b79ed98323..f0581361ab 100644 --- a/backend/tests/test_stream_bridge.py +++ b/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"): diff --git a/config.example.yaml b/config.example.yaml index 69d5245095..013ee2644e 100644 --- a/config.example.yaml +++ b/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