From aca9be716b2a466ed8189d128f9775ad2f33aafa Mon Sep 17 00:00:00 2001
From: OliverBryant <2713999266@qq.com>
Date: Tue, 8 Sep 2026 16:33:29 +0800
Subject: [PATCH 1/5] feat(mcp): hold a connector write for approval above both
 transports

An MCP write that needs a human in front of it currently pauses by
returning waiting_for_user, and the model then writes the arguments a
second time when it resumes -- so what executes is not what was shown.
This adds the seam that fixes that: a host installs a hook, the hook
answers "run it" or "pause", and an approved call is replayed with the
arguments the approver saw, verbatim, because they were never handed
back to the model.

The gate is consulted in a wrapper on the host, not inside the adapter.
A supported npx/uvx connector does not run in this process at all: the
sandbox serializes the adapter and tool_runner.py rebuilds it with
importlib and cloudpickle in a guest, where no host hook and no database
exist. A check inside the adapter is therefore absent for exactly the
transport that most needs it. WriteGateTool wraps whatever the loader
produced -- a bare MCPToolAdapter or a SandboxedToolWrapper around one --
applied at the single line where both loaders converge.

Both of the wrapper's entry points are covered. run_json_async consults
the gate; run_json_sync refuses, because the sandbox wrapper's sync path
works (it drives its own async path through asyncio.run) and forwarding
it would run a sandboxed write with nothing consulted.

A replay re-enters the target's own run_json_async under a ContextVar
that suppresses re-gating, so the authorization check, the delegated-
credential refresh retry and the safe connector-error mapping all still
run for an approved call, rather than reaching past them into the
low-level call.

A gated tool also reports concurrency_safe=False, but only while a hook
is installed: ReAct copies one answer to every interaction in a
concurrent segment, so one approve would authorize every independently
frozen write in the batch. With no hook installed the wrapper is a total
pass-through, metadata included.

Two limits are stated where the seam is defined rather than left for a
reader to discover: an approval is not bound to a connector identity
(nothing in the tree exposes an immutable server/account identity to
bind to), and only a surface that delivers the chosen option value
verbatim can grant one -- xagent's own ClarificationForm submits display
labels, so a pause surfaced there is voided rather than executed.
---
 .../core/tools/adapters/vibe/mcp_adapter.py   |   9 +-
 .../core/tools/adapters/vibe/write_gate.py    | 173 ++++++++++++
 .../tools/adapters/vibe/write_gate_tool.py    | 258 ++++++++++++++++++
 3 files changed, 439 insertions(+), 1 deletion(-)
 create mode 100644 src/xagent/core/tools/adapters/vibe/write_gate.py
 create mode 100644 src/xagent/core/tools/adapters/vibe/write_gate_tool.py

diff --git a/src/xagent/core/tools/adapters/vibe/mcp_adapter.py b/src/xagent/core/tools/adapters/vibe/mcp_adapter.py
index f3aa3a934..87aa2c91d 100644
--- a/src/xagent/core/tools/adapters/vibe/mcp_adapter.py
+++ b/src/xagent/core/tools/adapters/vibe/mcp_adapter.py
@@ -54,6 +54,7 @@
     should_sandbox_mcp_connection,
 )
 from .tool_naming_limits import MAX_AGENT_TOOL_NAME_LENGTH
+from .write_gate_tool import gate_mcp_tools
 
 
 class MCPFailurePhase(str, Enum):
