fix(gateway): enforce run-create authorization on stateless endpoints (#5030)

* fix(gateway): enforce authz on stateless runs

* fix(gateway): guard scheduled run creation

README.md

@@ -905,7 +905,7 @@ uv run python -m deerflow.skills.review.cli ../skills/public/data-analysis --for
 
 Tools follow the same philosophy. DeerFlow comes with a core toolset — web search, web fetch, rendered web capture, file operations, bash execution — and supports custom tools via MCP servers and Python functions. Swap anything. Add anything.
 
-Advanced deployments can enable pluggable authorization with `authorization.enabled` in `config.yaml`. A configured `AuthorizationProvider` filters denied tools before they reach the model or deferred-tool catalog, then the same provider is checked again before every business-tool execution through the existing guardrail middleware. Gateway `threads:*` and `runs:*` route permissions are derived from the same provider, while existing owner checks and admin-only management gates remain in force. A generated `tool_search` may bypass the second tool check only when it fronts the current build's already-filtered deferred catalog. Model access follows the same provider: the Gateway `models` list is filtered per principal, `model:use` is enforced on model detail requests and again when the runtime resolves the agent's model, and a denied default model falls back to the first remaining candidate that also passes `model:use`. The built-in RBAC provider supports per-role `tools`, `routes`, `models`, `skills`, and `sandbox` allow/deny policies and validates that `default_role` names a configured role; authorization is disabled by default. See `config.example.yaml` and the [authorization RFC](docs/plans/2026-07-10-pluggable-authorization-rfc.md).
+Advanced deployments can enable pluggable authorization with `authorization.enabled` in `config.yaml`. A configured `AuthorizationProvider` filters denied tools before they reach the model or deferred-tool catalog, then the same provider is checked again before every business-tool execution through the existing guardrail middleware. Gateway `threads:*` and `runs:*` route permissions are derived from the same provider, while existing owner checks and admin-only management gates remain in force. Every HTTP route that starts or enables a future Agent run requires `runs:create`: this includes the stateless `POST /api/runs/stream` and `POST /api/runs/wait` endpoints plus scheduled-task create, update, resume, and manual-trigger mutations. Scheduled-task mutations retain their existing `threads:write` requirement, and the stateless routes separately enforce ownership when the optional thread ID is supplied in the request body. A generated `tool_search` may bypass the second tool check only when it fronts the current build's already-filtered deferred catalog. Model access follows the same provider: the Gateway `models` list is filtered per principal, `model:use` is enforced on model detail requests and again when the runtime resolves the agent's model, and a denied default model falls back to the first remaining candidate that also passes `model:use`. The built-in RBAC provider supports per-role `tools`, `routes`, `models`, `skills`, and `sandbox` allow/deny policies and validates that `default_role` names a configured role; authorization is disabled by default. See `config.example.yaml` and the [authorization RFC](docs/plans/2026-07-10-pluggable-authorization-rfc.md).
 
 Advanced deployments can also extend the agent runtime itself by declaring zero-argument `AgentMiddleware` classes under `extensions.middlewares` in `config.yaml` or `extensions_config.json`. DeerFlow loads the same configured class list into the lead-agent and subagent pipelines after their built-in runtime middlewares and loop/token guards, but before the terminal-response/safety/clarification tail, so enterprise forks can add domain guardrails, tool-call governance, or observability hooks without patching the built-in middleware builders. Missing packages, invalid classes, and broken modules fail loudly at agent creation. Treat `config.yaml` and `extensions_config.json` as trusted operator-controlled files: middleware paths are code execution, just like custom tool, model, sandbox, guardrail, MCP server, and MCP interceptor declarations. Gateway skill/MCP toggle endpoints preserve this field but do not expose an API write path for `extensions.middlewares`. Per-context parameterization and separate lead-only/subagent-only middleware lists are not supported yet.
 

backend/app/gateway/AGENTS.md

@@ -59,7 +59,7 @@ reads/searches.
 | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `<think>` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) |
 | **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, an empty run-event message feed is seeded from an existing checkpoint head so legacy checkpoint-only history receives earlier thread-global sequence numbers and remains visible after the new run; a thread with no checkpoint or an already-populated feed skips this compatibility path. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage`  - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its configured `context_window`.  |
 | **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific |
-| **Runs** (`/api/runs`) | `POST /stream` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id |
+| **Runs** (`/api/runs`) | `POST /stream`, `/wait` - stateless runs requiring `runs:create`; optional body `thread_id` is owner-checked. Scheduled-task create/update/resume/trigger also require `threads:write` plus `runs:create`. `GET /{rid}/messages`, `/feedback` - run messages/feedback |
 | **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. |
 | **GitHub Event-Driven Agents** | Custom agents can declare a `github:` block in their `config.yaml` to bind to repos and event triggers. Webhook fan-out publishes one `InboundMessage` per matching binding to the channel bus; `GitHubChannel` routes those messages through `ChannelManager`. The response `dispatch` summarizes matched/fired/skipped agents. |
 

backend/app/gateway/routers/runs.py

@@ -31,6 +31,7 @@ def _resolve_thread_id(body: RunCreateRequest) -> str:
 
 
 @router.post("/stream")
+@require_permission("runs", "create")
 async def stateless_stream(body: RunCreateRequest, request: Request) -> StreamingResponse:
     """Create a run and stream events via SSE.
 
@@ -56,6 +57,7 @@ async def stateless_stream(body: RunCreateRequest, request: Request) -> Streamin
 
 
 @router.post("/wait", response_model=dict)
+@require_permission("runs", "create")
 async def stateless_wait(body: RunCreateRequest, request: Request) -> dict:
     """Create a run and block until completion.
 

backend/app/gateway/routers/scheduled_tasks.py

@@ -81,6 +81,7 @@ async def list_scheduled_tasks(request: Request):
 
 @router.post("/scheduled-tasks")
 @require_permission("threads", "write")
+@require_permission("runs", "create")
 async def create_scheduled_task(request: Request, body: ScheduledTaskCreateRequest):
     config = get_config()
     repo = get_scheduled_task_repo(request)
@@ -153,6 +154,7 @@ async def get_scheduled_task(task_id: str, request: Request):
 
 @router.patch("/scheduled-tasks/{task_id}")
 @require_permission("threads", "write")
+@require_permission("runs", "create")
 async def update_scheduled_task(task_id: str, request: Request, body: ScheduledTaskUpdateRequest):
     config = get_config()
     repo = get_scheduled_task_repo(request)
@@ -272,6 +274,7 @@ async def pause_scheduled_task(task_id: str, request: Request):
 
 @router.post("/scheduled-tasks/{task_id}/resume")
 @require_permission("threads", "write")
+@require_permission("runs", "create")
 async def resume_scheduled_task(task_id: str, request: Request):
     repo = get_scheduled_task_repo(request)
     user = await get_optional_user_from_request(request)
@@ -300,6 +303,7 @@ async def resume_scheduled_task(task_id: str, request: Request):
 
 @router.post("/scheduled-tasks/{task_id}/trigger")
 @require_permission("threads", "write")
+@require_permission("runs", "create")
 async def trigger_scheduled_task(task_id: str, request: Request):
     repo = get_scheduled_task_repo(request)
     service = get_scheduled_task_service(request)

backend/tests/test_authorization_route_permissions.py

@@ -1,10 +1,10 @@
 """Route-level authorization tests for the Gateway permission decorators."""
 
 from types import SimpleNamespace
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
-from fastapi import FastAPI, Request
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.testclient import TestClient
 
 from app.gateway.auth.models import User
@@ -16,6 +16,7 @@ from app.gateway.authz import (
     require_permission,
     resolve_route_permissions,
 )
+from app.gateway.routers import runs, scheduled_tasks
 from deerflow.authz.provider import AuthzDecision, AuthzReason
 from deerflow.authz.rbac import RbacAuthorizationProvider
 from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig
@@ -268,6 +269,107 @@ def test_auth_middleware_marks_internal_route_principal(monkeypatch):
     assert permission_resolver.await_args.kwargs == {"is_internal": True}
 
 
+_STATELESS_RUN_PATHS = ("/api/runs/stream", "/api/runs/wait")
+
+
+def _enable_auth_disabled_for_route_test(monkeypatch) -> None:
+    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
+    monkeypatch.delenv("DEER_FLOW_ENV", raising=False)
+    monkeypatch.delenv("ENVIRONMENT", raising=False)
+
+
+def _make_stateless_runs_app() -> FastAPI:
+    app = FastAPI()
+    app.add_middleware(AuthMiddleware)
+    app.include_router(runs.router)
+    app.state.stream_bridge = MagicMock()
+    app.state.run_manager = MagicMock()
+    return app
+
+
+@pytest.mark.parametrize("path", _STATELESS_RUN_PATHS)
+def test_stateless_run_creation_requires_runs_create(monkeypatch, path):
+    _enable_auth_disabled_for_route_test(monkeypatch)
+    monkeypatch.setattr(
+        "app.gateway.auth_middleware.resolve_route_permissions",
+        AsyncMock(return_value=[Permissions.RUNS_READ]),
+    )
+    start_run = AsyncMock(side_effect=HTTPException(status_code=418, detail="run creation reached"))
+    monkeypatch.setattr(runs, "start_run", start_run)
+
+    with TestClient(_make_stateless_runs_app()) as client:
+        response = client.post(path, json={})
+
+    assert response.status_code == 403
+    assert response.json() == {"detail": "Permission denied: runs:create"}
+    start_run.assert_not_awaited()
+
+
+@pytest.mark.parametrize("path", _STATELESS_RUN_PATHS)
+def test_stateless_run_creation_allows_runs_create(monkeypatch, path):
+    _enable_auth_disabled_for_route_test(monkeypatch)
+    monkeypatch.setattr(
+        "app.gateway.auth_middleware.resolve_route_permissions",
+        AsyncMock(return_value=[Permissions.RUNS_CREATE]),
+    )
+    start_run = AsyncMock(side_effect=HTTPException(status_code=418, detail="run creation reached"))
+    monkeypatch.setattr(runs, "start_run", start_run)
+
+    with TestClient(_make_stateless_runs_app()) as client:
+        response = client.post(path, json={})
+
+    assert response.status_code == 418
+    assert response.json() == {"detail": "run creation reached"}
+    start_run.assert_awaited_once()
+
+
+_SCHEDULED_RUN_CREATION_REQUESTS = (
+    (
+        "POST",
+        "/api/scheduled-tasks",
+        {
+            "title": "Daily summary",
+            "prompt": "Summarize the latest activity",
+            "schedule_type": "cron",
+            "schedule_spec": {"cron": "0 9 * * *"},
+            "timezone": "UTC",
+        },
+    ),
+    ("PATCH", "/api/scheduled-tasks/task-1", {"title": "Updated summary"}),
+    ("POST", "/api/scheduled-tasks/task-1/resume", None),
+    ("POST", "/api/scheduled-tasks/task-1/trigger", None),
+)
+
+
+def _make_scheduled_tasks_app() -> FastAPI:
+    app = FastAPI()
+    app.add_middleware(AuthMiddleware)
+    app.include_router(scheduled_tasks.router)
+    return app
+
+
+@pytest.mark.parametrize(("method", "path", "payload"), _SCHEDULED_RUN_CREATION_REQUESTS)
+@pytest.mark.parametrize(
+    ("permissions", "denied_permission"),
+    [
+        ([Permissions.THREADS_WRITE], Permissions.RUNS_CREATE),
+        ([Permissions.RUNS_CREATE], Permissions.THREADS_WRITE),
+    ],
+)
+def test_scheduled_run_creation_requires_thread_write_and_runs_create(monkeypatch, method, path, payload, permissions, denied_permission):
+    _enable_auth_disabled_for_route_test(monkeypatch)
+    monkeypatch.setattr(
+        "app.gateway.auth_middleware.resolve_route_permissions",
+        AsyncMock(return_value=permissions),
+    )
+
+    with TestClient(_make_scheduled_tasks_app()) as client:
+        response = client.request(method, path, json=payload)
+
+    assert response.status_code == 403
+    assert response.json() == {"detail": f"Permission denied: {denied_permission}"}
+
+
 # ── Provider cache tests ────────────────────────────────────────────────
 
 

backend/tests/test_run_creation_route_contract.py

@@ -0,0 +1,66 @@
+"""Static authorization contract for HTTP routes that can create Agent runs."""
+
+from __future__ import annotations
+
+import ast
+import re
+from pathlib import Path
+
+ROUTERS_DIR = Path(__file__).resolve().parent.parent / "app" / "gateway" / "routers"
+
+# These scheduled-task mutations create or re-enable work that the background
+# scheduler later launches through the normal Gateway run lifecycle.
+SCHEDULED_RUN_ENABLING_HANDLERS = {
+    "create_scheduled_task",
+    "update_scheduled_task",
+    "resume_scheduled_task",
+}
+
+_ROUTE_DECORATOR_RE = re.compile(r"router\.(get|post|delete|put|patch)")
+
+
+def _is_route_handler(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
+    return any(
+        isinstance(decorator, ast.Call) and isinstance(decorator.func, ast.Attribute) and isinstance(decorator.func.value, ast.Name) and _ROUTE_DECORATOR_RE.fullmatch(f"{decorator.func.value.id}.{decorator.func.attr}")
+        for decorator in node.decorator_list
+    )
+
+
+def _calls_run_launcher(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
+    for child in ast.walk(node):
+        if not isinstance(child, ast.Call):
+            continue
+        if isinstance(child.func, ast.Name) and child.func.id == "start_run":
+            return True
+        if isinstance(child.func, ast.Attribute) and child.func.attr == "dispatch_task":
+            return True
+    return False
+
+
+def _requires_runs_create(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
+    for decorator in node.decorator_list:
+        if not isinstance(decorator, ast.Call):
+            continue
+        if not isinstance(decorator.func, ast.Name) or decorator.func.id != "require_permission":
+            continue
+        if len(decorator.args) < 2:
+            continue
+        resource, action = decorator.args[:2]
+        if isinstance(resource, ast.Constant) and resource.value == "runs" and isinstance(action, ast.Constant) and action.value == "create":
+            return True
+    return False
+
+
+def test_every_run_creation_route_requires_runs_create():
+    violations = []
+    for path in sorted(ROUTERS_DIR.glob("*.py")):
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        for node in ast.walk(tree):
+            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) or not _is_route_handler(node):
+                continue
+            if node.name not in SCHEDULED_RUN_ENABLING_HANDLERS and not _calls_run_launcher(node):
+                continue
+            if not _requires_runs_create(node):
+                violations.append(f"{path.name}:{node.name}")
+
+    assert not violations, "run-creating routes without runs:create:\n" + "\n".join(violations)

backend/tests/test_scheduled_task_router_behavior.py

@@ -4,6 +4,7 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
+from _router_auth_helpers import call_unwrapped
 from fastapi import HTTPException
 
 from app.gateway.routers import scheduled_tasks
@@ -188,7 +189,8 @@ async def test_create_scheduled_task_uses_repo():
         scheduled_tasks.get_config = lambda: config
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
-        created = await scheduled_tasks.create_scheduled_task.__wrapped__(
+        created = await call_unwrapped(
+            scheduled_tasks.create_scheduled_task,
             request=request,
             body=body,
         )
@@ -231,7 +233,8 @@ async def test_create_fresh_thread_task_does_not_require_thread_id():
         scheduled_tasks.get_config = lambda: config
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
-        created = await scheduled_tasks.create_scheduled_task.__wrapped__(
+        created = await call_unwrapped(
+            scheduled_tasks.create_scheduled_task,
             request=request,
             body=body,
         )
@@ -273,7 +276,8 @@ async def test_trigger_scheduled_task_dispatches_manual_run():
         scheduled_tasks.get_scheduled_task_service = lambda _request: service
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
-        result = await scheduled_tasks.trigger_scheduled_task.__wrapped__(
+        result = await call_unwrapped(
+            scheduled_tasks.trigger_scheduled_task,
             task_id=task["id"],
             request=request,
         )
@@ -317,7 +321,8 @@ async def test_trigger_scheduled_task_returns_conflict_when_dispatch_conflicts()
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
         with pytest.raises(Exception) as exc_info:
-            await scheduled_tasks.trigger_scheduled_task.__wrapped__(
+            await call_unwrapped(
+                scheduled_tasks.trigger_scheduled_task,
                 task_id=task["id"],
                 request=request,
             )
@@ -360,7 +365,8 @@ async def test_update_scheduled_task_writes_repo():
         scheduled_tasks.get_config = lambda: config
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
-        result = await scheduled_tasks.update_scheduled_task.__wrapped__(
+        result = await call_unwrapped(
+            scheduled_tasks.update_scheduled_task,
             task_id=task["id"],
             request=request,
             body=scheduled_tasks.ScheduledTaskUpdateRequest(title="Updated title"),
@@ -419,7 +425,8 @@ async def test_update_rechecks_atomic_mutability_after_router_precheck(tmp_path)
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=SimpleNamespace(id="user-1"))
 
         patch_call = asyncio.create_task(
-            scheduled_tasks.update_scheduled_task.__wrapped__(
+            call_unwrapped(
+                scheduled_tasks.update_scheduled_task,
                 task_id=task["id"],
                 request=SimpleNamespace(),
                 body=scheduled_tasks.ScheduledTaskUpdateRequest(prompt="changed after admission"),
@@ -476,7 +483,8 @@ async def test_delete_scheduled_task_deletes_repo_row():
         scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
-        result = await scheduled_tasks.delete_scheduled_task.__wrapped__(
+        result = await call_unwrapped(
+            scheduled_tasks.delete_scheduled_task,
             task_id=task["id"],
             request=request,
         )
@@ -513,12 +521,14 @@ async def test_pause_and_resume_scheduled_task_update_status():
         scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
-        paused = await scheduled_tasks.pause_scheduled_task.__wrapped__(
+        paused = await call_unwrapped(
+            scheduled_tasks.pause_scheduled_task,
             task_id=task["id"],
             request=request,
         )
         paused_status = paused["status"]
-        resumed = await scheduled_tasks.resume_scheduled_task.__wrapped__(
+        resumed = await call_unwrapped(
+            scheduled_tasks.resume_scheduled_task,
             task_id=task["id"],
             request=request,
         )
@@ -555,7 +565,8 @@ async def test_pause_cancels_waiting_occurrence_before_pausing_task():
     try:
         scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
-        result = await scheduled_tasks.pause_scheduled_task.__wrapped__(
+        result = await call_unwrapped(
+            scheduled_tasks.pause_scheduled_task,
             task_id=task["id"],
             request=request,
         )
@@ -593,7 +604,8 @@ async def test_delete_rejects_occurrence_that_has_started_launching():
         scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
         with pytest.raises(Exception) as exc_info:
-            await scheduled_tasks.delete_scheduled_task.__wrapped__(
+            await call_unwrapped(
+                scheduled_tasks.delete_scheduled_task,
                 task_id=task["id"],
                 request=request,
             )
@@ -632,7 +644,8 @@ async def test_pause_rejects_running_task():
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
         with pytest.raises(Exception) as exc_info:
-            await scheduled_tasks.pause_scheduled_task.__wrapped__(
+            await call_unwrapped(
+                scheduled_tasks.pause_scheduled_task,
                 task_id=task["id"],
                 request=request,
             )
@@ -676,7 +689,8 @@ async def test_update_rejects_running_task():
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
         with pytest.raises(Exception) as exc_info:
-            await scheduled_tasks.update_scheduled_task.__wrapped__(
+            await call_unwrapped(
+                scheduled_tasks.update_scheduled_task,
                 task_id=task["id"],
                 request=request,
                 body=scheduled_tasks.ScheduledTaskUpdateRequest(title="Updated title"),
@@ -720,7 +734,8 @@ async def test_update_rejects_queued_task_definition_until_occurrence_finishes()
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
         with pytest.raises(Exception) as exc_info:
-            await scheduled_tasks.update_scheduled_task.__wrapped__(
+            await call_unwrapped(
+                scheduled_tasks.update_scheduled_task,
                 task_id=task["id"],
                 request=request,
                 body=scheduled_tasks.ScheduledTaskUpdateRequest(prompt="Changed while queued"),
@@ -773,7 +788,8 @@ async def test_list_thread_scheduled_tasks_filters_by_thread_id():
         scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
-        result = await scheduled_tasks.list_thread_scheduled_tasks.__wrapped__(
+        result = await call_unwrapped(
+            scheduled_tasks.list_thread_scheduled_tasks,
             thread_id="thread-1",
             request=request,
         )
@@ -825,7 +841,8 @@ async def test_list_scheduled_task_runs_returns_persisted_rows_without_side_effe
         scheduled_tasks.get_scheduled_task_run_repo = lambda _request: run_repo
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
-        result = await scheduled_tasks.list_scheduled_task_runs.__wrapped__(
+        result = await call_unwrapped(
+            scheduled_tasks.list_scheduled_task_runs,
             task_id=task["id"],
             request=request,
         )
@@ -864,7 +881,8 @@ async def test_create_once_task_enforces_minimum_delay():
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
         with pytest.raises(Exception) as exc_info:
-            await scheduled_tasks.create_scheduled_task.__wrapped__(
+            await call_unwrapped(
+                scheduled_tasks.create_scheduled_task,
                 request=request,
                 body=body,
             )
@@ -909,7 +927,8 @@ async def test_update_terminal_once_task_with_future_run_at_rearms_it():
         scheduled_tasks.get_config = lambda: _Config()
         scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
 
-        result = await scheduled_tasks.update_scheduled_task.__wrapped__(
+        result = await call_unwrapped(
+            scheduled_tasks.update_scheduled_task,
             task_id=task["id"],
             request=request,
             body=scheduled_tasks.ScheduledTaskUpdateRequest(schedule_spec={"run_at": future_run_at}),

backend/tests/test_thread_regenerate_prepare.py

@@ -6,6 +6,7 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
 
 import pytest
+from _router_auth_helpers import call_unwrapped
 from fastapi import HTTPException
 from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
 from langgraph.checkpoint.base import empty_checkpoint, uuid6
@@ -251,8 +252,8 @@ def test_run_wait_readers_return_materialized_final_values() -> None:
                 return_value=(accessor, snapshot.config),
             ),
         ):
-            thread_result = await thread_runs.wait_run.__wrapped__("thread-1", body, request)
-            stateless_result = await runs.stateless_wait(body, request)
+            thread_result = await call_unwrapped(thread_runs.wait_run, "thread-1", body, request)
+            stateless_result = await call_unwrapped(runs.stateless_wait, body, request)
         return thread_result, stateless_result
 
     thread_result, stateless_result = asyncio.run(_scenario())
@@ -303,8 +304,8 @@ def test_run_wait_readers_preserve_terminal_error_without_checkpoint() -> None:
                 return_value=(accessor, snapshot.config),
             ),
         ):
-            thread_result = await thread_runs.wait_run.__wrapped__("thread-1", body, request)
-            stateless_result = await runs.stateless_wait(body, request)
+            thread_result = await call_unwrapped(thread_runs.wait_run, "thread-1", body, request)
+            stateless_result = await call_unwrapped(runs.stateless_wait, body, request)
         return thread_result, stateless_result
 
     thread_result, stateless_result = asyncio.run(_scenario())
@@ -340,7 +341,7 @@ def test_run_wait_readers_preserve_terminal_error_when_accessor_builder_fails(ro
                     side_effect=RuntimeError("graph construction failed"),
                 ),
             ):
-                return await thread_runs.wait_run.__wrapped__("thread-1", body, request)
+                return await call_unwrapped(thread_runs.wait_run, "thread-1", body, request)
 
         with (
             patch.object(runs, "get_stream_bridge", return_value=object()),
@@ -352,7 +353,7 @@ def test_run_wait_readers_preserve_terminal_error_when_accessor_builder_fails(ro
                 side_effect=RuntimeError("graph construction failed"),
             ),
         ):
-            return await runs.stateless_wait(body, request)
+            return await call_unwrapped(runs.stateless_wait, body, request)
 
     result = asyncio.run(_scenario())