Merge branch 'master-dev' into 'master'

offline-backend-20260512/.gitattributes

@@ -0,0 +1,3 @@
+*.sh text eol=lf
+offline-runtime-requirements.txt text eol=lf
+wheelhouse/SHA256SUMS text eol=lf

offline-backend-20260512/README.md

@@ -0,0 +1,66 @@
+# DeerFlow 离线后端部署补充
+
+本目录提供与原始 `start_backend.sh` 配套的 wheelhouse。启动脚本保持原有结构，只新增一行 Redis Python 客户端的离线安装命令。
+
+## 文件
+
+- `offline-runtime-requirements.txt`：固定版本的完整依赖闭包。
+- `prepare_wheelhouse.sh`：联网机器下载 Linux/Python 3.12 wheel，并验证传递依赖完整性。
+- `start_backend.sh`：原始 Docker 启动脚本，加一行 Redis 缺包安装命令。
+- `wheelhouse/`：22 个 wheel 和 `SHA256SUMS`，需要随发布目录一起复制。
+
+## 联网机器准备
+
+默认目标是后端镜像使用的 CPython 3.12、Linux x86_64：
+
+```bash
+cd offline-backend-20260512
+bash prepare_wheelhouse.sh
+```
+
+直接依赖为：
+
+- `croniter==6.2.2`
+- `hindsight-client==0.6.2`
+- `redis==5.2.1`
+
+上述依赖在 Python 3.12/Linux x86_64 下的 19 个传递依赖也已锁定在 `offline-runtime-requirements.txt`。
+
+国内镜像示例：
+
+```bash
+PIP_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple bash prepare_wheelhouse.sh
+```
+
+ARM64 镜像需要改目标平台：
+
+```bash
+TARGET_PLATFORM=manylinux2014_aarch64 bash prepare_wheelhouse.sh
+```
+
+## 离线机器启动
+
+保持原脚本要求的发布目录结构，包括 Docker 镜像 tar、`.env`、`config.yaml`、`extensions_config.json`、`skill_evolution.json`、`skills/`、`wheelhouse/`、`backend/uv.lock` 和 `backend/`。
+
+可先校验 wheel：
+
+```bash
+cd offline-backend-20260512/wheelhouse
+sha256sum -c SHA256SUMS
+cd ..
+```
+
+然后启动：
+
+```bash
+bash start_backend.sh
+docker logs -f deerflow-backend
+```
+
+原脚本已有 `croniter` 和 `hindsight-client` 的离线补装逻辑；新增的一行命令只负责在缺少 `redis` 模块时，从 `/app/wheelhouse` 安装 `redis==5.2.1`。
+
+## Redis 说明
+
+`redis==5.2.1` 只是 Python 客户端，不包含 Redis 服务端。`config.yaml` 中 `concurrency.redis.enabled: false` 时无需部署 Redis；启用后必须另行准备 Redis 服务或离线 Redis Docker 镜像。
+
+如果 Redis 运行在另一个容器，`redis://127.0.0.1:6379/0` 指向后端容器自身，不能直接使用。应将两个容器加入同一个 Docker network，并将 URL 改为类似 `redis://redis:6379/0`，同时为不同环境配置不同的 `namespace`。

offline-backend-20260512/backend/CLAUDE.md