@@ -2194,7 +2195,13 @@ def tool_builder(
                 server_tools = direct_result.tools
                 failures.extend(direct_result.failures)
 
-            agent_tools.extend(server_tools)
+            # The one place both transports meet: ``server_tools`` is either
+            # sandbox-wrapped tools or bare adapters by this point, and the
+            # gate wraps whichever it is. Deliberately not inside the adapter
+            # -- a sandboxed connector rebuilds that class in a guest process
+            # where no host hook or database exists, so a gate placed there
+            # is absent exactly for the transport that most needs it.
+            agent_tools.extend(gate_mcp_tools(server_tools))
             if server_tools:
                 loaded_servers.append(server_name)
             logger.info(f"Found {len(server_tools)} tools from server {server_name}")
diff --git a/src/xagent/core/tools/adapters/vibe/write_gate.py b/src/xagent/core/tools/adapters/vibe/write_gate.py
new file mode 100644
index 000000000..3ef28c27b
--- /dev/null
+++ b/src/xagent/core/tools/adapters/vibe/write_gate.py
@@ -0,0 +1,173 @@
+"""The seam a host uses to require approval before an MCP write executes.
+
+A gated call is not executed. The hook is handed the call exactly as it was
+about to run -- tool name and arguments -- and answers with either "run it"
+or a pause carrying the frozen payload's identity. On approval the same
+arguments are executed verbatim, because they were never handed back to the
+model to be written a second time.
+
+**Why a host-injected hook rather than a direct call into the web layer.**
+This module is imported by the tool adapters, which
+``sandboxed_tool/tool_runner.py`` reconstructs inside the sandbox for every
+npx/uvx MCP tool -- a process where sqlalchemy is not installed. Anything
+here that reached for a database would turn every sandboxed tool call into a
+``ModuleNotFoundError``. So this file stays free of the web layer and the
+host injects the policy instead.
+
+Where the gate is *consulted* is a separate question, answered in
+``write_gate_tool.py``: on the host, in ``WriteGateTool``, above both the
+direct adapter and the sandbox wrapper. A sandboxed write is therefore gated
+before anything crosses into the guest, which an earlier revision -- asking
+from inside the adapter, where a supported npx/uvx connector only ever runs
+in the guest -- got exactly backwards.
+
+Not registering a hook *is* the off switch. No hook means no gate: nothing
+to consult, nothing to record, and no behavior change at all -- including
+the tool metadata the scheduler reads.
+
+**Not a trust boundary.** The hook decides using, among other things, a
+server's own ``readOnlyHint``/``destructiveHint`` annotations, which the MCP
+spec says a client must not trust from an untrusted server. What this seam
+guarantees is narrower and worth stating exactly: *if* a call is gated, the
+arguments that eventually execute are the ones that were shown, byte for
+byte. It does not guarantee that every dangerous call gets gated.
+
+**Known limits of this version.** Both are deliberate, and neither is
+described elsewhere in this module as if it were solved:
+
+*An approval is not bound to a connector identity.* It is bound to the
+interaction it was shown for, and nothing more. If a same-name MCP server is
+repointed to another endpoint, reauthorized to a different account, or
+deleted and recreated between the pause and the answer, the approved
+arguments still execute -- against whatever that name resolves to when the
+answer arrives. Carrying an immutable server/account identity would need the
+MCP connection layer to expose one, which it does not today.
+
+*Only a surface that delivers the chosen option value verbatim can grant an
+approval.* The pause offers machine values (``approve``/``reject``), and the
+host decides what counts as consent. xagent's own ``ClarificationForm``
+cannot take part: it submits each answer's *display label* rather than the
+option's value -- for every interaction type, not just ``confirm``, which it
+renders as a localized yes/no switch that ignores ``options`` entirely. A
+pause surfaced through that form therefore reaches the host as ``"Yes"`` and
+is voided, not executed. The supported surface in this version is a host
+that passes the value through untouched (Toby delivers the Slack button's
+value as the resume message).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any, Callable, Optional
+
+
+@dataclass(frozen=True)
+class GatedCall:
+    """One MCP call presented to the gate before it runs."""
+
+    tool_name: str
+    """The tool's runtime name, as the model called it."""
+
+    server_name: str
+    """Normalized identity of the MCP server the tool came from."""
+
+    arguments: Mapping[str, Any]
+    """The arguments the call would have executed with.
+
+    Exactly as the model produced them, which is also exactly what a replay
+    re-enters the tool with: the adapter's own deterministic handling of them
+    then runs identically both times.
+    """
+
+    write_hint: str
+    """The server's own write declaration: an ``MCPWriteHint`` value.
+
+    Carried as a plain string so this module stays independent of the
+    adapter's enum. A hook must treat everything except ``"read_only"`` as a
+    write: ``"undeclared"`` is the common case, not a promise of safety.
+    """
+
+
+@dataclass(frozen=True)
+class GateDecision:
+    """What the host decided about one gated call.
+
+    ``interaction_id`` is the identity the frozen payload was stored under
+    and the identity the resume callback will be handed back. It is the
+    hook's to mint: the adapter neither generates nor interprets it, it only
+    carries it into the pause so the two halves meet.
+    """
+
+    approval_required: bool
+    interaction_id: str = ""
+    message: str = ""
+
+
+WriteGateHook = Callable[[GatedCall], Optional[GateDecision]]
+
+# Given the interaction's identity, whether the user approved, and a callable
+# that runs one frozen argument set, the host loads the payload it recorded,
+# settles that approval exactly once, and returns the tool result. The
+# executor is passed in rather than imported because only the tool knows how
+# to place a call on its own connection -- and going through it is what keeps
+# the replay on the tool's normal authorization and error-mapping path.
+WriteGateResumeHook = Callable[..., Any]
+
+_HOOK: WriteGateHook | None = None
+_RESUME_HOOK: WriteGateResumeHook | None = None
+
+
+def set_write_gate_hook(hook: WriteGateHook | None) -> None:
+    """Install (or clear) the process-wide approval hook.
+
+    Idempotent and last-writer-wins, matching ``set_connector_runtime_resolver``.
+    Passing ``None`` restores ungated execution.
+    """
+    global _HOOK
+    _HOOK = hook
+
+
+def get_write_gate_hook() -> WriteGateHook | None:
+    """Return the installed hook, or ``None`` when nothing gates writes."""
+    return _HOOK
+
+
+def set_write_gate_resume_hook(hook: WriteGateResumeHook | None) -> None:
+    """Install (or clear) the hook that settles an approved or rejected call."""
+    global _RESUME_HOOK
+    _RESUME_HOOK = hook
+
+
+def get_write_gate_resume_hook() -> WriteGateResumeHook | None:
+    """Return the installed resume hook, or ``None``."""
+    return _RESUME_HOOK
+
+
+def consult_write_gate(call: GatedCall) -> GateDecision | None:
+    """Ask the installed hook about ``call``; ``None`` means "just run it".
+
+    A hook that raises is treated as no decision and the call proceeds. That
+    direction is deliberate and is the opposite of what a security boundary
+    would do, for the reason in the module docstring: this seam makes an
+    approved call faithful, it is not what keeps a dangerous call from
+    running. Failing closed here would let a transient database error strand
+    every connector call in a workspace behind an approval nobody can grant,
+    which trades a bounded loss of gating for an unbounded loss of function.
+    The host owns the decision to fail closed on its own side, where it can
+    tell a policy miss from an outage.
+    """
+    hook = _HOOK
+    if hook is None:
+        return None
+    try:
+        return hook(call)
+    except Exception:  # noqa: BLE001 - see the docstring
+        import logging
+
+        logging.getLogger(__name__).warning(
+            "Write gate hook failed for %s; executing ungated",
+            call.tool_name,
+            exc_info=True,
+        )
+        return None
diff --git a/src/xagent/core/tools/adapters/vibe/write_gate_tool.py b/src/xagent/core/tools/adapters/vibe/write_gate_tool.py
new file mode 100644
index 000000000..4da3c3226
--- /dev/null
+++ b/src/xagent/core/tools/adapters/vibe/write_gate_tool.py
@@ -0,0 +1,258 @@
+"""The host-side boundary where a connector write waits for approval.
+
+**Why this is a wrapper and not a check inside the adapter.** A supported
+npx/uvx connector does not execute in this process at all: the sandbox
+serializes the adapter class and its constructor data, and
+``sandboxed_tool/tool_runner.py`` rebuilds it with ``importlib`` and
+``cloudpickle`` in a guest process, then calls ``run_json_async`` there.
+Anything the host installs as module state -- a hook, a database session --
+does not exist on that side, so a gate placed inside the adapter is simply
+absent for the transport that most needs it. The first version of this
+feature did exactly that and left every sandboxed write ungated.
+
+Placed here, the gate wraps *whatever* the loader produced: a bare
+``MCPToolAdapter`` for a direct connection, or a ``SandboxedToolWrapper``
+around one for npx/uvx. Both look identical from outside, both are entered
+in the host process, and neither can be reached without passing through
+this object first.
+
+The wrapper is also what makes resumption honest. It replays an approved
+call by re-entering the target's own ``run_json_async`` with a flag that
+suppresses re-gating, so the authorization check, ``UserContext``, the
+delegated-credential refresh retry and the safe connector-error mapping all
+still run -- rather than reaching past them into the low-level call, which
+is what an earlier revision did.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Iterable, Mapping
+from contextvars import ContextVar
+from typing import Any, Optional, Type
+
+from pydantic import BaseModel
+
+from ...user_interaction import WAITING_FOR_USER_STATUS
+from .base import AbstractBaseTool, ToolMetadata
+from .write_gate import (
+    GatedCall,
+    consult_write_gate,
+    get_write_gate_hook,
+    get_write_gate_resume_hook,
+)
+
+logger = logging.getLogger(__name__)
+
+# Set while a wrapper is replaying an approved call, so the nested
+# ``run_json_async`` runs its normal authorization, retry and error handling
+# without asking the gate a second question about a call the user already
+# answered. A ContextVar rather than an attribute: the replay is awaited on
+# the same task, and a flag on the instance would leak across concurrent
+# calls to the same tool.
+_REPLAYING: ContextVar[bool] = ContextVar("xagent_write_gate_replaying", default=False)
+
+
+def is_replaying_approved_call() -> bool:
+    """Whether the current task is replaying an already-approved call."""
+    return _REPLAYING.get()
+
+
+class WriteGateTool(AbstractBaseTool):
+    """Holds a connector write until a human answers, then replays it verbatim.
+
+    Delegates every descriptive surface to the wrapped tool, so the model and
+    the tool-selection layers cannot tell the difference: what a gate changes
+    is *when* a call runs, never what it looks like.
+    """
+
+    def __init__(self, target_tool: AbstractBaseTool) -> None:
+        self._target = target_tool
+
+    @property
+    def target(self) -> AbstractBaseTool:
+        """The wrapped tool, for callers that need the transport itself."""
+        return self._target
+
+    @property
+    def name(self) -> str:
+        return self._target.name
+
+    @property
+    def description(self) -> str:
+        return self._target.description
+
+    @property
+    def tags(self) -> list[str]:
+        return self._target.tags
+
+    @property
+    def metadata(self) -> ToolMetadata:
+        """The target's metadata, except that a gated write is never batched.
+
+        ReAct groups consecutive ``concurrency_safe`` calls into one segment
+        and then delivers **one** answer string to every pending interaction
+        in it. A single ``approve`` would therefore authorize every
+        independently frozen write in the batch, and the approver never got
+        to judge them separately.
+
+        Reporting ``False`` is not a workaround for that scheduling detail;
+        it is what the flag actually means here. The pattern's own contract
+        (see its I5 note) makes ``concurrency_safe`` an idempotency promise
+        as well as a concurrency one, and a call parked for human approval
+        is not idempotent -- replaying it publishes twice. So a gated tool
+        runs alone, one approval decides one write, and the answer cannot be
+        spread across calls the user never saw.
+
+        Scope: only a connection whose operator opted into
+        ``concurrency_safe`` is affected, and only while a gate is actually
+        installed. Reads on such a connection lose batching too --
+        ``metadata`` is read before there is a call to ask the hook about, so
+        it cannot know whether *this* invocation would pause. That is the
+        conservative direction; the alternative is a batch whose single
+        answer authorizes a write nobody was shown.
+        """
+        metadata = self._target.metadata
+        if not metadata.concurrency_safe:
+            return metadata
+        if get_write_gate_hook() is None:
+            # Nothing in this process can pause a call, so nothing can spread
+            # one answer across a batch. Returning the target's own metadata
+            # is what keeps the off switch total: an unregistered gate must
+            # not silently cost an operator the batching they configured.
+            return metadata
+        return metadata.model_copy(update={"concurrency_safe": False})
+
+    @property
+    def is_sandboxed(self) -> bool:
+        return bool(getattr(self._target, "is_sandboxed", False))
+
+    def args_type(self) -> Type[BaseModel]:
+        return self._target.args_type()
+
+    def return_type(self) -> Type[BaseModel]:
+        return self._target.return_type()
+
+    def state_type(self) -> Optional[Type[BaseModel]]:
+        return self._target.state_type()
+
+    def return_value_as_string(self, value: Any) -> str:
+        return self._target.return_value_as_string(value)
+
+    def is_async(self) -> bool:
+        return True
+
+    def run_json_sync(self, args: Mapping[str, Any]) -> Any:
+        """Refused: nothing on this path can pause for an answer.
+
+        ``MCPToolAdapter.run_json_sync`` already raises -- MCP tools are
+        async-only -- so the direct transport was never reachable this way.
+        The sandbox wrapper's is different: it drives its own async path
+        through ``asyncio.run`` and works. Delegating here would therefore
+        run a sandboxed npx/uvx write with no gate consulted at all, through
+        exactly the transport this wrapper exists to cover.
+
+        Raising instead keeps "no execution without passing the gate" a
+        property of the object rather than of one of its two entry points,
+        and makes both transports refuse identically -- which is the point of
+        wrapping them in the same place.
+        """
+        raise RuntimeError(
+            f"MCP tool {self.name} is async only; please use run_json_async()"
+        )
+
+    async def run_json_async(self, args: Mapping[str, Any]) -> Any:
+        """Ask the gate, then either run the call or park it for approval."""
+        if is_replaying_approved_call():
+            # Already answered. Fall through so the target performs its own
+            # authorization, retry and error mapping on the replay.
+            return await self._target.run_json_async(args)
+
+        metadata = self.metadata
+        decision = consult_write_gate(
+            GatedCall(
+                tool_name=self.name,
+                # Both fields are read through ``metadata`` rather than off
+                # the target. The adapter mirrors its normalized server
+                # identity and its write declaration there, and
+                # ``SandboxedToolWrapper`` delegates ``metadata`` while
+                # forwarding no other attribute -- so reaching for
+                # ``target.source_server`` reported no server at all for
+                # exactly the npx/uvx transport this wrapper exists to cover.
+                server_name=metadata.source_server or "",
+                arguments=args,
+                write_hint=self._write_hint_value(metadata),
+            )
+        )
+        if decision is None or not decision.approval_required:
+            return await self._target.run_json_async(args)
+
+        return {
+            "status": WAITING_FOR_USER_STATUS,
+            "interaction_id": decision.interaction_id,
+            "message": decision.message or f"Approve running {self.name}?",
+            "interactions": [
+                {
+                    "type": "confirm",
+                    "field": "approve",
+                    "label": "Approve",
+                    "options": [
+                        {"label": "Approve", "value": "approve"},
+                        {"label": "Reject", "value": "reject"},
+                    ],
+                }
+            ],
+        }
+
+    async def resume_user_interaction(
+        self,
+        *,
+        interaction_id: str,
+        response: str,
+    ) -> Any:
+        """Replay the approved call, or void it.
+
+        The model is not consulted: the whole point is that the arguments
+        which execute are the ones that were shown. The replay re-enters the
+        target's ``run_json_async`` under ``_REPLAYING``, so authorization,
+        credential refresh and error mapping behave exactly as they do for
+        an ungated call.
+        """
+        hook = get_write_gate_resume_hook()
+        if hook is None:
+            return {
+                "success": False,
+                "status": "error",
+                "error": "This approval can no longer be completed.",
+            }
+        return await hook(
+            interaction_id=interaction_id,
+            response=response,
+            tool_name=self.name,
+            executor=self._replay,
+        )
+
+    async def _replay(self, arguments: Mapping[str, Any]) -> Any:
+        """Run one frozen argument set through the target's normal path."""
+        token = _REPLAYING.set(True)
+        try:
+            return await self._target.run_json_async(arguments)
+        finally:
+            _REPLAYING.reset(token)
+
+    def _write_hint_value(self, metadata: ToolMetadata) -> str:
+        """The target's write declaration, as a plain string."""
+        hint = metadata.mcp_write_hint
+        return hint if isinstance(hint, str) else "undeclared"
+
+
+def gate_mcp_tools(tools: "Iterable[AbstractBaseTool]") -> list[AbstractBaseTool]:
+    """Wrap loaded MCP tools so a write cannot execute before approval.
+
+    Applied once where the direct and sandboxed loaders converge, so neither
+    transport can be gated by accident and neither can be missed. Wrapping is
+    unconditional and cheap: whether a given call actually needs approval is
+    the installed hook's decision, made per call, and with no hook installed
+    every wrapper is a pass-through.
+    """
+    return [WriteGateTool(tool) for tool in tools]

From 60f56cb2c3479c36680d7601df55f85733284bd8 Mon Sep 17 00:00:00 2001
From: OliverBryant <2713999266@qq.com>
Date: Tue, 8 Sep 2026 16:33:29 +0800
Subject: [PATCH 2/5] test(mcp): pin the gate against the transport that
 bypassed it

The first version of this feature was green while every sandboxed write
went out ungated, because the tests called the adapter directly -- the
one place a supported npx/uvx connector never reaches. So these drive a
real SandboxedToolWrapper and assert on its guest dispatch itself
(sandbox.exec carrying --args-b64), which means a gate absent for the
sandbox transport, or an approval whose arguments are re-derived on the
way into the guest, fails here rather than in production.

Covered: nothing crosses into the guest before approval; an approved
call carries the shown arguments across the wrapper/runner seam byte for
byte; a rejection never dispatches; the host policy sees a real server
name through a wrapper that forwards no attributes; a replay enters the
target's own entry point and is not gated twice; a gated write is never
batched while a hook is installed and metadata is untouched when none
is; a failing hook executes ungated (deliberate, and the opposite of a
security boundary); and the sync entry point refuses.

The four existing loader tests now assert through a shared helper that
every tool leaves load_mcp_tools_as_agent_tools inside a WriteGateTool,
so deleting that one wiring line stops being silent. Those four were
failing on this branch and nobody saw it: the lint failure fail-fast'd
pytest away on every run.
---
 .../adapters/vibe/test_mcp_init_timeout.py    |   3 +-
 .../tools/adapters/vibe/test_write_gate.py    | 379 ++++++++++++++++++
 tests/core/tools/conftest.py                  |  23 ++
 .../tools/test_mcp_sandbox_integration.py     |   7 +-
 4 files changed, 408 insertions(+), 4 deletions(-)
 create mode 100644 tests/core/tools/adapters/vibe/test_write_gate.py
 create mode 100644 tests/core/tools/conftest.py

diff --git a/tests/core/tools/adapters/vibe/test_mcp_init_timeout.py b/tests/core/tools/adapters/vibe/test_mcp_init_timeout.py
index 3898b6d32..2c9d5f432 100644
--- a/tests/core/tools/adapters/vibe/test_mcp_init_timeout.py
+++ b/tests/core/tools/adapters/vibe/test_mcp_init_timeout.py
@@ -5,6 +5,7 @@
 
 import pytest
 
+from tests.core.tools.conftest import gated_targets
 from xagent.config import MCP_TOOL_INIT_TIMEOUT_SECONDS
 from xagent.core.tools.adapters.vibe import mcp_adapter as mcp_adapter_module
 from xagent.core.tools.adapters.vibe.config import MCPFailurePolicy
@@ -43,7 +44,7 @@ async def fake_load_direct(server_name, connection, **kwargs):
         }
     )
 
