Agent: preserve target denial reasons through MCP

docs/MCP-AGENT-HOST.md

@@ -104,6 +104,17 @@ The first MCP adapter intentionally exposes named operations rather than arbitra
 
 This is narrower than the raw paired-target service by design.
 
+Expected target-side policy and reachability failures are returned as structured results rather than opaque MCP server errors:
+
+```text
+ok: false
+error:
+  kind: denied | unavailable | invalid_request
+  message: <target/transport reason>
+```
+
+This lets the MCP host distinguish an authorization denial from transport loss or an invalid target request without exposing credentials.
+
 ## Codex connection
 
 Codex supports MCP servers in both the CLI and IDE extension, with shared configuration.

src/relmote/mcp_server.py

@@ -73,6 +73,26 @@ def _target_context(target: str) -> dict[str, Any]:
             "execution_location": "target",
         }
 
+    @staticmethod
+    def _expected_error(
+        target: str,
+        exc: Exception,
+    ) -> dict[str, Any]:
+        if isinstance(exc, PermissionError):
+            kind = "denied"
+        elif isinstance(exc, ConnectionError):
+            kind = "unavailable"
+        else:
+            kind = "invalid_request"
+        return {
+            "ok": False,
+            "context": RelmoteMCPTools._target_context(target),
+            "error": {
+                "kind": kind,
+                "message": str(exc),
+            },
+        }
+
     def targets(self) -> dict[str, Any]:
         targets = []
         for item in self.service.targets():
@@ -86,29 +106,50 @@ def targets(self) -> dict[str, Any]:
         }
 
     def status(self, target: str) -> dict[str, Any]:
+        try:
+            value = self.service.status(target)
+        except (PermissionError, ConnectionError, ValueError) as exc:
+            return self._expected_error(target, exc)
         return {
+            "ok": True,
             "context": self._target_context(target),
-            "target_status": self.service.status(target),
+            "target_status": value,
         }
 
     def list(self, target: str, path: str = ".") -> dict[str, Any]:
+        context = {
+            **self._target_context(target),
+            "path": path,
+            "path_location": "target_workspace",
+        }
+        try:
+            entries = self.service.list(target, path)
+        except (PermissionError, ConnectionError, ValueError) as exc:
+            value = self._expected_error(target, exc)
+            value["context"] = context
+            return value
         return {
-            "context": {
-                **self._target_context(target),
-                "path": path,
-                "path_location": "target_workspace",
-            },
-            "entries": self.service.list(target, path),
+            "ok": True,
+            "context": context,
+            "entries": entries,
         }
 
     def read(self, target: str, path: str) -> dict[str, Any]:
