diff --git a/datadog_checks_dev/datadog_checks/dev/conditions.py b/datadog_checks_dev/datadog_checks/dev/conditions.py
deleted file mode 100644
index d4c228c259780..0000000000000
--- a/datadog_checks_dev/datadog_checks/dev/conditions.py
+++ /dev/null
@@ -1,229 +0,0 @@
-# (C) Datadog, Inc. 2018-present
-# All rights reserved
-# Licensed under a 3-clause BSD style license (see LICENSE)
-import re
-import socket
-import time
-from contextlib import closing
-from typing import Callable, Dict, List, Tuple, Union  # noqa: F401
-from urllib.request import urlopen
-
-from .errors import RetryError
-from .structures import LazyFunction
-from .subprocess import run_command
-from .utils import file_exists
-
-
-class WaitFor(LazyFunction):
-    def __init__(
-        self,
-        func,  # type: Callable
-        attempts=60,  # type: int
-        wait=1,  # type: int
-        args=(),  # type: Tuple
-        kwargs=None,  # type: Dict
-    ):
-        if kwargs is None:
-            kwargs = {}
-
-        self.func = func
-        self.attempts = attempts
-        self.wait = wait
-        self.args = args
-        self.kwargs = kwargs
-
-    def __call__(self):
-        last_result = None
-        last_error = None
-
-        for _ in range(self.attempts):
-            try:
-                result = self.func(*self.args, **self.kwargs)
-            except Exception as e:
-                last_error = str(e)
-                time.sleep(self.wait)
-                continue
-            else:
-                last_result = result
-
-            if last_result is None or last_result is True:
-                return True
-
-            time.sleep(self.wait)
-        else:
-            raise RetryError(
-                'Result: {}\nError: {}\nFunction: {}, Args: {}, Kwargs: {}\n'.format(
-                    repr(last_result), last_error, self.func.__name__, self.args, self.kwargs
-                )
-            )
-
-
-class CheckEndpoints(LazyFunction):
-    def __init__(
-        self,
-        endpoints,  # type: Union[str, List[str]]
-        timeout=1,  # type: int
-        attempts=60,  # type: int
-        wait=1,  # type: int
-        send_request=None,
-    ):
-        self.endpoints = [endpoints] if isinstance(endpoints, str) else endpoints
-        self.timeout = timeout
-        self.attempts = attempts
-        self.wait = wait
-        self.send_request = urlopen if send_request is None else send_request
-
-    def __call__(self):
-        last_endpoint = ''
-        last_error = None
-
-        for _ in range(self.attempts):
-            for endpoint in self.endpoints:
-                last_endpoint = endpoint
-                try:
-                    response = self.send_request(endpoint, timeout=self.timeout)
-                except Exception as e:
-                    last_error = str(e)
-                    break
-                else:
-                    status_code = response.getcode()
-                    if 400 <= status_code < 600:
-                        last_error = 'status {}'.format(status_code)
-                        break
-            else:
-                break
-
-            time.sleep(self.wait)
-        else:
-            raise RetryError('Endpoint: {}\n' 'Error: {}'.format(last_endpoint, last_error))
-
-
-class CheckCommandOutput(LazyFunction):
-    def __init__(
-        self,
-        command,  # type: Union[str, List[str]]
-        patterns,  # type: Union[str, List[str]]
-        matches=1,  # type: Union[str, int]  #Either 'all' or a number
-        stdout=True,  # type: bool
-        stderr=True,  # type: bool
-        attempts=60,  # type: int
-        wait=1,  # type: int
-    ):
-        """
-        Checks if the provided patterns are present in the output of a command
-
-        :param command: The command to run
-        :param patterns: List of patterns to match
-        :param matches: How many of the provided patterns need to match, it can be a number or "all"
-        :param stdout: Whether to search for the provided patterns in stdout
-        :param stderr: Whether to search for the provided patterns in stderr
-        :param attempts: How many times to try searching for the patterns
-        :param wait: How long, in seconds, to wait between attempts
-        """
-        self.command = command
-        self.stdout = stdout
-        self.stderr = stderr
-        self.attempts = attempts
-        self.wait = wait
-
-        if not (self.stdout or self.stderr):
-            raise ValueError('Must capture stdout, stderr, or both.')
-
-        if isinstance(patterns, str):
-            patterns = [patterns]
-
-        self.patterns = [re.compile(pattern, re.M) if isinstance(pattern, str) else pattern for pattern in patterns]
-
-        if matches == 'all':
-            self.matches = len(patterns)
-        else:
-            self.matches = int(matches)
-
-    def __call__(self):
-        log_output = ''
-        exit_code = 0
-
-        for _ in range(self.attempts):
-            result = run_command(self.command, capture=True)
-            exit_code = result.code
-
-            if self.stdout and self.stderr:
-                log_output = result.stdout + result.stderr
-            elif self.stdout:
-                log_output = result.stdout
-            else:
-                log_output = result.stderr
-
-            matches = 0
-            missing_patterns = set(self.patterns)
-            for pattern in self.patterns:
-                if pattern.search(log_output):
-                    matches += 1
-                    missing_patterns.remove(pattern)
-
-            if matches >= self.matches:
-                return matches
-
-            time.sleep(self.wait)
-        else:
-            patterns = '\t- '.join([''] + [str(p) for p in self.patterns])
-            missing_patterns = '\t- '.join([''] + [str(p) for p in missing_patterns])
-            raise RetryError(
-                u'Command: {}\nFailed to match `{}` of the patterns.\n'
-                u'Provided patterns: {}\n'
-                u'Missing patterns: {}\n'
-                u'Exit code: {}\n'
-                u'Captured Output: {}'.format(
-                    self.command, self.matches, patterns, missing_patterns, exit_code, log_output
-                )
-            )
-
-
-class CheckDockerLogs(CheckCommandOutput):
-    def __init__(
-        self,
-        identifier,  # type: str
-        patterns,  # type: Union[str, List[str]]
-        matches=1,  # type: Union[str, int]
-        stdout=True,  # type: bool
-        stderr=True,  # type: bool
-        attempts=60,  # type: int
-        wait=1,  # type: int
-    ):
-        """
-        Checks if the provided patterns are present in docker logs
-
-        :param identifier: The docker image identifier
-        :param patterns: List of patterns to match
-        :param matches: How many of the provided patterns need to match, it can be a number or "all"
-        :param stdout: Whether to search for the provided patterns in stdout
-        :param stderr: Whether to search for the provided patterns in stderr
-        :param attempts: How many times to try searching for the patterns
-        :param wait: How long, in seconds, to wait between attempts
-        """
-        if file_exists(identifier):
-            command = ['docker', 'compose', '-f', identifier, 'logs']
-        else:
-            command = ['docker', 'logs', identifier]
-
-        super(CheckDockerLogs, self).__init__(
-            command, patterns, matches=matches, stdout=stdout, stderr=stderr, attempts=attempts, wait=wait
-        )
-        self.identifier = identifier
-
-
-class WaitForPortListening(WaitFor):
-    """Wait until a server is available on `host:port`."""
-
-    def __init__(
-        self,
-        host,  # type: str
-        port,  # type: int
-        attempts=60,  # type: int
-        wait=1,  # type: int
-    ):
-        super(WaitForPortListening, self).__init__(self.connect, attempts, wait, args=(host, port))
-
-    def connect(self, host, port):
-        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
-            s.connect((host, port))
diff --git a/datadog_checks_dev/datadog_checks/dev/tooling/e2e/__init__.py b/datadog_checks_dev/datadog_checks/dev/tooling/e2e/__init__.py
index 00e4755791fbd..3a97e8c2be290 100644
--- a/datadog_checks_dev/datadog_checks/dev/tooling/e2e/__init__.py
+++ b/datadog_checks_dev/datadog_checks/dev/tooling/e2e/__init__.py
@@ -5,4 +5,4 @@
 from .core import create_interface, derive_interface
 from .run import start_environment, stop_environment
 
