feat(authz): middleware-declared tool path Layer 1 now covers tools declared by middlewares (#6105)

* feat(authz): extension tool authorization PR2 — middleware-declared tool path

Layer 1 now covers tools declared by middlewares, which LangChain merges
into the bound tool set after the host's explicit-list filter:

- new agents/middlewares/tool_declarations.py: record the ordinary pass's
  verdict, collect declarations from the built stack, decide only the
  delta seeded by the first pass (a name denied for this build stays
  denied), narrow the stack build-locally on independent state-preserving
  copies (direct __dict__ write, refused up front for tools data
  descriptors, never mutating caller-owned instances), and verify the
  exact bound stack after the final normalization copy.
- IsolatedMiddleware ships a state-preserving __copy__ so narrowing
  survives full/delta normalization and copies of copies.
- TodoMiddleware degrades when write_todos is denied: no system-prompt
  injection, no completion/context-loss reminders, no jump_to.
- Layer 2 forwards host-resolved tool provenance into AuthzRequest.context
  and binds the infrastructure exemption to the concrete host-created
  tool object instead of a name set (tool=None goes to the provider).
- Wired into all four assembly paths (lead x2, native subagent with the
  decision offloaded via asyncio.to_thread, DeerFlowClient).

Tests: new test_middleware_declared_tools.py (35) and
test_layer2_tool_provenance.py (7); blocking-IO anchor pinning the
subagent offload (teeth verified red->green); placement-guarantee and
enforcement suites extended; executor test doubles converted for the
async _create_agent. Behavior change called out in CHANGELOG: with
authorization enabled and a tools policy that excludes write_todos,
plan-mode builds no longer bind it (built-in RBAC default unchanged).

* fix(authz): close residual gaps from review — narrow non-collectable declarations, dedupe candidates, warn on skipped pass

- apply_declared_tool_view now removes non-BaseTool declarations from the
  bound view when authorization is enabled (they would otherwise bind via
  LangChain's create_tool auto-conversion with no provider decision);
  verify_declared_tool_view fails loudly if one survives normalization
- decide_declared_tools dedupes never-submitted names so the provider sees
  each candidate once
- subagent executor logs a warning when the provider is set but the seeded
  Layer-1 state is missing, instead of skipping the declaration pass silently
- docs: middleware-declared tools must be BaseTool instances with usable names

* fix(authz): fail loudly on non-list/tuple middleware tools declarations

_iter_declared warned and returned () for a 'tools' attribute that is not
a list/tuple, but LangChain's factory iterates the attribute as-is — so a
set/generator/dict_values declaration bound with no Layer-1 decision.
Raise DeclaredToolViewError instead, matching the module's fail-loud
standard for every other un-narrowable shape. Disabled path remains a
strict no-op. Docs updated to match.

CHANGELOG.md

@@ -37,6 +37,27 @@ This release closes that milestone with **301 merged pull requests**.
   discard previously collected pages. `total_count` still covers all filtered
   matches and no new dependency, storage layer, or HTTP endpoint is involved.
   ([#5570])
+- **agents:** Middleware-declared tools are now covered by Layer-1 tool
+  authorization on every assembly path (lead agent, native subagents, and the
+  embedded client). LangChain merges each middleware's `tools` into the bound
+  tool set *after* the host's explicit-list filter, so declarations such as an
+  extension-contributed tool or plan mode's `write_todos` previously bypassed
+  the `tools` policy entirely. Each build now collects declarations from the
+  assembled stack, decides only the names the ordinary pass never saw —
+  seeded with that pass's verdicts so a name denied for the build (including
+  by a fail-closed provider failure) stays denied — and narrows the stack on
+  independent state-preserving copies without mutating caller-owned
+  middleware instances. **Behavior change:** with `authorization.enabled`
+  and an explicit `tools` policy that does not allow `write_todos`, plan-mode
+  builds no longer bind `write_todos`, and `TodoMiddleware` correspondingly
+  stops injecting the todo system prompt and incomplete-todo completion
+  reminders; the built-in RBAC default (missing tool policy = unrestricted)
+  is unchanged, as is every deployment with authorization disabled. Layer 2
+  also forwards host-resolved tool provenance into
+  `AuthzRequest.context["tool_provenance"]` and binds the infrastructure
+  exemption (the generated `tool_search` helper) to the concrete host-created
+  tool object instead of its name, so a same-named foreign tool can no longer
+  inherit the exemption.
 - **agents:** Custom agents can persist a default knowledge scope in agent
   settings, so a specialized agent starts each conversation with its own
   corpus instead of all operator-approved knowledge bases. When a new turn

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

@@ -48,6 +48,7 @@ from deerflow.agents.middlewares.terminal_response_middleware import TerminalRes
 from deerflow.agents.middlewares.title_middleware import TitleMiddleware
 from deerflow.agents.middlewares.todo_middleware import TodoMiddleware
 from deerflow.agents.middlewares.token_usage_middleware import TokenUsageMiddleware
+from deerflow.agents.middlewares.tool_declarations import layer_one_outcome, narrow_declared_tools, verify_declared_tool_view
 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.task_continuity.tools import append_task_continuity_tools
@@ -1164,6 +1165,7 @@ def _assemble_lead_agent(config: RunnableConfig, *, app_config: AppConfig) -> Le
             context=cfg,
             app_config=resolved_app_config,
         )
+        layer_one = layer_one_outcome(authorization_candidates, authorized_tools)
         configured_tools = [tool for tool in authorized_tools if id(tool) in configured_tool_ids]
         late_tools = [tool for tool in authorized_tools if id(tool) not in configured_tool_ids]
         final_tools, setup = assemble_deferred_tools(configured_tools, enabled=resolved_app_config.tool_search.enabled)
@@ -1193,6 +1195,13 @@ def _assemble_lead_agent(config: RunnableConfig, *, app_config: AppConfig) -> Le
             skill_authorization=skill_authorization,
             subagent_execution_capacity=subagent_execution_capacity,
         )
+        middlewares, declared_authorized = narrow_declared_tools(
+            middlewares,
+            outcome=layer_one,
+            context=cfg,
+            app_config=resolved_app_config,
+            authorization_provider=_authz_provider,
+        )
         system_prompt = apply_prompt_template(
             subagent_enabled=subagent_enabled,
             max_concurrent_subagents=max_concurrent_subagents,
@@ -1218,10 +1227,12 @@ def _assemble_lead_agent(config: RunnableConfig, *, app_config: AppConfig) -> Le
             memory_enabled=memory_enabled,
             bash_available=has_bash_tool(authorized_tools),
         )
+        bound_middlewares = normalize_middleware_state_schemas(middlewares, mode)
+        verify_declared_tool_view(bound_middlewares, authorized_names=declared_authorized)
         graph = create_agent(
             model=chat_model,
             tools=final_tools,
-            middleware=normalize_middleware_state_schemas(middlewares, mode),
+            middleware=bound_middlewares,
             system_prompt=system_prompt,
             state_schema=get_thread_state_schema(mode),
             context_schema=dict,
@@ -1313,6 +1324,7 @@ def _assemble_lead_agent(config: RunnableConfig, *, app_config: AppConfig) -> Le
         context=cfg,
         app_config=resolved_app_config,
     )
+    layer_one = layer_one_outcome(authorization_candidates, authorized_tools)
     configured_tools = [tool for tool in authorized_tools if id(tool) in configured_tool_ids]
     late_tools = [tool for tool in authorized_tools if id(tool) not in configured_tool_ids]
     final_tools, setup = assemble_deferred_tools(configured_tools, enabled=resolved_app_config.tool_search.enabled)
@@ -1337,6 +1349,13 @@ def _assemble_lead_agent(config: RunnableConfig, *, app_config: AppConfig) -> Le
         skill_authorization=skill_authorization,
         subagent_execution_capacity=subagent_execution_capacity,
     )
+    middlewares, declared_authorized = narrow_declared_tools(
+        middlewares,
+        outcome=layer_one,
+        context=cfg,
+        app_config=resolved_app_config,
+        authorization_provider=_authz_provider,
+    )
     system_prompt = apply_prompt_template(
         subagent_enabled=subagent_enabled,
         max_concurrent_subagents=max_concurrent_subagents,
@@ -1354,10 +1373,12 @@ def _assemble_lead_agent(config: RunnableConfig, *, app_config: AppConfig) -> Le
         memory_enabled=memory_enabled,
         bash_available=has_bash_tool(authorized_tools),
     )
+    bound_middlewares = normalize_middleware_state_schemas(middlewares, mode)
+    verify_declared_tool_view(bound_middlewares, authorized_names=declared_authorized)
     graph = create_agent(
         model=chat_model,
         tools=final_tools,
-        middleware=normalize_middleware_state_schemas(middlewares, mode),
+        middleware=bound_middlewares,
         system_prompt=system_prompt,
         state_schema=get_thread_state_schema(mode),
         context_schema=dict,

backend/packages/harness/deerflow/agents/middlewares/todo_middleware.py

@@ -118,6 +118,20 @@ class TodoMiddleware(TodoListMiddleware):
 
     state_schema = ThreadState
 
+    @property
+    def _todo_capability_enabled(self) -> bool:
+        """False when authorization Layer 1 narrowed away this build's ``write_todos``.
+
+        An empty narrowed ``tools`` disables the middleware's todo behavior for
+        the build: no system-prompt injection, no context-loss or completion
+        reminders, no ``jump_to: model`` — the model cannot call a tool that is
+        not bound, so demanding updates wastes model rounds against an
+        unreachable tool. Keyed off instance state so the state-preserving
+        copy chain carries the degradation automatically and the caller-owned
+        instance is never touched.
+        """
+        return bool(getattr(self, "tools", None))
+
     def release_policy_parameters(self) -> dict[str, object]:
         from deerflow_extension_api import canonical_hash
 
@@ -135,6 +149,10 @@ class TodoMiddleware(TodoListMiddleware):
         runtime: Runtime,
     ) -> dict[str, Any] | None:
         """Inject a todo-list reminder when write_todos has left the context window."""
+        if not self._todo_capability_enabled:
+            # write_todos was denied for this build; context-loss detection only
+            # exists to protect todo state the model cannot update anyway.
+            return None
         todos: list[Todo] = state.get("todos") or []  # type: ignore[assignment]
         if not todos:
             return None
