diff --git a/CHANGELOG.md b/CHANGELOG.md index 2255c102e8..ef5f14d253 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -135,6 +135,21 @@ This release closes that milestone with **301 merged pull requests**. duplicate keys, and syntactically valid large numbers are allowed; a UTF-8 BOM is rejected. Syntax only — no schema or business-field checks; existing criteria and verdict semantics are unchanged. ([#5947]) +- **gateway:** The multi-process startup gate now fires on an explicit + `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`) as well + as `GATEWAY_WORKERS > 1`. The worker-count variables only see one process + tree, so a Kubernetes Deployment with several one-worker Pods passed every + check while each Pod's startup orphan reconciliation wrote the other Pods' + lease-less runs off as crashed on every rolling update. Both paths require + Postgres, `run_events.backend: db`, `run_ownership.heartbeat_enabled: true` + and now also a Redis stream bridge, and refuse an explicit + `sandbox.ownership.type: memory`, process-local browser tools and a + scheduler without `scheduler.multi_instance`. The agent-storage divergence + warning, the inbound webhook dedupe warning and the WeChat QR-login guard + honor the same declaration. **Behavior change:** `GATEWAY_WORKERS > 1` with + the memory stream bridge no longer starts; configure `stream_bridge.type: + redis` or `DEER_FLOW_STREAM_BRIDGE_REDIS_URL` (docker-compose and the Helm + chart already inject it). ([#6328]) #### Memory @@ -7799,3 +7814,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6313]: https://github.com/bytedance/deer-flow/pull/6313 [#6319]: https://github.com/bytedance/deer-flow/pull/6319 [#6326]: https://github.com/bytedance/deer-flow/pull/6326 +[#6328]: https://github.com/bytedance/deer-flow/pull/6328 diff --git a/README.md b/README.md index b4d8198502..039da2429c 100644 --- a/README.md +++ b/README.md @@ -472,7 +472,7 @@ Browser login uses `HttpOnly` session cookies. The login page offers a "keep me DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser-facing scheme and origin behind a proxy. The bundled nginx sets `X-Forwarded-Proto`, but preserves an upstream HTTPS value and does not overwrite every forwarded header. Configure the outer trusted proxy to replace or strip client-supplied forwarding headers before traffic reaches DeerFlow. > [!IMPORTANT] -> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination. +> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination. > > In single-process JSONL deployments, cancelling an admitted event-store mutation waits for its background file I/O, rollback, and bookkeeping to settle before releasing the thread write lock. This prevents an older cancelled write from recreating deleted records or rolling back a later successful write. Cancellation can therefore wait on slow storage; it does not stop an in-flight filesystem operation. Callers still waiting to acquire the lock can cancel without starting a mutation. A batch spanning multiple threads drains its current thread group before propagating cancellation; subsequent thread groups do not start. > diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 4ce0008179..79fc184adf 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -18,6 +18,7 @@ The backend runs a LangGraph-based super agent with sandbox execution, persisten - Scheduled tasks dispatch through the normal Gateway run path. `launch_scheduled_thread_run` reads `get_app_config()` at dispatch and passes `scheduler.recursion_limit` (default 1000, matching the web UI; clamped by `max_recursion_limit`), so YAML changes apply on the next run without restarting Gateway. - Run-history `status` filters are occurrence states, not task states. `ScheduledTaskRunStatus` in `persistence/scheduled_tasks/model.py` is the shared API/repository vocabulary and must match the active and terminal occurrence-status sets. Keep owner lookup before reading history, and apply SQL task/status predicates before pagination; omitted status preserves the existing response. - 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 launch claims return to the durable queue, expired run leases are atomically taken over, stale launch writes are fenced by lease ownership, and the Postgres advisory-locked budget makes `max_concurrent_runs` a shared global cap for `launching`/`running` rows. +- The multi-process startup gate also fires on `deployment.multi_instance: true`; contract in `docs/CONFIGURATION.md`. - Long-running MCP work uses a separate durable task runtime (`McpTaskService` + `mcp_tasks`, lease-based recovery) rather than keeping remote task IDs or status polling inside the Agent loop; only submit remains Agent-visible, the database is the source of truth, and `ThreadState` receives only a bounded current-thread projection. Full contract (leases, cancellation fencing, delivery idempotency, management-tool exposure): [packages/harness/deerflow/mcp/AGENTS.md](packages/harness/deerflow/mcp/AGENTS.md). - MCP task notification retries, dead-lettering, and the cancel endpoint's worker-stopped 503 are part of that same contract — see [packages/harness/deerflow/mcp/AGENTS.md](packages/harness/deerflow/mcp/AGENTS.md). - Scheduled-task dispatch permits one active occurrence per task via `uq_scheduled_task_run_active` (`task_id WHERE status IN ('queued','launching','running')`). Durable `queued` rows survive restarts; only lease-fenced `launching` may call Gateway launch; `running` references the durable run. Stable admission idempotency keys reuse that run after recovery. Reused-thread `ConflictError` returns `launching` to `queued`; other launch errors become `failed`. Atomic queue claims enforce `max_concurrent_runs`, excluding waiting rows; the budget count and its UPDATE are separate statements, so writers must serialize before the count (Postgres advisory lock; SQLite `BEGIN IMMEDIATE`, whose deferred transaction otherwise reserves the writer only at the UPDATE) or claims on distinct rows overshoot the cap. Repeated triggers coalesce; same-thread FIFO blocks behind older active rows. Queue admission, PATCH/resume, pause and delete lock the parent before the occurrence, freezing active task definitions. Pause/delete atomically cancel `queued` work but reject `launching`/`running`; PATCH/resume reject all active states. Only queued conflicts offer pause cancellation. Manual triggers may queue/run while paused. Recovery locks task/run pairs in task-id/run-id order and restores `run_id`, `started_at` and live errors before releasing launch claims. Launch/failure/timeout updates use one parent-first transaction to prevent interleaved claims. Queue timeout fails the occurrence and advances scheduled work to prevent immediate requeue. Repository boundaries coerce serialized timestamps before SQL `DateTime` binding. diff --git a/backend/app/channels/dedupe_store.py b/backend/app/channels/dedupe_store.py index effb949816..5a59f8279c 100644 --- a/backend/app/channels/dedupe_store.py +++ b/backend/app/channels/dedupe_store.py @@ -17,6 +17,8 @@ from typing import Any, Protocol from sqlalchemy import text +from deerflow.config.deployment_config import multi_instance_declaration + logger = logging.getLogger(__name__) INBOUND_DEDUPE_TTL_SECONDS = 10 * 60 @@ -269,7 +271,11 @@ def make_inbound_dedupe_store(app_config: Any | None = None) -> InboundDedupeSto db_is_postgres = db_backend == "postgres" workers, worker_env = _gateway_workers() - multi_worker = workers > 1 + # The worker count only sees one process tree; Kubernetes replicas declare + # themselves through deployment.multi_instance / DEER_FLOW_MULTI_INSTANCE. + declaration = multi_instance_declaration(app_config) + multi_worker = workers > 1 or declaration is not None + topology = f"{worker_env}>1" if workers > 1 else getattr(declaration, "knob", None) if backend == "postgres": if not db_is_postgres: @@ -282,7 +288,7 @@ def make_inbound_dedupe_store(app_config: Any | None = None) -> InboundDedupeSto if backend == "memory": if multi_worker: logger.warning( - f"dedupe_storage=memory with {worker_env}>1: inbound webhook dedupe " + f"dedupe_storage=memory with {topology}: inbound webhook dedupe " "is per-pod and will NOT drop redeliveries routed to a different replica. " "Use dedupe_storage=postgres (or remove the setting to let 'auto' pick it) " "for multi-worker deployments. See issue #4120." diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 71cec1c1f6..f2171ff788 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -180,8 +180,8 @@ without changed outputs keep ordinary chat behavior. Journal mechanics (callback attribution, receipt idempotency and retries, orphan recovery): `packages/harness/deerflow/runtime/AGENTS.md`. Multi-worker deployments require `run_events.backend: db` for shared, ordered delivery events; the -startup gate rejects process-local memory and JSONL event stores when -`GATEWAY_WORKERS > 1`. +startup gate rejects memory/JSONL event stores when `GATEWAY_WORKERS > 1` +or `deployment.multi_instance: true`. **RunManager / RunStore contract**: - LangGraph-compatible run requests validate their supported subset before creating a run. `runtime/stream_modes.py` is the shared backend contract for public stream modes and the worker's `graph.astream` mapping; the public `messages-tuple` mode maps to LangGraph's internal `messages` mode, while public `messages`, `events`, and other unsupported modes are rejected instead of being dropped or replaced with `values`. `app/gateway/run_models.py::RunCreateRequest` is shared by HTTP and internal scheduled launch paths, retains only truthful compatibility defaults for unimplemented options (`if_not_exists="create"` plus `None` placeholders), returns 422 for unsupported values including `on_completion="complete"`, `on_completion="continue"`, and `multitask_strategy="enqueue"`, and forbids undeclared SDK options so fields such as `checkpoint_during` and `durability` cannot be silently discarded. A placeholder must still accept the stock SDK's own default: `langgraph_sdk` drops only `None` from its run payload, so `stream_resumable=False` reaches every request and means "non-resumable", which is what DeerFlow serves — rejecting it 422'd every IM channel run (#4466). `tests/test_run_request_validation.py::test_gateway_accepts_langgraph_sdk_default_payload` pins the real SDK payload against this boundary; channel tests mock the SDK client and cannot catch this class of drift. diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index 00a3305dc2..77ac3ea486 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -29,6 +29,7 @@ from langgraph.types import Checkpointer from deerflow.community.browser_automation.session import browser_multi_worker_error from deerflow.config.app_config import AppConfig, get_app_config +from deerflow.config.deployment_config import multi_instance_declaration from deerflow.persistence.feedback import FeedbackRepository from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, RunContext, RunManager, StreamBridge from deerflow.runtime.events.store.base import RunEventStore @@ -86,9 +87,55 @@ def _gateway_worker_count() -> tuple[int, str]: return 1, _WORKER_COUNT_ENV_VARS[0] +def _multi_process_signal(config: AppConfig) -> tuple[str, str] | None: + """Return ``(reason, rollback)`` when this process must assume peer Gateway processes. + + ``reason`` names the knob that established the topology (``GATEWAY_WORKERS=2``, + ``DEER_FLOW_MULTI_INSTANCE=1`` or ``deployment.multi_instance=true``) and + ``rollback`` is the single-instance remediation a refusal message offers. + When the worker count and a declaration are both active, ``rollback`` + withdraws both: resetting only the worker count would bounce the operator + through a second refusal on the declaration at the next start. + + The worker-count variables only see one process tree, so a Kubernetes + Deployment with ``replicas > 1`` and one worker per Pod is invisible to + them; the explicit declaration exists for exactly that topology. + """ + workers, worker_env = _gateway_worker_count() + declaration = multi_instance_declaration(config) + if workers > 1: + rollback = f"Set {worker_env}=1" + if declaration is not None: + rollback = f"{rollback} and {declaration.rollback}" + return f"{worker_env}={workers}", rollback + if declaration is None: + return None + step = declaration.rollback + return declaration.knob, f"{step[0].upper()}{step[1:]} and run a single Gateway instance" + + +def _stream_bridge_is_cross_process(config: AppConfig) -> bool: + """Return whether live run events can reach SSE clients on every instance. + + Mirrors ``runtime/stream_bridge/async_provider.py::_resolve_config``: the + config.yaml section wins, and an omitted section falls back to the + ``DEER_FLOW_STREAM_BRIDGE_REDIS_URL`` variable the Docker and Helm + deployments inject. + """ + bridge = getattr(config, "stream_bridge", None) + if bridge is not None: + return getattr(bridge, "type", None) == "redis" + return bool((os.environ.get("DEER_FLOW_STREAM_BRIDGE_REDIS_URL") or "").strip()) + + def _enforce_postgres_for_multi_worker(config: AppConfig) -> None: """Refuse unsafe multi-process configurations before persistence starts. + A deployment counts as multi-process when ``GATEWAY_WORKERS`` / + ``WEB_CONCURRENCY`` is above 1, or when the operator declares peers with + ``deployment.multi_instance: true`` / ``DEER_FLOW_MULTI_INSTANCE=1`` (the + worker count cannot see other Pods, and a Pod that starts next to a peer + without these prerequisites writes the peer's live runs off as orphans). Multi-instance scheduler recovery also needs the durable run ownership contract even when each Pod runs a single Gateway worker. @@ -105,13 +152,16 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None: every run has a NULL lease, so reconciliation treats all inflight runs as orphans and Worker B would kill Worker A's live runs on every rolling update or scale-up. + 6. The stream bridge must be Redis. The memory bridge is process-local, + so SSE streams, reconnects and ``/wait`` only work on the owner. + 7. ``sandbox.ownership.type`` must not be an explicit ``memory``: an + in-process ownership store cannot see peers, so reconciliation would + adopt and idle-destroy another instance's live containers (#4206). This gate runs once at startup before any persistence engine is initialised so the error message is clear and the process exits immediately. """ - workers, worker_env = _gateway_worker_count() - scheduler = getattr(config, "scheduler", None) multi_instance_requested = bool(getattr(scheduler, "multi_instance", False)) multi_instance_scheduler = bool(getattr(scheduler, "enabled", False) and multi_instance_requested) @@ -127,34 +177,54 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None: if multi_instance_requested and (run_ownership is None or not run_ownership.heartbeat_enabled): raise SystemExit("scheduler.multi_instance=true requires run_ownership.heartbeat_enabled=true so peer runs retain a valid lease. Set scheduler.multi_instance=false or enable run ownership heartbeats.") - if workers <= 1: + signal = _multi_process_signal(config) + if signal is None: return + reason, rollback = signal if config.scheduler.enabled and not multi_instance_scheduler: - raise SystemExit(f"{worker_env}={workers} cannot run with scheduler.enabled=true because each worker starts its own scheduler. Set {worker_env}=1, scheduler.multi_instance=true, or scheduler.enabled=false.") + raise SystemExit(f"{reason} cannot run with scheduler.enabled=true because each worker starts its own scheduler. {rollback}, scheduler.multi_instance=true, or scheduler.enabled=false.") if _browser_tools_enabled_in_config(config): - raise SystemExit(browser_multi_worker_error(workers)) + workers, _worker_env = _gateway_worker_count() + if workers > 1 and multi_instance_declaration(config) is None: + raise SystemExit(browser_multi_worker_error(workers)) + raise SystemExit(f"{reason} cannot enable agentic browser tools: browser sessions are process-local and a request can land on any instance. {rollback} or disable the browser_navigate tool.") if backend != "postgres": - raise SystemExit(f"{worker_env}={workers} requires database.backend='postgres', but database.backend is '{backend}'. SQLite cannot support concurrent multi-process access. Set {worker_env}=1 or switch to Postgres.") + raise SystemExit(f"{reason} requires database.backend='postgres', but database.backend is '{backend}'. SQLite cannot support concurrent multi-process access. {rollback} or switch to Postgres.") if run_events_backend != "db": raise SystemExit( - f"{worker_env}={workers} requires run_events.backend='db', but run_events.backend is '{run_events_backend}'. " + f"{reason} requires run_events.backend='db', but run_events.backend is '{run_events_backend}'. " "Memory and JSONL event stores are process-local, so delivery receipt singleton guarantees cannot hold across workers. " - f"Set {worker_env}=1 or configure run_events.backend: db." + f"{rollback} or configure run_events.backend: db." ) if run_ownership is None or not run_ownership.heartbeat_enabled: raise SystemExit( - f"{worker_env}={workers} requires run_ownership.heartbeat_enabled=true. " + f"{reason} requires run_ownership.heartbeat_enabled=true. " "Without heartbeat, every run has a NULL lease, so reconciliation " - "treats all inflight runs as orphans — Worker B would kill Worker A's " - "live runs on every rolling update or scale-up. " + "treats all inflight runs as orphans — a starting instance would kill its " + "peers' live runs on every rolling update or scale-up. " "Set run_ownership.heartbeat_enabled=true in config.yaml." ) + if not _stream_bridge_is_cross_process(config): + raise SystemExit( + f"{reason} requires stream_bridge.type='redis' (or DEER_FLOW_STREAM_BRIDGE_REDIS_URL), " + "but the stream bridge is the process-local memory bridge: SSE streams, reconnects and /wait " + f"only work on the instance that owns the run. {rollback} or configure the redis stream bridge." + ) + + ownership = getattr(getattr(config, "sandbox", None), "ownership", None) + if ownership is not None and getattr(ownership, "type", None) == "memory": + raise SystemExit( + f"{reason} cannot run with sandbox.ownership.type='memory': an in-process ownership store cannot see " + "peer instances, so startup reconciliation would adopt and idle-destroy containers another instance is " + "using (#4206). Set sandbox.ownership.type='redis', or omit the section so it is inferred from the redis stream bridge." + ) + def _validate_agent_storage(config: AppConfig) -> None: """Fail fast on an agent-storage backend the database cannot support. @@ -178,14 +248,13 @@ def _validate_agent_storage(config: AppConfig) -> None: f"but database.backend is '{db_backend}'. A 'memory' database is per-process and cannot " "share agent definitions across nodes. Set database.backend, or use agent_storage.backend='file'." ) - workers, worker_env = _gateway_worker_count() - if workers > 1 and db_backend == "postgres" and backend == "file": + signal = _multi_process_signal(config) + if signal is not None and db_backend == "postgres" and backend == "file": logger.warning( - "%s=%s with database.backend='postgres' but agent_storage.backend='file': " + "%s with database.backend='postgres' but agent_storage.backend='file': " "custom agents and managed subagents are stored per-node on local disk and are not visible " "across workers/nodes. Set agent_storage.backend='db' to share them.", - worker_env, - workers, + signal[0], ) diff --git a/backend/app/gateway/routers/channel_connections.py b/backend/app/gateway/routers/channel_connections.py index dca57623fc..3e8858953c 100644 --- a/backend/app/gateway/routers/channel_connections.py +++ b/backend/app/gateway/routers/channel_connections.py @@ -21,6 +21,7 @@ from app.channels.wechat_qr_login import QRLoginError, WechatQRLogin from app.gateway.deps import require_admin_user from app.gateway.persistent_writes import run_drained_write from deerflow.config.channel_connections_config import ChannelConnectionsConfig +from deerflow.config.deployment_config import multi_instance_declaration from deerflow.persistence.channel_connections import ChannelConnectionRepository from deerflow.persistence.engine import get_session_factory from deerflow.utils.file_io import await_drained @@ -772,10 +773,14 @@ async def _require_wechat_qr_login(request: Request) -> ChannelConnectionsConfig workers = int(os.environ.get("GATEWAY_WORKERS") or os.environ.get("WEB_CONCURRENCY") or "1") except ValueError: workers = 0 - if workers != 1: + # Kubernetes replicas run one worker each; they declare their peers instead. + if workers != 1 or multi_instance_declaration(_get_app_config()) is not None: raise HTTPException( status_code=503, - detail="WeChat QR login requires a single Gateway worker. Set GATEWAY_WORKERS=1 (or WEB_CONCURRENCY=1 when using Uvicorn directly), or enter a bot token manually.", + detail=( + "WeChat QR login requires a single Gateway worker. Set GATEWAY_WORKERS=1 (or WEB_CONCURRENCY=1 when using Uvicorn directly), " + "run one Gateway instance without deployment.multi_instance / DEER_FLOW_MULTI_INSTANCE, or enter a bot token manually." + ), ) return config diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 2bac50f0ad..a34dad587a 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -518,7 +518,7 @@ Notes: - **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 counts `launching`/`running` occurrences. Waiting `queued` rows remain outside the execution cap; capacity 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. -- Multi-worker deployments (`GATEWAY_WORKERS > 1`) must use the Postgres database backend, enable run ownership heartbeats, and set `run_events.backend: db`. SQLite silently ignores row-level locks, while memory and JSONL run-event stores are process-local and cannot enforce singleton delivery receipts across workers; startup rejects these combinations. The process-local agentic browser tool group is incompatible with multiple Gateway workers; keep the Gateway to one worker process while `browser_navigate` is enabled — `GATEWAY_WORKERS=1`, plus `WEB_CONCURRENCY` unset or `1` on launches that pass uvicorn no worker count (`backend/Dockerfile`, `scripts/serve.sh`), where uvicorn takes the process count from it. Browser control also requires the backend `browser` extra (`cd backend && uv sync --extra browser && uv run playwright install chromium`); startup detects enabled browser config and fails fast when Playwright is missing, and `/api/features` reports `browser_control.enabled=false` until the runtime is available. +- Multi-worker deployments (`GATEWAY_WORKERS > 1`) and declared multi-instance deployments (`deployment.multi_instance: true` / `DEER_FLOW_MULTI_INSTANCE=1`, see [Deployment topology](#deployment-topology-multi-instance)) must use the Postgres database backend, enable run ownership heartbeats, set `run_events.backend: db`, and use the Redis stream bridge. SQLite silently ignores row-level locks, while memory and JSONL run-event stores are process-local and cannot enforce singleton delivery receipts across workers; startup rejects these combinations. The process-local agentic browser tool group is incompatible with multiple Gateway workers; keep the Gateway to one worker process while `browser_navigate` is enabled — `GATEWAY_WORKERS=1`, plus `WEB_CONCURRENCY` unset or `1` on launches that pass uvicorn no worker count (`backend/Dockerfile`, `scripts/serve.sh`), where uvicorn takes the process count from it. Browser control also requires the backend `browser` extra (`cd backend && uv sync --extra browser && uv run playwright install chromium`); startup detects enabled browser config and fails fast when Playwright is missing, and `/api/features` reports `browser_control.enabled=false` until the runtime is available. - The MVP supports thread reuse and fresh-thread-per-run execution modes. - Create/update accept optional `assistant_id` (`lead_agent` by default, or an existing custom agent for the task owner). - Create/update accept `once`, `cron`, and `interval`. Interval uses `schedule_spec.every_seconds` (UTC `now + N`, no missed-beat catch-up). N is at least `min_once_delay_seconds` (default 60) and at most 30 days. @@ -526,6 +526,21 @@ Notes: - Scheduled task definitions and task-run history are persisted in the application database. - With `channel_connections.enabled: true`, the scheduler enqueues IM outcome notifications for the task owner's connected identities. A delivery worker (same poll cadence as the scheduler) pushes them. Scheduled runs that finish as success or failed notify. For goal tasks, an `unmet` scheduled occurrence sends a goal-unmet notice instead, and an automatic pause after three unmet occurrences sends an auto-pause notice; a stop requested by the agent adds no separate notice. Manual triggers and interrupts stay silent, and so do occurrences that end without a finished run (launch error, queue timeout, restart recovery). Channel/transport outages park rows without consuming the retry budget, for up to about a day; platform rejections exhaust ~15 minutes of counted retries then settle `failed`. The worker re-checks the binding right before sending: a target the owner has disconnected since enqueue is dropped as `failed`, never pushed. Only WeCom currently implements proactive `send_notification`. +### Deployment topology (multi-instance) + +```yaml +deployment: + multi_instance: false +``` + +Notes: + +- `GATEWAY_WORKERS` / `WEB_CONCURRENCY` only count the uvicorn workers of one process tree. A Kubernetes Deployment with `replicas > 1` runs one worker per Pod, so every Pod reports a single worker and the multi-worker startup gate stays inert — while each Pod's startup orphan reconciliation still writes the other Pods' lease-less runs off as crashed on every rolling update. +- 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. +- The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, and the WeChat QR-login guard, which otherwise only look at the worker count. +- Restart-required: the gate runs once at startup. Restart all Gateway instances together after changing it. + ### Agent Storage Custom agent **definitions** (`config.yaml` + `SOUL.md`) are stored per-user on diff --git a/backend/packages/harness/deerflow/config/app_config.py b/backend/packages/harness/deerflow/config/app_config.py index a4c0121d13..59e31a230f 100644 --- a/backend/packages/harness/deerflow/config/app_config.py +++ b/backend/packages/harness/deerflow/config/app_config.py @@ -20,6 +20,7 @@ from deerflow.config.channel_connections_config import ChannelConnectionsConfig from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict from deerflow.config.database_config import DatabaseConfig from deerflow.config.dedupe_storage_config import DedupeStorageConfig +from deerflow.config.deployment_config import DeploymentConfig from deerflow.config.extensions_config import ExtensionsConfig from deerflow.config.file_signature import ConfigSignature as _ConfigSignature from deerflow.config.file_signature import get_config_signature as _get_config_signature @@ -386,6 +387,13 @@ class AppConfig(BaseModel): field_doc="Stream bridge connecting agent workers to SSE endpoints.", ), ) + deployment: DeploymentConfig = Field( + default_factory=DeploymentConfig, + description=format_field_description( + "deployment", + field_doc="Deployment topology declaration: whether more than one Gateway instance shares this database (drives the multi-process startup safety gate).", + ), + ) run_ownership: RunOwnershipConfig = Field( default_factory=RunOwnershipConfig, description=format_field_description( diff --git a/backend/packages/harness/deerflow/config/deployment_config.py b/backend/packages/harness/deerflow/config/deployment_config.py new file mode 100644 index 0000000000..cf16ce3f69 --- /dev/null +++ b/backend/packages/harness/deerflow/config/deployment_config.py @@ -0,0 +1,107 @@ +"""Deployment topology declaration for the Gateway startup safety gates. + +``GATEWAY_WORKERS`` / ``WEB_CONCURRENCY`` only count the uvicorn workers of one +process tree. A Kubernetes Deployment with ``replicas > 1`` runs one worker per +Pod, so every Pod reports a single worker and the multi-process safety gate in +``app/gateway/deps.py`` stays inert -- while each Pod's startup orphan +reconciliation still treats the other Pods' lease-less runs as crashed. + +This module is the explicit declaration that closes that gap: an operator (or +deploy tooling such as a Helm chart, which can export the env variable from +its replica count) states that more than one Gateway instance shares the +database, and the gate then enforces the same prerequisites it enforces for +``GATEWAY_WORKERS > 1``. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any, Literal + +from pydantic import BaseModel, Field + +#: Environment override for ``deployment.multi_instance``, so deploy tooling can +#: set the declaration from its replica count instead of editing config.yaml. +MULTI_INSTANCE_ENV_VAR = "DEER_FLOW_MULTI_INSTANCE" + +_FALSY = frozenset({"", "0", "false", "no", "off"}) + + +class DeploymentConfig(BaseModel): + """Operator-declared deployment topology.""" + + multi_instance: bool = Field( + default=False, + description=( + "Declare that more than one Gateway instance (uvicorn worker, container, or Kubernetes Pod) shares this database. " + "GATEWAY_WORKERS counts only one process tree, so replicas > 1 with one worker per Pod are invisible to the startup " + "safety gate without this flag. When true, startup requires database.backend='postgres', run_events.backend='db', " + "run_ownership.heartbeat_enabled=true, and a Redis stream bridge; it refuses an explicit memory sandbox ownership store, " + "process-local browser tools, and an enabled scheduler without scheduler.multi_instance. " + "DEER_FLOW_MULTI_INSTANCE=1 declares the same thing from the environment." + ), + ) + + +@dataclass(frozen=True) +class MultiInstanceDeclaration: + """An explicit multi-instance declaration and the knob that made it. + + Callers branch on ``source`` and compose messages from ``knob`` and + ``rollback``; the display text is never the thing to inspect. + """ + + source: Literal["env", "config"] + value: str + + @property + def knob(self) -> str: + """The setting as the operator spelled it, for refusal and warning text.""" + if self.source == "env": + return f"{MULTI_INSTANCE_ENV_VAR}={self.value}" + return "deployment.multi_instance=true" + + @property + def rollback(self) -> str: + """The one step that withdraws this declaration, as a lower-case imperative.""" + if self.source == "env": + return f"unset {MULTI_INSTANCE_ENV_VAR}" + return "set deployment.multi_instance=false" + + def __str__(self) -> str: + return self.knob + + +def multi_instance_declared_by_env() -> str | None: + """Return the ``DEER_FLOW_MULTI_INSTANCE`` value when it declares a multi-instance deployment. + + Blank, ``0``, ``false``, ``no`` and ``off`` mean "not declared", so a + templated ``DEER_FLOW_MULTI_INSTANCE=false`` stays inert. Every other value + is a declaration: a typo in this variable must fail closed (run the gate) + rather than silently disable it. + """ + raw = os.environ.get(MULTI_INSTANCE_ENV_VAR) + if raw is None: + return None + value = raw.strip() + if value.lower() in _FALSY: + return None + return value + + +def multi_instance_declaration(config: Any) -> MultiInstanceDeclaration | None: + """Return the explicit multi-instance declaration in effect, or ``None``. + + The environment is consulted first because deploy tooling that sets it + knows the actual topology; ``deployment.multi_instance`` in config.yaml is + the operator's manual spelling. The result carries the knob that made the + declaration so a refusal message tells the operator what to change. + """ + env_value = multi_instance_declared_by_env() + if env_value is not None: + return MultiInstanceDeclaration(source="env", value=env_value) + deployment = getattr(config, "deployment", None) + if bool(getattr(deployment, "multi_instance", False)): + return MultiInstanceDeclaration(source="config", value="true") + return None diff --git a/backend/packages/harness/deerflow/config/reload_boundary.py b/backend/packages/harness/deerflow/config/reload_boundary.py index 99949bb007..7f71c18c8e 100644 --- a/backend/packages/harness/deerflow/config/reload_boundary.py +++ b/backend/packages/harness/deerflow/config/reload_boundary.py @@ -82,6 +82,10 @@ STARTUP_ONLY_FIELDS: dict[str, str] = { ), "subagent_runtime": ("the shared native-subagent admission controller and isolated execution loop are configured once during Gateway lifespan startup; changing process slots, queue policy, or queue bounds requires a restart."), "subagent_batches": ("the durable subagent batch service is constructed and started once during Gateway lifespan startup; scheduler limits, leases, and recovery behavior are captured by that service instance."), + "deployment": ( + "langgraph_runtime() evaluates deployment.multi_instance once at startup inside the multi-process safety gate (_enforce_postgres_for_multi_worker); " + "the gate never re-runs on config.yaml edits, so declaring or withdrawing a multi-instance topology needs a Gateway restart on every instance." + ), "run_ownership": ( "RunOwnershipConfig is captured once into RunManager at langgraph_runtime() startup; the lease heartbeat background task is created and " "started there, and heartbeat_enabled / lease_seconds / grace_seconds are not re-read on config.yaml edits." diff --git a/backend/tests/test_channel_connections_router.py b/backend/tests/test_channel_connections_router.py index 1721639643..2500b469d1 100644 --- a/backend/tests/test_channel_connections_router.py +++ b/backend/tests/test_channel_connections_router.py @@ -26,6 +26,7 @@ def _stub_app_config(monkeypatch): monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "0") monkeypatch.delenv("GATEWAY_WORKERS", raising=False) monkeypatch.delenv("WEB_CONCURRENCY", raising=False) + monkeypatch.delenv("DEER_FLOW_MULTI_INSTANCE", raising=False) set_app_config(AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}})) yield reset_app_config() @@ -1452,6 +1453,8 @@ def test_wechat_qr_confirmation_preserves_bot_id_unless_replaced(monkeypatch, re {"GATEWAY_WORKERS": "invalid"}, # A blank GATEWAY_WORKERS is unset, so WEB_CONCURRENCY still decides. {"GATEWAY_WORKERS": "", "WEB_CONCURRENCY": "2"}, + # Kubernetes replicas run one worker each and declare their peers instead. + {"GATEWAY_WORKERS": "1", "DEER_FLOW_MULTI_INSTANCE": "1"}, ], ) @pytest.mark.parametrize("method,suffix", [("POST", ""), ("POST", "/session/poll"), ("DELETE", "/session")]) diff --git a/backend/tests/test_inbound_dedupe.py b/backend/tests/test_inbound_dedupe.py index 47a8098f63..6eab7522ee 100644 --- a/backend/tests/test_inbound_dedupe.py +++ b/backend/tests/test_inbound_dedupe.py @@ -28,6 +28,7 @@ def isolated_worker_env(monkeypatch): """ monkeypatch.delenv("GATEWAY_WORKERS", raising=False) monkeypatch.delenv("WEB_CONCURRENCY", raising=False) + monkeypatch.delenv("DEER_FLOW_MULTI_INSTANCE", raising=False) class _FakeDedupe: @@ -107,6 +108,29 @@ def test_factory_warns_when_memory_explicit_under_multi_worker(monkeypatch, capl assert any("dedupe_storage=memory with GATEWAY_WORKERS>1" in r.message for r in caplog.records) +def test_factory_warns_when_memory_explicit_under_a_declared_multi_instance_deployment(monkeypatch, caplog): + """Kubernetes replicas run one worker each and declare their peers instead of a worker count.""" + import logging + + monkeypatch.setenv("DEER_FLOW_MULTI_INSTANCE", "1") + app = _FakeApp("memory", "sqlite") + with caplog.at_level(logging.WARNING): + store = make_inbound_dedupe_store(app) + assert isinstance(store, MemoryInboundDedupeStore) + assert any("dedupe_storage=memory with DEER_FLOW_MULTI_INSTANCE=1" in r.message for r in caplog.records) + + +def test_factory_warns_when_auto_resolves_memory_under_a_declared_multi_instance_deployment(monkeypatch, caplog): + import logging + + monkeypatch.setenv("DEER_FLOW_MULTI_INSTANCE", "true") + app = _FakeApp("auto", "sqlite") + with caplog.at_level(logging.WARNING): + store = make_inbound_dedupe_store(app) + assert isinstance(store, MemoryInboundDedupeStore) + assert any("Multi-worker deployment detected but dedupe_storage=auto" in r.message for r in caplog.records) + + def test_factory_warns_when_auto_resolves_memory_under_multi_worker(monkeypatch, caplog): import logging diff --git a/backend/tests/test_multi_worker_postgres_gate.py b/backend/tests/test_multi_worker_postgres_gate.py index 9d2801c5f3..23d9cd33b2 100644 --- a/backend/tests/test_multi_worker_postgres_gate.py +++ b/backend/tests/test_multi_worker_postgres_gate.py @@ -21,7 +21,9 @@ from fastapi import FastAPI from app.gateway.deps import _enforce_postgres_for_multi_worker, _validate_agent_storage, langgraph_runtime from app.gateway.routers.browser import _browser_tools_enabled from deerflow.config.database_config import DatabaseConfig +from deerflow.config.deployment_config import MULTI_INSTANCE_ENV_VAR, DeploymentConfig, multi_instance_declaration from deerflow.config.run_ownership_config import RunOwnershipConfig +from deerflow.config.stream_bridge_config import StreamBridgeConfig def _config_with_backend( @@ -32,18 +34,33 @@ def _config_with_backend( run_events_backend: str = "db", scheduler_enabled: bool = False, scheduler_multi_instance: bool = False, + deployment_multi_instance: bool = False, + stream_bridge_type: str | None = None, + ownership_type: str | None = None, ) -> SimpleNamespace: run_ownership = RunOwnershipConfig(heartbeat_enabled=heartbeat_enabled) if heartbeat_enabled is not None else None tools = [SimpleNamespace(name="browser_navigate")] if browser_enabled else [] + stream_bridge = StreamBridgeConfig(type=stream_bridge_type) if stream_bridge_type is not None else None + ownership = SimpleNamespace(type=ownership_type) if ownership_type is not None else None return SimpleNamespace( database=DatabaseConfig(backend=backend), run_ownership=run_ownership, run_events=SimpleNamespace(backend=run_events_backend), scheduler=SimpleNamespace(enabled=scheduler_enabled, multi_instance=scheduler_multi_instance), tools=tools, + deployment=DeploymentConfig(multi_instance=deployment_multi_instance), + stream_bridge=stream_bridge, + sandbox=SimpleNamespace(ownership=ownership), ) +def _cluster_ready(**overrides): + """A Postgres deployment with every multi-instance prerequisite satisfied.""" + kwargs = dict(heartbeat_enabled=True, stream_bridge_type="redis") + kwargs.update(overrides) + return _config_with_backend("postgres", **kwargs) + + @pytest.fixture(autouse=True) def isolated_worker_env(monkeypatch): """Keep the suite independent of the invoking shell's worker count. @@ -56,6 +73,8 @@ def isolated_worker_env(monkeypatch): """ monkeypatch.delenv("GATEWAY_WORKERS", raising=False) monkeypatch.delenv("WEB_CONCURRENCY", raising=False) + monkeypatch.delenv(MULTI_INSTANCE_ENV_VAR, raising=False) + monkeypatch.delenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", raising=False) # --------------------------------------------------------------------------- @@ -181,7 +200,7 @@ def test_agent_storage_warning_names_the_variable_that_set_the_count(monkeypatch def test_gate_allows_multi_worker_with_postgres_and_heartbeat(monkeypatch): monkeypatch.setenv("GATEWAY_WORKERS", "2") - _enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True)) + _enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True, stream_bridge_type="redis")) def test_gate_rejects_multi_worker_with_scheduler_enabled(monkeypatch): @@ -227,6 +246,7 @@ def test_gate_allows_multi_instance_scheduler_with_multiple_workers(monkeypatch) heartbeat_enabled=True, scheduler_enabled=True, scheduler_multi_instance=True, + stream_bridge_type="redis", ) ) @@ -458,3 +478,232 @@ async def test_langgraph_runtime_invokes_gate_before_persistence_setup(monkeypat init_engine_from_config.assert_not_called() make_stream_bridge.assert_not_called() make_store.assert_not_called() + + +# --------------------------------------------------------------------------- +# Cross-process stream bridge: multi-worker now needs it too +# --------------------------------------------------------------------------- + + +def test_multi_worker_requires_a_cross_process_stream_bridge(monkeypatch): + """GATEWAY_WORKERS > 1 with the memory bridge used to start and then 409 every SSE join on a peer.""" + monkeypatch.setenv("GATEWAY_WORKERS", "2") + for bridge in (None, "memory"): + with pytest.raises(SystemExit, match="stream_bridge.type='redis'"): + _enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True, stream_bridge_type=bridge)) + + +def test_multi_worker_accepts_the_env_redis_stream_bridge(monkeypatch): + """docker-compose and the Helm chart inject DEER_FLOW_STREAM_BRIDGE_REDIS_URL instead of a config section.""" + monkeypatch.setenv("GATEWAY_WORKERS", "2") + monkeypatch.setenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", "redis://redis:6379/0") + _enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True)) + + +def test_multi_worker_rejects_explicit_memory_sandbox_ownership(monkeypatch): + monkeypatch.setenv("GATEWAY_WORKERS", "2") + with pytest.raises(SystemExit, match="sandbox.ownership.type='memory'"): + _enforce_postgres_for_multi_worker(_cluster_ready(ownership_type="memory")) + + +# --------------------------------------------------------------------------- +# Explicit multi-instance declaration: Kubernetes replicas run one worker per Pod +# --------------------------------------------------------------------------- + + +def test_declared_multi_instance_is_gated_with_a_single_worker(monkeypatch): + """replicas > 1 with GATEWAY_WORKERS=1 per Pod must not bypass the gate. + + Without the declaration every Pod reports one worker, passes every check, + and its startup orphan reconciliation writes the peers' lease-less runs off + as crashed on every rolling update. + """ + monkeypatch.setenv("GATEWAY_WORKERS", "1") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite", deployment_multi_instance=True)) + msg = str(exc_info.value) + assert "deployment.multi_instance=true" in msg + assert "requires database.backend='postgres'" in msg + assert "deployment.multi_instance=false" in msg, "must name the rollback knob" + + +def test_declared_multi_instance_accepts_a_fully_shared_deployment(monkeypatch): + monkeypatch.setenv("GATEWAY_WORKERS", "1") + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True)) + + +def test_declared_multi_instance_requires_heartbeat(): + with pytest.raises(SystemExit, match="heartbeat_enabled=true"): + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, heartbeat_enabled=False)) + + +@pytest.mark.parametrize("run_events_backend", ["memory", "jsonl"]) +def test_declared_multi_instance_requires_db_run_events(run_events_backend): + with pytest.raises(SystemExit, match="run_events.backend='db'"): + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, run_events_backend=run_events_backend)) + + +def test_declared_multi_instance_requires_a_redis_stream_bridge(): + """The memory bridge is process-local, so a peer never sees the owner's SSE events.""" + for bridge in (None, "memory"): + with pytest.raises(SystemExit, match="stream_bridge.type='redis'"): + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, stream_bridge_type=bridge)) + + +def test_declared_multi_instance_accepts_the_env_redis_stream_bridge(monkeypatch): + monkeypatch.setenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", "redis://redis:6379/0") + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, stream_bridge_type=None)) + + +def test_declared_multi_instance_rejects_explicit_memory_sandbox_ownership(): + with pytest.raises(SystemExit, match="sandbox.ownership.type='memory'"): + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, ownership_type="memory")) + + +def test_declared_multi_instance_accepts_redis_or_inferred_sandbox_ownership(): + """An omitted ownership section is inferred from the redis stream bridge (ownership/factory.py).""" + for ownership in (None, "redis"): + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, ownership_type=ownership)) + + +def test_declared_multi_instance_rejects_single_instance_scheduler(): + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, scheduler_enabled=True)) + msg = str(exc_info.value) + assert "each worker starts its own scheduler" in msg + assert "scheduler.multi_instance=true" in msg + assert "deployment.multi_instance=false" in msg + + +def test_declared_multi_instance_accepts_multi_instance_scheduler(): + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, scheduler_enabled=True, scheduler_multi_instance=True)) + + +def test_declared_multi_instance_rejects_process_local_browser(): + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, browser_enabled=True)) + msg = str(exc_info.value) + assert "process-local" in msg + assert "deployment.multi_instance=false" in msg + + +@pytest.mark.parametrize("value", ["1", "true", "True", "yes", "on", "replicas"]) +def test_env_declaration_is_gated(monkeypatch, value): + """DEER_FLOW_MULTI_INSTANCE is what deploy tooling sets from the replica count. + + Any spelling that is not an explicit "off" counts: a typo must fail closed + (run the gate) rather than silently disable it. + """ + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, value) + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + msg = str(exc_info.value) + assert f"{MULTI_INSTANCE_ENV_VAR}={value}" in msg + assert f"Unset {MULTI_INSTANCE_ENV_VAR}" in msg, "must name the rollback knob" + + +@pytest.mark.parametrize("value", ["", "0", "false", "False", "no", "off", " "]) +def test_falsy_env_declaration_stays_inert(monkeypatch, value): + """A templated DEER_FLOW_MULTI_INSTANCE=false (one replica) must not trip the gate.""" + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, value) + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite", scheduler_enabled=True)) + + +def test_env_declaration_accepts_a_fully_shared_deployment(monkeypatch): + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1") + _enforce_postgres_for_multi_worker(_cluster_ready()) + + +def test_worker_count_is_reported_over_the_declaration(monkeypatch): + """When both apply, the refusal names the worker count: that is the knob uvicorn acts on.""" + monkeypatch.setenv("GATEWAY_WORKERS", "2") + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite")) + assert "GATEWAY_WORKERS=2" in str(exc_info.value) + + +@pytest.mark.parametrize( + ("declared_by", "expected_rollback"), + [ + ("env", f"Set GATEWAY_WORKERS=1 and unset {MULTI_INSTANCE_ENV_VAR}"), + ("config", "Set GATEWAY_WORKERS=1 and set deployment.multi_instance=false"), + ], +) +def test_worker_count_rollback_also_withdraws_the_declaration(monkeypatch, declared_by, expected_rollback): + """With both knobs active, a rollback that only resets the worker count is not enough. + + Following ``Set GATEWAY_WORKERS=1`` alone leaves the declaration tripping the + gate at the next start, so the operator would bounce through a second + refusal before learning about the other knob. The message names both. + """ + monkeypatch.setenv("GATEWAY_WORKERS", "2") + if declared_by == "env": + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1") + config = _config_with_backend("sqlite", deployment_multi_instance=declared_by == "config") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(config) + assert expected_rollback in str(exc_info.value) + + +def test_offered_env_rollback_clears_the_gate(monkeypatch): + """The remediation a refusal offers must make the next start succeed.""" + monkeypatch.setenv("GATEWAY_WORKERS", "2") + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1") + sqlite = _config_with_backend("sqlite") + with pytest.raises(SystemExit): + _enforce_postgres_for_multi_worker(sqlite) + + # Resetting only the worker count is the half-step the message must not suggest. + monkeypatch.setenv("GATEWAY_WORKERS", "1") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(sqlite) + assert f"{MULTI_INSTANCE_ENV_VAR}=1" in str(exc_info.value) + + # The full rollback the message offered clears the gate. + monkeypatch.delenv(MULTI_INSTANCE_ENV_VAR) + _enforce_postgres_for_multi_worker(sqlite) + + +def test_offered_config_rollback_clears_the_gate(monkeypatch): + monkeypatch.setenv("GATEWAY_WORKERS", "2") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite", deployment_multi_instance=True)) + assert "Set GATEWAY_WORKERS=1 and set deployment.multi_instance=false" in str(exc_info.value) + + monkeypatch.setenv("GATEWAY_WORKERS", "1") + _enforce_postgres_for_multi_worker(_config_with_backend("sqlite", deployment_multi_instance=False)) + + +def test_browser_refusal_names_the_declaration_when_both_knobs_are_active(monkeypatch): + """The browser refusal must not send the operator through the same double bounce.""" + monkeypatch.setenv("GATEWAY_WORKERS", "2") + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1") + with pytest.raises(SystemExit) as exc_info: + _enforce_postgres_for_multi_worker(_cluster_ready(browser_enabled=True)) + msg = str(exc_info.value) + assert "browser" in msg + assert f"Set GATEWAY_WORKERS=1 and unset {MULTI_INSTANCE_ENV_VAR}" in msg + + +def test_agent_storage_warning_fires_for_a_declared_multi_instance_deployment(caplog): + with caplog.at_level("WARNING"): + _validate_agent_storage(_cluster_ready(deployment_multi_instance=True)) + messages = [r.message for r in caplog.records if "not visible across workers" in r.message] + assert messages and "deployment.multi_instance=true" in messages[0] + + +def test_deployment_declaration_helpers(monkeypatch): + assert DeploymentConfig().multi_instance is False + assert multi_instance_declaration(None) is None + + by_config = multi_instance_declaration(SimpleNamespace(deployment=DeploymentConfig(multi_instance=True))) + assert by_config is not None + assert (by_config.source, by_config.knob, by_config.rollback) == ("config", "deployment.multi_instance=true", "set deployment.multi_instance=false") + assert str(by_config) == by_config.knob + + # The environment wins over config.yaml and keeps the operator's spelling. + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "replicas") + by_env = multi_instance_declaration(SimpleNamespace(deployment=DeploymentConfig(multi_instance=True))) + assert by_env is not None + assert (by_env.source, by_env.knob, by_env.rollback) == ("env", f"{MULTI_INSTANCE_ENV_VAR}=replicas", f"unset {MULTI_INSTANCE_ENV_VAR}") diff --git a/config.example.yaml b/config.example.yaml index 2cd87ad194..dc3fcc0184 100644 --- a/config.example.yaml +++ b/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: 52 +config_version: 53 # Optional operator-owned literal prompt extensions (see CONFIGURATION.md): # lead_prompt_overlay: @@ -2754,6 +2754,27 @@ mcp_tasks: max_result_bytes: 65536 result_preview_max_chars: 2000 +# ============================================================================ +# Deployment Topology +# ============================================================================ +# Declare whether more than one Gateway instance shares this database. +# +# GATEWAY_WORKERS / WEB_CONCURRENCY only count the uvicorn workers of one +# process tree, so a Kubernetes Deployment with replicas > 1 (one worker per +# Pod) is invisible to the startup safety gate. Without this declaration each +# Pod passes every check, and its startup orphan reconciliation writes the +# other Pods' lease-less runs off as crashed on every rolling update. +# +# With multi_instance: true (or the environment variable +# DEER_FLOW_MULTI_INSTANCE=1, which deploy tooling such as a Helm chart can +# set from its replica count) startup requires database.backend: postgres, run_events.backend: db, +# run_ownership.heartbeat_enabled: true and a Redis stream bridge, and refuses +# an explicit sandbox.ownership.type: memory, process-local browser tools, and +# scheduler.enabled: true without scheduler.multi_instance: true. +# Restart required; apply the same value to every instance. +deployment: + multi_instance: false + # ============================================================================ # Run Ownership Configuration # ============================================================================ diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md index 0b290d3159..69e4b4f2d2 100644 --- a/deploy/helm/deer-flow/README.md +++ b/deploy/helm/deer-flow/README.md @@ -135,7 +135,7 @@ they resolve from the `secrets` map): ```yaml config: | - config_version: 52 + config_version: 53 models: - name: gpt-4 use: langchain_openai:ChatOpenAI diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml index 3aa2a3dad2..943664cc72 100644 --- a/deploy/helm/deer-flow/values.yaml +++ b/deploy/helm/deer-flow/values.yaml @@ -263,7 +263,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: 52 + config_version: 53 log_level: info recursion_limit: 100 max_recursion_limit: 1000