+        context = {
+            **self._target_context(target),
+            "path": path,
+            "path_location": "target_workspace",
+        }
+        try:
+            text = self.service.read(target, path)
+        except (PermissionError, ConnectionError, ValueError) as exc:
+            value = self._expected_error(target, exc)
+            value["context"] = context
+            return value
         return {
-            "context": {
-                **self._target_context(target),
-                "path": path,
-                "path_location": "target_workspace",
-            },
-            "text": self.service.read(target, path),
+            "ok": True,
+            "context": context,
+            "text": text,
         }
 
     def exec(
@@ -134,19 +175,26 @@ def exec(
         else:  # Defensive for direct Python callers; MCP validates Literal.
             raise ValueError(f"unsupported MCP execution operation: {operation}")
 
-        result = self.service.exec(
-            target,
-            argv,
-            cwd=cwd,
-            timeout=timeout,
-        )
+        context = {
+            **self._target_context(target),
+            "cwd": cwd,
+            "cwd_location": "target_workspace",
+            "operation": operation,
+        }
+        try:
+            result = self.service.exec(
+                target,
+                argv,
+                cwd=cwd,
+                timeout=timeout,
+            )
+        except (PermissionError, ConnectionError, ValueError) as exc:
+            value = self._expected_error(target, exc)
+            value["context"] = context
+            return value
         return {
-            "context": {
-                **self._target_context(target),
-                "cwd": cwd,
-                "cwd_location": "target_workspace",
-                "operation": operation,
-            },
+            "ok": True,
+            "context": context,
             "result": result,
         }
 
@@ -227,8 +275,10 @@ def relmote_exec(
         Execution happens in the Target's authorized workspace, not on the
         local Agent Host. git_status and git_diff accept no extra arguments.
         pytest may receive explicit pytest arguments. Target-side workspace,
-        capability, and revocation policy remains authoritative. A denial must
-        not be bypassed through another transport.
+        capability, and revocation policy remains authoritative. Expected
+        denials/unavailable states are returned as structured ok=false results
+        so the reason stays visible to the MCP host. A denial must not be
+        bypassed through another transport.
         """
         return tools.exec(
             target,

tests/test_mcp_server.py

@@ -54,15 +54,18 @@ def test_mcp_tools_reuse_paired_target_service_boundary():
     assert targets["targets"][0]["workspace_location"] == "target"
 
     status = tools.status("target-a")
+    assert status["ok"] is True
     assert status["context"]["target_role"] == "paired_target"
     assert status["context"]["execution_location"] == "target"
     assert status["target_status"]["state"] == "active"
 
     listed = tools.list("target-a", "src")
+    assert listed["ok"] is True
     assert listed["context"]["path_location"] == "target_workspace"
     assert listed["entries"][0]["name"] == "README.md"
 
     read = tools.read("target-a", "README.md")
+    assert read["ok"] is True
     assert read["context"]["path_location"] == "target_workspace"
     assert read["text"] == "hello\n"
 
@@ -83,6 +86,7 @@ def test_mcp_exec_maps_named_operations_not_arbitrary_git_arguments():
     pytest_result = tools.exec("target-a", "pytest", args=["-q"])
 
     for value in (status_result, diff_result, pytest_result):
+        assert value["ok"] is True
         assert value["context"]["execution_location"] == "target"
         assert value["context"]["cwd_location"] == "target_workspace"
         assert value["result"]["returncode"] == 0
@@ -131,3 +135,46 @@ def test_context_tool_makes_execution_location_explicit():
     assert value["target_paths_belong_to"] == "target_workspace"
     assert value["target_execution_location"] == "target"
     assert value["agent_host_filesystem_exposed_as_target"] is False
+
+
+class DenyingService(FakeService):
+    def status(self, target):
+        raise PermissionError("agent session is not active")
+
+    def list(self, target, path="."):
+        raise ConnectionError("Relmote Agent API unavailable: connection refused")
+
+    def read(self, target, path):
+        raise ValueError("workspace path escapes configured root")
+
+    def exec(self, target, argv, *, cwd=".", timeout=30):
+        raise PermissionError(
+            "Relmote Agent API: agent capability not granted: terminal.exec"
+        )
+
+
+def test_expected_target_failures_are_structured_not_opaque_mcp_errors():
+    tools = RelmoteMCPTools(DenyingService())
+
+    status = tools.status("target-a")
+    assert status["ok"] is False
+    assert status["error"] == {
+        "kind": "denied",
+        "message": "agent session is not active",
+    }
+
+    listed = tools.list("target-a", ".")
+    assert listed["ok"] is False
+    assert listed["error"]["kind"] == "unavailable"
+    assert "connection refused" in listed["error"]["message"]
+
+    read = tools.read("target-a", "../secret")
+    assert read["ok"] is False
+    assert read["error"]["kind"] == "invalid_request"
+    assert "escapes configured root" in read["error"]["message"]
+
+    executed = tools.exec("target-a", "git_status")
+    assert executed["ok"] is False
+    assert executed["error"]["kind"] == "denied"
+    assert "terminal.exec" in executed["error"]["message"]
+    assert executed["context"]["execution_location"] == "target"