feat(authz): enforce model authorization at Gateway routes and runtime (#4063 Phase 3) (#4540)

* feat(authz): enforce model authorization at Gateway routes and runtime (#4063 Phase 3)

Phase 3 / Models — the first of three resource-type PRs (Models, Skills,
Sandbox). The RBAC provider already maps "model" → config key "models"
(rbac.py _RESOURCE_POLICY_KEYS), so no schema change is needed.

Gateway route layer (mirrors Phase 2A):
- resolve_model_authorization() in authz.py returns (provider, principal),
  reusing _get_cached_route_provider and build_principal_from_context,
  including the INTERNAL_SYSTEM_ROLE → None pop for internal callers.
- list_models filters via provider.filter_resources(principal, "model", names).
- get_model checks provider.authorize("model", "use"). Deny → 403 (not 404,
  since the model exists but the role lacks permission).

Runtime resolution layer (mirrors Phase 1B):
- _authorize_model_name() in agent.py runs after _resolve_model_name. On deny,
  falls back to the first allowed model (RFC §9: graceful, not crash). All
  models denied + fail_closed → ValueError (matches existing contract).

authorization.enabled: false is a complete no-op on both layers. Anonymous
requests (user=None) bypass filtering. 18 new tests + 314 existing tests pass.

* fix(authz): enforce model:use on the embedded DeerFlowClient path (Phase 3 follow-up)

Round 4 review (willem-bd): _authorize_model_name only covered the Gateway
runtime path (_make_lead_agent). The parallel lead-agent construction path
DeerFlowClient._ensure_agent (client.py) filtered tools but not the model,
so a library/embedded consumer with role-scoped model policies could run a
model the role is denied model:use for.

- Insert _authorize_model_name in _ensure_agent, mirroring _make_lead_agent.
- Resolve None default to the first configured model before the gate so the
  implicit default (create_chat_model(name=None)) is also authorized.
- Update test_authorization_filters_framework_tools_and_reuses_provider: the
  stub provider now returns an allow decision for model:use (checked during
  assembly) and patches resolve_authorization_provider in the agent namespace.
- Add 3 DeerFlowClient._ensure_agent path tests (real-path fallback,
  None-default resolution, disabled no-op); 24 tests total.

* docs(authz): document get_model provider-unavailable fail-open path + test

zhfeng review (round 5): get_model's docstring only mentioned the deny→403
path, not the provider-resolution-error + fail-open path (which allows the
request, mirroring list_models's documented fail-open semantics). The
behavior itself is correct and symmetric with list_models, but it was
undocumented and the _AuthorizationUnavailable path had no test coverage.

- Extend get_model docstring to state the provider-error fail-closed/fail-open
  outcome, matching list_models's wording.
- Add test_get_model_provider_unavailable_fail_closed_vs_open exercising the
  _AuthorizationUnavailable path (provider cannot be resolved at all), pinning
  fail-closed→403 / fail-open→200.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/gateway/authz.py

@@ -40,7 +40,7 @@ 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.provider import AuthorizationProvider, AuthzDecision, AuthzRequest, Principal
 from deerflow.authz.runtime import resolve_authorization_provider
 from deerflow.config.authorization_config import AuthorizationConfig
 
@@ -263,6 +263,61 @@ async def resolve_route_permissions(user: User, *, is_internal: bool) -> list[st
     return [p for p in results if p is not None]
 
 
+class _AuthorizationUnavailable(Exception):
+    """Raised internally when the provider cannot be resolved for a route check.
+
+    Carries the ``fail_closed`` flag so the caller can decide between deny-all
+    and legacy allow-all without re-reading config.
+    """
+
+    def __init__(self, *, fail_closed: bool) -> None:
+        self.fail_closed = fail_closed
+
+
+def resolve_model_authorization(user: User, *, is_internal: bool) -> tuple[AuthorizationProvider | None, Principal | None]:
+    """Return ``(provider, principal)`` for model-route authorization.
+
+    When authorization is disabled, returns ``(None, None)`` so callers can
+    short-circuit to legacy behavior (all models visible). When enabled,
+    resolves the cached provider and builds a Principal identical to
+    ``resolve_route_permissions`` (including the ``INTERNAL_SYSTEM_ROLE``
+    → ``None`` pop so internal callers fall under ``default_role``).
+
+    Raises ``_AuthorizationUnavailable`` (carrying the ``fail_closed`` flag)
+    when the provider cannot be resolved; callers translate that into the
+    appropriate deny response (empty list / 403).
+    """
+    config = _get_route_authorization_config()
+    if config.enabled is not True:
+        return None, None
+
+    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 model routes", exc_info=True)
+        raise _AuthorizationUnavailable(fail_closed=config.fail_closed)
+
+    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,
+    )
+    return provider, principal
+
+
 async def _authenticate(request: Request) -> AuthContext:
     """Authenticate request and return AuthContext.
 

backend/app/gateway/routers/models.py

@@ -1,9 +1,19 @@
-from fastapi import APIRouter, Depends, HTTPException
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel, Field
 
-from app.gateway.deps import get_config
+from app.gateway.authz import (
+    _AuthorizationUnavailable,
+    _is_internal_caller,
+    resolve_model_authorization,
+)
+from app.gateway.deps import get_config, get_optional_user_from_request
+from deerflow.authz.provider import AuthzDecision, AuthzRequest
 from deerflow.config.app_config import AppConfig
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(prefix="/api", tags=["models"])
 
 
@@ -37,12 +47,19 @@ class ModelsListResponse(BaseModel):
     summary="List All Models",
     description="Retrieve a list of all available AI models configured in the system.",
 )
-async def list_models(config: AppConfig = Depends(get_config)) -> ModelsListResponse:
+async def list_models(
+    request: Request,
+    config: AppConfig = Depends(get_config),
+) -> ModelsListResponse:
     """List all available models from configuration.
 
     Returns model information suitable for frontend display,
     excluding sensitive fields like API keys and internal configuration.
 
+    When ``authorization.enabled`` is true, only models the caller's role may
+    ``list`` are returned (filtered via ``provider.filter_resources``). A
+    provider error yields an empty list (fail-closed) or all models (fail-open).
+
     Returns:
         A list of all configured models with their metadata and token usage display settings.
 
@@ -73,6 +90,28 @@ async def list_models(config: AppConfig = Depends(get_config)) -> ModelsListResp
         }
         ```
     """
+    visible_models = config.models
+    fail_closed = config.authorization.fail_closed
+
+    user = await get_optional_user_from_request(request)
+    if user is not None:
+        try:
+            provider, principal = resolve_model_authorization(user, is_internal=_is_internal_caller(request, user))
+        except _AuthorizationUnavailable as exc:
+            if exc.fail_closed:
+                visible_models = []
+        else:
+            if provider is not None and principal is not None:
+                try:
+                    allowed_names = provider.filter_resources(principal, "model", [m.name for m in config.models])
+                    if not isinstance(allowed_names, list) or any(not isinstance(n, str) for n in allowed_names):
+                        raise TypeError("AuthorizationProvider.filter_resources must return list[str]")
+                    allowed_set = set(allowed_names)
+                    visible_models = [m for m in config.models if m.name in allowed_set]
+                except Exception:
+                    logger.warning("Authorization provider failed while filtering models", exc_info=True)
+                    visible_models = [] if fail_closed else config.models
+
     models = [
         ModelResponse(
             name=model.name,
@@ -82,7 +121,7 @@ async def list_models(config: AppConfig = Depends(get_config)) -> ModelsListResp
             supports_thinking=model.supports_thinking,
             supports_reasoning_effort=model.supports_reasoning_effort,
         )
-        for model in config.models
+        for model in visible_models
     ]
     return ModelsListResponse(
         models=models,
@@ -96,7 +135,11 @@ async def list_models(config: AppConfig = Depends(get_config)) -> ModelsListResp
     summary="Get Model Details",
     description="Retrieve detailed information about a specific AI model by its name.",
 )
-async def get_model(model_name: str, config: AppConfig = Depends(get_config)) -> ModelResponse:
+async def get_model(
+    model_name: str,
+    request: Request,
+    config: AppConfig = Depends(get_config),
+) -> ModelResponse:
     """Get a specific model by name.
 
     Args:
@@ -106,7 +149,10 @@ async def get_model(model_name: str, config: AppConfig = Depends(get_config)) ->
         Model information if found.
 
     Raises:
-        HTTPException: 404 if model not found.
+        HTTPException: 404 if model not found; 403 if the caller's role may not
+        ``use`` the model (only when ``authorization.enabled`` is true). A
+        provider resolution error yields 403 (fail-closed) or allows the request
+        (fail-open), mirroring ``list_models``'s provider-error semantics.
 
     Example Response:
         ```json
@@ -122,6 +168,33 @@ async def get_model(model_name: str, config: AppConfig = Depends(get_config)) ->
     if model is None:
         raise HTTPException(status_code=404, detail=f"Model '{model_name}' not found")
 
+    # Phase 3: enforce model:use authorization (deny → 403, not 404, since the
+    # model exists but the role lacks permission to use it).
+    fail_closed = config.authorization.fail_closed
+    user = await get_optional_user_from_request(request)
+    if user is not None:
+        try:
+            provider, principal = resolve_model_authorization(user, is_internal=_is_internal_caller(request, user))
+        except _AuthorizationUnavailable:
+            if fail_closed:
+                raise HTTPException(status_code=403, detail=f"Model '{model_name}' is not available for your role")
+        else:
+            if provider is not None and principal is not None:
+                try:
+                    decision = provider.authorize(AuthzRequest(principal=principal, resource="model", action="use", target=model_name))
+                    if not isinstance(decision, AuthzDecision):
+                        raise TypeError("AuthorizationProvider.authorize must return AuthzDecision")
+                    allowed = decision.allow
+                except Exception:
+                    logger.warning(
+                        "Authorization provider failed while checking model:use for %s",
+                        model_name,
+                        exc_info=True,
+                    )
+                    allowed = not fail_closed
+                if not allowed:
+                    raise HTTPException(status_code=403, detail=f"Model '{model_name}' is not available for your role")
+
     return ModelResponse(
         name=model.name,
         model=model.model,

backend/packages/harness/deerflow/agents/lead_agent/agent.py

@@ -26,6 +26,8 @@ from __future__ import annotations
 
 import logging
 import secrets
+from collections.abc import Mapping
+from typing import Any
 
 from langchain.agents import create_agent
 from langchain.agents.middleware import AgentMiddleware
@@ -47,6 +49,9 @@ from deerflow.agents.middlewares.token_usage_middleware import TokenUsageMiddlew
 from deerflow.agents.middlewares.tool_error_handling_middleware import build_lead_runtime_middlewares
 from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware
 from deerflow.agents.thread_state import get_thread_state_schema, normalize_middleware_state_schemas
+from deerflow.authz.principal import build_principal_from_context
+from deerflow.authz.provider import AuthzDecision, AuthzRequest
+from deerflow.authz.runtime import resolve_authorization_provider
 from deerflow.authz.tool_filter import apply_tool_authorization
 from deerflow.config.agents_config import load_agent_config, validate_agent_name
 from deerflow.config.app_config import AppConfig, get_app_config
@@ -135,6 +140,101 @@ def _resolve_model_name(requested_model_name: str | None = None, *, app_config:
     return default_model_name
 
 
+def _authorize_model_name(
+    model_name: str,
+    *,
+    context: Mapping[str, Any],
+    app_config: AppConfig,
+) -> str:
+    """Enforce ``model:use`` authorization on the resolved model name.
+
+    When ``authorization.enabled`` is false this is a no-op (returns
+    *model_name* unchanged). When enabled, the resolved model is checked
+    against the provider's policy via ``authorize("model", "use")`` so the
+    runtime path and the Gateway ``get_model`` route enforce the same
+    action-scoped contract (matters for custom providers that distinguish
+    ``list`` from ``use``). On deny, a graceful fallback to the first
+    ``filter_resources``-allowed model is attempted (RFC §9: "fall back to an
+    allowed default, not error, to avoid breaking runs"). If no model is
+    allowed and ``fail_closed`` is true, ``ValueError`` is raised (matching
+    the existing "no models configured" contract); fail-open returns the
+    original name.
+
+    Mirrors the Principal/provider pattern of ``apply_tool_authorization`` so
+    the tool path and the model path share one identity source.
+    """
+    authz_config = app_config.authorization
+    if authz_config.enabled is not True:
+        return model_name
+
+    provider = resolve_authorization_provider(authz_config)
+    if provider is None:
+        return model_name
+
+    principal = build_principal_from_context(context, default_role=authz_config.default_role)
+    all_names = [m.name for m in app_config.models]
+
+    # Check the resolved model against the action-scoped ``model:use`` policy.
+    # This aligns with the Gateway ``get_model`` route, which also checks
+    # ``authorize("model", "use")``. For the built-in RBAC provider (which
+    # ignores ``action``) this is equivalent to a membership check; for a
+    # custom provider that distinguishes ``list`` from ``use``, it prevents
+    # a model visible via ``filter_resources`` but denied for ``use`` from
+    # being silently selected at runtime.
+    try:
+        decision = provider.authorize(AuthzRequest(principal=principal, resource="model", action="use", target=model_name))
+        if not isinstance(decision, AuthzDecision):
+            raise TypeError("AuthorizationProvider.authorize must return AuthzDecision")
+        if decision.allow:
+            return model_name
+    except Exception:
+        logger.warning("Authorization provider failed while checking model:use for '%s'", model_name, exc_info=True)
+        if authz_config.fail_closed:
+            raise ValueError("No models are authorized for the current role (authorization provider error).")
+        return model_name
+
+    # Denied — graceful fallback: pick the first model that ``filter_resources``
+    # says is visible AND that also passes ``authorize("model", "use")``. For the
+    # built-in RBAC provider (which ignores ``action``) this is equivalent to
+    # picking the first visible name; for a custom provider that distinguishes
+    # ``list`` from ``use``, it ensures the fallback is actually usable.
+    try:
+        allowed_names = provider.filter_resources(principal, "model", all_names)
+        if not isinstance(allowed_names, list) or any(not isinstance(n, str) for n in allowed_names):
+            raise TypeError("AuthorizationProvider.filter_resources must return list[str]")
+    except Exception:
+        logger.warning("Authorization provider failed while resolving allowed models", exc_info=True)
+        if authz_config.fail_closed:
+            raise ValueError("No models are authorized for the current role (authorization provider error).")
+        return model_name
+
+    for candidate in allowed_names:
+        if candidate == model_name:
+            continue  # already denied above
+        try:
+            cb_decision = provider.authorize(AuthzRequest(principal=principal, resource="model", action="use", target=candidate))
+            if isinstance(cb_decision, AuthzDecision) and cb_decision.allow:
+                logger.warning(
+                    "Model '%s' is not authorized for the current role; fallback to '%s'.",
+                    model_name,
+                    candidate,
+                )
+                return candidate
+        except Exception:
+            logger.warning(
+                "Authorization provider failed while checking model:use fallback for '%s'",
+                candidate,
+                exc_info=True,
+            )
+            if authz_config.fail_closed:
+                raise ValueError("No models are authorized for the current role (authorization provider error).")
+            return model_name
+    if authz_config.fail_closed:
+        raise ValueError("No models are authorized for the current role.")
+    logger.warning("No models are authorized for the current role; fail_open allows '%s'.", model_name)
+    return model_name
+
+
 def _create_summarization_middleware(*, app_config: AppConfig | None = None, run_model_name: str | None = None) -> DeerFlowSummarizationMiddleware | None:
     """Create and configure the summarization middleware from config.
 
@@ -581,6 +681,10 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig):
     # Final model name resolution: request → agent config → global default, with fallback for unknown names
     model_name = _resolve_model_name(requested_model_name or agent_model_name, app_config=resolved_app_config)
 
+    # Phase 3: enforce model:use authorization. On deny, fall back to the first
+    # allowed model (graceful) rather than crashing the run (RFC §9).
+    model_name = _authorize_model_name(model_name, context=cfg, app_config=resolved_app_config)
+
     model_config = resolved_app_config.get_model_config(model_name)
 
     if model_config is None:

backend/packages/harness/deerflow/client.py

@@ -33,7 +33,7 @@ from langchain.agents.middleware import AgentMiddleware
 from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
 from langchain_core.runnables import RunnableConfig
 
-from deerflow.agents.lead_agent.agent import build_middlewares
+from deerflow.agents.lead_agent.agent import _authorize_model_name, build_middlewares
 from deerflow.agents.lead_agent.prompt import apply_prompt_template, get_enabled_skills_for_config
 from deerflow.agents.thread_state import get_thread_state_schema, normalize_middleware_state_schemas
 from deerflow.authz.principal import build_principal_from_context
@@ -290,6 +290,16 @@ class DeerFlowClient:
 
         thinking_enabled = cfg.get("thinking_enabled", True)
         model_name = cfg.get("model_name")
+        # Phase 3: enforce model:use authorization on the embedded/library path
+        # too, mirroring the Gateway runtime path in ``_make_lead_agent`` so the
+        # role-scoped model policy cannot be bypassed by constructing the agent
+        # through ``DeerFlowClient``. Resolve the ``None`` default to a concrete
+        # name first (what ``create_chat_model(name=None)`` would pick) so the
+        # policy covers the implicit default model. ``cfg`` already carries the
+        # identity that ``apply_tool_authorization`` reads below.
+        if model_name is None and self._app_config.models:
+            model_name = self._app_config.models[0].name
+        model_name = _authorize_model_name(model_name, context=cfg, app_config=self._app_config)
         subagent_enabled = cfg.get("subagent_enabled", False)
         max_concurrent_subagents = cfg.get("max_concurrent_subagents", 3)
         max_total_subagents = cfg.get("max_total_subagents", self._app_config.subagents.max_total_per_run)

backend/tests/test_client.py

@@ -1088,6 +1088,8 @@ class TestExtractText:
 
 class TestEnsureAgent:
     def test_authorization_filters_framework_tools_and_reuses_provider(self, client, mock_app_config):
+        from deerflow.authz.provider import AuthzDecision, AuthzReason
+
         class Provider:
             name = "test"
 
@@ -1095,10 +1097,13 @@ class TestEnsureAgent:
                 return [name for name in candidates if name == "safe_tool"]
 
             def authorize(self, request):
-                raise AssertionError("not called while assembling")
+                # Phase 3: model:use is now checked during assembly; allow it so
+                # the model name passes through. Tool-level authorize is still
+                # not invoked here (filter_resources drives tool assembly).
+                return AuthzDecision(allow=True, reasons=[AuthzReason(code="authz.allowed")])
 
             async def aauthorize(self, request):
-                raise AssertionError("not called while assembling")
+                return self.authorize(request)
 
         provider = Provider()
         mock_app_config.authorization = AuthorizationConfig(
@@ -1121,6 +1126,7 @@ class TestEnsureAgent:
             patch("deerflow.client.build_skill_search_setup", return_value=SimpleNamespace(describe_skill_tool=describe_tool, skill_names=frozenset({"example"}))),
             patch.object(client, "_get_tools", return_value=[safe_tool, denied_tool]),
             patch("deerflow.authz.tool_filter.resolve_authorization_provider", return_value=provider),
+            patch("deerflow.agents.lead_agent.agent.resolve_authorization_provider", return_value=provider),
             patch("deerflow.runtime.checkpointer.get_checkpointer", return_value=None),
         ):
             client._ensure_agent(client._get_runnable_config("t1"), context={"user_role": "user"})

backend/tests/test_models_authorization.py

@@ -0,0 +1,724 @@
+"""Phase 3 model-level authorization tests.
+
+Covers two enforcement layers:
+- Gateway routes (``list_models``, ``get_model``) — request-scoped Principal,
+  mirrors Phase 2A's ``resolve_route_permissions``.
+- Runtime model resolution (``_authorize_model_name``) — context-scoped
+  Principal with graceful fallback, mirrors Phase 1B's ``apply_tool_authorization``.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.gateway.routers import models as models_router
+from deerflow.authz.provider import AuthzDecision, AuthzReason
+from deerflow.authz.rbac import RbacAuthorizationProvider
+from deerflow.config.app_config import AppConfig
+from deerflow.config.authorization_config import AuthorizationConfig
+from deerflow.config.model_config import ModelConfig
+from deerflow.config.sandbox_config import SandboxConfig
+from deerflow.config.token_usage_config import TokenUsageConfig
+
+# ── Helpers ────────────────────────────────────────────────────────────
+
+
+def _user(**overrides):
+    values = {
+        "id": "user-123",
+        "system_role": "user",
+        "oauth_provider": "github",
+        "oauth_id": "oauth-456",
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+def _make_app_config(model_names: list[str]) -> AppConfig:
+    """Build a minimal AppConfig with the given model names."""
+    return AppConfig(
+        models=[ModelConfig(name=n, model=n, use="langchain_openai:ChatOpenAI") for n in model_names],
+        sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"),
+        token_usage=TokenUsageConfig(enabled=False),
+        authorization=AuthorizationConfig(),
+    )
+
+
+def _enable_authorization(monkeypatch, provider, *, fail_closed: bool = True, default_role: str = "user") -> None:
+    config = AuthorizationConfig(
+        enabled=True,
+        fail_closed=fail_closed,
+        default_role=default_role,
+    )
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+    monkeypatch.setattr("app.gateway.authz._get_cached_route_provider", lambda c: provider)
+
+
+def _make_models_app(app_config: AppConfig) -> FastAPI:
+    """Build a FastAPI app with the models router and a pinned config."""
+    app = FastAPI()
+    app.include_router(models_router.router)
+    # Pin the config dependency so routes use our test AppConfig.
+    app.dependency_overrides[models_router.get_config] = lambda: app_config
+    return app
+
+
+class _RecordingProvider:
+    """Provider that records all requests and can deny/error specific targets."""
+
+    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.authorize_requests: list = []
+        self.filter_requests: list = []
+
+    def authorize(self, request):
+        self.authorize_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")],
+        )
+
+    async def aauthorize(self, request):
+        return self.authorize(request)
+
+    def filter_resources(self, principal, resource_type, candidates):
+        self.filter_requests.append((resource_type, list(candidates)))
+        if resource_type in self.errors:
+            raise RuntimeError(f"provider failed for {resource_type}")
+        return [c for c in candidates if c not in self.denied]
+
+
+# ── list_models tests ──────────────────────────────────────────────────
+
+
+def test_list_models_disabled_returns_all(monkeypatch):
+    """When authorization is disabled, all models are visible."""
+    config = AuthorizationConfig(enabled=False)
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+    cached = AsyncMock(side_effect=AssertionError("disabled must not resolve provider"))
+    monkeypatch.setattr("app.gateway.authz._get_cached_route_provider", cached)
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models")
+
+    assert response.status_code == 200
+    names = [m["name"] for m in response.json()["models"]]
+    assert names == ["gpt-4", "claude-3"]
+    cached.assert_not_called()
+
+
+def test_list_models_anonymous_user_returns_all(monkeypatch):
+    """Anonymous requests (user=None) are not filtered."""
+    provider = _RecordingProvider()
+    _enable_authorization(monkeypatch, provider)
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=None),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models")
+
+    assert response.status_code == 200
+    names = [m["name"] for m in response.json()["models"]]
+    assert names == ["gpt-4", "claude-3"]
+    assert provider.filter_requests == []
+
+
+def test_list_models_rbac_filters_by_allow(monkeypatch):
+    """Role with allowlist sees only allowed models."""
+    provider = RbacAuthorizationProvider(
+        roles={"user": {"models": {"allow": ["gpt-4"]}}},
+    )
+    _enable_authorization(monkeypatch, provider)
+
+    app_config = _make_app_config(["gpt-4", "claude-3", "llama-3"])
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models")
+
+    assert response.status_code == 200
+    names = [m["name"] for m in response.json()["models"]]
+    assert names == ["gpt-4"]
+
+
+def test_list_models_rbac_filters_by_deny(monkeypatch):
+    """Role with deny hides denied models."""
+    provider = RbacAuthorizationProvider(
+        roles={"user": {"models": {"allow": "*", "deny": ["claude-3"]}}},
+    )
+    _enable_authorization(monkeypatch, provider)
+
+    app_config = _make_app_config(["gpt-4", "claude-3", "llama-3"])
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models")
+
+    assert response.status_code == 200
+    names = [m["name"] for m in response.json()["models"]]
+    assert names == ["gpt-4", "llama-3"]
+
+
+def test_list_models_wildcard_returns_all(monkeypatch):
+    """Role with allow: '*' sees all models."""
+    provider = RbacAuthorizationProvider(
+        roles={"user": {"models": {"allow": "*"}}},
+    )
+    _enable_authorization(monkeypatch, provider)
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models")
+
+    assert response.status_code == 200
+    names = [m["name"] for m in response.json()["models"]]
+    assert names == ["gpt-4", "claude-3"]
+
+
+@pytest.mark.parametrize(
+    ("fail_closed", "expected_count"),
+    [(True, 0), (False, 3)],
+)
+def test_list_models_provider_error_fail_closed_vs_open(monkeypatch, fail_closed, expected_count):
+    """Provider error → empty (fail-closed) or all (fail-open)."""
+    provider = _RecordingProvider(errors={"model"})
+    _enable_authorization(monkeypatch, provider, fail_closed=fail_closed)
+
+    app_config = _make_app_config(["gpt-4", "claude-3", "llama-3"])
+    app_config.authorization.fail_closed = fail_closed
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models")
+
+    assert response.status_code == 200
+    assert len(response.json()["models"]) == expected_count
+
+
+# ── get_model tests ────────────────────────────────────────────────────
+
+
+def test_get_model_disabled_returns_model(monkeypatch):
+    """When authorization is disabled, get_model works as before."""
+    config = AuthorizationConfig(enabled=False)
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+
+    app_config = _make_app_config(["gpt-4"])
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models/gpt-4")
+
+    assert response.status_code == 200
+    assert response.json()["name"] == "gpt-4"
+
+
+def test_get_model_404_when_not_found(monkeypatch):
+    """Non-existent model returns 404 regardless of authorization."""
+    provider = RbacAuthorizationProvider(roles={"user": {"models": {"allow": "*"}}})
+    _enable_authorization(monkeypatch, provider)
+
+    app_config = _make_app_config(["gpt-4"])
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models/nonexistent")
+
+    assert response.status_code == 404
+
+
+def test_get_model_denied_returns_403(monkeypatch):
+    """Role denied model:use → 403 (not 404)."""
+    provider = RbacAuthorizationProvider(
+        roles={"user": {"models": {"allow": ["claude-3"]}}},
+    )
+    _enable_authorization(monkeypatch, provider)
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models/gpt-4")
+
+    assert response.status_code == 403
+
+
+def test_get_model_allowed_returns_200(monkeypatch):
+    """Role allowed model:use → 200."""
+    provider = RbacAuthorizationProvider(
+        roles={"user": {"models": {"allow": ["gpt-4", "claude-3"]}}},
+    )
+    _enable_authorization(monkeypatch, provider)
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models/gpt-4")
+
+    assert response.status_code == 200
+    assert response.json()["name"] == "gpt-4"
+
+
+@pytest.mark.parametrize(
+    ("fail_closed", "expected_status"),
+    [(True, 403), (False, 200)],
+)
+def test_get_model_provider_error_fail_closed_vs_open(monkeypatch, fail_closed, expected_status):
+    """Provider error on model:use → 403 (fail-closed) or 200 (fail-open)."""
+    provider = _RecordingProvider(errors={"gpt-4"})
+    _enable_authorization(monkeypatch, provider, fail_closed=fail_closed)
+
+    app_config = _make_app_config(["gpt-4"])
+    app_config.authorization.fail_closed = fail_closed
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models/gpt-4")
+
+    assert response.status_code == expected_status
+
+
+@pytest.mark.parametrize(
+    ("fail_closed", "expected_status"),
+    [(True, 403), (False, 200)],
+)
+def test_get_model_provider_unavailable_fail_closed_vs_open(monkeypatch, fail_closed, expected_status):
+    """Provider *resolution* failure → 403 (fail-closed) or 200 (fail-open).
+
+    Distinct from ``test_get_model_provider_error_fail_closed_vs_open``: that
+    test exercises a provider that resolves but errors inside ``authorize``.
+    This one exercises ``_AuthorizationUnavailable`` (the provider cannot be
+    resolved at all, e.g. misconfigured class path) and pins the fail-open
+    path so the docstring's "provider resolution error yields 403 (fail-closed)
+    or allows the request (fail-open)" claim is backed by a test.
+    """
+    config = AuthorizationConfig(
+        enabled=True,
+        fail_closed=fail_closed,
+        default_role="user",
+    )
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+
+    # Force provider resolution to raise → _AuthorizationUnavailable.
+    def _boom(_config):
+        raise RuntimeError("provider class path invalid")
+
+    monkeypatch.setattr("app.gateway.authz._get_cached_route_provider", _boom)
+
+    app_config = _make_app_config(["gpt-4"])
+    app_config.authorization.fail_closed = fail_closed
+    monkeypatch.setattr(
+        "app.gateway.routers.models.get_optional_user_from_request",
+        AsyncMock(return_value=_user()),
+    )
+
+    with TestClient(_make_models_app(app_config)) as client:
+        response = client.get("/api/models/gpt-4")
+
+    assert response.status_code == expected_status
+
+
+# ── Runtime model resolution tests (_authorize_model_name) ─────────────
+
+
+def _rbac_context(**overrides):
+    """Build a minimal run context dict for build_principal_from_context."""
+    values = {
+        "user_id": "user-123",
+        "user_role": "user",
+        "oauth_provider": "github",
+        "oauth_id": "oauth-456",
+        "is_internal": False,
+    }
+    values.update(overrides)
+    return values
+
+
+def _enable_runtime_authorization(monkeypatch, provider) -> AuthorizationConfig:
+    """Patch resolve_authorization_provider in agent.py to return *provider*.
+
+    Returns an enabled AuthorizationConfig the caller assigns to app_config.
+    """
+    monkeypatch.setattr(
+        "deerflow.agents.lead_agent.agent.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    return AuthorizationConfig(enabled=True, fail_closed=True, default_role="user")
+
+
+def test_authorize_model_name_disabled_is_noop():
+    """When authorization is disabled, model name is returned unchanged."""
+    from deerflow.agents.lead_agent.agent import _authorize_model_name
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    # AuthorizationConfig() defaults to enabled=False.
+    result = _authorize_model_name("gpt-4", context=_rbac_context(), app_config=app_config)
+    assert result == "gpt-4"
+
+
+def test_authorize_model_name_allowed_returns_same(monkeypatch):
+    """Allowed model → returned unchanged."""
+    from deerflow.agents.lead_agent.agent import _authorize_model_name
+
+    provider = RbacAuthorizationProvider(
+        roles={"user": {"models": {"allow": ["gpt-4", "claude-3"]}}},
+    )
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    app_config.authorization = _enable_runtime_authorization(monkeypatch, provider)
+
+    result = _authorize_model_name("gpt-4", context=_rbac_context(), app_config=app_config)
+    assert result == "gpt-4"
+
+
+def test_authorize_model_name_denied_falls_back_gracefully(monkeypatch):
+    """Denied model → falls back to first allowed model (RFC §9)."""
+    from deerflow.agents.lead_agent.agent import _authorize_model_name
+
+    provider = RbacAuthorizationProvider(
+        roles={"user": {"models": {"allow": ["claude-3"]}}},
+    )
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    app_config.authorization = _enable_runtime_authorization(monkeypatch, provider)
+
+    result = _authorize_model_name("gpt-4", context=_rbac_context(), app_config=app_config)
+    assert result == "claude-3"
+
+
+def test_authorize_model_name_all_denied_fail_closed_raises(monkeypatch):
+    """All models denied + fail_closed → ValueError."""
+    from deerflow.agents.lead_agent.agent import _authorize_model_name
+
+    provider = RbacAuthorizationProvider(
+        roles={"user": {"models": {"allow": []}}},
+    )
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    app_config.authorization = _enable_runtime_authorization(monkeypatch, provider)
+
+    with pytest.raises(ValueError, match="No models are authorized"):
+        _authorize_model_name("gpt-4", context=_rbac_context(), app_config=app_config)
+
+
+def test_authorize_model_name_all_denied_fail_open_returns_original(monkeypatch):
+    """All models denied + fail_open → returns original model name."""
+    from deerflow.agents.lead_agent.agent import _authorize_model_name
+
+    provider = RbacAuthorizationProvider(
+        roles={"user": {"models": {"allow": []}}},
+    )
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    app_config.authorization = AuthorizationConfig(
+        enabled=True,
+        fail_closed=False,
+        default_role="user",
+    )
+    monkeypatch.setattr(
+        "deerflow.agents.lead_agent.agent.resolve_authorization_provider",
+        lambda config: provider,
+    )
+
+    result = _authorize_model_name("gpt-4", context=_rbac_context(), app_config=app_config)
+    assert result == "gpt-4"
+
+
+def test_authorize_model_name_custom_provider_list_vs_use_divergence(monkeypatch):
+    """Custom provider that allows 'list' but denies 'use' → model is denied.
+
+    Regression for willem-bd's forward-looking note: a custom provider that
+    distinguishes ``list`` from ``use`` must not let a model through the
+    runtime path just because ``filter_resources`` includes it. The runtime
+    path checks ``authorize("model", "use")`` first; only on deny does it
+    fall back to ``filter_resources`` to pick a replacement.
+    """
+
+    class _ListButNotUseProvider:
+        """Allows listing gpt-4 but denies using it."""
+
+        name = "list-not-use"
+
+        def authorize(self, request):
+            if request.resource == "model" and request.action == "use" and request.target == "gpt-4":
+                return AuthzDecision(allow=False, reasons=[AuthzReason(code="authz.denied")])
+            return AuthzDecision(allow=True, reasons=[AuthzReason(code="authz.allowed")])
+
+        async def aauthorize(self, request):
+            return self.authorize(request)
+
+        def filter_resources(self, principal, resource_type, candidates):
+            # gpt-4 is "visible" (listable) but not "usable"
+            return list(candidates)
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    app_config.authorization = AuthorizationConfig(enabled=True, fail_closed=True, default_role="user")
+    monkeypatch.setattr(
+        "deerflow.agents.lead_agent.agent.resolve_authorization_provider",
+        lambda config: _ListButNotUseProvider(),
+    )
+
+    # gpt-4 is listable but denied for use → falls back to claude-3
+    from deerflow.agents.lead_agent.agent import _authorize_model_name
+
+    result = _authorize_model_name("gpt-4", context=_rbac_context(), app_config=app_config)
+    assert result != "gpt-4"
+    assert result == "claude-3"
+
+
+def test_authorize_model_name_custom_provider_no_usable_fallback_fail_closed(monkeypatch):
+    """All visible models denied for use + fail_closed → ValueError.
+
+    Regression for willem-bd's edge-case note: when ``filter_resources``
+    returns only models that are themselves denied for ``use``, the fallback
+    must NOT silently reselect a denied model. With ``fail_closed=True`` it
+    must raise; with ``fail_closed=False`` it returns the original name.
+    """
+
+    class _AllListNoneUseProvider:
+        """Lists all models but denies use for every one of them."""
+
+        name = "all-list-none-use"
+
+        def authorize(self, request):
+            if request.resource == "model" and request.action == "use":
+                return AuthzDecision(allow=False, reasons=[AuthzReason(code="authz.denied")])
+            return AuthzDecision(allow=True, reasons=[AuthzReason(code="authz.allowed")])
+
+        async def aauthorize(self, request):
+            return self.authorize(request)
+
+        def filter_resources(self, principal, resource_type, candidates):
+            return list(candidates)  # all visible
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    app_config.authorization = AuthorizationConfig(enabled=True, fail_closed=True, default_role="user")
+    monkeypatch.setattr(
+        "deerflow.agents.lead_agent.agent.resolve_authorization_provider",
+        lambda config: _AllListNoneUseProvider(),
+    )
+
+    from deerflow.agents.lead_agent.agent import _authorize_model_name
+
+    # gpt-4 denied for use; fallback candidates also denied → ValueError
+    with pytest.raises(ValueError, match="No models are authorized"):
+        _authorize_model_name("gpt-4", context=_rbac_context(), app_config=app_config)
+
+
+def test_authorize_model_name_custom_provider_no_usable_fallback_fail_open(monkeypatch):
+    """All visible models denied for use + fail_open → returns original name."""
+
+    class _AllListNoneUseProvider:
+        name = "all-list-none-use"
+
+        def authorize(self, request):
+            if request.resource == "model" and request.action == "use":
+                return AuthzDecision(allow=False, reasons=[AuthzReason(code="authz.denied")])
+            return AuthzDecision(allow=True, reasons=[AuthzReason(code="authz.allowed")])
+
+        async def aauthorize(self, request):
+            return self.authorize(request)
+
+        def filter_resources(self, principal, resource_type, candidates):
+            return list(candidates)
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    app_config.authorization = AuthorizationConfig(enabled=True, fail_closed=False, default_role="user")
+    monkeypatch.setattr(
+        "deerflow.agents.lead_agent.agent.resolve_authorization_provider",
+        lambda config: _AllListNoneUseProvider(),
+    )
+
+    from deerflow.agents.lead_agent.agent import _authorize_model_name
+
+    result = _authorize_model_name("gpt-4", context=_rbac_context(), app_config=app_config)
+    assert result == "gpt-4"
+
+
+# ── DeerFlowClient._ensure_agent path ─────────────────────────────────
+# Regression for willem-bd's Round 4 coverage observation: the embedded/library
+# lead-agent construction path (``DeerFlowClient._ensure_agent``) must enforce
+# ``model:use`` too, not just the Gateway runtime path (``_make_lead_agent``).
+# Otherwise a consumer that enables ``authorization`` with role-scoped model
+# policies gets tools filtered yet can still run a model the role is denied
+# ``use`` for, diverging from the contract this PR establishes.
+
+
+def test_client_ensure_agent_enforces_model_use_when_authorized(monkeypatch):
+    """``_ensure_agent`` routes the resolved model through ``_authorize_model_name``.
+
+    Real-path test: we let the genuine ``_authorize_model_name`` run against a
+    real RBAC provider (only ``resolve_authorization_provider`` is patched, as
+    in the runtime tests above) so the full ``client → authz gate → RBAC →
+    fallback → create_chat_model`` chain is exercised — not just "the gate was
+    called". The provider allows only ``claude-3`` for ``use``, so the denied
+    ``gpt-4`` must be swapped for ``claude-3`` before reaching the model factory.
+    """
+    from langchain_core.runnables import RunnableConfig
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    app_config.authorization = AuthorizationConfig(enabled=True, fail_closed=True, default_role="user")
+
+    provider = RbacAuthorizationProvider(roles={"user": {"models": {"allow": ["claude-3"]}}})
+    monkeypatch.setattr(
+        "deerflow.agents.lead_agent.agent.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    captured_name = _stub_client_assembly(monkeypatch)
+
+    client = _bare_client(app_config)
+    config: RunnableConfig = {"configurable": {"model_name": "gpt-4", "user_id": "user-123", "user_role": "user"}}
+    client._ensure_agent(config)
+
+    # Denied ``gpt-4`` was swapped for the authorized fallback ``claude-3``.
+    assert captured_name["name"] == "claude-3"
+
+
+def test_client_ensure_agent_resolves_none_default_before_authorization(monkeypatch):
+    """A ``None`` model name is resolved to the default before the authz gate.
+
+    Guards the ``create_chat_model(name=None)`` semantic: when the caller omits
+    ``model_name`` the implicit default (first configured model) must still pass
+    ``model:use`` — otherwise the embedded path could run an unauthorized default.
+    """
+    from langchain_core.runnables import RunnableConfig
+
+    app_config = _make_app_config(["gpt-4", "claude-3"])
+    app_config.authorization = AuthorizationConfig(enabled=True, fail_closed=True, default_role="user")
+
+    # Deny the default ``gpt-4``; the gate must fallback to ``claude-3``.
+    provider = RbacAuthorizationProvider(roles={"user": {"models": {"allow": ["claude-3"]}}})
+    monkeypatch.setattr(
+        "deerflow.agents.lead_agent.agent.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    captured_name = _stub_client_assembly(monkeypatch)
+
+    client = _bare_client(app_config)
+    # No model_name supplied → defaults to ``gpt-4`` (first configured) → denied → fallback.
+    config: RunnableConfig = {"configurable": {"user_id": "user-123", "user_role": "user"}}
+    client._ensure_agent(config)
+
+    assert captured_name["name"] == "claude-3"
+
+
+def test_client_ensure_agent_noop_when_authorization_disabled(monkeypatch):
+    """When ``authorization.enabled`` is false, ``_ensure_agent`` leaves the model unchanged."""
+    from langchain_core.runnables import RunnableConfig
+
+    app_config = _make_app_config(["gpt-4"])
+    # AuthorizationConfig() defaults to enabled=False.
+
+    captured_name = _stub_client_assembly(monkeypatch)
+    client = _bare_client(app_config)
+    config: RunnableConfig = {"configurable": {"model_name": "gpt-4"}}
+    client._ensure_agent(config)
+
+    # Disabled → gate is a no-op: original name passed straight through.
+    assert captured_name["name"] == "gpt-4"
+
+
+def _stub_client_assembly(monkeypatch) -> dict[str, str]:
+    """Stub the heavy dependencies ``_ensure_agent`` pulls in after the authz gate.
+
+    Returns a dict the caller can inspect to see what ``create_chat_model`` got.
+    Everything here is downstream of the contract under test, so we replace it
+    with no-ops to keep the test focused on the ``_authorize_model_name`` call.
+    """
+    captured: dict[str, str] = {}
+    monkeypatch.setattr(
+        "deerflow.client.create_chat_model",
+        lambda **kwargs: captured.__setitem__("name", kwargs.get("name")) or object(),
+    )
+    monkeypatch.setattr("deerflow.client.create_agent", lambda **kwargs: object())
+    monkeypatch.setattr("deerflow.client.build_middlewares", lambda *args, **kwargs: [])
+    monkeypatch.setattr("deerflow.client.DeerFlowClient._get_tools", staticmethod(lambda *, model_name, subagent_enabled: []))  # noqa: ARG005
+    monkeypatch.setattr("deerflow.client.get_enabled_skills_for_config", lambda app_config: [])  # noqa: ARG005
+    monkeypatch.setattr(
+        "deerflow.client.build_skill_search_setup",
+        lambda skills, *, enabled, container_base_path: SimpleNamespace(describe_skill_tool=None, skill_names=frozenset()),  # noqa: ARG005
+    )
+    monkeypatch.setattr(
+        "deerflow.client.assemble_deferred_tools",
+        lambda tools, *, enabled: ([], SimpleNamespace(deferred_names=frozenset())),  # noqa: ARG005
+    )
+    monkeypatch.setattr("deerflow.client.build_mcp_routing_middleware", lambda *args, **kwargs: None)  # noqa: ARG005
+    monkeypatch.setattr("deerflow.client.get_mcp_routing_hints_prompt_section", lambda *args, **kwargs: "")  # noqa: ARG005
+    monkeypatch.setattr("deerflow.client.apply_prompt_template", lambda **kwargs: "")  # noqa: ARG005
+    monkeypatch.setattr("deerflow.client.get_thread_state_schema", lambda *args, **kwargs: object())  # noqa: ARG005
+    monkeypatch.setattr("deerflow.client.normalize_middleware_state_schemas", lambda schemas, mode, freq: [])  # noqa: ARG005
+    monkeypatch.setattr("deerflow.client.get_effective_user_id", lambda: "user-123")
+    # ``apply_tool_authorization`` (called with the empty tool list above) still
+    # resolves a provider via ``tool_filter.resolve_authorization_provider``; route
+    # it at an allow-all RBAC provider so the empty list stays empty.
+    monkeypatch.setattr(
+        "deerflow.authz.tool_filter.resolve_authorization_provider",
+        lambda config: RbacAuthorizationProvider(roles={"user": {"tools": {"allow": "*"}}}),
+    )
+    return captured
+
+
+def _bare_client(app_config):
+    """Construct a ``DeerFlowClient`` without running ``__init__``."""
+    from deerflow.client import DeerFlowClient
+
+    client = DeerFlowClient.__new__(DeerFlowClient)
+    client._app_config = app_config
+    client._agent_name = "default"
+    client._available_skills = None
+    client._checkpoint_channel_mode = "full"
+    client._checkpoint_snapshot_frequency = None
+    client._middlewares = []
+    client._agent = None
+    client._agent_config_key = None
+    # Non-None so ``_ensure_agent`` skips the real (postgres/sqlite) checkpointer
+    # resolution — the value is never used because ``create_agent`` is stubbed.
+    client._checkpointer = object()
+    return client

config.example.yaml

@@ -2325,12 +2325,15 @@ run_ownership:
 #         admin:
 #           tools: {allow: "*"}
 #           routes: {allow: "*"}
+#           models: {allow: "*"}
 #         user:
 #           tools: {allow: "*", deny: ["update_agent"]}
 #           routes: {allow: "*"}
+#           models: {allow: "*"}
 #         guest:
 #           tools: {allow: ["web_search", "read_file"]}
 #           routes: {allow: ["threads:read", "runs:read"]}
+#           models: {allow: ["gpt-4o-mini"]}
 authorization:
   enabled: false
 

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

@@ -308,6 +308,47 @@ Phase 1 最低验证要求：
 - **延期：** Models、Skills、Sandbox 权限；前端 effective-permissions 展示；
   management route 的 provider 迁移。
 
+### 2026-07-28 — Phase 3 / Models authorization (list / use)
+
+- **背景：** Phase 2A 合并后，route-level 权限已由 provider 派生，但模型仍然对所有
+  已认证用户开放——`list_models` 返回全部模型，`_resolve_model_name` 不检查角色。
+  RFC §9 Phase 3 要求覆盖 Models/Skills/Sandbox 三个资源类型。
+- **决策（Gateway 路由层）：** 新增 `resolve_model_authorization(user, *, is_internal)`
+  返回 `(provider, principal)`，复用 Phase 2A 的 `_get_cached_route_provider` 和
+  `build_principal_from_context`，包括 `INTERNAL_SYSTEM_ROLE → None` pop。
+  `list_models` 使用 `provider.filter_resources(principal, "model", names)` 批量过滤；
+  `get_model` 使用 `provider.authorize(AuthzRequest(resource="model", action="use",
+  target=model_name))`。deny → 403（模型存在但角色无权使用），provider 解析失败 →
+  `_AuthorizationUnavailable`（携带 `fail_closed` 标志）。
+- **决策（运行时解析层）：** 新增 `_authorize_model_name(model_name, *, context,
+  app_config)` 在 `_resolve_model_name` 之后执行。deny 时按 RFC §9 优雅降级：回退到
+  `filter_resources` 返回的第一个允许模型并记录 warning，而不是崩溃。全部模型被拒 +
+  `fail_closed` → `ValueError`（与现有"无模型配置"契约一致）；fail-open 返回原名。
+  fallback 阶段对每个候选重新调 `authorize("model", "use", candidate)` 验证，避免
+  custom provider 在 `filter_resources`（action-agnostic）里可见但 `use` 被拒的模型被
+  静默选中。
+- **决策（embedded/library 路径）：** `_authorize_model_name` 同样接入
+  `DeerFlowClient._ensure_agent`（client.py），与 Gateway runtime 路径 `_make_lead_agent`
+  对称。否则 library/embedded 消费者启用 `authorization` + role-scoped model policy 时，
+  tools 会被过滤但模型仍可绕过 `model:use`。调用前先把 `None` 默认解析为第一个配置模型
+  （与 `create_chat_model(name=None)` 的语义一致），确保隐式默认模型也经过授权。
+- **否决方案：** 不为 `get_model` 引入 `"read"` action——RFC §9 将 `get_model` 映射到
+  `model:use`，引入第三个 action 会增加 RBAC 配置面而无实际收益。不在 `_resolve_model_name`
+  内部做授权——该函数是纯解析（request → config → default fallback），授权检查放在调用
+  点之后，保持单一职责。
+- **兼容性：** `authorization.enabled: false` 时两条路径均为 no-op（路由返回全部模型，
+  解析返回原名）。匿名请求（user=None）不触发过滤。RBAC provider 的 `_RESOURCE_POLICY_KEYS`
+  已包含 `"model": "models"`，无需 schema 变更。
+- **证据：** `tests/test_models_authorization.py`（24 tests）覆盖 disabled/anonymous/
+  RBAC allow/deny/wildcard/fail-closed/fail-open 路由场景，disabled/allowed/
+  graceful-fallback/all-denied-fail-closed/all-denied-fail-open/custom-provider-list-vs-use/
+  no-usable-fallback 运行时场景，以及 `DeerFlowClient._ensure_agent` 的 model:use 强制 +
+  None 默认解析 + disabled no-op 集成场景；
+  `test_authorization_*.py` + `test_lead_agent_model_resolution.py` +
+  `test_auth_middleware.py` 共 318 tests 全部通过。
+- **延期：** Skills、Sandbox 权限（Phase 3 后续 PR）；前端 effective-permissions 展示；
+  management route 的 provider 迁移。
+
 ### 新记录模板
 
 ```markdown