-    assert result.tools == (healthy_tool,)
+    assert gated_targets(result.tools) == (healthy_tool,)
     assert result.loaded_servers == ("healthy",)
     assert len(result.failures) == 1
     assert result.failures[0].server_name == "stalled"
diff --git a/tests/core/tools/adapters/vibe/test_write_gate.py b/tests/core/tools/adapters/vibe/test_write_gate.py
new file mode 100644
index 000000000..91e99d52d
--- /dev/null
+++ b/tests/core/tools/adapters/vibe/test_write_gate.py
@@ -0,0 +1,379 @@
+"""A gated MCP write must not execute before it is approved -- on either transport.
+
+The first version of this feature consulted the gate inside
+``MCPToolAdapter.run_json_async``, which is the one place a supported npx/uvx
+connector never reaches: the sandbox serializes that adapter and
+``tool_runner.py`` rebuilds it in a guest process where no host hook and no
+database exist. Every sandboxed write went out ungated while the host tests
+were green, because those tests called the adapter directly.
+
+So the tests here refuse a hand-rolled stand-in for that boundary. They drive
+a real ``SandboxedToolWrapper`` and watch its guest dispatch (``sandbox.exec``
+carrying ``--args-b64``), which means a gate that is absent for the sandbox
+transport, or an approval whose arguments are re-derived on the way into the
+guest, fails here rather than in production.
+"""
+
+import base64
+import json
+from typing import Any, Mapping, Optional
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from tests.core.tools.adapters.sandboxed_tool.conftest import FakeBaseTool
+from xagent.core.tools.adapters.vibe.mcp_adapter import MCPWriteHint
+from xagent.core.tools.adapters.vibe.sandboxed_tool.sandbox_config import sandbox_config
+from xagent.core.tools.adapters.vibe.sandboxed_tool.sandboxed_tool_wrapper import (
+    SandboxedToolWrapper,
+)
+from xagent.core.tools.adapters.vibe.write_gate import (
+    GatedCall,
+    GateDecision,
+    set_write_gate_hook,
+    set_write_gate_resume_hook,
+)
+from xagent.core.tools.adapters.vibe.write_gate_tool import (
+    gate_mcp_tools,
+    is_replaying_approved_call,
+)
+from xagent.core.tools.user_interaction import WAITING_FOR_USER_STATUS
+
+# The arguments a human would have been shown. Deliberately not a flat
+# {"text": "hi"}: nesting and a non-ASCII body are where a "re-serialize it
+# again on the way out" bug shows up as a difference instead of a coincidence.
+SHOWN_ARGUMENTS = {
+    "channel": "C123",
+    "blocks": [{"type": "section", "text": "内容 A"}],
+    "unfurl": False,
+}
+
+
+@sandbox_config()
+class _FakeMCPTool(FakeBaseTool):
+    """One MCP tool adapter's surface, without a connection behind it.
+
+    ``source_server``/``concurrency_safe``/``write_hint`` are the three
+    attributes ``AbstractBaseTool.metadata`` reads off a concrete tool, so a
+    wrapper that goes through ``metadata`` sees exactly what it would see
+    from a real adapter -- and one that reaches for a raw attribute instead
+    sees nothing once this is behind the sandbox wrapper.
+    """
+
+    source_server = "slack"
+    concurrency_safe = True
+
+    def __init__(self) -> None:
+        self.direct_calls: list[dict[str, Any]] = []
+        self.replay_flags: list[bool] = []
+
+    @property
+    def name(self) -> str:
+        return "slack_post_message"
+
+    @property
+    def write_hint(self) -> MCPWriteHint:
+        return MCPWriteHint.UNDECLARED
+
+    async def run_json_async(self, args: Mapping[str, Any]) -> Any:
+        self.direct_calls.append(dict(args))
+        self.replay_flags.append(is_replaying_approved_call())
+        return {"success": True, "posted": dict(args)}
+
+
+class _RecordingGate:
+    """A host gate that always demands approval and remembers what it saw."""
+
+    def __init__(self) -> None:
+        self.calls: list[GatedCall] = []
+
+    def __call__(self, call: GatedCall) -> Optional[GateDecision]:
+        self.calls.append(call)
+        return GateDecision(
+            approval_required=True,
+            interaction_id="interaction-1",
+            message="Post this to Slack?",
+        )
+
+
+class _RecordingResume:
+    """The host half: holds the frozen arguments and spends them once."""
+
+    def __init__(self, frozen: dict[str, Any]) -> None:
+        self.frozen = frozen
+        self.calls: list[tuple[str, str, str]] = []
+
+    async def __call__(
+        self,
+        *,
+        interaction_id: str,
+        response: str,
+        tool_name: str,
+        executor: Any,
+    ) -> Any:
+        # The keyword set is the contract the saas side implements against.
+        # Pinned by signature: a new kwarg on the caller breaks this here
+        # rather than at runtime in the other repository.
+        self.calls.append((interaction_id, response, tool_name))
+        if response != "approve":
+            return {"success": False, "status": "voided"}
+        return await executor(self.frozen)
+
+
+def _make_sandbox() -> MagicMock:
+    """A sandbox that reports success without running anything."""
+    payload = {"success": True, "output": "sent"}
+
+    def _exec(*args: Any, **kwargs: Any) -> MagicMock:
+        result = MagicMock()
+        result.exit_code = 0
+        result.stdout = json.dumps(payload) if args[0] == "cat" else ""
+        result.stderr = ""
+        return result
+
+    sandbox = MagicMock()
+    sandbox.name = "sandbox-test"
+    sandbox.exec = AsyncMock(side_effect=_exec)
+    sandbox.write_file = AsyncMock()
+    return sandbox
+
+
+def _guest_arguments(sandbox: MagicMock) -> list[dict[str, Any]]:
+    """The argument payloads that actually crossed into the guest process.
+
+    Read back out of the ``--args-b64`` the wrapper puts on the tool-runner
+    command line, so this asserts on what the guest would decode -- not on
+    what the host meant to send.
+    """
+    payloads = []
+    for call in sandbox.exec.call_args_list:
+        argv = list(call.args)
+        if "--args-b64" not in argv:
+            continue
+        encoded = argv[argv.index("--args-b64") + 1]
+        payloads.append(json.loads(base64.b64decode(encoded).decode("utf-8")))
+    return payloads
+
+
+@pytest.fixture(autouse=True)
+def _clear_gate_hooks():
+    """The hooks are process-global; a leak would silently gate other tests."""
+    yield
+    set_write_gate_hook(None)
+    set_write_gate_resume_hook(None)
+
+
+def _sandboxed_gated_tool() -> tuple[Any, _FakeMCPTool, MagicMock]:
+    target = _FakeMCPTool()
+    sandbox = _make_sandbox()
+    wrapper = SandboxedToolWrapper(target, sandbox)
+    (gated,) = gate_mcp_tools([wrapper])
+    return gated, target, sandbox
+
+
+async def test_sandboxed_write_does_not_reach_the_guest_before_approval():
+    """The regression the whole redesign exists for.
+
+    A gate inside the adapter cannot stop this call: by the time the adapter
+    runs, it is running in the guest. Asserting on ``sandbox.exec`` is what
+    makes that failure visible from the host side.
+    """
+    gate = _RecordingGate()
+    set_write_gate_hook(gate)
+    gated, target, sandbox = _sandboxed_gated_tool()
+
+    result = await gated.run_json_async(SHOWN_ARGUMENTS)
+
+    assert result["status"] == WAITING_FOR_USER_STATUS
+    assert result["interaction_id"] == "interaction-1"
+    assert _guest_arguments(sandbox) == []
+    assert sandbox.exec.await_count == 0
+    assert target.direct_calls == []
+    assert len(gate.calls) == 1
+
+
+async def test_approved_sandboxed_call_carries_the_shown_arguments_into_the_guest():
+    """What executes is what was shown, across the wrapper/runner seam."""
+    set_write_gate_hook(_RecordingGate())
+    resume = _RecordingResume(dict(SHOWN_ARGUMENTS))
+    set_write_gate_resume_hook(resume)
+    gated, _target, sandbox = _sandboxed_gated_tool()
+
+    paused = await gated.run_json_async(SHOWN_ARGUMENTS)
+    assert paused["status"] == WAITING_FOR_USER_STATUS
+
+    await gated.resume_user_interaction(
+        interaction_id=paused["interaction_id"], response="approve"
+    )
+
+    assert _guest_arguments(sandbox) == [SHOWN_ARGUMENTS]
+    assert resume.calls == [("interaction-1", "approve", "slack_post_message")]
+
+
+async def test_rejected_sandboxed_call_never_reaches_the_guest():
+    set_write_gate_hook(_RecordingGate())
+    set_write_gate_resume_hook(_RecordingResume(dict(SHOWN_ARGUMENTS)))
+    gated, _target, sandbox = _sandboxed_gated_tool()
+
+    paused = await gated.run_json_async(SHOWN_ARGUMENTS)
+    settled = await gated.resume_user_interaction(
+        interaction_id=paused["interaction_id"], response="reject"
+    )
+
+    assert settled["success"] is False
+    assert _guest_arguments(sandbox) == []
+
+
+async def test_the_gate_sees_the_server_name_through_the_sandbox_wrapper():
+    """``SandboxedToolWrapper`` forwards ``metadata`` and nothing else.
+
+    Reading ``target.source_server`` therefore reported no server at all for
+    exactly the transport this wrapper exists to cover, leaving the host
+    policy to decide about an anonymous call.
+    """
+    gate = _RecordingGate()
+    set_write_gate_hook(gate)
+    gated, _target, _sandbox = _sandboxed_gated_tool()
+
+    await gated.run_json_async(SHOWN_ARGUMENTS)
+
+    (call,) = gate.calls
+    assert call.server_name == "slack"
+    assert call.tool_name == "slack_post_message"
+    assert call.write_hint == MCPWriteHint.UNDECLARED.value
+    assert call.arguments == SHOWN_ARGUMENTS
+
+
+async def test_approved_direct_call_replays_through_the_targets_own_entry_point():
+    """Not past it.
+
+    The replay re-enters ``run_json_async``, so the target's authorization
+    check, delegated-credential refresh retry and connector-error mapping all
+    still run for an approved call. A replay that reached for a lower-level
+    call would skip every one of them, and the row is already spent.
+    """
+    set_write_gate_hook(_RecordingGate())
+    resume = _RecordingResume(dict(SHOWN_ARGUMENTS))
+    set_write_gate_resume_hook(resume)
+    target = _FakeMCPTool()
+    (gated,) = gate_mcp_tools([target])
+
+    paused = await gated.run_json_async(SHOWN_ARGUMENTS)
+    assert target.direct_calls == []
+
+    result = await gated.resume_user_interaction(
+        interaction_id=paused["interaction_id"], response="approve"
+    )
+
+    assert target.direct_calls == [SHOWN_ARGUMENTS]
+    assert result["posted"] == SHOWN_ARGUMENTS
+
+
+async def test_replay_is_not_gated_a_second_time():
+    """And the replay marker does not outlive the replay."""
+    gate = _RecordingGate()
+    set_write_gate_hook(gate)
+    set_write_gate_resume_hook(_RecordingResume(dict(SHOWN_ARGUMENTS)))
+    target = _FakeMCPTool()
+    (gated,) = gate_mcp_tools([target])
+
+    paused = await gated.run_json_async(SHOWN_ARGUMENTS)
+    await gated.resume_user_interaction(
+        interaction_id=paused["interaction_id"], response="approve"
+    )
+
+    assert len(gate.calls) == 1
+    assert target.replay_flags == [True]
+    assert is_replaying_approved_call() is False
+
+
+async def test_a_gated_write_is_never_batched_with_another():
+    """ReAct copies one answer to every interaction in a concurrent segment.
+
+    One ``approve`` would then authorize every independently frozen write in
+    the batch, including the ones the approver was never shown.
+    """
+    set_write_gate_hook(_RecordingGate())
+    target = _FakeMCPTool()
+    (gated,) = gate_mcp_tools([target])
+
+    assert target.metadata.concurrency_safe is True
+    assert gated.metadata.concurrency_safe is False
+    # Everything else about the tool is unchanged: the model must not be able
+    # to tell a gated tool from an ungated one.
+    assert gated.metadata.model_dump(exclude={"concurrency_safe"}) == (
+        target.metadata.model_dump(exclude={"concurrency_safe"})
+    )
+
+
+async def test_no_installed_gate_means_no_behavior_change_at_all():
+    """The off switch is "install nothing", and it has to be total.
+
+    Suppressing batching unconditionally would quietly cost every operator
+    who opted into ``concurrency_safe`` on a connection their batching, in a
+    process where nothing can pause a call in the first place.
+    """
+    target = _FakeMCPTool()
+    (gated,) = gate_mcp_tools([target])
+
+    assert gated.metadata == target.metadata
+
+    result = await gated.run_json_async(SHOWN_ARGUMENTS)
+
+    assert target.direct_calls == [SHOWN_ARGUMENTS]
+    assert result["success"] is True
+
+
+async def test_a_failing_gate_executes_ungated():
+    """Deliberate, and the opposite of what a security boundary would do.
+
+    This seam makes an approved call faithful; it is not what keeps a
+    dangerous call from running. Failing closed here would strand every
+    connector call in a workspace behind an approval nobody can grant.
+    """
+
+    def _broken_gate(call: GatedCall) -> Optional[GateDecision]:
+        raise RuntimeError("policy lookup failed")
+
+    set_write_gate_hook(_broken_gate)
+    target = _FakeMCPTool()
+    (gated,) = gate_mcp_tools([target])
+
+    result = await gated.run_json_async(SHOWN_ARGUMENTS)
+
+    assert result["success"] is True
+    assert target.direct_calls == [SHOWN_ARGUMENTS]
+
+
+async def test_an_approval_with_no_resume_hook_reports_an_error_instead_of_executing():
+    set_write_gate_hook(_RecordingGate())
+    target = _FakeMCPTool()
+    (gated,) = gate_mcp_tools([target])
+
+    paused = await gated.run_json_async(SHOWN_ARGUMENTS)
+    settled = await gated.resume_user_interaction(
+        interaction_id=paused["interaction_id"], response="approve"
+    )
+
+    assert settled["success"] is False
+    assert settled["status"] == "error"
+    assert target.direct_calls == []
+
+
+def test_the_sync_entry_point_cannot_run_a_gated_write():
+    """A gate is only a gate if there is no second door.
+
+    The direct adapter refuses ``run_json_sync`` on its own (MCP tools are
+    async-only), but ``SandboxedToolWrapper.run_json_sync`` works -- it just
+    drives its async path through ``asyncio.run``. Forwarding here would run
+    a sandboxed npx/uvx write with nothing consulted, through the one
+    transport the wrapper exists to cover.
+    """
+    set_write_gate_hook(_RecordingGate())
+    _gated, target, sandbox = _sandboxed_gated_tool()
+
+    with pytest.raises(RuntimeError, match="async only"):
+        _gated.run_json_sync(SHOWN_ARGUMENTS)
+
+    assert _guest_arguments(sandbox) == []
+    assert target.direct_calls == []
diff --git a/tests/core/tools/conftest.py b/tests/core/tools/conftest.py
new file mode 100644
index 000000000..16dfadac6
--- /dev/null
+++ b/tests/core/tools/conftest.py
@@ -0,0 +1,23 @@
+"""Shared helpers for the MCP tool-loading tests."""
+
+from typing import Any
+
+from xagent.core.tools.adapters.vibe.write_gate_tool import WriteGateTool
+
+
+def gated_targets(tools: Any) -> tuple[Any, ...]:
+    """The loader's tools, unwrapped from the approval gate it applies.
+
+    Every loaded MCP tool leaves ``load_mcp_tools_as_agent_tools`` inside a
+    ``WriteGateTool``: that function is the one place the direct and
+    sandboxed transports converge, which is why the gate is applied there
+    rather than inside the adapter -- a sandboxed npx/uvx connector rebuilds
+    the adapter in a guest process where no host hook exists, so a gate
+    placed there is absent for exactly the transport that most needs it.
+
+    Asserting through this helper keeps each caller's test about the thing it
+    was written for -- which transport ran, what a timeout did -- while still
+    failing if a tool ever escapes the loader ungated.
+    """
+    assert all(isinstance(tool, WriteGateTool) for tool in tools), tools
+    return tuple(tool.target for tool in tools)
diff --git a/tests/core/tools/test_mcp_sandbox_integration.py b/tests/core/tools/test_mcp_sandbox_integration.py
index 1a034bf32..a8b47ebd1 100644
--- a/tests/core/tools/test_mcp_sandbox_integration.py
+++ b/tests/core/tools/test_mcp_sandbox_integration.py
@@ -5,6 +5,7 @@
 import pytest
 from mcp.types import Tool as MCPTool
 