@@ -310,6 +328,11 @@ class TodoMiddleware(TodoListMiddleware):
         A retry cap of ``_MAX_COMPLETION_REMINDERS`` (default 2) prevents
         infinite loops when the agent cannot make further progress.
         """
+        if not self._todo_capability_enabled:
+            # write_todos was denied for this build: no completion reminders and
+            # no jump_to — the model cannot update todos through an unbound tool.
+            return None
+
         # 1. Preserve base class logic (parallel write_todos detection).
         base_result = super().after_model(state, runtime)
         if base_result is not None:
@@ -381,6 +404,10 @@ class TodoMiddleware(TodoListMiddleware):
         request: ModelRequest,
         handler: Callable[[ModelRequest], ModelResponse],
     ) -> ModelCallResult:
+        if not self._todo_capability_enabled:
+            # write_todos was denied for this build: no system-prompt injection
+            # and no completion reminders for a tool the model cannot call.
+            return handler(request)
         # The base class appends the `write_todos` system prompt to the request;
         # without calling it the model is never told about the todo list feature.
         # Augment with pending completion reminders on the request that already
@@ -398,6 +425,10 @@ class TodoMiddleware(TodoListMiddleware):
         request: ModelRequest,
         handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
     ) -> ModelCallResult:
+        if not self._todo_capability_enabled:
+            # See wrap_model_call: capability denied — pass the request through.
+            return await handler(request)
+
         # See wrap_model_call: preserve the base class system-prompt injection.
         reminders = self._drain_completion_reminders(request.runtime)
 

backend/packages/harness/deerflow/agents/middlewares/tool_declarations.py

@@ -0,0 +1,338 @@
+"""Layer 1 authorization for middleware-declared tools.
+
+LangChain's ``create_agent`` folds ``middleware.tools`` into the bound toolset
+*after* the host's explicit tool list was filtered
+(``langchain/agents/factory.py``: ``middleware_tools + regular_tools``), so a
+middleware declaration would otherwise bypass Layer 1 entirely. This module
+closes that gap with a five-step pass that runs once per build, after the
+middleware stack exists:
+
+1. **Record** — :func:`layer_one_outcome` captures the ordinary pass's verdict
+   as two name sets at the existing Layer-1 call site.
+2. **Collect** — :func:`collect_declared_tools` reads ``getattr(m, "tools",
+   ())`` off the built stack. The attribute must be a ``list`` or ``tuple``:
+   LangChain's factory iterates whatever it finds as-is, so any other
+   container (a ``set``, a generator, ``dict_values``, ...) would bind with
+   no decision at all — that shape fails the build loudly with
+   :class:`DeclaredToolViewError` rather than riding through. Entries that
+   are not ``BaseTool`` instances with a usable name cannot be authorized by
+   name; with authorization enabled they are warned about here and removed
+   from the bound stack at view time (LangChain would otherwise convert them
+   through ``create_tool`` and bind them unchecked). A contributor who needs
+   a plain callable must wrap it in a ``StructuredTool``.
+3. **Decide, seeded** — :func:`decide_declared_tools` submits only names the
+   ordinary pass never saw, through the *same* provider instance, context, and
+   app config. A name the ordinary pass already decided reuses that verdict:
+   a name denied for this build stays denied for this build, whatever the
+   middlewares declare (the seed also keeps a fail-closed provider failure
+   from being silently undone by a recovered second call).
+4. **Narrow, build-locally** — :func:`apply_declared_tool_view` returns a new
+   stack. Fully-authorized middlewares keep their identity and position; a
+   middleware with a denied or non-collectable declaration is replaced by an
+   independent copy whose ``tools`` is a new tuple of its authorized
+   declarations. The original (possibly caller-owned, possibly shared across
+   builds) is never mutated.
+5. **Verify after the last copy** — :func:`verify_declared_tool_view`
+   re-collects declared names from the exact stack being bound (after the
+   final ``normalize_middleware_state_schemas``) and raises
+   :class:`DeclaredToolViewError` if any survived narrowing. This is the
+   backstop for third-party ``__copy__`` implementations that rebuild from
+   constructor arguments and would otherwise restore denied declarations
+   during ``delta`` normalization.
+
+Copy contract for declaring middlewares: ``copy.copy`` must return an
+*independent* instance that *preserves current instance state*. A ``__copy__``
+returning ``self`` is rejected at view time; one that rebuilds from
+constructor arguments is caught by the post-chain verification; one that
+raises fails the build loudly. The narrowed assignment is safe by
+precondition: it writes ``clone.__dict__["tools"]`` directly (bypassing
+``__setattr__`` overrides and property setters) and only after
+``inspect.getattr_static`` confirms no data descriptor named ``tools`` exists
+on the class — a forwarding property would redirect the write into storage a
+shallow copy shares, so that shape is refused *before any write* and the
+build fails with the caller's instance fully intact.
+
+When ``authorization.enabled`` is not true the ordinary pass returns
+``provider=None`` and every step here is skipped: nothing is collected,
+nothing is narrowed, and the stack handed to ``create_agent`` is the original
+list object (:func:`narrow_declared_tools` returns it unchanged).
+"""
+
+from __future__ import annotations
+
+import copy
+import inspect
+import logging
+from collections.abc import Container, Mapping, Sequence
+from dataclasses import dataclass
+from typing import Any
+
+from langchain_core.tools import BaseTool
+
+from deerflow.authz.provider import AuthorizationProvider
+from deerflow.authz.tool_filter import apply_tool_authorization
+from deerflow.config.app_config import AppConfig
+
+logger = logging.getLogger(__name__)
+
+
+class DeclaredToolViewError(RuntimeError):
+    """A middleware-declared tool view could not be built or verified safely.
+
+    Always a configuration/integration error worth stopping the build for:
+    never falls back to mutating a caller-owned instance, and never silently
+    re-admits a denied declaration.
+    """
+
+
+@dataclass(frozen=True, slots=True)
+class LayerOneOutcome:
+    """What the ordinary Layer 1 pass decided for this build (names, not objects)."""
+
+    submitted: frozenset[str]
+    """Every candidate name Layer 1 saw."""
+    allowed: frozenset[str]
+    """Names Layer 1 kept."""
+
+
+def layer_one_outcome(candidates: Sequence[BaseTool], authorized: Sequence[BaseTool]) -> LayerOneOutcome:
+    """Capture the ordinary Layer 1 pass's verdict at its existing call site.
+
+    When authorization is disabled the pass returns its input unchanged, so
+    ``submitted == allowed`` and nothing can be denied downstream.
+    """
+    return LayerOneOutcome(
+        submitted=frozenset(tool.name for tool in candidates),
+        allowed=frozenset(tool.name for tool in authorized),
+    )
+
+
+def _collectable(tool: object) -> bool:
+    """A declaration this module can authorize: a ``BaseTool`` with a usable name."""
+    return isinstance(tool, BaseTool) and isinstance(tool.name, str) and bool(tool.name)
+
+
+def _middleware_name(middleware: object) -> str:
+    return getattr(middleware, "name", None) or type(middleware).__name__
+
+
+def _iter_declared(middleware: object) -> tuple[Any, ...]:
+    tools = getattr(middleware, "tools", None)
+    if tools is None:
+        return ()
+    if not isinstance(tools, (list, tuple)):
+        # Fail loudly, not a warning-and-skip: LangChain's factory iterates
+        # the attribute as-is, so a non-sequence container (set, generator,
+        # dict_values, ...) would still bind with no Layer-1 decision. Only
+        # reachable on the authorization-enabled paths — when authorization
+        # is disabled this module never inspects the stack.
+        raise DeclaredToolViewError(
+            f"Middleware {_middleware_name(middleware)} ({type(middleware).__name__}) declares 'tools' as {type(tools).__name__}, not a list/tuple; LangChain would iterate and bind it unchecked. Declare tools as a list or tuple."
+        )
+    return tuple(tools)
+
+
+def collect_declared_tools(middlewares: Sequence[object]) -> list[BaseTool]:
+    """Return the collectable ``BaseTool`` declarations in middleware order.
+
+    Entries that are not collectable are dropped with a warning: they cannot
+    be authorized by name, and :func:`apply_declared_tool_view` removes them
+    from the bound stack when authorization is enabled. The middleware itself
+    is left untouched here — dropping only excludes the entry from the
+    authorization decision.
+    """
+    declared: list[BaseTool] = []
+    for middleware in middlewares:
+        for tool in _iter_declared(middleware):
+            if _collectable(tool):
+                declared.append(tool)
+            else:
+                logger.warning(
+                    "Middleware %s declares a tool that is not a BaseTool with a usable name (%r); it cannot be authorized and is removed from the bound stack when authorization is enabled",
+                    _middleware_name(middleware),
+                    tool,
+                )
+    return declared
+
+
+def decide_declared_tools(
+    declared: Sequence[BaseTool],
+    *,
+    outcome: LayerOneOutcome,
+    context: Mapping[str, Any],
+    app_config: AppConfig,
+    authorization_provider: AuthorizationProvider | None,
+) -> frozenset[str]:
+    """Authorize declared names, seeded by the ordinary pass's verdict.
+
+    Only names the ordinary pass never submitted are sent to the provider —
+    through the *same* provider instance, context, and app config, so both
+    passes decide with one principal. A name in ``outcome.submitted`` reuses
+    the recorded verdict and is never re-submitted; in particular a denial
+    (including a fail-closed provider failure) cannot be undone by a
+    same-named middleware declaration.
+
+    ``authorization_provider=None`` (authorization disabled) authorizes every
+    declared name, matching ``filter_tools_by_authorization``'s keep-all
+    behavior.
+    """
+    # Dedupe by first occurrence of the name: two middlewares declaring the
+    # same never-submitted name must reach the provider as one candidate, so
+    # counting or audit-logging providers see each name once.
+    seen: set[str] = set()
+    new_declarations: list[BaseTool] = []
+    for tool in declared:
+        if tool.name in outcome.submitted or tool.name in seen:
+            continue
+        seen.add(tool.name)
+        new_declarations.append(tool)
+    if not new_declarations or authorization_provider is None:
+        newly_authorized = {tool.name for tool in new_declarations}
+        return frozenset(outcome.allowed | newly_authorized)
+    authorized_new, _ = apply_tool_authorization(
+        list(new_declarations),
+        context=context,
+        app_config=app_config,
+        authorization_provider=authorization_provider,
+    )
+    return frozenset(outcome.allowed | {tool.name for tool in authorized_new})
+
+
+def _independent_copy(middleware: object) -> Any:
+    try:
+        clone = copy.copy(middleware)
+    except Exception as exc:
+        raise DeclaredToolViewError(
+            f"Middleware {_middleware_name(middleware)} ({type(middleware).__name__}) declares tools denied by policy "
+            "but cannot be copied for build-local narrowing "
+            f"({type(exc).__name__}: {exc}). Give it a state-preserving __copy__ or remove the denied declaration."
+        ) from exc
+    if clone is middleware:
+        raise DeclaredToolViewError(
+            f"Middleware {_middleware_name(middleware)} ({type(middleware).__name__}) has a __copy__ that returns self; "
+            "narrowing its declared tools would mutate an instance the build does not own. "
+            "__copy__ must return an independent, state-preserving instance."
+        )
+    return clone
+
+
+def _assign_narrowed_tools(clone: Any, original: object, narrowed: tuple[Any, ...]) -> None:
+    """Write the narrowed view into *clone* without touching shared storage.
+
+    Safe by precondition, not by post-check: a data descriptor named ``tools``
+    on the class (e.g. a forwarding property) would redirect the write into
+    storage a shallow copy shares with the caller's instance, so that shape is
+    refused *before any write*. The write itself goes straight into
+    ``clone.__dict__``, bypassing ``__setattr__`` overrides and property
+    setters, so it can only ever affect the clone's own instance dict.
+    """
+    try:
+        static = inspect.getattr_static(clone, "tools")
+    except AttributeError:
+        static = None
+    if static is not None and (hasattr(static, "__set__") or hasattr(static, "__delete__")):
+        raise DeclaredToolViewError(
+            f"Middleware {_middleware_name(original)} ({type(original).__name__}) defines 'tools' as a data descriptor "
+            f"({type(static).__name__}); narrowing it safely is impossible because the write would reach storage "
+            "shared with the original instance. Store declared tools as a plain instance attribute."
+        )
+    try:
+        instance_dict = clone.__dict__
+    except AttributeError as exc:
+        raise DeclaredToolViewError(f"Middleware {_middleware_name(original)} ({type(original).__name__}) has no instance __dict__; build-local narrowing of its declared tools is impossible.") from exc
+    instance_dict["tools"] = narrowed
+
+
+def apply_declared_tool_view(
+    middlewares: Sequence[object],
+    *,
+    authorized_names: Container[str],
+) -> list[object]:
+    """Return a build-local stack with declarations narrowed to *authorized_names*.
+
+    A middleware whose declarations are all authorized keeps its identity and
+    position; a middleware with at least one denied or non-collectable
+    declaration is replaced by an independent copy whose ``tools`` is a new
+    tuple of its authorized declarations. Non-collectable entries cannot be
+    authorized by name, so they are removed here too — otherwise LangChain's
+    factory would fold them into ``ToolNode`` verbatim (auto-converting plain
+    callables through ``create_tool``) and they would bind with no provider
+    decision at all. Never mutates the caller's instances, never reorders,
+    never appends.
+    """
+    view = list(middlewares)
+    for index, middleware in enumerate(view):
+        tools = _iter_declared(middleware)
+        if not tools:
+            continue
+        kept = tuple(tool for tool in tools if _collectable(tool) and tool.name in authorized_names)
+        if len(kept) == len(tools):
+            continue
+        clone = _independent_copy(middleware)
+        _assign_narrowed_tools(clone, middleware, kept)
+        view[index] = clone
+    return view
+
+
+def verify_declared_tool_view(
+    middlewares: Sequence[object],
+    *,
+    authorized_names: Container[str] | None,
+) -> None:
+    """Re-check the exact stack being bound after the last copy in the chain.
+
+    Runs after the final ``normalize_middleware_state_schemas`` and before
+    ``create_agent`` / the assembly descriptor. A third-party ``__copy__``
+    that rebuilds from constructor arguments would restore denied declarations
+    during ``delta`` normalization *after* the view was applied; that cannot
+    be prevented at view time, so it is caught here and fails the build
+    loudly. ``authorized_names=None`` (authorization disabled) skips the
+    check.
+    """
+    if authorized_names is None:
+        return
+    for middleware in middlewares:
+        for tool in _iter_declared(middleware):
+            if not _collectable(tool):
+                raise DeclaredToolViewError(
+                    f"Middleware {_middleware_name(middleware)} ({type(middleware).__name__}) still declares "
+                    f"a tool that is not a BaseTool with a usable name ({tool!r}) after middleware normalization. "
+                    "Non-collectable declarations cannot be authorized and must not survive narrowing; "
+                    "its __copy__ must preserve instance state instead of rebuilding from constructor arguments."
+                )
+            if tool.name not in authorized_names:
+                raise DeclaredToolViewError(
+                    f"Middleware {_middleware_name(middleware)} ({type(middleware).__name__}) still declares "
+                    f"tool {tool.name!r}, which is not authorized for this build, after middleware normalization. "
+                    "Its __copy__ must preserve instance state instead of rebuilding from constructor arguments."
+                )
+
+
+def narrow_declared_tools(
+    middlewares: Sequence[object],
+    *,
+    outcome: LayerOneOutcome,
+    context: Mapping[str, Any],
+    app_config: AppConfig,
+    authorization_provider: AuthorizationProvider | None,
+) -> tuple[list[object], frozenset[str] | None]:
+    """Collect → seeded decide → narrowed view, for the synchronous assembly paths.
+
+    Returns ``(view, authorized_names)``; the subagent path performs the same
+    steps through the primitives so the decision can cross
+    ``asyncio.to_thread``. When authorization is disabled
+    (``authorization_provider is None``) this is a strict no-op: the returned
+    view is the original list object and ``authorized_names`` is ``None``,
+    which also disables :func:`verify_declared_tool_view`.
+    """
+    if authorization_provider is None:
+        return list(middlewares) if not isinstance(middlewares, list) else middlewares, None
+    declared = collect_declared_tools(middlewares)
+    authorized_names = decide_declared_tools(
+        declared,
+        outcome=outcome,
+        context=context,
+        app_config=app_config,
+        authorization_provider=authorization_provider,
+    )
+    return apply_declared_tool_view(middlewares, authorized_names=authorized_names), authorized_names

backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py

@@ -3,7 +3,7 @@
 import asyncio
 import logging
 import secrets
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Iterable
 from typing import TYPE_CHECKING, override
 
 from langchain.agents import AgentState
@@ -351,7 +351,7 @@ def _build_runtime_middlewares(
     lazy_init: bool = True,
     receipts_render_mode: str = "delegation_only",
     authorization_provider=None,
-    authorization_infrastructure_tool_names: frozenset[str] = frozenset(),
+    authorization_infrastructure_tools: Iterable[object] = (),
     available_skills: set[str] | None = None,
     owns_agent_skill_projection: bool = True,
     skill_authorization=None,
@@ -456,7 +456,7 @@ def _build_runtime_middlewares(
                     GuardrailAuthorizationAdapter(
                         authorization_provider,
                         default_role=authorization_config.default_role,
-                        infrastructure_tool_names=authorization_infrastructure_tool_names,
+                        infrastructure_tools=authorization_infrastructure_tools,
                     ),
                     fail_closed=authorization_config.fail_closed,
                 )
@@ -557,9 +557,9 @@ def build_lead_runtime_middlewares(
         authorization_provider=authorization_provider,
         available_skills=available_skills,
         owns_agent_skill_projection=owns_agent_skill_projection,
+        authorization_infrastructure_tools=((deferred_setup.tool_search_tool,) if authorization_provider is not None and deferred_setup is not None and deferred_setup.tool_search_tool is not None else ()),
         skill_authorization=skill_authorization,
         user_id=user_id,
-        authorization_infrastructure_tool_names=(frozenset({deferred_setup.tool_search_tool.name}) if authorization_provider is not None and deferred_setup is not None and deferred_setup.tool_search_tool is not None else frozenset()),
     )
 
 
@@ -596,7 +596,7 @@ def build_subagent_runtime_middlewares(
         # the subagent context — no ledger, no citations, Layer 1 goes inert.
         receipts_render_mode="always",
         authorization_provider=authorization_provider,
-        authorization_infrastructure_tool_names=(frozenset({deferred_setup.tool_search_tool.name}) if authorization_provider is not None and deferred_setup is not None and deferred_setup.tool_search_tool is not None else frozenset()),
+        authorization_infrastructure_tools=((deferred_setup.tool_search_tool,) if authorization_provider is not None and deferred_setup is not None and deferred_setup.tool_search_tool is not None else ()),
         owns_agent_skill_projection=False,
         skill_authorization=skill_authorization,
         user_id=user_id,

backend/packages/harness/deerflow/authz/adapter.py

@@ -38,10 +38,13 @@ class GuardrailAuthorizationAdapter:
             ``AuthorizationConfig.default_role``.
         resource_type: Resource type for all ``AuthzRequest`` instances.
         action: Action for all ``AuthzRequest`` instances.
-        infrastructure_tool_names: Framework tools created from an already
-            authorized capability set. These may execute without a second
-            provider decision; callers must derive the names from the current
-            build's concrete deferred setup rather than from static config.
+        infrastructure_tools: Framework tool *objects* created by this build
+            from an already authorized capability set. The exemption binds to
+            object identity — never to a name or a provenance label — so a
+            same-named foreign tool, or one self-declaring
+            ``deerflow_tool_source: "builtin"``, is not exempt. Callers pass
+            the current build's concrete tool objects (e.g. the deferred
+            setup's generated ``tool_search``), not static config.
     """
 
     name = "authorization"
@@ -53,17 +56,23 @@ class GuardrailAuthorizationAdapter:
         default_role: str = "user",
         resource_type: str = "tool",
         action: str = "call",
-        infrastructure_tool_names: Iterable[str] = (),
+        infrastructure_tools: Iterable[object] = (),
     ) -> None:
         self._provider = provider
         self._default_role = default_role
         self._resource_type = resource_type
         self._action = action
-        self._infrastructure_tool_names = frozenset(infrastructure_tool_names)
+        self._infrastructure_tools = tuple(infrastructure_tools)
 
     def _infrastructure_decision(self, request: GuardrailRequest) -> GuardrailDecision | None:
-        """Allow framework tools created from an already-filtered capability set."""
-        if request.tool_name not in self._infrastructure_tool_names:
+        """Allow this build's host-created framework tools without a second decision.
+
+        The exemption binds to the concrete tool *object* the host created for
+        this build, not to a name or a provenance label: a request that enters
+        the guardrail with ``tool=None`` (an unregistered/dynamic name) is not
+        exempt and goes to the provider.
+        """
+        if request.tool_identity is None or not any(request.tool_identity is tool for tool in self._infrastructure_tools):
             return None
         return GuardrailDecision(
             allow=True,
@@ -98,6 +107,7 @@ class GuardrailAuthorizationAdapter:
                 "is_subagent": gr.is_subagent,
                 "agent_id": gr.agent_id,
                 "timestamp": gr.timestamp,
+                "tool_provenance": gr.tool_provenance,
             },
         )
 

backend/packages/harness/deerflow/client.py

@@ -35,6 +35,7 @@ from langchain_core.runnables import RunnableConfig
 
 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, has_bash_tool
+from deerflow.agents.middlewares.tool_declarations import layer_one_outcome, narrow_declared_tools, verify_declared_tool_view
 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.config.agents_config import AGENT_NAME_PATTERN, load_agent_config
@@ -494,6 +495,7 @@ class DeerFlowClient:
             context=cfg,
             app_config=self._app_config,
         )
+        layer_one = layer_one_outcome([*tools, *late_tools], authorized_tools)
         tools = [tool for tool in authorized_tools if id(tool) in configured_tool_ids]
         late_tools = [tool for tool in authorized_tools if id(tool) not in configured_tool_ids]
         final_tools, deferred_setup = assemble_deferred_tools(tools, enabled=self._app_config.tool_search.enabled)
@@ -505,6 +507,34 @@ class DeerFlowClient:
         )
         mcp_routing_hints_section = get_mcp_routing_hints_prompt_section(authorized_tools, deferred_names=deferred_setup.deferred_names)
 
