Expose bound MCP tools to delegated Codex workers

docs/reference/local-delegation.md

@@ -74,6 +74,32 @@ workspace、Codex Session 与持久 delegation operation。`delegation inspect`
 规划投影会读回例如 `gpt-5.6-sol@xhigh`，start/resume 也把同一配置送入原 Session。
 这不扩大 requester grant，也不把 profile 冒充验收或结果返回回执。
 
+For a Codex binding, the delegation host also supplies one invocation-scoped
+`loopx_delegation` stdio MCP server to every fresh or resumed worker Session.
+Its command pins the selected worker `agent_id`, workspace, Goal, registry,
+runtime and operator execution configuration before Codex starts. The model
+cannot select or rewrite those values. Codex receives the server through
+per-invocation configuration, so LoopX does not modify the user's global Codex
+MCP settings and a resumed worker keeps the same binding. The server is required
+for this managed worker route and its already identity-scoped tools are approved
+inside that route; failure to start the server rejects the Turn instead of
+silently continuing without tools. The native tools are
+the collaboration and authorized delegation operations from that bound server;
+they do not add shell, Todo or external-action authority.
+
+The shell commands below remain the compatibility path for an already running
+Agent Session, a host without MCP support, or an operator who deliberately uses
+shell-only coordination. Both surfaces call the same `Delegations` service and
+preserve the same binding, operation and acceptance rules; the shell path is
+not a second control-plane implementation.
+
+中文：Codex binding 会为每个新建或续接的 worker Session 注入一次调用范围内的
+`loopx_delegation` stdio MCP server。启动前，host 已固定 worker `agent_id`、
+workspace、Goal、registry、runtime 与 operator execution configuration；模型不能
+选择或改写这些值。该配置不会修改用户的全局 Codex MCP 设置，也不会授予 shell、
+Todo 或外部动作权限。下方 shell 命令继续作为既有 Session、无 MCP host 或显式
+shell-only 协调的兼容入口；两种入口复用同一个 `Delegations` 服务和同一套验收规则。
+
 ## Use an existing Agent conversation through its shell
 
 An attached Codex or other shell-capable Agent can use the same execution