+from tests.core.tools.conftest import gated_targets
 from xagent.core.tools.adapters.vibe.mcp_adapter import (
     MCPFailurePhase,
     MCPLoadResult,
@@ -127,7 +128,7 @@ async def test_sandboxed_stdio_server_uses_sandbox_path(self):
                 sandbox=sandbox,
             )
 
-        assert result.tools == (wrapped_tool,)
+        assert gated_targets(result.tools) == (wrapped_tool,)
         assert result.loaded_servers == ("demo",)
         assert result.failures == ()
         mock_list.assert_awaited_once_with(sandbox, connection)
@@ -162,7 +163,7 @@ async def test_non_sandbox_connection_uses_direct_path(self):
                 sandbox=MagicMock(),
             )
 
-        assert result.tools == (direct_tool,)
+        assert gated_targets(result.tools) == (direct_tool,)
         assert result.loaded_servers == ("demo",)
         assert result.failures == ()
         mock_direct.assert_awaited_once()
@@ -266,7 +267,7 @@ async def test_sandbox_wrap_failure_preserves_other_wrapped_tools(self):
                 {"demo": connection}, sandbox=sandbox
             )
 
-        assert result.tools == (wrapped_tool,)
+        assert gated_targets(result.tools) == (wrapped_tool,)
         assert result.loaded_servers == ("demo",)
         assert len(result.failures) == 1
         assert result.failures[0].phase is MCPFailurePhase.SANDBOX_TOOL_WRAP

