From 1804b7c5227b7a6b51c0c192331119ee6dcf1031 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Sat, 14 Nov 2026 18:33:41 +0900 Subject: [PATCH] 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. --- backend/app/gateway/authz.py | 7 +- backend/app/gateway/deps.py | 15 +- backend/app/gateway/routers/models.py | 4 +- backend/app/gateway/routers/skills.py | 2 +- backend/app/gateway/routers/suggestions.py | 3 +- backend/app/gateway/run_inputs.py | 6 +- backend/app/gateway/services.py | 30 ++-- .../harness/deerflow/config/AGENTS.md | 4 + ...route_scoped_authorization_config_reads.py | 170 ++++++++++++++++++ .../test_run_start_config_reads.py | 150 ++++++++++++++++ .../test_external_system_message_boundary.py | 2 +- backend/tests/test_goal_input_validation.py | 2 +- backend/tests/test_run_inputs.py | 2 +- .../tests/test_sandbox_reference_ownership.py | 2 +- 14 files changed, 372 insertions(+), 27 deletions(-) create mode 100644 backend/tests/blocking_io/test_route_scoped_authorization_config_reads.py create mode 100644 backend/tests/blocking_io/test_run_start_config_reads.py diff --git a/backend/app/gateway/authz.py b/backend/app/gateway/authz.py index 47a4f1eef4..be105466dd 100644 --- a/backend/app/gateway/authz.py +++ b/backend/app/gateway/authz.py @@ -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( diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index c7e420dbaa..989d4cee38 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -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( diff --git a/backend/app/gateway/routers/models.py b/backend/app/gateway/routers/models.py index 50f57d2cb0..f3103e2b29 100644 --- a/backend/app/gateway/routers/models.py +++ b/backend/app/gateway/routers/models.py @@ -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) diff --git a/backend/app/gateway/routers/skills.py b/backend/app/gateway/routers/skills.py index 631e826ac9..9a18c79669 100644 --- a/backend/app/gateway/routers/skills.py +++ b/backend/app/gateway/routers/skills.py @@ -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 diff --git a/backend/app/gateway/routers/suggestions.py b/backend/app/gateway/routers/suggestions.py index 6607babc5d..9322f28655 100644 --- a/backend/app/gateway/routers/suggestions.py +++ b/backend/app/gateway/routers/suggestions.py @@ -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" diff --git a/backend/app/gateway/run_inputs.py b/backend/app/gateway/run_inputs.py index 9db0a04d6b..236a9450ff 100644 --- a/backend/app/gateway/run_inputs.py +++ b/backend/app/gateway/run_inputs.py @@ -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 diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 46baa35c3c..1c0fd7f10e 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -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; diff --git a/backend/packages/harness/deerflow/config/AGENTS.md b/backend/packages/harness/deerflow/config/AGENTS.md index b9cce49ee1..7d2146dc34 100644 --- a/backend/packages/harness/deerflow/config/AGENTS.md +++ b/backend/packages/harness/deerflow/config/AGENTS.md @@ -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. diff --git a/backend/tests/blocking_io/test_route_scoped_authorization_config_reads.py b/backend/tests/blocking_io/test_route_scoped_authorization_config_reads.py new file mode 100644 index 0000000000..e089ff8047 --- /dev/null +++ b/backend/tests/blocking_io/test_route_scoped_authorization_config_reads.py @@ -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) diff --git a/backend/tests/blocking_io/test_run_start_config_reads.py b/backend/tests/blocking_io/test_run_start_config_reads.py new file mode 100644 index 0000000000..9aca0792db --- /dev/null +++ b/backend/tests/blocking_io/test_run_start_config_reads.py @@ -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() diff --git a/backend/tests/test_external_system_message_boundary.py b/backend/tests/test_external_system_message_boundary.py index 4cb798ac4c..1ecbb95a76 100644 --- a/backend/tests/test_external_system_message_boundary.py +++ b/backend/tests/test_external_system_message_boundary.py @@ -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) diff --git a/backend/tests/test_goal_input_validation.py b/backend/tests/test_goal_input_validation.py index 015743985c..65b66d9b56 100644 --- a/backend/tests/test_goal_input_validation.py +++ b/backend/tests/test_goal_input_validation.py @@ -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() diff --git a/backend/tests/test_run_inputs.py b/backend/tests/test_run_inputs.py index c313c64234..44e1d77664 100644 --- a/backend/tests/test_run_inputs.py +++ b/backend/tests/test_run_inputs.py @@ -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"} diff --git a/backend/tests/test_sandbox_reference_ownership.py b/backend/tests/test_sandbox_reference_ownership.py index a3b36b2018..9885031209 100644 --- a/backend/tests/test_sandbox_reference_ownership.py +++ b/backend/tests/test_sandbox_reference_ownership.py @@ -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)