-E2E_SUPPORTED_TYPES = {'docker', 'local'}
+E2E_SUPPORTED_TYPES = {'docker', 'local', 'vagrant'}
diff --git a/ddev/src/ddev/e2e/agent/Vagrantfile.template b/ddev/src/ddev/e2e/agent/Vagrantfile.template
new file mode 100644
index 0000000000000..c100fbb4d736b
--- /dev/null
+++ b/ddev/src/ddev/e2e/agent/Vagrantfile.template
@@ -0,0 +1,38 @@
+# -*- mode: ruby -*-
+# vi: set ft=ruby :
+
+$set_environment_variables = <<SCRIPT
+tee "/etc/profile.d/myvars.sh" > "/dev/null" <<EOF
+
+export DD_API_KEY="{{ dd_api_key }}"
+export LOCAL_IP=$(hostname -I | cut -d ' ' -f 1)
+export DD_SITE="datadoghq.com"
+export DD_HOSTNAME=$(hostname)
+{{ exported_env_vars_str }}
+
+EOF
+SCRIPT
+
+
+Vagrant.configure("2") do |config|
+  config.vm.box = "{{ vagrant_box }}"
+  config.vm.box_version = "1.1"
+
+  {{ synced_folders_str | safe }}
+  config.vm.network "private_network", ip: "172.30.1.5"
+
+  config.vm.define "{{ vm_hostname }}" do |node|
+    node.vm.hostname = "{{ vm_hostname }}"
+
+    node.vm.provision "shell", inline: $set_environment_variables, run: "always"
+
+    node.vm.provision "shell", inline: <<-SHELL, run: "always"
+      apt update
+      echo "DD_ENV_VARS: {{ agent_install_env_vars_str }}" # TODO: remove this, debugging purposes only
+      {{ agent_install_env_vars_str }} bash -c "$(curl -L https://install.datadoghq.com/scripts/install_script_agent7.sh)"
+      service datadog-agent start && echo "Agent started successfully"
+      echo "VM {{ vm_hostname }} is ready"
+    SHELL
+  end
+
+end 
\ No newline at end of file
diff --git a/ddev/src/ddev/e2e/agent/__init__.py b/ddev/src/ddev/e2e/agent/__init__.py
index 531b78f4b6347..f0dcfe5a29556 100644
--- a/ddev/src/ddev/e2e/agent/__init__.py
+++ b/ddev/src/ddev/e2e/agent/__init__.py
@@ -10,9 +10,14 @@
 
 
 def get_agent_interface(agent_type: str) -> type[AgentInterface]:
-    if agent_type == 'docker':
+    if agent_type == "docker":
         from ddev.e2e.agent.docker import DockerAgent
 
         return DockerAgent
 
-    raise NotImplementedError(f'Unsupported Agent type: {agent_type}')
+    if agent_type == "vagrant":
+        from ddev.e2e.agent.vagrant import VagrantAgent
+
+        return VagrantAgent
+
+    raise NotImplementedError(f"Unsupported Agent type: {agent_type}")
diff --git a/ddev/src/ddev/e2e/agent/vagrant.py b/ddev/src/ddev/e2e/agent/vagrant.py
new file mode 100644
index 0000000000000..3bf11377975f7
--- /dev/null
+++ b/ddev/src/ddev/e2e/agent/vagrant.py
@@ -0,0 +1,480 @@
+# (C) Datadog, Inc. 2025-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+import os
+import shlex
+import sys
+import hashlib
+
+from contextlib import AbstractContextManager, contextmanager, nullcontext, closing
+from functools import cache, cached_property, partial
+from jinja2 import Template
+from typing import TYPE_CHECKING, Callable, Type
+
+import shutil
+
+from ddev.e2e.agent.interface import AgentInterface
+from ddev.utils.structures import EnvVars
+from ddev.e2e.agent.constants import AgentEnvVars
+from ddev.utils.fs import Path
+
+
+if TYPE_CHECKING:
+    import subprocess
+    from ddev.platform import Platform
+
+
+
+@contextmanager
+def disable_integration_before_install(config_file: Path):
+    """
+    Disable integration by renaming the config to "conf.yaml.example".
+
+    As we exit the context manager we rename it back to "conf.yaml" to re-enable the integration.
+    This assumes the config_file is accessible from the host and is part of a synced folder.
+    """
+    old_name = config_file.name
+    new_config_file = config_file.rename(config_file.parent / (config_file.name + ".example"))
+    try:
+        yield
+    finally:
+        # Ensure the file is renamed back even if errors occur within the context
+        if new_config_file.is_file():  # Check if it was successfully renamed
+            new_config_file.rename(config_file.parent / old_name)
+        elif not config_file.exists() and (config_file.parent / old_name).exists():
+            # If the original was somehow restored or yield failed before new_config_file was used.
+            pass  # Already in desired state or original file does not exist to be renamed.
+
+
+
+class VagrantAgent(AgentInterface):
+    def __init__(
+        self,
+        platform: Platform,
+        integration: Path,
+        env: str,
+        metadata: dict,
+        config_file: Path,
+    ):
+        super().__init__(platform, integration, env, metadata, config_file)
+        self._initialize_vagrant(overwrite=False)
+
+    def _initialize_vagrant(self, overwrite: bool = False, **kwargs):
+        # Initialize and create the directory for Vagrant files specific to this VM
+        home_dir = Path.home()
+        self._temp_vagrant_dir = home_dir / ".ddev" / "vagrant" / self._vm_name
+        self._temp_vagrant_dir.mkdir(parents=True, exist_ok=True)
+
+        # Generate Vagrantfile if it doesn't exist
+        vagrantfile_path = self._temp_vagrant_dir / "Vagrantfile"
+        if not vagrantfile_path.exists() or overwrite:
+            old_file_hash = None
+            if overwrite:
+                old_file_hash = hashlib.sha256(vagrantfile_path.read_text().encode()).hexdigest()
+                print(f"Overwriting Vagrantfile found at '{vagrantfile_path}'.")
+                vagrantfile_path.unlink()
+                print(f"Vagrantfile deleted at {vagrantfile_path}")
+            else:
+                print(f"Vagrantfile not found at {vagrantfile_path}, generating new one.")
+    
+            
+            vagrantfile_content = self._generate_vagrantfile_content(**kwargs)
+            vagrantfile_path.write_text(vagrantfile_content)
+            print(f"Vagrantfile generated at {vagrantfile_path}")
+            new_file_hash = hashlib.sha256(vagrantfile_content.encode()).hexdigest()
+            if old_file_hash and old_file_hash != new_file_hash:
+                self.metadata["vagrant_provision"] = True
+        else:
+            print(f"Using existing Vagrantfile at {vagrantfile_path}")
+
+        # Set VAGRANT_CWD to self.temp_vagrant_dir
+        os.environ["VAGRANT_CWD"] = str(self._temp_vagrant_dir)
+        print(f"Vagrant working directory set to: {self._temp_vagrant_dir}")
+    
+    def _get_vagrantfile_template(self) -> Template:
+        template_path = Path(__file__).parent / "Vagrantfile.template"
+        if not template_path.is_file():
+            raise FileNotFoundError(f"Vagrantfile template not found at {template_path}")
+        return Template(template_path.read_text())
+
+    def _generate_vagrantfile_content(self, **kwargs) -> str:
+        agent_install_env_vars = kwargs.get("agent_install_env_vars", {})
+        synced_folders = kwargs.get("synced_folders", [])
+        exported_env_vars = kwargs.get("exported_env_vars", {})
+
+        agent_install_env_vars_str = " ".join([f'{key}="{value}"' for key, value in agent_install_env_vars.items()]) if agent_install_env_vars else ""
+        exported_env_vars_str = "\n".join([f'export {key}="{value}"' for key, value in exported_env_vars.items()]) if exported_env_vars else ""
+
+        vm_hostname = self._vm_name  # Already sanitized and unique
+        vagrant_box = self.metadata.get("vagrant_box", "net9/ubuntu-24.04-arm64")  # Default box
+
+
+        synced_folders_str = ""
+        for volume in synced_folders:
+            path, target = volume.split(":")
+            synced_folders_str += f'config.vm.synced_folder "{path}", "{target}"\n'
+        
+        template = self._get_vagrantfile_template()
+
+        return template.render(
+            dd_api_key=os.environ.get("DD_API_KEY", ""),
+            exported_env_vars_str=exported_env_vars_str,
+            vagrant_box=vagrant_box,
+            synced_folders_str=synced_folders_str,
+            vm_hostname=vm_hostname,
+            agent_install_env_vars_str=agent_install_env_vars_str,
+        )
+
+    @cached_property
+    def _isatty(self) -> bool:
+        isatty: Callable[[], bool] | None = getattr(sys.stdout, "isatty", None)
+        if isatty is not None:
+            try:
+                return isatty()
+            except ValueError:  # Raised if sys.stdout is not a tty (e.g. in a pipe)
+                pass
+        return False
+
+    @cached_property
+    def _vm_name(self) -> str:
+        # vm name can only conntain letters, numbers, hyphens and dots.
+        return f"dd_vagrant_{super().get_id()}".replace("_", "-")
+
+    @cached_property
+    def _is_windows_vm(self) -> bool:
+        return self.metadata.get("vagrant_guest_os", "linux").lower() == "windows"
+
+    @cached_property
+    def _package_mount_dir(self) -> str:
+        # Default path INSIDE the VM where host packages are synced/mounted.
+        # Assumes a synced folder like `/home` or `C:\vagrant` on the guest.
+        base_synced_dir = "C:\\vagrant" if self._is_windows_vm else "/home"
+        return os.path.join(base_synced_dir, "packages/").replace(
+            "\\", "\\\\"
+        )  # Ensure correct path sep for os.join and escape for f-strings
+
+    @cached_property
+    def _config_mount_dir(self) -> str:
+        # Path INSIDE the VM where agent configs are expected.
+        name = self.integration.name
+        if self._is_windows_vm:
+            return f"C:\\ProgramData\\Datadog\\conf.d\\{name}.d"
+        else:
+            return f"/etc/datadog-agent/conf.d/{name}.d"
+
+    @cached_property
+    def _python_path(self) -> str:
+        # Path to python executable INSIDE the VM.
+        py_major = self.python_version[0]
+        if self._is_windows_vm:
+            return f"C:\\Program Files\\Datadog\\Datadog Agent\\embedded{py_major}\\python.exe"
+        else:
+            return f"/opt/datadog-agent/embedded/bin/python{py_major}"
+
+    def _format_command(self, guest_command_parts: list[str], interactive: bool = False) -> list[str]:
+        # Returns the host-side command to execute something in the guest.
+        if interactive:
+            return ["vagrant", "ssh", self._vm_name]
+
+        # Prepare the command string to be executed inside the VM via ssh -c "..."
+        inner_cmd_list = []
+        inner_cmd_list.extend(guest_command_parts)
+
+        # Handle sudo for non-Windows guests if metadata suggests it's needed for the command.
+        # This is a basic approach. More complex sudo needs might require specific command metadata.
+        if not self._is_windows_vm and self.metadata.get("vagrant_command_needs_sudo", True):
+            if not inner_cmd_list or inner_cmd_list[0] != "sudo":
+                inner_cmd_list.insert(0, "sudo -E")
+
+        inner_cmd_str = " ".join(shlex.quote(part) for part in inner_cmd_list)
+        host_command = ["vagrant", "ssh", self._vm_name, "-c", inner_cmd_str.replace("'", "")]
+        return host_command
+
+    def _captured_process(self, host_command: list[str], **kwargs) -> subprocess.CompletedProcess:
+        kwargs.setdefault("check", False)
+
+        return self.platform.run_command(
+            host_command,
+            stdout=self.platform.modules.subprocess.PIPE,
+            stderr=self.platform.modules.subprocess.PIPE,
+            **kwargs,
+        )
+
+    def _run_command(self, host_command: list[str], **kwargs) -> subprocess.CompletedProcess:
+        # `check=True` by default if not specified by caller.
+        kwargs.setdefault("check", True)
+        kwargs.setdefault("env", os.environ)
+
+        return self.platform.run_command(host_command, **kwargs)
+
+    def _show_logs(self) -> None:
+        print(f"Attempting to fetch agent logs from VM: {self._vm_name}...")
+        log_file_path_guest = (
+            "C:\\ProgramData\\Datadog\\logs\\agent.log" if self._is_windows_vm else "/var/log/datadog/agent.log"
+        )
+
+        guest_cmd_parts = ["type" if self._is_windows_vm else "cat", log_file_path_guest]
+
+        # Reading agent logs on Linux might require sudo.
+        if not self._is_windows_vm and self.metadata.get("vagrant_logs_require_sudo", True):
+            if guest_cmd_parts[0] != "sudo":  # Avoid double sudo if 'cat' was replaced by 'sudo cat'
+                guest_cmd_parts.insert(0, "sudo")
+
+        host_cmd_to_fetch_logs = self._format_command(guest_cmd_parts)
+
+        process = self._captured_process(host_cmd_to_fetch_logs)
+        stdout = process.stdout.decode("utf-8", errors="replace").strip() if process.stdout else ""
+        stderr = process.stderr.decode("utf-8", errors="replace").strip() if process.stderr else ""
+
+        if process.returncode == 0 and stdout:
+            print(f"Agent logs from VM ({self._vm_name}):\n{stdout}")
+        elif process.returncode == 0 and not stdout:
+            print(f"Successfully fetched agent logs from VM ({self._vm_name}), but logs are empty.")
+        else:
+            raise Exception(
+                f"Failed to fetch agent logs from VM ({self._vm_name}). RC: {process.returncode}\n"
+                f"Stdout:\n{stdout}\n"
+                f"Stderr:\n{stderr}"
+            )
+
+    def get_id(self) -> str:
+        return self._vm_name
+
+    def start(self, *, agent_build: str, local_packages: dict[Path, str], env_vars: dict[str, str]) -> None:
+        # prepare agent install scripts environment variables
+        agent_install_env_vars = {}
+
+        if agent_build:
+            pipeline_id, major_version, arch = agent_build.split("-")
+            agent_install_env_vars["TESTING_APT_URL"] = "s3.amazonaws.com/apttesting.datad0g.com"
+            agent_install_env_vars["TESTING_APT_REPO_VERSION"] = f"pipeline-{pipeline_id}-a{major_version}-{arch} {major_version}"
+            agent_install_env_vars["TESTING_YUM_URL"] = "s3.amazonaws.com/yumtesting.datad0g.com"
+            agent_install_env_vars["TESTING_YUM_VERSION_PATH"] = f"testing/pipeline-{pipeline_id}-a{major_version}/{major_version}"
+
+
+        # prepare synced_folders
+        synced_folders = []    
+        synced_folders.extend(self.metadata.get('vagrant_synced_folders', []))
+        if self.config_file.is_file():
+            synced_folders.append(f'{self.config_file.parent}:{self._config_mount_dir}')
+            if local_packages:
+                ensure_local_pkg = partial(disable_integration_before_install, self.config_file)
+
+        # It is safe to assume that the directory name is unique across all repos
+        for local_package in local_packages:
+            synced_folders.append(f'{local_package}:{self._package_mount_dir}/{local_package.name}')
+
+        # Host environment variables for `vagrant up` command itself
+        host_operation_env_vars = EnvVars(os.environ)
+        host_operation_env_vars.update(env_vars)  # User-provided can override system for the vagrant command
+
+        # Prepare environment variables intended for the Agent process inside the VM.
+        exported_env_vars = {}
+        exported_env_vars.update(self.metadata.get('env', {}))
+        if AgentEnvVars.API_KEY not in env_vars:
+            exported_env_vars[AgentEnvVars.API_KEY] = "a" * 32
+
+        # By default, the hostname is the VM hostname (set in VagrantFile with DD_HOSTNAME)
+        if self.metadata.get("dd_hostname"):
+            exported_env_vars[AgentEnvVars.HOSTNAME] = self.metadata.get("dd_hostname")
+
+        exported_env_vars[AgentEnvVars.APM_ENABLED] = self.metadata.get("dd_apm_enabled", "false")
+        exported_env_vars[AgentEnvVars.TELEMETRY_ENABLED] = self.metadata.get("dd_telemetry_enabled", "true")
+        exported_env_vars[AgentEnvVars.EXPVAR_PORT] = self.metadata.get("dd_expvar_port", "5000")
+        if (proxy_data := self.metadata.get("proxy")) is not None:
+            if (http_proxy := proxy_data.get("http")) is not None:
+                exported_env_vars[AgentEnvVars.PROXY_HTTP] = http_proxy
+            if (https_proxy := proxy_data.get("https")) is not None:
+                exported_env_vars[AgentEnvVars.PROXY_HTTPS] = https_proxy
+
+        ensure_local_pkg: Type[AbstractContextManager] | Callable[[], AbstractContextManager] = nullcontext
+        if self.config_file.is_file() and local_packages:
+            print(f"Local packages to install; temporarily disabling integration config: {self.config_file}")
+            ensure_local_pkg = partial(disable_integration_before_install, self.config_file)
+
+
+        # We can now generate the Vagrantfile content
+        self._initialize_vagrant(overwrite=True, exported_env_vars=exported_env_vars, agent_install_env_vars=agent_install_env_vars, synced_folders=synced_folders)
+
+        print(f"Starting Vagrant environment for VM: {self._vm_name} with agent build: '{agent_build}'")
+
+        up_command_host = ["vagrant", "up", self._vm_name]
+        if self.metadata.get("vagrant_provision", False):
+            print(f"VagrantFile changed, provisioning Vagrant VM: {self._vm_name}")
+            up_command_host.append("--provision")
+
+        # Agent installation script via Python is removed. Assumed to be in Vagrantfile provisioning.
+        # Custom start_commands from metadata are still run.
+        start_commands = []
+        if custom_start_cmds := self.metadata.get("start_commands"):
+            if isinstance(custom_start_cmds, list):
+                start_commands.extend(custom_start_cmds)
+            else:
+                start_commands.append(str(custom_start_cmds))
+
+        post_install_guest_commands = self.metadata.get("post_install_commands", [])
+
+        with ensure_local_pkg():
+            self._initialize(
+                up_command_host,
+                local_packages,
+                start_commands,
+                post_install_guest_commands,
+                host_operation_env_vars,
+            )
+
+        if local_packages or start_commands or post_install_guest_commands:
+            print("Local packages installed or custom start/post-install commands run. Restarting agent service.")
+            self.restart_agent_service()
+        else:
+            print("No local packages or custom start/post-install commands. Agent service restart skipped.")
+
+    def _initialize(
+        self,
+        up_command_host: list[str],
+        local_packages: dict[Path, str],
+        start_guest_commands: list[str],  # Renamed from start_commands for clarity here
+        post_install_guest_commands: list[str],
+        host_operation_env_vars: EnvVars,
+    ):
+        print(
+            f"Bringing up Vagrant VM: {self._vm_name} from {self._temp_vagrant_dir} with command: {' '.join(up_command_host)}"
+        )
+        # _captured_process will now automatically use self._temp_vagrant_dir as cwd for vagrant commands
+        process = self._captured_process(up_command_host, env=host_operation_env_vars)
+        stdout = process.stdout.decode("utf-8", errors="replace") if process.stdout else ""
+        stderr = process.stderr.decode("utf-8", errors="replace") if process.stderr else ""
+        if process.returncode:
+            raise RuntimeError(
+                f"Failed to start Vagrant VM `{self._vm_name}` (RC: {process.returncode}).\n"
+                f"Stdout:\n{stdout}\nStderr:\n{stderr}"
+            )
+        print(f"Vagrant VM `{self._vm_name}` started successfully.\n{stdout}")
+
+        if start_guest_commands:
+            self._run_commands(start_guest_commands, "start")
+
+        # Install local packages
+        if local_packages:
+            print(f"Installing local packages in VM `{self._vm_name}`...")
+            base_pip_command = [self._python_path, '-m', 'pip', 'install', '--disable-pip-version-check', '-e']
+            for local_package, features in local_packages.items():
+                package_mount = f'{self._package_mount_dir}{local_package.name}{features}'
+                formatted_cmd = self._format_command([*base_pip_command, package_mount])
+                process = self._captured_process(formatted_cmd)
+                stdout = process.stdout.decode("utf-8", errors="replace") if process.stdout else ""
+                stderr = process.stderr.decode("utf-8", errors="replace") if process.stderr else ""
+                if process.returncode:
+                    raise RuntimeError(
+                        f"Failed to run command `{' '.join(formatted_cmd)}` in VM `{self._vm_name}` (RC: {process.returncode}).\n"
+                        f"Stdout:\n{stdout}\nStderr:\n{stderr}"
+                    )
+                print(f"Successfully installed  localge package `{local_package.name}` in Agent Vagrant VM `{self._vm_name}`.")
+
+        # Execute post_install_commands (guest commands)
+        if post_install_guest_commands:
+            self._run_commands(start_guest_commands, "post-install")
+
+    def _run_commands(self, commands: list[str], command_type: str) -> None:
+        print(f"Running {command_type} commands in VM `{self._vm_name}`...")
+        for guest_cmd_str in commands:
+            cmd_parts_guest = self.platform.modules.shlex.split(guest_cmd_str)
+            formatted_host_cmd = self._format_command(cmd_parts_guest)
+            process = self._captured_process(formatted_host_cmd)
+            stdout = process.stdout.decode("utf-8", errors="replace") if process.stdout else ""
+            stderr = process.stderr.decode("utf-8", errors="replace") if process.stderr else ""
+            if process.returncode:
+                raise RuntimeError(
+                    f"Failed to run {command_type} command `{' '.join(cmd_parts_guest)}` in VM `{self._vm_name}` (RC: {process.returncode}).\n"
+                    f"Stdout:\n{stdout}\nStderr:\n{stderr}"
+                )
+            print(f"Successfully ran {command_type} command `{' '.join(cmd_parts_guest)}`.\n{stdout}")
+
+    def stop(self) -> None:
+        print(f"Stopping Vagrant VM: {self._vm_name}...")
+        stop_guest_commands = self.metadata.get("stop_commands", [])
+        if stop_guest_commands:
+            self._run_commands(stop_guest_commands, "stop")
+        
+        # Halt the VM
+        halt_cmd_host = ["vagrant", "halt", self._vm_name]
+        print(f"Halting VM: {self._vm_name} with command: {' '.join(halt_cmd_host)}")
+        process_halt = self._captured_process(halt_cmd_host)  # cwd is no longer set by _captured_process
+        if process_halt.returncode:
+            print(
+                f"Failed to halt Vagrant VM `{self._vm_name}` (RC: {process_halt.returncode}). "
+                f"Stderr: {process_halt.stderr.decode('utf-8', errors='replace') if process_halt.stderr else 'N/A'}"
+            )
+        else:
+            print(f"VM {self._vm_name} halted.")
+
+        # Destroy the VM
+        destroy_cmd_host = ["vagrant", "destroy", self._vm_name, "--force"]
+        print(f"Destroying VM: {self._vm_name} with command: {' '.join(destroy_cmd_host)}")
+        process_destroy = self._captured_process(destroy_cmd_host)
+        if process_destroy.returncode:
+            raise RuntimeError(
+                f"Failed to destroy Vagrant VM `{self._vm_name}` (RC: {process_destroy.returncode}).\n"
+                f"Stdout: {process_destroy.stdout.decode('utf-8', errors='replace') if process_destroy.stdout else 'N/A'}\n"
+                f"Stderr: {process_destroy.stderr.decode('utf-8', errors='replace') if process_destroy.stderr else 'N/A'}"
+            )
+        print(f"VM {self._vm_name} destroyed.")
+
+        # delete the temp vagrant dir
+        shutil.rmtree(self._temp_vagrant_dir)
+        print(f"Vagrant working directory deleted: {self._temp_vagrant_dir}")
+
+    def restart(self) -> None:
+        print(f"Restarting (reloading) Vagrant VM: {self._vm_name}...")
+        reload_cmd_host = ["vagrant", "reload", self._vm_name]
+        if self.metadata.get("vagrant_provision_on_reload", True):  # Default to reprovisioning on reload
+            reload_cmd_host.append("--provision")
+
+        process = self._captured_process(reload_cmd_host)
+        stdout = process.stdout.decode("utf-8", errors="replace") if process.stdout else ""
+        stderr = process.stderr.decode("utf-8", errors="replace") if process.stderr else ""
+        if process.returncode:
+            raise RuntimeError(
+                f"Failed to restart (reload) Vagrant VM `{self._vm_name}` (RC: {process.returncode}).\n"
+                f"Stdout:\n{stdout}\nStderr:\n{stderr}"
+            )
+        print(f"Vagrant VM `{self._vm_name}` restarted successfully.\n{stdout}")
+
+    def restart_agent_service(self) -> None:
+        # Restarts the Datadog Agent service *inside* the VM
+        print(f"Restarting Datadog Agent service in VM: {self._vm_name}...")
+        if self._is_windows_vm:
+            # Example for Windows: restart using sc. Ensure agent service name is correct.
+            agent_service_name = self.metadata.get("vagrant_windows_agent_service_name", "DatadogAgent")
+            # Stop and then start, as 'restart' is not always available/reliable with sc for all services
+            guest_cmds = [["sc", "stop", agent_service_name], ["sc", "start", agent_service_name]]
+        else:
+            # Example for Linux: systemd. Allow override via metadata.
+            restart_cmd_str = self.metadata.get(
+                "vagrant_linux_agent_restart_command", "sudo systemctl restart datadog-agent.service"
+            )
+            guest_cmds = [restart_cmd_str]
+        
+        self._run_commands(guest_cmds, "restart-agent-service")
+        print("Datadog Agent service restart sequence completed.")
+
+    def invoke(self, args: list[str]) -> None:
+        # Runs an 'agent <command>' inside the VM
+        agent_bin = "/opt/datadog-agent/bin/agent/agent" if not self._is_windows_vm else "C:\\Program Files\\Datadog\\Datadog Agent\\bin\\agent.exe"
+
+        guest_cmd_parts = [agent_bin] + args
+        host_cmd = self._format_command(guest_cmd_parts)
+
+        print(f"Invoking agent command in VM `{self._vm_name}`: {' '.join(host_cmd)}")
+        # Using _run_command directly to stream output and check=True by default
+        self._run_command(host_cmd)
+
+    def enter_shell(self) -> None:
+        print(f"Entering interactive shell for VM: {self._vm_name}...")
+        # _format_command with interactive=True gives ['vagrant', 'ssh', self._vm_name]
+        host_cmd = self._format_command([], interactive=True)
+        # For interactive shells, check=True might exit if shell exits non-zero,
+        # but usually, it's what's desired.
+        self._run_command(host_cmd, check=True)
\ No newline at end of file
diff --git a/docs/developer/ddev/plugins.md b/docs/developer/ddev/plugins.md
index 607d29e19438a..470dc3ad8ec51 100644
--- a/docs/developer/ddev/plugins.md
+++ b/docs/developer/ddev/plugins.md
@@ -137,7 +137,7 @@ is desired. This fixture is responsible for starting and stopping environments a
 
 #### Metadata
 