+        middlewares, declared_authorized = narrow_declared_tools(
+            build_middlewares(
+                config,
+                model_name=model_name,
+                agent_name=self._agent_name,
+                available_skills=available_skills,
+                memory_enabled=memory_enabled,
+                custom_middlewares=self._middlewares,
+                app_config=self._app_config,
+                deferred_setup=deferred_setup,
+                mcp_routing_middleware=mcp_routing_middleware,
+                user_id=effective_user_id,
+                authorization_provider=_authz_provider,
+                skill_authorization=skill_authorization,
+                subagent_execution_capacity=subagent_execution_capacity,
+            ),
+            outcome=layer_one,
+            context=cfg,
+            app_config=self._app_config,
+            authorization_provider=_authz_provider,
+        )
+        bound_middlewares = normalize_middleware_state_schemas(
+            middlewares,
+            self._checkpoint_channel_mode,
+            self._checkpoint_snapshot_frequency,
+        )
+        verify_declared_tool_view(bound_middlewares, authorized_names=declared_authorized)
+
         kwargs: dict[str, Any] = {
             # attach_tracing=False because ``stream()`` injects tracing
             # callbacks at the graph invocation root so a single embedded run
@@ -512,25 +542,7 @@ class DeerFlowClient:
             # Attaching them again on the model would emit duplicate spans.
             "model": create_chat_model(name=model_name, thinking_enabled=thinking_enabled, attach_tracing=False),
             "tools": final_tools,
-            "middleware": normalize_middleware_state_schemas(
-                build_middlewares(
-                    config,
-                    model_name=model_name,
-                    agent_name=self._agent_name,
-                    available_skills=available_skills,
-                    memory_enabled=memory_enabled,
-                    custom_middlewares=self._middlewares,
-                    app_config=self._app_config,
-                    deferred_setup=deferred_setup,
-                    mcp_routing_middleware=mcp_routing_middleware,
-                    user_id=effective_user_id,
-                    authorization_provider=_authz_provider,
-                    skill_authorization=skill_authorization,
-                    subagent_execution_capacity=subagent_execution_capacity,
-                ),
-                self._checkpoint_channel_mode,
-                self._checkpoint_snapshot_frequency,
-            ),
+            "middleware": bound_middlewares,
             "system_prompt": apply_prompt_template(
                 subagent_enabled=subagent_enabled,
                 max_concurrent_subagents=max_concurrent_subagents,

backend/packages/harness/deerflow/extensions/isolation.py

@@ -188,6 +188,22 @@ class IsolatedMiddleware(AgentMiddleware):
         self.transformers = getattr(inner, "transformers", ())
         self.state_schema = getattr(inner, "state_schema", AgentMiddleware.state_schema)
 
+    def __copy__(self) -> IsolatedMiddleware:
+        """State-preserving shallow copy; never rebuilds from ``inner``.
+
+        ``normalize_middleware_state_schemas`` (delta mode) and the
+        authorization declaration view both copy middleware instances. A
+        rebuild-based copy would re-run ``__init__`` and re-mirror
+        ``inner.tools``, silently restoring declarations Layer 1 narrowed away
+        on this build; the default copy protocol would call the argument-
+        requiring ``__new__`` above and fail. Bypassing both keeps the
+        narrowed ``tools``, transformers, and adapted ``state_schema`` through
+        the whole copy chain.
+        """
+        clone = object.__new__(type(self))
+        clone.__dict__.update(self.__dict__)
+        return clone
+
     @property
     def name(self) -> str:
         """Stable graph and trace identity for this isolated contribution."""

backend/packages/harness/deerflow/guardrails/middleware.py

@@ -74,6 +74,17 @@ class GuardrailMiddleware(AgentMiddleware[AgentState]):
         return context if isinstance(context, dict) else {}
 
     def _build_request(self, request: ToolCallRequest, context: dict) -> GuardrailRequest:
+        # The concrete bound tool object (or None for an unregistered/dynamic
+        # name): provenance is forwarded as provider context, and the object
+        # identity backs the authorization adapter's infrastructure exemption
+        # so a same-named foreign tool can never inherit it.
+        # Local import: a module-level import of deerflow.tools.tool_provenance
+        # pulls in the deerflow.tools package __init__ (builtins → task_tool →
+        # subagents), which is circular while the subagent executor is itself
+        # being imported. At tool-call time the chain is fully initialized.
+        from deerflow.tools.tool_provenance import tool_provenance_context
+
+        tool = getattr(request, "tool", None)
         return GuardrailRequest(
             tool_name=str(request.tool_call.get("name", "")),
             tool_input=request.tool_call.get("args", {}),
@@ -90,6 +101,8 @@ class GuardrailMiddleware(AgentMiddleware[AgentState]):
             channel_user_id=context.get("channel_user_id"),
             is_internal=context.get("is_internal") is True,
             authz_attributes=normalize_authz_attributes(context.get("authz_attributes")),
+            tool_provenance=tool_provenance_context(tool),
+            tool_identity=tool,
         )
 
     def _build_denied_message(self, request: ToolCallRequest, decision: GuardrailDecision) -> ToolMessage:

backend/packages/harness/deerflow/guardrails/provider.py

@@ -28,6 +28,12 @@ class GuardrailRequest:
     channel_user_id: str | None = None
     is_internal: bool = False
     authz_attributes: dict[str, Any] = field(default_factory=dict)
+    # Host-resolved tool provenance (display/enterprise context) and the bound
+    # tool object itself (opaque identity for the infrastructure exemption).
+    # ``tool_identity`` is never serialized and never forwarded into
+    # AuthzRequest; a missing tool yields ``{}`` / ``None``.
+    tool_provenance: dict[str, Any] = field(default_factory=dict)
+    tool_identity: Any = None
 
 
 @dataclass

backend/packages/harness/deerflow/subagents/executor.py

@@ -988,7 +988,7 @@ class SubagentExecutor:
             self._resolved_app_config = get_app_config()
         return self._resolved_app_config
 
-    def _create_agent(
+    async def _create_agent(
         self,
         tools: list[BaseTool] | None = None,
         *,
@@ -1000,6 +1000,12 @@ class SubagentExecutor:
         ``deferred_setup`` (assembled in ``_build_initial_state``) carries the
         deferred MCP tool names + catalog hash so the subagent gets the same
         DeferredToolFilterMiddleware the lead agent has. ``None`` is a no-op.
+
+        Async because the middleware-declared-tool decision below calls the
+        authorization provider's ``filter_resources`` — potentially external
+        policy-service I/O — and this method runs on the shared event loop
+        (``_aexecute`` awaits it), so the decision is offloaded to a worker
+        thread with the same provider instance and context.
         """
         app_config = self._get_resolved_app_config()
         if self.model_name is None:
@@ -1047,6 +1053,52 @@ class SubagentExecutor:
         if mcp_routing_middleware is not None:
             middleware_kwargs["mcp_routing_middleware"] = mcp_routing_middleware
         middlewares = build_subagent_runtime_middlewares(**middleware_kwargs)
+        # Authorization Layer 1 for middleware-declared tools (e.g. extension
+        # contributions): collect from the built stack, decide with the same
+        # provider/principal as the ordinary pass (offloaded off the event
+        # loop — a custom provider's filter_resources may do policy-service
+        # I/O), then narrow the view build-locally. The verification before
+        # create_agent is the backstop for third-party __copy__ behavior.
+        declared_authorized = None
+        layer_one = getattr(self, "_layer_one_outcome", None)
+        authz_context = getattr(self, "_authz_context", None)
+        # Production sets provider, context, and outcome together in
+        # ``_build_initial_state``; a missing piece means no ordinary pass ran
+        # on this executor (direct ``_create_agent`` calls in tests), so there
+        # is no verdict set to seed the decision with and the pass is skipped.
+        if authz_provider is not None and layer_one is not None and authz_context is not None:
+            from deerflow.agents.middlewares.tool_declarations import (
+                apply_declared_tool_view,
+                collect_declared_tools,
+                decide_declared_tools,
+                verify_declared_tool_view,
+            )
+
+            declared = collect_declared_tools(middlewares)
+            declared_authorized = await asyncio.to_thread(
+                decide_declared_tools,
+                declared,
+                outcome=layer_one,
+                context=authz_context,
+                app_config=app_config,
+                authorization_provider=authz_provider,
+            )
+            middlewares = apply_declared_tool_view(middlewares, authorized_names=declared_authorized)
+            verify_declared_tool_view(middlewares, authorized_names=declared_authorized)
+        elif authz_provider is not None:
+            # Defense in depth: the provider is configured but the seeded
+            # decision state is missing, so the declaration pass is skipped.
+            # Layer 2 still enforces at call time, but middleware-declared
+            # tools bypass Layer 1 in this build — log so a future refactor
+            # that breaks the ``_build_initial_state`` invariant is visible
+            # instead of silently reopening the channel.
+            missing = [name for name, value in (("_layer_one_outcome", layer_one), ("_authz_context", authz_context)) if value is None]
+            logger.warning(
+                "Authorization provider is configured but %s %s not set for subagent %r; skipping the middleware-declared tool pass (Layer 2 still enforces at call time)",
+                " and ".join(missing),
+                "is" if len(missing) == 1 else "are",
+                self.config.name,
+            )
         # Collect every guard middleware that exposes ``consume_stop_reason``
         # (TokenBudgetMiddleware, LoopDetectionMiddleware) so _aexecute can read
         # each after the run and surface whichever cap fired. Duck-typed
@@ -1347,6 +1399,7 @@ class SubagentExecutor:
 
         # Apply authorization Layer 1: filter tools before deferred assembly
         # so denied tools can never enter the DeferredToolCatalog.
+        from deerflow.agents.middlewares.tool_declarations import layer_one_outcome
         from deerflow.authz.tool_filter import apply_tool_authorization
 
         authz_context = {
@@ -1367,6 +1420,10 @@ class SubagentExecutor:
             context=authz_context,
             app_config=resolved_app_config,
         )
+        # The declaration pass in ``_create_agent`` reuses this run's provider,
+        # context, and verdicts — one principal and one verdict set per run.
+        self._authz_context = authz_context
+        self._layer_one_outcome = layer_one_outcome(authorization_candidates, authorized_tools)
         configured_tools = [tool for tool in authorized_tools if id(tool) in configured_tool_ids]
         late_tools = [tool for tool in authorized_tools if id(tool) not in configured_tool_ids]
 
@@ -1607,7 +1664,7 @@ class SubagentExecutor:
                 return result
 
             state, final_tools, deferred_setup = await self._build_initial_state(task)
-            agent = self._create_agent(
+            agent = await self._create_agent(
                 final_tools,
                 deferred_setup=deferred_setup,
                 extensions=loaded_extensions,

backend/tests/blocking_io/test_declared_tool_decision_offloop.py

@@ -0,0 +1,225 @@
+"""Regression: the subagent middleware-declared-tool decision runs off the event loop.
+
+``SubagentExecutor._create_agent`` is awaited by ``_aexecute`` on the shared
+event loop, and its authorization declaration pass calls the provider's
+``filter_resources`` — which a custom provider may implement with external
+policy-service IO. The decision must stay behind ``asyncio.to_thread``; if it
+is flattened back to a plain call, the blocking-probe provider below trips the
+strict Blockbuster gate (this directory's conftest). The meta-check at the
+bottom proves the probe has teeth by calling the decision inline on the loop.
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+import threading
+from types import ModuleType, SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+# Same cycle-breaking parent mocks as tests/test_subagent_executor.py: the real
+# executor must be imported behind them (conftest.py keeps a mock in
+# sys.modules["deerflow.subagents.executor"] for collection).
+_MOCKED_MODULE_NAMES = [
+    "deerflow.agents",
+    "deerflow.agents.thread_state",
+    "deerflow.agents.middlewares",
+    "deerflow.agents.middlewares.thread_data_middleware",
+    "deerflow.sandbox",
+    "deerflow.sandbox.middleware",
+    "deerflow.sandbox.security",
+    "deerflow.models",
+    "deerflow.skills.storage",
+]
+
+
+def _import_real_executor():
+    """Import the real SubagentExecutor at module scope (imports do one-time IO
+    that must not run inside a gated test item), then restore sys.modules."""
+    original_modules = {name: sys.modules.get(name) for name in _MOCKED_MODULE_NAMES}
+    original_executor = sys.modules.get("deerflow.subagents.executor")
+    # Preload the real leafs the executor imports at runtime so no import IO
+    # happens inside the gated test either.
+    tool_declarations_module = importlib.import_module("deerflow.agents.middlewares.tool_declarations")
+    audit_context_module = importlib.import_module("deerflow.agents.middlewares.audit_context")
+
+    sys.modules.pop("deerflow.subagents.executor", None)
+    subagents_pkg = sys.modules.get("deerflow.subagents")
+    had_executor_attr = subagents_pkg is not None and hasattr(subagents_pkg, "executor")
+    if had_executor_attr:
+        delattr(subagents_pkg, "executor")
+
+    for name in _MOCKED_MODULE_NAMES:
+        sys.modules[name] = MagicMock()
+    sys.modules["deerflow.agents.middlewares.tool_declarations"] = tool_declarations_module
+    sys.modules["deerflow.agents.middlewares.audit_context"] = audit_context_module
+    try:
+        module = importlib.import_module("deerflow.subagents.executor")
+    finally:
+        for name, original in original_modules.items():
+            if original is None:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = original
+        if original_executor is not None:
+            sys.modules["deerflow.subagents.executor"] = original_executor
+    return module
+
+
+executor_module = _import_real_executor()
+SubagentExecutor = executor_module.SubagentExecutor
+SubagentConfig = importlib.import_module("deerflow.subagents.config").SubagentConfig
+LayerOneOutcome = importlib.import_module("deerflow.agents.middlewares.tool_declarations").LayerOneOutcome
+tool_declarations = sys.modules["deerflow.agents.middlewares.tool_declarations"]
+
+pytestmark = pytest.mark.asyncio
+
+
+def _leaf_module(name: str, **attrs):
+    module = ModuleType(name)
+    for key, value in attrs.items():
+        setattr(module, key, value)
+    return module
+
+
+def _app_config():
+    return SimpleNamespace(
+        models=[SimpleNamespace(name="default-model")],
+        authorization=SimpleNamespace(enabled=True, fail_closed=True, default_role="user"),
+    )
+
+
+def _blocking_probe_provider(probe_file, observed_threads: list):
+    """An AuthorizationProvider whose filter_resources performs real blocking file IO."""
+    from deerflow.authz.provider import AuthzDecision
+
+    class _Provider:
+        name = "blocking-probe"
+
+        def authorize(self, request):
+            return AuthzDecision(allow=True)
+
+        async def aauthorize(self, request):
+            return self.authorize(request)
+
+        def filter_resources(self, principal, resource_type, candidates):
+            observed_threads.append(threading.current_thread())
+            body = probe_file.read_text(encoding="utf-8")  # trips the gate on the loop
+            return [candidate for candidate in candidates if candidate in body]
+
+    return _Provider()
+
+
+def _executor_with_declaration(monkeypatch, declaring):
+    app_config = _app_config()
+    captured: dict = {}
+
+    def fake_build_subagent_runtime_middlewares(**kwargs):
+        return [declaring]
+
+    def fake_create_agent(**kwargs):
+        captured["agent"] = kwargs
+        return MagicMock()
+
+    monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: MagicMock())
+    monkeypatch.setattr(executor_module, "create_agent", fake_create_agent)
+    monkeypatch.setitem(
+        sys.modules,
+        "deerflow.agents.middlewares.tool_error_handling_middleware",
+        _leaf_module(
+            "deerflow.agents.middlewares.tool_error_handling_middleware",
+            build_subagent_runtime_middlewares=fake_build_subagent_runtime_middlewares,
+        ),
+    )
+
+    # Phase 3's skill-authorization resolution requires a real
+    # AuthorizationConfig; the SimpleNamespace app_config above is scoped to
+    # the declaration pass, which uses the explicitly-set provider.
+    monkeypatch.setattr(SubagentExecutor, "_resolve_skill_authorization", lambda self: None)
+
+    executor = SubagentExecutor(
+        config=SubagentConfig(name="researcher", description="d", system_prompt="p"),
+        tools=[],
+        app_config=app_config,
+        parent_model="parent-model",
+    )
+    executor.model_name = "test-model"
+    return executor, captured
+
+
+def _tool(name: str):
+    from langchain_core.tools import StructuredTool
+
+    return StructuredTool.from_function(lambda: name, name=name, description=name)
+
+
+async def test_declared_tool_decision_runs_off_loop(monkeypatch, tmp_path):
+    """The seeded declaration decision must not execute provider IO on the loop."""
+    from langchain.agents.middleware import AgentMiddleware
+
+    probe = tmp_path / "policy.txt"
+    probe.write_text("allowed_decl", encoding="utf-8")
+    observed_threads: list = []
+
+    class _DeclaringMiddleware(AgentMiddleware):
+        def __init__(self, tools):
+            super().__init__()
+            self.tools = tools
+
+    declaring = _DeclaringMiddleware([_tool("allowed_decl"), _tool("denied_decl")])
+    executor, captured = _executor_with_declaration(monkeypatch, declaring)
+    executor._authz_provider = _blocking_probe_provider(probe, observed_threads)
+    executor._authz_context = {"user_role": "user"}
+    executor._layer_one_outcome = LayerOneOutcome(submitted=frozenset({"regular"}), allowed=frozenset({"regular"}))
+
+    await executor._create_agent()
+
+    assert observed_threads, "the provider was never consulted — the anchor is vacuous"
+    assert all(thread is not threading.current_thread() for thread in observed_threads), "filter_resources ran on the event-loop thread"
+    (bound,) = captured["agent"]["middleware"]
+    assert [tool.name for tool in bound.tools] == ["allowed_decl"]
+
+
+async def test_inline_decision_trips_the_gate(tmp_path):
+    """Meta-check: the same decision called inline (no offload) really does hit
+    the gate — BlockingError fires inside the probe — so the anchor above is
+    not vacuously green. ``filter_tools_by_authorization``'s fail-closed
+    handler then swallows that error into a denial, which is exactly why the
+    anchor's teeth are the thread identity assertion, not the exception."""
+    from blockbuster import BlockingError
+
+    probe = tmp_path / "policy.txt"
+    probe.write_text("allowed_decl", encoding="utf-8")
+    caught: list = []
+
+    class _Provider:
+        name = "blocking-probe"
+
+        def authorize(self, request):
+            from deerflow.authz.provider import AuthzDecision
+
+            return AuthzDecision(allow=True)
+
+        async def aauthorize(self, request):
+            return self.authorize(request)
+
+        def filter_resources(self, principal, resource_type, candidates):
+            try:
+                body = probe.read_text(encoding="utf-8")
+            except BlockingError as exc:
+                caught.append(exc)
+                raise
+            return [candidate for candidate in candidates if candidate in body]
+
+    authorized = tool_declarations.decide_declared_tools(
+        [_tool("allowed_decl")],
+        outcome=LayerOneOutcome(submitted=frozenset(), allowed=frozenset()),
+        context={},
+        app_config=_app_config(),
+        authorization_provider=_Provider(),
+    )
+
+    assert len(caught) == 1  # the gate fired through the deerflow decision frame
+    assert authorized == frozenset()  # … and fail_closed converted it to a denial

backend/tests/test_authorization_enforcement.py

@@ -171,6 +171,7 @@ class TestAuthorizationGuardrailWiring:
             "args": {"query": "mcp_allowed"},
             "id": "call-search",
         }
+        request.tool = deferred_setup.tool_search_tool
         request.runtime = SimpleNamespace(context={"user_role": "guest"})
         expected = MagicMock()
         handler = MagicMock(return_value=expected)
@@ -196,6 +197,7 @@ class TestAuthorizationGuardrailWiring:
         authorization_middleware = next(middleware for middleware in middlewares if isinstance(middleware, GuardrailMiddleware) and isinstance(middleware.provider, GuardrailAuthorizationAdapter))
         request = MagicMock()
         request.tool_call = {"name": "tool_search", "args": {}, "id": "call-search"}
+        request.tool = None
         request.runtime = SimpleNamespace(context={"user_role": "guest"})
         handler = MagicMock()
 
@@ -232,6 +234,7 @@ class TestAuthorizationGuardrailWiring:
             "args": {"query": "mcp_allowed"},
             "id": "call-search",
         }
+        request.tool = deferred_setup.tool_search_tool
         request.runtime = SimpleNamespace(context={"user_role": "guest"})
         expected = MagicMock()
         handler = AsyncMock(return_value=expected)

backend/tests/test_batch_extension_snapshot.py

@@ -85,7 +85,7 @@ async def test_batch_keeps_its_construction_snapshot(monkeypatch, _subagent_env,
     async def initial(self, task):
         return ({}, self.tools, None)
 
-    def create(self, tools, *, deferred_setup=None, extensions=None):
+    async def create(self, tools, *, deferred_setup=None, extensions=None):
         seen["extensions"] = extensions
         return _CapturingSubagent(seen)
 

backend/tests/test_extension_placement_guarantees.py

@@ -260,3 +260,38 @@ def test_middleware_implements_detects_overrides():
 
     assert middleware_implements(_Wraps(), "wrap_tool_call") is True
     assert middleware_implements(_Plain(), "wrap_tool_call") is False
+
+
+def test_declaring_extension_middleware_keeps_ordering_through_narrowing():
+    """A contributor declaring tools composes cleanly, and Layer-1 narrowing —
+    a build-local copy at the same position — cannot break placement
+    guarantees or the wrapper's provenance identity."""
+    from langchain_core.tools import StructuredTool
+
+    from deerflow.agents.middlewares.tool_declarations import apply_declared_tool_view
+    from deerflow.extensions.ordering import assert_ordering
+
+    def _tool(name):
+        return StructuredTool.from_function(lambda: name, name=name, description=name)
+
+    class _DeclaringProbe(_Probe):
+        def __init__(self, tag: str) -> None:
+            super().__init__(tag)
+            self.tools = [_tool("decl_kept"), _tool("decl_denied")]
+
+    stack = _stack_with(MiddlewarePlacement(_DeclaringProbe("declaring"), Placement.MODEL_PHYSICAL))
+    index = _index_of_probe(stack, "declaring")
+    wrapper = stack[index]
+
+    view = apply_declared_tool_view(stack, authorized_names=frozenset({"decl_kept"}))
+
+    narrowed = view[index]
+    assert narrowed is not wrapper
+    assert [tool.name for tool in narrowed.tools] == ["decl_kept"]
+    assert _index_of_probe(view, "declaring") == index  # position preserved
+    assert narrowed.name == wrapper.name  # graph identity preserved
+    assert narrowed.source == wrapper.source  # provenance identity preserved
+    assert_ordering(view, {})  # the invariants still hold on the narrowed view
+    # The caller-owned wrapper and the contributor's inner are untouched.
+    assert [tool.name for tool in wrapper.tools] == ["decl_kept", "decl_denied"]
+    assert [tool.name for tool in wrapper.inner.tools] == ["decl_kept", "decl_denied"]

backend/tests/test_extension_subagent_lifecycle.py

@@ -133,6 +133,15 @@ async def _noop_initial_state(self, task):
     return ({}, [], None)
 
 
+def _create_agent_stub(seen: dict | None = None):
+    """Async stand-in for SubagentExecutor._create_agent (the real one is async)."""
+
+    async def _create(self, tools, **kwargs):
+        return _CompletingAgent(seen)
+
+    return _create
+
+
 @pytest.mark.asyncio
 async def test_subagent_success_emits_shaped_start_and_completed_stop(monkeypatch, env):
     recorder = _Recorder()
@@ -140,11 +149,7 @@ async def test_subagent_success_emits_shaped_start_and_completed_stop(monkeypatc
     executor = _executor(env)
     seen: dict = {}
     monkeypatch.setattr(env.SubagentExecutor, "_build_initial_state", _noop_initial_state)
-    monkeypatch.setattr(
-        env.SubagentExecutor,
-        "_create_agent",
-        lambda self, tools, **kwargs: _CompletingAgent(seen),
-    )
+    monkeypatch.setattr(env.SubagentExecutor, "_create_agent", _create_agent_stub(seen))
 
     result = await executor._aexecute("do the thing")
 
@@ -178,11 +183,7 @@ async def test_subagent_failure_and_cancellation_map_to_distinct_outcomes(monkey
     assert recorder.stops[-1][1] is TaskOutcome.FAILED
 
     monkeypatch.setattr(env.SubagentExecutor, "_build_initial_state", _noop_initial_state)
-    monkeypatch.setattr(
-        env.SubagentExecutor,
-        "_create_agent",
-        lambda self, tools, **kwargs: _CompletingAgent(),
-    )
+    monkeypatch.setattr(env.SubagentExecutor, "_create_agent", _create_agent_stub())
     holder = env.SubagentResult(
         task_id="cancel-me",
         trace_id="trace",
@@ -218,11 +219,7 @@ async def test_subagent_without_parent_run_skips_lifecycle_but_keeps_task_store(
     executor = _executor(env, run_id=None)
     seen: dict = {}
     monkeypatch.setattr(env.SubagentExecutor, "_build_initial_state", _noop_initial_state)
-    monkeypatch.setattr(
-        env.SubagentExecutor,
-        "_create_agent",
-        lambda self, tools, **kwargs: _CompletingAgent(seen),
-    )
+    monkeypatch.setattr(env.SubagentExecutor, "_create_agent", _create_agent_stub(seen))
 
     result = await executor._aexecute("direct")
 
@@ -244,7 +241,7 @@ async def test_subagent_keeps_one_snapshot_across_build_context_and_hooks(monkey
         set_loaded_extensions(_loaded(second))
         return ({}, [], None)
 
-    def _capture_agent(self, tools, *, deferred_setup=None, extensions=None):
+    async def _capture_agent(self, tools, *, deferred_setup=None, extensions=None):
         seen["extensions"] = extensions
         return _CompletingAgent(seen)
 

backend/tests/test_extension_task_store_runtime.py

@@ -208,12 +208,21 @@ class _CapturingSubagent:
         yield {"messages": [AIMessage(content="done")]}
 
 
+def _create_agent_stub(seen: dict):
+    """Async stand-in for SubagentExecutor._create_agent (the real one is async)."""
+
+    async def _create(self, tools, **kwargs):
+        return _CapturingSubagent(seen)
+
+    return _create
+
+
 async def _run_subagent(monkeypatch, env, *, seen: dict, result_holder=None):
     async def _initial_state(self, task):
         return ({}, [], None)
 
     monkeypatch.setattr(env.SubagentExecutor, "_build_initial_state", _initial_state)
-    monkeypatch.setattr(env.SubagentExecutor, "_create_agent", lambda self, tools, **kwargs: _CapturingSubagent(seen))
+    monkeypatch.setattr(env.SubagentExecutor, "_create_agent", _create_agent_stub(seen))
     config = env.SubagentConfig(name="researcher", description="d", system_prompt="p", tools=[])
     executor = env.SubagentExecutor(config=config, tools=[], thread_id="thread-1", run_id=None)
     return await executor._aexecute("do the thing", result_holder=result_holder)
@@ -356,7 +365,7 @@ async def test_subagent_builder_receives_the_same_extension_snapshot_as_the_stor
         set_loaded_extensions(ExtensionRegistry().build())
         return ({}, [], None)
 
-    def _create_agent(self, tools, *, deferred_setup=None, extensions=None):
+    async def _create_agent(self, tools, *, deferred_setup=None, extensions=None):
         seen["extensions"] = extensions
         return _CapturingSubagent(seen)
 
@@ -388,7 +397,7 @@ async def test_subagent_prefers_the_parent_run_snapshot_over_the_singleton(monke
     async def _initial_state(self, task):
         return ({}, [], None)
 
-    def _create_agent(self, tools, *, deferred_setup=None, extensions=None):
+    async def _create_agent(self, tools, *, deferred_setup=None, extensions=None):
         seen["extensions"] = extensions
         return _CapturingSubagent(seen)
 

backend/tests/test_layer2_tool_provenance.py

@@ -0,0 +1,144 @@
+"""Layer-2 tool provenance forwarding and the identity-bound infrastructure exemption.
+
+Covers (spec: PR 2, §6.5):
+- ``GuardrailMiddleware._build_request`` resolves host provenance from the
+  bound tool object and forwards it into ``AuthzRequest.context``;
+- the authorization adapter's infrastructure exemption binds to the concrete
+  host-created tool *object* — never to a name or a provenance label;
+- a request entering the guardrail with ``tool=None`` goes to the provider.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from langchain_core.tools import StructuredTool
+
+from deerflow.authz.adapter import GuardrailAuthorizationAdapter
+from deerflow.authz.provider import AuthzDecision, AuthzReason, Principal
+from deerflow.guardrails.middleware import GuardrailMiddleware
+from deerflow.tools.tool_provenance import DECLARED_TOOL_SOURCE_METADATA_KEY, tag_plugin_tool
+
+
+def _tool(name: str) -> StructuredTool:
+    return StructuredTool.from_function(lambda: name, name=name, description=name)
+
+
+def _build_guardrail_request(tool, *, tool_name: str):
+    middleware = GuardrailMiddleware(MagicMock(), fail_closed=True)
+    request = MagicMock()
+    request.tool_call = {"name": tool_name, "args": {}, "id": "call-1"}
+    request.tool = tool
+    request.runtime = SimpleNamespace(context={"user_role": "user"})
+    return middleware._build_request(request, {"user_role": "user"})
+
+
+class _CapturingProvider:
+    name = "capturing"
+
+    def __init__(self, allow: bool = True) -> None:
+        self.allow = allow
+        self.requests = []
+
+    def authorize(self, request):
+        self.requests.append(request)
+        return AuthzDecision(
+            allow=self.allow,
+            reasons=[] if self.allow else [AuthzReason(code="test.deny", message="denied")],
+        )
+
+    async def aauthorize(self, request):
+        return self.authorize(request)
+
+    def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
+        return list(candidates)
+
+
+class TestProvenanceForwarding:
+    def test_plugin_tool_provenance_reaches_the_authz_context(self):
+        from deerflow.extensions.plugin_tools import plugin_tool_name
+
+        tool = _tool(plugin_tool_name("acme.search", "web_query"))
+        tag_plugin_tool(tool, namespace="acme.search", declaration="web_query", installation="acme:1.0.0", operation="query")
+
+        gr = _build_guardrail_request(tool, tool_name=tool.name)
+
+        assert gr.tool_identity is tool
+        assert gr.tool_provenance["source"] == "plugin:acme.search"
+        assert gr.tool_provenance["namespace"] == "acme.search"
+        assert gr.tool_provenance["declaration"] == "web_query"
+        assert gr.tool_provenance["installation"] == "acme:1.0.0"
+        assert gr.tool_provenance["operation"] == "query"
+
+        provider = _CapturingProvider()
+        GuardrailAuthorizationAdapter(provider).evaluate(gr)
+
+        forwarded = provider.requests[0].context["tool_provenance"]
+        assert forwarded["source"] == "plugin:acme.search"
+        assert forwarded["namespace"] == "acme.search"
+        # The opaque identity never leaves the guardrail layer.
+        assert "tool_identity" not in provider.requests[0].context
+
+    def test_mis_attributed_plugin_tag_is_dropped(self):
+        tool = _tool("totally_different_name")
+        tag_plugin_tool(tool, namespace="acme.search", declaration="web_query", installation="acme:1.0.0")
+
+        gr = _build_guardrail_request(tool, tool_name=tool.name)
+
+        assert gr.tool_provenance.get("source") != "plugin:acme.search"
+        assert "namespace" not in gr.tool_provenance
+
+    def test_missing_tool_yields_empty_provenance_and_no_identity(self):
+        gr = _build_guardrail_request(None, tool_name="anything")
+
+        assert gr.tool_provenance == {}
+        assert gr.tool_identity is None
+
+
+class TestIdentityBoundExemption:
+    def test_exemption_applies_to_the_host_created_object(self):
+        host_tool = _tool("tool_search")
+        provider = _CapturingProvider()
+        adapter = GuardrailAuthorizationAdapter(provider, infrastructure_tools=[host_tool])
+
+        decision = adapter.evaluate(_build_guardrail_request(host_tool, tool_name="tool_search"))
+
+        assert decision.allow is True
+        assert decision.policy_id == "authz:infrastructure"
+        assert provider.requests == []  # no provider call for the exempt tool
+
+    def test_same_named_foreign_tool_is_not_exempt(self):
+        host_tool = _tool("tool_search")
+        foreign_tool = _tool("tool_search")  # same name, different object
+        provider = _CapturingProvider(allow=False)
+        adapter = GuardrailAuthorizationAdapter(provider, infrastructure_tools=[host_tool])
+
+        decision = adapter.evaluate(_build_guardrail_request(foreign_tool, tool_name="tool_search"))
+
+        assert decision.allow is False
+        assert len(provider.requests) == 1
+
+    def test_builtin_label_does_not_buy_the_exemption(self):
+        host_tool = _tool("tool_search")
+        impostor = _tool("tool_search")
+        impostor.metadata = {DECLARED_TOOL_SOURCE_METADATA_KEY: "builtin"}
+        provider = _CapturingProvider(allow=False)
+        adapter = GuardrailAuthorizationAdapter(provider, infrastructure_tools=[host_tool])
+
+        decision = adapter.evaluate(_build_guardrail_request(impostor, tool_name="tool_search"))
+
+        assert decision.allow is False
+        assert len(provider.requests) == 1
+        # The label still flows as display context — it just decides nothing.
+        assert provider.requests[0].context["tool_provenance"]["source"] == "builtin"
+
+    def test_tool_none_request_goes_to_the_provider(self):
+        host_tool = _tool("tool_search")
+        provider = _CapturingProvider(allow=False)
+        adapter = GuardrailAuthorizationAdapter(provider, infrastructure_tools=[host_tool])
+
+        decision = adapter.evaluate(_build_guardrail_request(None, tool_name="tool_search"))
+
+        assert decision.allow is False
+        assert len(provider.requests) == 1

backend/tests/test_middleware_declared_tools.py

@@ -0,0 +1,870 @@
+"""Layer 1 authorization for middleware-declared tools (spec: PR 2, tool path).
+
+Covers the five-step declaration pass in
+``deerflow.agents.middlewares.tool_declarations`` — record the ordinary pass,
+collect declarations, seeded incremental decision, build-local narrowing view,
+post-chain verification — plus the copy contract, the ``TodoMiddleware``
+degradation, and the wiring on the lead / client / subagent assembly paths.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import logging
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+from langchain.agents.middleware import AgentMiddleware
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.tools import StructuredTool
+
+from deerflow.agents.middlewares.todo_middleware import TodoMiddleware
+from deerflow.agents.middlewares.tool_declarations import (
+    DeclaredToolViewError,
+    LayerOneOutcome,
+    apply_declared_tool_view,
+    collect_declared_tools,
+    decide_declared_tools,
+    layer_one_outcome,
+    narrow_declared_tools,
+    verify_declared_tool_view,
+)
+from deerflow.agents.thread_state import normalize_middleware_state_schemas
+from deerflow.authz.provider import AuthzDecision, Principal
+from deerflow.authz.tool_filter import apply_tool_authorization
+from deerflow.config.app_config import AppConfig
+from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig
+from deerflow.config.sandbox_config import SandboxConfig
+from deerflow.extensions.isolation import IsolatedMiddleware
+
+
+def _tool(name: str) -> StructuredTool:
+    return StructuredTool.from_function(lambda: name, name=name, description=name)
+
+
+class _FilterProvider:
+    """AuthorizationProvider test double filtering by an allowed-name list."""
+
+    name = "filter"
+
+    def __init__(self, allowed: list[str]) -> None:
+        self.allowed = allowed
+        self.calls: list[tuple[Principal, str, list[str]]] = []
+
+    def authorize(self, request):
+        return AuthzDecision(allow=True)
+
+    async def aauthorize(self, request):
+        return self.authorize(request)
+
+    def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
+        self.calls.append((principal, resource_type, list(candidates)))
+        return [candidate for candidate in candidates if candidate in self.allowed]
+
+
+class _DeclaringMiddleware(AgentMiddleware):
+    """Plain middleware declaring tools as an instance attribute."""
+
+    def __init__(self, tools: list) -> None:
+        super().__init__()
+        self.tools = tools
+
+
+def _app_config(*, fail_closed: bool = True) -> AppConfig:
+    return AppConfig(
+        models=[],
+        sandbox=SandboxConfig(use="test"),
+        authorization=AuthorizationConfig(
+            enabled=True,
+            fail_closed=fail_closed,
+            provider=AuthorizationProviderConfig(use="unused:Provider"),
+        ),
+    )
+
+
+def _outcome(submitted: list[str], allowed: list[str]) -> LayerOneOutcome:
+    return LayerOneOutcome(submitted=frozenset(submitted), allowed=frozenset(allowed))
+
+
+def _narrow(stack, *, provider, outcome, context=None, app_config=None):
+    return narrow_declared_tools(
+        stack,
+        outcome=outcome,
+        context=context or {},
+        app_config=app_config or _app_config(),
+        authorization_provider=provider,
+    )
+
+
+class TestDisabledAuthorization:
+    def test_returns_the_original_stack_and_skips_everything(self):
+        middleware = _DeclaringMiddleware([_tool("decl")])
+        stack = [middleware]
+
+        view, authorized_names = _narrow(
+            stack,
+            provider=None,
+            outcome=_outcome([], []),
+        )
+
+        assert view is stack
+        assert authorized_names is None
+        verify_declared_tool_view(view, authorized_names=authorized_names)  # no-op
+        assert view[0] is middleware
+
+    def test_non_sequence_tools_attribute_is_not_even_inspected(self):
+        """The disabled path is a strict no-op: a set-declared stack rides
+        through untouched (LangChain will iterate it as-is, matching the
+        pre-authorization behavior)."""
+        middleware = _DeclaringMiddleware({object()})
+        stack = [middleware]
+
+        view, authorized_names = _narrow(stack, provider=None, outcome=_outcome([], []))
+
+        assert view is stack
+        assert authorized_names is None
+        verify_declared_tool_view(view, authorized_names=authorized_names)  # no-op
+
+
+class TestCollectAndDecide:
+    def test_collect_keeps_base_tools_in_middleware_order(self):
+        first, second = _tool("first"), _tool("second")
+        plain = AgentMiddleware()
+        stack = [_DeclaringMiddleware([first]), plain, _DeclaringMiddleware([second])]
+
+        assert collect_declared_tools(stack) == [first, second]
+
+    def test_non_basetool_declaration_is_dropped_with_a_warning(self, caplog):
+        middleware = _DeclaringMiddleware([lambda: "not a tool", _tool("real_decl")])
+
+        with caplog.at_level(logging.WARNING):
+            declared = collect_declared_tools([middleware])
+
+        assert [tool.name for tool in declared] == ["real_decl"]
+        assert "cannot be authorized" in caplog.text
+        # The middleware itself is untouched — the entry was never a decision input.
+        assert len(middleware.tools) == 2
+
+    @pytest.mark.parametrize(
+        "container",
+        [
+            pytest.param(lambda entry: {entry}, id="set"),
+            pytest.param(lambda entry: (x for x in [entry]), id="generator"),
+            pytest.param(lambda entry: {"decl": entry}.values(), id="dict_values"),
+        ],
+    )
+    def test_non_sequence_tools_attribute_fails_the_build_loudly(self, container):
+        """LangChain's factory iterates ``middleware.tools`` as-is, so a
+        non-list/tuple container would bind with no Layer-1 decision — refuse
+        the build instead of warning and skipping."""
+        middleware = _DeclaringMiddleware(container(object()))
+
+        with pytest.raises(DeclaredToolViewError, match="not a list/tuple"):
+            collect_declared_tools([middleware])
+        # Same refusal through the narrowing view and the post-chain verify:
+        # no path re-admits the shape.
+        with pytest.raises(DeclaredToolViewError, match="not a list/tuple"):
+            _narrow([middleware], provider=_FilterProvider(["decl"]), outcome=_outcome([], []))
+        with pytest.raises(DeclaredToolViewError, match="not a list/tuple"):
+            verify_declared_tool_view([middleware], authorized_names=frozenset({"decl"}))
+
+    def test_decision_skips_the_provider_when_the_delta_is_empty(self):
+        provider = _FilterProvider([])
+        declared = [_tool("seen")]
+
+        authorized = decide_declared_tools(
+            declared,
+            outcome=_outcome(["seen"], ["seen"]),
+            context={},
+            app_config=_app_config(),
+            authorization_provider=provider,
+        )
+
+        assert authorized == frozenset({"seen"})
+        assert provider.calls == []
+
+    def test_decision_reuses_the_layer_one_provider_and_principal(self):
+        provider = _FilterProvider(["regular", "allowed_decl"])
+        app_config = _app_config()
+        candidates = [_tool("regular")]
+        authorized_tools, resolved = apply_tool_authorization(
+            candidates,
+            context={"user_role": "reviewer"},
+            app_config=app_config,
+            authorization_provider=provider,
+        )
+        outcome = layer_one_outcome(candidates, authorized_tools)
+
+        authorized = decide_declared_tools(
+            [_tool("allowed_decl"), _tool("denied_decl")],
+            outcome=outcome,
+            context={"user_role": "reviewer"},
+            app_config=app_config,
+            authorization_provider=resolved,
+        )
+
+        assert authorized == frozenset({"regular", "allowed_decl"})
+        assert len(provider.calls) == 2
+        first_principal, _, first_candidates = provider.calls[0]
+        second_principal, _, second_candidates = provider.calls[1]
+        assert first_candidates == ["regular"]
+        assert second_candidates == ["allowed_decl", "denied_decl"]
+        assert second_principal == first_principal  # one principal across both passes
+        assert second_principal.role == "reviewer"
+
+    def test_same_named_new_declarations_reach_the_provider_once(self):
+        """Two middlewares declaring the same never-submitted name submit one
+        candidate — counting or audit-logging providers must not see a doubled
+        candidate."""
+        provider = _FilterProvider(["dup"])
+
+        authorized = decide_declared_tools(
+            [_tool("dup"), _tool("dup")],
+            outcome=_outcome([], []),
+            context={},
+            app_config=_app_config(),
+            authorization_provider=provider,
+        )
+
+        assert authorized == frozenset({"dup"})
+        assert len(provider.calls) == 1
+        assert provider.calls[0][2] == ["dup"]
+
+
+class TestNarrowingView:
+    def test_denied_declaration_is_removed_on_an_independent_copy(self):
+        allowed, denied = _tool("allowed_decl"), _tool("denied_decl")
+        original = _DeclaringMiddleware([allowed, denied])
+        provider = _FilterProvider(["allowed_decl"])
+
+        view, names = _narrow([original], provider=provider, outcome=_outcome([], []))
+
+        assert names == frozenset({"allowed_decl"})
+        narrowed = view[0]
+        assert narrowed is not original
+        assert narrowed.tools == (allowed,)
+        # The caller-owned instance is fully intact.
+        assert original.tools == [allowed, denied]
+
+    def test_fully_authorized_middlewares_keep_identity_and_position(self):
+        first = _DeclaringMiddleware([_tool("a")])
+        plain = AgentMiddleware()
+        second = _DeclaringMiddleware([_tool("b")])
+        provider = _FilterProvider(["a", "b"])
+
+        view, _ = _narrow([first, plain, second], provider=provider, outcome=_outcome([], []))
+
+        assert view == [first, plain, second]
+        assert all(view[i] is original for i, original in enumerate([first, plain, second]))
+
+    def test_authorized_declaration_binds_through_the_langchain_merge(self):
+        declared = _tool("decl")
+        view, names = _narrow([_DeclaringMiddleware([declared])], provider=_FilterProvider(["decl"]), outcome=_outcome([], []))
+
+        assert "decl" in names
+        # Binding convention: declarations are never added to final_tools; the
+        # bound set is LangChain's middleware_tools + regular_tools merge.
+        bound_names = [tool.name for middleware in view for tool in getattr(middleware, "tools", [])]
+        assert bound_names == ["decl"]
+
+    def test_non_collectable_declaration_is_removed_from_the_bound_view(self):
+        """A plain callable cannot be authorized by name; with authorization
+        enabled it must not bind (LangChain would auto-convert it into the
+        ToolNode unchecked)."""
+        allowed = _tool("allowed_decl")
+        original = _DeclaringMiddleware([lambda: "not a tool", allowed])
+        provider = _FilterProvider(["allowed_decl"])
+
+        view, names = _narrow([original], provider=provider, outcome=_outcome([], []))
+
+        assert names == frozenset({"allowed_decl"})
+        narrowed = view[0]
+        assert narrowed is not original
+        assert narrowed.tools == (allowed,)
+        assert len(original.tools) == 2  # caller-owned instance untouched
+
+    def test_middleware_with_only_non_collectable_declarations_binds_none(self):
+        original = _DeclaringMiddleware([lambda: "not a tool"])
+
+        view, names = _narrow([original], provider=_FilterProvider([]), outcome=_outcome([], []))
+
+        assert names == frozenset()
+        assert view[0] is not original
+        assert view[0].tools == ()
+        assert len(original.tools) == 1
+
+    def test_verify_fails_loudly_when_a_non_collectable_declaration_survives(self):
+        """Backstop: a hostile __copy__ that rebuilds from constructor arguments
+        would restore non-collectable entries after narrowing — same fail-loud
+        standard as for denied names."""
+        restored = _DeclaringMiddleware([lambda: "not a tool", _tool("kept")])
+
+        with pytest.raises(DeclaredToolViewError, match="not a BaseTool"):
+            verify_declared_tool_view([restored], authorized_names=frozenset({"kept"}))
+
+
+class TestSeed:
+    def test_fail_closed_denial_cannot_be_rebound_by_a_declaration(self):
+        """A provider that fails on the ordinary pass and recovers must not let a
+        same-named declaration resurrect the tool it denied under fail_closed."""
+
+        class _FlakyProvider(_FilterProvider):
+            def filter_resources(self, principal, resource_type, candidates):
+                if not self.calls:
+                    self.calls.append((principal, resource_type, list(candidates)))
+                    raise RuntimeError("policy service down")
+                return super().filter_resources(principal, resource_type, candidates)
+
+        provider = _FlakyProvider(["decl"])
+        app_config = _app_config(fail_closed=True)
+        candidates = [_tool("decl")]
+        authorized_tools, resolved = apply_tool_authorization(
+            candidates,
+            context={},
+            app_config=app_config,
+            authorization_provider=provider,
+        )
+        assert authorized_tools == []  # fail_closed denied every candidate
+        outcome = layer_one_outcome(candidates, authorized_tools)
+
+        view, names = _narrow([_DeclaringMiddleware([_tool("decl")])], provider=resolved, outcome=outcome, app_config=app_config)
+
+        assert names == frozenset()
+        assert view[0].tools == ()
+        # The recovered provider was never consulted again for the seeded name.
+        assert len(provider.calls) == 1
+
+    def test_ordinary_allow_is_never_re_decided_for_a_same_named_declaration(self):
+        provider = _FilterProvider(["decl"])
+        app_config = _app_config()
+        candidates = [_tool("decl")]
+        authorized_tools, resolved = apply_tool_authorization(
+            candidates,
+            context={},
+            app_config=app_config,
+            authorization_provider=provider,
+        )
+        outcome = layer_one_outcome(candidates, authorized_tools)
+        declared = _tool("decl")  # a *different* object with the same name
+
+        view, names = _narrow([_DeclaringMiddleware([declared])], provider=resolved, outcome=outcome, app_config=app_config)
+
+        assert "decl" in names
+        assert len(provider.calls) == 1  # verdict reused, never re-submitted
+
+    def test_same_named_declaration_cannot_shadow_an_authorized_regular_tool(self):
+        """Under the name-keyed merge the regular tools= entry binds; the declared
+        object inheriting its verdict is not a second capability."""
+        regular = StructuredTool.from_function(lambda: "regular", name="shared", description="regular")
+        declared = StructuredTool.from_function(lambda: "declared", name="shared", description="declared")
+        provider = _FilterProvider(["shared"])
+        outcome = _outcome(["shared"], ["shared"])
+
+        view, names = _narrow([_DeclaringMiddleware([declared])], provider=provider, outcome=outcome)
+
+        assert names == frozenset({"shared"})
+        assert provider.calls == []  # seeded name: no second decision
+        # LangChain merges middleware_tools + regular_tools, later entry wins.
+        middleware_tools = [tool for middleware in view for tool in getattr(middleware, "tools", [])]
+        merged = {tool.name: tool for tool in [*middleware_tools, regular]}
+        assert merged["shared"] is regular
+
+
+class TestCopyFailureModes:
+    def test_copy_that_raises_fails_the_build_loudly(self):
+        class _Uncopyable(_DeclaringMiddleware):
+            def __copy__(self):
+                raise TypeError("no copies")
+
+        original = _Uncopyable([_tool("denied")])
+
+        with pytest.raises(DeclaredToolViewError, match="cannot be copied"):
+            apply_declared_tool_view([original], authorized_names=frozenset())
+        assert [tool.name for tool in original.tools] == ["denied"]
+
+    def test_copy_returning_self_is_rejected_without_touching_the_original(self):
+        class _SelfCopy(_DeclaringMiddleware):
+            def __copy__(self):
+                return self
+
+        original = _SelfCopy([_tool("denied")])
+
+        with pytest.raises(DeclaredToolViewError, match="returns self"):
+            apply_declared_tool_view([original], authorized_names=frozenset())
+        assert [tool.name for tool in original.tools] == ["denied"]
+
+        class _Rebuilding(AgentMiddleware):
+            state_schema = object  # non-None so delta normalization copies it
+
+            def __init__(self, tools):
+                super().__init__()
+                self._ctor_tools = list(tools)
+                self.tools = list(tools)
+
+            def __copy__(self):
+                return type(self)(self._ctor_tools)
+
+        original = _Rebuilding([_tool("denied")])
+        view = apply_declared_tool_view([original], authorized_names=frozenset())
+        assert view[0].tools == ()
+
+        normalized = normalize_middleware_state_schemas(view, "delta")
+
+        with pytest.raises(DeclaredToolViewError, match="'denied'"):
+            verify_declared_tool_view(normalized, authorized_names=frozenset())
+
+    def test_state_preserving_copy_passes_the_whole_chain(self):
+        class _StatePreserving(AgentMiddleware):
+            state_schema = object  # non-None so delta normalization copies it
+
+            def __init__(self, tools):
+                super().__init__()
+                self.tools = list(tools)
+
+            def __copy__(self):
+                clone = object.__new__(type(self))
+                clone.__dict__.update(self.__dict__)
+                return clone
+
+        original = _StatePreserving([_tool("denied"), _tool("kept")])
+        view = apply_declared_tool_view([original], authorized_names=frozenset({"kept"}))
+        normalized = normalize_middleware_state_schemas(view, "delta")
+
+        verify_declared_tool_view(normalized, authorized_names=frozenset({"kept"}))
+        assert [tool.name for tool in normalized[0].tools] == ["kept"]
+        assert [tool.name for tool in original.tools] == ["denied", "kept"]
+
+
+class TestSafeAssignmentPrecondition:
+    def test_forwarding_property_is_refused_before_any_write(self):
+        setter_calls: list = []
+
+        class _PropertyTools(AgentMiddleware):
+            def __init__(self, tools):
+                super().__init__()
+                self.__dict__["_shared"] = {"tools": list(tools)}
+
+            @property
+            def tools(self):
+                return self._shared["tools"]
+
+            @tools.setter
+            def tools(self, value):
+                setter_calls.append(value)
+                self._shared["tools"] = value
+
+        original = _PropertyTools([_tool("denied"), _tool("kept")])
+
+        with pytest.raises(DeclaredToolViewError, match="data descriptor"):
+            apply_declared_tool_view([original], authorized_names=frozenset({"kept"}))
+
+        assert setter_calls == []  # the setter was never invoked
+        assert [tool.name for tool in original.tools] == ["denied", "kept"]
+
+        # A later allowing build on the same instance still binds the tool.
+        view = apply_declared_tool_view([original], authorized_names=frozenset({"denied", "kept"}))
+        assert view[0] is original
+        assert [tool.name for tool in view[0].tools] == ["denied", "kept"]
+
+    def test_setattr_override_is_bypassed_by_the_direct_dict_write(self):
+        class _SetattrTrap(AgentMiddleware):
+            def __init__(self, tools):
+                super().__init__()
+                self.__dict__["tools"] = list(tools)
+
+            def __setattr__(self, key, value):
+                if key == "tools":
+                    raise AssertionError("attribute assignment must never be used for narrowing")
+                super().__setattr__(key, value)
+
+        original = _SetattrTrap([_tool("denied"), _tool("kept")])
+
+        view = apply_declared_tool_view([original], authorized_names=frozenset({"kept"}))
+
+        assert view[0] is not original
+        assert [tool.name for tool in view[0].tools] == ["kept"]
+        assert [tool.name for tool in original.tools] == ["denied", "kept"]
+
+
+class TestIsolatedMiddlewareCopyChain:
+    def _wrap(self, inner: AgentMiddleware) -> IsolatedMiddleware:
+        return IsolatedMiddleware(inner, "test:install", lambda diagnostic: None)
+
+    def test_copy_copy_on_the_wrapper_preserves_state(self):
+        inner = _DeclaringMiddleware([_tool("a"), _tool("b")])
+        wrapper = self._wrap(inner)
+
+        clone = copy.copy(wrapper)
+
+        assert clone is not wrapper
+        assert [tool.name for tool in clone.tools] == ["a", "b"]
+        assert clone.inner is inner
+        assert clone.name == wrapper.name
+
+    def test_narrowing_survives_full_and_delta_normalization_and_a_copy_of_a_copy(self):
+        allowed, denied = _tool("allowed_decl"), _tool("denied_decl")
+        inner = _DeclaringMiddleware([allowed, denied])
+        wrapper = self._wrap(inner)
+        provider = _FilterProvider(["allowed_decl"])
+
+        view, names = _narrow([wrapper], provider=provider, outcome=_outcome([], []))
+        narrowed = view[0]
+        assert narrowed is not wrapper
+        assert narrowed.tools == (allowed,)
+
+        full = normalize_middleware_state_schemas(view, "full")
+        assert [tool.name for middleware in full for tool in middleware.tools] == ["allowed_decl"]
+        verify_declared_tool_view(full, authorized_names=names)
+
+        delta = normalize_middleware_state_schemas(view, "delta")
+        verify_declared_tool_view(delta, authorized_names=names)
+        assert [tool.name for middleware in delta for tool in middleware.tools] == ["allowed_decl"]
+
+        recopied = copy.copy(delta[0])
+        assert [tool.name for tool in recopied.tools] == ["allowed_decl"]
+
+        # Neither the wrapper nor the contributor's inner was mutated.
+        assert [tool.name for tool in wrapper.tools] == ["allowed_decl", "denied_decl"]
+        assert [tool.name for tool in inner.tools] == ["allowed_decl", "denied_decl"]
+
+    def test_denied_declaration_is_not_callable_end_to_end(self):
+        """Through a real create_agent graph: the model asking for the denied
+        declaration reaches an unbound name, never the tool implementation."""
+        from _agent_e2e_helpers import build_single_tool_call_model
+        from langchain.agents import create_agent
+
+        calls: list[str] = []
+        denied = StructuredTool.from_function(lambda: calls.append("denied") or "denied", name="denied_ext", description="denied")
+        inner = _DeclaringMiddleware([denied])
+        wrapper = self._wrap(inner)
+        view, names = _narrow([wrapper], provider=_FilterProvider([]), outcome=_outcome([], []))
+
+        model = build_single_tool_call_model(tool_name="denied_ext", tool_args={})
+        agent = create_agent(model=model, tools=[], middleware=view)
+        agent.invoke({"messages": [HumanMessage(content="run it")]})
+
+        assert calls == []
+        verify_declared_tool_view(view, authorized_names=names)
+
+    def test_authorized_declaration_is_callable_end_to_end(self):
+        from _agent_e2e_helpers import build_single_tool_call_model
+        from langchain.agents import create_agent
+
+        calls: list[str] = []
+        allowed = StructuredTool.from_function(lambda: calls.append("allowed") or "allowed", name="allowed_ext", description="allowed")
+        inner = _DeclaringMiddleware([allowed])
+        wrapper = self._wrap(inner)
+        view, _ = _narrow([wrapper], provider=_FilterProvider(["allowed_ext"]), outcome=_outcome([], []))
+
+        model = build_single_tool_call_model(tool_name="allowed_ext", tool_args={})
+        agent = create_agent(model=model, tools=[], middleware=view)
+        agent.invoke({"messages": [HumanMessage(content="run it")]})
+
+        assert calls == ["allowed"]
+
+
+class TestCrossBuildIsolation:
+    def test_deny_then_allow_on_one_caller_owned_instance(self):
+        allowed, denied = _tool("allowed_decl"), _tool("denied_decl")
+        original = _DeclaringMiddleware([allowed, denied])
+        before = dict(original.__dict__)
+
+        denying_view, _ = _narrow([original], provider=_FilterProvider(["allowed_decl"]), outcome=_outcome([], []))
+        assert [tool.name for tool in denying_view[0].tools] == ["allowed_decl"]
+        assert original.__dict__ == before  # byte-identical after the denying build
+
+        allowing_view, _ = _narrow([original], provider=_FilterProvider(["allowed_decl", "denied_decl"]), outcome=_outcome([], []))
+        assert allowing_view[0] is original
+        assert [tool.name for tool in allowing_view[0].tools] == ["allowed_decl", "denied_decl"]
+
+    def test_interleaved_builds_with_different_principals_see_their_own_view(self):
+        class _RoleProvider(_FilterProvider):
+            def filter_resources(self, principal, resource_type, candidates):
+                self.calls.append((principal, resource_type, list(candidates)))
+                allowed = {"admin": ["decl_a", "decl_b"], "guest": ["decl_a"]}[principal.role]
+                return [candidate for candidate in candidates if candidate in allowed]
+
+        original = _DeclaringMiddleware([_tool("decl_a"), _tool("decl_b")])
+        provider = _RoleProvider([])
+
+        guest_view, guest_names = _narrow([original], provider=provider, outcome=_outcome([], []), context={"user_role": "guest"})
+        admin_view, admin_names = _narrow([original], provider=provider, outcome=_outcome([], []), context={"user_role": "admin"})
+
+        assert [tool.name for tool in guest_view[0].tools] == ["decl_a"]
+        assert guest_names == frozenset({"decl_a"})
+        assert [tool.name for tool in admin_view[0].tools] == ["decl_a", "decl_b"]
+        assert admin_names == frozenset({"decl_a", "decl_b"})
+        assert [tool.name for tool in original.tools] == ["decl_a", "decl_b"]
+
+
+class TestPlanModeWriteTodos:
+    def test_denied_write_todos_is_removed_from_the_bound_name_set(self):
+        provider = _FilterProvider(["bash"])  # policy covers tools, write_todos not allowed
+        todo = TodoMiddleware()
+        outcome = _outcome(["bash"], ["bash"])
+
+        view, names = _narrow([todo], provider=provider, outcome=outcome)
+
+        bound_names = {tool.name for middleware in view for tool in getattr(middleware, "tools", [])} | {"bash"}
+        assert "write_todos" not in bound_names
+        assert view[0] is not todo
+        assert [tool.name for tool in todo.tools] == ["write_todos"]  # original untouched
+
+    def test_unrestricted_policy_still_binds_write_todos(self):
+        provider = _FilterProvider(["bash", "write_todos"])
+        todo = TodoMiddleware()
+
+        view, _ = _narrow([todo], provider=provider, outcome=_outcome(["bash"], ["bash"]))
+
+        assert view[0] is todo
+        assert [tool.name for tool in view[0].tools] == ["write_todos"]
+
+
+def _runtime():
+    runtime = MagicMock()
+    runtime.context = {"thread_id": "test-thread", "run_id": "test-run"}
+    return runtime
+
+
+def _model_request(messages):
+    from langchain.agents.middleware.types import ModelRequest
+
+    return ModelRequest(
+        model=object(),
+        messages=list(messages),
+        state={"messages": list(messages)},
+        runtime=_runtime(),
+    )
+
+
+class TestTodoDegradation:
+    def _narrowed_todo(self) -> TodoMiddleware:
+        todo = TodoMiddleware()
+        view = apply_declared_tool_view([todo], authorized_names=frozenset({"bash"}))
+        narrowed = view[0]
+        assert narrowed.tools == ()
+        return narrowed
+
+    def _state_with_incomplete_todos(self):
+        return {
+            "messages": [HumanMessage(content="hi"), AIMessage(content="all done, no more tool calls")],
+            "todos": [{"status": "in_progress", "content": "finish the task"}],
+        }
+
+    def test_no_system_prompt_injection_when_denied(self):
+        narrowed = self._narrowed_todo()
+        request = _model_request([HumanMessage(content="hi")])
+        seen = []
+
+        narrowed.wrap_model_call(request, lambda req: seen.append(req) or MagicMock())
+
+        assert seen == [request]  # untouched request: no todo system prompt
+        assert seen[0].system_message is None
+
+    def test_system_prompt_injection_preserved_when_allowed(self):
+        todo = TodoMiddleware()
+        request = _model_request([HumanMessage(content="hi")])
+        seen = []
+
+        todo.wrap_model_call(request, lambda req: seen.append(req) or MagicMock())
+
+        assert seen[0].system_message is not None
+        assert "write_todos" in str(seen[0].system_message.content)
+
+    def test_no_completion_reminder_or_jump_when_denied(self):
+        narrowed = self._narrowed_todo()
+        state = self._state_with_incomplete_todos()
+
+        assert narrowed.after_model(state, _runtime()) is None
+        assert asyncio.run(narrowed.aafter_model(state, _runtime())) is None
+
+    def test_completion_enforcement_preserved_when_allowed(self):
+        todo = TodoMiddleware()
+        state = self._state_with_incomplete_todos()
+
+        result = todo.after_model(state, _runtime())
+
+        assert result == {"jump_to": "model"}
+
+    def test_context_loss_detection_inert_when_denied(self):
+        narrowed = self._narrowed_todo()
+        state = self._state_with_incomplete_todos()
+
+        assert narrowed.before_model(state, _runtime()) is None
+        assert asyncio.run(narrowed.abefore_model(state, _runtime())) is None
+
+    def test_context_loss_reminder_preserved_when_allowed(self):
+        todo = TodoMiddleware()
+        state = self._state_with_incomplete_todos()
+
+        result = todo.before_model(state, _runtime())
+
+        assert result is not None
+        assert result["messages"][0].name == "todo_reminder"
+
+    def test_async_model_call_passthrough_when_denied(self):
+        narrowed = self._narrowed_todo()
+        request = _model_request([HumanMessage(content="hi")])
+        seen = []
+
+        async def handler(req):
+            seen.append(req)
+            return MagicMock()
+
+        asyncio.run(narrowed.awrap_model_call(request, handler))
+
+        assert seen == [request]
+        assert seen[0].system_message is None
+
+
+class TestLeadAgentWiring:
+    @pytest.mark.parametrize("is_bootstrap", [False, True])
+    def test_denied_declaration_absent_from_the_bound_middleware_stack(self, monkeypatch, is_bootstrap):
+        from deerflow.agents.lead_agent import agent as lead_agent_module
+        from deerflow.config.model_config import ModelConfig
+
+        config = AppConfig(
+            models=[
+                ModelConfig(
+                    name="test-model",
+                    display_name="Test model",
+                    use="langchain_openai:ChatOpenAI",
+                    model="test-model",
+                )
+            ],
+            sandbox=SandboxConfig(use="test"),
+            authorization=AuthorizationConfig(
+                enabled=True,
+                provider=AuthorizationProviderConfig(
+                    use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                    config={"roles": {"user": {"tools": {"allow": ["safe_tool", "history_read", "allowed_decl"]}}}},
+                ),
+            ),
+        )
+        config.skills.deferred_discovery = True
+        config.task_continuity.enabled = True
+
+        monkeypatch.setattr(lead_agent_module, "_resolve_model_name", lambda *args, **kwargs: "test-model")
+        monkeypatch.setattr(lead_agent_module, "create_chat_model", lambda **kwargs: object())
+        monkeypatch.setattr(lead_agent_module, "create_agent", lambda **kwargs: kwargs)
+        monkeypatch.setattr(lead_agent_module, "apply_prompt_template", lambda **kwargs: "prompt")
+        monkeypatch.setattr(lead_agent_module, "build_tracing_callbacks", lambda: [])
+        monkeypatch.setattr(lead_agent_module, "_load_enabled_available_skills", lambda *args, **kwargs: [])
+        monkeypatch.setattr(
+            lead_agent_module,
+            "build_skill_search_setup",
+            lambda *args, **kwargs: SimpleNamespace(describe_skill_tool=None, skill_names=frozenset()),
+            raising=False,
+        )
+        monkeypatch.setattr("deerflow.skills.describe.build_skill_search_setup", lead_agent_module.build_skill_search_setup)
+        monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [_tool("safe_tool")])
+        monkeypatch.setattr(lead_agent_module, "should_use_memory_tools", lambda memory_config: False)
+
+        declaring = _DeclaringMiddleware([_tool("allowed_decl"), _tool("denied_decl")])
+        monkeypatch.setattr(lead_agent_module, "build_middlewares", lambda *args, **kwargs: [declaring])
+
+        runtime_context = {"user_role": "user"}
+        if is_bootstrap:
+            runtime_context["is_bootstrap"] = True
+        result = lead_agent_module._make_lead_agent({"context": runtime_context}, app_config=config)
+
+        # Binding convention: declarations never enter final_tools ...
+        assert "allowed_decl" not in [tool.name for tool in result["tools"]]
+        assert "denied_decl" not in [tool.name for tool in result["tools"]]
+        # ... and the bound middleware stack carries only the authorized declaration.
+        (bound,) = result["middleware"]
+        assert bound is not declaring
+        assert [tool.name for tool in bound.tools] == ["allowed_decl"]
+        assert [tool.name for tool in declaring.tools] == ["allowed_decl", "denied_decl"]
+
+
+class TestClientWiring:
+    def test_denied_declaration_absent_from_the_bound_middleware_stack(self, tmp_path):
+        import deerflow.skills.storage as _storage_mod
+        from deerflow.client import DeerFlowClient
+        from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
+
+        provider = _FilterProvider(["safe_tool", "allowed_decl"])
+        app_config = MagicMock()
+        model = MagicMock()
+        model.name = "test-model"
+        model.model = "test-model"
+        model.supports_thinking = False
+        model.supports_reasoning_effort = False
+        model.model_dump.return_value = {"name": "test-model", "use": "langchain_openai:ChatOpenAI"}
+        app_config.models = [model]
+        app_config.token_usage.enabled = False
+        app_config.skills.deferred_discovery = False
+        app_config.skills.container_path = "/mnt/skills"
+        app_config.tool_search.enabled = False
+        app_config.database.checkpoint_channel_mode = "full"
+        app_config.database.checkpoint_delta.snapshot_frequency = 10
+        app_config.authorization = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(use="unused:Provider"),
+        )
+
+        _storage_mod._default_skill_storage = LocalSkillStorage(host_path=str(tmp_path))
+        declaring = _DeclaringMiddleware([_tool("allowed_decl"), _tool("denied_decl")])
+        safe_tool = _tool("safe_tool")
+
+        with patch("deerflow.client.get_app_config", return_value=app_config):
+            client = DeerFlowClient()
+        client._app_config = app_config
+
+        with (
+            patch("deerflow.client.create_chat_model"),
+            patch("deerflow.client.create_agent", return_value=MagicMock()) as mock_create_agent,
+            patch("deerflow.client.build_middlewares", return_value=[declaring]),
+            patch("deerflow.client.apply_prompt_template", return_value="prompt"),
+            patch("deerflow.client.get_enabled_skills_for_config", return_value=[]),
+            patch.object(client, "_get_tools", return_value=[safe_tool]),
+            patch("deerflow.authz.tool_filter.resolve_authorization_provider", return_value=provider),
+            patch("deerflow.agents.lead_agent.agent.resolve_authorization_provider", return_value=provider),
+            # Phase 3 resolves a skill-authorization provider through
+            # skill_filter's own import; a separate double keeps the tool-path
+            # call assertions below exact.
+            patch("deerflow.authz.skill_filter.resolve_authorization_provider", return_value=_FilterProvider([])),
+            patch("deerflow.runtime.checkpointer.get_checkpointer", return_value=None),
+        ):
+            client._ensure_agent(client._get_runnable_config("t1"), context={"user_role": "user"})
+
+        (bound,) = mock_create_agent.call_args.kwargs["middleware"]
+        assert bound is not declaring
+        assert [tool.name for tool in bound.tools] == ["allowed_decl"]
+        assert [tool.name for tool in declaring.tools] == ["allowed_decl", "denied_decl"]
+        assert [tool.name for tool in mock_create_agent.call_args.kwargs["tools"]] == ["safe_tool"]
+        # One provider instance, two passes: the declared names were the second call.
+        assert [call[2] for call in provider.calls] == [["safe_tool"], ["allowed_decl", "denied_decl"]]
+
+
+class TestDescriptorConsistency:
+    def test_descriptor_reports_the_narrowed_set(self):
+        from deerflow.agents.assembly_descriptor import build_assembly_descriptor
+
+        allowed, denied = _tool("allowed_decl"), _tool("denied_decl")
+        original = _DeclaringMiddleware([allowed, denied])
+        view, _ = _narrow([original], provider=_FilterProvider(["allowed_decl"]), outcome=_outcome([], []))
+
+        descriptor = build_assembly_descriptor(
+            namespace="deerflow",
+            agent_name="test",
+            requested_model="test-model",
+            effective_model="test-model",
+            model_config=None,
+            thinking_enabled=False,
+            reasoning_effort=None,
+            rendered_base_prompt="prompt",
+            prompt_template_id="test",
+            tools=[_tool("regular")],
+            middlewares=view,
+            deferred_names=frozenset(),
+            enabled_skills=[],
+            effective_policies={},
+        )
+
+        reported_names = [tool.name for tool in descriptor.tools]
+        assert "allowed_decl" in reported_names
+        assert "denied_decl" not in reported_names

backend/tests/test_skills_authorization.py

@@ -1075,7 +1075,9 @@ def test_subagent_executor_resolves_skill_authorization_for_chain(monkeypatch):
     executor.authz_attributes = None
     executor.tools = []
 
-    executor._create_agent(tools=[])
+    import asyncio
+
+    asyncio.run(executor._create_agent(tools=[]))
 
     wired = captured["skill_authorization"]
     assert isinstance(wired, ResolvedSkillAuthorization)
@@ -1441,7 +1443,9 @@ def test_subagent_executor_shares_one_skill_authorization_instance(monkeypatch):
     executor.tools = []
 
     asyncio.run(executor._load_skills())
-    executor._create_agent(tools=[])
+    import asyncio
+
+    asyncio.run(executor._create_agent(tools=[]))
 
     layer1 = filter_captured["authorization"]
     layer2 = middleware_captured["skill_authorization"]

backend/tests/test_subagent_checkpointer_isolation.py

@@ -11,6 +11,7 @@ once, and extracts the last AIMessage. It never resumes, so persistence
 is unnecessary and inheriting the parent checkpointer is harmful.
 """
 
+import asyncio
 import sys
 from types import ModuleType, SimpleNamespace
 from unittest.mock import MagicMock
@@ -133,7 +134,7 @@ class TestSubagentCheckpointerIsolation:
         monkeypatch.setattr(executor_module, "create_chat_model", fake_create_chat_model)
         monkeypatch.setattr(executor_module, "resolve_subagent_model_name", lambda config, parent, app_config=None: "test-model")
 
-        result = executor._create_agent()
+        result = asyncio.run(executor._create_agent())
 
         assert captured_kwargs.get("checkpointer") is False, f"Expected checkpointer=False in create_agent() kwargs, got: {captured_kwargs.get('checkpointer')!r}"
         assert result.checkpointer is False

backend/tests/test_subagent_executor.py

@@ -88,6 +88,7 @@ def _setup_executor_classes():
     original_modules = {name: sys.modules.get(name) for name in _MOCKED_MODULE_NAMES}
     original_executor = sys.modules.get("deerflow.subagents.executor")
     original_audit_context = sys.modules.get("deerflow.agents.middlewares.audit_context")
+    original_tool_declarations = sys.modules.get("deerflow.agents.middlewares.tool_declarations")
     original_tool_search = sys.modules.get("deerflow.tools.builtins.tool_search")
     original_sandbox_provider = sys.modules.get("deerflow.sandbox.sandbox_provider")
     original_sandbox_overwrite = sys.modules.get("deerflow.sandbox.overwrite")
@@ -96,6 +97,7 @@ def _setup_executor_classes():
     # with cycle-breaking test doubles. Keeping the concrete leaf modules in
     # sys.modules makes this fixture independent of test collection order.
     audit_context_module = importlib.import_module("deerflow.agents.middlewares.audit_context")
+    tool_declarations_module = importlib.import_module("deerflow.agents.middlewares.tool_declarations")
     tool_search_module = importlib.import_module("deerflow.tools.builtins.tool_search")
     sandbox_provider_module = importlib.import_module("deerflow.sandbox.sandbox_provider")
     sandbox_overwrite_module = importlib.import_module("deerflow.sandbox.overwrite")
@@ -113,6 +115,7 @@ def _setup_executor_classes():
     storage_module.get_or_new_user_skill_storage = lambda user_id, **kwargs: SimpleNamespace(load_skills=lambda *, enabled_only: [])
     sys.modules["deerflow.skills.storage"] = storage_module
     sys.modules["deerflow.agents.middlewares.audit_context"] = audit_context_module
+    sys.modules["deerflow.agents.middlewares.tool_declarations"] = tool_declarations_module
     sys.modules["deerflow.tools.builtins.tool_search"] = tool_search_module
     sys.modules["deerflow.sandbox.sandbox_provider"] = sandbox_provider_module
     sys.modules["deerflow.sandbox.overwrite"] = sandbox_overwrite_module
@@ -164,6 +167,10 @@ def _setup_executor_classes():
         sys.modules["deerflow.agents.middlewares.audit_context"] = original_audit_context
     else:
         sys.modules.pop("deerflow.agents.middlewares.audit_context", None)
+    if original_tool_declarations is not None:
+        sys.modules["deerflow.agents.middlewares.tool_declarations"] = original_tool_declarations
+    else:
+        sys.modules.pop("deerflow.agents.middlewares.tool_declarations", None)
     if original_tool_search is not None:
         sys.modules["deerflow.tools.builtins.tool_search"] = original_tool_search
     else:
@@ -178,6 +185,15 @@ def _setup_executor_classes():
         sys.modules.pop("deerflow.sandbox.overwrite", None)
 
 
+def _async_create_agent_double(agent):
+    """Async stand-in for SubagentExecutor._create_agent (the real one is async)."""
+
+    async def _create(*_args, **_kwargs):
+        return agent
+
+    return _create
+
+
 # Helper classes that wrap real classes for testing
 class MockHumanMessage:
     """Mock HumanMessage for testing - wraps real class from fixture."""
@@ -364,7 +380,7 @@ class TestAgentConstruction:
         provider = object()
         executor._authz_provider = provider
 
-        result = executor._create_agent()
+        result = asyncio.run(executor._create_agent())
 
         assert result is agent
         assert captured["middlewares"]["authorization_provider"] is provider
@@ -393,6 +409,151 @@ class TestAgentConstruction:
         assert captured["agent"]["tools"] == []
         assert captured["agent"]["system_prompt"] is None  # system_prompt is merged into initial state messages
 
+    def test_create_agent_narrows_denied_middleware_declared_tools(
+        self,
+        classes,
+        base_config,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        """Layer 1 declaration pass: a denied middleware-declared tool is removed
+        from the bound stack on an independent copy, decided by the same provider
+        instance and principal as the ordinary pass."""
+        from langchain.agents.middleware import AgentMiddleware
+        from langchain_core.tools import StructuredTool
+
+        from deerflow.subagents import executor as executor_module
+
+        SubagentExecutor = classes["SubagentExecutor"]
+
+        def _tool(name):
+            return StructuredTool.from_function(lambda: name, name=name, description=name)
+
+        class _Provider:
+            name = "test"
+
+            def __init__(self):
+                self.calls = []
+
+            def authorize(self, request):
+                from deerflow.authz.provider import AuthzDecision
+
+                return AuthzDecision(allow=True)
+
+            async def aauthorize(self, request):
+                return self.authorize(request)
+
+            def filter_resources(self, principal, resource_type, candidates):
+                self.calls.append((principal, list(candidates)))
+                return [candidate for candidate in candidates if candidate == "allowed_decl"]
+
+        class _Declaring(AgentMiddleware):
+            def __init__(self, tools):
+                super().__init__()
+                self.tools = tools
+
+        app_config = SimpleNamespace(
+            models=[SimpleNamespace(name="default-model")],
+            authorization=SimpleNamespace(enabled=True, fail_closed=True, default_role="user"),
+        )
+        declaring = _Declaring([_tool("allowed_decl"), _tool("denied_decl")])
+        captured: dict[str, dict] = {}
+
+        def fake_build_subagent_runtime_middlewares(**kwargs):
+            captured["middlewares"] = kwargs
+            return [declaring]
+
+        def fake_create_agent(**kwargs):
+            captured["agent"] = kwargs
+            return MagicMock()
+
+        monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: MagicMock())
+        monkeypatch.setattr(executor_module, "create_agent", fake_create_agent)
+        monkeypatch.setitem(
+            sys.modules,
+            "deerflow.agents.middlewares.tool_error_handling_middleware",
+            _module(
+                "deerflow.agents.middlewares.tool_error_handling_middleware",
+                build_subagent_runtime_middlewares=fake_build_subagent_runtime_middlewares,
+            ),
+        )
+
+        executor = SubagentExecutor(
+            config=base_config,
+            tools=[],
+            app_config=app_config,
+            parent_model="parent-model",
+        )
+        provider = _Provider()
+        executor._authz_provider = provider
+        executor._authz_context = {"user_role": "user"}
+        from deerflow.agents.middlewares.tool_declarations import LayerOneOutcome
+
+        executor._layer_one_outcome = LayerOneOutcome(submitted=frozenset({"regular"}), allowed=frozenset({"regular"}))
+        # Phase 3's skill-authorization resolution needs a real AuthorizationConfig;
+        # this test's SimpleNamespace app_config deliberately lacks one — the
+        # declaration pass under test uses the explicitly-set provider instead.
+        monkeypatch.setattr(executor, "_resolve_skill_authorization", lambda: None)
+
+        asyncio.run(executor._create_agent())
+
+        (bound,) = captured["agent"]["middleware"]
+        assert bound is not declaring
+        assert [tool.name for tool in bound.tools] == ["allowed_decl"]
+        # The caller-owned instance keeps every declaration.
+        assert [tool.name for tool in declaring.tools] == ["allowed_decl", "denied_decl"]
+        # Only the never-submitted names went to the provider, with the run's principal.
+        assert len(provider.calls) == 1
+        principal, candidates = provider.calls[0]
+        assert candidates == ["allowed_decl", "denied_decl"]
+        assert principal.role == "user"
+
+    def test_create_agent_warns_when_the_declared_tool_pass_is_skipped_with_a_provider(
+        self,
+        classes,
+        base_config,
+        monkeypatch: pytest.MonkeyPatch,
+        caplog,
+    ):
+        """Fail-open must not be silent: a configured provider without the seeded
+        Layer-1 state skips the declaration pass but logs a warning."""
+        from deerflow.subagents import executor as executor_module
+
+        SubagentExecutor = classes["SubagentExecutor"]
+
+        app_config = SimpleNamespace(
+            models=[SimpleNamespace(name="default-model")],
+            authorization=SimpleNamespace(enabled=True, fail_closed=True, default_role="user"),
+        )
+
+        monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: MagicMock())
+        monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: MagicMock())
+        monkeypatch.setitem(
+            sys.modules,
+            "deerflow.agents.middlewares.tool_error_handling_middleware",
+            _module(
+                "deerflow.agents.middlewares.tool_error_handling_middleware",
+                build_subagent_runtime_middlewares=lambda **kwargs: [],
+            ),
+        )
+
+        executor = SubagentExecutor(
+            config=base_config,
+            tools=[],
+            app_config=app_config,
+            parent_model="parent-model",
+        )
+        executor._authz_provider = object()
+        # _layer_one_outcome and _authz_context deliberately unset.
+        # Same SimpleNamespace limitation as above: skip Phase 3 resolution so
+        # the skip-warning branch is what runs.
+        monkeypatch.setattr(executor, "_resolve_skill_authorization", lambda: None)
+
+        with caplog.at_level(logging.WARNING, logger="deerflow.subagents.executor"):
+            asyncio.run(executor._create_agent())
+
+        assert "skipping the middleware-declared tool pass" in caplog.text
+        assert "_layer_one_outcome" in caplog.text
+
     def test_create_agent_scales_max_turns_into_a_super_step_budget(
         self,
         classes,
@@ -435,7 +596,7 @@ class TestAgentConstruction:
             app_config=SimpleNamespace(models=[SimpleNamespace(name="default-model")]),
             parent_model="parent-model",
         )
-        executor._create_agent()
+        asyncio.run(executor._create_agent())
 
         # model + tools + two after_model nodes, once per turn.
         assert executor._recursion_limit == base_config.max_turns * 4
@@ -485,7 +646,7 @@ class TestAgentConstruction:
             parent_model="parent-model",
         )
         with caplog.at_level(logging.WARNING, logger=executor_module.logger.name):
-            executor._create_agent()
+            asyncio.run(executor._create_agent())
 
         assert "_Jumper.after_model" in caplog.text
         assert "lower bound" in caplog.text
@@ -1374,7 +1535,7 @@ class TestAgentConstruction:
         deferred_setup = DeferredToolSetup(object(), frozenset({"mcp_calc"}), "hash123")
         executor = SubagentExecutor(config=base_config, tools=[], app_config=app_config, parent_model="parent-model")
 
-        executor._create_agent(tools=[], deferred_setup=deferred_setup)
+        asyncio.run(executor._create_agent(tools=[], deferred_setup=deferred_setup))
 
         assert captured["middlewares"]["deferred_setup"] is deferred_setup
 
@@ -4279,7 +4440,7 @@ async def test_subagent_mcp_uses_captured_thread_incarnation(classes, monkeypatc
         return ({"messages": [classes["AIMessage"](content="", tool_calls=[{"name": tool.name, "args": {}, "id": "probe"}])]}, [], None)
 
     monkeypatch.setattr(executor, "_build_initial_state", build_initial_state)
-    monkeypatch.setattr(executor, "_create_agent", lambda *args, **kwargs: child)
+    monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(child))
     result = await executor._aexecute("probe MCP")
 
     if expected_scope is None:
@@ -4331,7 +4492,7 @@ class TestSubagentCheckpointLineage:
             return ({"messages": [classes["HumanMessage"](content=task)]}, [], None)
 
         monkeypatch.setattr(executor, "_build_initial_state", build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *args, **kwargs: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4433,7 +4594,7 @@ class TestSubagentCheckpointLineage:
             return ({"messages": [classes["HumanMessage"](content=task)]}, [], None)
 
         monkeypatch.setattr(executor, "_build_initial_state", build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *args, **kwargs: child_graph)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(child_graph))
 
         async def delegate(_state):
             task_id = executor.execute_async("run the child graph")
@@ -4544,7 +4705,7 @@ class TestSubagentTracingWiring:
         executor = self._make_executor(classes, user_id="alice")
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         result = await executor._aexecute("do something")
 
@@ -4585,7 +4746,7 @@ class TestSubagentTracingWiring:
         executor = self._make_executor(classes, deerflow_trace_id="parent-trace-1")
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         seen: list[str | None] = []
         original = executor._aexecute_admitted
@@ -4628,7 +4789,7 @@ class TestSubagentTracingWiring:
         executor = self._make_executor(classes, user_id="alice", name="general_purpose", deerflow_trace_id="gateway-trace-sub")
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4659,7 +4820,7 @@ class TestSubagentTracingWiring:
         executor = self._make_executor(classes, user_id="alice")
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4689,7 +4850,7 @@ class TestSubagentTracingWiring:
         executor = self._make_executor(classes, user_id=None)
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4732,7 +4893,7 @@ class TestSubagentTracingWiring:
         )
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4761,7 +4922,7 @@ class TestSubagentTracingWiring:
         executor = self._make_executor(classes, user_id="alice")
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4857,7 +5018,7 @@ class TestSubagentGuardrailAttribution:
         )
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4890,7 +5051,7 @@ class TestSubagentGuardrailAttribution:
         )
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4916,7 +5077,7 @@ class TestSubagentGuardrailAttribution:
         )
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4942,7 +5103,7 @@ class TestSubagentGuardrailAttribution:
         )
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -4981,7 +5142,7 @@ class TestSubagentGuardrailAttribution:
         )
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -5003,7 +5164,7 @@ class TestSubagentGuardrailAttribution:
         executor = self._make_executor(classes)
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -5042,7 +5203,7 @@ class TestSubagentGuardrailAttribution:
         )
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -5073,7 +5234,7 @@ class TestSubagentGuardrailAttribution:
         )
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -5108,7 +5269,7 @@ class TestSubagentGuardrailAttribution:
         assert executor.authz_attributes == {"dept": "eng"}
         fake_agent = _FakeStreamAgent()
         monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
-        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(fake_agent))
 
         await executor._aexecute("do something")
 
@@ -5207,7 +5368,7 @@ class TestInterruptedTokenUsage:
             execution_capacity=SubagentExecutionCapacity(SubagentRuntimeConfig()),
         )
         monkeypatch.setattr(executor, "_build_initial_state", AsyncMock(return_value=({"messages": [classes["HumanMessage"](content="Echo ok")]}, [], None)))
-        monkeypatch.setattr(executor, "_create_agent", lambda *args, **kwargs: graph)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(graph))
         monkeypatch.setattr(executor_module, "build_tracing_callbacks", lambda: [])
         monkeypatch.setattr(executor_module, "inject_langfuse_metadata", lambda *args, **kwargs: None)
         original_terminal = classes["SubagentResult"].try_set_terminal
@@ -5288,7 +5449,7 @@ class TestInterruptedTokenUsage:
         graph = SimpleNamespace(astream=lambda state, config, **kwargs: Stream(config["callbacks"][0]))
         executor = classes["SubagentExecutor"](config=base_config, tools=[])
         monkeypatch.setattr(executor, "_build_initial_state", AsyncMock(return_value=({}, [], None)))
-        monkeypatch.setattr(executor, "_create_agent", lambda *args, **kwargs: graph)
+        monkeypatch.setattr(executor, "_create_agent", _async_create_agent_double(graph))
         holder = classes["SubagentResult"](task_id="cleanup-usage", trace_id="trace", status=classes["SubagentStatus"].RUNNING)
         running = asyncio.create_task(executor._aexecute_admitted("Task", holder))
         try:

backend/tests/test_worker_stream_subgraph_namespace.py

@@ -365,6 +365,15 @@ def _collect_ids(payload: object) -> set[str]:
     return ids
 
 
+def _async_returning(value):
+    """Wrap *value* in an async callable (the real _create_agent is async)."""
+
+    async def _create(*_args, **_kwargs):
+        return value
+
+    return _create
+
+
 def _build_delegating_parent_graph(executor_module, monkeypatch, *, child_emits_error_fallback: bool = False, emit_task_running: bool = False):
     """Real parent graph whose node delegates a scripted child through the
     real ``SubagentExecutor`` and emits ``task_*`` custom events the way the
@@ -445,7 +454,7 @@ def _build_delegating_parent_graph(executor_module, monkeypatch, *, child_emits_
         return ({"messages": [HumanMessage(content=task, id="child-task-sentinel")]}, [], None)
 
     monkeypatch.setattr(executor, "_build_initial_state", build_initial_state)
-    monkeypatch.setattr(executor, "_create_agent", lambda *_args, **_kwargs: child_graph)
+    monkeypatch.setattr(executor, "_create_agent", _async_returning(child_graph))
 
     async def delegate(_state):
         writer = get_stream_writer()

docs/full-stack-plugins.md

@@ -242,6 +242,30 @@ as `plugin_actions` and `plugin_management`. A role with no policy for a
 resource is **unrestricted** for it, so list these keys explicitly wherever you
 want to constrain them; `config.example.yaml` shows the shape.
 
+### Middleware-declared tools
+
+A contributed middleware may declare tools through LangChain's `middleware.tools`
+attribute. When authorization is enabled, every build collects those declarations
+after the middleware stack is assembled and applies the same `tools` policy as for
+explicitly configured tools, seeded with the build's Layer-1 verdicts (a name
+denied for the build cannot be resurrected by declaring it). A middleware whose
+declaration is denied is replaced in that build's stack by an independent,
+state-preserving copy carrying only the authorized declarations — the contributor's
+own instance is never mutated, so a later build under a more permissive principal
+binds the declaration again. Declarations must be `BaseTool` instances with a
+usable name: anything else (for example a plain callable, which LangChain would
+otherwise auto-convert and bind unchecked) cannot be authorized by name and is
+removed from the bound stack with a warning when authorization is enabled — wrap
+callables in a `StructuredTool`. Three author-visible rules follow: declare tools as a
+plain instance attribute holding a `list` or `tuple` (a `tools` property cannot be
+narrowed safely and fails the build, and any other container shape — a `set`,
+generator, `dict_values`, … — would be iterated and bound by LangChain as-is, so it
+likewise fails the build), and if you customize `__copy__`, it must return an
+independent copy that preserves current instance state — returning `self` or
+rebuilding from constructor arguments fails the build loudly rather than silently
+restoring a denied tool. With `authorization.enabled: false` nothing is collected or
+narrowed.
+
 ### Guarding a contributed management route
 
 A contributed router is not covered by the registered-action dispatcher, so it