From c23350f17b4516a85b981980f2c1e4ad3a28b110 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Tue, 30 Jun 2026 14:56:29 +0800 Subject: [PATCH] fix: isolate user concurrency from internal agent runs --- .../gateway/roundtable_inprocess_gateway.py | 9 ++- .../app/gateway/routers/multi_agent.py | 8 ++- .../backend/app/gateway/services.py | 70 ++++++++++++++++++- offline-backend-20260512/backend/config.yaml | 5 +- .../deerflow/config/concurrency_config.py | 5 +- .../deerflow/runtime/runs/concurrency_gate.py | 18 +++-- .../harness/deerflow/runtime/runs/manager.py | 1 + .../backend/tests/test_model_concurrency.py | 56 +++++++++++++++ 8 files changed, 162 insertions(+), 10 deletions(-) diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py index fadc3093..c34a7867 100644 --- a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py +++ b/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( diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py index 83046e24..c3bcfe0a 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py +++ b/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, diff --git a/offline-backend-20260512/backend/app/gateway/services.py b/offline-backend-20260512/backend/app/gateway/services.py index 8331a565..22cdd9e9 100644 --- a/offline-backend-20260512/backend/app/gateway/services.py +++ b/offline-backend-20260512/backend/app/gateway/services.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio import json import logging +import os import re import time import uuid @@ -41,12 +42,68 @@ 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, str], + lease_metadata: dict[str, Any], heartbeat_interval_seconds: int, run_agent_kwargs: dict[str, Any], ) -> None: @@ -750,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. @@ -793,21 +851,29 @@ async def start_run( 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) + 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( diff --git a/offline-backend-20260512/backend/config.yaml b/offline-backend-20260512/backend/config.yaml index 1b01c238..474607d1 100644 --- a/offline-backend-20260512/backend/config.yaml +++ b/offline-backend-20260512/backend/config.yaml @@ -18,7 +18,10 @@ concurrency: enabled: false url: redis://127.0.0.1:6379/0 password: null - namespace: deerflow:concurrency + # 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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/concurrency_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/concurrency_config.py index d741fda1..381e099b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/concurrency_config.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/concurrency_config.py @@ -14,7 +14,10 @@ class RedisConcurrencyConfig(BaseModel): default=None, description="Optional Redis password. You may also include credentials directly in url.", ) - namespace: str = Field(default="deerflow:concurrency", description="Redis key prefix for concurrency leases.") + 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) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/concurrency_gate.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/concurrency_gate.py index b2bd9e2f..b483a2e3 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/concurrency_gate.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/concurrency_gate.py @@ -110,7 +110,9 @@ end redis.call('ZADD', KEYS[1], expires_at, run_id) redis.call('ZADD', KEYS[2], expires_at, run_id) -redis.call('ZADD', KEYS[3], 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', ''} @@ -146,7 +148,7 @@ return {'acquired', ''} logger.debug("Failed to close Redis concurrency client", exc_info=True) def _ns(self) -> str: - return self.config.namespace.strip().rstrip(":") or "deerflow:concurrency" + 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() @@ -263,6 +265,7 @@ return {'acquired', ''} 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 @@ -272,7 +275,10 @@ return {'acquired', ''} try: client = await self._redis() pipe = client.pipeline(transaction=False) - for key in keys[:4]: + 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) @@ -287,6 +293,7 @@ return {'acquired', ''} 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 @@ -294,7 +301,10 @@ return {'acquired', ''} try: client = await self._redis() pipe = client.pipeline(transaction=False) - for key in keys[:4]: + 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) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py index edef5296..07daf14c 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -273,6 +273,7 @@ class RunManager: 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) diff --git a/offline-backend-20260512/backend/tests/test_model_concurrency.py b/offline-backend-20260512/backend/tests/test_model_concurrency.py index 86091e12..7468d9d4 100644 --- a/offline-backend-20260512/backend/tests/test_model_concurrency.py +++ b/offline-backend-20260512/backend/tests/test_model_concurrency.py @@ -10,11 +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 @@ -212,6 +218,15 @@ async def test_run_manager_rejects_user_model_limit_atomically( 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", @@ -247,3 +262,44 @@ def test_get_model_concurrency_limit_skips_when_disabled( ) 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"