From 2de26e1ef92ae0b9b34d181ab147e291760e0110 Mon Sep 17 00:00:00 2001
From: OliverBryant <2713999266@qq.com>
Date: Tue, 8 Sep 2026 17:51:13 +0800
Subject: [PATCH 3/5] fix(mcp): mark a wrapped tool's server loaded in the
 STRICT summary

_build_mcp_load_summary decided which servers had loaded by reading
source_server off each tool. Every wrapper in this pipeline delegates
metadata and forwards no other attribute, so a wrapped tool answered
None: its tools were counted in successful_tool_count while its server
was never added to loaded_keys, and the loop that follows then
synthesized no_tools_returned for it -- an internally contradictory
summary that fails STRICT setup for a connector that had just answered.

This was already the behaviour on the sandbox transport before any gate
existed. Verified on origin/main in a clean worktree: a
SandboxedToolWrapper reports getattr(tool, "source_server") as None
while metadata.source_server is correct. Reading through the metadata
contract therefore fixes the npx/uvx case as well, where delegating the
attribute on the new wrapper alone would have left it broken.

The tests drive a metadata-only wrapper -- the contract every wrapper
here actually keeps, not a stand-in for one class -- plus the gate, plus
both stacked, which is the production npx/uvx shape. One test guards the
other direction: a server that genuinely returned nothing must still be
reported, since that is the condition STRICT setup exists to catch.
---
 .../core/tools/adapters/vibe/mcp_tools.py     |   9 +-
 .../adapters/vibe/test_mcp_load_summary.py    | 158 ++++++++++++++++++
 2 files changed, 166 insertions(+), 1 deletion(-)
 create mode 100644 tests/core/tools/adapters/vibe/test_mcp_load_summary.py

