diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
index f1d0c3426e0..20b8a0bda64 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
+++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
@@ -61,7 +61,7 @@ it to that middleware's declaration in the same change.
    forgeries). Consumers pop it; the publisher and the consumer share only that
    contract module.
 10. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations for security logging before tool execution. Command classification is **defense-in-depth and audit, not a security boundary** — the sandbox itself is the isolation boundary. Command substitution is judged by *position*, not by the presence of `$(`: a substitution in **command position** (`$(curl url)`, `` `curl url` ``, the word after a `|`/`&&`/`;`, or any `eval`/`source` argument) executes fetched or interpreted content and is blocked, while **value position** (`x=$(curl url)`, `echo $(curl url)`, an argument, a `for` word list) only captures output and passes (#4611). `_HIGH_RISK_COMMAND_POSITION_PATTERNS` is therefore matched anchored against each split sub-command, never against the whole compound string, and `_split_compound_command(split_pipes=True)` supplies those sub-commands; rules that span a pipe (`| sh`, `base64 -d | ...`) still rely on `_classify_command`'s whole-command Pass 1. `_COMMAND_POSITION_PREFIX` extends the anchor over leading variable assignments and exec wrappers (`FOO=1 $(curl url)`, `env`/`command`/`builtin`/`exec`/`nohup`/`time`/`sudo`/`doas`), which are still command position; its assignment branch requires whitespace before the substitution, which is exactly what keeps `x=$(curl url)` in value position. Two execution contexts are deliberately **position-blind** and matched against the whole command in Pass 1, because they execute what they receive wherever they appear (including as an argument to something else, e.g. `xargs sh -c "$(curl url)"`): an `eval`/`source` argument, and an interpreter's **code-string flag** — `-c` (shells, `python`), `-e` (`perl`/`ruby`/`node`), `-p` (`perl`/`node`), `-r` (`php`) — plus the here-string (`<<<`) that reaches the same place through stdin. All three substitution spellings (`$(cmd`, `<(cmd`, `` `cmd ``) share one `_RISKY_SUBSTITUTION` opener so a rule cannot cover one spelling and miss another. An unquoted newline splits like `;`, because it separates statements the same way: leaving it joined let `echo hi\n$(curl url)` evade the anchored rules that its `;` spelling triggers. A heredoc body is data rather than statements, so `_split_compound_command` records headers (`<<EOF`, `<<-EOF`, `<<'EOF'`) and consumes their bodies verbatim at the newline that starts them — otherwise a body line beginning with `$(curl url)` would be promoted to a command position the shell never creates. Two things that look like headers must not open one, or a body that never terminates swallows every following statement: `<<<` is a here-string (both a lookahead and a lookbehind are needed, or the trailing `<<` of `<<< "text"` reads as a heredoc with delimiter `text`), and a `<<` inside `$(( ... ))` / `(( ... ))` is a bit shift, so arithmetic depth is tracked alongside the quote flags. That is a heuristic, not shell parsing: it exists only to avoid manufacturing command positions *and* to avoid destroying real ones. An unterminated body consumes the rest of the string; an unclosed `((` only disables heredoc detection, so newlines keep splitting and the failure direction stays towards seeing more command positions rather than fewer. Known, deliberate gaps: process substitution outside `eval`/`source` (`. <(curl u)`) is not detected — closing it would require real shell parsing, which is out of scope for this layer. Two-step forms (`x=$(curl u); eval "$x"`) are inherent rather than incidental: any rule that allows output capture allows the first statement, and connecting it to the later `eval` needs dataflow analysis, not pattern matching. There is currently no config gate: the middleware is appended unconditionally in `_build_runtime_middlewares`, so it applies to both the lead agent and subagents.
-11. **ReadBeforeWriteMiddleware** - *(optional, if `read_before_write.enabled`, default on)* Outermost write gate (issue #3857): `read_file` stamps a content hash onto its ToolMessage; `write_file` (append/overwrite-existing) and `str_replace` are blocked unless the newest mark for that path matches the file's current hash. Sits outside ToolProgressMiddleware and ToolErrorHandlingMiddleware so a blocked write returns immediately without consuming a ToolProgress slot. Blocked results call `normalize_tool_result` directly to stamp `deerflow_tool_meta` (`recoverable_by_model=True`) before returning, keeping the result well-formed for any outer consumer. Marks live on messages, so summarization dropping the read result invalidates the gate automatically; writes never refresh marks, forcing a re-read between consecutive edits. Gate check + tool execution are serialized per (thread, path) so same-turn parallel writes cannot reuse one stale mark; on sandboxes whose `read_file` reports failures as `"Error: ..."` strings instead of raising (AIO/E2B), uninspectable targets fail open (creation proceeds, no mark stamped)
+11. **ReadBeforeWriteMiddleware** - *(optional, if `read_before_write.enabled`, default on)* Outermost write gate (issue #3857): `read_file` stamps a content hash onto its ToolMessage; `write_file` (append/overwrite-existing) and `str_replace` are blocked unless the newest mark for that path matches the file's current hash. Sits outside ToolProgressMiddleware and ToolErrorHandlingMiddleware so a blocked write returns immediately without consuming a ToolProgress slot. Blocked results call `normalize_tool_result` directly to stamp `deerflow_tool_meta` (`recoverable_by_model=True`) before returning, keeping the result well-formed for any outer consumer. Marks live on messages, so summarization dropping the read result invalidates the gate automatically; writes never refresh marks, forcing a re-read between consecutive edits. Gate check + tool execution are serialized per (thread, path) so same-turn parallel writes cannot reuse one stale mark; on sandboxes whose `read_file` reports failures as `"Error: ..."` strings instead of raising (AIO/E2B), uninspectable targets fail open (creation proceeds, no mark stamped). The middleware also owns the sandbox authorization scope for these composed calls: pre-write inspection, the tool body, and post-read hashing share one sync/async provider decision, while `SandboxAuthorizationError` bypasses the generic inspection fail-open paths and becomes an error ToolMessage.
 12. **ToolProgressMiddleware** - *(optional, if `tool_progress.enabled`)* State-machine-based stagnation guard (RFC #3177). Outer wrapper around ToolErrorHandlingMiddleware so its `wrap_tool_call` receives results already stamped with `deerflow_tool_meta`. Tracks per-(thread, tool) consecutive "no-new-info" calls across three error categories: (a) `recoverable_by_model=True` (no_results, not_found, permission, Jaccard-duplicate success): ACTIVE → WARNED (terminal — hint re-injected on each subsequent problem); (b) `recoverable_by_model=False, action≠stop` (rate_limited, transient): ACTIVE → WARNED → BLOCKED after `warn_escalation_count` more problems; (c) `recoverable_by_model=False, action=stop` (auth, config, internal): immediately BLOCKED on first occurrence. **Division of labor with LoopDetectionMiddleware:** ToolProgressMiddleware is a result-quality guard — fires after tool execution and blocks specific tools that stop producing new information; LoopDetectionMiddleware is a call-pattern guard — fires after the model responds and hard-stops the whole turn when the model repeatedly issues identical tool_calls. Both can inject HumanMessage hints in the same model call without conflict; neither reads the other's internal state.
 13. **ToolReceiptMiddleware + ToolErrorHandlingMiddleware** - `ToolReceiptMiddleware` is *(optional, if `verification.receipts_enabled`, default on)*. It is the **outermost `wrap_tool_call` layer** — registered ahead of entries 9-12 — because Guardrail/SandboxAudit/ReadBeforeWrite/ToolProgress can short-circuit a call with their own ToolMessage (and SandboxAudit rebuilds medium-risk results); an inner receipt layer would silently gap the ledger on those results (ordering constraints in `deerflow.extensions.ordering`). Normal results still carry the `deerflow_tool_meta` status ToolErrorHandlingMiddleware stamps on the inner return path; short-circuit messages self-stamp meta or fall back to `message.status`. It stamps deterministic provenance (tool name, status, args/output hashes, byte count, timestamp) onto direct `ToolMessage` results and every matching `ToolMessage` carried in `Command.update.messages`, including delegated `task`, `present_file`, `view_image`, and `tool_search` results; before model calls it derives a hidden receipt ledger (display ids r1..rN) from message state, and when the 2,000-character budget is exceeded the newest receipts are retained in chronological order with their original ids plus an older-receipts omission marker. Rendering returns both the text and its retained receipt subset; every response that received a ledger carries only that exact server-owned subset, never omitted receipts. Snapshot validation accepts a strictly consecutive positive original-id range (for example `r24`–`r30`) rather than requiring `r1`, so subagent terminal citation verification resolves ids against evidence present in the citing turn even when later summarization drops and renumbers tool messages. Model-generated citation IDs are digit-bounded before integer conversion; oversized IDs are ignored as malformed input rather than raising through task write-back. Citation parsing deduplicates exact `(id, anchor)` pairs, not IDs alone, so repeated identical references stay compact while every distinct anchor claim is verified. Gateway strips delegated receipts/verdicts from external messages. `ToolErrorHandlingMiddleware` receives `AppConfig`, converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting, stamps every result with `deerflow_tool_meta` (status / error_type / recoverable_by_model / recommended_next_action / source) via `tool_result_meta.normalize_tool_result`, stamps structured metadata for task exception wrappers, and stamps skill-read metadata for downstream durable-context capture. Task tool result text is generated from the same status/result/error inputs as the structured metadata so callers do not hand-write a second protocol string.
 
diff --git a/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py
index 1198d75773f..5108b0c58f0 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py
@@ -39,8 +39,13 @@
 from langgraph.prebuilt.tool_node import ToolCallRequest
 from langgraph.types import Command
 
-from deerflow.agents.middlewares.tool_result_meta import normalize_tool_result
-from deerflow.sandbox.tools import read_current_file_content
+from deerflow.agents.middlewares.tool_result_meta import normalize_tool_result, stamp_exception_meta
+from deerflow.sandbox.exceptions import SandboxAuthorizationError
+from deerflow.sandbox.tools import (
+    read_current_file_content,
+    sandbox_authorization_scope,
+    sandbox_authorization_scope_async,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -105,21 +110,29 @@ def wrap_tool_call(
             path = self._requested_path(request)
             if path is None:
                 return handler(request)
-            with self._lock_for(request, path):
-                blocked = self._check_write_gate(request)
-                if blocked is not None:
-                    # Stamp deerflow_tool_meta so ToolProgressMiddleware can classify
-                    # the blocked write even though it bypasses ToolErrorHandlingMiddleware.
-                    return normalize_tool_result(blocked)
-                return handler(request)
+            try:
+                with sandbox_authorization_scope(request.runtime):
+                    with self._lock_for(request, path):
+                        blocked = self._check_write_gate(request)
+                        if blocked is not None:
+                            # Stamp deerflow_tool_meta so ToolProgressMiddleware can classify
+                            # the blocked write even though it bypasses ToolErrorHandlingMiddleware.
+                            return normalize_tool_result(blocked)
+                        return handler(request)
+            except SandboxAuthorizationError as exc:
+                return self._authorization_error_result(request, exc)
         if name in _READ_TOOLS:
             path = self._requested_path(request)
             if path is None:
                 return handler(request)
-            with self._lock_for(request, path):
-                result = handler(request)
-                self._attach_read_mark(request, result)
-                return result
+            try:
+                with sandbox_authorization_scope(request.runtime):
+                    with self._lock_for(request, path):
+                        result = handler(request)
+                        self._attach_read_mark(request, result)
+                        return result
+            except SandboxAuthorizationError as exc:
+                return self._authorization_error_result(request, exc)
         return handler(request)
 
     @override
@@ -133,32 +146,54 @@ async def awrap_tool_call(
             path = self._requested_path(request)
             if path is None:
                 return await handler(request)
-            # threading.Lock may be released from a different thread than the
-            # acquiring one, so acquiring in a worker thread and releasing on
-            # the event-loop thread is safe.
-            lock = self._lock_for(request, path)
-            await asyncio.to_thread(lock.acquire)
             try:
-                blocked = await asyncio.to_thread(self._check_write_gate, request)
-                if blocked is not None:
-                    return normalize_tool_result(blocked)
-                return await handler(request)
-            finally:
-                lock.release()
+                async with sandbox_authorization_scope_async(request.runtime):
+                    # threading.Lock may be released from a different thread than the
+                    # acquiring one, so acquiring in a worker thread and releasing on
+                    # the event-loop thread is safe.
+                    lock = self._lock_for(request, path)
+                    await asyncio.to_thread(lock.acquire)
+                    try:
+                        blocked = await asyncio.to_thread(self._check_write_gate, request)
+                        if blocked is not None:
+                            return normalize_tool_result(blocked)
+                        return await handler(request)
+                    finally:
+                        lock.release()
+            except SandboxAuthorizationError as exc:
+                return self._authorization_error_result(request, exc)
         if name in _READ_TOOLS:
             path = self._requested_path(request)
             if path is None:
                 return await handler(request)
-            lock = self._lock_for(request, path)
-            await asyncio.to_thread(lock.acquire)
             try:
-                result = await handler(request)
-                await asyncio.to_thread(self._attach_read_mark, request, result)
-                return result
-            finally:
-                lock.release()
+                async with sandbox_authorization_scope_async(request.runtime):
+                    lock = self._lock_for(request, path)
+                    await asyncio.to_thread(lock.acquire)
+                    try:
+                        result = await handler(request)
+                        await asyncio.to_thread(self._attach_read_mark, request, result)
+                        return result
+                    finally:
+                        lock.release()
+            except SandboxAuthorizationError as exc:
+                return self._authorization_error_result(request, exc)
         return await handler(request)
 
+    @staticmethod
+    def _authorization_error_result(request: ToolCallRequest, exc: SandboxAuthorizationError) -> ToolMessage:
+        """Return the normal tool-level denial instead of failing open or the run."""
+        tool_name = str(request.tool_call.get("name") or "unknown_tool")
+        tool_call_id = str(request.tool_call.get("id") or "missing-tool-call-id")
+        detail = str(exc).strip() or exc.__class__.__name__
+        message = ToolMessage(
+            content=f"Error: {detail}",
+            tool_call_id=tool_call_id,
+            name=tool_name,
+            status="error",
+        )
+        return stamp_exception_meta(message, f"{exc.__class__.__name__}: {detail}")
+
     # -- locking ---------------------------------------------------------
 
     def _lock_for(self, request: ToolCallRequest, path: str) -> threading.Lock:
@@ -193,6 +228,8 @@ def _check_write_gate(self, request: ToolCallRequest) -> ToolMessage | None:
         except FileNotFoundError:
             # write_file creates the file; str_replace surfaces its own error.
             return None
+        except SandboxAuthorizationError:
+            raise
         except Exception:
             logger.warning("read-before-write gate could not inspect %r; allowing the write (fail-open)", path, exc_info=True)
             return None
@@ -246,6 +283,8 @@ def _attach_read_mark(self, request: ToolCallRequest, result: ToolMessage | Comm
             return
         try:
             content = self._content_reader(request.runtime, path)
+        except SandboxAuthorizationError:
+            raise
         except Exception:
             logger.debug("read-before-write mark skipped for %r: file not hashable", path, exc_info=True)
             return
diff --git a/backend/packages/harness/deerflow/authz/runtime.py b/backend/packages/harness/deerflow/authz/runtime.py
index b0f7756680e..6ba1957e161 100644
--- a/backend/packages/harness/deerflow/authz/runtime.py
+++ b/backend/packages/harness/deerflow/authz/runtime.py
@@ -1,30 +1,44 @@
-"""Provider factory — resolves and constructs the configured AuthorizationProvider.
+"""Provider factory — discovers and constructs the configured provider.
 
-This is the single entry point for creating an authorization provider from
-``AuthorizationConfig``. It does not cache instances (Phase 1B resolves once
-per agent build and passes the same instance to Layer 1 and Layer 2).
+The two-phase API lets async callers offload class-path discovery/import while
+constructing loop-affine providers on their running event loop. The synchronous
+``resolve_authorization_provider`` convenience function composes both phases.
+Instances are not cached (Phase 1B resolves once per agent build and passes the
+same instance to Layer 1 and Layer 2).
 """
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+from typing import Any
+
 from deerflow.authz.provider import AuthorizationProvider
 from deerflow.config.authorization_config import AuthorizationConfig
 from deerflow.reflection import resolve_variable
 
 
-def resolve_authorization_provider(
+@dataclass(frozen=True, slots=True)
+class AuthorizationProviderSpec:
+    """A discovered provider class and its constructor inputs."""
+
+    class_path: str
+    provider_cls: type[Any]
+    kwargs: dict[str, Any]
+
+
+def resolve_authorization_provider_spec(
     config: AuthorizationConfig,
-) -> AuthorizationProvider | None:
-    """Resolve the authorization provider from config.
+) -> AuthorizationProviderSpec | None:
+    """Discover a provider class without constructing the provider.
 
     Returns:
-        A constructed ``AuthorizationProvider`` instance, or ``None`` if
-        authorization is disabled.
+        Constructor inputs for the configured provider, or ``None`` if
+        authorization is disabled. This discovery phase may import a custom
+        module and is safe to offload from an async event loop.
 
     Raises:
         ValueError: If ``enabled`` is True but no provider is configured,
-            or if the class path is invalid / construction fails / the
-            instance does not satisfy the ``AuthorizationProvider`` Protocol.
+            or if the class path is invalid.
     """
     if not config.enabled:
         return None
@@ -39,13 +53,24 @@ def resolve_authorization_provider(
         raise ValueError(f"Failed to resolve authorization provider class '{class_path}': {err}") from err
 
     kwargs = dict(config.provider.config) if config.provider.config else {}
+    return AuthorizationProviderSpec(class_path=class_path, provider_cls=provider_cls, kwargs=kwargs)
+
+
+def construct_authorization_provider(
+    spec: AuthorizationProviderSpec | None,
+    config: AuthorizationConfig,
+) -> AuthorizationProvider | None:
+    """Construct and validate a previously discovered provider spec."""
+    if spec is None:
+        return None
+
     try:
-        instance = provider_cls(**kwargs)
+        instance = spec.provider_cls(**spec.kwargs)
     except Exception as err:
-        raise ValueError(f"Failed to construct authorization provider '{class_path}': {err}") from err
+        raise ValueError(f"Failed to construct authorization provider '{spec.class_path}': {err}") from err
 
     if not isinstance(instance, AuthorizationProvider):
-        raise ValueError(f"Authorization provider '{class_path}' does not satisfy the AuthorizationProvider Protocol")
+        raise ValueError(f"Authorization provider '{spec.class_path}' does not satisfy the AuthorizationProvider Protocol")
 
     from deerflow.authz.rbac import RbacAuthorizationProvider
 
@@ -53,6 +78,13 @@ def resolve_authorization_provider(
         try:
             instance.validate_role(config.default_role, field="authorization.default_role")
         except ValueError as err:
-            raise ValueError(f"Invalid authorization default_role for provider '{class_path}': {err}") from err
+            raise ValueError(f"Invalid authorization default_role for provider '{spec.class_path}': {err}") from err
 
     return instance
+
+
+def resolve_authorization_provider(
+    config: AuthorizationConfig,
+) -> AuthorizationProvider | None:
+    """Discover, construct, and validate the configured provider synchronously."""
+    return construct_authorization_provider(resolve_authorization_provider_spec(config), config)
diff --git a/backend/packages/harness/deerflow/authz/sandbox_authz.py b/backend/packages/harness/deerflow/authz/sandbox_authz.py
index 1e4063caa5a..8a951149f7a 100644
--- a/backend/packages/harness/deerflow/authz/sandbox_authz.py
+++ b/backend/packages/harness/deerflow/authz/sandbox_authz.py
@@ -1,7 +1,7 @@
 """Sandbox execution authorization gate.
 
-Checks ``authorize("sandbox", "execute")`` before sandbox acquisition so a
-role-scoped policy can deny sandbox execution entirely. On deny, a
+Checks ``authorize("sandbox", "execute")`` before sandbox use so a role-scoped
+policy can deny sandbox execution entirely. On deny, a
 :class:`~deerflow.sandbox.exceptions.SandboxAuthorizationError` propagates up
 through the tool's execution; the agent's tool-error handling converts it to a
 friendly ``ToolMessage`` ("sandbox not permitted for your role") rather than
@@ -14,13 +14,14 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 from collections.abc import Mapping
 from typing import Any
 
 from deerflow.authz.principal import build_principal_from_context
-from deerflow.authz.provider import AuthzDecision, AuthzRequest
-from deerflow.authz.runtime import resolve_authorization_provider
+from deerflow.authz.provider import AuthorizationProvider, AuthzDecision, AuthzRequest, Principal
+from deerflow.authz.runtime import construct_authorization_provider, resolve_authorization_provider, resolve_authorization_provider_spec
 from deerflow.config.app_config import AppConfig
 from deerflow.sandbox.exceptions import SandboxAuthorizationError
 
@@ -45,6 +46,11 @@ def safe_app_config() -> AppConfig | None:
         return None
 
 
+async def safe_app_config_async() -> AppConfig | None:
+    """Load AppConfig without running config file I/O on the event loop."""
+    return await asyncio.to_thread(safe_app_config)
+
+
 # Sandbox is a single shared resource (the execution environment), not a named
 # catalog like tools/models/skills. The target is therefore a sentinel "*" that
 # means "the sandbox as a whole"; RBAC ``allow: ["*"]`` / ``allow: true`` permits
@@ -52,21 +58,19 @@ def safe_app_config() -> AppConfig | None:
 _SANDBOX_TARGET = "*"
 
 
-def authorize_sandbox_execution(*, context: Mapping[str, Any], app_config: AppConfig | None) -> None:
-    """Check ``authorize("sandbox", "execute")`` before sandbox acquisition.
-
-    ``app_config=None`` (unreadable config) is treated the same as
-    ``authorization.enabled: false`` — a no-op. On deny (or provider error
-    with ``fail_closed``), raises :class:`SandboxAuthorizationError`; on
-    provider error with fail-open, returns silently (legacy allow behavior).
-    """
+def _resolve_authorization_inputs(
+    *,
+    context: Mapping[str, Any],
+    app_config: AppConfig | None,
+) -> tuple[AuthorizationProvider, Any, Principal] | None:
+    """Resolve the enabled provider and principal shared by both call paths."""
     # Guard against Mock/SimpleNamespace app_config objects in tests that
     # don't carry a real AuthorizationConfig. getattr avoids AttributeError
     # and the ``is not True`` identity check avoids truthy Mock attributes
     # (mirrors filter_available_skills_by_authorization in skill_filter.py).
     authz_config = getattr(app_config, "authorization", None)
     if authz_config is None or getattr(authz_config, "enabled", None) is not True:
-        return
+        return None
 
     # Provider *resolution* failures follow the same fail_closed/fail_open
     # decision as authorize() errors — a raw ValueError here would otherwise
@@ -77,25 +81,107 @@ def authorize_sandbox_execution(*, context: Mapping[str, Any], app_config: AppCo
         logger.warning("Failed to resolve authorization provider for sandbox:execute", exc_info=True)
         if authz_config.fail_closed:
             raise SandboxAuthorizationError() from None
-        # fail-open: allow sandbox acquisition despite the resolution error.
-        return
+        # fail-open: allow sandbox use despite the resolution error.
+        return None
     if provider is None:
-        return
+        return None
 
     principal = build_principal_from_context(context, default_role=authz_config.default_role)
+    return provider, authz_config, principal
+
+
+async def _resolve_authorization_inputs_async(
+    *,
+    context: Mapping[str, Any],
+    app_config: AppConfig | None,
+) -> tuple[AuthorizationProvider, Any, Principal] | None:
+    """Discover off-loop, then construct custom providers on this event loop."""
+    authz_config = getattr(app_config, "authorization", None)
+    if authz_config is None or getattr(authz_config, "enabled", None) is not True:
+        return None
+
     try:
-        decision = provider.authorize(AuthzRequest(principal=principal, resource="sandbox", action="execute", target=_SANDBOX_TARGET))
-        if not isinstance(decision, AuthzDecision):
-            raise TypeError("AuthorizationProvider.authorize must return AuthzDecision")
-        if decision.allow:
-            return
-        # Explicit deny → block sandbox acquisition with a friendly error.
+        if getattr(authz_config, "provider", None) is None:
+            # There is no module to import. Preserve the synchronous resolver's
+            # missing-provider validation without an unnecessary worker hop.
+            provider = resolve_authorization_provider(authz_config)
+        else:
+            spec = await asyncio.to_thread(resolve_authorization_provider_spec, authz_config)
+            # Async providers may create loop-affine clients in __init__.
+            provider = construct_authorization_provider(spec, authz_config)
+    except Exception:
+        logger.warning("Failed to resolve authorization provider for sandbox:execute", exc_info=True)
+        if authz_config.fail_closed:
+            raise SandboxAuthorizationError() from None
+        return None
+    if provider is None:
+        return None
+
+    principal = build_principal_from_context(context, default_role=authz_config.default_role)
+    return provider, authz_config, principal
+
+
+def _authorization_request(principal: Principal) -> AuthzRequest:
+    return AuthzRequest(principal=principal, resource="sandbox", action="execute", target=_SANDBOX_TARGET)
+
+
+def _enforce_decision(decision: AuthzDecision, *, principal: Principal, method_name: str) -> None:
+    if not isinstance(decision, AuthzDecision):
+        raise TypeError(f"AuthorizationProvider.{method_name} must return AuthzDecision")
+    if not decision.allow:
         raise SandboxAuthorizationError(role=principal.role)
+
+
+def authorize_sandbox_execution(*, context: Mapping[str, Any], app_config: AppConfig | None) -> None:
+    """Synchronously check ``authorize("sandbox", "execute")`` before use.
+
+    ``app_config=None`` (unreadable config) is treated the same as
+    ``authorization.enabled: false`` — a no-op. On deny (or provider error
+    with ``fail_closed``), raises :class:`SandboxAuthorizationError`; on
+    provider error with fail-open, returns silently (legacy allow behavior).
+    """
+    inputs = _resolve_authorization_inputs(context=context, app_config=app_config)
+    if inputs is None:
+        return
+    provider, authz_config, principal = inputs
+
+    try:
+        decision = provider.authorize(_authorization_request(principal))
+        _enforce_decision(decision, principal=principal, method_name="authorize")
+    except SandboxAuthorizationError:
+        raise
+    except Exception:
+        logger.warning("Authorization provider failed while checking sandbox:execute", exc_info=True)
+        if authz_config.fail_closed:
+            raise SandboxAuthorizationError(role=principal.role)
+        # fail-open: allow sandbox use despite the provider error.
+        return
+
+
+async def authorize_sandbox_execution_async(*, context: Mapping[str, Any], app_config: AppConfig | None) -> None:
+    """Asynchronously check ``authorize("sandbox", "execute")`` before use.
+
+    Provider discovery may import a custom module, so it runs off the event
+    loop. Provider construction stays on the event loop because a valid async
+    provider may create loop-affine clients in ``__init__``.
+    """
+    context_snapshot = dict(context)
+    inputs = await _resolve_authorization_inputs_async(
+        context=context_snapshot,
+        app_config=app_config,
+    )
+    if inputs is None:
+        return
+    provider, authz_config, principal = inputs
+
+    try:
+        decision = await provider.aauthorize(_authorization_request(principal))
+        _enforce_decision(decision, principal=principal, method_name="aauthorize")
     except SandboxAuthorizationError:
         raise
     except Exception:
         logger.warning("Authorization provider failed while checking sandbox:execute", exc_info=True)
         if authz_config.fail_closed:
             raise SandboxAuthorizationError(role=principal.role)
-        # fail-open: allow sandbox acquisition despite the provider error.
+        # fail-open: allow sandbox use despite the provider error.
         return
diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md
index 8909fc0e7a9..b54061bc187 100644
--- a/backend/packages/harness/deerflow/sandbox/AGENTS.md
+++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md
@@ -3,7 +3,7 @@
 **Interface**: Abstract `Sandbox` with `execute_command(command, env=None)`, `read_file`, `write_file`, `list_dir`, `glob`, and `grep`. `grep` accepts either one text file or a directory tree. The optional `env` injects per-call environment variables (request-scoped secrets — see Request-Scoped Secrets below); `LocalSandbox` merges it via `subprocess.run(env=...)` and `AioSandbox` routes env-bearing commands through the `bash.exec(env=...)` API on a fresh session.
 **Provider Pattern**: `SandboxProvider` with `acquire`, `acquire_async`, `get`, `release` lifecycle. Async agent/tool paths call async sandbox lifecycle hooks so Docker sandbox creation, discovery, cross-process locking, readiness polling, and release stay off the event loop.
 **Shared components** (RFC #4741): remote providers derive their deterministic sandbox id through `derive_sandbox_scope_token` (`sandbox/identity.py`, keyword-only; the sha256/16-hex derivation is a compatibility contract — changing it orphans existing containers), and serialize provider-selected acquire/release transitions through `AcquireSerializer` (`sandbox/acquire_serialization.py`): per-key `threading.Lock` table with holder/waiter refcount reclamation (no unbounded per-thread lock growth), a bounded dedicated executor so async waits never touch the event loop or the default executor, worker-owned cancellation cleanup that does not depend on a cancelled event loop task resuming, and idempotent `close()` called from provider `shutdown()`/`reset()`. AIO/E2B key by `(user_id, thread_id)`; BoxLite/Tenki/OpenSandbox key by the derived sandbox id. `thread_id=None` acquires (random uuid ids) never enter the serializer.
-**Authorization gate** (`sandbox:execute`, RFC #4063 Phase 3): every sandbox acquisition passes through `authorize_sandbox_execution` (`deerflow/authz/sandbox_authz.py`) - a binary `authorize(principal, "sandbox", "execute", target="*")` check before `provider.acquire`. The gate lives at the single acquisition entry point (`ensure_sandbox_initialized` / `_acquire_sandbox_async` in `tools.py`, and `SandboxMiddleware.before_agent` / `abefore_agent`), so it cannot be bypassed regardless of which sandbox-dependent tool triggers it; the reuse path (sandbox already in state) skips the re-check. Deny raises `SandboxAuthorizationError` (`sandbox/exceptions.py`), which propagates out of tool execution as a friendly error `ToolMessage` ("sandbox execution is not permitted for your role") - the eager path catches it and skips acquisition instead, deferring the deny to the first sandbox-touching tool call so both paths share the same semantics. Provider errors (both `authorize()` and provider resolution) follow `authorization.fail_closed` / `fail_open`; no readable `config.yaml` or `authorization.enabled: false` makes the gate a no-op (`safe_app_config` tolerates missing config). Gateway auxiliary sync paths (uploads/artifacts routers) call `try_acquire_sandbox_for_request` (`app/gateway/authz.py`), which gates via `authorize_sandbox_for_request` and skips the sync on deny - the upload/artifact edit itself still succeeds. Tests: `tests/test_sandbox_authorization.py`.
+**Authorization gate** (`sandbox:execute`, RFC #4063 Phase 3): every sandbox-backed tool call passes through the gate in `deerflow/authz/sandbox_authz.py` - a binary `authorize(principal, "sandbox", "execute", target="*")` check before either reusing a persisted sandbox id or calling `provider.acquire`. Rechecking reuse is required because authorization config and user roles can change while the sandbox remains cached. Sync tool invocations call `authorize_sandbox_execution`; async tool invocations await `authorize_sandbox_execution_async` exactly once. A task-local `ContextVar` scopes that single decision across the complete composed tool invocation, including `ReadBeforeWriteMiddleware`'s pre-write inspection, tool body, and post-read mark; the value is copied into `asyncio.to_thread` workers. Authorization denial is converted to the normal error `ToolMessage` at the composed middleware boundary and is explicitly excluded from the gate's generic fail-open handlers. Async config loading and provider class discovery/import are offloaded before `aauthorize()` so reused sandbox calls do not hash config files or import custom modules on the event loop; provider construction remains on the running event loop because async providers may initialize loop-affine clients. The gate lives at the single tool initialization entry point (`ensure_sandbox_initialized` / `ensure_sandbox_initialized_async` in `tools.py`), while `SandboxMiddleware.before_agent` / `abefore_agent` apply the matching sync/async check to eager acquisition. Deny raises `SandboxAuthorizationError` (`sandbox/exceptions.py`), which propagates out of ordinary tool execution as a friendly error `ToolMessage` ("sandbox execution is not permitted for your role") - the eager path catches it and skips acquisition instead, deferring the deny to the first sandbox-touching tool call so both paths share the same semantics. Provider errors (authorization calls and provider resolution) follow `authorization.fail_closed` / `fail_open`; no readable `config.yaml` or `authorization.enabled: false` makes the gate a no-op (`safe_app_config` tolerates missing config). Gateway auxiliary sync paths (uploads/artifacts routers) call `try_acquire_sandbox_for_request` (`app/gateway/authz.py`), which gates via `authorize_sandbox_for_request` and skips the sync on deny - the upload/artifact edit itself still succeeds. Tests: `tests/test_sandbox_authorization.py` and `tests/blocking_io/test_sandbox_authorization.py`.
 **Environment policy** (`sandbox/env_policy.py`): `execute_command` no longer inherits the full `os.environ`. `build_sandbox_env()` scrubs secret-looking names (`*KEY*`/`*SECRET*`/`*TOKEN*`/`*PASS*`/`*CREDENTIAL*`) from the inherited environment before layering injected request secrets on top, so platform credentials (e.g. `OPENAI_API_KEY`) never leak into skill subprocesses. Benign vars (`PATH`, `HOME`, `LANG`, `VIRTUAL_ENV`, ...) are preserved.
 **Implementations**:
 - `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-thread `LocalSandbox` (id `local:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/acp-workspace` to that thread's host directories, so the public `Sandbox` API honours the `/mnt/user-data` contract uniformly with AIO. `acquire()` / `acquire(None)` keeps the legacy generic singleton (id `local`) for callers without a thread context. Per-thread sandboxes are held in an LRU cache (default 256 entries) guarded by a `threading.Lock`. Public, custom, legacy, and managed integration skill mappings point at stable enabled-only projection roots rather than raw skill directories. On Windows, Git Bash/MSYS argument-conversion exclusions are limited to safe non-root virtual path prefixes; do not restore a blanket conversion disable, because host-native CLI launchers need normal MSYS path conversion for their own installation paths.
diff --git a/backend/packages/harness/deerflow/sandbox/middleware.py b/backend/packages/harness/deerflow/sandbox/middleware.py
index c0d98f560f4..ed7e8e7586a 100644
--- a/backend/packages/harness/deerflow/sandbox/middleware.py
+++ b/backend/packages/harness/deerflow/sandbox/middleware.py
@@ -12,7 +12,12 @@
 from langgraph.types import Command
 
 from deerflow.agents.thread_state import SandboxStateField, ThreadDataState
-from deerflow.authz.sandbox_authz import authorize_sandbox_execution, safe_app_config
+from deerflow.authz.sandbox_authz import (
+    authorize_sandbox_execution,
+    authorize_sandbox_execution_async,
+    safe_app_config,
+    safe_app_config_async,
+)
 from deerflow.runtime.user_context import resolve_runtime_user_id
 from deerflow.sandbox import get_sandbox_provider
 from deerflow.sandbox.exceptions import SandboxAuthorizationError
@@ -116,9 +121,9 @@ async def abefore_agent(self, state: SandboxMiddlewareState, runtime: Runtime) -
             # ``ensure_sandbox_initialized`` denies per-tool with the RFC §9
             # friendly message on the first sandbox-touching tool call.
             try:
-                authorize_sandbox_execution(
+                await authorize_sandbox_execution_async(
                     context=runtime.context or {},
-                    app_config=safe_app_config(),
+                    app_config=await safe_app_config_async(),
                 )
             except SandboxAuthorizationError:
                 logger.info("Sandbox execution denied for this role; skipping eager sandbox acquisition (thread_id=%s)", thread_id)
diff --git a/backend/packages/harness/deerflow/sandbox/tools.py b/backend/packages/harness/deerflow/sandbox/tools.py
index f927b36bcd0..6f8bc2c04bc 100644
--- a/backend/packages/harness/deerflow/sandbox/tools.py
+++ b/backend/packages/harness/deerflow/sandbox/tools.py
@@ -5,14 +5,21 @@
 import posixpath
 import re
 import shlex
-from collections.abc import Callable
+from collections.abc import AsyncIterator, Callable, Iterator
+from contextlib import asynccontextmanager, contextmanager
+from contextvars import ContextVar
 from functools import lru_cache
 from pathlib import Path
 
 from langchain.tools import tool
 
 from deerflow.agents.thread_state import ThreadDataState
-from deerflow.authz.sandbox_authz import authorize_sandbox_execution, safe_app_config
+from deerflow.authz.sandbox_authz import (
+    authorize_sandbox_execution,
+    authorize_sandbox_execution_async,
+    safe_app_config,
+    safe_app_config_async,
+)
 from deerflow.config import get_app_config
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
 from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
@@ -34,6 +41,15 @@
 
 logger = logging.getLogger(__name__)
 
+# The read-before-write middleware can enter the sandbox before or after the
+# tool body. Scope this marker to the complete composed invocation so all of
+# those entries share one live provider decision. Context variables are copied
+# into ``asyncio.to_thread`` workers, keeping the handoff task-local.
+_SANDBOX_AUTHORIZATION_CHECKED: ContextVar[bool] = ContextVar(
+    "deerflow_sandbox_authorization_checked",
+    default=False,
+)
+
 _ABSOLUTE_PATH_PATTERN = re.compile(r"(?<![:\w])(?<!:/)/(?:[^\s\"'`;&|<>()]+)")
 # A ``{...}`` block holding a single identifier-like placeholder (e.g. ``{id}``
 # in a REST template or ``{port}`` in an f-string). Bash brace expansion such as
@@ -1371,6 +1387,42 @@ def sandbox_from_runtime(runtime: Runtime | None = None) -> Sandbox:
     return sandbox
 
 
+@contextmanager
+def sandbox_authorization_scope(runtime: Runtime) -> Iterator[None]:
+    """Authorize once for one complete synchronous sandbox tool invocation."""
+    if _SANDBOX_AUTHORIZATION_CHECKED.get():
+        yield
+        return
+
+    authorize_sandbox_execution(
+        context=runtime.context or {},
+        app_config=safe_app_config(),
+    )
+    token = _SANDBOX_AUTHORIZATION_CHECKED.set(True)
+    try:
+        yield
+    finally:
+        _SANDBOX_AUTHORIZATION_CHECKED.reset(token)
+
+
+@asynccontextmanager
+async def sandbox_authorization_scope_async(runtime: Runtime) -> AsyncIterator[None]:
+    """Authorize once for one complete asynchronous sandbox tool invocation."""
+    if _SANDBOX_AUTHORIZATION_CHECKED.get():
+        yield
+        return
+
+    await authorize_sandbox_execution_async(
+        context=runtime.context or {},
+        app_config=await safe_app_config_async(),
+    )
+    token = _SANDBOX_AUTHORIZATION_CHECKED.set(True)
+    try:
+        yield
+    finally:
+        _SANDBOX_AUTHORIZATION_CHECKED.reset(token)
+
+
 def ensure_sandbox_initialized(runtime: Runtime | None = None) -> Sandbox:
     """Ensure sandbox is initialized, acquiring lazily if needed.
 
@@ -1395,6 +1447,15 @@ def ensure_sandbox_initialized(runtime: Runtime | None = None) -> Sandbox:
     if runtime.state is None:
         raise SandboxRuntimeError("Tool runtime state not available")
 
+    # Authorization is a live execution policy, not a lifetime property of a
+    # sandbox id. Re-check before both reuse and acquisition so a role or policy
+    # change takes effect on the next sandbox-backed tool call.
+    if not _SANDBOX_AUTHORIZATION_CHECKED.get():
+        authorize_sandbox_execution(
+            context=runtime.context or {},
+            app_config=safe_app_config(),
+        )
+
     # Check if sandbox already exists in state
     # Discarding fork_restored is safe: after_agent short-circuits on the
     # still-wrapped state before the context-based release branch, so this
@@ -1417,15 +1478,6 @@ def ensure_sandbox_initialized(runtime: Runtime | None = None) -> Sandbox:
     if thread_id is None:
         raise SandboxRuntimeError("Thread ID not available in runtime context")
 
-    # Phase 3: enforce sandbox:execute authorization before acquiring. On deny
-    # a SandboxAuthorizationError propagates up through the tool so the agent's
-    # tool-error handling returns a friendly message (RFC §9). Skipped on the
-    # reuse path above (already authorized when first acquired).
-    authorize_sandbox_execution(
-        context=runtime.context or {},
-        app_config=safe_app_config(),
-    )
-
     provider = get_sandbox_provider()
     sandbox_id = provider.acquire(thread_id, user_id=resolve_runtime_user_id(runtime))
 
@@ -1455,6 +1507,14 @@ async def ensure_sandbox_initialized_async(runtime: Runtime | None = None) -> Sa
     if runtime.state is None:
         raise SandboxRuntimeError("Tool runtime state not available")
 
+    # Keep the async path aligned with the sync path: persisted sandbox state
+    # must not bypass a newly-revoked sandbox:execute grant.
+    if not _SANDBOX_AUTHORIZATION_CHECKED.get():
+        await authorize_sandbox_execution_async(
+            context=runtime.context or {},
+            app_config=await safe_app_config_async(),
+        )
+
     # Same discard as the sync path above: the reuse path never releases,
     # because after_agent short-circuits on the still-wrapped state first.
     sandbox_state, _ = unwrap_sandbox(runtime.state.get("sandbox"))
@@ -1473,13 +1533,6 @@ async def ensure_sandbox_initialized_async(runtime: Runtime | None = None) -> Sa
     if thread_id is None:
         raise SandboxRuntimeError("Thread ID not available in runtime context")
 
-    # Phase 3: enforce sandbox:execute authorization before acquiring (async
-    # counterpart of the sync gate in ``ensure_sandbox_initialized``).
-    authorize_sandbox_execution(
-        context=runtime.context or {},
-        app_config=safe_app_config(),
-    )
-
     provider = get_sandbox_provider()
     sandbox_id = await provider.acquire_async(thread_id, user_id=resolve_runtime_user_id(runtime))
 
@@ -1501,17 +1554,18 @@ async def _run_sync_tool_after_async_sandbox_init(
 ) -> str:
     """Initialize lazily via async provider, then run sync tool body off-thread."""
     try:
-        await ensure_sandbox_initialized_async(runtime)
+        async with sandbox_authorization_scope_async(runtime):
+            await ensure_sandbox_initialized_async(runtime)
+
+            if func is None:
+                return "Error: Tool implementation not available"
+
+            return await asyncio.to_thread(func, runtime, *args)
     except SandboxError as e:
         return f"Error: {e}"
     except Exception as e:
         return f"Error: Unexpected error initializing sandbox: {_sanitize_error(e, runtime)}"
 
-    if func is None:
-        return "Error: Tool implementation not available"
-
-    return await asyncio.to_thread(func, runtime, *args)
-
 
 def ensure_thread_directories_exist(runtime: Runtime | None) -> None:
     """Ensure thread data directories (workspace, uploads, outputs) exist.
diff --git a/backend/tests/blocking_io/test_sandbox_authorization.py b/backend/tests/blocking_io/test_sandbox_authorization.py
new file mode 100644
index 00000000000..234276e33f2
--- /dev/null
+++ b/backend/tests/blocking_io/test_sandbox_authorization.py
@@ -0,0 +1,64 @@
+"""Sandbox authorization resolution must stay off the async event loop."""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from deerflow.authz import sandbox_authz
+from deerflow.config.app_config import AppConfig
+from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig
+from deerflow.config.model_config import ModelConfig
+from deerflow.config.sandbox_config import SandboxConfig
+from deerflow.sandbox import tools as sandbox_tools
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_reused_async_sandbox_offloads_config_and_provider_resolution(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Config hashing and class discovery stay off-loop; construction does not."""
+    probe = tmp_path / "sandbox-authz-probe"
+    await asyncio.to_thread(probe.write_text, "probe", encoding="utf-8")
+
+    app_config = AppConfig(
+        models=[ModelConfig(name="gpt-4", model="gpt-4", use="langchain_openai:ChatOpenAI")],
+        sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"),
+        authorization=AuthorizationConfig(
+            enabled=True,
+            fail_closed=True,
+            default_role="user",
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"user": {"sandbox": {"allow": "*"}}}},
+            ),
+        ),
+    )
+
+    def blocking_config_load():
+        probe.read_text(encoding="utf-8")
+        return app_config
+
+    discover_provider = sandbox_authz.resolve_authorization_provider_spec
+
+    def blocking_provider_discovery(config):
+        probe.read_text(encoding="utf-8")
+        return discover_provider(config)
+
+    monkeypatch.setattr(sandbox_authz, "safe_app_config", blocking_config_load)
+    monkeypatch.setattr(sandbox_authz, "resolve_authorization_provider_spec", blocking_provider_discovery)
+
+    sandbox = MagicMock()
+    sandbox_provider = MagicMock()
+    sandbox_provider.get.return_value = sandbox
+    monkeypatch.setattr(sandbox_tools, "get_sandbox_provider", lambda: sandbox_provider)
+    runtime = SimpleNamespace(
+        state={"sandbox": {"sandbox_id": "sbx-existing"}},
+        context={"thread_id": "t1", "user_id": "u1", "user_role": "user"},
+        config=None,
+    )
+
+    assert await sandbox_tools.ensure_sandbox_initialized_async(runtime) is sandbox
diff --git a/backend/tests/test_sandbox_authorization.py b/backend/tests/test_sandbox_authorization.py
index 1a43e290c03..dad3cf9a1f4 100644
--- a/backend/tests/test_sandbox_authorization.py
+++ b/backend/tests/test_sandbox_authorization.py
@@ -1,7 +1,7 @@
 """Phase 3 sandbox-level authorization tests.
 
-Covers the ``authorize("sandbox", "execute")`` gate at the sandbox-acquisition
-entry point. When denied, a :class:`SandboxAuthorizationError` propagates up
+Covers the ``authorize("sandbox", "execute")`` gate at the sandbox-use entry
+point. When denied, a :class:`SandboxAuthorizationError` propagates up
 through the tool so the agent's tool-error handling returns a friendly message
 (RFC §9), rather than crashing the run.
 
@@ -13,7 +13,9 @@
 
 from __future__ import annotations
 
-from types import SimpleNamespace
+import asyncio
+import sys
+from types import ModuleType, SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -22,7 +24,7 @@
 from deerflow.authz.rbac import RbacAuthorizationProvider
 from deerflow.authz.sandbox_authz import authorize_sandbox_execution
 from deerflow.config.app_config import AppConfig
-from deerflow.config.authorization_config import AuthorizationConfig
+from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig
 from deerflow.config.model_config import ModelConfig
 from deerflow.config.sandbox_config import SandboxConfig
 from deerflow.sandbox.exceptions import SandboxAuthorizationError
@@ -238,6 +240,34 @@ def test_ensure_sandbox_initialized_denies_on_authz_reject(monkeypatch):
     sandbox_provider.acquire.assert_not_called()
 
 
+def test_ensure_sandbox_initialized_rechecks_authz_for_reused_sandbox(monkeypatch):
+    """A persisted sandbox id must not outlive a revoked execute grant."""
+    from deerflow.sandbox import tools as sandbox_tools
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    sandbox_provider = MagicMock()
+    sandbox_provider.get.return_value = MagicMock()
+    monkeypatch.setattr(sandbox_tools, "get_sandbox_provider", lambda: sandbox_provider)
+    runtime = SimpleNamespace(
+        state={"sandbox": {"sandbox_id": "sbx-existing"}},
+        context={"thread_id": "t1", "user_id": "u1", "user_role": "user"},
+        config=None,
+    )
+
+    with pytest.raises(SandboxAuthorizationError):
+        sandbox_tools.ensure_sandbox_initialized(runtime)
+    sandbox_provider.get.assert_not_called()
+    sandbox_provider.acquire.assert_not_called()
+
+
 def test_ensure_sandbox_initialized_allows_on_authz_permit(monkeypatch):
     """ensure_sandbox_initialized proceeds to acquire on allow."""
     from deerflow.sandbox import tools as sandbox_tools
@@ -662,6 +692,299 @@ def test_ensure_sandbox_initialized_async_denies_on_authz_reject(monkeypatch):
     sandbox_provider.acquire_async.assert_not_called()
 
 
+def test_ensure_sandbox_initialized_async_rechecks_authz_for_reused_sandbox(monkeypatch):
+    """Async tool calls also re-check a revoked grant before sandbox reuse."""
+    from deerflow.sandbox import tools as sandbox_tools
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    sandbox_provider = MagicMock()
+    sandbox_provider.get.return_value = MagicMock()
+    sandbox_provider.acquire_async = AsyncMock(return_value="sbx-new")
+    monkeypatch.setattr(sandbox_tools, "get_sandbox_provider", lambda: sandbox_provider)
+    runtime = SimpleNamespace(
+        state={"sandbox": {"sandbox_id": "sbx-existing"}},
+        context={"thread_id": "t1", "user_id": "u1", "user_role": "user"},
+        config=None,
+    )
+    import asyncio
+
+    with pytest.raises(SandboxAuthorizationError):
+        asyncio.run(sandbox_tools.ensure_sandbox_initialized_async(runtime))
+    sandbox_provider.get.assert_not_called()
+    sandbox_provider.acquire_async.assert_not_called()
+
+
+def test_async_provider_is_constructed_on_running_event_loop(monkeypatch):
+    """A loop-affine custom provider must not be constructed in a worker."""
+    from deerflow.authz.sandbox_authz import authorize_sandbox_execution_async
+
+    provider_module = ModuleType("loop_affine_authz_test_provider")
+    constructed_on = []
+
+    class LoopAffineProvider:
+        name = "loop-affine"
+
+        def __init__(self):
+            self.loop = asyncio.get_running_loop()
+            constructed_on.append(self.loop)
+
+        def authorize(self, request):
+            return AuthzDecision(allow=True)
+
+        async def aauthorize(self, request):
+            assert asyncio.get_running_loop() is self.loop
+            return AuthzDecision(allow=True)
+
+        def filter_resources(self, principal, resource_type, candidates):
+            return list(candidates)
+
+    provider_module.LoopAffineProvider = LoopAffineProvider
+    monkeypatch.setitem(sys.modules, provider_module.__name__, provider_module)
+
+    app_config = _make_app_config()
+    app_config.authorization = AuthorizationConfig(
+        enabled=True,
+        fail_closed=True,
+        default_role="user",
+        provider=AuthorizationProviderConfig(use=f"{provider_module.__name__}:LoopAffineProvider"),
+    )
+
+    async def _run() -> None:
+        running_loop = asyncio.get_running_loop()
+        await authorize_sandbox_execution_async(context=_context(), app_config=app_config)
+        assert constructed_on == [running_loop]
+
+    asyncio.run(_run())
+
+
+def test_async_sandbox_tool_authorizes_once_via_async_provider(monkeypatch):
+    """One async tool invocation must make one async authorization decision."""
+    from deerflow.sandbox import tools as sandbox_tools
+
+    provider = MagicMock()
+    provider.authorize.return_value = AuthzDecision(allow=True)
+    provider.aauthorize = AsyncMock(return_value=AuthzDecision(allow=True))
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    sandbox = MagicMock()
+    sandbox.list_dir.return_value = []
+    sandbox_provider = MagicMock()
+    sandbox_provider.get.return_value = sandbox
+    monkeypatch.setattr(sandbox_tools, "get_sandbox_provider", lambda: sandbox_provider)
+    monkeypatch.setattr(sandbox_tools, "is_local_sandbox", lambda runtime: False)
+    runtime = SimpleNamespace(
+        state={"sandbox": {"sandbox_id": "sbx-existing"}},
+        context={"thread_id": "t1", "user_id": "u1", "user_role": "user"},
+        config=None,
+    )
+
+    import asyncio
+
+    result = asyncio.run(
+        sandbox_tools.ls_tool.coroutine(
+            runtime=runtime,
+            description="list workspace",
+            path="/mnt/user-data/workspace",
+        )
+    )
+
+    assert result == "(empty)"
+    provider.aauthorize.assert_awaited_once()
+    provider.authorize.assert_not_called()
+
+
+def _composed_file_request(name, args, runtime, messages=()):
+    from langgraph.prebuilt.tool_node import ToolCallRequest
+
+    return ToolCallRequest(
+        tool_call={"name": name, "args": args, "id": f"call-{name}"},
+        tool=None,
+        state={"messages": list(messages)},
+        runtime=runtime,
+    )
+
+
+def _install_composed_file_authz(monkeypatch, *, allow=True):
+    from deerflow.sandbox import tools as sandbox_tools
+
+    provider = MagicMock()
+    provider.authorize.return_value = AuthzDecision(allow=allow)
+    provider.aauthorize = AsyncMock(return_value=AuthzDecision(allow=allow))
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    sandbox = MagicMock()
+    sandbox.id = "sbx-existing"
+    sandbox_provider = MagicMock()
+    sandbox_provider.get.return_value = sandbox
+    monkeypatch.setattr(sandbox_tools, "get_sandbox_provider", lambda: sandbox_provider)
+    monkeypatch.setattr(sandbox_tools, "is_local_sandbox", lambda runtime: False)
+    runtime = SimpleNamespace(
+        state={"sandbox": {"sandbox_id": "sbx-existing"}},
+        context={"thread_id": "t1", "user_id": "u1", "user_role": "user"},
+        config=None,
+    )
+    return provider, sandbox, runtime
+
+
+@pytest.mark.parametrize("tool_name", ["read_file", "write_file", "str_replace"])
+def test_composed_sync_file_tool_authorizes_once(monkeypatch, tool_name):
+    """Read-before-write re-entry shares one synchronous provider decision."""
+    from langchain_core.messages import ToolMessage
+
+    from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware
+    from deerflow.sandbox import tools as sandbox_tools
+
+    provider, sandbox, runtime = _install_composed_file_authz(monkeypatch)
+    path = "/mnt/user-data/outputs/report.md"
+    middleware = ReadBeforeWriteMiddleware()
+    messages = ()
+
+    if tool_name == "read_file":
+        sandbox.read_file.return_value = "v1"
+        args = {"description": "read report", "path": path}
+
+        def handler(_request):
+            content = sandbox_tools.read_file_tool.func(runtime, args["description"], path)
+            return ToolMessage(content=content, tool_call_id="call-read_file", name="read_file")
+
+    elif tool_name == "write_file":
+        sandbox.read_file.side_effect = FileNotFoundError(path)
+        args = {"description": "write report", "path": path, "content": "v1"}
+
+        def handler(_request):
+            content = sandbox_tools.write_file_tool.func(runtime, args["description"], path, args["content"])
+            return ToolMessage(content=content, tool_call_id="call-write_file", name="write_file")
+
+    else:
+        import hashlib
+
+        sandbox.read_file.return_value = "v1"
+        args = {"description": "edit report", "path": path, "old_str": "v1", "new_str": "v2"}
+        read_mark = ToolMessage(content="v1", tool_call_id="prior-read", name="read_file")
+        read_mark.additional_kwargs["deerflow_read_mark"] = {
+            "path": path,
+            "hash": hashlib.sha256(b"v1").hexdigest(),
+        }
+        messages = (read_mark,)
+
+        def handler(_request):
+            content = sandbox_tools.str_replace_tool.func(runtime, args["description"], path, args["old_str"], args["new_str"])
+            return ToolMessage(content=content, tool_call_id="call-str_replace", name="str_replace")
+
+    result = middleware.wrap_tool_call(_composed_file_request(tool_name, args, runtime, messages), handler)
+
+    assert result.status != "error"
+    provider.authorize.assert_called_once()
+    provider.aauthorize.assert_not_called()
+
+
+@pytest.mark.parametrize("tool_name", ["read_file", "write_file", "str_replace"])
+def test_composed_async_file_tool_authorizes_once(monkeypatch, tool_name):
+    """Async gate, tool body, and read-mark worker share one async decision."""
+    import asyncio
+
+    from langchain_core.messages import ToolMessage
+
+    from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware
+    from deerflow.sandbox import tools as sandbox_tools
+
+    provider, sandbox, runtime = _install_composed_file_authz(monkeypatch)
+    path = "/mnt/user-data/outputs/report.md"
+    middleware = ReadBeforeWriteMiddleware()
+    messages = ()
+
+    if tool_name == "read_file":
+        sandbox.read_file.return_value = "v1"
+        args = {"description": "read report", "path": path}
+
+        async def handler(_request):
+            content = await sandbox_tools.read_file_tool.coroutine(runtime, args["description"], path)
+            return ToolMessage(content=content, tool_call_id="call-read_file", name="read_file")
+
+    elif tool_name == "write_file":
+        sandbox.read_file.side_effect = FileNotFoundError(path)
+        args = {"description": "write report", "path": path, "content": "v1"}
+
+        async def handler(_request):
+            content = await sandbox_tools.write_file_tool.coroutine(runtime, args["description"], path, args["content"])
+            return ToolMessage(content=content, tool_call_id="call-write_file", name="write_file")
+
+    else:
+        import hashlib
+
+        sandbox.read_file.return_value = "v1"
+        args = {"description": "edit report", "path": path, "old_str": "v1", "new_str": "v2"}
+        read_mark = ToolMessage(content="v1", tool_call_id="prior-read", name="read_file")
+        read_mark.additional_kwargs["deerflow_read_mark"] = {
+            "path": path,
+            "hash": hashlib.sha256(b"v1").hexdigest(),
+        }
+        messages = (read_mark,)
+
+        async def handler(_request):
+            content = await sandbox_tools.str_replace_tool.coroutine(runtime, args["description"], path, args["old_str"], args["new_str"])
+            return ToolMessage(content=content, tool_call_id="call-str_replace", name="str_replace")
+
+    result = asyncio.run(middleware.awrap_tool_call(_composed_file_request(tool_name, args, runtime, messages), handler))
+
+    assert result.status != "error"
+    provider.aauthorize.assert_awaited_once()
+    provider.authorize.assert_not_called()
+
+
+@pytest.mark.parametrize("is_async", [False, True])
+def test_composed_write_authorization_deny_is_not_swallowed(monkeypatch, is_async):
+    """The gate's generic fail-open path must never turn an authz deny into allow."""
+    import asyncio
+
+    from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware
+
+    provider, sandbox, runtime = _install_composed_file_authz(monkeypatch, allow=False)
+    path = "/mnt/user-data/outputs/report.md"
+    args = {"description": "write report", "path": path, "content": "v1"}
+    request = _composed_file_request("write_file", args, runtime)
+    middleware = ReadBeforeWriteMiddleware()
+    handler = MagicMock(side_effect=AssertionError("denied handler must not run"))
+
+    if is_async:
+
+        async def async_handler(_request):
+            handler(_request)
+
+        result = asyncio.run(middleware.awrap_tool_call(request, async_handler))
+        provider.aauthorize.assert_awaited_once()
+        provider.authorize.assert_not_called()
+    else:
+        result = middleware.wrap_tool_call(request, handler)
+        provider.authorize.assert_called_once()
+        provider.aauthorize.assert_not_called()
+
+    assert result.status == "error"
+    assert "not permitted" in str(result.content).lower()
+    handler.assert_not_called()
+    sandbox.read_file.assert_not_called()
+
+
 def test_abefore_agent_deny_skips_acquisition(monkeypatch):
     """Async eager path deny: acquisition skipped, no run-level error.
 
diff --git a/docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md b/docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md
index 36812a2e06a..d85d9d7f310 100644
--- a/docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md
+++ b/docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md
@@ -403,6 +403,41 @@ Phase 1 最低验证要求：
   effective-permissions 展示；management route 的 provider 迁移；
   feishu/dingtalk 文件同步路径的 sandbox gate（身份传递机制待定）。
 
+### 2026-08-27 — Phase 3 / PR #5006 组合调用单次决策与异步阻塞收口
+
+- **背景：** review 在默认启用的 `ReadBeforeWriteMiddleware` 组合路径复现了一次工具
+  调用产生两次 provider 决策：读工具在 tool body 后重新读取以写 mark，写工具在
+  tool body 前读取以检查 gate。异步路径还在 event loop 上同步加载配置并解析 provider。
+- **决策（调用作用域）：** `sandbox_authorization_scope` / async counterpart 用 task-local
+  `ContextVar` 覆盖完整的组合工具调用，而不只覆盖 offload 的同步 tool body。读写 gate、
+  tool body 和 mark stamping 共用一次实时授权决策；下一个独立工具调用仍重新授权。
+- **决策（deny 语义）：** `ReadBeforeWriteMiddleware` 在作用域入口把
+  `SandboxAuthorizationError` 转成标准 error `ToolMessage`，并在 `_check_write_gate` 与
+  `_attach_read_mark` 中显式重新抛出该异常，禁止通用 fail-open 分支吞掉授权拒绝。
+- **决策（event-loop 边界）：** async config 加载通过 `safe_app_config_async()` offload；
+  `_resolve_authorization_inputs()` 也在线程中执行，避免每次复用 sandbox 时在 event loop
+  上 stat/hash 配置文件或 import/构造自定义 provider。只有 provider 的 `aauthorize()`
+  在异步调用路径上直接 await。
+- **证据：** `tests/test_sandbox_authorization.py` 新增 sync/async `read_file` 与
+  `write_file` 组合覆盖，断言每次调用恰好一个 provider 决策并验证 deny 不被 fail-open；
+  `tests/blocking_io/test_sandbox_authorization.py` 用真实阻塞文件探针固定配置与 provider
+  解析均不在 event loop 上执行。
+- **兼容性：** `authorization.enabled: false` 仍为 no-op；未启用
+  `ReadBeforeWriteMiddleware` 的普通 sandbox 工具继续在各自调用入口重新授权；同步与异步
+  deny 均保持工具级错误而非 run 级异常。
+
+#### PR #5006 review 补充：异步 provider 的构造线程
+
+- 自定义 provider 的模块发现可能触发阻塞 import，但 provider 构造函数也可能创建
+  asyncio loop-affine 客户端。`runtime.py` 因此把解析拆成两阶段：
+  `resolve_authorization_provider_spec()` 在线程池完成 class-path 发现，
+  `construct_authorization_provider()` 在调用方事件循环构造并校验实例。
+- 同步 `resolve_authorization_provider()` 继续组合这两个阶段，保持原有调用契约与错误语义。
+  async sandbox gate 的发现和构造任一失败仍统一遵循 `fail_closed` / `fail_open`。
+- 回归覆盖同时固定两个边界：阻塞文件探针证明 config hash 与 class discovery 不占用
+  event loop；loop-affine provider 在 `__init__` 调用 `asyncio.get_running_loop()` 并在
+  `aauthorize()` 验证仍是同一个 loop。
+
 ### 新记录模板
 
 ```markdown