feat(scheduler): make scheduled-run recursion_limit configurable (#4848)

Adds scheduler.recursion_limit to config.yaml (default 1000, clamped by
max_recursion_limit) so scheduled background runs can use a different
recursion limit than the web UI. The value is read at dispatch time, so
a YAML edit applies to the next scheduled run without a Gateway restart.

Also logs a warning when the resolver falls back to the default or
clamps the configured value.
This commit is contained in:
Aleksandr Sapronov 2026-08-24 06:09:08 +07:00 committed by GitHub
parent b36504edf2
commit 613b90b0e6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 223 additions and 9 deletions

View File

@ -104,6 +104,7 @@ Skill quality review note:
Scheduled-task note:
- The scheduled-task MVP adds a workspace page at `/workspace/scheduled-tasks` plus a background scheduler service gated by `config.yaml -> scheduler.enabled`.
- Scheduled background runs are intentionally non-interactive: they execute through the normal run lifecycle, but the lead-agent toolset excludes `ask_clarification` when `context.non_interactive=true`. The key is honored only for internally-authenticated callers (the scheduler launch path); client-supplied `context.non_interactive` is dropped.
- Scheduled launches use `scheduler.recursion_limit` (default 1000, matching the web UI's `recursion_limit: 1000`, clamped by `max_recursion_limit`). The value is read at dispatch, so a YAML edit applies to the next scheduled run without a Gateway restart.
## Commands: Root vs. Module

View File

@ -123,6 +123,9 @@ This section accumulates work toward the **2.1.0** milestone
`present_files` no longer report success when delivery fails. ([#4365])
- **uploads:** Lazy-load historical files via a `list_uploaded_files` tool
instead of injecting the full manifest. ([#4174])
- **scheduler:** `scheduler.recursion_limit` in `config.yaml` sets the LangGraph
super-step cap for scheduled runs (default 1000, matching the web UI's
interactive budget, clamped by `max_recursion_limit`).
#### Memory

View File

@ -1361,6 +1361,8 @@ Current MVP limits:
Enable background polling with `config.yaml -> scheduler.enabled`. Manual trigger uses the same scheduled-task resource and execution path.
Scheduled runs use `scheduler.recursion_limit` in `config.yaml` (default `1000`, matching the web UI's interactive budget). Values above `max_recursion_limit` are clamped. This field is read at dispatch, so the next scheduled run picks it up without a Gateway restart.
The background scheduler is single-instance by default. For a multi-pod deployment, set `scheduler.multi_instance: true` and use shared Postgres, `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; startup and periodic recovery then preserve live peer runs, atomically take over only expired leases, and fence stale post-launch writes. `max_concurrent_runs` is a shared global cap across Pods, including short-lived dispatch reservations. Without those settings, enable the scheduler on exactly one Gateway pod. These scheduler fields are startup-only; restart all Gateway Pods together when changing them.
### Upgrade Notes

View File

@ -17,7 +17,7 @@ DeerFlow is a LangGraph-based AI super agent system with a full-stack architectu
- Gateway streams `write_file` and `str_replace` argument deltas in bounded batches when clients also subscribe to `values`; messages-only consumers retain the original per-chunk contract, while `values` preserves the complete tool call.
- With `stream_subgraphs`, subgraph frames keep their namespace in the SSE event name (`values|<ns>`, LangGraph Platform style) instead of impersonating root frames — a delegated subagent inherits the parent checkpoint namespace, so publishing its `values` snapshot as bare `values` replaces the whole thread view in SDK clients (#4399). Root-only consumers (file-tool chunk batcher, subagent event persistence, LLM error-fallback detection) ignore namespaced frames. The web frontend does not request subgraph streaming; subtask progress rides root-namespace `task_*` custom events.
- Background subagent identity is deliberately split: the provider `tool_call_id` remains the correlation key for `ToolMessage`, `task_*` SSE events, persisted lifecycle events, frontend cards, and the public `ExtensionData.scope_id` contract (stored as `SubagentResult.external_task_id`), while `SubagentExecutor.execute_async()` generates a full server-side `execution_id` for `SubagentResult.task_id`, the process-wide registry, polling, cancellation, timeout handling, and cleanup. Provider IDs are not globally unique across parent runs, so they must never become registry ownership keys; scheduler closures retain their own `SubagentResult` rather than resolving ownership again through the mutable registry. Terminal subagent token usage travels in the current run's `ToolMessage.additional_kwargs` and is attributed from message state, never through a process-global provider-ID cache.
- Scheduled-task executions must reuse that same Gateway run lifecycle. The scheduler may decide *when* work runs, but it must dispatch through the existing run path rather than introducing a parallel execution stack.
- Scheduled-task executions must reuse that same Gateway run lifecycle. The scheduler may decide *when* work runs, but it must dispatch through the existing run path rather than introducing a parallel execution stack. Scheduled launches pass `scheduler.recursion_limit` (default 1000, matching the web UI's `recursion_limit: 1000`, clamped by `max_recursion_limit`) via `launch_scheduled_thread_run`; the value is read from `get_app_config()` at dispatch.
- The background scheduler is single-instance by default. `scheduler.multi_instance=true` opts into lease-aware recovery across Gateway instances and requires shared Postgres, `run_ownership.heartbeat_enabled=true`, and `run_events.backend=db`; otherwise startup rejects the configuration. Live scheduled runs are preserved when a peer starts; expired leases are atomically taken over, stale post-launch writes are fenced by the dispatch lease owner, and the Postgres advisory-locked budget makes `max_concurrent_runs` a shared global cap (including pre-launch reservations).
- Long-running MCP work uses a separate durable task runtime rather than keeping remote task IDs or status polling inside the Agent loop. Explicit `task_toolsets` bind raw submit/status/cancel names; only submit remains Agent-visible, and its wrapper persists the remote handle before returning a local ID. `McpTaskService` claims due rows with leases, resolves a protocol-specific `McpTaskDriver`, and writes normalized snapshots back to `mcp_tasks`; expired leases are the restart-recovery mechanism, and a result returned after expiry or after a cancel request must be discarded even when the owner token still matches. The first cancel request fences an in-flight poll lease, while repeats preserve an active cancellation lease so they cannot issue concurrent remote cancels; cancellation backoff starts when the remote attempt finishes, so a slow timeout cannot consume the retry delay. Cancellation, polling, and notification batches isolate per-task exceptions; an unexpected cancellation/poll failure leaves that record's lease to expire, while notification failures release only the affected lease for retry. Input-required and terminal event snapshots are delivered by idempotent Agent runs and marked delivered only after run success; the trusted notification instruction stays outside the input boundary while the serialized remote event is framed as untrusted data. A busy-thread conflict is normalized back to the service boundary so the queued snapshot coalesces to the latest task event. A missing dispatched run becomes a failed delivery attempt, while transient run-store hydration errors stay distinguishable and retry the same lookup. The database is the source of truth; `ThreadState` receives only a bounded current-thread projection, and display names are neutralized at that model-state boundary. The installed process-local submitter is the source of truth for management-tool exposure; hot `mcp_tasks` edits take effect only after restart, and active skills must explicitly declare the list/cancel business tools.
- MCP notification failures use a consecutive counter separate from the idempotency-key `dispatch_attempt`, capped exponential backoff, latest-event rebuilding before a run launches, and a five-attempt budget before `dead_letter`. A permanently missing/mismatched target thread is dead-lettered immediately instead of being recreated or reclaimed. HTTP and Agent cancellation requests return after the durable cancel fence; the background loop alone owns the potentially slow remote call and retry schedule. The HTTP cancel endpoint rejects requests with 503 when the loop is not running (`mcp_tasks_available` false, e.g. `mcp_tasks.enabled=false` with SQL persistence), so a cancellation is never acknowledged without a worker to perform it. The bounded notification error/count/status join poll and cancellation diagnostics in the task detail API and expanded card.

View File

@ -591,6 +591,43 @@ def _resolve_max_recursion_limit() -> int:
return _DEFAULT_MAX_RECURSION_LIMIT
def _resolve_scheduler_recursion_limit() -> int:
"""Resolve the scheduled-run recursion_limit from ``AppConfig.scheduler``.
Falls back to ``_DEFAULT_RECURSION_LIMIT`` when the app config cannot be
loaded so a missing ``config.yaml`` in tests still launches with the server
default rather than crashing dispatch. The value is clamped to
``max_recursion_limit`` here, at dispatch, so an operator value above the
ceiling never reaches ``build_run_config`` as an unclamped value which
would otherwise be misattributed as a "client" overage and emit a
``clamped client recursion_limit`` warning on every scheduled run.
``build_run_config`` keeps its own clamp for genuine client-supplied
bodies (defense in depth; its warning then only fires for real clients).
Both silent paths now emit an operator-visible warning: the pre-clamp
(operator value above ``max_recursion_limit``) and the config-load failure
fallback (``_DEFAULT_RECURSION_LIMIT`` returned instead of the operator
value).
"""
try:
raw = get_app_config().scheduler.recursion_limit
max_limit = _resolve_max_recursion_limit()
if raw > max_limit:
logger.warning(
"scheduler.recursion_limit %d exceeds max_recursion_limit %d; clamped to %d for scheduled runs",
raw,
max_limit,
max_limit,
)
return min(raw, max_limit)
except Exception:
logger.warning(
"failed to load app config; falling back to recursion_limit=%d for scheduled runs",
_DEFAULT_RECURSION_LIMIT,
)
return _DEFAULT_RECURSION_LIMIT
def _clamp_recursion_limit(value: Any, max_limit: int) -> int:
"""Clamp a client-supplied ``recursion_limit`` into a safe server range.
@ -1420,7 +1457,7 @@ async def launch_scheduled_thread_run(
input={"messages": [{"role": "user", "content": prompt}]},
command=None,
metadata=metadata or {},
config=None,
config={"recursion_limit": _resolve_scheduler_recursion_limit()},
# ``user_id`` mirrors what IM channels put in ``body.context`` so
# runtime-context consumers without a ContextVar fallback (e.g.
# user-scoped GuardrailMiddleware providers) see the owning user;

View File

@ -127,8 +127,10 @@ in a single run. The unified Gateway path defaults to `100` in
`build_run_config` (see `backend/app/gateway/services.py`), which is a safer
starting point for plan-mode or subagent-heavy runs. Clients can still set
`recursion_limit` explicitly in the request body; increase it if you run deeply
nested subagent graphs. For safety, the Gateway clamps any client-supplied value
to a configurable server ceiling (`max_recursion_limit` in `config.yaml`,
nested subagent graphs. Scheduled-task launches do not take a client body: they
use `scheduler.recursion_limit` from `config.yaml` (default `1000`, matching
the web UI). For safety, the Gateway clamps any supplied
value to a configurable server ceiling (`max_recursion_limit` in `config.yaml`,
default `1000`) so a single run cannot execute unbounded graph steps (runaway
LLM cost / DoS); invalid or non-positive values fall back to the `100` default.
@ -1037,4 +1039,5 @@ curl -X POST http://localhost:2026/api/langgraph/threads/abc123/runs/stream \
> The unified Gateway path defaults `config.recursion_limit` to 100 for
> plan-mode and subagent-heavy runs. Clients may still set
> `config.recursion_limit` explicitly — see the [Create Run](#create-run)
> section for details.
> section for details. Scheduled-task launches use
> `scheduler.recursion_limit` from `config.yaml` instead of a client body.

View File

@ -242,6 +242,7 @@ scheduler:
lease_seconds: 120
max_concurrent_runs: 3
min_once_delay_seconds: 60
recursion_limit: 1000
```
Notes:
@ -250,7 +251,8 @@ Notes:
- `multi_instance: true` opts into lease-aware scheduler recovery across Gateway instances. It requires Postgres, `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; otherwise startup fails fast. Leave it false for the default single-instance scheduler.
- `max_concurrent_runs` is a shared global cap in multi-instance mode. It counts active `queued`/`running` scheduled-run rows plus valid pre-launch dispatch leases, and Postgres serializes the budget read with due-task claims so long runs or concurrent Pods cannot exceed it.
- Multi-instance reconciliation uses the run ownership lease: a live peer run is preserved, an expired lease is atomically taken over before its scheduled row is interrupted, and a stale Pod cannot overwrite a newer Pod's parent-task bookkeeping.
- All scheduler fields are restart-required; edits need a Gateway restart.
- `recursion_limit` is the LangGraph super-step cap for scheduler-launched runs (default 1000, matching the web UI's interactive budget). Values above `max_recursion_limit` (default 1000) are clamped. This field is read at dispatch, so a YAML edit applies to the next scheduled run without a Gateway restart.
- Poller fields (`enabled`, `multi_instance`, `poll_interval_seconds`, `lease_seconds`, `max_concurrent_runs`, `min_once_delay_seconds`) are restart-required; edits need a Gateway restart.
- **Upgrade note:** before upgrading a deployment with `GATEWAY_WORKERS > 1` and `scheduler.enabled: true`, either run the scheduler on exactly one Gateway worker or enable `scheduler.multi_instance: true` with shared Postgres, `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`. The startup gate now rejects the unsafe combination instead of allowing it to start silently.
- **Upgrade note:** in multi-instance mode, `max_concurrent_runs` is cluster-wide rather than per Pod and includes active scheduled runs plus dispatch reservations. Plan capacity accordingly; it does not multiply with the replica count.
- **Upgrade note:** `scheduler.multi_instance` and its related scheduler, ownership, and run-event settings are startup-only. Restart all Gateway Pods together after changing them; a ConfigMap update without a coordinated restart leaves the running service on its previous mode.

View File

@ -10,7 +10,7 @@ Setup: Copy `config.example.yaml` to `config.yaml` in the **project root** direc
**Config Hot-Reload Boundary**: Gateway dependencies route through `get_app_config()` on every request, so per-run fields like `models[*].max_tokens`, `summarization.*`, `title.*`, `memory.*`, `subagents.*`, `verification.*`, `tools[*]`, and the agent system prompt pick up `config.yaml` edits on the next message. `AppConfig` is intentionally **not** cached on `app.state``lifespan()` keeps a local `startup_config` variable for one-shot bootstrap work and passes it to `langgraph_runtime(app, startup_config)`.
Infrastructure fields are **restart-required**. The authoritative list lives in `packages/harness/deerflow/config/reload_boundary.py::STARTUP_ONLY_FIELDS` and is mirrored by the standardised `"startup-only:"` prefix on the corresponding `Field(description=...)` in `AppConfig`, so IDE hover on those fields surfaces the reason inline (no need to context-switch into this table). Currently registered: `plugins`, `database`, `checkpointer`, `run_events`, `stream_bridge`, `sandbox`, `log_level`, `logging`, `channels`, `channel_connections`, `scheduler`, `mcp_tasks`, `run_ownership`. Adding a new restart-required field requires updating the registry; drift is pinned by `tests/test_reload_boundary.py`.
Infrastructure fields are **restart-required**. The authoritative list lives in `packages/harness/deerflow/config/reload_boundary.py::STARTUP_ONLY_FIELDS` and is mirrored by the standardised `"startup-only:"` prefix on the corresponding `Field(description=...)` in `AppConfig`, so IDE hover on those fields surfaces the reason inline (no need to context-switch into this table). Currently registered: `plugins`, `database`, `checkpointer`, `run_events`, `stream_bridge`, `sandbox`, `log_level`, `logging`, `channels`, `channel_connections`, `scheduler`, `mcp_tasks`, `run_ownership`. Adding a new restart-required field requires updating the registry; drift is pinned by `tests/test_reload_boundary.py`. `scheduler.recursion_limit` is the exception inside that section: it is read from `get_app_config()` at each scheduled dispatch, so a YAML edit applies to the next run without restarting the poller.
**Persistence backend resolution**: the unified `database` section selects the
Gateway's LangGraph checkpointer, LangGraph Store, and DeerFlow SQL repositories.

View File

@ -69,7 +69,8 @@ STARTUP_ONLY_FIELDS: dict[str, str] = {
"scheduler": (
"ScheduledTaskService is constructed and started once during Gateway lifespan startup; enabled, poll_interval_seconds, lease_seconds, "
"max_concurrent_runs, and multi_instance are captured into the service instance and the background poller task is not rebuilt on config.yaml edits. "
"Changing multi-instance recovery prerequisites or lease behavior requires restarting every Gateway Pod together."
"Changing multi-instance recovery prerequisites or lease behavior requires restarting every Gateway Pod together. "
"scheduler.recursion_limit is not captured there: launch_scheduled_thread_run reads it from get_app_config() on each dispatch, so a YAML edit applies to the next scheduled run without a Gateway restart."
),
"mcp_tasks": (
"McpTaskService is constructed and started once during Gateway lifespan startup; enabled, poll_interval_seconds, lease_seconds, "

View File

@ -8,3 +8,14 @@ class SchedulerConfig(BaseModel):
lease_seconds: int = Field(default=120, ge=5, le=3600)
max_concurrent_runs: int = Field(default=3, ge=1, le=32)
min_once_delay_seconds: int = Field(default=60, ge=1, le=86400)
recursion_limit: int = Field(
default=1000,
ge=1,
description=(
"LangGraph recursion_limit for scheduler-launched runs. Read at dispatch "
"time (not captured into ScheduledTaskService). The default matches the "
"web UI's interactive budget (1000) so scheduled and interactive runs "
"behave identically out of the box. Values above "
"AppConfig.max_recursion_limit are clamped."
),
)

View File

@ -2274,6 +2274,7 @@ def test_launch_scheduled_thread_run_marks_context_non_interactive(_stub_app_con
assert captured["thread_id"] == "thread-scheduled"
assert isinstance(captured["body"], RunCreateRequest)
assert captured["body"].config == {"recursion_limit": 1000}
assert captured["context"] == {"non_interactive": True, "user_id": "user-1"}
assert captured["metadata"] == {"scheduled_task_id": "task-1"}
assert captured["if_not_exists"] == "create"
@ -2281,6 +2282,125 @@ def test_launch_scheduled_thread_run_marks_context_non_interactive(_stub_app_con
assert result == {"run_id": "run-1", "thread_id": "thread-scheduled"}
def test_launch_scheduled_thread_run_uses_configured_recursion_limit(_stub_app_config):
import asyncio
from types import SimpleNamespace
from unittest.mock import patch
from app.gateway.services import launch_scheduled_thread_run
from deerflow.config.app_config import AppConfig, set_app_config
set_app_config(
AppConfig.model_validate(
{
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
"scheduler": {"recursion_limit": 1000},
}
)
)
async def _scenario():
captured: dict[str, object] = {}
async def fake_start_run(body, thread_id, request):
captured["config"] = body.config
return SimpleNamespace(run_id="run-1", thread_id=thread_id)
with patch("app.gateway.services.start_run", side_effect=fake_start_run):
await launch_scheduled_thread_run(
thread_id="thread-scheduled",
assistant_id="lead_agent",
prompt="Run in background",
app=SimpleNamespace(state=SimpleNamespace()),
owner_user_id="user-1",
)
return captured
captured = asyncio.run(_scenario())
assert captured["config"] == {"recursion_limit": 1000}
def test_launch_scheduled_thread_run_recursion_limit_is_clamped_to_ceiling(_stub_app_config, caplog):
"""A scheduler.recursion_limit above max_recursion_limit is clamped at dispatch, so the run request never carries an unclamped value."""
import asyncio
import logging
from types import SimpleNamespace
from unittest.mock import patch
from app.gateway.services import launch_scheduled_thread_run
from deerflow.config.app_config import AppConfig, set_app_config
set_app_config(
AppConfig.model_validate(
{
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
"max_recursion_limit": 1000,
"scheduler": {"recursion_limit": 5000},
}
)
)
async def _scenario():
captured: dict[str, object] = {}
async def fake_start_run(body, thread_id, request):
captured["config"] = body.config
return SimpleNamespace(run_id="run-1", thread_id=thread_id)
with patch("app.gateway.services.start_run", side_effect=fake_start_run):
await launch_scheduled_thread_run(
thread_id="thread-scheduled",
assistant_id="lead_agent",
prompt="Run in background",
app=SimpleNamespace(state=SimpleNamespace()),
owner_user_id="user-1",
)
return captured
caplog.set_level(logging.WARNING, logger="app.gateway.services")
captured = asyncio.run(_scenario())
assert captured["config"] == {"recursion_limit": 1000}
assert any("scheduler.recursion_limit 5000 exceeds max_recursion_limit 1000" in r.message for r in caplog.records)
def test_launch_scheduled_thread_run_falls_back_when_config_unloadable(_stub_app_config, caplog):
"""When the app config cannot be loaded, dispatch falls back to the server default recursion limit and logs a warning."""
import asyncio
import logging
from types import SimpleNamespace
from unittest.mock import patch
from app.gateway.services import launch_scheduled_thread_run
async def _scenario():
captured: dict[str, object] = {}
async def fake_start_run(body, thread_id, request):
captured["config"] = body.config
return SimpleNamespace(run_id="run-1", thread_id=thread_id)
with (
patch(
"app.gateway.services.get_app_config",
side_effect=RuntimeError("config unavailable"),
),
patch("app.gateway.services.start_run", side_effect=fake_start_run),
):
await launch_scheduled_thread_run(
thread_id="thread-scheduled",
assistant_id="lead_agent",
prompt="Run in background",
app=SimpleNamespace(state=SimpleNamespace()),
owner_user_id="user-1",
)
return captured
caplog.set_level(logging.WARNING, logger="app.gateway.services")
captured = asyncio.run(_scenario())
assert captured["config"] == {"recursion_limit": 100}
assert any("failed to load app config; falling back to recursion_limit=100" in r.message for r in caplog.records)
def test_launch_scheduled_thread_run_rejects_legacy_auth_token():
"""The internal launcher shares run admission; task API/model state has no metadata field."""
import asyncio

View File

@ -0,0 +1,26 @@
"""Tests for SchedulerConfig schema."""
import pytest
from pydantic import ValidationError
from deerflow.config.scheduler_config import SchedulerConfig
def test_scheduler_config_defaults():
config = SchedulerConfig()
assert config.enabled is False
assert config.recursion_limit == 1000
def test_scheduler_config_accepts_positive_recursion_limit():
assert SchedulerConfig(recursion_limit=1).recursion_limit == 1
assert SchedulerConfig(recursion_limit=1000).recursion_limit == 1000
def test_scheduler_config_rejects_non_positive_recursion_limit():
with pytest.raises(ValidationError):
SchedulerConfig(recursion_limit=0)
with pytest.raises(ValidationError):
SchedulerConfig(recursion_limit=-5)

View File

@ -2116,7 +2116,12 @@ agent_storage:
# Scheduled Tasks Configuration
# ============================================================================
# Background scheduler for one-time and recurring (cron) agent runs.
# All fields are restart-required (captured at Gateway lifespan startup).
# Poller fields (enabled, multi_instance, poll_interval_seconds, lease_seconds,
# max_concurrent_runs, min_once_delay_seconds) are restart-required.
# recursion_limit is read at dispatch and applies to the next scheduled run
# without a Gateway restart. The default (1000) matches the web UI's
# interactive budget so scheduled and interactive runs behave identically out
# of the box. Values above max_recursion_limit are clamped.
#
# The scheduler is single-instance by default. Set multi_instance: true only
# when every Gateway instance shares Postgres, run ownership heartbeats, and
@ -2124,10 +2129,12 @@ agent_storage:
#
# scheduler:
# enabled: false # Master switch for the background poller
# multi_instance: false # Opt into lease-aware recovery across Gateway instances
# poll_interval_seconds: 5 # How often to scan for due tasks
# lease_seconds: 120 # Claim lease; a crashed process's task becomes reclaimable after this
# max_concurrent_runs: 3 # Global cap on active scheduled runs/reservations across multi-instance Pods
# min_once_delay_seconds: 60 # Minimum future offset for one-time tasks at creation time
# recursion_limit: 1000 # LangGraph super-step cap for scheduled runs (matches the web UI)
scheduler:
enabled: false
multi_instance: false # Opt into lease-aware recovery across Gateway instances
@ -2135,6 +2142,7 @@ scheduler:
lease_seconds: 120
max_concurrent_runs: 3
min_once_delay_seconds: 60
recursion_limit: 1000
# ============================================================================
# Long-running MCP Tasks Configuration