examples/codex-cli-native-mcp-materialization-smoke.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+"""Prove a fresh real Codex Turn can call an invocation-bound MCP tool."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+import shutil
+import sys
+import tempfile
+import uuid
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from loopx.control_plane.turn_driver.codex_cli import (  # noqa: E402
+    CODEX_STDIO_MCP_SERVER_SCHEMA_VERSION,
+    run_codex_cli_host,
+)
+
+
+def _request() -> dict[str, object]:
+    return {
+        "schema_version": "loopx_turn_host_request_v0",
+        "turn_key": "sha256:" + "a" * 64,
+        "route": "ready_for_host",
+        "session": {
+            "schema_version": "loopx_turn_session_binding_v0",
+            "action": "start_new",
+        },
+        "turn_envelope": {
+            "schema_version": "loopx_turn_envelope_v0",
+            "goal_id": "native-mcp-materialization",
+            "agent_id": "native-mcp-worker",
+            "action": {
+                "selected_todo": {
+                    "todo_id": "todo_native_mcp_materialization",
+                    "text": (
+                        "Call the native MCP tool read_bound_identity. Do not use "
+                        "shell or infer its result. Return validated_progress and "
+                        "include the exact nonce returned by the tool in summary."
+                    ),
+                }
+            },
+        },
+        "result_contract": {
+            "schema_version": "loopx_turn_result_v0",
+            "completed_phases": ["host_execute", "typed_result"],
+        },
+    }
+
+
+def _server(path: Path, nonce: str) -> None:
+    path.write_text(
+        "\n".join(
+            [
+                "from mcp.server.fastmcp import FastMCP",
+                'server = FastMCP("loopx-native-proof")',
+                "@server.tool()",
+                "def read_bound_identity():",
+                '    \"\"\"Read the host-bound identity and proof nonce.\"\"\"',
+                "    return "
+                + repr(
+                    {
+                        "goal_id": "native-mcp-materialization",
+                        "agent_id": "native-mcp-worker",
+                        "nonce": nonce,
+                    }
+                ),
+                'if __name__ == "__main__":',
+                '    server.run(transport="stdio")',
+                "",
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--codex-bin", default="codex")
+    parser.add_argument("--model")
+    parser.add_argument("--reasoning-effort", default="high")
+    parser.add_argument("--timeout-seconds", type=float, default=180.0)
+    args = parser.parse_args()
+    codex_bin = shutil.which(args.codex_bin)
+    if codex_bin is None:
+        parser.error(f"Codex CLI is unavailable: {args.codex_bin}")
+
+    nonce = "native-mcp-" + uuid.uuid4().hex
+    with tempfile.TemporaryDirectory(prefix="loopx-native-mcp-") as raw:
+        root = Path(raw)
+        project = root / "project"
+        project.mkdir()
+        server = root / "server.py"
+        _server(server, nonce)
+        result = run_codex_cli_host(
+            _request(),
+            runtime_root=root / "runtime",
+            project=project,
+            codex_bin=codex_bin,
+            sandbox="read-only",
+            model=args.model,
+            reasoning_effort=args.reasoning_effort,
+            mcp_server={
+                "schema_version": CODEX_STDIO_MCP_SERVER_SCHEMA_VERSION,
+                "name": "loopx_native_proof",
+                "command": [sys.executable, str(server)],
+            },
+            timeout_seconds=args.timeout_seconds,
+        )
+    assert result["result_kind"] == "validated_progress", result
+    assert nonce in result["summary"], result
+    print(
+        json.dumps(
+            {
+                "ok": True,
+                "result_kind": result["result_kind"],
+                "native_tool_proof": True,
+            },
+            indent=2,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

loopx/cli_commands/turn.py

@@ -995,6 +995,7 @@ def run_built_in_host(
                         sandbox=args.codex_sandbox,
                         model=args.codex_model,
                         reasoning_effort=args.codex_reasoning_effort,
+                        mcp_server=args.codex_mcp_server_json,
                         timeout_seconds=max(1.0, args.timeout_seconds - 5.0),
                     )
 

loopx/cli_commands/turn_registration.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import argparse
+import json
 from collections.abc import Callable
 
 from ..control_plane.turn_driver.host_binding import (
@@ -226,6 +227,15 @@ def register_turn_commands(
               "disables the inner sandbox; callers must provide their own isolation. "
               "The setting is passed explicitly for both new and resumed sessions."),
     )
+    run_once.add_argument(
+        "--codex-mcp-server-json",
+        type=json.loads,
+        help=(
+            "Trusted codex_stdio_mcp_server_v0 JSON for one invocation-scoped "
+            "stdio MCP server. The command is passed to fresh and resumed Codex "
+            "sessions without modifying user configuration."
+        ),
+    )
     run_once.add_argument(
         "--dsh-provider",
         help=(

loopx/collaboration_mcp.py

@@ -405,11 +405,36 @@ def _execution_arguments(self, binding: dict, operation_id: str) -> list[str]:
                      "--registry", str(self.registry), "--goal-id", self.goal_id,
                      "--agent-id", self.agent_id, "--execution-config", str(self.config),
                      "--workspace", binding["workspace"], "--operation-id", operation_id]
+        native_tools: list[str] = []
+        if turn_host_arg_option(binding["host_args"], "--host") == "codex-cli":
+            mcp_server = {
+                "schema_version": "codex_stdio_mcp_server_v0",
+                "name": "loopx_delegation",
+                "command": [
+                    sys.executable,
+                    "-m",
+                    "loopx.collaboration_mcp",
+                    "--runtime-root",
+                    str(self.root),
+                    "--registry",
+                    str(self.registry),
+                    "--goal-id",
+                    self.goal_id,
+                    "--agent-id",
+                    binding["agent_id"],
+                    "--workspace",
+                    binding["workspace"],
+                    "--execution-config",
+                    str(self.config),
+                ],
+            }
+            native_tools = ["--codex-mcp-server-json", json.dumps(mcp_server)]
         return ["--execution-mode", "isolated-headless", "--project", binding["workspace"],
                      "--scan-root", binding["workspace"], "--no-global-sync",
                      "--timeout-seconds", str(binding["timeout_seconds"]),
                      "--validation-command-json", json.dumps(validator),
-                     "--validation-failure-kind", "repair_required", *binding["host_args"]]
+                     "--validation-failure-kind", "repair_required", *native_tools,
+                     *binding["host_args"]]
 
     def _execute(self, path: Path, row: dict, binding: dict) -> None:
         request_id = row["identity"]["request_id"]

loopx/control_plane/turn_driver/codex_cli.py

@@ -5,6 +5,7 @@
 import hashlib
 import json
 import os
+import re
 import shutil
 import signal
 import subprocess
@@ -31,6 +32,7 @@
 
 
 CODEX_CLI_SESSION_SCHEMA_VERSION = "loopx_codex_cli_session_v1"
+CODEX_STDIO_MCP_SERVER_SCHEMA_VERSION = "codex_stdio_mcp_server_v0"
 CODEX_CLI_RESULT_KINDS = (
     "validated_progress",
     "repair_required",
@@ -103,12 +105,74 @@
     "provider_capacity": 3,
     "provider_overloaded": 3,
 }
+_MCP_SERVER_NAME = re.compile(r"^[A-Za-z][A-Za-z0-9_-]{0,63}$")
+_MCP_COMMAND_MAX_ITEMS = 64
+_MCP_COMMAND_MAX_BYTES = 16_000
 
 
 def _mapping(value: Any) -> dict[str, Any]:
     return dict(value) if isinstance(value, Mapping) else {}
 
 
+def normalize_codex_stdio_mcp_server(
+    value: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Validate one invocation-scoped stdio MCP server without persisting it.
+
+    The command is trusted host configuration. It is never included in the
+    model prompt or the durable Codex session binding; Codex receives it as
+    per-invocation config for both a fresh session and its resume.
+    """
+
+    if value is None:
+        return None
+    server = dict(value)
+    if server.get("schema_version") != CODEX_STDIO_MCP_SERVER_SCHEMA_VERSION:
+        raise ValueError("unsupported Codex stdio MCP server configuration")
+    if set(server) != {"schema_version", "name", "command"}:
+        raise ValueError("Codex stdio MCP server has unsupported fields")
+    name = server.get("name")
+    if not isinstance(name, str) or _MCP_SERVER_NAME.fullmatch(name) is None:
+        raise ValueError("Codex stdio MCP server name is invalid")
+    command = server.get("command")
+    if (
+        not isinstance(command, list)
+        or not 1 <= len(command) <= _MCP_COMMAND_MAX_ITEMS
+        or any(
+            not isinstance(item, str) or not item or len(item) > 4096
+            for item in command
+        )
+        or len(json.dumps(command, ensure_ascii=False).encode("utf-8"))
+        > _MCP_COMMAND_MAX_BYTES
+    ):
+        raise ValueError("Codex stdio MCP server command is invalid")
+    return {
+        "schema_version": CODEX_STDIO_MCP_SERVER_SCHEMA_VERSION,
+        "name": name,
+        "command": list(command),
+    }
+
+
+def _codex_mcp_config_arguments(
+    value: Mapping[str, Any] | None,
+) -> list[str]:
+    server = normalize_codex_stdio_mcp_server(value)
+    if server is None:
+        return []
+    name = server["name"]
+    command = server["command"]
+    pairs = [
+        f"mcp_servers.{name}.command={json.dumps(command[0])}",
+        f"mcp_servers.{name}.args={json.dumps(command[1:])}",
+        f"mcp_servers.{name}.enabled=true",
+        f"mcp_servers.{name}.required=true",
+        f'mcp_servers.{name}.default_tools_approval_mode="approve"',
+        f"mcp_servers.{name}.startup_timeout_sec=30",
+        f"mcp_servers.{name}.tool_timeout_sec=60",
+    ]
+    return [item for pair in pairs for item in ("-c", pair)]
+
+
 def _lineage(request: Mapping[str, Any]) -> dict[str, str]:
     envelope = _mapping(request.get("turn_envelope"))
     todo = selected_turn_todo(envelope)
