feat(authz): derive Gateway route permissions from AuthorizationProvider (#4439)

* feat(authz): derive Gateway route permissions from AuthorizationProvider (Phase 2A, #4063)

Phase 2A: replace legacy _ALL_PERMISSIONS with provider-derived route
permissions. When authorization.enabled, each threads:*/runs:* permission
is evaluated independently via provider.aauthorize(resource='route').
Disabled mode preserves legacy behavior. owner_check and require_admin_user
remain unchanged.

10 new tests: disabled/enabled/RBAC policy/fail-closed/fail-open/
authenticate integration/middleware integration.

* fix(authz): move AuthorizationConfig import to runtime for fallback in _get_route_authorization_config

README.md

@@ -763,7 +763,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 tool 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. A generated `tool_search` may bypass that second check only when it fronts the current build's already-filtered deferred catalog. The built-in RBAC provider supports per-role tool 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. A generated `tool_search` may bypass the second tool check only when it fronts the current build's already-filtered deferred catalog. The built-in RBAC provider supports per-role `tools` and `routes` 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/AGENTS.md

@@ -340,6 +340,8 @@ Lead-agent middlewares are assembled in strict order across three functions: the
 
 Authorization identity plumbing is independent of whether authorization enforcement is enabled. Gateway removes client-supplied `is_internal` / `authz_attributes` / `channel_user_id`, derives `is_internal` only from the server-owned `request.state.auth_source`, and accepts `channel_user_id` only from an internally authenticated IM caller's top-level `body.context`; free-form `body.config` can never supply it. `build_principal_from_context` is the shared Principal builder for assembly-time authorization and `GuardrailAuthorizationAdapter`; it applies `default_role`, strict-boolean internal provenance, and copy-on-read `authz_attributes`. The built-in RBAC provider validates `authorization.default_role` during provider resolution so an unknown fallback role fails agent construction instead of degrading into an empty tool set. Task delegation carries `is_internal` plus copied attributes through `SubagentExecutor`, while `GuardrailMiddleware` maps the same runtime fields into `GuardrailRequest`. Phase 1B applies Layer 1 before deferred-tool assembly on the lead, native-subagent, and embedded-client paths, then passes the same provider instance into Layer 2. Framework-provided `describe_skill` and memory tools are included in Layer 1 but restored to their legacy post-`tool_search` ordering afterward. `DeerFlowClient.stream()` treats its in-process caller as trusted and accepts the same identity fields as keyword overrides; it includes the complete Principal in its agent cache key and deep-copies nested attributes so caller mutation cannot make a stale tool set look current.
 
+Gateway route authorization uses `authz.py::resolve_route_permissions()` as the single provider integration point for both `AuthMiddleware` and decorator-only authentication. When enabled, it evaluates the six registered `threads:*` / `runs:*` permissions as `resource="route"` requests whose targets are the full `resource:action` strings. Decisions use the async provider API and are cached for the request in `AuthContext`; decorators do not call the provider again. Provider resolution or decision errors follow `authorization.fail_closed`, scoped per permission for decision errors. When authorization is disabled, the legacy complete permission set is returned without resolving a provider. Existing `owner_check` enforcement and `require_admin_user()` management gates remain independent and unchanged. Tests: `tests/test_authorization_route_permissions.py`, `tests/test_auth.py`, and `tests/test_auth_middleware.py`.
+
 Before changing a later authorization phase, read the [authorization RFC](../docs/plans/2026-07-10-pluggable-authorization-rfc.md) and its [implementation notes](../docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md). The notes are the cumulative handoff record for merged PR behavior, reviewer feedback, trust-boundary decisions, deferred scope, and required regression coverage.
 
 **Lead-only middlewares** (`build_middlewares`, appended after the base):

backend/app/gateway/auth_middleware.py

@@ -24,7 +24,7 @@ from app.gateway.auth_disabled import (
     get_auth_disabled_user,
     is_auth_disabled,
 )
-from app.gateway.authz import _ALL_PERMISSIONS, AuthContext
+from app.gateway.authz import AuthContext, resolve_route_permissions
 from app.gateway.internal_auth import INTERNAL_AUTH_HEADER_NAME, get_internal_user, is_valid_internal_auth_token
 from deerflow.runtime.user_context import reset_current_user, set_current_user
 
@@ -151,7 +151,11 @@ class AuthMiddleware(BaseHTTPMiddleware):
         # JWT-decode + DB-lookup pipeline a second time per request).
         request.state.user = user
         request.state.auth_source = auth_source
-        request.state.auth = AuthContext(user=user, permissions=_ALL_PERMISSIONS)
+        permissions = await resolve_route_permissions(
+            user,
+            is_internal=auth_source == AUTH_SOURCE_INTERNAL,
+        )
+        request.state.auth = AuthContext(user=user, permissions=permissions)
         token = set_current_user(user)
         try:
             return await call_next(request)

backend/app/gateway/authz.py

@@ -29,19 +29,27 @@ Inspired by LangGraph Auth system: https://github.com/langchain-ai/langgraph/blo
 
 from __future__ import annotations
 
+import asyncio
 import functools
 import inspect
+import logging
 from collections.abc import Callable
 from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
 
 from fastapi import HTTPException, Request
 
+from deerflow.authz.principal import build_principal_from_context
+from deerflow.authz.provider import AuthorizationProvider, AuthzDecision, AuthzRequest
+from deerflow.authz.runtime import resolve_authorization_provider
+from deerflow.config.authorization_config import AuthorizationConfig
+
 if TYPE_CHECKING:
     from app.gateway.auth.models import User
 
 P = ParamSpec("P")
 T = TypeVar("T")
+logger = logging.getLogger(__name__)
 
 
 # Permission constants
@@ -128,6 +136,133 @@ def _make_test_request_stub() -> Any:
     return SimpleNamespace(state=SimpleNamespace(), cookies={}, _deerflow_test_bypass_auth=True)
 
 
+def _get_route_authorization_config() -> AuthorizationConfig:
+    """Return the hot-reloaded authorization config for this request.
+
+    Falls back to a disabled config when AppConfig is not available (e.g. test
+    environments without a config.yaml), preserving legacy all-permissions behavior.
+    """
+    from deerflow.config.app_config import get_app_config
+
+    try:
+        return get_app_config().authorization
+    except (FileNotFoundError, RuntimeError):
+        return AuthorizationConfig()
+
+
+# --- Provider cache (W1/F1) ---
+# Keyed by config object identity (id) so the expensive model_dump() signature
+# is only recomputed when get_app_config() returns a new object (hot-reload).
+_route_provider_cache: dict[str, AuthorizationProvider] = {}
+_route_provider_config_id: int | None = None
+_route_provider_config_sig: str | None = None
+
+
+def _get_cached_route_provider(config: AuthorizationConfig) -> AuthorizationProvider | None:
+    """Resolve (or reuse) the authorization provider for route permissions.
+
+    The provider is cached per config object identity. When ``get_app_config()``
+    returns a new object (hot-reload), the signature is recomputed and compared;
+    only an actual content change triggers re-resolution. This avoids calling
+    ``model_dump()`` on every request — the fast path is a single ``id()`` check.
+    """
+    global _route_provider_config_id, _route_provider_config_sig, _route_provider_cache
+
+    config_id = id(config)
+
+    # Fast path: same config object as last time → return cached provider.
+    if config_id == _route_provider_config_id and _route_provider_cache:
+        return _route_provider_cache.get("provider")
+
+    # Config object changed (hot-reload): compute signature to check if
+    # content actually changed or just the wrapper object identity.
+    sig = repr(sorted(config.model_dump().items()))
+    if sig == _route_provider_config_sig and _route_provider_cache:
+        # Same content, different object — update id, reuse provider.
+        _route_provider_config_id = config_id
+        return _route_provider_cache.get("provider")
+
+    # Content changed (or first call): re-resolve into a local first,
+    # then publish id + sig + provider together to avoid a race window.
+    _route_provider_cache.clear()
+
+    provider = resolve_authorization_provider(config)
+    if provider is not None:
+        _route_provider_cache["provider"] = provider
+    _route_provider_config_id = config_id
+    _route_provider_config_sig = sig
+    return provider
+
+
+async def resolve_route_permissions(user: User, *, is_internal: bool) -> list[str]:
+    """Return the route permissions granted to an authenticated user.
+
+    Disabled authorization preserves the legacy all-permissions behavior.
+    When enabled, every registered ``resource:action`` permission is evaluated
+    independently so a provider failure affects only the route being checked.
+    Provider instances are cached per config signature (hot-reload safe).
+    """
+    config = _get_route_authorization_config()
+    if config.enabled is not True:
+        return list(_ALL_PERMISSIONS)
+
+    try:
+        provider = _get_cached_route_provider(config)
+        if provider is None:
+            raise ValueError("authorization is enabled but provider resolution returned None")
+    except Exception:
+        logger.warning("Failed to resolve authorization provider for Gateway routes", exc_info=True)
+        return [] if config.fail_closed else list(_ALL_PERMISSIONS)
+
+    # Align with Phase 1B's tool path: internal callers (IM channel workers,
+    # scheduler) have system_role="internal", which is not a real RBAC role.
+    # Omit it so default_role applies, mirroring inject_authenticated_user_context
+    # which pops user_role for internal callers without a resolved owner.
+    from app.gateway.internal_auth import INTERNAL_SYSTEM_ROLE
+
+    user_role = getattr(user, "system_role", None)
+    if user_role == INTERNAL_SYSTEM_ROLE:
+        user_role = None
+
+    principal = build_principal_from_context(
+        {
+            "user_id": str(user.id),
+            "user_role": user_role,
+            "oauth_provider": getattr(user, "oauth_provider", None),
+            "oauth_id": getattr(user, "oauth_id", None),
+            "is_internal": is_internal,
+        },
+        default_role=config.default_role,
+    )
+
+    # Evaluate all permissions in parallel (W2).
+    async def _evaluate(permission: str) -> str | None:
+        _, action = permission.split(":", maxsplit=1)
+        request = AuthzRequest(
+            principal=principal,
+            resource="route",
+            action=action,
+            target=permission,
+        )
+        try:
+            decision = await provider.aauthorize(request)
+            if not isinstance(decision, AuthzDecision):
+                raise TypeError("AuthorizationProvider.aauthorize must return AuthzDecision")
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.warning(
+                "Authorization provider failed while evaluating route permission %s",
+                permission,
+                exc_info=True,
+            )
+            return permission if not config.fail_closed else None
+        return permission if decision.allow else None
+
+    results = await asyncio.gather(*[_evaluate(p) for p in _ALL_PERMISSIONS])
+    return [p for p in results if p is not None]
+
+
 async def _authenticate(request: Request) -> AuthContext:
     """Authenticate request and return AuthContext.
 
@@ -140,8 +275,32 @@ async def _authenticate(request: Request) -> AuthContext:
     if user is None:
         return AuthContext(user=None, permissions=[])
 
-    # In future, permissions could be stored in user record
-    return AuthContext(user=user, permissions=_ALL_PERMISSIONS)
+    is_internal = _is_internal_caller(request, user)
+    permissions = await resolve_route_permissions(user, is_internal=is_internal)
+    return AuthContext(user=user, permissions=permissions)
+
+
+def _is_internal_caller(request: Request, user: Any) -> bool:
+    """Determine if the request originates from a trusted internal caller.
+
+    Checks three signals (any one suffices):
+    1. ``request.state.auth_source == AUTH_SOURCE_INTERNAL`` (set by AuthMiddleware).
+    2. ``user.system_role == INTERNAL_SYSTEM_ROLE`` (synthetic internal user).
+    3. The request carries a valid internal auth token header (decorator-only path
+       where AuthMiddleware may not have stamped ``auth_source`` yet).
+    """
+    from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL
+    from app.gateway.internal_auth import INTERNAL_AUTH_HEADER_NAME, INTERNAL_SYSTEM_ROLE, is_valid_internal_auth_token
+
+    if getattr(getattr(request, "state", None), "auth_source", None) == AUTH_SOURCE_INTERNAL:
+        return True
+    if getattr(user, "system_role", None) == INTERNAL_SYSTEM_ROLE:
+        return True
+    # Decorator-only path: check the internal token header directly.
+    internal_token = request.headers.get(INTERNAL_AUTH_HEADER_NAME) if hasattr(request, "headers") else None
+    if internal_token and is_valid_internal_auth_token(internal_token):
+        return True
+    return False
 
 
 def require_auth[**P, T](func: Callable[P, T]) -> Callable[P, T]:

backend/tests/test_auth_middleware.py

@@ -5,6 +5,17 @@ from starlette.testclient import TestClient
 
 from app.gateway.auth_middleware import AuthMiddleware, _is_public
 from app.gateway.csrf_middleware import CSRFMiddleware
+from deerflow.config.authorization_config import AuthorizationConfig
+
+
+@pytest.fixture(autouse=True)
+def _default_route_authorization_config(monkeypatch):
+    """Keep minimal middleware apps independent of a repository config.yaml."""
+    monkeypatch.setattr(
+        "app.gateway.authz._get_route_authorization_config",
+        lambda: AuthorizationConfig(),
+    )
+
 
 # ── _is_public unit tests ─────────────────────────────────────────────────
 

backend/tests/test_authorization_route_permissions.py

@@ -0,0 +1,355 @@
+"""Route-level authorization tests for the Gateway permission decorators."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from fastapi import FastAPI, Request
+from fastapi.testclient import TestClient
+
+from app.gateway.auth.models import User
+from app.gateway.auth_middleware import AuthMiddleware
+from app.gateway.authz import (
+    Permissions,
+    _authenticate,
+    _get_cached_route_provider,
+    require_permission,
+    resolve_route_permissions,
+)
+from deerflow.authz.provider import AuthzDecision, AuthzReason
+from deerflow.authz.rbac import RbacAuthorizationProvider
+from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig
+
+
+class _RecordingProvider:
+    name = "recording"
+
+    def __init__(
+        self,
+        *,
+        denied: set[str] | None = None,
+        errors: set[str] | None = None,
+    ) -> None:
+        self.denied = denied or set()
+        self.errors = errors or set()
+        self.requests = []
+
+    def authorize(self, request):
+        raise AssertionError("route authorization must use the async provider API")
+
+    async def aauthorize(self, request):
+        self.requests.append(request)
+        if request.target in self.errors:
+            raise RuntimeError(f"provider failed for {request.target}")
+        allowed = request.target not in self.denied
+        return AuthzDecision(
+            allow=allowed,
+            reasons=[AuthzReason(code="authz.allowed" if allowed else "authz.denied")],
+        )
+
+    def filter_resources(self, principal, resource_type, candidates):
+        raise AssertionError("route authorization must preserve per-action requests")
+
+
+def _user(**overrides):
+    values = {
+        "id": "user-123",
+        "system_role": "user",
+        "oauth_provider": "github",
+        "oauth_id": "oauth-456",
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+def _enable_authorization(monkeypatch, provider, *, fail_closed: bool = True) -> None:
+    config = AuthorizationConfig(
+        enabled=True,
+        fail_closed=fail_closed,
+        default_role="user",
+    )
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+    # Bypass the provider cache so each test gets its own provider instance.
+    monkeypatch.setattr("app.gateway.authz._get_cached_route_provider", lambda c: provider)
+
+
+@pytest.mark.asyncio
+async def test_route_permissions_disabled_preserves_all_permissions(monkeypatch):
+    config = AuthorizationConfig(enabled=False)
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+    # Bypass cache + ensure provider is never resolved when disabled.
+    cached = AsyncMock(side_effect=AssertionError("disabled authorization must not resolve a provider"))
+    monkeypatch.setattr("app.gateway.authz._get_cached_route_provider", cached)
+
+    permissions = await resolve_route_permissions(_user(), is_internal=False)
+
+    assert permissions == [
+        Permissions.THREADS_READ,
+        Permissions.THREADS_WRITE,
+        Permissions.THREADS_DELETE,
+        Permissions.RUNS_CREATE,
+        Permissions.RUNS_READ,
+        Permissions.RUNS_CANCEL,
+    ]
+    cached.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_route_permissions_use_async_provider_and_trusted_principal(monkeypatch):
+    provider = _RecordingProvider(denied={Permissions.THREADS_DELETE, Permissions.RUNS_CANCEL})
+    _enable_authorization(monkeypatch, provider)
+
+    permissions = await resolve_route_permissions(_user(), is_internal=True)
+
+    assert permissions == [
+        Permissions.THREADS_READ,
+        Permissions.THREADS_WRITE,
+        Permissions.RUNS_CREATE,
+        Permissions.RUNS_READ,
+    ]
+    assert [(request.resource, request.action, request.target) for request in provider.requests] == [
+        ("route", "read", Permissions.THREADS_READ),
+        ("route", "write", Permissions.THREADS_WRITE),
+        ("route", "delete", Permissions.THREADS_DELETE),
+        ("route", "create", Permissions.RUNS_CREATE),
+        ("route", "read", Permissions.RUNS_READ),
+        ("route", "cancel", Permissions.RUNS_CANCEL),
+    ]
+    principal = provider.requests[0].principal
+    assert principal.user_id == "user-123"
+    assert principal.role == "user"
+    assert principal.oauth_provider == "github"
+    assert principal.oauth_id == "oauth-456"
+    assert principal.is_internal is True
+
+
+@pytest.mark.asyncio
+async def test_route_permissions_fail_closed_denies_only_the_failed_permission(monkeypatch):
+    provider = _RecordingProvider(errors={Permissions.RUNS_CANCEL})
+    _enable_authorization(monkeypatch, provider, fail_closed=True)
+
+    permissions = await resolve_route_permissions(_user(), is_internal=False)
+
+    assert permissions == [
+        Permissions.THREADS_READ,
+        Permissions.THREADS_WRITE,
+        Permissions.THREADS_DELETE,
+        Permissions.RUNS_CREATE,
+        Permissions.RUNS_READ,
+    ]
+
+
+@pytest.mark.asyncio
+async def test_route_permissions_fail_open_allows_the_failed_permission(monkeypatch):
+    provider = _RecordingProvider(errors={Permissions.RUNS_CANCEL})
+    _enable_authorization(monkeypatch, provider, fail_closed=False)
+
+    permissions = await resolve_route_permissions(_user(), is_internal=False)
+
+    assert permissions == [
+        Permissions.THREADS_READ,
+        Permissions.THREADS_WRITE,
+        Permissions.THREADS_DELETE,
+        Permissions.RUNS_CREATE,
+        Permissions.RUNS_READ,
+        Permissions.RUNS_CANCEL,
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("fail_closed", "expected"),
+    [
+        (True, []),
+        (
+            False,
+            [
+                Permissions.THREADS_READ,
+                Permissions.THREADS_WRITE,
+                Permissions.THREADS_DELETE,
+                Permissions.RUNS_CREATE,
+                Permissions.RUNS_READ,
+                Permissions.RUNS_CANCEL,
+            ],
+        ),
+    ],
+)
+async def test_route_permissions_apply_failure_mode_to_provider_resolution(monkeypatch, fail_closed, expected):
+    config = AuthorizationConfig(
+        enabled=True,
+        fail_closed=fail_closed,
+        default_role="user",
+    )
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+
+    def fail_cached(c):
+        raise ValueError("invalid provider configuration")
+
+    monkeypatch.setattr("app.gateway.authz._get_cached_route_provider", fail_cached)
+
+    assert await resolve_route_permissions(_user(), is_internal=False) == expected
+
+
+@pytest.mark.asyncio
+async def test_route_permissions_use_builtin_rbac_route_policy(monkeypatch):
+    provider = RbacAuthorizationProvider(
+        roles={
+            "user": {
+                "routes": {
+                    "allow": [Permissions.THREADS_READ, Permissions.RUNS_READ],
+                }
+            }
+        }
+    )
+    _enable_authorization(monkeypatch, provider)
+
+    permissions = await resolve_route_permissions(_user(), is_internal=False)
+
+    assert permissions == [Permissions.THREADS_READ, Permissions.RUNS_READ]
+
+
+@pytest.mark.asyncio
+async def test_authenticate_uses_route_permission_resolution(monkeypatch):
+    user = User(email="route-authz@example.com", password_hash="hash")
+    permission_resolver = AsyncMock(return_value=[Permissions.THREADS_READ])
+    monkeypatch.setattr("app.gateway.deps.get_optional_user_from_request", AsyncMock(return_value=user))
+    monkeypatch.setattr("app.gateway.authz.resolve_route_permissions", permission_resolver)
+    request = SimpleNamespace(state=SimpleNamespace())
+
+    auth_context = await _authenticate(request)
+
+    assert auth_context.user is user
+    assert auth_context.permissions == [Permissions.THREADS_READ]
+    permission_resolver.assert_awaited_once_with(user, is_internal=False)
+
+
+def _make_middleware_app() -> FastAPI:
+    app = FastAPI()
+    app.add_middleware(AuthMiddleware)
+
+    @app.get("/api/threads")
+    @require_permission("threads", "read")
+    async def read_threads(request: Request):
+        return {"ok": True}
+
+    @app.delete("/api/threads")
+    @require_permission("threads", "delete")
+    async def delete_threads(request: Request):
+        return {"ok": True}
+
+    return app
+
+
+def test_auth_middleware_stamps_provider_derived_permissions(monkeypatch):
+    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
+    permission_resolver = AsyncMock(return_value=[Permissions.THREADS_READ])
+    monkeypatch.setattr("app.gateway.auth_middleware.resolve_route_permissions", permission_resolver)
+
+    with TestClient(_make_middleware_app()) as client:
+        assert client.get("/api/threads").status_code == 200
+        assert client.delete("/api/threads").status_code == 403
+
+    assert permission_resolver.await_count == 2
+    for call in permission_resolver.await_args_list:
+        assert call.kwargs == {"is_internal": False}
+
+
+def test_auth_middleware_marks_internal_route_principal(monkeypatch):
+    from app.gateway.internal_auth import create_internal_auth_headers
+
+    permission_resolver = AsyncMock(return_value=[Permissions.THREADS_READ])
+    monkeypatch.setattr("app.gateway.auth_middleware.resolve_route_permissions", permission_resolver)
+
+    with TestClient(_make_middleware_app()) as client:
+        response = client.get("/api/threads", headers=create_internal_auth_headers())
+
+    assert response.status_code == 200
+    permission_resolver.assert_awaited_once()
+    assert permission_resolver.await_args.kwargs == {"is_internal": True}
+
+
+# ── Provider cache tests ────────────────────────────────────────────────
+
+
+class TestRouteProviderCache:
+    """Verify the provider cache returns the same instance for unchanged config
+    and re-resolves when config content changes."""
+
+    def test_same_config_returns_same_provider(self):
+        """Calling twice with the same config object returns the same instance."""
+        import app.gateway.authz as authz_module
+
+        # Reset cache
+        authz_module._route_provider_cache.clear()
+        authz_module._route_provider_config_id = None
+        authz_module._route_provider_config_sig = None
+
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"user": {"routes": {"allow": "*"}}}},
+            ),
+        )
+
+        p1 = _get_cached_route_provider(config)
+        p2 = _get_cached_route_provider(config)
+        assert p1 is not None
+        assert p2 is p1
+
+    def test_changed_config_returns_new_provider(self):
+        """A config with different content triggers re-resolution."""
+        import app.gateway.authz as authz_module
+
+        authz_module._route_provider_cache.clear()
+        authz_module._route_provider_config_id = None
+        authz_module._route_provider_config_sig = None
+
+        config1 = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"user": {"routes": {"allow": "*"}}}},
+            ),
+        )
+        config2 = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"user": {"routes": {"allow": []}}}},
+            ),
+        )
+
+        p1 = _get_cached_route_provider(config1)
+        p2 = _get_cached_route_provider(config2)
+        assert p1 is not None
+        assert p2 is not None
+        assert p1 is not p2
+
+    def test_same_content_different_object_reuses_provider(self):
+        """Same content in a new object (e.g. hot-reload with no changes) reuses provider."""
+        import app.gateway.authz as authz_module
+
+        authz_module._route_provider_cache.clear()
+        authz_module._route_provider_config_id = None
+        authz_module._route_provider_config_sig = None
+
+        config1 = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"user": {"routes": {"allow": "*"}}}},
+            ),
+        )
+        # Same content, different object
+        config2 = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"user": {"routes": {"allow": "*"}}}},
+            ),
+        )
+
+        p1 = _get_cached_route_provider(config1)
+        p2 = _get_cached_route_provider(config2)
+        assert p1 is p2