@@ -998,6 +998,7 @@ Both the `view_image` tool and `ViewImageMiddleware` are registered whenever `vi
 ## Documentation
 
 See `docs/` directory for detailed documentation:
+- [../README.md](../README.md) - Offline Docker wheelhouse preparation and startup workflow. The pinned versions in `../offline-runtime-requirements.txt` and `../start_backend.sh` must stay synchronized. `start_backend.sh` stays aligned with the deployment script and adds only the Redis wheel fallback line. The download target defaults to CPython 3.12 on Linux x86_64 and verifies dependency closure using wheelhouse only.
 - [CONFIGURATION.md](docs/CONFIGURATION.md) - Configuration options
 - [ARCHITECTURE.md](docs/ARCHITECTURE.md) - Architecture details
 - [API.md](docs/API.md) - API reference

offline-backend-20260512/backend/README.md

@@ -411,6 +411,7 @@ uv run pytest
 - [Configuration Guide](docs/CONFIGURATION.md)
 - [Architecture Details](docs/ARCHITECTURE.md)
 - [API Reference](docs/API.md)
+- [Offline Docker wheelhouse deployment (ZH)](../README.md)
 - [File Upload](docs/FILE_UPLOAD.md)
 - [Path Examples](docs/PATH_EXAMPLES.md)
 - [Context Summarization](docs/summarization.md)

offline-backend-20260512/backend/app/gateway/app.py

@@ -853,8 +853,9 @@ async def _concurrency_sampler_loop(app: FastAPI) -> None:
         logger.info("Concurrency sampler disabled by CONCURRENCY_SAMPLE_ENABLED")
         return
 
-    store = getattr(app.state, "concurrency_sample_store", None)
-    run_mgr = getattr(app.state, "run_manager", None)
+    store = getattr(app.state, "concurrency_sample_store", None)
+    run_mgr = getattr(app.state, "run_manager", None)
+    concurrency_gate = getattr(app.state, "concurrency_gate", None)
     if store is None or run_mgr is None:
         logger.info("Concurrency sampler not started (no store / run manager — memory backend?)")
         return
@@ -875,11 +876,16 @@ async def _concurrency_sampler_loop(app: FastAPI) -> None:
             try:
                 # Runtime toggle (admin can turn sampling off without a restart).
                 # mtime-cached read, so this is cheap every tick.
-                from deerflow.config.system_settings import get_concurrency_monitor_settings
-
-                if get_concurrency_monitor_settings().enabled:
-                    active = await run_mgr.list_active()
-                    await store.record(len(active))
+                from deerflow.config.system_settings import get_concurrency_monitor_settings
+
+                if get_concurrency_monitor_settings().enabled:
+                    active_count = None
+                    if concurrency_gate is not None:
+                        active_count = await concurrency_gate.count_active()
+                    if active_count is None:
+                        active = await run_mgr.list_active()
+                        active_count = len(active)
+                    await store.record(active_count)
                     ticks += 1
                     if ticks % purge_every_ticks == 0:
                         from datetime import UTC, datetime, timedelta

offline-backend-20260512/backend/app/gateway/deps.py

@@ -367,8 +367,11 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
         run_events_config = getattr(config, "run_events", None)
         app.state.run_event_store = make_run_event_store(run_events_config)
 
-        # RunManager with store backing for persistence
-        app.state.run_manager = RunManager(store=app.state.run_store)
+        # RunManager with store backing for persistence
+        app.state.run_manager = RunManager(store=app.state.run_store)
+        from deerflow.runtime import make_redis_concurrency_gate
+
+        app.state.concurrency_gate = make_redis_concurrency_gate(config)
 
         try:
             yield
@@ -382,7 +385,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
                 await queue.stop()
                 set_auto_ingest_queue(None)
-            await close_engine()
+            await close_engine()
+            gate = getattr(app.state, "concurrency_gate", None)
+            if gate is not None:
+                await gate.aclose()
 
 
 # ---------------------------------------------------------------------------
@@ -408,7 +414,12 @@ get_run_manager: Callable[[Request], RunManager] = _require("run_manager", "Run
 get_checkpointer: Callable[[Request], Checkpointer] = _require("checkpointer", "Checkpointer")
 get_run_event_store: Callable[[Request], RunEventStore] = _require("run_event_store", "Run event store")
 get_feedback_repo: Callable[[Request], FeedbackRepository] = _require("feedback_repo", "Feedback")
-get_run_store: Callable[[Request], RunStore] = _require("run_store", "Run store")
+get_run_store: Callable[[Request], RunStore] = _require("run_store", "Run store")
+
+
+def get_concurrency_gate(request: Request):
+    """Return the optional cross-worker concurrency gate."""
+    return getattr(request.app.state, "concurrency_gate", None)
 
 
 def get_store(request: Request):

offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py

@@ -32,6 +32,7 @@ from app.gateway.roundtable_model_fallback import (
 )
 from app.gateway.roundtable_run_policy import roundtable_run_policy
 from app.gateway.roundtable_seat_skills import append_seat_skill_directive
+from app.gateway.services import RUN_SCOPE_BACKGROUND_CHILD
 from app.gateway.routers._roundtable_seed import (
     COORDINATOR_AGENT_ID,
     DASHBOARD_AGENT_ID,
@@ -346,6 +347,7 @@ async def _run_agent_turn(
     body = RunCreateRequest(
         assistant_id="lead_agent",
         input={"messages": [{"type": "human", "content": [{"type": "text", "text": message}], "additional_kwargs": {}}]},
+        metadata={"run_scope": RUN_SCOPE_BACKGROUND_CHILD},
         config={"recursion_limit": 1000},
         context={
             "agent_name": agent_name,
@@ -374,7 +376,12 @@ async def _run_agent_turn(
     # 落 checkpoint）。额度满时本轮在此 await 排队，等同用户某个在跑轮次释放额度后再开。
     async with _user_background_semaphore(user_id):
         try:
-            record = await start_run(body, thread_id, request)  # type: ignore[arg-type]
+            record = await start_run(  # type: ignore[arg-type]
+                body,
+                thread_id,
+                request,
+                run_scope=RUN_SCOPE_BACKGROUND_CHILD,
+            )
         except Exception as exc:  # noqa: BLE001 — start_run 报错（409 抢占 / DB 抖动等）也要落诊断
             if diag is not None:
                 await diag.record(

offline-backend-20260512/backend/app/gateway/routers/admin_active_runs.py

@@ -22,7 +22,7 @@ from fastapi import APIRouter, Query, Request
 from pydantic import BaseModel
 from sqlalchemy import select
 
-from app.gateway.deps import get_concurrency_sample_store, get_run_manager
+from app.gateway.deps import get_concurrency_gate, get_concurrency_sample_store, get_run_manager
 from app.gateway.routers.admin_users import _require_admin
 from deerflow.config.system_settings import load_system_settings, save_system_settings
 from deerflow.persistence.agents.model import AgentRow
@@ -149,6 +149,15 @@ def _elapsed_seconds(created_at: str, now: datetime) -> int:
     return max(0, int(delta))
 
 
+def _elapsed_seconds_from_epoch(created_at: float, now: datetime) -> int:
+    try:
+        if created_at <= 0:
+            return 0
+        return max(0, int(now.timestamp() - float(created_at)))
+    except Exception:
+        return 0
+
+
 def _kind_for(metadata: dict | None) -> str:
     """Coarse human label for the run's origin, from its thread metadata."""
     meta = metadata or {}
@@ -218,14 +227,46 @@ async def list_active_runs(request: Request) -> ActiveRunsResponse:
     """List every in-flight run with who/what is driving it (admin only)."""
     await _require_admin(request)
     run_mgr = get_run_manager(request)
-    records = await run_mgr.list_active()
+    gate = get_concurrency_gate(request)
+    leases = await gate.list_active() if gate is not None else None
+    records = [] if leases is not None else await run_mgr.list_active()
     now = datetime.now(UTC)
-    maps = await _enrich(records)
+    enrich_source = leases if leases is not None else records
+    maps = await _enrich(enrich_source)
     thread_meta = maps["thread_meta"]
     user_emails = maps["user_emails"]
     agent_names = maps["agent_names"]
 
     items: list[ActiveRunItem] = []
+    if leases is not None:
+        for r in leases:
+            meta_row = thread_meta.get(r.thread_id)
+            user_id = r.user_id or (meta_row.user_id if meta_row else "") or ""
+            metadata = (meta_row.metadata_json if meta_row else None) or {}
+            assistant_id = r.assistant_id or ""
+            if assistant_id in _DEFAULT_ASSISTANT_IDS:
+                agent_name = "榛樿鍔╂墜"
+            else:
+                agent_name = agent_names.get(assistant_id) or assistant_id
+            items.append(
+                ActiveRunItem(
+                    run_id=r.run_id,
+                    thread_id=r.thread_id,
+                    status="running",
+                    created_at=datetime.fromtimestamp(r.created_at, tz=UTC).isoformat() if r.created_at else "",
+                    elapsed_seconds=_elapsed_seconds_from_epoch(r.created_at, now),
+                    user_id=user_id,
+                    user_name=user_emails.get(user_id) or user_id or "鏈煡鐢ㄦ埛",
+                    agent_id=assistant_id,
+                    agent_name=agent_name,
+                    model_name=r.model_name,
+                    thread_title=(meta_row.display_name if meta_row else "") or "",
+                    kind=_kind_for(metadata),
+                    multitask_strategy="reject",
+                )
+            )
+        return ActiveRunsResponse(count=len(items), server_time=now.isoformat(), runs=items)
+
     for r in records:
         meta_row = thread_meta.get(r.thread_id)
         user_id = (meta_row.user_id if meta_row else "") or ""

offline-backend-20260512/backend/app/gateway/routers/multi_agent.py

@@ -65,6 +65,7 @@ from langgraph.checkpoint.base import empty_checkpoint
 from app.gateway.roundtable_diag import record_foreground_diag
 from app.gateway.roundtable_model_fallback import fallback_model_chain, is_llm_error_message, reason_text
 from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy
+from app.gateway.services import RUN_SCOPE_ORCHESTRATION_CHILD, internal_run_scope_headers
 from deerflow.agents.roundtable_orchestrator.extraction_spec import build_summary_directive
 
 from app.gateway.roundtable_seat_skills import append_seat_skill_directive
@@ -325,7 +326,11 @@ def _auth_headers(request: Request) -> dict[str, str]:
     token = request.headers.get("authorization")
     if not token:
         raise HTTPException(status_code=401, detail="Missing Authorization header")
-    return {"Authorization": token, "Content-Type": "application/json"}
+    return {
+        "Authorization": token,
+        "Content-Type": "application/json",
+        **internal_run_scope_headers(),
+    }
 
 
 async def _create_thread(
@@ -694,6 +699,7 @@ def _run_payload(
                 }
             ]
         },
+        "metadata": {"run_scope": RUN_SCOPE_ORCHESTRATION_CHILD},
         "config": {"recursion_limit": 1000},
         "context": {
             "agent_name": agent_name,

offline-backend-20260512/backend/app/gateway/routers/system_settings.py

@@ -24,7 +24,6 @@ from deerflow.config.system_settings import (
     PublishSquare,
     PublishSquaresSettings,
     SkillCompressionSettings,
-    SystemSettings,
     get_module_tag_ids,
     list_publish_squares,
     load_system_settings,
@@ -226,12 +225,16 @@ class ModelConcurrencyResponse(BaseModel):
     enabled: bool = Field(..., description="Whether per-model concurrency limiting is active")
     default_limit: int = Field(..., description="Fallback max concurrent runs for models without an explicit override")
     model_limits: dict[str, int] = Field(default_factory=dict, description="Per-model overrides: model_name -> max concurrent runs")
+    user_default_limit: int = Field(..., description="Fallback max concurrent runs per user and model. 0 disables the per-user fallback")
+    user_model_limits: dict[str, int] = Field(default_factory=dict, description="Per-user per-model overrides: model_name -> max concurrent runs for one user")
 
 
 class ModelConcurrencyUpdateRequest(BaseModel):
     enabled: bool | None = None
     default_limit: int | None = Field(default=None, ge=1, le=10_000)
     model_limits: dict[str, int] | None = None
+    user_default_limit: int | None = Field(default=None, ge=0, le=10_000)
+    user_model_limits: dict[str, int] | None = None
 
 
 def _skill_compression_to_response(settings: SkillCompressionSettings) -> SkillCompressionResponse:
@@ -255,10 +258,24 @@ def _model_concurrency_to_response(settings: ModelConcurrencySettings) -> ModelC
         if normalized_limit < 1:
             continue
         cleaned[model_name] = min(normalized_limit, 10_000)
+    cleaned_user: dict[str, int] = {}
+    for name, limit in (settings.user_model_limits or {}).items():
+        model_name = str(name).strip()
+        if not model_name:
+            continue
+        try:
+            normalized_limit = int(limit)
+        except (TypeError, ValueError):
+            continue
+        if normalized_limit < 1:
+            continue
+        cleaned_user[model_name] = min(normalized_limit, 10_000)
     return ModelConcurrencyResponse(
         enabled=settings.enabled,
         default_limit=max(1, min(int(settings.default_limit), 10_000)),
         model_limits=cleaned,
+        user_default_limit=max(0, min(int(getattr(settings, "user_default_limit", 0) or 0), 10_000)),
+        user_model_limits=cleaned_user,
     )
 
 
@@ -336,6 +353,22 @@ async def update_model_concurrency(
                 continue
             cleaned[model_name] = min(normalized_limit, 10_000)
         update["model_limits"] = cleaned
+    if "user_default_limit" in fields and fields["user_default_limit"] is not None:
+        update["user_default_limit"] = max(0, min(int(fields["user_default_limit"]), 10_000))
+    if "user_model_limits" in fields and fields["user_model_limits"] is not None:
+        cleaned_user: dict[str, int] = {}
+        for name, limit in (body.user_model_limits or {}).items():
+            model_name = str(name).strip()
+            if not model_name:
+                continue
+            try:
+                normalized_limit = int(limit)
+            except (TypeError, ValueError):
+                continue
+            if normalized_limit < 1:
+                continue
+            cleaned_user[model_name] = min(normalized_limit, 10_000)
+        update["user_model_limits"] = cleaned_user
     settings = current.model_concurrency.model_copy(update=update)
     saved = save_system_settings(current.model_copy(update={"model_concurrency": settings}))
     return _model_concurrency_to_response(saved.model_concurrency)

offline-backend-20260512/backend/app/gateway/services.py

@@ -10,15 +10,17 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+import os
 import re
 import time
+import uuid
 from collections.abc import Mapping
 from typing import Any
 
 from fastapi import HTTPException, Request
 from langchain_core.messages import HumanMessage
 
-from app.gateway.deps import get_run_context, get_run_manager, get_stream_bridge
+from app.gateway.deps import get_concurrency_gate, get_run_context, get_run_manager, get_stream_bridge
 from app.gateway.utils import sanitize_log_param
 from deerflow.config import get_app_config
 from deerflow.config.agents_config import load_agent_config, validate_agent_id
@@ -36,9 +38,105 @@ from deerflow.runtime import (
     UnsupportedStrategyError,
     run_agent,
 )
+from deerflow.runtime.user_context import get_effective_user_id
 
 logger = logging.getLogger(__name__)
 
+RUN_SCOPE_EXTERNAL_ROOT = "external_root"
+RUN_SCOPE_ORCHESTRATION_CHILD = "orchestration_child"
+RUN_SCOPE_BACKGROUND_CHILD = "background_child"
+_INTERNAL_RUN_SCOPES = {RUN_SCOPE_ORCHESTRATION_CHILD, RUN_SCOPE_BACKGROUND_CHILD}
+_ALL_RUN_SCOPES = {RUN_SCOPE_EXTERNAL_ROOT, *_INTERNAL_RUN_SCOPES}
+_INTERNAL_RUN_SCOPE_HEADER = "x-deerflow-internal-run-scope-token"
+_INTERNAL_RUN_SCOPE_TOKEN = (
+    os.getenv("DEERFLOW_INTERNAL_RUN_SCOPE_TOKEN", "").strip()
+    or uuid.uuid4().hex
+)
+
+
+def internal_run_scope_headers() -> dict[str, str]:
+    """Headers used by same-process loopback calls to mark internal child runs."""
+
+    return {_INTERNAL_RUN_SCOPE_HEADER: _INTERNAL_RUN_SCOPE_TOKEN}
+
+
+def _metadata_run_scope(body: Any) -> str | None:
+    metadata = getattr(body, "metadata", None) or {}
+    if not isinstance(metadata, dict):
+        return None
+    raw = str(metadata.get("run_scope") or "").strip()
+    return raw or None
+
+
+def _request_is_loopback(request: Request) -> bool:
+    client = getattr(request, "client", None)
+    host = str(getattr(client, "host", "") or "").strip()
+    return host in {"127.0.0.1", "::1", "localhost"}
+
+
+def _has_internal_run_scope_header(request: Request) -> bool:
+    headers = getattr(request, "headers", None)
+    token = headers.get(_INTERNAL_RUN_SCOPE_HEADER) if headers is not None else None
+    return str(token or "") == _INTERNAL_RUN_SCOPE_TOKEN
+
+
+def _resolve_run_scope(body: Any, request: Request, run_scope: str | None) -> str:
+    """Return the trusted run scope used for concurrency accounting."""
+
+    if run_scope:
+        scope = str(run_scope).strip()
+        return scope if scope in _ALL_RUN_SCOPES else RUN_SCOPE_EXTERNAL_ROOT
+
+    scope = _metadata_run_scope(body)
+    if (
+        scope in _INTERNAL_RUN_SCOPES
+        and _request_is_loopback(request)
+        and _has_internal_run_scope_header(request)
+    ):
+        return scope
+    if scope == RUN_SCOPE_EXTERNAL_ROOT:
+        return RUN_SCOPE_EXTERNAL_ROOT
+    return RUN_SCOPE_EXTERNAL_ROOT
+
+
+async def _run_agent_with_concurrency_lease(
+    *,
+    concurrency_gate: Any | None,
+    lease_acquired: bool,
+    lease_metadata: dict[str, Any],
+    heartbeat_interval_seconds: int,
+    run_agent_kwargs: dict[str, Any],
+) -> None:
+    """Run the agent while refreshing and finally releasing an optional lease."""
+
+    heartbeat_task: asyncio.Task | None = None
+
+    async def _heartbeat_loop() -> None:
+        if concurrency_gate is None or not lease_acquired:
+            return
+        interval = max(5, int(heartbeat_interval_seconds or 20))
+        while True:
+            await asyncio.sleep(interval)
+            await concurrency_gate.heartbeat(**lease_metadata)
+
+    if concurrency_gate is not None and lease_acquired:
+        heartbeat_task = asyncio.create_task(
+            _heartbeat_loop(),
+            name=f"concurrency-heartbeat:{lease_metadata.get('run_id')}",
+        )
+
+    try:
+        await run_agent(**run_agent_kwargs)
+    finally:
+        if heartbeat_task is not None:
+            heartbeat_task.cancel()
+            try:
+                await heartbeat_task
+            except asyncio.CancelledError:
+                pass
+        if concurrency_gate is not None and lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+
 
 # ---------------------------------------------------------------------------
 # SSE formatting
@@ -498,6 +596,49 @@ def _resolve_effective_run_model_name(body: Any) -> str:
     return default_model_name
 
 
+def _resolve_effective_model_name(body: Any, app_config: Any | None = None) -> str:
+    """Compatibility wrapper for resolving the effective run model."""
+    if app_config is None:
+        return _resolve_effective_run_model_name(body)
+
+    default_model_name = app_config.models[0].name if app_config.models else None
+    if not default_model_name:
+        raise ValueError("No chat models are configured. Please configure at least one model in config.yaml.")
+
+    requested_model_name = _extract_requested_model_name(body)
+    if requested_model_name and app_config.get_model_config(requested_model_name):
+        return requested_model_name
+
+    agent_id = _extract_effective_agent_id(body)
+    if agent_id:
+        try:
+            agent_config = load_agent_config(agent_id)
+        except Exception:
+            agent_config = None
+        agent_model_name = str(getattr(agent_config, "model", "") or "").strip()
+        if agent_model_name and app_config.get_model_config(agent_model_name):
+            return agent_model_name
+
+    return default_model_name
+
+
+def _resolve_model_concurrency_limit(model_name: str, app_config: Any) -> int | None:
+    """Read legacy per-model max-concurrency values from model config extras."""
+    model_config = app_config.get_model_config(model_name)
+    extras = getattr(model_config, "model_extra", None) or {}
+    for key in ("max_concurrency", "model_concurrency_limit"):
+        raw = extras.get(key)
+        if raw is None:
+            continue
+        try:
+            limit = int(raw)
+        except (TypeError, ValueError):
+            continue
+        if limit > 0:
+            return limit
+    return None
+
+
 def _get_model_concurrency_limit(model_name: str) -> int | None:
     """Return the configured limit for a model, or ``None`` when limiting is off."""
     settings = load_system_settings().model_concurrency
@@ -512,6 +653,25 @@ def _get_model_concurrency_limit(model_name: str) -> int | None:
     return max(1, limit)
 
 
+def _get_user_model_concurrency_limit(model_name: str) -> int | None:
+    """Return the configured per-user limit for a model, or ``None`` when off."""
+    settings = load_system_settings().model_concurrency
+    if not settings.enabled:
+        return None
+
+    raw_limit = settings.user_model_limits.get(
+        model_name,
+        getattr(settings, "user_default_limit", 0) or 0,
+    )
+    try:
+        limit = int(raw_limit)
+    except (TypeError, ValueError):
+        limit = int(getattr(settings, "user_default_limit", 0) or 0)
+    if limit < 1:
+        return None
+    return max(1, limit)
+
+
 async def _sync_legacy_agent_records(store) -> None:
     """Import existing folder-backed agents as built-in records.
 
@@ -647,6 +807,7 @@ async def start_run(
     request: Request,
     *,
     enforce_access: bool = True,
+    run_scope: str | None = None,
 ) -> RunRecord:
     """Create a RunRecord and launch the background agent task.
 
@@ -674,6 +835,7 @@ async def start_run(
     bridge = get_stream_bridge(request)
     run_mgr = get_run_manager(request)
     run_ctx = get_run_context(request)
+    concurrency_gate = get_concurrency_gate(request)
 
     disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_
 
@@ -688,10 +850,56 @@ async def start_run(
     t_post_enforce = time.perf_counter()
 
     effective_model_name = _resolve_effective_run_model_name(body)
+    effective_user_id = get_effective_user_id()
+    effective_run_scope = _resolve_run_scope(body, request, run_scope)
     model_concurrency_limit = _get_model_concurrency_limit(effective_model_name)
+    user_model_concurrency_limit = (
+        _get_user_model_concurrency_limit(effective_model_name)
+        if effective_run_scope == RUN_SCOPE_EXTERNAL_ROOT
+        else None
+    )
+    run_id = str(uuid.uuid4())
 
     run_metadata = dict(body.metadata or {})
     run_metadata["model_name"] = effective_model_name
+    run_metadata["user_id"] = effective_user_id
+    run_metadata["run_scope"] = effective_run_scope
+
+    redis_lease_acquired = False
+    redis_gate_degraded = True
+    user_model_accounted = user_model_concurrency_limit is not None
+    lease_metadata = {
+        "run_id": run_id,
+        "thread_id": thread_id,
+        "model_name": effective_model_name,
+        "user_id": effective_user_id,
+        "user_model_accounted": user_model_accounted,
+    }
+    if concurrency_gate is not None:
+        acquire = await concurrency_gate.try_acquire(
+            run_id=run_id,
+            thread_id=thread_id,
+            model_name=effective_model_name,
+            user_id=effective_user_id,
+            assistant_id=body.assistant_id or "",
+            model_limit=model_concurrency_limit,
+            user_model_limit=user_model_concurrency_limit,
+            reject_thread=body.multitask_strategy == "reject",
+        )
+        redis_lease_acquired = acquire.acquired
+        redis_gate_degraded = acquire.degraded
+        if acquire.rejected:
+            if acquire.reason == "thread":
+                raise HTTPException(
+                    status_code=409,
+                    detail="Current thread already has an active run. Please wait or stop it before sending another message.",
+                )
+            if acquire.reason == "user_model":
+                raise HTTPException(
+                    status_code=429,
+                    detail="This account has reached the concurrency limit for the selected model. Please try again later.",
+                )
+            raise HTTPException(status_code=429, detail="The selected model is busy. Please try again later.")
 
     try:
         record = await run_mgr.create_or_reject(
@@ -701,19 +909,35 @@ async def start_run(
             metadata=run_metadata,
             kwargs={"input": body.input, "config": body.config},
             multitask_strategy=body.multitask_strategy,
-            model_concurrency_limit=(effective_model_name, model_concurrency_limit) if model_concurrency_limit is not None else None,
+            model_concurrency_limit=(effective_model_name, model_concurrency_limit) if redis_gate_degraded and model_concurrency_limit is not None else None,
+            user_model_concurrency_limit=(effective_model_name, effective_user_id, user_model_concurrency_limit) if redis_gate_degraded and user_model_concurrency_limit is not None else None,
+            run_id=run_id,
         )
     except ModelConcurrencyLimitError as exc:
-        raise HTTPException(status_code=429, detail="当前使用人数较多，请稍等一下再试。") from exc
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+        if getattr(exc, "scope", "") == "user_model":
+            detail = "This account has reached the concurrency limit for the selected model. Please try again later."
+        else:
+            detail = "The selected model is busy. Please try again later."
+        raise HTTPException(status_code=429, detail=detail) from exc
     except ConflictError as exc:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
         # 同一会话上一轮回答还没结束，又提交了新问题 → 拒绝。给用户中文提示，
         # 而不是把内部英文「Thread ... already has an active run」直接抛到前端。
         raise HTTPException(
             status_code=409,
-            detail="当前对话还在生成回答，请等待本轮回复结束，或先点「停止」再发送新的问题。",
+            detail="Current thread already has an active run. Please wait or stop it before sending another message.",
         ) from exc
     except UnsupportedStrategyError as exc:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
         raise HTTPException(status_code=501, detail=str(exc)) from exc
+    except Exception:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+        raise
     t_post_create = time.perf_counter()
 
     # Upsert thread metadata so the thread appears in /threads/search,
@@ -751,7 +975,12 @@ async def start_run(
         logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
     t_post_thread_upsert = time.perf_counter()
 
-    agent_factory = resolve_agent_factory(body.assistant_id)
+    try:
+        agent_factory = resolve_agent_factory(body.assistant_id)
+    except Exception:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+        raise
     graph_input = normalize_input(body.input)
 
     # Admin-configurable prompt prefix: when the caller opted in via
@@ -789,24 +1018,38 @@ async def start_run(
     stream_modes = normalize_stream_modes(body.stream_mode)
 
     t_pre_task = time.perf_counter()
-    task = asyncio.create_task(
-        run_agent(
-            bridge,
-            run_mgr,
-            record,
-            ctx=run_ctx,
-            agent_factory=agent_factory,
-            graph_input=graph_input,
-            config=config,
-            stream_modes=stream_modes,
-            stream_subgraphs=body.stream_subgraphs,
-            interrupt_before=body.interrupt_before,
-            interrupt_after=body.interrupt_after,
-            # 透传 SDK ``Command``：resume / update / goto。Pregel 会据此从最近
-            # checkpoint 处恢复，而不是把 graph_input 当新一轮输入跑。
-            command=getattr(body, "command", None),
+    heartbeat_interval_seconds = 20
+    if concurrency_gate is not None:
+        heartbeat_interval_seconds = int(
+            getattr(getattr(concurrency_gate, "config", None), "heartbeat_interval_seconds", 20) or 20
         )
-    )
+    try:
+        task = asyncio.create_task(
+            _run_agent_with_concurrency_lease(
+                concurrency_gate=concurrency_gate,
+                lease_acquired=redis_lease_acquired,
+                lease_metadata=lease_metadata,
+                heartbeat_interval_seconds=heartbeat_interval_seconds,
+                run_agent_kwargs={
+                    "bridge": bridge,
+                    "run_manager": run_mgr,
+                    "record": record,
+                    "ctx": run_ctx,
+                    "agent_factory": agent_factory,
+                    "graph_input": graph_input,
+                    "config": config,
+                    "stream_modes": stream_modes,
+                    "stream_subgraphs": body.stream_subgraphs,
+                    "interrupt_before": body.interrupt_before,
+                    "interrupt_after": body.interrupt_after,
+                    "command": getattr(body, "command", None),
+                },
+            )
+        )
+    except Exception:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+        raise
     record.task = task
 
     # Title sync is handled by worker.py's finally block which reads the

offline-backend-20260512/backend/config.yaml

@@ -9,6 +9,29 @@ log_level: info
 gateway:
   port: 8001
 
+# Optional Redis-backed global concurrency coordination.
+# Leave disabled to keep the existing in-process behavior.
+# When enabled, Redis operations are short-timeout and fail-open by default:
+# if Redis is slow or unavailable, user Q&A continues with local-only limits.
+concurrency:
+  redis:
+    enabled: false
+    url: redis://127.0.0.1:6379/0
+    password: null
+    # Prefix with the deployment/system id because Redis may be shared by
+    # multiple projects. Keep this unique per system to avoid cross-project
+    # leases affecting each other.
+    namespace: cmzs_deerflow:concurrency
+    socket_connect_timeout_ms: 80
+    socket_timeout_ms: 80
+    acquire_timeout_ms: 120
+    operation_timeout_ms: 80
+    circuit_breaker_failures: 3
+    circuit_breaker_cooldown_seconds: 30
+    lease_ttl_seconds: 90
+    heartbeat_interval_seconds: 20
+    fail_open: true
+
 models:
   - name: deepseek-chat
     display_name: DeepSeek Chat

offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py

@@ -11,8 +11,9 @@ from pydantic import BaseModel, ConfigDict, Field
 from deerflow.config.acp_config import ACPAgentConfig, load_acp_config_from_dict
 from deerflow.config.agents_api_config import AgentsApiConfig, load_agents_api_config_from_dict
 from deerflow.config.browser_context_config import BrowserContextConfig
-from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
-from deerflow.config.database_config import DatabaseConfig
+from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
+from deerflow.config.concurrency_config import ConcurrencyConfig
+from deerflow.config.database_config import DatabaseConfig
 from deerflow.config.extensions_config import ExtensionsConfig
 from deerflow.config.gateway_config import GatewayConfig
 from deerflow.config.guardrails_config import GuardrailsConfig, load_guardrails_config_from_dict
@@ -137,8 +138,9 @@ class AppConfig(BaseModel):
     gateway: GatewayConfig = Field(default_factory=GatewayConfig, description="Gateway FastAPI listening port (used by intra-process loopback)")
     run_events: RunEventsConfig = Field(default_factory=RunEventsConfig, description="Run event storage configuration")
     checkpointer: CheckpointerConfig | None = Field(default=None, description="Checkpointer configuration")
-    stream_bridge: StreamBridgeConfig | None = Field(default=None, description="Stream bridge configuration")
-    browser_context: BrowserContextConfig = Field(default_factory=BrowserContextConfig, description="Optional browser page capture integration")
+    stream_bridge: StreamBridgeConfig | None = Field(default=None, description="Stream bridge configuration")
+    concurrency: ConcurrencyConfig = Field(default_factory=ConcurrencyConfig, description="Optional cross-worker concurrency coordination")
+    browser_context: BrowserContextConfig = Field(default_factory=BrowserContextConfig, description="Optional browser page capture integration")
     knowledge: KnowledgeConfig = Field(default_factory=KnowledgeConfig, description="Product knowledge base (obsidian-wiki compatible vault) configuration")
     knowledge_ingest: KnowledgeIngestConfig = Field(default_factory=KnowledgeIngestConfig, description="External knowledge-ingest endpoint configuration")
     task_deeplink: TaskDeeplinkConfig = Field(default_factory=TaskDeeplinkConfig, description="任务深链(rwfx)目的树闸门与跳转地址配置(前端经 /api/public/config/task-deeplink 读取)")

offline-backend-20260512/backend/packages/harness/deerflow/config/concurrency_config.py

@@ -0,0 +1,38 @@
+"""Runtime concurrency coordination configuration."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class RedisConcurrencyConfig(BaseModel):
+    """Optional Redis-backed cross-worker concurrency coordination."""
+
+    enabled: bool = Field(default=False, description="Enable Redis-backed global run concurrency coordination.")
+    url: str | None = Field(default=None, description="Redis URL, for example redis://127.0.0.1:6379/0.")
+    password: str | None = Field(
+        default=None,
+        description="Optional Redis password. You may also include credentials directly in url.",
+    )
+    namespace: str = Field(
+        default="cmzs_deerflow:concurrency",
+        description="Redis key prefix for concurrency leases. Use a deployment/system prefix when Redis is shared.",
+    )
+    socket_connect_timeout_ms: int = Field(default=80, ge=1, le=10_000)
+    socket_timeout_ms: int = Field(default=80, ge=1, le=10_000)
+    acquire_timeout_ms: int = Field(default=120, ge=1, le=10_000)
+    operation_timeout_ms: int = Field(default=80, ge=1, le=10_000)
+    circuit_breaker_failures: int = Field(default=3, ge=1, le=100)
+    circuit_breaker_cooldown_seconds: int = Field(default=30, ge=1, le=3600)
+    lease_ttl_seconds: int = Field(default=90, ge=10, le=3600)
+    heartbeat_interval_seconds: int = Field(default=20, ge=5, le=600)
+    fail_open: bool = Field(
+        default=True,
+        description="Allow runs to proceed with local-only limits when Redis is unavailable or slow.",
+    )
+
+
+class ConcurrencyConfig(BaseModel):
+    """Top-level concurrency coordination config."""
+
+    redis: RedisConcurrencyConfig = Field(default_factory=RedisConcurrencyConfig)

offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py

@@ -345,6 +345,16 @@ class ModelConcurrencySettings(BaseModel):
         default_factory=dict,
         description="Optional per-model overrides: model_name -> max concurrent runs.",
     )
+    user_default_limit: int = Field(
+        default=0,
+        ge=0,
+        le=10_000,
+        description="Fallback max concurrent runs per user and model. 0 disables the per-user fallback.",
+    )
+    user_model_limits: dict[str, int] = Field(
+        default_factory=dict,
+        description="Optional per-user per-model overrides: model_name -> max concurrent runs for one user.",
+    )
     model_config = ConfigDict(extra="ignore")
 
 

offline-backend-20260512/backend/packages/harness/deerflow/runtime/__init__.py

@@ -7,14 +7,19 @@ directly from ``deerflow.runtime``.
 
 from .checkpointer import checkpointer_context, get_checkpointer, make_checkpointer, reset_checkpointer
 from .runs import (
+    AcquireDecision,
+    AcquireResult,
+    ActiveLease,
     ConflictError,
     DisconnectMode,
     ModelConcurrencyLimitError,
+    RedisConcurrencyGate,
     RunContext,
     RunManager,
     RunRecord,
     RunStatus,
     UnsupportedStrategyError,
+    make_redis_concurrency_gate,
     run_agent,
 )
 from .serialization import serialize, serialize_channel_values, serialize_lc_object, serialize_messages_tuple
@@ -28,14 +33,19 @@ __all__ = [
     "make_checkpointer",
     "reset_checkpointer",
     # runs
+    "AcquireDecision",
+    "AcquireResult",
+    "ActiveLease",
     "ConflictError",
     "DisconnectMode",
     "ModelConcurrencyLimitError",
+    "RedisConcurrencyGate",
     "RunContext",
     "RunManager",
     "RunRecord",
     "RunStatus",
     "UnsupportedStrategyError",
+    "make_redis_concurrency_gate",
     "run_agent",
     # serialization
     "serialize",

offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/__init__.py

@@ -2,16 +2,22 @@
 
 from .manager import ConflictError, ModelConcurrencyLimitError, RunManager, RunRecord, UnsupportedStrategyError
 from .schemas import DisconnectMode, RunStatus
+from .concurrency_gate import AcquireDecision, AcquireResult, ActiveLease, RedisConcurrencyGate, make_redis_concurrency_gate
 from .worker import RunContext, run_agent
 
 __all__ = [
+    "AcquireDecision",
+    "AcquireResult",
+    "ActiveLease",
     "ConflictError",
     "DisconnectMode",
     "ModelConcurrencyLimitError",
+    "RedisConcurrencyGate",
     "RunContext",
     "RunManager",
     "RunRecord",
     "RunStatus",
     "UnsupportedStrategyError",
+    "make_redis_concurrency_gate",
     "run_agent",
 ]

offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/concurrency_gate.py

@@ -0,0 +1,379 @@
+"""Optional cross-worker run concurrency gates.
+
+The Redis gate is deliberately fail-open by default: Redis coordinates global
+limits when it is fast and healthy, but it must not block the chat path when it
+is slow or unavailable.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from deerflow.config.concurrency_config import RedisConcurrencyConfig
+
+logger = logging.getLogger(__name__)
+
+
+class AcquireDecision(str, Enum):
+    ACQUIRED = "acquired"
+    REJECTED = "rejected"
+    DEGRADED = "degraded"
+
+
+@dataclass(frozen=True)
+class AcquireResult:
+    decision: AcquireDecision
+    reason: str = ""
+    source: str = "redis"
+
+    @property
+    def acquired(self) -> bool:
+        return self.decision == AcquireDecision.ACQUIRED
+
+    @property
+    def rejected(self) -> bool:
+        return self.decision == AcquireDecision.REJECTED
+
+    @property
+    def degraded(self) -> bool:
+        return self.decision == AcquireDecision.DEGRADED
+
+
+@dataclass(frozen=True)
+class ActiveLease:
+    run_id: str
+    thread_id: str = ""
+    model_name: str = ""
+    user_id: str = ""
+    assistant_id: str = ""
+    created_at: float = 0.0
+    expires_at: float = 0.0
+
+
+class CircuitBreaker:
+    """Tiny in-process circuit breaker for Redis failures."""
+
+    def __init__(self, *, failure_threshold: int, cooldown_seconds: int) -> None:
+        self._failure_threshold = max(1, int(failure_threshold))
+        self._cooldown_seconds = max(1, int(cooldown_seconds))
+        self._failures = 0
+        self._opened_until = 0.0
+
+    def allow(self) -> bool:
+        return time.monotonic() >= self._opened_until
+
+    def record_success(self) -> None:
+        self._failures = 0
+        self._opened_until = 0.0
+
+    def record_failure(self) -> None:
+        self._failures += 1
+        if self._failures >= self._failure_threshold:
+            self._opened_until = time.monotonic() + self._cooldown_seconds
+
+    @property
+    def open(self) -> bool:
+        return not self.allow()
+
+
+class RedisConcurrencyGate:
+    """Redis lease-based global concurrency limiter."""
+
+    _ACQUIRE_SCRIPT = """
+local now = tonumber(ARGV[1])
+local expires_at = tonumber(ARGV[2])
+local run_id = ARGV[3]
+local model_limit = tonumber(ARGV[4])
+local user_model_limit = tonumber(ARGV[5])
+local reject_thread = tonumber(ARGV[6])
+local payload = ARGV[7]
+local ttl_ms = tonumber(ARGV[8])
+
+for i = 1, 4 do
+  redis.call('ZREMRANGEBYSCORE', KEYS[i], '-inf', now)
+end
+
+if reject_thread == 1 and redis.call('ZCARD', KEYS[4]) > 0 then
+  return {'rejected', 'thread'}
+end
+if model_limit > 0 and redis.call('ZCARD', KEYS[2]) >= model_limit then
+  return {'rejected', 'model'}
+end
+if user_model_limit > 0 and redis.call('ZCARD', KEYS[3]) >= user_model_limit then
+  return {'rejected', 'user_model'}
+end
+
+redis.call('ZADD', KEYS[1], expires_at, run_id)
+redis.call('ZADD', KEYS[2], expires_at, run_id)
+if user_model_limit > 0 then
+  redis.call('ZADD', KEYS[3], expires_at, run_id)
+end
+redis.call('ZADD', KEYS[4], expires_at, run_id)
+redis.call('SET', KEYS[5], payload, 'PX', ttl_ms)
+return {'acquired', ''}
+"""
+
+    def __init__(self, config: RedisConcurrencyConfig) -> None:
+        self.config = config
+        self._client: Any | None = None
+        self._breaker = CircuitBreaker(
+            failure_threshold=config.circuit_breaker_failures,
+            cooldown_seconds=config.circuit_breaker_cooldown_seconds,
+        )
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self.config.enabled and self.config.url)
+
+    def health(self) -> dict[str, Any]:
+        return {
+            "enabled": self.enabled,
+            "source": "redis",
+            "circuit_open": self._breaker.open,
+            "fail_open": self.config.fail_open,
+        }
+
+    async def aclose(self) -> None:
+        client = self._client
+        self._client = None
+        if client is not None:
+            try:
+                await client.aclose()
+            except Exception:
+                logger.debug("Failed to close Redis concurrency client", exc_info=True)
+
+    def _ns(self) -> str:
+        return self.config.namespace.strip().rstrip(":") or "cmzs_deerflow:concurrency"
+
+    def _keys(self, *, run_id: str, model_name: str, user_id: str, thread_id: str) -> list[str]:
+        ns = self._ns()
+        return [
+            f"{ns}:all",
+            f"{ns}:model:{model_name}",
+            f"{ns}:user_model:{model_name}:{user_id}",
+            f"{ns}:thread:{thread_id}",
+            f"{ns}:run:{run_id}",
+        ]
+
+    async def _redis(self) -> Any:
+        if self._client is not None:
+            return self._client
+        try:
+            from redis.asyncio import Redis
+        except Exception as exc:  # pragma: no cover - exercised when dependency is absent
+            raise RuntimeError("redis package is not installed") from exc
+
+        timeout = max(0.001, self.config.socket_timeout_ms / 1000)
+        connect_timeout = max(0.001, self.config.socket_connect_timeout_ms / 1000)
+        self._client = Redis.from_url(
+            self.config.url,
+            password=self.config.password,
+            socket_timeout=timeout,
+            socket_connect_timeout=connect_timeout,
+            decode_responses=True,
+        )
+        return self._client
+
+    async def _with_timeout(self, coro, timeout_ms: int) -> Any:
+        return await asyncio.wait_for(coro, timeout=max(0.001, timeout_ms / 1000))
+
+    def _degraded(self, reason: str) -> AcquireResult:
+        return AcquireResult(AcquireDecision.DEGRADED, reason=reason)
+
+    def _on_failure(self, action: str, exc: BaseException) -> None:
+        self._breaker.record_failure()
+        logger.warning(
+            "Redis concurrency %s failed; fail_open=%s circuit_open=%s: %s",
+            action,
+            self.config.fail_open,
+            self._breaker.open,
+            exc,
+        )
+
+    async def try_acquire(
+        self,
+        *,
+        run_id: str,
+        thread_id: str,
+        model_name: str,
+        user_id: str,
+        assistant_id: str = "",
+        model_limit: int | None = None,
+        user_model_limit: int | None = None,
+        reject_thread: bool = True,
+    ) -> AcquireResult:
+        if not self.enabled:
+            return self._degraded("disabled")
+        if not self._breaker.allow():
+            return self._degraded("circuit_open")
+
+        now = time.time()
+        expires_at = now + self.config.lease_ttl_seconds
+        payload = {
+            "run_id": run_id,
+            "thread_id": thread_id,
+            "model_name": model_name,
+            "user_id": user_id,
+            "assistant_id": assistant_id,
+            "created_at": now,
+            "expires_at": expires_at,
+        }
+        import json
+
+        keys = self._keys(run_id=run_id, model_name=model_name, user_id=user_id, thread_id=thread_id)
+        ttl_ms = int(self.config.lease_ttl_seconds * 1000)
+        try:
+            client = await self._redis()
+            raw = await self._with_timeout(
+                client.eval(
+                    self._ACQUIRE_SCRIPT,
+                    len(keys),
+                    *keys,
+                    str(now),
+                    str(expires_at),
+                    run_id,
+                    str(max(0, int(model_limit or 0))),
+                    str(max(0, int(user_model_limit or 0))),
+                    "1" if reject_thread else "0",
+                    json.dumps(payload, ensure_ascii=False),
+                    str(ttl_ms),
+                ),
+                self.config.acquire_timeout_ms,
+            )
+            self._breaker.record_success()
+        except Exception as exc:
+            self._on_failure("acquire", exc)
+            if self.config.fail_open:
+                return self._degraded(type(exc).__name__)
+            raise
+
+        status = raw[0] if isinstance(raw, list) and raw else ""
+        reason = raw[1] if isinstance(raw, list) and len(raw) > 1 else ""
+        if status == "acquired":
+            return AcquireResult(AcquireDecision.ACQUIRED)
+        return AcquireResult(AcquireDecision.REJECTED, reason=str(reason or "limit"))
+
+    async def heartbeat(
+        self,
+        *,
+        run_id: str,
+        thread_id: str,
+        model_name: str,
+        user_id: str,
+        user_model_accounted: bool = True,
+    ) -> None:
+        if not self.enabled or not self._breaker.allow():
+            return
+        now = time.time()
+        expires_at = now + self.config.lease_ttl_seconds
+        keys = self._keys(run_id=run_id, model_name=model_name, user_id=user_id, thread_id=thread_id)
+        try:
+            client = await self._redis()
+            pipe = client.pipeline(transaction=False)
+            refresh_keys = [keys[0], keys[1], keys[3]]
+            if user_model_accounted:
+                refresh_keys.append(keys[2])
+            for key in refresh_keys:
+                pipe.zadd(key, {run_id: expires_at})
+            pipe.expire(keys[4], self.config.lease_ttl_seconds)
+            await self._with_timeout(pipe.execute(), self.config.operation_timeout_ms)
+            self._breaker.record_success()
+        except Exception as exc:
+            self._on_failure("heartbeat", exc)
+
+    async def release(
+        self,
+        *,
+        run_id: str,
+        thread_id: str,
+        model_name: str,
+        user_id: str,
+        user_model_accounted: bool = True,
+    ) -> None:
+        if not self.enabled or not self._breaker.allow():
+            return
+        keys = self._keys(run_id=run_id, model_name=model_name, user_id=user_id, thread_id=thread_id)
+        try:
+            client = await self._redis()
+            pipe = client.pipeline(transaction=False)
+            release_keys = [keys[0], keys[1], keys[3]]
+            if user_model_accounted:
+                release_keys.append(keys[2])
+            for key in release_keys:
+                pipe.zrem(key, run_id)
+            pipe.delete(keys[4])
+            await self._with_timeout(pipe.execute(), self.config.operation_timeout_ms)
+            self._breaker.record_success()
+        except Exception as exc:
+            self._on_failure("release", exc)
+
+    async def count_active(self) -> int | None:
+        if not self.enabled or not self._breaker.allow():
+            return None
+        key = f"{self._ns()}:all"
+        now = time.time()
+        try:
+            client = await self._redis()
+            pipe = client.pipeline(transaction=False)
+            pipe.zremrangebyscore(key, "-inf", now)
+            pipe.zcard(key)
+            result = await self._with_timeout(pipe.execute(), self.config.operation_timeout_ms)
+            self._breaker.record_success()
+            return int(result[-1] or 0)
+        except Exception as exc:
+            self._on_failure("count", exc)
+            return None
+
+    async def list_active(self, *, limit: int = 1000) -> list[ActiveLease] | None:
+        if not self.enabled or not self._breaker.allow():
+            return None
+        import json
+
+        all_key = f"{self._ns()}:all"
+        now = time.time()
+        try:
+            client = await self._redis()
+            await self._with_timeout(client.zremrangebyscore(all_key, "-inf", now), self.config.operation_timeout_ms)
+            run_ids = await self._with_timeout(client.zrange(all_key, 0, max(0, limit - 1)), self.config.operation_timeout_ms)
+            if not run_ids:
+                self._breaker.record_success()
+                return []
+            keys = [f"{self._ns()}:run:{run_id}" for run_id in run_ids]
+            payloads = await self._with_timeout(client.mget(keys), self.config.operation_timeout_ms)
+            leases: list[ActiveLease] = []
+            for run_id, payload in zip(run_ids, payloads, strict=False):
+                if not payload:
+                    continue
+                try:
+                    data = json.loads(payload)
+                except Exception:
+                    continue
+                leases.append(
+                    ActiveLease(
+                        run_id=str(data.get("run_id") or run_id),
+                        thread_id=str(data.get("thread_id") or ""),
+                        model_name=str(data.get("model_name") or ""),
+                        user_id=str(data.get("user_id") or ""),
+                        assistant_id=str(data.get("assistant_id") or ""),
+                        created_at=float(data.get("created_at") or 0),
+                        expires_at=float(data.get("expires_at") or 0),
+                    )
+                )
+            self._breaker.record_success()
+            return leases
+        except Exception as exc:
+            self._on_failure("list", exc)
+            return None
+
+
+def make_redis_concurrency_gate(config: Any) -> RedisConcurrencyGate | None:
+    concurrency = getattr(config, "concurrency", None)
+    redis_config = getattr(concurrency, "redis", None)
+    if redis_config is None or not getattr(redis_config, "enabled", False) or not getattr(redis_config, "url", None):
+        return None
+    return RedisConcurrencyGate(redis_config)

offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py

@@ -218,6 +218,8 @@ class RunManager:
         kwargs: dict | None = None,
         multitask_strategy: str = "reject",
         model_concurrency_limit: tuple[str, int] | None = None,
+        user_model_concurrency_limit: tuple[str, str, int] | None = None,
+        run_id: str | None = None,
     ) -> RunRecord:
         """Atomically check for inflight runs and create a new one.
 
@@ -228,7 +230,7 @@ class RunManager:
         This method holds the lock across both the check and the insert,
         eliminating the TOCTOU race in separate ``has_inflight`` + ``create``.
         """
-        run_id = str(uuid.uuid4())
+        run_id = run_id or str(uuid.uuid4())
         now = _now_iso()
 
         _supported_strategies = ("reject", "interrupt", "rollback")
@@ -261,7 +263,20 @@ class RunManager:
                 model_name, limit = model_concurrency_limit
                 active_for_model = sum(1 for r in self._runs.values() if r.status in (RunStatus.pending, RunStatus.running) and str((r.metadata or {}).get("model_name") or "").strip() == model_name)
                 if active_for_model >= max(1, int(limit)):
-                    raise ModelConcurrencyLimitError(model_name, limit)
+                    raise ModelConcurrencyLimitError(model_name, limit, scope="model")
+
+            if user_model_concurrency_limit is not None:
+                model_name, user_id, limit = user_model_concurrency_limit
+                active_for_user_model = sum(
+                    1
+                    for r in self._runs.values()
+                    if r.status in (RunStatus.pending, RunStatus.running)
+                    and str((r.metadata or {}).get("model_name") or "").strip() == model_name
+                    and str((r.metadata or {}).get("user_id") or "").strip() == user_id
+                    and str((r.metadata or {}).get("run_scope") or "external_root").strip() == "external_root"
+                )
+                if active_for_user_model >= max(1, int(limit)):
+                    raise ModelConcurrencyLimitError(model_name, limit, scope="user_model", user_id=user_id)
 
             record = RunRecord(
                 run_id=run_id,
@@ -316,9 +331,11 @@ class ConflictError(Exception):
 class ModelConcurrencyLimitError(Exception):
     """Raised when a model-specific in-flight run limit has been reached."""
 
-    def __init__(self, model_name: str, limit: int) -> None:
+    def __init__(self, model_name: str, limit: int, *, scope: str = "model", user_id: str | None = None) -> None:
         self.model_name = model_name
         self.limit = limit
+        self.scope = scope
+        self.user_id = user_id
         super().__init__(f"Model {model_name} has reached its concurrency limit")
 
 

offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/model_concurrency.py

@@ -0,0 +1,39 @@
+"""Small in-memory gate used by legacy model-concurrency tests."""
+
+from __future__ import annotations
+
+import asyncio
+
+
+class ModelConcurrencyGate:
+    """Track per-model active counts with atomic try-acquire semantics."""
+
+    def __init__(self) -> None:
+        self._lock = asyncio.Lock()
+        self._counts: dict[str, int] = {}
+
+    async def try_acquire(self, model_name: str, limit: int) -> bool:
+        key = str(model_name).strip()
+        if not key:
+            return True
+        async with self._lock:
+            current = self._counts.get(key, 0)
+            if current >= max(1, int(limit)):
+                return False
+            self._counts[key] = current + 1
+            return True
+
+    async def release(self, model_name: str) -> None:
+        key = str(model_name).strip()
+        if not key:
+            return
+        async with self._lock:
+            current = self._counts.get(key, 0)
+            if current <= 1:
+                self._counts.pop(key, None)
+            else:
+                self._counts[key] = current - 1
+
+    async def snapshot(self) -> dict[str, int]:
+        async with self._lock:
+            return dict(self._counts)

offline-backend-20260512/backend/packages/harness/pyproject.toml

@@ -23,9 +23,10 @@ dependencies = [
     "markdownify>=1.2.2",
     "markitdown[all,xlsx]>=0.0.1a2",
     "pydantic>=2.12.5",
-    "pyyaml>=6.0.3",
-    "readabilipy>=0.3.0",
-    "tavily-python>=0.7.17",
+    "pyyaml>=6.0.3",
+    "readabilipy>=0.3.0",
+    "redis>=5.0.0",
+    "tavily-python>=0.7.17",
     "firecrawl-py>=1.15.0",
     "tiktoken>=0.8.0",
     "ddgs>=9.10.0",

offline-backend-20260512/backend/tests/test_model_concurrency.py

@@ -10,10 +10,17 @@ import pytest
 
 from app.gateway.routers import system_settings as ss_router
 from app.gateway.services import (
+    RUN_SCOPE_EXTERNAL_ROOT,
+    RUN_SCOPE_ORCHESTRATION_CHILD,
     _get_model_concurrency_limit,
+    _get_user_model_concurrency_limit,
+    _resolve_run_scope,
     _resolve_effective_run_model_name,
+    internal_run_scope_headers,
 )
 from deerflow.config import system_settings as ss
+from deerflow.config.concurrency_config import RedisConcurrencyConfig
+from deerflow.runtime.runs.concurrency_gate import RedisConcurrencyGate
 from deerflow.runtime.runs.manager import ModelConcurrencyLimitError, RunManager
 
 
@@ -52,23 +59,31 @@ async def test_model_concurrency_settings_round_trip(
             enabled=True,
             default_limit=8,
             model_limits={"glm5": 10, "deepseek": 4},
+            user_default_limit=2,
+            user_model_limits={"glm5": 1},
         ),
         req,
     )
     assert saved.enabled is True
     assert saved.default_limit == 8
     assert saved.model_limits == {"glm5": 10, "deepseek": 4}
+    assert saved.user_default_limit == 2
+    assert saved.user_model_limits == {"glm5": 1}
 
     file_path = tmp_runtime_home / "system_settings.json"
     payload = json.loads(file_path.read_text(encoding="utf-8"))
     assert payload["model_concurrency"]["enabled"] is True
     assert payload["model_concurrency"]["default_limit"] == 8
     assert payload["model_concurrency"]["model_limits"]["glm5"] == 10
+    assert payload["model_concurrency"]["user_default_limit"] == 2
+    assert payload["model_concurrency"]["user_model_limits"]["glm5"] == 1
 
     loaded = await ss_router.get_model_concurrency(req)
     assert loaded.enabled is True
     assert loaded.default_limit == 8
     assert loaded.model_limits["deepseek"] == 4
+    assert loaded.user_default_limit == 2
+    assert loaded.user_model_limits["glm5"] == 1
 
 
 def test_resolve_effective_run_model_name_prefers_requested_model(
@@ -183,6 +198,53 @@ async def test_run_manager_rejects_model_limit_atomically(
         )
 
 
+@pytest.mark.asyncio
+async def test_run_manager_rejects_user_model_limit_atomically(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        "app.gateway.services.load_system_settings",
+        lambda: SimpleNamespace(
+            model_concurrency=SimpleNamespace(
+                enabled=True,
+                default_limit=10,
+                model_limits={},
+                user_default_limit=3,
+                user_model_limits={"glm5": 1},
+            )
+        ),
+    )
+    assert _get_user_model_concurrency_limit("glm5") == 1
+    assert _get_user_model_concurrency_limit("other") == 3
+
+    run_mgr = RunManager()
+    await run_mgr.create_or_reject(
+        "t-child",
+        metadata={
+            "model_name": "glm5",
+            "user_id": "u1",
+            "run_scope": RUN_SCOPE_ORCHESTRATION_CHILD,
+        },
+        user_model_concurrency_limit=("glm5", "u1", 1),
+    )
+    await run_mgr.create_or_reject("t1", metadata={"model_name": "glm5", "user_id": "u1"})
+    await run_mgr.create_or_reject(
+        "t2",
+        metadata={"model_name": "glm5", "user_id": "u2"},
+        user_model_concurrency_limit=("glm5", "u2", 1),
+    )
+
+    with pytest.raises(ModelConcurrencyLimitError) as exc_info:
+        await run_mgr.create_or_reject(
+            "t3",
+            metadata={"model_name": "glm5", "user_id": "u1"},
+            user_model_concurrency_limit=("glm5", "u1", 1),
+        )
+
+    assert exc_info.value.scope == "user_model"
+    assert exc_info.value.user_id == "u1"
+
+
 def test_get_model_concurrency_limit_skips_when_disabled(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -193,7 +255,51 @@ def test_get_model_concurrency_limit_skips_when_disabled(
                 enabled=False,
                 default_limit=1,
                 model_limits={"glm5": 1},
+                user_default_limit=1,
+                user_model_limits={"glm5": 1},
             )
         ),
     )
     assert _get_model_concurrency_limit("glm5") is None
+    assert _get_user_model_concurrency_limit("glm5") is None
+
+
+def test_internal_run_scope_is_only_trusted_for_loopback_or_server_argument() -> None:
+    body = SimpleNamespace(metadata={"run_scope": RUN_SCOPE_ORCHESTRATION_CHILD})
+
+    trusted_loopback_request = SimpleNamespace(
+        client=SimpleNamespace(host="127.0.0.1"),
+        headers=internal_run_scope_headers(),
+    )
+    untrusted_loopback_request = SimpleNamespace(
+        client=SimpleNamespace(host="127.0.0.1"),
+        headers={},
+    )
+    public_request = SimpleNamespace(
+        client=SimpleNamespace(host="10.0.0.8"),
+        headers=internal_run_scope_headers(),
+    )
+
+    assert (
+        _resolve_run_scope(body, trusted_loopback_request, None)
+        == RUN_SCOPE_ORCHESTRATION_CHILD
+    )
+    assert (
+        _resolve_run_scope(body, untrusted_loopback_request, None)
+        == RUN_SCOPE_EXTERNAL_ROOT
+    )
+    assert _resolve_run_scope(body, public_request, None) == RUN_SCOPE_EXTERNAL_ROOT
+    assert (
+        _resolve_run_scope(
+            SimpleNamespace(metadata={}),
+            public_request,
+            RUN_SCOPE_ORCHESTRATION_CHILD,
+        )
+        == RUN_SCOPE_ORCHESTRATION_CHILD
+    )
+
+
+def test_redis_concurrency_namespace_has_system_prefix() -> None:
+    assert RedisConcurrencyConfig().namespace == "cmzs_deerflow:concurrency"
+    gate = RedisConcurrencyGate(RedisConcurrencyConfig(namespace=""))
+    assert gate._ns() == "cmzs_deerflow:concurrency"

offline-backend-20260512/offline-runtime-requirements.txt

@@ -0,0 +1,27 @@
+# Direct runtime packages layered into the prebuilt offline backend image.
+# Keep these three pins synchronized with start_backend.sh.
+croniter==6.2.2
+hindsight-client==0.6.2
+redis==5.2.1
+
+# Fully pinned dependency closure for CPython 3.12 on Linux. Keeping the
+# transitive packages here makes a newly generated wheelhouse reproducible.
+aiohappyeyeballs==2.7.1
+aiohttp==3.14.2
+aiohttp-retry==2.9.1
+aiosignal==1.4.0
+annotated-types==0.7.0
+async-timeout==5.0.1
+attrs==26.1.0
+frozenlist==1.8.0
+idna==3.18
+multidict==6.7.1
+propcache==0.5.2
+pydantic==2.13.4
+pydantic-core==2.46.4
+python-dateutil==2.9.0.post0
+six==1.17.0
+typing-extensions==4.16.0
+typing-inspection==0.4.2
+urllib3==2.7.0
+yarl==1.24.5

offline-backend-20260512/prepare_wheelhouse.sh

@@ -0,0 +1,99 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+PYTHON_BIN="${PYTHON_BIN:-python3}"
+REQUIREMENTS_FILE="${REQUIREMENTS_FILE:-${SCRIPT_DIR}/offline-runtime-requirements.txt}"
+WHEELHOUSE_DIR="${WHEELHOUSE_DIR:-${SCRIPT_DIR}/wheelhouse}"
+PIP_INDEX_URL="${PIP_INDEX_URL:-https://pypi.org/simple}"
+
+# The supplied backend image is CPython 3.12 on Debian x86_64. These values can
+# be overridden when building an offline bundle for a different image/CPU.
+TARGET_PLATFORM="${TARGET_PLATFORM:-manylinux2014_x86_64}"
+TARGET_PYTHON_VERSION="${TARGET_PYTHON_VERSION:-312}"
+TARGET_IMPLEMENTATION="${TARGET_IMPLEMENTATION:-cp}"
+TARGET_ABI="${TARGET_ABI:-cp312}"
+
+die() {
+  echo "ERROR: $*" >&2
+  exit 1
+}
+
+require_cmd() {
+  command -v "$1" >/dev/null 2>&1 || die "command not found: $1"
+}
+
+require_wheel() {
+  local pattern="$1"
+  compgen -G "${WHEELHOUSE_DIR}/${pattern}" >/dev/null || die "wheel not downloaded: ${pattern}"
+}
+
+require_cmd "${PYTHON_BIN}"
+require_cmd sha256sum
+[ -f "${REQUIREMENTS_FILE}" ] || die "requirements file not found: ${REQUIREMENTS_FILE}"
+"${PYTHON_BIN}" -m pip --version >/dev/null 2>&1 || die "pip is not available for ${PYTHON_BIN}"
+
+mkdir -p "${WHEELHOUSE_DIR}"
+
+download_args=(
+  download
+  --dest "${WHEELHOUSE_DIR}"
+  --only-binary=:all:
+  --platform "${TARGET_PLATFORM}"
+  --python-version "${TARGET_PYTHON_VERSION}"
+  --implementation "${TARGET_IMPLEMENTATION}"
+  --abi "${TARGET_ABI}"
+)
+
+echo "Downloading Linux/Python runtime wheels"
+echo "  requirements: ${REQUIREMENTS_FILE}"
+echo "  destination:  ${WHEELHOUSE_DIR}"
+echo "  platform:     ${TARGET_PLATFORM}"
+echo "  Python/ABI:   ${TARGET_PYTHON_VERSION}/${TARGET_ABI}"
+
+"${PYTHON_BIN}" -m pip "${download_args[@]}" \
+  --index-url "${PIP_INDEX_URL}" \
+  --requirement "${REQUIREMENTS_FILE}"
+
+require_wheel 'croniter-6.2.2-*.whl'
+require_wheel 'hindsight_client-0.6.2-*.whl'
+require_wheel 'redis-5.2.1-*.whl'
+
+# Resolve the same requirements a second time using only wheelhouse. This
+# catches a missing transitive dependency before the bundle reaches offline.
+verify_dir="$(mktemp -d)"
+cleanup() {
+  if [ -n "${verify_dir:-}" ] && [ -d "${verify_dir}" ]; then
+    rm -rf -- "${verify_dir}"
+  fi
+}
+trap cleanup EXIT
+
+verify_args=(
+  download
+  --dest "${verify_dir}"
+  --no-index
+  --find-links "${WHEELHOUSE_DIR}"
+  --only-binary=:all:
+  --platform "${TARGET_PLATFORM}"
+  --python-version "${TARGET_PYTHON_VERSION}"
+  --implementation "${TARGET_IMPLEMENTATION}"
+  --abi "${TARGET_ABI}"
+  --requirement "${REQUIREMENTS_FILE}"
+)
+"${PYTHON_BIN}" -m pip "${verify_args[@]}" >/dev/null
+
+shopt -s nullglob
+wheels=("${WHEELHOUSE_DIR}"/*.whl)
+[ "${#wheels[@]}" -gt 0 ] || die "wheelhouse is empty: ${WHEELHOUSE_DIR}"
+(
+  cd "${WHEELHOUSE_DIR}"
+  sha256sum -- *.whl | LC_ALL=C sort > SHA256SUMS
+)
+
+echo
+echo "Wheelhouse is complete for the pinned runtime requirements."
+echo "  wheel count: ${#wheels[@]}"
+echo "  checksums:   ${WHEELHOUSE_DIR}/SHA256SUMS"
+echo "Copy the wheelhouse directory with the deployment bundle."

offline-backend-20260512/start_backend.sh

@@ -0,0 +1,140 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+IMAGE_TAG="${IMAGE_TAG:-deerflow-backend-offline:20260505}"
+CONTAINER_NAME="${CONTAINER_NAME:-deerflow-backend}"
+HOST_PORT="${HOST_PORT:-8001}"
+IMAGE_TAR="${IMAGE_TAR:-${SCRIPT_DIR}/deerflow-backend-offline-image-20260505.tar}"
+# Default timezone for the container. The offline image is built from a base
+# layer that ships with UTC, which made scheduled tasks fire at the wrong wall
+# clock and made log timestamps confusing for ops. ``Asia/Shanghai`` keeps the
+# scheduler aligned with the operator's expectation; override with TZ=... when
+# deploying to a different region.
+CONTAINER_TZ="${CONTAINER_TZ:-${TZ:-Asia/Shanghai}}"
+
+die() {
+  echo "ERROR: $*" >&2
+  exit 1
+}
+
+require_cmd() {
+  command -v "$1" >/dev/null 2>&1 || die "command not found: $1"
+}
+
+require_file() {
+  [ -f "$1" ] || die "required file not found: $1"
+}
+
+require_dir() {
+  [ -d "$1" ] || die "required directory not found: $1"
+}
+
+require_env_key() {
+  local key="$1"
+  grep -Eq "^[[:space:]]*${key}=" "${SCRIPT_DIR}/.env" || die "required .env key not found: ${key}"
+}
+
+require_cmd docker
+
+if [ ! -f "${SCRIPT_DIR}/.env" ]; then
+  if [ -f "${SCRIPT_DIR}/.env.example" ]; then
+    cp "${SCRIPT_DIR}/.env.example" "${SCRIPT_DIR}/.env"
+    echo "Generated .env from .env.example. Please edit .env and run this script again."
+    exit 1
+  fi
+  die "missing .env"
+fi
+
+if [ ! -f "${SCRIPT_DIR}/config.yaml" ]; then
+  if [ -f "${SCRIPT_DIR}/config.yaml.example" ]; then
+    cp "${SCRIPT_DIR}/config.yaml.example" "${SCRIPT_DIR}/config.yaml"
+    echo "Generated config.yaml from config.yaml.example. Please edit model/API settings and run this script again."
+    exit 1
+  fi
+  die "missing config.yaml"
+fi
+
+require_env_key "AUTH_JWT_SECRET"
+require_env_key "SCHEDULED_TASKS_ENABLED"
+require_env_key "ELEMENT_ENABLED"
+require_file "${SCRIPT_DIR}/config.yaml"
+require_file "${SCRIPT_DIR}/extensions_config.json"
+require_file "${SCRIPT_DIR}/skill_evolution.json"
+require_file "${SCRIPT_DIR}/backend/pyproject.toml"
+require_file "${SCRIPT_DIR}/backend/app/gateway/app.py"
+require_file "${SCRIPT_DIR}/backend/packages/harness/deerflow/runtime/scheduler/element.py"
+require_dir "${SCRIPT_DIR}/skills"
+require_dir "${SCRIPT_DIR}/wheelhouse"
+# Knowledge base directory, mounted into the container at /mnt/knowledge_base.
+require_dir "${SCRIPT_DIR}/backend/knowledge/knowledge_base"
+mkdir -p "${SCRIPT_DIR}/data"
+
+if ! docker image inspect "${IMAGE_TAG}" >/dev/null 2>&1; then
+  require_file "${IMAGE_TAR}"
+  echo "Loading Docker image from ${IMAGE_TAR}"
+  docker load -i "${IMAGE_TAR}"
+fi
+
+if docker ps -a --format '{{.Names}}' | awk -v name="${CONTAINER_NAME}" '$0 == name { found=1 } END { exit(found ? 0 : 1) }'; then
+  echo "Removing existing container ${CONTAINER_NAME}"
+  docker rm -f "${CONTAINER_NAME}" >/dev/null
+fi
+
+echo "Starting backend container ${CONTAINER_NAME}"
+docker run -d \
+  --name "${CONTAINER_NAME}" \
+  --restart unless-stopped \
+  -p "${HOST_PORT}:8001" \
+  --env-file "${SCRIPT_DIR}/.env" \
+  -e TZ="${CONTAINER_TZ}" \
+  -e DEER_FLOW_PROJECT_ROOT=/app \
+  -e DEER_FLOW_HOME=/app/backend/.deer-flow \
+  -e DEER_FLOW_CONFIG_PATH=/app/config.yaml \
+  -e DEER_FLOW_EXTENSIONS_CONFIG_PATH=/app/extensions_config.json \
+  -e DEER_FLOW_SKILL_EVOLUTION_CONFIG_PATH=/app/skill_evolution.json \
+  -e DEER_FLOW_SKILLS_PATH=/app/skills \
+  -e UV_LINK_MODE=copy \
+  -v "${SCRIPT_DIR}/backend/app:/app/backend/app" \
+  -v "${SCRIPT_DIR}/backend/packages:/app/backend/packages" \
+  -v "${SCRIPT_DIR}/backend/scripts:/app/backend/scripts" \
+  -v "${SCRIPT_DIR}/backend/pyproject.toml:/app/backend/pyproject.toml:ro" \
+  -v "${SCRIPT_DIR}/backend/uv.lock:/app/backend/uv.lock:ro" \
+  -v "${SCRIPT_DIR}/backend/langgraph.json:/app/backend/langgraph.json:ro" \
+  -v "${SCRIPT_DIR}/config.yaml:/app/config.yaml:ro" \
+  -v "${SCRIPT_DIR}/extensions_config.json:/app/extensions_config.json:ro" \
+  -v "${SCRIPT_DIR}/skill_evolution.json:/app/skill_evolution.json:ro" \
+  -v "${SCRIPT_DIR}/skills:/app/skills:ro" \
+  -v "${SCRIPT_DIR}/wheelhouse:/app/wheelhouse:ro" \
+  -v "${SCRIPT_DIR}/backend/knowledge/knowledge_base:/mnt/knowledge_base:ro" \
+  -v "${SCRIPT_DIR}/data:/app/backend/.deer-flow" \
+  "${IMAGE_TAG}" \
+  sh -c 'cd backend
+# Apply timezone inside the container. ``TZ`` alone is enough for Python /
+# uvicorn log timestamps, but linking /etc/localtime keeps C-level tools
+# (e.g. anything that reads /etc/localtime directly) on the same clock.
+if [ -n "${TZ:-}" ] && [ -f "/usr/share/zoneinfo/${TZ}" ]; then
+  ln -sf "/usr/share/zoneinfo/${TZ}" /etc/localtime 2>/dev/null || true
+  echo "${TZ}" > /etc/timezone 2>/dev/null || true
+fi
+if [ "${SCHEDULER_INSTALL_DEPS:-1}" != "0" ]; then
+  uv run --no-sync python -c "import croniter" >/dev/null 2>&1 || {
+    PYTHON_BIN="$(uv run --no-sync python -c "import sys; print(sys.executable)")"
+    uv pip install --python "${PYTHON_BIN}" --no-index --find-links /app/wheelhouse ${SCHEDULER_PIP_PACKAGES:-croniter==6.2.2}
+  }
+fi
+if [ "${HINDSIGHT_INSTALL_DEPS:-1}" != "0" ]; then
+  uv run --no-sync python -c "import hindsight_client" >/dev/null 2>&1 || {
+    PYTHON_BIN="$(uv run --no-sync python -c "import sys; print(sys.executable)")"
+    uv pip install --python "${PYTHON_BIN}" --no-index --find-links /app/wheelhouse ${HINDSIGHT_PIP_PACKAGES:-hindsight-client==0.6.2}
+  }
+fi
+uv run --no-sync python -c "import redis" >/dev/null 2>&1 || uv pip install --python "$(uv run --no-sync python -c "import sys; print(sys.executable)")" --no-index --find-links /app/wheelhouse ${REDIS_PIP_PACKAGES:-redis==5.2.1}
+PYTHONPATH=. uv run --no-sync python -m uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001 --workers ${GATEWAY_WORKERS:-2}'
+
+echo
+echo "Backend started."
+echo "  URL:       http://<server-ip>:${HOST_PORT}"
+echo "  Logs:      docker logs -f ${CONTAINER_NAME}"
+echo "  Stop:      ./stop_backend.sh"