mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-15 07:41:20 +00:00
fix(gateway): read the run and authorization config off the event loop (#8137)
* fix(gateway): read the run and authorization config off the event loop get_app_config() stats and hashes config.yaml and extensions_config.json on every call and reparses them after an edit. Run admission read it inline on the loop two to four times per start (get_run_context, build_run_config, the model allowlist, the approval switch, conversation references), and the model list, model details, skill list, suggestions and sandbox gates read the authorization section inline once each. start_run now builds its run context in a worker and reuses that context's app_config for the rest of the start, so one run start sees one snapshot. Scheduled dispatch resolves its recursion limit in a worker, run inputs take only the AgentRuns capability, and the route-scoped authorization helpers run through asyncio.to_thread, as /api/models/resolve already did. * docs(changelog): reference #8137 for the off-loop run and authorization config reads * fix(gateway): validate the tool approval mode against the run's config snapshot The approval-mode check added on main read get_app_config() inline in start_run, so every run start revalidated config.yaml on the event loop again. It now uses the snapshot the run context was built with, like the model allowlist and the approval switch. * test(gateway): seed the skill catalog the authorization anchor lists The skill listing anchor relied on the checkout's skills/public tree containing academic-paper-review. It now writes one allowed and one hidden skill under its own skills.path, so it passes without the repo catalog and still fails if the policy is not applied.
This commit is contained in:
parent
a68dc2a1a0
commit
1804b7c522
@ -191,7 +191,10 @@ def _get_route_authorization_config() -> AuthorizationConfig:
|
||||
environments without a config.yaml), preserving legacy all-permissions behavior.
|
||||
``get_app_config()`` stats and hashes ``config.yaml`` and
|
||||
``extensions_config.json`` on every call, so async request paths use
|
||||
:func:`_route_authorization_config_async`.
|
||||
:func:`_route_authorization_config_async`, and call the synchronous
|
||||
helpers that read it (``resolve_model_authorization``,
|
||||
``resolve_skill_authorization``, ``authorize_model_use``,
|
||||
``authorize_sandbox_for_request``) through ``asyncio.to_thread``.
|
||||
"""
|
||||
from deerflow.config.app_config import get_app_config
|
||||
|
||||
@ -820,7 +823,7 @@ async def try_acquire_sandbox_for_request(
|
||||
|
||||
user = await get_optional_user_from_request(request) if request is not None else None
|
||||
if user is not None:
|
||||
authorize_sandbox_for_request(user, is_internal=_is_internal_caller(request, user), app_config=app_config)
|
||||
await asyncio.to_thread(authorize_sandbox_for_request, user, is_internal=_is_internal_caller(request, user), app_config=app_config)
|
||||
except SandboxAuthorizationError:
|
||||
logger.info("Sandbox sync skipped: sandbox execution not permitted for this caller (thread_id=%s)", thread_id)
|
||||
return SandboxRequestLease(
|
||||
|
||||
@ -1290,6 +1290,13 @@ def _run_completion_hook(*, scheduled_task_service: Any | None, batch_continuati
|
||||
return handle
|
||||
|
||||
|
||||
def get_agent_runs(request: Request):
|
||||
"""Return the host-bound AgentRuns capability for *request*, or ``None``."""
|
||||
host = getattr(request.app.state, "agent_runs_host", None)
|
||||
# Internal/channel and PAT runs deliberately receive no retained delegation.
|
||||
return host.bind(request) if host is not None else None
|
||||
|
||||
|
||||
def get_run_context(request: Request) -> RunContext:
|
||||
"""Build a :class:`RunContext` from ``app.state`` singletons.
|
||||
|
||||
@ -1299,10 +1306,12 @@ def get_run_context(request: Request) -> RunContext:
|
||||
``event_store`` / ``run_events_config`` pair stays frozen to the snapshot
|
||||
captured in :func:`langgraph_runtime` so callers never see a store bound to
|
||||
one backend paired with a config pointing at another.
|
||||
|
||||
Loading ``app_config`` stats and hashes ``config.yaml`` (and reparses it
|
||||
after an edit), so async callers build the context with
|
||||
``asyncio.to_thread``.
|
||||
"""
|
||||
host = getattr(request.app.state, "agent_runs_host", None)
|
||||
# Internal/channel and PAT runs deliberately receive no retained delegation.
|
||||
agent_runs = host.bind(request) if host is not None else None
|
||||
agent_runs = get_agent_runs(request)
|
||||
checkpointer = get_checkpointer(request)
|
||||
scheduled_task_service = getattr(request.app.state, "scheduled_task_service", None)
|
||||
return RunContext(
|
||||
|
||||
@ -197,7 +197,7 @@ async def list_models(
|
||||
owner = await _resolve_bound_owner_for_model_list(request)
|
||||
principal_user = owner if owner is not None else user
|
||||
try:
|
||||
provider, principal = resolve_model_authorization(principal_user, is_internal=_is_internal_caller(request, user))
|
||||
provider, principal = await asyncio.to_thread(resolve_model_authorization, principal_user, is_internal=_is_internal_caller(request, user))
|
||||
except _AuthorizationUnavailable as exc:
|
||||
if exc.fail_closed:
|
||||
visible_models = []
|
||||
@ -263,6 +263,6 @@ async def get_model(
|
||||
# model exists but the role lacks permission to use it).
|
||||
user = await get_optional_user_from_request(request)
|
||||
if user is not None:
|
||||
authorize_model_use(user, model_name, is_internal=_is_internal_caller(request, user), app_config=config)
|
||||
await asyncio.to_thread(authorize_model_use, user, model_name, is_internal=_is_internal_caller(request, user), app_config=config)
|
||||
|
||||
return _model_response(model)
|
||||
|
||||
@ -492,7 +492,7 @@ async def _filter_visible_skills(
|
||||
return skills
|
||||
|
||||
try:
|
||||
provider, principal = resolve_skill_authorization(user, is_internal=_is_internal_caller(request, user))
|
||||
provider, principal = await asyncio.to_thread(resolve_skill_authorization, user, is_internal=_is_internal_caller(request, user))
|
||||
except _AuthorizationUnavailable as exc:
|
||||
return [] if exc.fail_closed else skills
|
||||
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
|
||||
@ -122,7 +123,7 @@ async def generate_suggestions(
|
||||
# default when the caller omits model_name. Keep permission failures out
|
||||
# of the best-effort LLM error handler below so a denial remains a 403.
|
||||
user = await get_current_user_from_request(request)
|
||||
authorize_model_use(user, body.model_name, is_internal=_is_internal_caller(request, user), app_config=config)
|
||||
await asyncio.to_thread(authorize_model_use, user, body.model_name, is_internal=_is_internal_caller(request, user), app_config=config)
|
||||
|
||||
system_instruction = (
|
||||
"You are generating follow-up questions to help the user continue the conversation.\n"
|
||||
|
||||
@ -9,7 +9,7 @@ from uuid import NAMESPACE_URL, uuid5
|
||||
from fastapi import HTTPException, Request
|
||||
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
||||
|
||||
from app.gateway.deps import get_current_user, get_run_context, get_run_manager
|
||||
from app.gateway.deps import get_agent_runs, get_current_user, get_run_manager
|
||||
from deerflow.runtime.runs.inputs import InputCancelled, InputConflict, RunInputs
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -44,7 +44,7 @@ async def submit_run_input(thread_id: str, run_id: str, body: RunInputRequest, r
|
||||
run = await manager.get(run_id, user_id=user_id)
|
||||
if run is None or run.thread_id != thread_id:
|
||||
raise HTTPException(404, "Run not found")
|
||||
if get_run_context(request).agent_runs is None:
|
||||
if get_agent_runs(request) is None:
|
||||
raise HTTPException(503, "Run inputs require an authenticated session")
|
||||
row = await _store(request).get(run_id, user_id=user_id)
|
||||
if row is None or row.get("run_inputs") is None:
|
||||
@ -178,7 +178,7 @@ async def change_run_input(thread_id: str, run_id: str, input_id: str, request:
|
||||
inputs = RunInputs(_store(request), run_id, user_id)
|
||||
try:
|
||||
if retry:
|
||||
handle = get_run_context(request).agent_runs
|
||||
handle = get_agent_runs(request)
|
||||
if handle is None:
|
||||
raise HTTPException(503, "Run inputs require an authenticated session")
|
||||
from deerflow.runtime.run_origin import admit_origin
|
||||
|
||||
@ -55,7 +55,7 @@ from deerflow.agents.middlewares.tool_receipt import TOOL_RECEIPT_KEY, TOOL_RECE
|
||||
from deerflow.agents.middlewares.tool_transform_meta import TOOL_TRANSFORMS_KEY
|
||||
from deerflow.agents.middlewares.view_image_middleware import _IMAGE_CONTEXT_MESSAGE_MARKER_KEY
|
||||
from deerflow.config.agents_config import load_agent_config
|
||||
from deerflow.config.app_config import get_app_config
|
||||
from deerflow.config.app_config import AppConfig, get_app_config
|
||||
from deerflow.config.database_config import resolve_checkpoint_graph_cache_max
|
||||
from deerflow.knowledge_scope import KNOWLEDGE_SCOPE_KEY, KNOWLEDGE_SCOPE_RUNTIME_KEY
|
||||
from deerflow.mcp_scope import (
|
||||
@ -1403,10 +1403,11 @@ _DEFAULT_RECURSION_LIMIT = 100
|
||||
_DEFAULT_MAX_RECURSION_LIMIT = 1000
|
||||
|
||||
|
||||
def _resolve_gateway_recursion_limits() -> tuple[int, int]:
|
||||
def _resolve_gateway_recursion_limits(app_config: AppConfig | None = None) -> tuple[int, int]:
|
||||
"""Resolve the run default and ceiling from one hot-reloaded snapshot."""
|
||||
try:
|
||||
app_config = get_app_config()
|
||||
if app_config is None:
|
||||
app_config = get_app_config()
|
||||
raw = app_config.recursion_limit
|
||||
max_limit = app_config.max_recursion_limit
|
||||
if raw > max_limit:
|
||||
@ -1517,6 +1518,7 @@ def build_run_config(
|
||||
metadata: dict[str, Any] | None,
|
||||
*,
|
||||
assistant_id: str | None = None,
|
||||
app_config: AppConfig | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Build a RunnableConfig dict for the agent.
|
||||
|
||||
@ -1534,13 +1536,16 @@ def build_run_config(
|
||||
This mirrors the channel manager's ``_resolve_run_params`` logic. IM
|
||||
channel runs send no ``recursion_limit`` unless their session config or
|
||||
channel policy sets one, so they get the same configured default here.
|
||||
|
||||
*app_config* is the snapshot the recursion limits come from; without it
|
||||
they are read through ``get_app_config()``, which does file IO.
|
||||
"""
|
||||
# Lead-agent recursion budget (LangGraph super-steps for the lead graph
|
||||
# only). Independent of subagent depth: a `task()` dispatch runs the whole
|
||||
# subagent inside ONE lead tools-node step, and subagents enforce their own
|
||||
# limit via `subagents.max_turns`. Do not conflate this budget with the
|
||||
# general-purpose subagent's max_turns.
|
||||
default_recursion_limit, max_recursion_limit = _resolve_gateway_recursion_limits()
|
||||
default_recursion_limit, max_recursion_limit = _resolve_gateway_recursion_limits(app_config)
|
||||
config: dict[str, Any] = {"recursion_limit": default_recursion_limit}
|
||||
if request_config:
|
||||
caller_configurable = request_config.get("configurable")
|
||||
@ -2383,7 +2388,11 @@ async def start_run(
|
||||
stream_modes = normalize_stream_modes(body.stream_mode)
|
||||
bridge = get_stream_bridge(request)
|
||||
run_mgr = get_run_manager(request)
|
||||
run_ctx = get_run_context(request)
|
||||
# Building the context loads the live AppConfig, which stats and hashes
|
||||
# config.yaml and reparses it after an edit, so it runs in a worker. The
|
||||
# rest of the run start reuses that snapshot.
|
||||
run_ctx = await asyncio.to_thread(get_run_context, request)
|
||||
app_config = run_ctx.app_config
|
||||
admission_run_id: str | None = None
|
||||
producer_lease_started = False
|
||||
producer_lease_handed_off = False
|
||||
@ -2399,7 +2408,6 @@ async def start_run(
|
||||
|
||||
# Validate model against the allowlist when a model_name is provided.
|
||||
if model_name:
|
||||
app_config = get_app_config()
|
||||
resolved = app_config.get_model_config(model_name)
|
||||
if resolved is None:
|
||||
raise HTTPException(
|
||||
@ -2488,7 +2496,7 @@ async def start_run(
|
||||
thread_id=thread_id,
|
||||
accessor=accessor,
|
||||
read_config=read_config,
|
||||
enabled=get_app_config().hitl_approval.enabled,
|
||||
enabled=app_config.hitl_approval.enabled,
|
||||
body=body,
|
||||
)
|
||||
resume_map = getattr(validated_approval_request_ids, "resume_map", None)
|
||||
@ -2526,7 +2534,7 @@ async def start_run(
|
||||
run_metadata[DEERFLOW_TRACE_METADATA_KEY] = ensure_trace_id()
|
||||
|
||||
try:
|
||||
config = build_run_config(thread_id, body.config, run_metadata, assistant_id=body.assistant_id)
|
||||
config = build_run_config(thread_id, body.config, run_metadata, assistant_id=body.assistant_id, app_config=app_config)
|
||||
except ValueError as exc:
|
||||
# Raised only for the caller's own config sections or assistant_id.
|
||||
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
||||
@ -2545,7 +2553,7 @@ async def start_run(
|
||||
from deerflow.approvals.policy import resolve_mode
|
||||
|
||||
try:
|
||||
resolve_mode(get_app_config().hitl_approval, config.get("context") or {})
|
||||
resolve_mode(app_config.hitl_approval, config.get("context") or {})
|
||||
except (ValueError, TypeError) as exc:
|
||||
raise HTTPException(422, detail={"code": "invalid_approval_mode", "message": "The selected tool approval mode is not allowed"}) from exc
|
||||
if not is_internal_caller:
|
||||
@ -2653,7 +2661,7 @@ async def start_run(
|
||||
user_id=owner_user_id or (str(user.id) if user is not None else None),
|
||||
run_context=run_ctx,
|
||||
run_manager=run_mgr,
|
||||
app_config=get_app_config(),
|
||||
app_config=app_config,
|
||||
)
|
||||
reader, source_ids = prepared
|
||||
run_ctx = replace(run_ctx, conversation_reader=reader)
|
||||
@ -3146,7 +3154,7 @@ async def launch_scheduled_thread_run(
|
||||
input=graph_input,
|
||||
command=None,
|
||||
metadata=metadata or {},
|
||||
config={"recursion_limit": _resolve_scheduler_recursion_limit()},
|
||||
config={"recursion_limit": await asyncio.to_thread(_resolve_scheduler_recursion_limit)},
|
||||
# ``user_id`` mirrors what IM channels put in ``body.context`` so
|
||||
# runtime-context consumers without a ContextVar fallback (e.g.
|
||||
# user-scoped GuardrailMiddleware providers) see the owning user;
|
||||
|
||||
@ -33,6 +33,10 @@ When you add or rename a field, update `config.example.yaml` and raise its
|
||||
- **Read config through `get_app_config()` on each request.** It reloads when the
|
||||
resolved path or the content signature changes, so an edit applies to the next
|
||||
message. Do not cache `AppConfig` on `app.state` or in a module global.
|
||||
- **Async request paths read it in a worker.** Each read stats and hashes
|
||||
`config.yaml` and `extensions_config.json`. Gateway authorization helpers run
|
||||
through `asyncio.to_thread`; `start_run` builds `get_run_context()` in a worker
|
||||
and reuses that one snapshot.
|
||||
- **Startup-only fields are registered.** `reload_boundary.py::STARTUP_ONLY_FIELDS`
|
||||
is the list, and each such field's description starts with `startup-only:`. A
|
||||
new field that is read once at startup joins both.
|
||||
|
||||
@ -0,0 +1,170 @@
|
||||
"""Regression anchors: route-scoped authorization must read the config off the loop.
|
||||
|
||||
The model, skill, suggestion and sandbox gates re-read the hot-reloaded
|
||||
``authorization`` section through ``get_app_config()``, which stats and hashes
|
||||
``config.yaml`` and ``extensions_config.json`` on every call and reparses them
|
||||
after an edit. Async routes therefore call those synchronous helpers in a
|
||||
worker, as ``/api/models/resolve`` already did.
|
||||
|
||||
Each case runs against real config files with the built-in RBAC provider and
|
||||
asserts a decision only the policy on disk produces, so an anchor cannot pass
|
||||
because the route stopped reading the config.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from collections.abc import AsyncIterator
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from blockbuster import BlockingError
|
||||
from fastapi import FastAPI, Request
|
||||
|
||||
from app.gateway import authz
|
||||
from app.gateway.routers import models, skills, suggestions
|
||||
from deerflow.config.app_config import get_app_config, reset_app_config
|
||||
from deerflow.config.extensions_config import reset_extensions_config
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
_CONFIG_YAML = """sandbox:
|
||||
use: deerflow.sandbox.local:LocalSandboxProvider
|
||||
models:
|
||||
- name: allowed
|
||||
use: langchain_openai:ChatOpenAI
|
||||
model: allowed
|
||||
- name: restricted
|
||||
use: langchain_openai:ChatOpenAI
|
||||
model: restricted
|
||||
authorization:
|
||||
enabled: true
|
||||
fail_closed: true
|
||||
default_role: user
|
||||
provider:
|
||||
use: deerflow.authz.rbac:RbacAuthorizationProvider
|
||||
config:
|
||||
roles:
|
||||
user:
|
||||
models:
|
||||
allow: ["allowed"]
|
||||
skills:
|
||||
allow: ["visible-skill"]
|
||||
sandbox:
|
||||
allow: false
|
||||
"""
|
||||
|
||||
_USER = SimpleNamespace(id="user-123", system_role="user", oauth_provider=None, oauth_id=None)
|
||||
|
||||
|
||||
def _write_configs(directory: Path) -> tuple[Path, Path]:
|
||||
# A self-contained public catalog: one skill the policy allows, one it hides.
|
||||
skills_root = directory / "skills"
|
||||
for name in ("visible-skill", "hidden-skill"):
|
||||
skill_dir = skills_root / "public" / name
|
||||
skill_dir.mkdir(parents=True)
|
||||
(skill_dir / "SKILL.md").write_text(f"---\nname: {name}\ndescription: Route authorization anchor skill\n---\n\n# {name}\n", encoding="utf-8")
|
||||
config_path = directory / "config.yaml"
|
||||
extensions_path = directory / "extensions_config.json"
|
||||
config_path.write_text(_CONFIG_YAML + f"skills:\n path: {json.dumps(str(skills_root))}\n", encoding="utf-8")
|
||||
extensions_path.write_text(json.dumps({"mcpServers": {}, "skills": {}}), encoding="utf-8")
|
||||
return config_path, extensions_path
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def app(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[FastAPI]:
|
||||
config_path, extensions_path = await asyncio.to_thread(_write_configs, tmp_path)
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path))
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
|
||||
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path / "home"))
|
||||
monkeypatch.setattr(authz, "_route_provider_cache", {})
|
||||
monkeypatch.setattr(authz, "_route_provider_config_id", None)
|
||||
monkeypatch.setattr(authz, "_route_provider_config_sig", None)
|
||||
reset_app_config()
|
||||
reset_extensions_config()
|
||||
# Load once off the loop: each request then pays only the per-call
|
||||
# revalidation, which is what every gated request carries.
|
||||
await asyncio.to_thread(get_app_config)
|
||||
|
||||
application = FastAPI()
|
||||
|
||||
@application.middleware("http")
|
||||
async def authenticate(request: Request, call_next):
|
||||
request.state.user = _USER
|
||||
request.state.auth_source = "session"
|
||||
request.state.auth = authz.AuthContext(_USER, ["threads:read"])
|
||||
return await call_next(request)
|
||||
|
||||
application.state.thread_store = SimpleNamespace(check_access=AsyncMock(return_value=True))
|
||||
application.include_router(models.router)
|
||||
application.include_router(skills.router)
|
||||
application.include_router(suggestions.router)
|
||||
try:
|
||||
yield application
|
||||
finally:
|
||||
reset_app_config()
|
||||
reset_extensions_config()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(app: FastAPI) -> AsyncIterator[httpx.AsyncClient]:
|
||||
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://gateway.test") as http_client:
|
||||
yield http_client
|
||||
|
||||
|
||||
async def test_model_list_reads_authorization_off_the_loop(client: httpx.AsyncClient) -> None:
|
||||
response = await client.get("/api/models")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert [model["name"] for model in response.json()["models"]] == ["allowed"]
|
||||
|
||||
|
||||
async def test_model_details_read_authorization_off_the_loop(client: httpx.AsyncClient) -> None:
|
||||
assert (await client.get("/api/models/allowed")).status_code == 200
|
||||
assert (await client.get("/api/models/restricted")).status_code == 403
|
||||
|
||||
|
||||
async def test_skill_list_reads_authorization_off_the_loop(client: httpx.AsyncClient) -> None:
|
||||
response = await client.get("/api/skills")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert [skill["name"] for skill in response.json()["skills"]] == ["visible-skill"]
|
||||
|
||||
|
||||
async def test_suggestions_read_model_authorization_off_the_loop(client: httpx.AsyncClient) -> None:
|
||||
# The denial comes before any model is created or invoked.
|
||||
response = await client.post(
|
||||
"/api/threads/thread-1/suggestions",
|
||||
json={"messages": [{"role": "user", "content": "Hello"}], "model_name": "restricted"},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
async def test_sandbox_gate_reads_authorization_off_the_loop(app: FastAPI) -> None:
|
||||
request = Request({"type": "http", "app": app, "headers": [], "state": {}})
|
||||
request.state.user = _USER
|
||||
request.state.auth_source = "session"
|
||||
sandbox_provider = MagicMock()
|
||||
|
||||
lease = await authz.try_acquire_sandbox_for_request(
|
||||
request,
|
||||
sandbox_provider,
|
||||
"thread-1",
|
||||
user_id=_USER.id,
|
||||
app_config=await asyncio.to_thread(get_app_config),
|
||||
)
|
||||
|
||||
assert lease.denied is True
|
||||
sandbox_provider.assert_not_called()
|
||||
|
||||
|
||||
async def test_blocking_probe_inline_scoped_authorization_actually_trips_the_gate(app: FastAPI) -> None:
|
||||
"""Meta-check: resolving the same authorization inline on the loop is caught."""
|
||||
with pytest.raises(BlockingError):
|
||||
authz.resolve_model_authorization(_USER, is_internal=False)
|
||||
150
backend/tests/blocking_io/test_run_start_config_reads.py
Normal file
150
backend/tests/blocking_io/test_run_start_config_reads.py
Normal file
@ -0,0 +1,150 @@
|
||||
"""Regression anchors: starting a run must read the live config off the loop.
|
||||
|
||||
``start_run`` needs the hot-reloaded ``AppConfig`` for the run context, the
|
||||
recursion limits, the model allowlist and the approval switch. Each inline
|
||||
``get_app_config()`` stats and hashes ``config.yaml`` and
|
||||
``extensions_config.json`` and reparses them after an edit, so the run start
|
||||
builds its context, and with it one config snapshot, in a worker and reuses it.
|
||||
Run inputs and scheduled dispatch, which start or feed runs, are covered too.
|
||||
|
||||
The runs go through the real host → ``start_run`` → worker path against real
|
||||
config files, and assert values only the file on disk provides (the recursion
|
||||
limit, the model allowlist), so an anchor cannot pass because the read was
|
||||
skipped.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from blockbuster import BlockingError
|
||||
from deerflow_extension_api.agent_runs import AgentRunError
|
||||
from test_extension_agent_runs import host, runtime # noqa: F401 (fixtures)
|
||||
|
||||
import app.gateway.services as services # the runtime fixture imports it lazily; import before the gate
|
||||
from deerflow.config.app_config import get_app_config, reset_app_config
|
||||
from deerflow.config.extensions_config import reset_extensions_config
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
_CONFIG_YAML = """sandbox:
|
||||
use: deerflow.sandbox.local:LocalSandboxProvider
|
||||
recursion_limit: 37
|
||||
scheduler:
|
||||
recursion_limit: 23
|
||||
models:
|
||||
- name: allowed
|
||||
use: langchain_openai:ChatOpenAI
|
||||
model: allowed
|
||||
"""
|
||||
|
||||
|
||||
def _write_configs(directory: Path) -> tuple[Path, Path]:
|
||||
config_path = directory / "config.yaml"
|
||||
extensions_path = directory / "extensions_config.json"
|
||||
config_path.write_text(_CONFIG_YAML, encoding="utf-8")
|
||||
extensions_path.write_text(json.dumps({"mcpServers": {}, "skills": {}}), encoding="utf-8")
|
||||
return config_path, extensions_path
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def runs(runtime, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): # noqa: F811
|
||||
config_path, extensions_path = await asyncio.to_thread(_write_configs, tmp_path)
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path))
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
|
||||
# Replace the fixture's injected config with the files on disk, and load
|
||||
# them once off the loop: each run start then pays the per-call revalidation.
|
||||
reset_app_config()
|
||||
reset_extensions_config()
|
||||
await asyncio.to_thread(get_app_config)
|
||||
built: list[dict] = []
|
||||
build_run_config = services.build_run_config
|
||||
|
||||
def recording_build_run_config(*args, **kwargs):
|
||||
built.append(build_run_config(*args, **kwargs))
|
||||
return built[-1]
|
||||
|
||||
monkeypatch.setattr(services, "build_run_config", recording_build_run_config)
|
||||
agent_runs, _, _, entered, _ = runtime
|
||||
try:
|
||||
yield agent_runs, built, entered
|
||||
finally:
|
||||
reset_extensions_config()
|
||||
|
||||
|
||||
async def _start(agent_runs, thread_id: str, **kwargs) -> None:
|
||||
thread = await agent_runs.create_thread(thread_id=thread_id)
|
||||
run = await agent_runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "hi"}]}, **kwargs)
|
||||
assert (await agent_runs.wait(thread_id=thread, run_id=run.run_id, timeout=5)).status == "success"
|
||||
|
||||
|
||||
async def test_run_start_reads_the_config_off_the_loop(runs) -> None:
|
||||
agent_runs, built, _ = runs
|
||||
await _start(agent_runs, "config-read-plain")
|
||||
|
||||
assert {config["recursion_limit"] for config in built} == {37}
|
||||
|
||||
|
||||
async def test_run_start_with_a_model_reads_the_allowlist_off_the_loop(runs) -> None:
|
||||
agent_runs, built, _ = runs
|
||||
await _start(agent_runs, "config-read-model", context={"model_name": "allowed"})
|
||||
assert {config["recursion_limit"] for config in built} == {37}
|
||||
|
||||
thread = await agent_runs.create_thread(thread_id="config-read-unknown-model")
|
||||
with pytest.raises(AgentRunError) as rejected:
|
||||
await agent_runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "hi"}]}, context={"model_name": "missing"})
|
||||
assert rejected.value.status_code == 400
|
||||
|
||||
|
||||
async def test_resuming_a_paused_run_reads_the_config_off_the_loop(runs) -> None:
|
||||
agent_runs, built, _ = runs
|
||||
thread = await agent_runs.create_thread(thread_id="config-read-resume")
|
||||
first = await agent_runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "approve"}]})
|
||||
assert (await agent_runs.wait(thread_id=thread, run_id=first.run_id, timeout=5)).status == "waiting_for_input"
|
||||
|
||||
resumed = await agent_runs.resume(thread_id=thread, resume="yes")
|
||||
assert (await agent_runs.wait(thread_id=thread, run_id=resumed.run_id, timeout=5)).status == "success"
|
||||
assert {config["recursion_limit"] for config in built} == {37}
|
||||
|
||||
|
||||
async def test_run_input_submission_does_not_load_the_config_on_the_loop(runs) -> None:
|
||||
agent_runs, _, entered = runs
|
||||
thread = await agent_runs.create_thread(thread_id="config-read-input")
|
||||
run = await agent_runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "wait"}]})
|
||||
await asyncio.wait_for(entered.wait(), timeout=5)
|
||||
|
||||
# Loading the config inline in the capability check surfaces as a 503.
|
||||
accepted = await agent_runs.send_input(thread_id=thread, run_id=run.run_id, text="more", mode="queue", request_id="one")
|
||||
assert accepted["status"] == "pending"
|
||||
assert accepted["target_run_id"] == run.run_id
|
||||
await agent_runs.cancel(thread_id=thread, run_id=run.run_id)
|
||||
|
||||
|
||||
async def test_scheduled_dispatch_reads_the_recursion_limit_off_the_loop(runs, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
captured: list[dict] = []
|
||||
|
||||
async def fake_start_run(body, thread_id, request, **_kwargs):
|
||||
captured.append(body.config)
|
||||
return SimpleNamespace(run_id="run-1", thread_id=thread_id)
|
||||
|
||||
monkeypatch.setattr(services, "start_run", fake_start_run)
|
||||
await services.launch_scheduled_thread_run(
|
||||
thread_id="scheduled-thread",
|
||||
assistant_id="lead_agent",
|
||||
prompt="Run in background",
|
||||
app=SimpleNamespace(state=SimpleNamespace()),
|
||||
owner_user_id="user-1",
|
||||
)
|
||||
|
||||
assert captured == [{"recursion_limit": 23}]
|
||||
|
||||
|
||||
async def test_blocking_probe_inline_run_config_read_actually_trips_the_gate(runs) -> None:
|
||||
"""Meta-check: loading the same config inline on the loop is caught by the gate."""
|
||||
with pytest.raises(BlockingError):
|
||||
get_app_config()
|
||||
@ -107,7 +107,7 @@ def test_all_http_run_entrypoints_reject_before_starting_worker(monkeypatch, pat
|
||||
app.include_router(thread_runs.router)
|
||||
app.state.stream_bridge = SimpleNamespace()
|
||||
app.state.run_manager = SimpleNamespace(create_or_reject=AsyncMock())
|
||||
monkeypatch.setattr(services, "get_run_context", lambda _request: SimpleNamespace(thread_store=app.state.thread_store))
|
||||
monkeypatch.setattr(services, "get_run_context", lambda _request: SimpleNamespace(thread_store=app.state.thread_store, app_config=None))
|
||||
monkeypatch.setattr(services, "resolve_agent_factory", lambda _assistant: object())
|
||||
worker = AsyncMock()
|
||||
monkeypatch.setattr(services, "run_agent", worker)
|
||||
|
||||
@ -260,7 +260,7 @@ def test_run_routes_reject_invalid_goal_before_admission(monkeypatch, path, inva
|
||||
app.include_router(thread_runs.router)
|
||||
app.state.stream_bridge = SimpleNamespace()
|
||||
app.state.run_manager = SimpleNamespace(create_or_reject=AsyncMock())
|
||||
monkeypatch.setattr(services, "get_run_context", lambda _request: SimpleNamespace(thread_store=app.state.thread_store))
|
||||
monkeypatch.setattr(services, "get_run_context", lambda _request: SimpleNamespace(thread_store=app.state.thread_store, app_config=None))
|
||||
factory = Mock(side_effect=AssertionError("invalid goal reached graph assembly"))
|
||||
monkeypatch.setattr(services, "resolve_agent_factory", factory)
|
||||
worker = AsyncMock()
|
||||
|
||||
@ -302,7 +302,7 @@ async def test_http_input_admission_and_status_are_owner_scoped(input_store, mon
|
||||
app.include_router(router)
|
||||
app.state.run_manager = RunManager(store=input_store)
|
||||
await RunInputs(input_store, "run", USER_ID, "worker").reopen()
|
||||
monkeypatch.setattr("app.gateway.run_inputs.get_run_context", lambda _: SimpleNamespace(agent_runs=object()))
|
||||
monkeypatch.setattr("app.gateway.run_inputs.get_agent_runs", lambda _: object())
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||||
url = "/api/threads/thread/runs/run/inputs"
|
||||
body = {"request_id": "one", "text": "Add the appendix", "mode": "queue"}
|
||||
|
||||
@ -145,7 +145,7 @@ def test_all_external_run_entrypoints_reject_sandbox_before_worker(monkeypatch,
|
||||
app.include_router(thread_runs.router)
|
||||
app.state.stream_bridge = SimpleNamespace()
|
||||
app.state.run_manager = SimpleNamespace(create_or_reject=AsyncMock())
|
||||
monkeypatch.setattr(services, "get_run_context", lambda _request: SimpleNamespace(thread_store=app.state.thread_store))
|
||||
monkeypatch.setattr(services, "get_run_context", lambda _request: SimpleNamespace(thread_store=app.state.thread_store, app_config=None))
|
||||
monkeypatch.setattr(services, "resolve_agent_factory", lambda _assistant: object())
|
||||
worker = AsyncMock()
|
||||
monkeypatch.setattr(services, "run_agent", worker)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user