mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
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 <noreply@anthropic.com> * docs(changelog): link the two-Gateway harness entries to #6613 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * docs(scripts): keep the harness guidance under the AGENTS.md soft limit Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Nan Gao
GitHub
parent
7e0e3013d0
commit
35f6a98cd9
19
CHANGELOG.md
19
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
|
||||
|
||||
@ -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
|
||||
|
||||
15
Makefile
15
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:
|
||||
|
||||
@ -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.
|
||||
>
|
||||
|
||||
@ -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`.
|
||||
|
||||
705
backend/tests/test_dev_multi_instance_script.py
Normal file
705
backend/tests/test_dev_multi_instance_script.py
Normal file
@ -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-<port> 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
|
||||
@ -167,6 +167,12 @@ defaults `DEER_FLOW_PROJECT_ROOT` to the checkout, as `serve.sh` does, so
|
||||
Only "no config anywhere" creates `<checkout>/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
|
||||
|
||||
783
scripts/dev_multi_instance.py
Normal file
783
scripts/dev_multi_instance.py
Normal file
@ -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())
|
||||
675
scripts/dev_multi_instance.sh
Executable file
675
scripts/dev_multi_instance.sh
Executable file
@ -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 <<EOF
|
||||
Usage: scripts/dev_multi_instance.sh <command> [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 <a|b> [--kill] Stop one Gateway (SIGTERM; --kill sends SIGKILL to simulate a crash)
|
||||
start <a|b> Start a stopped Gateway again
|
||||
restart <a|b> 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:-<unrecorded>}; 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" </dev/null
|
||||
) >>"$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 <a|b> [--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 <a|b>"
|
||||
gateway_port "$name" >/dev/null
|
||||
start_gateway "$name"
|
||||
}
|
||||
|
||||
cmd_restart() {
|
||||
local name="${1:-}"
|
||||
[ -n "$name" ] && [ $# -eq 1 ] || die "Usage: restart <a|b>"
|
||||
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 "$@"
|
||||
Loading…
x
Reference in New Issue
Block a user