From 35f6a98cd964dd1c820c0d46e78067030b5f279d Mon Sep 17 00:00:00 2001 From: Nan Gao Date: Sat, 10 Oct 2026 21:26:01 +0800 Subject: [PATCH] feat(dev): add a two-Gateway multi-instance harness script (#6613) * feat(dev): add a two-Gateway multi-instance harness script There was no way to run the deployment.multi_instance topology locally: scripts/serve.sh pins one Gateway to port 8001, the local nginx config has a single upstream, and getting two processes past the multi-instance startup gate needs Postgres, Redis, a shared DEER_FLOW_HOME, shared auth secrets and a per-process DeerMem retrieval index by hand. scripts/dev_multi_instance.sh (make dev-multi / dev-multi-check / dev-multi-down) starts throwaway Postgres and Redis containers on loopback, renders a config from the developer's config.yaml (or config.example.yaml), and runs Gateway A on 8001 and Gateway B on 8011 with one shared home, one copy of extensions_config.json and generated AUTH_JWT_SECRET, DEER_FLOW_INTERNAL_AUTH_TOKEN and DEER_FLOW_CREDENTIALS_KEY values, plus an optional round-robin nginx on 2027. B starts after A answers /health: Alembic migrations are advisory-locked, but the LangGraph checkpointer/store setup() that follows is not. scripts/dev_multi_instance.py holds the testable parts: the config overlay (neutralizing and reporting settings the gate refuses), the secrets file, the nginx config, and `check`, which automates the scriptable cross-instance acceptance checks: readiness with stream_bridge ok, A-issued sessions and the internal token accepted by B, thread/upload/artifact visibility, a skill toggle seen by B, nginx balancing, and SSE join on B with Last-Event-ID resume on A. Tests pin that the rendered config passes the real _enforce_postgres_for_multi_worker gate. Co-Authored-By: Claude Opus 5.5 * docs(changelog): link the two-Gateway harness entries to #6613 Co-Authored-By: Claude Opus 5.5 * fix(dev): keep the multi-instance harness inside its own state Address review findings on the two-Gateway harness (#6613): - Containers carry a deerflow.harness.state-dir label; up/down remove only containers of their own state dir and refuse those of another checkout or DEERFLOW_MI_STATE_DIR instead of force-removing them. - Containers are removed with `docker rm -f -v`, so the anonymous volumes the postgres and redis images declare no longer leak on every up/down. - `up` installs an EXIT trap once it starts containers, so any later failure (including B's port being taken after preflight) tears down A and both containers; start_gateway returns failures instead of exiting the script, and the Gateway is disowned so stopping it prints no job notice. - The overlay removes any checkpointer section (a postgres one would write checkpoints and the Store into the developer's database), points an explicit sandbox.ownership and a Redis checkpoint_cache at the harness Redis (keeping ownership timing), drops DeerMem storage_path (both spellings) and the blob_storage root so data lands in the harness home, gives a local AIO sandbox a state-dir-specific container_prefix, and reports a non-DeerMem memory backend it keeps sharing. - nginx paths are rendered as quoted string literals ('$' and control characters are refused), so a state dir with spaces works. - `check` skips the run-based SSE check when model discovery did not succeed, unless --with-llm, and enforces the SSE time budget per raw line so heartbeat-only traffic cannot hold it open. Co-Authored-By: Claude Opus 5.5 * docs(scripts): keep the harness guidance under the AGENTS.md soft limit Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- CHANGELOG.md | 19 + CHANGELOG_zh.md | 15 + Makefile | 15 +- README.md | 2 +- backend/docs/CONFIGURATION.md | 36 + .../tests/test_dev_multi_instance_script.py | 705 ++++++++++++++++ scripts/AGENTS.md | 6 + scripts/dev_multi_instance.py | 783 ++++++++++++++++++ scripts/dev_multi_instance.sh | 675 +++++++++++++++ 9 files changed, 2254 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_dev_multi_instance_script.py create mode 100644 scripts/dev_multi_instance.py create mode 100755 scripts/dev_multi_instance.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index 24007f96f3..a2f2ecf6d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -351,6 +351,24 @@ This release closes that milestone with **439 merged pull requests**. redis` or `DEER_FLOW_STREAM_BRIDGE_REDIS_URL` (docker-compose and the Helm chart already inject it). ([#6328]) +- **dev:** `scripts/dev_multi_instance.sh` (`make dev-multi`, `make dev-multi-check`, + `make dev-multi-down`) runs two Gateways as one local multi-instance cluster: + throwaway Postgres and Redis containers on loopback, Gateway A on port 8001 and + Gateway B on 8011 sharing one `DEER_FLOW_HOME` and generated `AUTH_JWT_SECRET`, + `DEER_FLOW_INTERNAL_AUTH_TOKEN` and `DEER_FLOW_CREDENTIALS_KEY` values, each with + its own DeerMem retrieval index, and an optional round-robin nginx on port 2027. + The config is the developer's own `config.yaml` (or `config.example.yaml`) plus + the settings the multi-instance startup gate requires; settings the gate refuses + are changed and reported, and the checkpointer, Redis endpoints, DeerMem and + blob data roots and local AIO sandbox container prefix the config names are + redirected into the harness (a non-DeerMem memory backend is used as + configured). Containers are labelled with the state dir that owns them and + removed with their volumes; a failed `up` tears down what it started. `check` + automates the scriptable cross-instance checks: readiness, shared sessions and + internal token, thread, upload and artifact visibility, skill toggles, nginx + balancing, and SSE `Last-Event-ID` resume across instances. See "Local + two-Gateway harness" in `backend/docs/CONFIGURATION.md`. ([#6613]) + - **config:** `DEER_FLOW_ENV_FILE` selects one explicit UTF-8 dotenv file for the backend at startup, shared by configuration loading, authentication startup, and the debug entry point; relative paths resolve from the backend process working directory, existing process @@ -9534,3 +9552,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6590]: https://github.com/bytedance/deer-flow/pull/6590 [#6595]: https://github.com/bytedance/deer-flow/pull/6595 [#6611]: https://github.com/bytedance/deer-flow/pull/6611 +[#6613]: https://github.com/bytedance/deer-flow/pull/6613 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 3088bf8777..4190b85024 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -274,6 +274,20 @@ `stream_bridge.type: redis` 或 `DEER_FLOW_STREAM_BRIDGE_REDIS_URL` （docker-compose 与 Helm chart 已自动注入）。([#6328]) +- **开发：** 新增 `scripts/dev_multi_instance.sh`（`make dev-multi`、`make dev-multi-check`、 + `make dev-multi-down`），在本机把两个 Gateway 作为一个多实例集群运行：回环地址上的 + 临时 Postgres 与 Redis 容器，Gateway A（8001 端口）与 Gateway B（8011 端口）共享同一 + `DEER_FLOW_HOME` 以及生成的 `AUTH_JWT_SECRET`、`DEER_FLOW_INTERNAL_AUTH_TOKEN` 与 + `DEER_FLOW_CREDENTIALS_KEY`，各自使用独立的 DeerMem 检索索引，并可选在 2027 端口启动 + 轮询 nginx。配置取开发者自己的 `config.yaml`（或 `config.example.yaml`），叠加多实例 + 启动门控要求的设置；门控拒绝的设置会被修改并逐项提示，配置中引用的 checkpointer、Redis 端点、 + DeerMem 与 blob 数据目录以及本地 AIO 沙箱容器名前缀都会被重定向到 harness 内部（非 DeerMem 的 + 记忆后端按原配置使用）。容器带有所属状态目录的标签，删除时连同其匿名卷一并 + 移除；`up` 中途失败会拆除已启动的部分。`check` 自动执行可脚本化的跨实例 + 检查：就绪探针、共享会话与内部令牌、线程/上传/产物可见性、技能开关、nginx 负载均衡， + 以及跨实例的 SSE `Last-Event-ID` 续接。详见 `backend/docs/CONFIGURATION.md` 的 + “Local two-Gateway harness”一节。 ([#6613]) + - **配置：** 新增 `DEER_FLOW_ENV_FILE`，在后端启动时选择一个显式的 UTF-8 dotenv 文件，配置 加载、认证启动和调试入口共用；相对路径以后端进程工作目录解析，已有进程环境变量保持 优先，选择为空、缺失、不可读或编码无效时直接报错而不回退。不设置时保持默认 dotenv 发现； @@ -7665,3 +7679,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6590]: https://github.com/bytedance/deer-flow/pull/6590 [#6595]: https://github.com/bytedance/deer-flow/pull/6595 [#6611]: https://github.com/bytedance/deer-flow/pull/6611 +[#6613]: https://github.com/bytedance/deer-flow/pull/6613 diff --git a/Makefile b/Makefile index 9ccaea34eb..5875f8c41d 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ # DeerFlow - Unified Development Environment -.PHONY: help config config-upgrade check check-agent-guidance install extension-install extension-upgrade extension-list extension-enable extension-disable extension-remove setup doctor support-bundle detect-thread-boundaries detect-blocking-io dev dev-daemon start start-daemon nginx stop up down prod-logs clean docker-init docker-start docker-stop docker-logs docker-logs-frontend docker-logs-gateway docker-logs-redis setup-sandbox +.PHONY: help config config-upgrade check check-agent-guidance install extension-install extension-upgrade extension-list extension-enable extension-disable extension-remove setup doctor support-bundle detect-thread-boundaries detect-blocking-io dev dev-daemon dev-multi dev-multi-check dev-multi-down start start-daemon nginx stop up down prod-logs clean docker-init docker-start docker-stop docker-logs docker-logs-frontend docker-logs-gateway docker-logs-redis setup-sandbox BASH ?= bash BACKEND_UV_RUN = cd backend && uv run @@ -48,6 +48,9 @@ help: @echo " make start-daemon - Start prod services in background (daemon mode)" @echo " make nginx - Start nginx alone in the foreground (local dev config)" @echo " make stop - Stop all running services" + @echo " make dev-multi - Start two Gateways as a local multi-instance cluster (Postgres + Redis in Docker)" + @echo " make dev-multi-check - Run the cross-instance checks against that pair" + @echo " make dev-multi-down - Stop the pair and delete its containers and data" @echo " make clean - Stop local services and DELETE local runtime data (backend/.deer-flow: $(RUNTIME_DATA_CONTENTS)) and logs" @echo "" @echo "Docker Production Commands:" @@ -174,6 +177,16 @@ nginx: stop: @$(RUN_SHELL_SCRIPT) ./scripts/serve.sh --stop +# Two Gateways on shared Postgres, Redis and DEER_FLOW_HOME (scripts/dev_multi_instance.sh --help) +dev-multi: + @$(RUN_SHELL_SCRIPT) ./scripts/dev_multi_instance.sh up + +dev-multi-check: + @$(RUN_SHELL_SCRIPT) ./scripts/dev_multi_instance.sh check + +dev-multi-down: + @$(RUN_SHELL_SCRIPT) ./scripts/dev_multi_instance.sh down + # Clean up: deletes local runtime data, not just temporary files. The guard # runs before stop, which would otherwise stop a running stack's sandboxes. clean: diff --git a/README.md b/README.md index b1f56cadea..e09209731a 100644 --- a/README.md +++ b/README.md @@ -533,7 +533,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. 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. Declared instances that store credentials (`channel_connections`) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). 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. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); the remaining IM channel state (each instance's platform connections, follow-up buffers) still needs its 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. Declared instances that store credentials (`channel_connections`) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). 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. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); the remaining IM channel state (each instance's platform connections, follow-up buffers) still needs its own multi-worker coordination. To try this topology on one machine, `make dev-multi` starts two Gateways on shared Postgres, Redis and `DEER_FLOW_HOME`, and `make dev-multi-check` runs the cross-instance checks; see the local two-Gateway harness in [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#local-two-gateway-harness). > > 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/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 9047e4a811..1e32145c60 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -595,6 +595,42 @@ Notes: - A declared multi-instance deployment that enables a feature storing credentials (today `channel_connections.enabled: true`) also needs one shared `DEER_FLOW_CREDENTIALS_KEY`; see [Credentials encryption key](#credentials-encryption-key-deer_flow_credentials_key). - Restart-required: the gate runs once at startup. Restart all Gateway instances together after changing it. +#### Local two-Gateway harness + +`scripts/dev_multi_instance.sh` (also `make dev-multi`, `make dev-multi-check`, `make dev-multi-down`) runs this topology on one machine, so cross-instance behavior can be checked without a cluster: + +- Throwaway `postgres:17-alpine` and `redis:7-alpine` containers (`deerflow-mi-postgres`, `deerflow-mi-redis`) on `127.0.0.1:55432` and `127.0.0.1:56379`, Gateway A on `127.0.0.1:8001` and Gateway B on `127.0.0.1:8011` (8002 stays free for the provisioner). When nginx is installed, `http://localhost:2027` round-robins `/api` over both Gateways, names the Gateway that answered in an `X-DeerFlow-Upstream` response header, and forwards `/` to a frontend on port 3000 if you start one. Ports, container names and images are overridable; see `--help`. +- The generated `config.yaml` is your own `config.yaml` (otherwise `config.example.yaml`; `DEERFLOW_MI_BASE_CONFIG` picks another file) plus `deployment.multi_instance: true`, `database.backend: postgres`, `run_events.backend: db`, `run_ownership.heartbeat_enabled: true` and `stream_bridge.type: redis`. Settings the startup gate refuses are changed and reported: process-local browser tools are removed and an enabled scheduler gets `scheduler.multi_instance: true`. Enabled IM channels are disabled because both Gateways would connect the same bot; `DEERFLOW_MI_KEEP_CHANNELS=1` keeps them. +- Everything the pair writes stays in the harness, never in your own deployment: any `checkpointer` section is removed (it would win over `database` for checkpoints and the Store), an explicit `sandbox.ownership` and a Redis `database.checkpoint_cache` point at the harness Redis (ownership timing settings are kept), a DeerMem `storage_path` (either spelling) and `blob_storage.backend_config.root` are dropped so their data lands in the harness home, and a local AIO sandbox gets a `container_prefix` derived from the state dir, so the harness never adopts or reaps sandbox containers it did not start. A non-DeerMem memory backend is used as configured, and the harness says so. +- Both processes share one `DEER_FLOW_HOME`, one copy of `extensions_config.json` (MCP and skill edits never reach your own file), and generated `AUTH_JWT_SECRET`, `DEER_FLOW_INTERNAL_AUTH_TOKEN` and `DEER_FLOW_CREDENTIALS_KEY` values. Each process keeps its own DeerMem `retrieval_index_path`, which the shared config reads from `$DEERFLOW_MI_RETRIEVAL_INDEX_PATH`. +- Gateway B starts only after A answers `/health`. Alembic migrations are serialized by an advisory lock, but the LangGraph checkpointer and store `setup()` that runs next is not, so two cold starts against an empty database could race. +- State lives in `.deer-flow/multi-instance/` (gitignored). `down` stops the Gateways and nginx, removes the containers together with their anonymous data volumes and deletes everything there except `logs/`. Each container is labelled with the state dir that created it, and only that state dir removes it: a harness started from another checkout or `DEERFLOW_MI_STATE_DIR` refuses to start rather than remove them (give it its own `DEERFLOW_MI_*_CONTAINER` names and ports). If `up` fails partway, everything it already started is torn down. Do not run `make dev` or `make stop` alongside the harness: both reclaim port 8001 and this checkout's Gateway processes. + +```bash +scripts/dev_multi_instance.sh up # containers, config, Gateway A, then B, then nginx +scripts/dev_multi_instance.sh check # automated cross-instance checks (below) +scripts/dev_multi_instance.sh stop b --kill # SIGKILL B to simulate a crash; `start b` brings it back +scripts/dev_multi_instance.sh logs a -f +scripts/dev_multi_instance.sh down +``` + +`check` signs in as a throwaway account it creates on first use (on a fresh database it becomes the first admin, which the skill toggle needs; the credentials are in `.deer-flow/multi-instance/check-user.json`) and verifies that: + +- `GET /health/ready` reports `ready` and `stream_bridge: ok` on both Gateways; +- B accepts a session issued by A, and both accept the shared internal token while refusing a wrong one; +- a thread created on A is readable on B, and a file uploaded on A is listed on B, which serves identical bytes from `GET /api/threads/{id}/artifacts/{path}`; +- a public skill toggled with `PUT /api/skills/{name}` on A shows the new state on B within 30 s (the toggle is reverted afterwards); +- nginx reaches both Gateways; +- a run created on A can be joined on B and resumed on A from a `Last-Event-ID`. With no model configured the run fails before any LLM call; when models are configured, or the model list could not be read, this check is skipped unless you pass `check --with-llm`. Each stream read has a time budget that heartbeats do not extend. + +These still need a manual pass: + +- stop B with `stop b --kill` while A has an active run: A's run must not turn `error`; +- cancel a run on B from A: it takes effect within about one heartbeat (~10 s; a lower `run_ownership.lease_seconds` makes the boundary easier to observe); +- change an MCP filesystem server's args on A: B's local-bash allowlist follows; +- edit a skill on A: B's system-prompt skills section changes within 30 s; +- the IM channel gates (one leader per platform, failover after killing the leader, runtime-config changes), once channel leader election lands; run them with `DEERFLOW_MI_KEEP_CHANNELS=1`. + ### Credentials encryption key (`DEER_FLOW_CREDENTIALS_KEY`) DeerFlow encrypts credentials it stores at rest (today: per-connection IM channel credentials in the `channel_credentials` table) with a deployment key read **only** from the environment — there is no `config.yaml` key, and it is never derived from `AUTH_JWT_SECRET`. diff --git a/backend/tests/test_dev_multi_instance_script.py b/backend/tests/test_dev_multi_instance_script.py new file mode 100644 index 0000000000..abb19c97bf --- /dev/null +++ b/backend/tests/test_dev_multi_instance_script.py @@ -0,0 +1,705 @@ +"""Tests for the two-Gateway local harness (``scripts/dev_multi_instance.{sh,py}``). + +The shell entry point owns containers and processes; the Python helper owns the +parts that decide whether the pair can start at all. The key contract pinned +here: the generated ``config.yaml`` passes the *real* multi-instance startup +gate (``app.gateway.deps._enforce_postgres_for_multi_worker``) for any base +config, while leaving the developer's other sections untouched. +""" + +from __future__ import annotations + +import base64 +import importlib.util +import logging +import os +import shutil +import stat +import subprocess +import sys +from pathlib import Path + +import pytest +import yaml + +from app.gateway.deps import _enforce_postgres_for_multi_worker, _validate_memory_retrieval_index +from deerflow.config.app_config import AppConfig +from deerflow.config.deployment_config import MULTI_INSTANCE_ENV_VAR, multi_instance_declaration + +REPO_ROOT = Path(__file__).resolve().parents[2] +HELPER_PATH = REPO_ROOT / "scripts" / "dev_multi_instance.py" +SHELL_PATH = REPO_ROOT / "scripts" / "dev_multi_instance.sh" + +_spec = importlib.util.spec_from_file_location("deerflow_dev_multi_instance", HELPER_PATH) +assert _spec is not None and _spec.loader is not None +mi = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = mi # dataclasses resolve string annotations through sys.modules +_spec.loader.exec_module(mi) + +DATABASE_URL = "postgresql://deerflow:secret@127.0.0.1:55432/deerflow" +REDIS_URL = "redis://127.0.0.1:56379/0" + + +@pytest.fixture(autouse=True) +def _isolated_gate_env(monkeypatch): + """The gate also reads worker-count and declaration variables from the shell.""" + for name in ("GATEWAY_WORKERS", "WEB_CONCURRENCY", MULTI_INSTANCE_ENV_VAR, "DEER_FLOW_STREAM_BRIDGE_REDIS_URL", mi.RETRIEVAL_INDEX_ENV_VAR): + monkeypatch.delenv(name, raising=False) + + +SANDBOX_PREFIX = "deer-flow-mi-0123456789-sandbox" + + +def _overlay(base: dict, **kwargs): + return mi.build_multi_instance_config(base, database_url=DATABASE_URL, redis_url=REDIS_URL, sandbox_container_prefix=SANDBOX_PREFIX, **kwargs) + + +# --------------------------------------------------------------------------- +# Overlay semantics +# --------------------------------------------------------------------------- + + +def test_overlay_sets_every_multi_instance_prerequisite(): + config, _notes = _overlay({}) + + assert config["deployment"]["multi_instance"] is True + assert config["database"]["backend"] == "postgres" + assert config["database"]["postgres_url"] == DATABASE_URL + assert config["run_events"]["backend"] == "db" + assert config["run_ownership"]["heartbeat_enabled"] is True + assert config["stream_bridge"] == {"type": "redis", "redis_url": REDIS_URL} + # One shared file, a per-process index directory chosen by the environment. + assert config["memory"]["backend_config"]["retrieval_index_path"] == f"${mi.RETRIEVAL_INDEX_ENV_VAR}" + + +def test_overlay_preserves_unrelated_base_sections_without_mutating_the_base(): + base = { + "config_version": 57, + "models": [{"name": "m1", "use": "langchain_openai:ChatOpenAI", "model": "gpt-x"}], + "tools": [{"name": "web_search", "use": "deerflow.community.ddg_search.tools:web_search_tool"}], + "database": {"backend": "sqlite", "sqlite_dir": ".deer-flow/data", "pool_recycle": 120}, + "run_ownership": {"lease_seconds": 15, "grace_seconds": 5, "heartbeat_enabled": False}, + "run_events": {"backend": "memory", "max_trace_content": 2048}, + "memory": {"enabled": True, "backend_config": {"debounce_seconds": 5, "storage_path": ""}}, + "scheduler": {"enabled": False, "multi_instance": False}, + } + snapshot = yaml.safe_dump(base) + + config, _notes = _overlay(base) + + assert yaml.safe_dump(base) == snapshot, "the base mapping must not be mutated" + assert config["config_version"] == 57 + assert config["models"] == base["models"] + assert config["tools"] == base["tools"] + assert config["database"]["pool_recycle"] == 120 + assert config["run_ownership"]["lease_seconds"] == 15 + assert config["run_ownership"]["grace_seconds"] == 5 + assert config["run_events"]["max_trace_content"] == 2048 + assert config["memory"]["backend_config"]["debounce_seconds"] == 5 + # A disabled scheduler is left alone. + assert config["scheduler"] == {"enabled": False, "multi_instance": False} + + +def test_overlay_neutralizes_settings_the_gate_refuses(): + base = { + "tools": [ + {"name": "web_search", "use": "deerflow.community.ddg_search.tools:web_search_tool"}, + {"name": "browser_navigate", "use": "deerflow.community.browser_automation.tools:browser_navigate_tool"}, + {"name": "browser_click", "use": "deerflow.community.browser_automation.tools:browser_click_tool"}, + ], + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider", "ownership": {"type": "memory"}}, + "scheduler": {"enabled": True}, + "checkpointer": {"type": "sqlite", "connection_string": "checkpoints.db"}, + "channels": { + "langgraph_url": "http://localhost:8001/api", + "session": {"assistant_id": "lead_agent"}, + "telegram": {"enabled": True, "bot_token": "$TELEGRAM_BOT_TOKEN"}, + "slack": {"enabled": False}, + }, + } + + config, notes = _overlay(base) + + assert [tool["name"] for tool in config["tools"]] == ["web_search"] + assert config["sandbox"]["ownership"] == {"type": "redis", "redis_url": REDIS_URL} + assert config["sandbox"]["use"] == "deerflow.sandbox.local:LocalSandboxProvider" + assert "container_prefix" not in config["sandbox"], "only the local AIO provider names containers" + assert config["scheduler"] == {"enabled": True, "multi_instance": True} + assert "checkpointer" not in config + assert config["channels"]["telegram"]["enabled"] is False + assert config["channels"]["telegram"]["bot_token"] == "$TELEGRAM_BOT_TOKEN" + assert config["channels"]["slack"]["enabled"] is False + assert config["channels"]["langgraph_url"] == "http://localhost:8001/api" + joined = "\n".join(notes) + for fragment in ("browser_navigate", "sandbox.ownership", "scheduler.multi_instance", "checkpointer", "telegram"): + assert fragment in joined + + +def test_overlay_keeps_channels_when_asked(): + config, _notes = _overlay({"channels": {"telegram": {"enabled": True}}}, keep_channels=True) + + assert config["channels"]["telegram"]["enabled"] is True + + +def test_overlay_drops_a_postgres_checkpointer_pointing_at_the_developers_database(): + # An explicit checkpointer section wins over database.postgres_url for LangGraph + # checkpoints and the Store, so keeping it would write into the real database. + base = {"checkpointer": {"type": "postgres", "connection_string": "postgresql://me@db.internal:5432/prod"}} + + config, notes = _overlay(base) + + assert "checkpointer" not in config + assert any("checkpointer" in note for note in notes) + + +def test_overlay_points_explicit_sandbox_ownership_at_the_harness_redis(): + base = { + "sandbox": { + "use": "deerflow.sandbox.local:LocalSandboxProvider", + "ownership": {"type": "redis", "redis_url": "redis://prod-redis:6379/3", "renewal_interval_seconds": 7.5, "ttl_multiplier": 3}, + } + } + + config, notes = _overlay(base) + + assert config["sandbox"]["ownership"] == {"type": "redis", "redis_url": REDIS_URL, "renewal_interval_seconds": 7.5, "ttl_multiplier": 3} + assert any("sandbox.ownership" in note for note in notes) + + +def test_overlay_gives_local_aio_sandboxes_a_harness_container_prefix(): + base = {"sandbox": {"use": "deerflow.community.aio_sandbox:AioSandboxProvider", "container_prefix": "deer-flow-sandbox"}} + + config, notes = _overlay(base) + + assert config["sandbox"]["container_prefix"] == SANDBOX_PREFIX + assert any("container_prefix" in note for note in notes) + # Omitted prefix (the provider default) is replaced too. + config, _notes = _overlay({"sandbox": {"use": "deerflow.community.aio_sandbox:AioSandboxProvider"}}) + assert config["sandbox"]["container_prefix"] == SANDBOX_PREFIX + + +def test_overlay_leaves_provisioner_backed_aio_prefixes_alone(): + base = {"sandbox": {"use": "deerflow.community.aio_sandbox:AioSandboxProvider", "provisioner_url": "http://provisioner:8002"}} + + config, _notes = _overlay(base) + + assert "container_prefix" not in config["sandbox"] + + +def test_sandbox_container_prefix_is_stable_per_state_dir(tmp_path): + first = mi.sandbox_container_prefix(tmp_path / "a") + + assert first == mi.sandbox_container_prefix(tmp_path / "a") + assert first != mi.sandbox_container_prefix(tmp_path / "b") + assert first.startswith("deer-flow-mi-") and first.endswith("-sandbox") + assert all(ch.isalnum() or ch in "-_." for ch in first) + + +@pytest.mark.parametrize( + "memory", + [ + {"backend_config": {"storage_path": "/srv/real-memory"}}, + {"storage_path": "/srv/real-memory"}, # legacy pre-abstraction spelling, auto-migrated on load + ], +) +def test_overlay_keeps_deermem_data_inside_the_harness_home(memory): + config, notes = _overlay({"memory": memory}) + + assert "storage_path" not in config["memory"] + assert "storage_path" not in config["memory"]["backend_config"] + assert any("storage_path" in note for note in notes) + + +def test_overlay_keeps_other_data_roots_inside_the_harness(): + base = { + "blob_storage": {"enabled": True, "backend": "local_fs", "backend_config": {"root": "/mnt/shared/deerflow-blobs"}}, + "database": {"checkpoint_cache": {"type": "redis", "redis_url": "redis://prod-redis:6379/1", "max_entries": 64}}, + } + + config, notes = _overlay(base) + + assert "root" not in config["blob_storage"]["backend_config"] + assert config["database"]["checkpoint_cache"] == {"type": "redis", "redis_url": REDIS_URL, "max_entries": 64} + joined = "\n".join(notes) + assert "blob_storage" in joined and "checkpoint_cache" in joined + + +def test_overlay_leaves_non_deermem_memory_backends_alone_but_says_so(): + config, notes = _overlay({"memory": {"manager_class": "mem0", "backend_config": {"base_url": "https://mem0.example"}}}) + + assert config["memory"]["backend_config"] == {"base_url": "https://mem0.example"} + assert any("mem0" in note for note in notes) + + +# --------------------------------------------------------------------------- +# The generated file against the real loader and startup gate +# --------------------------------------------------------------------------- + + +def _load_generated(tmp_path: Path, monkeypatch, base_path: Path, *, index_dir: Path) -> AppConfig: + out = tmp_path / "state" / "config.yaml" + extensions = tmp_path / "extensions_config.json" + extensions.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8") + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions)) + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path / "home")) + monkeypatch.setenv(mi.RETRIEVAL_INDEX_ENV_VAR, str(index_dir)) + mi.render_config(base_path, out, database_url=DATABASE_URL, redis_url=REDIS_URL, state_dir=tmp_path / "state") + return AppConfig.from_file(str(out)) + + +def test_generated_config_from_the_example_passes_the_real_startup_gate(tmp_path, monkeypatch, caplog): + config = _load_generated(tmp_path, monkeypatch, REPO_ROOT / "config.example.yaml", index_dir=tmp_path / "index-a") + + assert multi_instance_declaration(config) is not None + _enforce_postgres_for_multi_worker(config) # must not raise SystemExit + + with caplog.at_level(logging.WARNING, logger="app.gateway.deps"): + _validate_memory_retrieval_index(config) + assert "retrieval index" not in caplog.text, "each process must keep its index outside the shared storage_path" + assert config.memory.backend_config["retrieval_index_path"] == str(tmp_path / "index-a") + + +def test_generated_config_from_a_hostile_base_passes_the_real_startup_gate(tmp_path, monkeypatch): + base_path = tmp_path / "base.yaml" + base_path.write_text( + yaml.safe_dump( + { + "config_version": 1, + "models": [], + "tools": [{"name": "browser_navigate", "use": "deerflow.community.browser_automation.tools:browser_navigate_tool", "group": "web"}], + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider", "ownership": {"type": "memory"}}, + "scheduler": {"enabled": True}, + "database": {"backend": "sqlite", "sqlite_dir": ".deer-flow/data"}, + "run_events": {"backend": "jsonl"}, + "stream_bridge": {"type": "memory"}, + } + ), + encoding="utf-8", + ) + monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1") + with pytest.raises(SystemExit): + # Sanity: the base itself is refused under a multi-instance declaration. + _enforce_postgres_for_multi_worker(AppConfig.from_file(str(base_path))) + monkeypatch.delenv(MULTI_INSTANCE_ENV_VAR) + + config = _load_generated(tmp_path, monkeypatch, base_path, index_dir=tmp_path / "index-b") + + _enforce_postgres_for_multi_worker(config) + + +def test_each_process_resolves_its_own_retrieval_index(tmp_path, monkeypatch): + out = tmp_path / "config.yaml" + mi.render_config(REPO_ROOT / "config.example.yaml", out, database_url=DATABASE_URL, redis_url=REDIS_URL, state_dir=tmp_path) + raw = yaml.safe_load(out.read_text(encoding="utf-8")) + resolved = [] + for name in ("a", "b"): + monkeypatch.setenv(mi.RETRIEVAL_INDEX_ENV_VAR, str(tmp_path / "index" / name)) + resolved.append(AppConfig.resolve_env_variables(raw)["memory"]["backend_config"]["retrieval_index_path"]) + + assert resolved[0] != resolved[1] + + +def test_render_config_writes_a_private_file_with_a_generated_header(tmp_path): + out = tmp_path / "nested" / "config.yaml" + + notes = mi.render_config(REPO_ROOT / "config.example.yaml", out, database_url=DATABASE_URL, redis_url=REDIS_URL, state_dir=tmp_path) + + assert isinstance(notes, list) + text = out.read_text(encoding="utf-8") + assert text.startswith("# Generated by scripts/dev_multi_instance.sh") + assert stat.S_IMODE(out.stat().st_mode) == 0o600, "the base config may carry model API keys" + assert yaml.safe_load(text)["deployment"]["multi_instance"] is True + + +# --------------------------------------------------------------------------- +# Shared secrets and nginx front +# --------------------------------------------------------------------------- + + +def test_secrets_file_is_private_complete_and_stable(tmp_path): + path = tmp_path / "secrets.env" + + first = mi.ensure_secrets_file(path) + second = mi.ensure_secrets_file(path) + + assert first == second, "an existing secrets file is reused, never regenerated" + assert set(first) == set(mi.SHARED_SECRET_NAMES) + assert all(value for value in first.values()) + assert len(base64.urlsafe_b64decode(first["DEER_FLOW_CREDENTIALS_KEY"])) == 32, "must be a raw Fernet key" + assert first["AUTH_JWT_SECRET"] != first["DEER_FLOW_INTERNAL_AUTH_TOKEN"] + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + # Shell-sourceable KEY=value lines. + assert sorted(line.split("=", 1)[0] for line in path.read_text(encoding="utf-8").splitlines() if line and not line.startswith("#")) == sorted(mi.SHARED_SECRET_NAMES) + + +def test_nginx_conf_balances_both_gateways(tmp_path): + text = mi.render_nginx_conf(listen_port=2027, gateway_ports=[8001, 8011], frontend_port=3000, state_dir=tmp_path) + + assert "listen 127.0.0.1:2027;" in text + assert "server 127.0.0.1:8001;" in text + assert "server 127.0.0.1:8011;" in text + assert "X-DeerFlow-Upstream $upstream_addr" in text + assert f'"{tmp_path / "run" / "nginx.pid"}"' in text + + +def test_nginx_conf_quotes_paths_with_whitespace(tmp_path): + state = tmp_path / "mi state" + + text = mi.render_nginx_conf(listen_port=2027, gateway_ports=[8001, 8011], frontend_port=3000, state_dir=state) + + assert f'pid "{state / "run" / "nginx.pid"}";' in text + assert f'access_log "{state / "logs" / "nginx-access.log"}";' in text + assert f'proxy_temp_path "{state / "nginx" / "temp" / "proxy"}";' in text + + +def test_nginx_conf_rejects_paths_nginx_would_interpolate(tmp_path): + with pytest.raises(ValueError): + mi.render_nginx_conf(listen_port=2027, gateway_ports=[8001], frontend_port=3000, state_dir=tmp_path / "$HOME") + + +@pytest.mark.skipif(shutil.which("nginx") is None, reason="nginx is not installed") +@pytest.mark.parametrize("state_name", ["state", "mi state"]) +def test_nginx_conf_is_accepted_by_nginx(tmp_path, state_name): + state = tmp_path / state_name + for sub in ("run", "logs", "nginx/temp"): # the shell creates these before nginx starts + (state / sub).mkdir(parents=True) + conf = state / "nginx" / "nginx.conf" + conf.write_text(mi.render_nginx_conf(listen_port=2027, gateway_ports=[8001, 8011], frontend_port=3000, state_dir=state), encoding="utf-8") + + result = subprocess.run(["nginx", "-t", "-p", str(state / "nginx"), "-c", str(conf)], capture_output=True, text=True, timeout=30) + + assert result.returncode == 0, result.stderr + + +# --------------------------------------------------------------------------- +# Check helpers +# --------------------------------------------------------------------------- + + +def test_parse_sse_events_keeps_ids_and_event_names(): + lines = [ + ": comment", + "id: 1-0", + "event: metadata", + 'data: {"run_id": "r"}', + "", + "event: heartbeat", + "data: {}", + "", + "id: 2-0", + "event: end", + "data: null", + "", + ] + + events = list(mi.parse_sse_events(lines)) + + assert [(event.id, event.event) for event in events] == [("1-0", "metadata"), (None, "heartbeat"), ("2-0", "end")] + assert events[0].data == '{"run_id": "r"}' + + +def test_check_exit_code_reflects_failures_only(): + assert mi.exit_code([mi.CheckResult("a", mi.PASS, ""), mi.CheckResult("b", mi.SKIP, "")]) == 0 + assert mi.exit_code([mi.CheckResult("a", mi.PASS, ""), mi.CheckResult("b", mi.FAIL, "")]) == 1 + + +def _offline_harness(tmp_path, handler, **kwargs): + import httpx + + h = mi.Harness(gateways=[("A", "http://a.test"), ("B", "http://b.test")], nginx_url=None, state_dir=tmp_path, with_llm=False, timeout=5.0, **kwargs) + h.http = httpx.Client(transport=httpx.MockTransport(handler)) + return h + + +def test_sse_resume_never_starts_a_run_when_model_discovery_failed(tmp_path): + def refuse(request): + raise AssertionError(f"unexpected request {request.method} {request.url}") + + h = _offline_harness(tmp_path, refuse) + h.thread_id = "t-1" + h.models = None # the internal-token check failed or was skipped + + with pytest.raises(mi.CheckSkipped, match="--with-llm"): + mi._check_sse_resume(h) + + +def test_sse_read_enforces_its_budget_while_only_heartbeats_arrive(tmp_path): + import time as _time + + def heartbeats(): + # A live idle run: comment heartbeats only, for far longer than the budget. + for _ in range(300): + _time.sleep(0.01) + yield b": heartbeat\n\n" + + import httpx + + h = _offline_harness(tmp_path, lambda request: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=heartbeats())) + started = _time.monotonic() + + with pytest.raises(mi.CheckFailed, match="budget"): + mi._read_stream(h, "http://a.test", "/api/threads/t/runs/r/join", budget=0.3) + + assert _time.monotonic() - started < 2.0 + + +# --------------------------------------------------------------------------- +# Shell entry point smoke tests (no Docker, no processes) +# --------------------------------------------------------------------------- + + +def test_shell_script_parses(): + result = subprocess.run(["bash", "-n", str(SHELL_PATH)], capture_output=True, text=True, timeout=30) + + assert result.returncode == 0, result.stderr + + +@pytest.mark.parametrize("flag", ["--help", "help", "-h"]) +def test_shell_script_help_lists_every_command_without_side_effects(tmp_path, flag): + env = {**os.environ, "DEERFLOW_MI_STATE_DIR": str(tmp_path / "state"), "PATH": os.environ.get("PATH", "")} + + result = subprocess.run(["bash", str(SHELL_PATH), flag], capture_output=True, text=True, timeout=30, env=env) + + assert result.returncode == 0, result.stderr + for command in ("up", "down", "status", "logs", "check", "restart"): + assert f" {command}" in result.stdout + assert not (tmp_path / "state").exists(), "help must not create harness state" + + +def test_shell_script_rejects_unknown_commands(tmp_path): + env = {**os.environ, "DEERFLOW_MI_STATE_DIR": str(tmp_path / "state")} + + result = subprocess.run(["bash", str(SHELL_PATH), "frobnicate"], capture_output=True, text=True, timeout=30, env=env) + + assert result.returncode != 0 + assert "Unknown command" in result.stderr + + +def test_makefile_exposes_harness_targets(): + makefile = (REPO_ROOT / "Makefile").read_text(encoding="utf-8") + + for target in ("dev-multi", "dev-multi-check", "dev-multi-down"): + assert f"\n{target}:" in makefile + assert f"make {target} " in makefile, f"`make help` should list {target}" + + +# --------------------------------------------------------------------------- +# Shell lifecycle against stubbed docker / uv / lsof and a fake Gateway +# --------------------------------------------------------------------------- + +_DOCKER_STUB = r"""#!/usr/bin/env bash +dir="$(cd "$(dirname "$0")" && pwd)" +state="$dir/containers" +mkdir -p "$state" +{ printf 'docker'; printf ' %s' "$@"; printf '\n'; } >>"$dir/docker.log" +cmd="$1" +shift +case "$cmd" in + info) exit 0 ;; + inspect) + fmt="" + if [ "$1" = "-f" ]; then fmt="$2"; shift 2; fi + [ -d "$state/$1" ] || exit 1 + case "$fmt" in + *state-dir*) cat "$state/$1/deerflow.harness.state-dir" 2>/dev/null; echo ;; + *deerflow.harness*) cat "$state/$1/deerflow.harness" 2>/dev/null; echo ;; + *) echo running ;; + esac + ;; + run) + name="" + labels="" + while [ $# -gt 0 ]; do + case "$1" in + --name) name="$2"; shift 2 ;; + --label) labels="$labels$2 +"; shift 2 ;; + -e | -p) shift 2 ;; + *) shift ;; + esac + done + mkdir -p "$state/$name" + printf '%s' "$labels" | while IFS= read -r kv; do + [ -n "$kv" ] && printf '%s' "${kv#*=}" >"$state/$name/${kv%%=*}" + done + echo "id-$name" + ;; + exec) + case "$*" in *redis-cli*) echo PONG ;; esac + ;; + rm) + for arg in "$@"; do + case "$arg" in -*) ;; *) rm -rf "$state/$arg" ;; esac + done + ;; +esac +exit 0 +""" + +_UV_STUB = r"""#!/usr/bin/env bash +dir="$(cd "$(dirname "$0")" && pwd)" +case "$*" in + *"print(sys.executable)"*) echo "$dir/python" ;; +esac +exit 0 +""" + +_PYTHON_STUB = r"""#!/usr/bin/env bash +dir="$(cd "$(dirname "$0")" && pwd)" +if [ "${1:-}" = "-m" ] && [ "${2:-}" = "uvicorn" ]; then + port="" + previous="" + for arg in "$@"; do + [ "$previous" = "--port" ] && port="$arg" + previous="$arg" + done + exec "$REAL_PYTHON" "$dir/fake_gateway.py" app.gateway.app:app "$port" +fi +exec "$REAL_PYTHON" "$@" +""" + +_FAKE_GATEWAY = """import sys +from http.server import BaseHTTPRequestHandler, HTTPServer + + +class Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(200) + self.end_headers() + self.wfile.write(b"{}") + + def log_message(self, *args): + pass + + +HTTPServer(("127.0.0.1", int(sys.argv[2])), Handler).serve_forever() +""" + +# Ports listed in busy-from- report busy from that call number on; every +# other query falls through to the real lsof (or "free" when it is missing). +_LSOF_STUB = r"""#!/usr/bin/env bash +dir="$(cd "$(dirname "$0")" && pwd)" +port="" +for arg in "$@"; do + case "$arg" in -iTCP:*) port="${arg#-iTCP:}" ;; esac +done +count=$(( $(cat "$dir/lsof-count-$port" 2>/dev/null || echo 0) + 1 )) +echo "$count" >"$dir/lsof-count-$port" +if [ -f "$dir/busy-from-$port" ] && [ "$count" -ge "$(cat "$dir/busy-from-$port")" ]; then + echo 99999 + exit 0 +fi +[ -n "$REAL_LSOF" ] || exit 1 +exec "$REAL_LSOF" "$@" +""" + + +def _free_port() -> int: + import socket + + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +@pytest.fixture +def stubbed_harness(tmp_path): + if os.name == "nt" or shutil.which("bash") is None or shutil.which("curl") is None: + pytest.skip("POSIX shell lifecycle test") + stubs = tmp_path / "stubs" + stubs.mkdir() + for name, body in (("docker", _DOCKER_STUB), ("uv", _UV_STUB), ("python", _PYTHON_STUB), ("lsof", _LSOF_STUB), ("fake_gateway.py", _FAKE_GATEWAY)): + path = stubs / name + path.write_text(body, encoding="utf-8") + path.chmod(0o755) + state = tmp_path / "mi state" # whitespace on purpose + ports = {"a": _free_port(), "b": _free_port(), "pg": _free_port(), "redis": _free_port()} + env = {key: value for key, value in os.environ.items() if not key.startswith(("DEERFLOW_MI_", "DEER_FLOW_"))} + env.update( + PATH=f"{stubs}{os.pathsep}{os.environ.get('PATH', '')}", + REAL_PYTHON=sys.executable, + REAL_LSOF=shutil.which("lsof") or "", + DEERFLOW_MI_STATE_DIR=str(state), + DEERFLOW_MI_BASE_CONFIG=str(REPO_ROOT / "config.example.yaml"), + DEERFLOW_MI_GATEWAY_A_PORT=str(ports["a"]), + DEERFLOW_MI_GATEWAY_B_PORT=str(ports["b"]), + DEERFLOW_MI_POSTGRES_PORT=str(ports["pg"]), + DEERFLOW_MI_REDIS_PORT=str(ports["redis"]), + DEERFLOW_MI_HEALTH_TIMEOUT="20", + ) + + def run(*args: str) -> subprocess.CompletedProcess: + return subprocess.run(["bash", str(SHELL_PATH), *args], capture_output=True, text=True, timeout=120, env=env) + + yield stubs, state, ports, run + + # Never leak a fake Gateway, whatever the script did. + for pid_file in (state / "run").glob("*.pid") if (state / "run").is_dir() else []: + try: + os.kill(int(pid_file.read_text().strip()), 9) + except (ValueError, ProcessLookupError): + pass + + +def _port_open(port: int) -> bool: + import socket + + with socket.socket() as sock: + sock.settimeout(0.5) + return sock.connect_ex(("127.0.0.1", port)) == 0 + + +def _container(stubs: Path, name: str, *, state_dir: str) -> Path: + path = stubs / "containers" / name + path.mkdir(parents=True) + (path / "deerflow.harness").write_text("dev-multi-instance", encoding="utf-8") + (path / "deerflow.harness.state-dir").write_text(state_dir, encoding="utf-8") + return path + + +def test_up_tears_everything_down_when_b_loses_its_port_after_preflight(stubbed_harness): + stubs, state, ports, run = stubbed_harness + (stubs / f"busy-from-{ports['b']}").write_text("2", encoding="utf-8") # free at preflight, taken when B starts + + result = run("up", "--skip-install", "--no-nginx") + + assert result.returncode != 0 + assert f"Port {ports['b']} is in use" in result.stderr + assert "Gateway A ready" in result.stdout, "A must have started before B failed" + assert not _port_open(ports["a"]), "Gateway A must be stopped when up fails" + assert "Terminated" not in result.stderr, "the Gateway is not a job of the script shell; killing it must not print job notices" + assert not list((stubs / "containers").iterdir()), "both containers must be removed" + removals = [line for line in (stubs / "docker.log").read_text().splitlines() if line.startswith("docker rm")] + assert removals and all(line.startswith("docker rm -f -v ") for line in removals), removals + + +def test_up_refuses_containers_owned_by_another_state_dir(stubbed_harness): + stubs, state, _ports, run = stubbed_harness + foreign = _container(stubs, "deerflow-mi-postgres", state_dir="/elsewhere/.deer-flow/multi-instance") + + result = run("up", "--skip-install", "--no-nginx") + + assert result.returncode != 0 + assert "/elsewhere/.deer-flow/multi-instance" in result.stderr + assert foreign.is_dir() + commands = (stubs / "docker.log").read_text() + assert "docker rm" not in commands and "docker run" not in commands + + down = run("down") + + assert foreign.is_dir(), "down must not remove another harness's container either" + assert "docker rm" not in (stubs / "docker.log").read_text(), down.stdout + down.stderr + + +def test_up_replaces_its_own_stale_containers_with_their_volumes(stubbed_harness): + stubs, state, ports, run = stubbed_harness + _container(stubs, "deerflow-mi-redis", state_dir=str(state)) + (stubs / f"busy-from-{ports['b']}").write_text("2", encoding="utf-8") # stop after A so the test stays short + + run("up", "--skip-install", "--no-nginx") + + log = (stubs / "docker.log").read_text().splitlines() + first_run = next(i for i, line in enumerate(log) if line.startswith("docker run")) + assert "docker rm -f -v deerflow-mi-redis" in log[:first_run] + labelled = [line for line in log if line.startswith("docker run")] + assert all(f"--label deerflow.harness.state-dir={state}" in line for line in labelled), labelled diff --git a/scripts/AGENTS.md b/scripts/AGENTS.md index 8e7812dcb6..7501855b99 100644 --- a/scripts/AGENTS.md +++ b/scripts/AGENTS.md @@ -167,6 +167,12 @@ defaults `DEER_FLOW_PROJECT_ROOT` to the checkout, as `serve.sh` does, so Only "no config anywhere" creates `/config.yaml` from the example. `backend/tests/test_config_version.py::test_config_upgrade_*` pins this. +## Multi-Instance Dev Harness + +`dev_multi_instance.{sh,py}` must keep the rendered config passing the real +multi-instance gate and every endpoint/data root inside its state dir +(`backend/tests/test_dev_multi_instance_script.py`). + ## Shell Script Invocation Contract Root Makefile recipes must invoke repository `.sh` files through diff --git a/scripts/dev_multi_instance.py b/scripts/dev_multi_instance.py new file mode 100644 index 0000000000..f8effec884 --- /dev/null +++ b/scripts/dev_multi_instance.py @@ -0,0 +1,783 @@ +#!/usr/bin/env python3 +"""Helpers for ``scripts/dev_multi_instance.sh``, the two-Gateway local harness. + +The shell entry point owns containers and processes. This module owns the parts +that decide whether the pair can start and whether it behaves as one cluster: + +* ``render-config`` overlays the multi-instance prerequisites on a base + ``config.yaml`` so the result passes the Gateway's multi-instance startup gate + (``app.gateway.deps._enforce_postgres_for_multi_worker``) and leaves every + other section of the developer's config alone. +* ``secrets`` generates the secrets every Gateway process must share. +* ``nginx-conf`` renders a minimal round-robin nginx front for both Gateways. +* ``check`` runs the scriptable cross-instance acceptance checks. + +It needs PyYAML and httpx, so run it with the backend environment:: + + uv run --no-sync --project backend python scripts/dev_multi_instance.py --help +""" + +from __future__ import annotations + +import argparse +import base64 +import copy +import hashlib +import json +import os +import secrets +import sys +import time +from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +# Each Gateway exports its own value; the shared config.yaml references it so +# one file serves both processes (the Helm chart does the same with a per-Pod +# emptyDir mounted at one path). +RETRIEVAL_INDEX_ENV_VAR = "DEERFLOW_MI_RETRIEVAL_INDEX_PATH" + +# Every Gateway process must see identical values. DEER_FLOW_CREDENTIALS_KEY is +# the at-rest credentials key: without one shared value a declared +# multi-instance pair with channel_connections enabled refuses to start. +SHARED_SECRET_NAMES = ( + "AUTH_JWT_SECRET", + "DEER_FLOW_INTERNAL_AUTH_TOKEN", + "DEER_FLOW_CREDENTIALS_KEY", + "DEERFLOW_MI_POSTGRES_PASSWORD", +) + +GENERATED_HEADER = "# Generated by scripts/dev_multi_instance.sh on every `up`; edit the base config instead.\n# Base config: {base}\n" + + +# --------------------------------------------------------------------------- +# Config overlay +# --------------------------------------------------------------------------- + + +def _section(config: dict[str, Any], name: str) -> dict[str, Any]: + value = config.get(name) + if not isinstance(value, dict): + value = {} + config[name] = value + return value + + +def _is_browser_tool(tool: Any) -> bool: + if not isinstance(tool, Mapping): + return False + return str(tool.get("name") or "").startswith("browser_") or "browser_automation" in str(tool.get("use") or "") + + +def _is_local_aio_sandbox(sandbox: Mapping[str, Any]) -> bool: + """AIO sandboxes run as named Docker containers on this host unless a provisioner owns them.""" + return "aio_sandbox" in str(sandbox.get("use") or "") and not sandbox.get("provisioner_url") + + +def sandbox_container_prefix(state_dir: Path | str) -> str: + """Docker name prefix for the local AIO sandboxes of one harness state directory. + + Stable for a state directory (warm sandboxes survive restarts) and distinct + from the provider default, so the harness never adopts or reaps the + sandbox containers of `make dev` or of another harness. + """ + digest = hashlib.sha256(os.path.abspath(state_dir).encode("utf-8")).hexdigest()[:10] + return f"deer-flow-mi-{digest}-sandbox" + + +def build_multi_instance_config( + base: Mapping[str, Any] | None, + *, + database_url: str, + redis_url: str, + sandbox_container_prefix: str, + keep_channels: bool = False, +) -> tuple[dict[str, Any], list[str]]: + """Return ``(config, notes)``: *base* plus every multi-instance prerequisite. + + Everything the harness writes must land in its own Postgres, Redis, home + and containers, never in the developer's own deployment, so every + persistence endpoint or data root the base config names is redirected. + ``notes`` describes each change beyond the always-applied keys, so the + harness can tell the developer what differs from their own config. + """ + config: dict[str, Any] = copy.deepcopy(dict(base or {})) + notes: list[str] = [] + + _section(config, "deployment")["multi_instance"] = True + + database = _section(config, "database") + database["backend"] = "postgres" + database["postgres_url"] = database_url + cache = database.get("checkpoint_cache") + if isinstance(cache, dict) and (cache.get("type") == "redis" or cache.get("redis_url")): + cache["redis_url"] = redis_url + notes.append("pointed database.checkpoint_cache at the harness Redis") + + _section(config, "run_events")["backend"] = "db" + _section(config, "run_ownership")["heartbeat_enabled"] = True + + bridge = _section(config, "stream_bridge") + bridge["type"] = "redis" + bridge["redis_url"] = redis_url + + checkpointer = config.pop("checkpointer", None) + if checkpointer is not None: + # An explicit checkpointer section wins over `database` for LangGraph + # checkpoints and the Store: a sqlite/memory one would split threads per + # process, a postgres one would write into the developer's database. + kind = checkpointer.get("type") if isinstance(checkpointer, Mapping) else None + notes.append(f"removed checkpointer (type {kind!r}); the harness Postgres holds checkpoints and the Store") + + scheduler = config.get("scheduler") + if isinstance(scheduler, dict) and scheduler.get("enabled") and not scheduler.get("multi_instance"): + scheduler["multi_instance"] = True + notes.append("set scheduler.multi_instance: true (the scheduler is enabled; the gate refuses a per-process scheduler)") + + tools = config.get("tools") + if isinstance(tools, list): + removed = [str(tool.get("name")) for tool in tools if _is_browser_tool(tool)] + if removed: + config["tools"] = [tool for tool in tools if not _is_browser_tool(tool)] + notes.append(f"removed process-local browser tools: {', '.join(removed)}") + + sandbox = config.get("sandbox") + if isinstance(sandbox, dict): + ownership = sandbox.get("ownership") + if ownership is not None: + # An explicit section stops ownership from inheriting the bridge's + # Redis, so its endpoint must be rewritten; timing settings stay. + ownership = dict(ownership) if isinstance(ownership, Mapping) else {} + previous = f"type {ownership.get('type', 'memory')!r}" + (f", {ownership['redis_url']}" if ownership.get("redis_url") else "") + ownership.update(type="redis", redis_url=redis_url) + sandbox["ownership"] = ownership + notes.append(f"pointed sandbox.ownership at the harness Redis (was {previous})") + if _is_local_aio_sandbox(sandbox): + previous_prefix = sandbox.get("container_prefix") or "deer-flow-sandbox" + sandbox["container_prefix"] = sandbox_container_prefix + notes.append(f"set sandbox.container_prefix: {sandbox_container_prefix} (was {previous_prefix!r}) so the harness never adopts or reaps other sandbox containers") + + channels = config.get("channels") + if not keep_channels and isinstance(channels, dict): + disabled = sorted(name for name, value in channels.items() if isinstance(value, dict) and value.get("enabled")) + for name in disabled: + channels[name]["enabled"] = False + if disabled: + notes.append(f"disabled IM channels {', '.join(disabled)}: both Gateways would connect the same bot (set DEERFLOW_MI_KEEP_CHANNELS=1 to keep them)") + + memory = _section(config, "memory") + manager_class = memory.get("manager_class") + if manager_class in (None, "", "deermem"): + backend_config = memory.get("backend_config") + if not isinstance(backend_config, dict): + backend_config = {} + memory["backend_config"] = backend_config + # Empty storage_path means the harness DEER_FLOW_HOME. Both spellings go: + # a legacy top-level memory.storage_path is migrated into backend_config on load. + dropped = [f"{label}={value!r}" for label, value in (("memory.storage_path", memory.pop("storage_path", None)), ("memory.backend_config.storage_path", backend_config.pop("storage_path", None))) if value] + if dropped: + notes.append(f"dropped {', '.join(dropped)}: DeerMem data stays in the harness home") + backend_config["retrieval_index_path"] = f"${RETRIEVAL_INDEX_ENV_VAR}" + else: + notes.append(f"memory.manager_class={manager_class!r} is not DeerMem: the harness uses that backend as configured, shared with your own deployment") + + blob_storage = config.get("blob_storage") + blob_config = blob_storage.get("backend_config") if isinstance(blob_storage, dict) else None + if isinstance(blob_config, dict) and blob_config.get("root"): + notes.append(f"dropped blob_storage.backend_config.root={blob_config.pop('root')!r}: blobs stay in the harness home") + + return config, notes + + +def _write_private_text(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(f".{path.name}.tmp") + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(text) + os.chmod(tmp, 0o600) + os.replace(tmp, path) + + +def render_config(base_path: Path, out_path: Path, *, database_url: str, redis_url: str, state_dir: Path, keep_channels: bool = False) -> list[str]: + """Write the generated multi-instance config (mode 0600) and return its notes.""" + import yaml + + base = yaml.safe_load(Path(base_path).read_text(encoding="utf-8-sig")) or {} + if not isinstance(base, dict): + raise SystemExit(f"{base_path} is not a YAML mapping") + config, notes = build_multi_instance_config( + base, + database_url=database_url, + redis_url=redis_url, + sandbox_container_prefix=sandbox_container_prefix(state_dir), + keep_channels=keep_channels, + ) + body = yaml.safe_dump(config, sort_keys=False, allow_unicode=True, default_flow_style=False) + _write_private_text(Path(out_path), GENERATED_HEADER.format(base=base_path) + body) + return notes + + +# --------------------------------------------------------------------------- +# Shared secrets +# --------------------------------------------------------------------------- + + +def generate_shared_secrets() -> dict[str, str]: + return { + "AUTH_JWT_SECRET": secrets.token_urlsafe(32), + "DEER_FLOW_INTERNAL_AUTH_TOKEN": secrets.token_urlsafe(32), + # A raw urlsafe-base64 32-byte Fernet key, not a passphrase. + "DEER_FLOW_CREDENTIALS_KEY": base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("ascii"), + "DEERFLOW_MI_POSTGRES_PASSWORD": secrets.token_hex(16), + } + + +def read_env_file(path: Path) -> dict[str, str]: + values: dict[str, str] = {} + for line in Path(path).read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + values[key.strip()] = value.strip() + return values + + +def ensure_secrets_file(path: Path) -> dict[str, str]: + """Create the shell-sourceable secrets file once (``O_EXCL``, 0600) and return it.""" + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + values = generate_shared_secrets() + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError: + return read_env_file(path) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write("# Shared by every Gateway of the dev_multi_instance harness. Do not commit.\n") + for name in SHARED_SECRET_NAMES: + fh.write(f"{name}={values[name]}\n") + return values + + +# --------------------------------------------------------------------------- +# nginx front +# --------------------------------------------------------------------------- + +_PROXY_HEADERS = """ proxy_set_header Host $http_host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme;""" + + +def _nginx_path(path: Path) -> str: + """Render *path* as an nginx double-quoted string literal. + + Quoting keeps whitespace and ``;``/``{`` inside one argument. ``$`` cannot be + escaped (nginx interpolates it in directives such as ``access_log``) and + control characters cannot be spelled, so both are refused. + """ + text = str(path) + if "$" in text or any(ord(char) < 32 for char in text): + raise ValueError(f"nginx cannot reference {text!r}: choose a state directory without '$' or control characters") + return '"' + text.replace("\\", "\\\\").replace('"', '\\"') + '"' + + +def render_nginx_conf(*, listen_port: int, gateway_ports: Sequence[int], frontend_port: int, state_dir: Path) -> str: + """Return a minimal nginx config that round-robins ``/api`` over every Gateway.""" + state_dir = Path(state_dir) + run_dir, log_dir, temp_dir = state_dir / "run", state_dir / "logs", state_dir / "nginx" / "temp" + servers = "\n".join(f" server 127.0.0.1:{port};" for port in gateway_ports) + return f"""# Generated by scripts/dev_multi_instance.sh; regenerated on every `up`. +worker_processes 1; +pid {_nginx_path(run_dir / "nginx.pid")}; +error_log {_nginx_path(log_dir / "nginx-error.log")} warn; + +events {{ + worker_connections 1024; +}} + +http {{ + access_log {_nginx_path(log_dir / "nginx-access.log")}; + client_body_temp_path {_nginx_path(temp_dir / "client_body")}; + proxy_temp_path {_nginx_path(temp_dir / "proxy")}; + fastcgi_temp_path {_nginx_path(temp_dir / "fastcgi")}; + uwsgi_temp_path {_nginx_path(temp_dir / "uwsgi")}; + scgi_temp_path {_nginx_path(temp_dir / "scgi")}; + + upstream deerflow_gateways {{ +{servers} + }} + + upstream deerflow_frontend {{ + server 127.0.0.1:{frontend_port}; + }} + + map $http_upgrade $connection_upgrade {{ + default upgrade; + '' ''; + }} + + server {{ + listen 127.0.0.1:{listen_port}; + server_name _; + + # Which Gateway answered: lets `check` and developers see the balancing. + add_header X-DeerFlow-Upstream $upstream_addr always; + + proxy_http_version 1.1; + proxy_buffering off; + proxy_cache off; + proxy_request_buffering off; + client_max_body_size 101M; + proxy_connect_timeout 10s; + proxy_send_timeout 600s; + proxy_read_timeout 600s; + + location /api/langgraph/ {{ + rewrite ^/api/langgraph/(.*) /api/$1 break; +{_PROXY_HEADERS} + proxy_set_header Connection ''; + proxy_pass http://deerflow_gateways; + }} + + location /api/ {{ +{_PROXY_HEADERS} + proxy_set_header Connection ''; + proxy_pass http://deerflow_gateways; + }} + + location /health {{ +{_PROXY_HEADERS} + proxy_pass http://deerflow_gateways; + }} + + location / {{ +{_PROXY_HEADERS} + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection $connection_upgrade; + proxy_pass http://deerflow_frontend; + }} + }} +}} +""" + + +# --------------------------------------------------------------------------- +# Cross-instance checks +# --------------------------------------------------------------------------- + +PASS, FAIL, SKIP = "PASS", "FAIL", "SKIP" + + +@dataclass +class CheckResult: + name: str + status: str + detail: str + + +@dataclass +class SSEEvent: + id: str | None + event: str | None + data: str + + +def parse_sse_events(lines: Iterable[str]) -> Iterator[SSEEvent]: + """Yield server-sent events from decoded lines (blank line = dispatch).""" + event_id: str | None = None + event_name: str | None = None + data: list[str] = [] + for raw in lines: + line = raw.rstrip("\r\n") + if not line: + if event_name is not None or data or event_id is not None: + yield SSEEvent(event_id, event_name, "\n".join(data)) + event_id, event_name, data = None, None, [] + continue + if line.startswith(":"): + continue + name, _, value = line.partition(":") + value = value[1:] if value.startswith(" ") else value + if name == "id": + event_id = value + elif name == "event": + event_name = value + elif name == "data": + data.append(value) + if event_name is not None or data or event_id is not None: + yield SSEEvent(event_id, event_name, "\n".join(data)) + + +def exit_code(results: Iterable[CheckResult]) -> int: + return 1 if any(result.status == FAIL for result in results) else 0 + + +class CheckFailed(Exception): + """A check's expectation did not hold; the message is the reported detail.""" + + +class CheckSkipped(Exception): + """A check cannot run in this setup; the message says why.""" + + +@dataclass +class Harness: + """State shared by the checks: two Gateways, an optional nginx, one session.""" + + gateways: list[tuple[str, str]] + nginx_url: str | None + state_dir: Path + with_llm: bool + timeout: float + internal_token: str = "" + cookies: dict[str, str] = field(default_factory=dict) + user: dict[str, Any] = field(default_factory=dict) + is_admin: bool = False + thread_id: str | None = None + models: list[Any] | None = None + + def __post_init__(self) -> None: + import httpx + + # Every target is loopback: never route it through an HTTP(S)_PROXY from the shell. + self.http = httpx.Client(timeout=httpx.Timeout(self.timeout, read=max(self.timeout, 60.0)), follow_redirects=False, trust_env=False) + + # -- request helpers ----------------------------------------------------- + + def headers(self, *, mutating: bool = False) -> dict[str, str]: + headers: dict[str, str] = {} + if self.cookies: + headers["Cookie"] = "; ".join(f"{name}={value}" for name, value in self.cookies.items()) + if mutating and "csrf_token" in self.cookies: + headers["X-CSRF-Token"] = self.cookies["csrf_token"] + return headers + + def request(self, method: str, base: str, path: str, *, expect: Iterable[int] = (200,), session: bool = True, **kwargs: Any): + mutating = method.upper() in {"POST", "PUT", "PATCH", "DELETE"} + headers = {**(self.headers(mutating=mutating) if session else {}), **kwargs.pop("headers", {})} + response = self.http.request(method, f"{base}{path}", headers=headers, **kwargs) + self.http.cookies.clear() # cookies travel explicitly, never through the jar + if response.status_code not in tuple(expect): + raise CheckFailed(f"{method} {base}{path} -> HTTP {response.status_code}: {response.text[:300]}") + return response + + @property + def a(self) -> str: + return self.gateways[0][1] + + @property + def b(self) -> str: + return self.gateways[1][1] + + # -- session --------------------------------------------------------------- + + def _adopt_session(self, response) -> None: + for name in ("access_token", "csrf_token"): + value = response.cookies.get(name) + if value: + self.cookies[name] = value + self.http.cookies.clear() + if "access_token" not in self.cookies: + raise CheckFailed(f"{response.request.url} set no access_token cookie") + + def ensure_session(self) -> str: + """Log in as the harness's throwaway user, creating it on first use.""" + creds_path = self.state_dir / "check-user.json" + if creds_path.is_file(): + creds = json.loads(creds_path.read_text(encoding="utf-8")) + response = self.request("POST", self.a, "/api/v1/auth/login/local", data={"username": creds["email"], "password": creds["password"]}, expect=(200, 401)) + if response.status_code == 200: + self._adopt_session(response) + self.is_admin = bool(creds.get("admin")) + return f"logged in as {creds['email']} on A" + creds = {"email": f"mi-check-{secrets.token_hex(4)}@example.com", "password": secrets.token_urlsafe(18)} + setup = self.request("GET", self.a, "/api/v1/auth/setup-status").json() + if setup.get("needs_setup"): + response = self.request("POST", self.a, "/api/v1/auth/initialize", json=creds, expect=(201,)) + creds["admin"] = True + else: + response = self.request("POST", self.a, "/api/v1/auth/register", json=creds, expect=(201,)) + creds["admin"] = False + self._adopt_session(response) + self.is_admin = creds["admin"] + _write_private_text(creds_path, json.dumps(creds)) + role = "admin" if self.is_admin else "user" + return f"created throwaway {role} {creds['email']} on A (credentials: {creds_path})" + + +def _check_health(h: Harness) -> str: + seen = [] + for name, base in h.gateways: + body = h.request("GET", base, "/health/ready").json() + if body.get("status") != "ready" or body.get("stream_bridge") != "ok": + raise CheckFailed(f"{name} /health/ready: {body}") + seen.append(f"{name}: database={body.get('database')} stream_bridge={body.get('stream_bridge')}") + return "; ".join(seen) + + +def _check_session(h: Harness) -> str: + created = h.ensure_session() + me_a = h.request("GET", h.a, "/api/v1/auth/me").json() + me_b = h.request("GET", h.b, "/api/v1/auth/me").json() + if me_a.get("id") != me_b.get("id"): + raise CheckFailed(f"A sees user {me_a.get('id')}, B sees {me_b.get('id')}") + h.user = me_a + return f"{created}; B accepts A's session (shared AUTH_JWT_SECRET)" + + +def _check_internal_token(h: Harness) -> str: + if not h.internal_token: + raise CheckSkipped("no DEER_FLOW_INTERNAL_AUTH_TOKEN in the secrets file") + bodies = [] + for _name, base in h.gateways: + # Negative control first: without a session, a wrong token must be refused. + h.request("GET", base, "/api/models", session=False, headers={"X-DeerFlow-Internal-Token": "not-the-token"}, expect=(401,)) + bodies.append(h.request("GET", base, "/api/models", session=False, headers={"X-DeerFlow-Internal-Token": h.internal_token}).json()) + models = bodies[0].get("models", []) if isinstance(bodies[0], dict) else [] + h.models = models + return f"both Gateways accept the shared DEER_FLOW_INTERNAL_AUTH_TOKEN ({len(models)} model(s) configured)" + + +def _require_thread(h: Harness) -> str: + if h.thread_id is None: + raise CheckSkipped("no thread (thread check did not pass)") + return h.thread_id + + +def _check_thread(h: Harness) -> str: + created = h.request("POST", h.a, "/api/threads", json={"metadata": {"source": "dev_multi_instance check"}}).json() + thread_id = created["thread_id"] + fetched = h.request("GET", h.b, f"/api/threads/{thread_id}").json() + if fetched.get("thread_id") != thread_id: + raise CheckFailed(f"B returned {fetched}") + h.thread_id = thread_id + return f"thread {thread_id} created on A is readable on B" + + +def _check_upload(h: Harness) -> str: + thread_id = _require_thread(h) + name = f"mi-check-{secrets.token_hex(4)}.txt" + payload = f"multi-instance upload {secrets.token_hex(16)}\n".encode() + uploaded = h.request("POST", h.a, f"/api/threads/{thread_id}/uploads", files={"files": (name, payload, "text/plain")}).json() + info = next((item for item in uploaded.get("files", []) if item.get("filename") == name), None) + if info is None: + raise CheckFailed(f"A did not report {name}: {uploaded}") + listed = h.request("GET", h.b, f"/api/threads/{thread_id}/uploads/list").json() + if name not in {item.get("filename") for item in listed.get("files", [])}: + raise CheckFailed(f"B does not list {name}: {listed}") + body = h.request("GET", h.b, info["artifact_url"]).content + if body != payload: + raise CheckFailed(f"B served {len(body)} bytes that differ from the {len(payload)} uploaded on A") + return f"uploaded on A; B lists it and serves identical bytes via {info['artifact_url']}" + + +def _check_skill_toggle(h: Harness) -> str: + if not h.is_admin: + raise CheckSkipped("the check user is not an admin (an admin already existed); PUT /api/skills is admin-only") + skills = h.request("GET", h.a, "/api/skills").json().get("skills", []) + skill = next((s for s in skills if s.get("category") == "public"), None) + if skill is None: + raise CheckSkipped("no public skill to toggle") + name, original = skill["name"], bool(skill.get("enabled", True)) + h.request("PUT", h.a, f"/api/skills/{name}", json={"enabled": not original}) + try: + started = time.monotonic() + deadline = started + 30 + while True: + seen = {s["name"]: s.get("enabled") for s in h.request("GET", h.b, "/api/skills").json().get("skills", [])} + if seen.get(name) is (not original): + return f"skill {name!r} toggled on A is {'enabled' if not original else 'disabled'} on B after {time.monotonic() - started:.1f}s (restored)" + if time.monotonic() > deadline: + raise CheckFailed(f"B still reports {name!r} enabled={seen.get(name)} 30s after A toggled it") + time.sleep(1) + finally: + h.request("PUT", h.a, f"/api/skills/{name}", json={"enabled": original}) + + +def _check_nginx(h: Harness) -> str: + if not h.nginx_url: + raise CheckSkipped("nginx is not running (up --no-nginx, or nginx not installed)") + expected = {base.split("://", 1)[1] for _name, base in h.gateways} + upstreams = set() + for _ in range(2 * len(h.gateways) + 2): + response = h.request("GET", h.nginx_url, "/health") + upstreams.add(response.headers.get("x-deerflow-upstream", "")) + if not expected <= upstreams: + raise CheckFailed(f"nginx reached {sorted(upstreams)}, expected every Gateway {sorted(expected)}") + detail = f"round-robin reached {', '.join(sorted(expected))}" + if h.thread_id: + for _ in range(len(h.gateways) * 2): + h.request("GET", h.nginx_url, f"/api/threads/{h.thread_id}") + detail += f"; thread {h.thread_id} readable through every upstream" + return detail + + +def bounded_lines(lines: Iterable[str], *, deadline: float, what: str, clock: Callable[[], float] = time.monotonic) -> Iterator[str]: + """Yield raw SSE lines until *deadline*, counting comment heartbeats too. + + The production stream keeps an idle live run open with ``: heartbeat`` + comments, which the event parser drops and which also reset httpx's read + timeout, so the budget must be enforced per raw line, not per event. + """ + for line in lines: + if clock() > deadline: + raise CheckFailed(f"{what} exceeded its time budget without reaching `end`") + yield line + + +def _read_stream(h: Harness, base: str, path: str, *, last_event_id: str | None = None, budget: float = 90.0) -> list[SSEEvent]: + headers = {**h.headers(), "Accept": "text/event-stream"} + if last_event_id: + headers["Last-Event-ID"] = last_event_id + events: list[SSEEvent] = [] + deadline = time.monotonic() + budget + try: + with h.http.stream("GET", f"{base}{path}", headers=headers) as response: + if response.status_code != 200: + response.read() + raise CheckFailed(f"GET {base}{path} -> HTTP {response.status_code}: {response.text[:300]}") + for event in parse_sse_events(bounded_lines(response.iter_lines(), deadline=deadline, what=f"GET {base}{path} ({budget:g}s budget)")): + events.append(event) + if event.event == "end": + break + finally: + h.http.cookies.clear() + return events + + +def _check_sse_resume(h: Harness) -> str: + thread_id = _require_thread(h) + if not h.with_llm: + # Only a confirmed empty model list makes the run free (it fails before any LLM call). + if h.models is None: + raise CheckSkipped("model discovery did not succeed, so a run might call an LLM; rerun with `check --with-llm`") + if h.models: + raise CheckSkipped(f"{len(h.models)} model(s) configured: a run would call the LLM; rerun with `check --with-llm`") + body = {"assistant_id": "lead_agent", "input": {"messages": [{"role": "user", "content": "Reply with the single word: pong"}]}} + run = h.request("POST", h.a, f"/api/threads/{thread_id}/runs", json=body).json() + run_id = run["run_id"] + path = f"/api/threads/{thread_id}/runs/{run_id}/join" + full = _read_stream(h, h.b, path) + ids = [event.id for event in full if event.id] + if not full or full[-1].event != "end": + raise CheckFailed(f"joining on B did not reach `end`: {[e.event for e in full]}") + if len(ids) < 2: + raise CheckSkipped(f"run {run_id} produced only {len(ids)} event id(s); nothing to resume from") + resumed = _read_stream(h, h.a, path, last_event_id=ids[0]) + resumed_ids = [event.id for event in resumed if event.id] + if resumed_ids != ids[1:]: + raise CheckFailed(f"resuming on A after {ids[0]} returned ids {resumed_ids}, expected {ids[1:]}") + status = h.request("GET", h.b, f"/api/threads/{thread_id}/runs/{run_id}").json().get("status") + kind = "no-model (fails fast, no LLM call)" if h.models == [] else "LLM" + return f"{kind} run {run_id} created on A (status {status}); B replayed {len(full)} events; A resumed from Last-Event-ID {ids[0]} with the remaining {len(resumed_ids)}" + + +CHECKS: list[tuple[str, Callable[[Harness], str]]] = [ + ("health/ready on both (stream_bridge ok)", _check_health), + ("session from A accepted by B", _check_session), + ("shared internal auth token", _check_internal_token), + ("thread created on A visible on B", _check_thread), + ("upload on A, list + download on B", _check_upload), + ("skill toggle on A seen by B", _check_skill_toggle), + ("nginx balances both Gateways", _check_nginx), + ("SSE join on B, Last-Event-ID resume on A", _check_sse_resume), +] + +MANUAL_CHECKS = ( + "`stop b --kill` while A has an active run: A's run must not turn `error`", + "cancel a run on B from A: takes effect within about one heartbeat (~10s)", + "change an MCP filesystem server's args on A: B's local-bash allowlist follows", + "edit a skill on A: B's system-prompt skills section changes within 30s", + "IM channel gates (one leader per platform, failover, runtime-config): once channel leader election lands", +) + + +def run_checks(h: Harness, *, out=sys.stdout) -> list[CheckResult]: + results: list[CheckResult] = [] + for name, check in CHECKS: + try: + result = CheckResult(name, PASS, check(h)) + except CheckSkipped as exc: + result = CheckResult(name, SKIP, str(exc)) + except CheckFailed as exc: + result = CheckResult(name, FAIL, str(exc)) + except Exception as exc: # connection errors, unexpected payloads + result = CheckResult(name, FAIL, f"{type(exc).__name__}: {exc}") + results.append(result) + print(f"[{result.status}] {result.name}: {result.detail}", file=out, flush=True) + counts = {status: sum(1 for r in results if r.status == status) for status in (PASS, FAIL, SKIP)} + print(f"\n{counts[PASS]} passed, {counts[FAIL]} failed, {counts[SKIP]} skipped", file=out) + print('Still manual (backend/docs/CONFIGURATION.md, "Local two-Gateway harness"):', file=out) + for item in MANUAL_CHECKS: + print(f" - {item}", file=out) + return results + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def _parse_gateway(value: str) -> tuple[str, str]: + name, sep, url = value.partition("=") + if not sep or not name or not url: + raise argparse.ArgumentTypeError("expected NAME=URL, e.g. A=http://127.0.0.1:8001") + return name, url.rstrip("/") + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + sub = parser.add_subparsers(dest="command", required=True) + + render = sub.add_parser("render-config", help="write the multi-instance config.yaml") + render.add_argument("--base", required=True, type=Path) + render.add_argument("--out", required=True, type=Path) + render.add_argument("--database-url", required=True) + render.add_argument("--redis-url", required=True) + render.add_argument("--state-dir", required=True, type=Path) + render.add_argument("--keep-channels", action="store_true") + + secrets_cmd = sub.add_parser("secrets", help="create the shared secrets file if missing") + secrets_cmd.add_argument("--out", required=True, type=Path) + + nginx = sub.add_parser("nginx-conf", help="write the round-robin nginx config") + nginx.add_argument("--out", required=True, type=Path) + nginx.add_argument("--state-dir", required=True, type=Path) + nginx.add_argument("--listen-port", required=True, type=int) + nginx.add_argument("--gateway-port", required=True, type=int, action="append") + nginx.add_argument("--frontend-port", type=int, default=3000) + + check = sub.add_parser("check", help="run the cross-instance checks") + check.add_argument("--gateway", required=True, type=_parse_gateway, action="append", help="NAME=URL; pass exactly two") + check.add_argument("--nginx", default=None) + check.add_argument("--state-dir", required=True, type=Path) + check.add_argument("--with-llm", action="store_true", help="allow checks that start a run on a configured model") + check.add_argument("--timeout", type=float, default=15.0) + + args = parser.parse_args(argv) + + if args.command == "render-config": + for note in render_config(args.base, args.out, database_url=args.database_url, redis_url=args.redis_url, state_dir=args.state_dir, keep_channels=args.keep_channels): + print(f" note: {note}") + return 0 + if args.command == "secrets": + ensure_secrets_file(args.out) + return 0 + if args.command == "nginx-conf": + _write_private_text(args.out, render_nginx_conf(listen_port=args.listen_port, gateway_ports=args.gateway_port, frontend_port=args.frontend_port, state_dir=args.state_dir)) + return 0 + if args.command == "check": + if len(args.gateway) != 2: + parser.error("check needs exactly two --gateway NAME=URL values") + secrets_path = args.state_dir / "secrets.env" + token = read_env_file(secrets_path).get("DEER_FLOW_INTERNAL_AUTH_TOKEN", "") if secrets_path.is_file() else "" + harness = Harness(gateways=args.gateway, nginx_url=(args.nginx or "").rstrip("/") or None, state_dir=args.state_dir, with_llm=args.with_llm, timeout=args.timeout, internal_token=token) + return exit_code(run_checks(harness)) + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/dev_multi_instance.sh b/scripts/dev_multi_instance.sh new file mode 100755 index 0000000000..19e30bac99 --- /dev/null +++ b/scripts/dev_multi_instance.sh @@ -0,0 +1,675 @@ +#!/usr/bin/env bash +# +# dev_multi_instance.sh — run two Gateways as one local multi-instance cluster. +# +# Starts throwaway Postgres and Redis containers, renders a multi-instance +# config.yaml from your own config (or config.example.yaml), and runs Gateway A +# and Gateway B on one shared DEER_FLOW_HOME with shared auth secrets, plus an +# optional round-robin nginx in front of both. Run with --help for commands. +# +# Config rendering, secrets, the nginx config and the checks live in +# dev_multi_instance.py (unit-tested); this file owns containers and processes. +# Everything lives under DEERFLOW_MI_STATE_DIR so it never collides with +# `make dev` state in backend/.deer-flow. + +set -euo pipefail + +REPO_ROOT="$(builtin cd "$(dirname "${BASH_SOURCE[0]}")/.." >/dev/null 2>&1 && pwd -P)" +BACKEND_DIR="$REPO_ROOT/backend" +HELPER="$REPO_ROOT/scripts/dev_multi_instance.py" + +STATE_DIR="${DEERFLOW_MI_STATE_DIR:-$REPO_ROOT/.deer-flow/multi-instance}" +# Gateways run from backend/, so every path handed to them must be absolute. +case "$STATE_DIR" in + /*) ;; + *) STATE_DIR="$PWD/$STATE_DIR" ;; +esac +GATEWAY_A_PORT="${DEERFLOW_MI_GATEWAY_A_PORT:-8001}" +GATEWAY_B_PORT="${DEERFLOW_MI_GATEWAY_B_PORT:-8011}" +NGINX_PORT="${DEERFLOW_MI_NGINX_PORT:-2027}" +FRONTEND_PORT="${DEERFLOW_MI_FRONTEND_PORT:-3000}" +PG_PORT="${DEERFLOW_MI_POSTGRES_PORT:-55432}" +REDIS_PORT="${DEERFLOW_MI_REDIS_PORT:-56379}" +PG_CONTAINER="${DEERFLOW_MI_POSTGRES_CONTAINER:-deerflow-mi-postgres}" +REDIS_CONTAINER="${DEERFLOW_MI_REDIS_CONTAINER:-deerflow-mi-redis}" +PG_IMAGE="${DEERFLOW_MI_POSTGRES_IMAGE:-postgres:17-alpine}" +REDIS_IMAGE="${DEERFLOW_MI_REDIS_IMAGE:-redis:7-alpine}" +HEALTH_TIMEOUT="${DEERFLOW_MI_HEALTH_TIMEOUT:-180}" + +LABEL_KEY="deerflow.harness" +LABEL_VALUE="dev-multi-instance" +# Container names are global, so each container also records the state dir +# that created it; only that state dir may remove it. +STATE_LABEL_KEY="deerflow.harness.state-dir" +STATE_MARKER=".dev-multi-instance" +RUN_DIR="$STATE_DIR/run" +LOG_DIR="$STATE_DIR/logs" +CONFIG_OUT="$STATE_DIR/config.yaml" +EXTENSIONS_OUT="$STATE_DIR/extensions_config.json" +SECRETS_FILE="$STATE_DIR/secrets.env" + +usage() { + cat < [options] + +Run two DeerFlow Gateways (A and B) on shared Postgres, Redis and +DEER_FLOW_HOME: the topology that deployment.multi_instance: true targets. + +Commands: + up [--skip-install] [--no-nginx] Start containers, render the config, start A, then B, then nginx + down Stop everything; delete the containers and harness data (logs are kept) + status Show containers, Gateways and nginx + logs [a|b|nginx|all] [-f] Print the last 100 log lines (-f follows) + check [--with-llm] Run the cross-instance checks against the running pair + stop [--kill] Stop one Gateway (SIGTERM; --kill sends SIGKILL to simulate a crash) + start Start a stopped Gateway again + restart stop + start + help Show this help + +Defaults: + Gateway A http://127.0.0.1:$GATEWAY_A_PORT Gateway B http://127.0.0.1:$GATEWAY_B_PORT + nginx http://localhost:$NGINX_PORT (round-robin /api over A and B; / goes to a frontend on :$FRONTEND_PORT) + Postgres 127.0.0.1:$PG_PORT ($PG_CONTAINER) Redis 127.0.0.1:$REDIS_PORT ($REDIS_CONTAINER) + State $STATE_DIR + +Environment overrides: + DEERFLOW_MI_BASE_CONFIG base config (default: \$DEER_FLOW_CONFIG_PATH, config.yaml, + backend/config.yaml, then config.example.yaml) + DEERFLOW_MI_STATE_DIR state directory + DEERFLOW_MI_GATEWAY_A_PORT, DEERFLOW_MI_GATEWAY_B_PORT, DEERFLOW_MI_NGINX_PORT, + DEERFLOW_MI_FRONTEND_PORT, DEERFLOW_MI_POSTGRES_PORT, DEERFLOW_MI_REDIS_PORT + DEERFLOW_MI_POSTGRES_CONTAINER, DEERFLOW_MI_REDIS_CONTAINER, + DEERFLOW_MI_POSTGRES_IMAGE, DEERFLOW_MI_REDIS_IMAGE + DEERFLOW_MI_KEEP_CHANNELS=1 keep IM channels enabled (both Gateways then connect every bot) + DEERFLOW_MI_HEALTH_TIMEOUT seconds to wait for each Gateway's /health (default 180) + +'make stop' and 'make dev' reclaim port 8001 and this checkout's Gateways, so +do not run them while the harness is up. +EOF +} + +die() { + echo "✗ $*" >&2 + exit 1 +} + +upper() { + printf '%s' "$1" | tr '[:lower:]' '[:upper:]' +} + +port_in_use() { + lsof -nP -iTCP:"$1" -sTCP:LISTEN -t >/dev/null 2>&1 +} + +pid_file() { + printf '%s/%s.pid' "$RUN_DIR" "$1" +} + +# Print the pid recorded for $1 when that process is still the one we started +# ($2 is a fragment of its command line), so a reused pid is never signalled. +live_pid() { + local pid command + pid="$(cat "$(pid_file "$1")" 2>/dev/null || true)" + [ -n "$pid" ] || return 1 + kill -0 "$pid" 2>/dev/null || return 1 + command="$(ps -p "$pid" -o command= 2>/dev/null || true)" + case "$command" in + *"$2"*) printf '%s' "$pid" ;; + *) return 1 ;; + esac +} + +gateway_pid() { + live_pid "gateway-$1" "app.gateway.app:app" +} + +nginx_pid() { + live_pid nginx "nginx" +} + +gateway_port() { + case "$1" in + a) printf '%s' "$GATEWAY_A_PORT" ;; + b) printf '%s' "$GATEWAY_B_PORT" ;; + *) die "Unknown Gateway '$1' (expected a or b)" ;; + esac +} + +# Resolve the backend interpreter once (in this shell, not a subshell) and run +# it directly: `uv run` would sit between us and uvicorn, so the recorded pid +# would not be the process that serves (SIGKILL would orphan the server). +ensure_python() { + if [ -z "${PYTHON_BIN:-}" ]; then + PYTHON_BIN="$(uv run --no-sync --project "$BACKEND_DIR" python -c 'import sys; print(sys.executable)')" || + die "Cannot run the backend environment. Run 'make install' (or 'cd backend && uv sync') first." + fi +} + +run_helper() { + ensure_python + "$PYTHON_BIN" "$HELPER" "$@" +} + +load_secrets() { + [ -f "$SECRETS_FILE" ] || die "Missing $SECRETS_FILE; run 'up' first." + set -a + # shellcheck disable=SC1090 + . "$SECRETS_FILE" + set +a +} + +# ── State directory ────────────────────────────────────────────────────────── + +ensure_state_dir() { + if [ -d "$STATE_DIR" ] && [ ! -f "$STATE_DIR/$STATE_MARKER" ] && [ -n "$(ls -A "$STATE_DIR" 2>/dev/null)" ]; then + die "$STATE_DIR is not empty and was not created by this harness; set DEERFLOW_MI_STATE_DIR to another directory." + fi + mkdir -p "$STATE_DIR" "$RUN_DIR" "$LOG_DIR" + touch "$STATE_DIR/$STATE_MARKER" +} + +# Delete everything the harness created except logs. Guarded by the marker so +# a mistyped DEERFLOW_MI_STATE_DIR can never point this at someone's data. +wipe_runtime_data() { + [ -f "$STATE_DIR/$STATE_MARKER" ] || return 0 + find "$STATE_DIR" -mindepth 1 -maxdepth 1 ! -name logs ! -name "$STATE_MARKER" -exec rm -rf {} + +} + +# ── Containers ─────────────────────────────────────────────────────────────── + +require_docker() { + command -v docker >/dev/null 2>&1 || die "docker is not installed." + docker info >/dev/null 2>&1 || die "Docker is not running. Start Docker Desktop (or the daemon) and retry." +} + +container_exists() { + docker inspect "$1" >/dev/null 2>&1 +} + +container_label() { + docker inspect -f "{{ index .Config.Labels \"$2\" }}" "$1" 2>/dev/null || true +} + +# Remove container $1, with its anonymous volumes (both images declare a +# VOLUME), when this state dir created it. A container from another harness +# state dir, or from outside the harness, is never touched: returns 1. +remove_container() { + local owner + container_exists "$1" || return 0 + if [ "$(container_label "$1" "$LABEL_KEY")" != "$LABEL_VALUE" ]; then + echo " ! container $1 was not created by this harness; leaving it alone" >&2 + return 1 + fi + owner="$(container_label "$1" "$STATE_LABEL_KEY")" + if [ "$owner" != "$STATE_DIR" ]; then + echo " ! container $1 belongs to the harness state dir ${owner:-}; leaving it alone" >&2 + echo " (take that harness down from there, or 'docker rm -f -v $1' if it is stale)" >&2 + return 1 + fi + docker rm -f -v "$1" >/dev/null + echo "✓ Removed container $1 and its volumes" +} + +wait_until() { + local timeout=$1 waited=0 + shift + until "$@" >/dev/null 2>&1; do + [ "$waited" -lt "$timeout" ] || return 1 + sleep 1 + waited=$((waited + 1)) + done +} + +redis_pong() { + [ "$(docker exec "$REDIS_CONTAINER" redis-cli ping 2>/dev/null)" = "PONG" ] +} + +start_containers() { + echo "Starting Postgres ($PG_IMAGE) on 127.0.0.1:$PG_PORT..." + POSTGRES_PASSWORD="$DEERFLOW_MI_POSTGRES_PASSWORD" docker run -d --name "$PG_CONTAINER" \ + --label "$LABEL_KEY=$LABEL_VALUE" --label "$STATE_LABEL_KEY=$STATE_DIR" \ + -e POSTGRES_USER=deerflow -e POSTGRES_DB=deerflow -e POSTGRES_PASSWORD \ + -p "127.0.0.1:$PG_PORT:5432" "$PG_IMAGE" >/dev/null || return 1 + echo "Starting Redis ($REDIS_IMAGE) on 127.0.0.1:$REDIS_PORT..." + docker run -d --name "$REDIS_CONTAINER" --label "$LABEL_KEY=$LABEL_VALUE" --label "$STATE_LABEL_KEY=$STATE_DIR" \ + -p "127.0.0.1:$REDIS_PORT:6379" "$REDIS_IMAGE" >/dev/null || return 1 + + # TCP readiness: the image's init-time server listens on the Unix socket only. + if ! wait_until 90 docker exec "$PG_CONTAINER" pg_isready -h 127.0.0.1 -U deerflow -d deerflow; then + docker logs --tail 30 "$PG_CONTAINER" >&2 || true + echo "✗ Postgres did not become ready" >&2 + return 1 + fi + echo "✓ Postgres ready" + if ! wait_until 30 redis_pong; then + docker logs --tail 30 "$REDIS_CONTAINER" >&2 || true + echo "✗ Redis did not answer PING" >&2 + return 1 + fi + echo "✓ Redis ready" +} + +# ── Gateways and nginx ─────────────────────────────────────────────────────── + +start_gateway() { + local name=$1 port log pid waited=0 + port="$(gateway_port "$name")" + log="$LOG_DIR/gateway-$name.log" + if pid="$(gateway_pid "$name")"; then + echo "✗ Gateway $(upper "$name") is already running (pid $pid)." >&2 + return 1 + fi + if port_in_use "$port"; then + echo "✗ Port $port is in use; free it or set DEERFLOW_MI_GATEWAY_$(upper "$name")_PORT." >&2 + return 1 + fi + if [ ! -f "$CONFIG_OUT" ]; then + echo "✗ Missing $CONFIG_OUT; run 'up' first." >&2 + return 1 + fi + load_secrets + mkdir -p "$STATE_DIR/index/$name" "$STATE_DIR/home" + ensure_python + + echo "Starting Gateway $(upper "$name") on 127.0.0.1:$port..." + { + echo "" + echo "===== $(date '+%Y-%m-%d %H:%M:%S') starting Gateway $(upper "$name") on 127.0.0.1:$port =====" + } >>"$log" + ( + cd "$BACKEND_DIR" + # One worker per Gateway: the pid file must name the process that serves. + export DEER_FLOW_CONFIG_PATH="$CONFIG_OUT" \ + DEER_FLOW_EXTENSIONS_CONFIG_PATH="$EXTENSIONS_OUT" \ + DEER_FLOW_HOME="$STATE_DIR/home" \ + DEER_FLOW_PROJECT_ROOT="$REPO_ROOT" \ + DEERFLOW_MI_RETRIEVAL_INDEX_PATH="$STATE_DIR/index/$name" \ + DEERFLOW_MI_DATABASE_URL="postgresql://deerflow:$DEERFLOW_MI_POSTGRES_PASSWORD@127.0.0.1:$PG_PORT/deerflow" \ + DEERFLOW_MI_REDIS_URL="redis://127.0.0.1:$REDIS_PORT/0" \ + DEER_FLOW_CHANNELS_LANGGRAPH_URL="http://127.0.0.1:$port/api" \ + DEER_FLOW_CHANNELS_GATEWAY_URL="http://127.0.0.1:$port" \ + GATEWAY_WORKERS=1 WEB_CONCURRENCY=1 PYTHONPATH=. \ + AUTH_JWT_SECRET DEER_FLOW_INTERNAL_AUTH_TOKEN DEER_FLOW_CREDENTIALS_KEY + exec nohup "$PYTHON_BIN" -m uvicorn app.gateway.app:app --host 127.0.0.1 --port "$port" >"$log" 2>&1 & + pid=$! + # The Gateway outlives this script: drop it from the job table so stopping + # it later (teardown, stop/restart) prints no "Terminated" job notice. + disown "$pid" 2>/dev/null || true + echo "$pid" >"$(pid_file "gateway-$name")" + + while ! curl -fsS --noproxy '*' -o /dev/null --max-time 2 "http://127.0.0.1:$port/health" 2>/dev/null; do + if ! kill -0 "$pid" 2>/dev/null; then + echo "✗ Gateway $(upper "$name") exited during startup. Last lines of $log:" >&2 + tail -n 40 "$log" >&2 + return 1 + fi + if [ "$waited" -ge "$HEALTH_TIMEOUT" ]; then + echo "✗ Gateway $(upper "$name") did not answer /health within ${HEALTH_TIMEOUT}s. Last lines of $log:" >&2 + tail -n 40 "$log" >&2 + return 1 + fi + sleep 1 + waited=$((waited + 1)) + done + echo "✓ Gateway $(upper "$name") ready on http://127.0.0.1:$port (pid $pid, ${waited}s)" +} + +# stop_process NAME CMD_FRAGMENT SIGNAL +stop_process() { + local name=$1 fragment=$2 signal=${3:-TERM} pid waited=0 + if pid="$(live_pid "$name" "$fragment")"; then + kill "-$signal" "$pid" 2>/dev/null || true + while kill -0 "$pid" 2>/dev/null; do + if [ "$waited" -ge 30 ]; then + echo " ! $name (pid $pid) ignored SIG$signal for 30s; sending SIGKILL" >&2 + kill -KILL "$pid" 2>/dev/null || true + break + fi + sleep 1 + waited=$((waited + 1)) + done + echo "✓ Stopped $name (pid $pid, SIG$signal)" + fi + rm -f "$(pid_file "$name")" +} + +start_nginx() { + local conf="$STATE_DIR/nginx/nginx.conf" waited=0 + if ! command -v nginx >/dev/null 2>&1; then + echo " nginx not found; skipping the round-robin front (Gateways are still reachable directly)" + return 0 + fi + if port_in_use "$NGINX_PORT"; then + echo " ! port $NGINX_PORT is in use; skipping nginx (set DEERFLOW_MI_NGINX_PORT)" >&2 + return 0 + fi + mkdir -p "$STATE_DIR/nginx/temp" + run_helper nginx-conf --out "$conf" --state-dir "$STATE_DIR" --listen-port "$NGINX_PORT" \ + --gateway-port "$GATEWAY_A_PORT" --gateway-port "$GATEWAY_B_PORT" --frontend-port "$FRONTEND_PORT" || return 1 + echo "Starting nginx on 127.0.0.1:$NGINX_PORT..." + if ! nginx -e "$LOG_DIR/nginx-error.log" -p "$STATE_DIR/nginx" -c "$conf" >>"$LOG_DIR/nginx.log" 2>&1; then + tail -n 20 "$LOG_DIR/nginx.log" "$LOG_DIR/nginx-error.log" >&2 2>/dev/null || true + echo "✗ nginx failed to start" >&2 + return 1 + fi + until nginx_pid >/dev/null; do + [ "$waited" -lt 10 ] || { echo "✗ nginx wrote no pid file" >&2; return 1; } + sleep 1 + waited=$((waited + 1)) + done + echo "✓ nginx ready on http://localhost:$NGINX_PORT (pid $(nginx_pid))" +} + +# Stop processes and remove containers; keep config and logs for inspection. +teardown() { + stop_process nginx "nginx" QUIT + stop_process gateway-b "app.gateway.app:app" TERM + stop_process gateway-a "app.gateway.app:app" TERM + if command -v docker >/dev/null 2>&1 && docker info >/dev/null 2>&1; then + remove_container "$PG_CONTAINER" || true + remove_container "$REDIS_CONTAINER" || true + else + echo " ! Docker is not reachable; containers $PG_CONTAINER / $REDIS_CONTAINER were not removed" >&2 + fi +} + +fail_up() { + echo "✗ $*" >&2 + exit 1 +} + +# EXIT trap while `up` starts things: any exit before the banner (fail_up, die, +# set -e, Ctrl+C) stops what was started instead of leaving it running. +UP_IN_PROGRESS=false +on_up_exit() { + local status=$? + trap - EXIT INT TERM + if $UP_IN_PROGRESS; then + echo " Tearing down the partial start; logs stay in $LOG_DIR" >&2 + teardown + [ "$status" -ne 0 ] || status=1 + fi + exit "$status" +} + +# ── Inputs ─────────────────────────────────────────────────────────────────── + +absolute_path() { + printf '%s/%s' "$(builtin cd "$(dirname "$1")" >/dev/null 2>&1 && pwd -P)" "$(basename "$1")" +} + +resolve_base_config() { + local candidate + if [ -n "${DEERFLOW_MI_BASE_CONFIG:-}" ]; then + [ -f "$DEERFLOW_MI_BASE_CONFIG" ] || die "DEERFLOW_MI_BASE_CONFIG=$DEERFLOW_MI_BASE_CONFIG does not exist." + BASE_CONFIG="$(absolute_path "$DEERFLOW_MI_BASE_CONFIG")" + return 0 + fi + for candidate in "${DEER_FLOW_CONFIG_PATH:-}" "$REPO_ROOT/config.yaml" "$REPO_ROOT/backend/config.yaml" "$REPO_ROOT/config.example.yaml"; do + [ -n "$candidate" ] && [ -f "$candidate" ] || continue + candidate="$(absolute_path "$candidate")" + [ "$candidate" != "$CONFIG_OUT" ] || continue + BASE_CONFIG="$candidate" + return 0 + done + die "No base config found (config.yaml or config.example.yaml)." +} + +copy_extensions_config() { + local candidate + for candidate in "${DEER_FLOW_EXTENSIONS_CONFIG_PATH:-}" "$REPO_ROOT/extensions_config.json" "$REPO_ROOT/backend/extensions_config.json"; do + [ -n "$candidate" ] && [ -f "$candidate" ] && [ "$candidate" != "$EXTENSIONS_OUT" ] || continue + cp "$candidate" "$EXTENSIONS_OUT" + chmod 600 "$EXTENSIONS_OUT" + echo " extensions: copied $candidate (Gateway MCP/skill edits go to the copy)" + return 0 + done + printf '{\n "mcpServers": {},\n "skills": {}\n}\n' >"$EXTENSIONS_OUT" + echo " extensions: none found; using an empty extensions_config.json" +} + +# Extras the base config needs (models, channels, ...) plus the two this +# topology always needs. The detector only emits validated extra names. +uv_extra_flags() { + local detected="" token previous="" flags="--extra postgres --extra redis" + if command -v python3 >/dev/null 2>&1; then + detected="$(DEER_FLOW_CONFIG_PATH="$BASE_CONFIG" python3 "$REPO_ROOT/scripts/detect_uv_extras.py" 2>/dev/null || true)" + fi + for token in $detected; do + if [ "$previous" = "--extra" ] && [ "$token" != postgres ] && [ "$token" != redis ]; then + flags="$flags --extra $token" + fi + previous="$token" + done + printf '%s' "$flags" +} + +# ── Commands ───────────────────────────────────────────────────────────────── + +cmd_up() { + local skip_install=false with_nginx=true arg port extras + for arg in "$@"; do + case "$arg" in + --skip-install) skip_install=true ;; + --no-nginx) with_nginx=false ;; + *) die "Unknown option for up: $arg" ;; + esac + done + + require_docker + command -v uv >/dev/null 2>&1 || die "uv is not installed." + command -v curl >/dev/null 2>&1 || die "curl is not installed." + command -v lsof >/dev/null 2>&1 || die "lsof is not installed." + if gateway_pid a >/dev/null || gateway_pid b >/dev/null || nginx_pid >/dev/null; then + die "The harness is already running; run 'down' first (or 'status')." + fi + + ensure_state_dir + resolve_base_config + + # A previous session of this state dir that was not taken down: clear it + # before the port checks. Containers of any other owner are refused. + remove_container "$PG_CONTAINER" || die "Container name $PG_CONTAINER is taken; set DEERFLOW_MI_POSTGRES_CONTAINER (and the ports) to run a second harness." + remove_container "$REDIS_CONTAINER" || die "Container name $REDIS_CONTAINER is taken; set DEERFLOW_MI_REDIS_CONTAINER (and the ports) to run a second harness." + for port in "$GATEWAY_A_PORT" "$GATEWAY_B_PORT" "$PG_PORT" "$REDIS_PORT"; do + if port_in_use "$port"; then + die "Port $port is in use. Stop whatever holds it ('make stop' for make dev) or override the port (see --help)." + fi + done + + wipe_runtime_data + mkdir -p "$RUN_DIR" "$LOG_DIR" "$STATE_DIR/home" "$STATE_DIR/index" + : >"$LOG_DIR/gateway-a.log" + : >"$LOG_DIR/gateway-b.log" + rm -f "$LOG_DIR"/nginx*.log + + echo "Base config: $BASE_CONFIG" + if ! $skip_install; then + extras="$(uv_extra_flags)" + echo "Syncing backend dependencies ($extras)..." + # Intentionally unquoted: splats validated `--extra NAME` pairs. + # shellcheck disable=SC2086 + (cd "$BACKEND_DIR" && uv sync --locked --quiet --all-packages $extras) || die "Backend dependency sync failed." + fi + + run_helper secrets --out "$SECRETS_FILE" || die "Could not create $SECRETS_FILE." + load_secrets + echo "Rendering $CONFIG_OUT..." + local keep_channels="" + if [ "${DEERFLOW_MI_KEEP_CHANNELS:-}" = "1" ]; then + keep_channels="--keep-channels" + fi + # The URLs stay $VAR references so the file holds no harness secret; each + # Gateway process exports them (see start_gateway). + # shellcheck disable=SC2016 + run_helper render-config --base "$BASE_CONFIG" --out "$CONFIG_OUT" --state-dir "$STATE_DIR" \ + --database-url '$DEERFLOW_MI_DATABASE_URL' --redis-url '$DEERFLOW_MI_REDIS_URL' $keep_channels || + die "Could not render the multi-instance config." + copy_extensions_config + + UP_IN_PROGRESS=true + trap on_up_exit EXIT + trap 'echo ""; fail_up "Interrupted"' INT TERM + start_containers || fail_up "Containers failed to start." + # Sequential cold start: Alembic migrations are advisory-locked, but the + # LangGraph checkpointer/store setup() is not, so B starts after A is up. + start_gateway a || fail_up "Gateway A failed to start." + start_gateway b || fail_up "Gateway B failed to start." + if $with_nginx; then + start_nginx || fail_up "nginx failed to start." + fi + UP_IN_PROGRESS=false + trap - EXIT INT TERM + + echo "" + echo "==========================================" + echo " ✓ Multi-instance harness is up" + echo "==========================================" + echo " Gateway A http://127.0.0.1:$GATEWAY_A_PORT (log $LOG_DIR/gateway-a.log)" + echo " Gateway B http://127.0.0.1:$GATEWAY_B_PORT (log $LOG_DIR/gateway-b.log)" + if nginx_pid >/dev/null; then + echo " nginx http://localhost:$NGINX_PORT (round-robin; start the frontend on :$FRONTEND_PORT for the UI)" + fi + echo " Postgres 127.0.0.1:$PG_PORT ($PG_CONTAINER, user/db deerflow, password in $SECRETS_FILE)" + echo " Redis 127.0.0.1:$REDIS_PORT ($REDIS_CONTAINER)" + echo " Shared home $STATE_DIR/home" + echo " Config $CONFIG_OUT" + echo "" + echo " Next: scripts/dev_multi_instance.sh check" + echo " scripts/dev_multi_instance.sh stop b --kill (crash B), then: start b" + echo " scripts/dev_multi_instance.sh down" +} + +cmd_down() { + [ $# -eq 0 ] || die "down takes no options" + teardown + wipe_runtime_data + echo "✓ Harness is down (logs kept in $LOG_DIR)" +} + +cmd_status() { + local name pid port code container + echo "State dir: $STATE_DIR" + if command -v docker >/dev/null 2>&1 && docker info >/dev/null 2>&1; then + for container in "$PG_CONTAINER" "$REDIS_CONTAINER"; do + if container_exists "$container"; then + echo " $container: $(docker inspect -f '{{.State.Status}}' "$container")" + else + echo " $container: absent" + fi + done + else + echo " containers: Docker is not reachable" + fi + for name in a b; do + port="$(gateway_port "$name")" + if pid="$(gateway_pid "$name")"; then + code="$(curl -s --noproxy '*' -o /dev/null -w '%{http_code}' --max-time 3 "http://127.0.0.1:$port/health/ready" || true)" + echo " Gateway $(upper "$name"): running (pid $pid) http://127.0.0.1:$port, /health/ready $code" + else + echo " Gateway $(upper "$name"): stopped" + fi + done + if pid="$(nginx_pid)"; then + echo " nginx: running (pid $pid) http://localhost:$NGINX_PORT" + else + echo " nginx: stopped" + fi +} + +cmd_logs() { + local follow=false arg files="" + for arg in "$@"; do + case "$arg" in + a | b) files="$files gateway-$arg.log" ;; + nginx) files="$files nginx-error.log nginx-access.log" ;; + all) files="$files gateway-a.log gateway-b.log" ;; + -f | --follow) follow=true ;; + *) die "Unknown option for logs: $arg" ;; + esac + done + [ -n "$files" ] || files="gateway-a.log gateway-b.log" + [ -d "$LOG_DIR" ] || die "No logs yet in $LOG_DIR." + cd "$LOG_DIR" + if $follow; then + # shellcheck disable=SC2086 + exec tail -n 100 -F $files + fi + for arg in $files; do + if [ -f "$arg" ]; then + echo "==> $LOG_DIR/$arg <==" + tail -n 100 "$arg" + fi + done +} + +cmd_check() { + local arg nginx_url="" + for arg in "$@"; do + case "$arg" in + --with-llm) ;; + *) die "Unknown option for check: $arg" ;; + esac + done + if ! gateway_pid a >/dev/null || ! gateway_pid b >/dev/null; then + die "Both Gateways must be running; run 'up' (or 'start a|b')." + fi + if nginx_pid >/dev/null; then + nginx_url="http://127.0.0.1:$NGINX_PORT" + fi + run_helper check --gateway "A=http://127.0.0.1:$GATEWAY_A_PORT" --gateway "B=http://127.0.0.1:$GATEWAY_B_PORT" \ + ${nginx_url:+--nginx "$nginx_url"} --state-dir "$STATE_DIR" "$@" +} + +cmd_stop() { + local name="${1:-}" signal=TERM + [ -n "$name" ] || die "Usage: stop [--kill]" + gateway_port "$name" >/dev/null + shift + case "${1:-}" in + "") ;; + --kill) signal=KILL ;; + *) die "Unknown option for stop: $1" ;; + esac + gateway_pid "$name" >/dev/null || die "Gateway $(upper "$name") is not running." + stop_process "gateway-$name" "app.gateway.app:app" "$signal" +} + +cmd_start() { + local name="${1:-}" + [ -n "$name" ] && [ $# -eq 1 ] || die "Usage: start " + gateway_port "$name" >/dev/null + start_gateway "$name" +} + +cmd_restart() { + local name="${1:-}" + [ -n "$name" ] && [ $# -eq 1 ] || die "Usage: restart " + gateway_port "$name" >/dev/null + stop_process "gateway-$name" "app.gateway.app:app" TERM + start_gateway "$name" +} + +main() { + local command="${1:-help}" + [ $# -eq 0 ] || shift + case "$command" in + up) cmd_up "$@" ;; + down) cmd_down "$@" ;; + status) cmd_status ;; + logs) cmd_logs "$@" ;; + check) cmd_check "$@" ;; + stop) cmd_stop "$@" ;; + start) cmd_start "$@" ;; + restart) cmd_restart "$@" ;; + help | -h | --help) usage ;; + *) + echo "Unknown command: $command" >&2 + usage >&2 + exit 2 + ;; + esac +} + +main "$@"