@@ -668,6 +732,7 @@ def _codex_command(
     model: str | None,
     reasoning_effort: str | None,
     session_id: str | None,
+    mcp_server: Mapping[str, Any] | None,
 ) -> list[str]:
     if session_id:
         command = [
@@ -709,6 +774,7 @@ def _codex_command(
                 f"model_reasoning_effort={json.dumps(reasoning_effort)}",
             ]
         )
+    command.extend(_codex_mcp_config_arguments(mcp_server))
     if session_id:
         command.append(session_id)
     command.append("-")
@@ -724,6 +790,7 @@ def run_codex_cli_host(
     sandbox: str = "read-only",
     model: str | None = None,
     reasoning_effort: str | None = None,
+    mcp_server: Mapping[str, Any] | None = None,
     timeout_seconds: float = 115.0,
 ) -> dict[str, Any]:
     if request.get("schema_version") != LOOPX_TURN_HOST_REQUEST_SCHEMA_VERSION:
@@ -732,6 +799,7 @@ def run_codex_cli_host(
         raise ValueError(f"Codex CLI sandbox must be one of {CODEX_CLI_SANDBOXES}")
     if reasoning_effort is not None:
         reasoning_effort = require_supported_reasoning_effort(reasoning_effort)
+    mcp_server = normalize_codex_stdio_mcp_server(mcp_server)
     resolved = shutil.which(codex_bin) if os.path.sep not in codex_bin else codex_bin
     if not resolved or not Path(resolved).exists():
         raise ValueError("Codex CLI executable is unavailable")
@@ -774,6 +842,7 @@ def run_codex_cli_host(
             model=model,
             reasoning_effort=reasoning_effort,
             session_id=session_id,
+            mcp_server=mcp_server,
         )
         proc = subprocess.Popen(
             command,

tests/test_delegation_preflight.py

@@ -1,6 +1,7 @@
 """A binding inspection must use the actual Turn without launching or spending."""
 
 import json
+import sys
 
 import pytest
 
@@ -194,6 +195,19 @@ def test_selected_codex_managed_agent_profile_is_projected_exactly(service):
     assert not any(result["effects"].values())
     assert not (root / "host-started").exists()
 
+    binding = runner.binding("analysis", require_active=True)
+    execution = runner._execution_arguments(binding, "native-tool-inspection")
+    encoded = execution[execution.index("--codex-mcp-server-json") + 1]
+    native = json.loads(encoded)
+    assert native["schema_version"] == "codex_stdio_mcp_server_v0"
+    assert native["name"] == "loopx_delegation"
+    command = native["command"]
+    assert command[:3] == [sys.executable, "-m", "loopx.collaboration_mcp"]
+    assert command[command.index("--agent-id") + 1] == "analyst"
+    assert command[command.index("--workspace") + 1] == binding["workspace"]
+    assert command[command.index("--execution-config") + 1] == str(runner.config)
+    assert "lead" not in command
+
 
 def test_preflight_does_not_call_an_invalidated_acceptance_ready(service):
     root, runner = service