diff --git a/src/xagent/core/tools/adapters/vibe/mcp_tools.py b/src/xagent/core/tools/adapters/vibe/mcp_tools.py
index b7351a708..5720b5f89 100644
--- a/src/xagent/core/tools/adapters/vibe/mcp_tools.py
+++ b/src/xagent/core/tools/adapters/vibe/mcp_tools.py
@@ -99,7 +99,14 @@ def _build_mcp_load_summary(
             continue
 
         successful_tool_count += 1
-        source_server = getattr(tool, "source_server", None)
+        # Read through the metadata contract, not off the tool. Every wrapper
+        # in this pipeline delegates ``metadata`` and forwards no other
+        # attribute -- ``SandboxedToolWrapper`` already did, so a healthy
+        # npx/uvx connector counted its tools here while never being marked
+        # loaded, and the loop below then synthesized ``no_tools_returned``
+        # for it and failed STRICT setup. Fixing the wrapper instead would
+        # have left the sandbox transport broken exactly as it is today.
+        source_server = getattr(getattr(tool, "metadata", None), "source_server", None)
         if type(source_server) is not str:
             continue
         key = normalize_mcp_server_name(source_server)
diff --git a/tests/core/tools/adapters/vibe/test_mcp_load_summary.py b/tests/core/tools/adapters/vibe/test_mcp_load_summary.py
new file mode 100644
index 000000000..7fad564e0
--- /dev/null
+++ b/tests/core/tools/adapters/vibe/test_mcp_load_summary.py
@@ -0,0 +1,158 @@
+"""A healthy connector must not be reported as having returned no tools.
+
+``_build_mcp_load_summary`` decided which servers loaded by reading
+``source_server`` off each tool. Every wrapper in this pipeline delegates
+``metadata`` and forwards no other attribute, so a wrapped tool answered
+``None``: its tools were counted, its server was never marked loaded, and the
+loop that follows then synthesized ``no_tools_returned`` for it -- failing
+STRICT setup for a connector that had just answered.
+
+That was already true on the sandbox transport before any gate existed, which
+is why the fix is here and not on one wrapper: reading through the metadata
+contract covers every wrapper that honours it, including the ones not written
+yet.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Mapping, Type
+
+import pytest
+from pydantic import BaseModel
+
+from xagent.core.tools.adapters.vibe.base import AbstractBaseTool
+from xagent.core.tools.adapters.vibe.mcp_tools import _build_mcp_load_summary
+from xagent.core.tools.adapters.vibe.write_gate_tool import gate_mcp_tools
+
+CONFIGS = [{"name": "slack"}]
+
+
+class _ArgsModel(BaseModel):
+    pass
+
+
+class _AdapterLike(AbstractBaseTool):
+    """The shape ``MCPToolAdapter`` presents: the attribute *and* the metadata."""
+
+    source_server = "slack"
+
+    @property
+    def name(self) -> str:
+        return "slack_post_message"
+
+    @property
+    def description(self) -> str:
+        return "post"
+
+    @property
+    def tags(self) -> list[str]:
+        return []
+
+    def args_type(self) -> Type[BaseModel]:
+        return _ArgsModel
+
+    def return_type(self) -> Type[BaseModel]:
+        return _ArgsModel
+
+    def run_json_sync(self, args: Mapping[str, Any]) -> Any:
+        return {}
+
+    async def run_json_async(self, args: Mapping[str, Any]) -> Any:
+        return {}
+
+
+class _MetadataOnlyWrapper(AbstractBaseTool):
+    """A wrapper that delegates ``metadata`` and forwards nothing else.
+
+    Not a stand-in for one particular class: this is the contract every
+    wrapper in the pipeline actually keeps. ``SandboxedToolWrapper`` has kept
+    exactly this shape since before the write gate existed.
+    """
+
+    def __init__(self, target: AbstractBaseTool) -> None:
+        self._target = target
+
+    @property
+    def name(self) -> str:
+        return self._target.name
+
+    @property
+    def description(self) -> str:
+        return self._target.description
+
+    @property
+    def tags(self) -> list[str]:
+        return self._target.tags
+
+    @property
+    def metadata(self):  # type: ignore[override]
+        return self._target.metadata
+
+    def args_type(self) -> Type[BaseModel]:
+        return self._target.args_type()
+
+    def return_type(self) -> Type[BaseModel]:
+        return self._target.return_type()
+
+    def run_json_sync(self, args: Mapping[str, Any]) -> Any:
+        return self._target.run_json_sync(args)
+
+    async def run_json_async(self, args: Mapping[str, Any]) -> Any:
+        return await self._target.run_json_async(args)
+
+
+def test_a_bare_adapter_marks_its_server_loaded() -> None:
+    """The baseline: this never broke, and must keep working."""
+    summary = _build_mcp_load_summary(CONFIGS, [_AdapterLike()])
+
+    assert summary.loaded_servers == ("slack",)
+    assert summary.failures == ()
+    assert summary.successful_tool_count == 1
+
+
+def test_a_metadata_only_wrapper_marks_its_server_loaded() -> None:
+    """The pre-existing bug, independent of the write gate.
+
+    Reading the raw attribute made this answer ``None``, so the server fell
+    through to ``no_tools_returned`` while its tool was still counted -- an
+    internally contradictory summary that failed STRICT setup for a connector
+    that had answered.
+    """
+    summary = _build_mcp_load_summary(CONFIGS, [_MetadataOnlyWrapper(_AdapterLike())])
+
+    assert summary.loaded_servers == ("slack",)
+    assert summary.failures == ()
+    assert summary.successful_tool_count == 1
+
+
+def test_a_gated_tool_marks_its_server_loaded() -> None:
+    """And the wrapper this PR adds, through the real ``gate_mcp_tools``."""
+    summary = _build_mcp_load_summary(CONFIGS, list(gate_mcp_tools([_AdapterLike()])))
+
+    assert summary.loaded_servers == ("slack",)
+    assert summary.failures == ()
+    assert summary.successful_tool_count == 1
+
+
+def test_a_gated_sandbox_wrapper_marks_its_server_loaded() -> None:
+    """Both wrappers stacked, which is the production npx/uvx shape."""
+    stacked = gate_mcp_tools([_MetadataOnlyWrapper(_AdapterLike())])
+
+    summary = _build_mcp_load_summary(CONFIGS, list(stacked))
+
+    assert summary.loaded_servers == ("slack",)
+    assert summary.failures == ()
+
+
+@pytest.mark.parametrize("tools", [[], None])
+def test_a_server_that_returned_nothing_is_still_reported(tools) -> None:
+    """The guard on the other side: the failure path must stay reachable.
+
+    A fix that marked every requested server loaded would silence the real
+    ``no_tools_returned``, which is the condition STRICT setup exists to catch.
+    """
+    summary = _build_mcp_load_summary(CONFIGS, list(tools or []))
+
+    assert summary.loaded_servers == ()
+    assert [failure.reason for failure in summary.failures] == ["no_tools_returned"]
+    assert summary.successful_tool_count == 0

From 79a8b25c0e1f8e4396644655472016a5dfdd8aa0 Mon Sep 17 00:00:00 2001
From: OliverBryant <2713999266@qq.com>
Date: Tue, 8 Sep 2026 18:50:36 +0800
Subject: [PATCH 4/5] test(mcp): give the load-summary double the metadata
 every real tool has

The summary now reads source_server through the metadata contract, which
broke a double declaring only the bare attribute. Fixed the double rather
than adding a fallback to the production read: every tool that reaches
_build_mcp_load_summary is an AbstractBaseTool -- whose metadata property
mirrors source_server -- or an UnavailableMCPTool, which the summary
takes on an earlier branch. A raw-attribute fallback would keep alive a
shape production cannot produce, and would reinstate the behaviour that
made a wrapped tool look like a server returning nothing.
---
 .../tools/adapters/vibe/test_selection_spec.py     | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/tests/core/tools/adapters/vibe/test_selection_spec.py b/tests/core/tools/adapters/vibe/test_selection_spec.py
index e22994e12..29658594a 100644
--- a/tests/core/tools/adapters/vibe/test_selection_spec.py
+++ b/tests/core/tools/adapters/vibe/test_selection_spec.py
@@ -500,8 +500,22 @@ async def test_mcp_summary_reports_partial_success_and_same_server_failure(
     from xagent.core.tools.adapters.vibe import mcp_tools
 
     class _LoadedTool:
+        # Carries ``metadata``, because every tool that reaches the summary
+        # does: they are all ``AbstractBaseTool`` instances (whose metadata
+        # property mirrors ``source_server``) or ``UnavailableMCPTool``,
+        # which the summary takes on an earlier branch. A double with only
+        # the bare attribute is thinner than anything production produces,
+        # and the summary reads the metadata contract so that a wrapper --
+        # which forwards ``metadata`` and no other attribute -- is not
+        # mistaken for a server that returned nothing.
         source_server = "gmail"
 
+        @property
+        def metadata(self):
+            from xagent.core.tools.adapters.vibe.base import ToolMetadata
+
+            return ToolMetadata(name="gmail_send", source_server=self.source_server)
+
     gmail_failure = UnavailableMCPTool(
         server_name="Gmail",
         server_id=1,

From 1b0230751b61c0663cd59f0adce92897b4071c3d Mon Sep 17 00:00:00 2001
From: OliverBryant <2713999266@qq.com>
Date: Tue, 8 Sep 2026 19:34:52 +0800
Subject: [PATCH 5/5] docs(mcp): say which half of a gated call is actually
 frozen

The gate is handed the model's arguments, and GatedCall.arguments claimed
they were "the arguments the call would have executed with". They are not
the wire payload: before executing, the adapter normalizes them against
the schema, applies the args model's defaults and coercion, strips any
runtime-bound field the model tried to set, and injects the current values
of its runtime bindings.

The first three are pure functions of those arguments and the schema, so
preview and replay agree -- the replay re-enters the same entry point with
the same arguments. The injected runtime values are not, and a resume runs
on a rebuilt instance, so a binding whose source changed while the
question waited sends the new value. That is the connector-identity limit
already declared here, seen in the argument dimension.

Recorded as a third known limit rather than papered over, with what
closing it would actually take. Freezing the prepared payload needs a
second guest entry point: tool_runner.py re-enters run_json_async with
whatever arguments it is handed, and preparation is not idempotent -- the
key set _runtime_bound_tool_argument_names strips is exactly the key set
_runtime_tool_arguments injects, so a round trip discards precisely the
half worth freezing. Detecting the change instead needs the host to store
the binding values with the approval and compare at replay, which is a
consumer this seam does not have; adding the field without one would be
another value nobody reads, which is what the connector_identity field
already was.
---
 .../core/tools/adapters/vibe/write_gate.py    | 43 ++++++++++++++++---
 1 file changed, 38 insertions(+), 5 deletions(-)

diff --git a/src/xagent/core/tools/adapters/vibe/write_gate.py b/src/xagent/core/tools/adapters/vibe/write_gate.py
index 3ef28c27b..1a6df318f 100644
--- a/src/xagent/core/tools/adapters/vibe/write_gate.py
+++ b/src/xagent/core/tools/adapters/vibe/write_gate.py
@@ -53,6 +53,30 @@
 is voided, not executed. The supported surface in this version is a host
 that passes the value through untouched (Toby delivers the Slack button's
 value as the resume message).
+
+*Runtime-bound arguments are re-derived at execution, not frozen.* What is
+frozen and replayed byte for byte is everything the model authored -- which
+is also everything the approver was shown. On top of that the adapter
+injects its runtime bindings' current values (``_runtime_tool_arguments``,
+``_runtime_mcp_meta``), resolved from the connector runtime of whichever
+adapter instance executes. A resume runs on a rebuilt instance, so if a
+binding's source changed while the question waited, the value that goes out
+is the new one. This is the connector-identity limit above, seen in the
+argument dimension rather than the endpoint dimension.
+
+Closing it needs one of two things that are deliberately not in this
+change. Freezing the *prepared* payload means the guest can no longer
+prepare it: ``sandboxed_tool/tool_runner.py`` re-enters ``run_json_async``
+with the arguments it is handed, and preparation is not idempotent: the key
+set ``_runtime_bound_tool_argument_names`` strips is exactly the key set
+``_runtime_tool_arguments`` injects, so a prepared payload fed back through
+that entry point has its frozen runtime values stripped as though the model
+had set them and then replaced with current ones -- the round trip discards
+precisely the half that was worth freezing. So it needs a second guest
+entry point and a marker in the execution spec. Detecting the change instead
+needs the *host* to store the binding values with the approval and compare
+them at replay, which is a consumer this seam does not have; adding the
+field without it would be another value nobody reads.
 """
 
 from __future__ import annotations
@@ -73,11 +97,20 @@ class GatedCall:
     """Normalized identity of the MCP server the tool came from."""
 
     arguments: Mapping[str, Any]
-    """The arguments the call would have executed with.
-
-    Exactly as the model produced them, which is also exactly what a replay
-    re-enters the tool with: the adapter's own deterministic handling of them
-    then runs identically both times.
+    """The model-authored arguments, exactly as the model produced them.
+
+    Not the payload that goes on the wire, and the difference is worth
+    stating precisely because this seam's whole promise is about fidelity.
+    Before a call executes, the adapter normalizes these against the tool's
+    schema, applies the args model's defaults and coercion, strips any
+    runtime-bound field the model tried to set, and injects the current
+    values of its runtime bindings.
+
+    The first three of those are pure functions of these arguments and the
+    tool's schema, so they produce the same result at preview and at replay
+    -- the replay re-enters the same public entry point with these exact
+    arguments. The injected runtime values are not: see the third known
+    limit in the module docstring.
     """
 
     write_hint: str