config.example.yaml

@@ -2152,15 +2152,18 @@ run_ownership:
 #   provider:
 #     use: deerflow.authz.rbac:RbacAuthorizationProvider
 #     config:
-#       # A known role with no `tools` policy is unrestricted for tools.
-#       # Define `tools` for every role whose tool access should be constrained.
+#       # A known role with no policy for a resource is unrestricted for it.
+#       # Define both `tools` and `routes` wherever access should be constrained.
 #       roles:
 #         admin:
 #           tools: {allow: "*"}
+#           routes: {allow: "*"}
 #         user:
 #           tools: {allow: "*", deny: ["update_agent"]}
+#           routes: {allow: "*"}
 #         guest:
 #           tools: {allow: ["web_search", "read_file"]}
+#           routes: {allow: ["threads:read", "runs:read"]}
 authorization:
   enabled: false
 

docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md

@@ -288,6 +288,26 @@ Phase 1 最低验证要求：
   `filter_tools_by_authorization()` helper；同时断言被拒绝工具不在最终 bound tools 中，
   且 Layer 2 收到的 provider 与 Layer 1 为同一对象。
 
+### 2026-07-24 — Phase 2A / Gateway route permissions
+
+- **背景：** `@require_permission` 已覆盖 Gateway 的 threads/runs 普通路由，但
+  `AuthMiddleware` 和 decorator-only `_authenticate()` 都向每个已认证用户写入固定
+  `_ALL_PERMISSIONS`，所以 Phase 1 的 provider 还不能限制 HTTP route。
+- **决策：** `resolve_route_permissions()` 是唯一 route provider 入口；两条认证路径
+  都调用它，并把结果缓存到 request-scoped `AuthContext`。每个已注册 permission
+  生成独立 `AuthzRequest(resource="route", action=<action>,
+  target="<resource>:<action>")`，通过 `aauthorize()` 求值。
+- **决策：** 单项 decision 异常只按 `authorization.fail_closed` 影响对应 permission，
+  不能因为求值其他五项的 incidental failure 扩大当前 route 的拒绝范围。provider
+  解析失败按同一配置返回空权限或 legacy 全权限。
+- **兼容性：** `authorization.enabled: false` 时不解析 provider，继续返回原有六项
+  threads/runs 权限。`owner_check` 与 `require_admin_user()` 保持独立且不变。
+- **证据：** 新增 route policy、trusted principal、async provider、fail-open /
+  fail-closed、middleware/decorator 共用和 built-in RBAC 覆盖；既有 auth 与 middleware
+  回归测试一并执行。
+- **延期：** Models、Skills、Sandbox 权限；前端 effective-permissions 展示；
+  management route 的 provider 迁移。
+
 ### 新记录模板
 
 ```markdown