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/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/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..1a6df318f --- /dev/null +++ b/src/xagent/core/tools/adapters/vibe/write_gate.py @@ -0,0 +1,206 @@ +"""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). + +*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 + +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 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 + """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] 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_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 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, 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