-- `env_type` - This is the type of interface that will be used to interact with the Agent. Currently, we support `docker` (default) and `local`.
+- `env_type` - This is the type of interface that will be used to interact with the Agent. Currently, we support `docker` (default), `vagrant` and `local`.
 - `env_vars` - A `dict` of environment variables and their values that will be present when starting the Agent.
 - `docker_volumes` - A `list` of `str` representing [Docker volume mounts][docker-volume-docs] if `env_type` is `docker` e.g. `/local/path:/agent/container/path:ro`.
 - `docker_platform` - The container architecture to use if `env_type` is `docker`. Currently, we support `linux` (default) and `windows`.
diff --git a/glusterfs/tests/common.py b/glusterfs/tests/common.py
index aec7349038114..f3cd883206ac7 100644
--- a/glusterfs/tests/common.py
+++ b/glusterfs/tests/common.py
@@ -7,7 +7,7 @@
 
 CHECK = "glusterfs"
 INSTANCE = {"use_sudo": False}
-E2E_INIT_CONFIG = {"gstatus_path": "/usr/local/bin/gstatus"}
+E2E_INIT_CONFIG = {}
 CONFIG = {"init_config": E2E_INIT_CONFIG, "instances": [INSTANCE]}
 
 EXPECTED_METRICS = [
diff --git a/glusterfs/tests/conftest.py b/glusterfs/tests/conftest.py
index 4e61b15c0b36c..9b22686777519 100644
--- a/glusterfs/tests/conftest.py
+++ b/glusterfs/tests/conftest.py
@@ -7,32 +7,34 @@
 
 import pytest
 
-from datadog_checks.base.utils.common import get_docker_hostname
-from datadog_checks.dev import WaitFor, docker_run, run_command
-
 from .common import CONFIG, INSTANCE
 
 HERE = os.path.dirname(os.path.abspath(__file__))
-HOST = get_docker_hostname()
 
 E2E_METADATA = {
+    "agent_type": "vagrant",
+    "vagrant_guest_os": "linux",
     "start_commands": [
-        "apt-get update",
-        'apt-get install -o Dpkg::Options::="--force-confdef" -o Dpkg::Options::="--force-confold" -y docker.io',
+        "sudo apt update",
+        "sudo apt install -y software-properties-common",
+        "sudo add-apt-repository -y ppa:gluster/glusterfs-7",
+        "sudo apt install -y glusterfs-server glusterfs-client",
+        "sudo systemctl start glusterd",
+        "sudo systemctl enable glusterd",
+        "sudo mkdir -p /gluster/brick1",
+        "sudo gluster volume create volume1 172.30.1.5:/gluster/brick1 force",
+        "sudo gluster volume start volume1",
+        "sudo mkdir -p /gluster/brick2",
+        "sudo gluster volume create volume2 172.30.1.5:/gluster/brick2 force",
+        "sudo gluster volume start volume2",
     ],
-    "docker_volumes": ["/var/run/docker.sock:/var/run/docker.sock"],
 }
 
 
 @pytest.fixture(scope="session")
 def dd_environment():
-    compose_file = os.path.join(HERE, "docker", "docker-compose.yaml")
-    with docker_run(
-        compose_file=compose_file,
-        conditions=[WaitFor(create_volume)],
-        down=delete_volume,
-    ):
-        yield CONFIG, E2E_METADATA
+    vm_config = copy.deepcopy(CONFIG)
+    yield vm_config, E2E_METADATA
 
 
 @pytest.fixture
@@ -53,33 +55,3 @@ def mock_gstatus_data():
 
     with mock.patch("datadog_checks.glusterfs.check.GlusterfsCheck.get_gstatus_data", return_value=data):
         yield
-
-
-def create_volume():
-    # The image for some reason doesn't actually start glusterd. So the below two lines will start it manually.
-    # Tried also adding the command to the Dockerfile, but that didn't work either. So leaving this here for now.
-    run_command(
-        "docker exec gluster-node-2 /usr/sbin/glusterd -p /var/run/glusterd.pid --log-level INFO",
-        capture=True,
-        check=False,
-    )
-    run_command(
-        "docker exec gluster-node-1 /usr/sbin/glusterd -p /var/run/glusterd.pid --log-level INFO",
-        capture=True,
-        check=False,
-    )
-
-    commands = [
-        "node-2 mkdir -p /export-test",
-        "node-1 mkdir -p /export-test",
-        "node-1 gluster peer probe gluster-node-2",
-        "node-1 gluster volume create gv0 replica 2 gluster-node-1:/export-test gluster-node-2:/export-test force",
-        "node-1 gluster volume start gv0",
-    ]
-
-    for command in commands:
-        run_command(f"docker exec gluster-{command}", capture=True, check=True)
-
-
-def delete_volume():
-    run_command("docker exec gluster-node-1 gluster volume delete gv0", capture=True, check=True)