diff --git a/datadog_checks_dev/CHANGELOG.md b/datadog_checks_dev/CHANGELOG.md
index 53ffc75da23957..519457b6c055cd 100644
--- a/datadog_checks_dev/CHANGELOG.md
+++ b/datadog_checks_dev/CHANGELOG.md
@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+***Added***:
+
+* Migrate E2E features ([#15931](https://github.com/DataDog/integrations-core/pull/15931))
+
 ## 25.1.2 / 2023-09-26
 
 ***Fixed***:
diff --git a/datadog_checks_dev/datadog_checks/dev/_env.py b/datadog_checks_dev/datadog_checks/dev/_env.py
index 87b9c3059a7255..1651eb1777a87b 100644
--- a/datadog_checks_dev/datadog_checks/dev/_env.py
+++ b/datadog_checks_dev/datadog_checks/dev/_env.py
@@ -19,7 +19,7 @@
 E2E_ENV_VAR_PREFIX = '{}_ENV_'.format(E2E_PREFIX)
 E2E_SET_UP = '{}_UP'.format(E2E_PREFIX)
 E2E_TEAR_DOWN = '{}_DOWN'.format(E2E_PREFIX)
-E2E_PARENT_PYTHON = '{}_PYTHON_PATH'.format(E2E_PREFIX)
+E2E_PARENT_PYTHON = '{}_PARENT_PYTHON'.format(E2E_PREFIX)
 E2E_RESULT_FILE = '{}_RESULT_FILE'.format(E2E_PREFIX)
 
 E2E_FIXTURE_NAME = 'dd_environment'
diff --git a/datadog_checks_dev/datadog_checks/dev/plugin/pytest.py b/datadog_checks_dev/datadog_checks/dev/plugin/pytest.py
index 1567a3f81e8ab0..49b71d7193ab0b 100644
--- a/datadog_checks_dev/datadog_checks/dev/plugin/pytest.py
+++ b/datadog_checks_dev/datadog_checks/dev/plugin/pytest.py
@@ -9,10 +9,11 @@
 import warnings
 from base64 import urlsafe_b64encode
 from collections import namedtuple  # Not using dataclasses for Py2 compatibility
+from io import open
 from typing import Dict, List, Optional, Tuple  # noqa: F401
 
 import pytest
-from six import PY2
+from six import PY2, ensure_text
 
 from .._env import (
     E2E_FIXTURE_NAME,
@@ -139,7 +140,7 @@ def dd_environment_runner(request):
         # Exit testing and pass data back up to command
         if E2E_RESULT_FILE in os.environ:
             with open(os.environ[E2E_RESULT_FILE], 'w', encoding='utf-8') as f:
-                f.write(json.dumps(data))
+                f.write(ensure_text(json.dumps(data)))
 
             # Rather than exiting we skip every test to avoid the following output:
             # !!!!!!!!!! _pytest.outcomes.Exit: !!!!!!!!!!
@@ -160,7 +161,8 @@ def pytest_report_teststatus(report, config):
         """
         # Skipping every test displays an `s` for each even when using
         # the minimum verbosity so we force zero output
-        return 'skipped', '', ''
+        if report.skipped:
+            return 'skipped', '', ''
 
 
 @pytest.fixture
@@ -187,8 +189,7 @@ def run_check(config=None, **kwargs):
         python_path = os.environ[E2E_PARENT_PYTHON]
         env = os.environ['HATCH_ENV_ACTIVE']
 
-        # TODO: switch to `ddev` when the old CLI is gone
-        check_command = [python_path, '-m', 'datadog_checks.dev', 'env', 'check', check, env, '--json']
+        check_command = [python_path, '-m', 'ddev', 'env', 'agent', check, env, 'check', '--json']
 
         if config:
             config = format_config(config)
@@ -197,7 +198,14 @@ def run_check(config=None, **kwargs):
             with open(config_file, 'wb') as f:
                 output = json.dumps(config).encode('utf-8')
                 f.write(output)
-            check_command.extend(['--config', config_file])
+            check_command.extend(['--config-file', config_file])
+
+        # TODO: remove these legacy flags when all usage of this fixture is migrated
+        if 'rate' in kwargs:
+            kwargs['check_rate'] = kwargs.pop('rate')
+
+        if 'times' in kwargs:
+            kwargs['check_times'] = kwargs.pop('times')
 
         for key, value in kwargs.items():
             if value is not False:
diff --git a/ddev/CHANGELOG.md b/ddev/CHANGELOG.md
index 0b00aa3425c9ac..a4d192f7e831c2 100644
--- a/ddev/CHANGELOG.md
+++ b/ddev/CHANGELOG.md
@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+***Added***:
+
+* Migrate E2E features ([#15931](https://github.com/DataDog/integrations-core/pull/15931))
+
 ## 5.1.1 / 2023-09-29
 
 ***Fixed***:
diff --git a/ddev/pyproject.toml b/ddev/pyproject.toml
index 6669a100be2ddf..aa53dc6468765a 100644
--- a/ddev/pyproject.toml
+++ b/ddev/pyproject.toml
@@ -28,7 +28,7 @@ classifiers = [
 dependencies = [
     "click~=8.1.6",
     "coverage",
-    "datadog-checks-dev[cli]~=25.0",
+    "datadog-checks-dev[cli]>=25.1.2,<26",
     "hatch>=1.6.3",
     "httpx",
     "jsonpointer",
diff --git a/ddev/src/ddev/cli/application.py b/ddev/src/ddev/cli/application.py
index 8e20095a279223..4293e1ad6a810b 100644
--- a/ddev/src/ddev/cli/application.py
+++ b/ddev/src/ddev/cli/application.py
@@ -4,10 +4,11 @@
 from __future__ import annotations
 
 import os
+from functools import cached_property
 from typing import cast
 
 from ddev.cli.terminal import Terminal
-from ddev.config.constants import AppEnvVars, VerbosityLevels
+from ddev.config.constants import AppEnvVars, ConfigEnvVars, VerbosityLevels
 from ddev.config.file import ConfigFile, RootConfig
 from ddev.repo.core import Repository
 from ddev.utils.fs import Path
@@ -18,7 +19,7 @@
 class Application(Terminal):
     def __init__(self, exit_func, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.platform = Platform(self.display_raw)
+        self.platform = Platform(self.output)
         self.__exit_func = exit_func
 
         self.config_file = ConfigFile()
@@ -40,6 +41,12 @@ def config(self) -> RootConfig:
     def repo(self) -> Repository:
         return self.__repo
 
+    @cached_property
+    def data_dir(self) -> Path:
+        from platformdirs import user_data_dir
+
+        return Path(os.getenv(ConfigEnvVars.DATA) or user_data_dir('ddev', appauthor=False)).expand()
+
     @property
     def github(self) -> GitHubManager:
         return self.__github
diff --git a/ddev/src/ddev/cli/config/__init__.py b/ddev/src/ddev/cli/config/__init__.py
index 2426ed6dd94eb3..28b854862802e5 100644
--- a/ddev/src/ddev/cli/config/__init__.py
+++ b/ddev/src/ddev/cli/config/__init__.py
@@ -43,7 +43,7 @@ def show(app, all_keys):
         from rich.syntax import Syntax
 
         text = app.config_file.read() if all_keys else app.config_file.read_scrubbed()
-        app.display_raw(Syntax(text.rstrip(), 'toml', background_color='default'))
+        app.output(Syntax(text.rstrip(), 'toml', background_color='default'))
 
 
 @config.command(short_help='Update the config file with any new fields')
@@ -146,4 +146,4 @@ def set_value(app, key, value):
     from rich.syntax import Syntax
 
     app.display_success('New setting:')
-    app.display_raw(Syntax(rendered_changed, 'toml', background_color='default'))
+    app.output(Syntax(rendered_changed, 'toml', background_color='default'))
diff --git a/ddev/src/ddev/cli/env/__init__.py b/ddev/src/ddev/cli/env/__init__.py
index b2858cdb14ff45..6f38d7bf9730cd 100644
--- a/ddev/src/ddev/cli/env/__init__.py
+++ b/ddev/src/ddev/cli/env/__init__.py
@@ -2,15 +2,16 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 import click
-from datadog_checks.dev.tooling.commands.env.check import check_run
-from datadog_checks.dev.tooling.commands.env.edit import edit
-from datadog_checks.dev.tooling.commands.env.ls import ls
-from datadog_checks.dev.tooling.commands.env.prune import prune
-from datadog_checks.dev.tooling.commands.env.reload import reload_env
-from datadog_checks.dev.tooling.commands.env.shell import shell
-from datadog_checks.dev.tooling.commands.env.start import start
-from datadog_checks.dev.tooling.commands.env.stop import stop
-from datadog_checks.dev.tooling.commands.env.test import test
+
+from ddev.cli.env.agent import agent
+from ddev.cli.env.check import check
+from ddev.cli.env.config import config
+from ddev.cli.env.reload import reload_command
+from ddev.cli.env.shell import shell
+from ddev.cli.env.show import show
+from ddev.cli.env.start import start
+from ddev.cli.env.stop import stop
+from ddev.cli.env.test import test_command
 
 
 @click.group(short_help='Manage environments')
@@ -20,12 +21,12 @@ def env():
     """
 
 
-env.add_command(check_run)
-env.add_command(edit)
-env.add_command(ls)
-env.add_command(prune)
-env.add_command(reload_env)
+env.add_command(agent)
+env.add_command(check)
+env.add_command(config)
+env.add_command(reload_command)
 env.add_command(shell)
+env.add_command(show)
 env.add_command(start)
 env.add_command(stop)
-env.add_command(test)
+env.add_command(test_command)
diff --git a/ddev/src/ddev/cli/env/agent.py b/ddev/src/ddev/cli/env/agent.py
new file mode 100644
index 00000000000000..f81991391feb83
--- /dev/null
+++ b/ddev/src/ddev/cli/env/agent.py
@@ -0,0 +1,68 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command(short_help='Invoke the Agent', context_settings={'ignore_unknown_options': True})
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.argument('args', required=True, nargs=-1)
+@click.option('--config-file', hidden=True)
+@click.pass_obj
+def agent(app: Application, *, intg_name: str, environment: str, args: tuple[str, ...], config_file: str | None):
+    """
+    Invoke the Agent.
+    """
+    import subprocess
+
+    from ddev.e2e.agent import get_agent_interface
+    from ddev.e2e.config import EnvDataStorage
+    from ddev.utils.fs import Path
+
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+
+    if not env_data.exists():
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+
+    metadata = env_data.read_metadata()
+    agent_type = metadata['env_type']
+    agent = get_agent_interface(agent_type)(app.platform, integration, environment, metadata, env_data.config_file)
+
+    full_args = list(args)
+    trigger_run = False
+    if full_args[0] == 'check':
+        trigger_run = True
+
+        # TODO: remove this when all invocations migrate to the actual command
+        if len(full_args) > 2 and full_args[1] == '--jmx-list':
+            full_args = ['jmx', 'list', full_args[2]]
+        # Automatically inject the integration name if not passed
+        elif len(full_args) == 1 or full_args[1].startswith('-'):
+            full_args.insert(1, intg_name)
+
+    if config_file is None or not trigger_run:
+        try:
+            agent.invoke(full_args)
+        except subprocess.CalledProcessError as e:
+            app.abort(code=e.returncode)
+
+        return
+
+    import json
+
+    temp_config_file = env_data.config_file.parent / f'{env_data.config_file.name}.bak.example'
+    env_data.config_file.replace(temp_config_file)
+    try:
+        env_data.write_config(json.loads(Path(config_file).read_text()))
+        agent.invoke(full_args)
+    finally:
+        temp_config_file.replace(env_data.config_file)
diff --git a/ddev/src/ddev/cli/env/check.py b/ddev/src/ddev/cli/env/check.py
new file mode 100644
index 00000000000000..57e032f5ba7573
--- /dev/null
+++ b/ddev/src/ddev/cli/env/check.py
@@ -0,0 +1,18 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+import click
+
+
+@click.command(context_settings={'ignore_unknown_options': True}, hidden=True)
+@click.argument('intg_name')
+@click.argument('environment')
+@click.argument('args', nargs=-1)
+@click.option('--config')
+@click.pass_context
+def check(ctx: click.Context, *, intg_name: str, environment: str, args: tuple[str, ...], config: str | None):
+    from ddev.cli.env.agent import agent
+
+    ctx.invoke(agent, intg_name=intg_name, environment=environment, args=('check', *args), config_file=config)
diff --git a/ddev/src/ddev/cli/env/config.py b/ddev/src/ddev/cli/env/config.py
new file mode 100644
index 00000000000000..d104bfc1d09116
--- /dev/null
+++ b/ddev/src/ddev/cli/env/config.py
@@ -0,0 +1,95 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.group(short_help='Manage the config file')
+def config():
+    pass
+
+
+@config.command(short_help='Edit the config file with your default editor')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.pass_obj
+def edit(app: Application, *, intg_name: str, environment: str):
+    """
+    Edit the config file with your default editor.
+    """
+    from ddev.e2e.config import EnvDataStorage
+
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+
+    if not env_data.exists():
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+
+    click.edit(filename=str(env_data.config_file))
+
+
+@config.command(short_help='Open the config location in your file manager')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.pass_obj
+def explore(app: Application, *, intg_name: str, environment: str):
+    """
+    Open the config location in your file manager.
+    """
+    from ddev.e2e.config import EnvDataStorage
+
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+
+    if not env_data.exists():
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+
+    click.launch(str(env_data.config_file), locate=True)
+
+
+@config.command(short_help='Show the location of the config file')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.pass_obj
+def find(app: Application, *, intg_name: str, environment: str):
+    """
+    Show the location of the config file.
+    """
+    from ddev.e2e.config import EnvDataStorage
+
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+
+    if not env_data.exists():
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+
+    app.output(f'[link={env_data.config_file}]{env_data.config_file}[/]')
+
+
+@config.command(short_help='Show the contents of the config file')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.pass_obj
+def show(app: Application, *, intg_name: str, environment: str):
+    """
+    Show the contents of the config file.
+    """
+    from ddev.e2e.config import EnvDataStorage
+
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+
+    if not env_data.exists():
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+
+    from rich.syntax import Syntax
+
+    text = env_data.config_file.read_text().rstrip()
+    app.output(Syntax(text, 'yaml', background_color='default'))
diff --git a/ddev/src/ddev/cli/env/reload.py b/ddev/src/ddev/cli/env/reload.py
new file mode 100644
index 00000000000000..9d9c2d9f09d1eb
--- /dev/null
+++ b/ddev/src/ddev/cli/env/reload.py
@@ -0,0 +1,38 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command('reload', short_help='Restart the Agent to detect environment changes')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.pass_obj
+def reload_command(app: Application, *, intg_name: str, environment: str):
+    """
+    Restart the Agent to detect environment changes.
+    """
+    from ddev.e2e.agent import get_agent_interface
+    from ddev.e2e.config import EnvDataStorage
+
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+
+    if not env_data.exists():
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+
+    metadata = env_data.read_metadata()
+    agent_type = metadata['env_type']
+    agent = get_agent_interface(agent_type)(app.platform, integration, environment, metadata, env_data.config_file)
+
+    try:
+        agent.restart()
+    except Exception as e:
+        app.abort(str(e))
diff --git a/ddev/src/ddev/cli/env/shell.py b/ddev/src/ddev/cli/env/shell.py
new file mode 100644
index 00000000000000..e9e70cf8067ded
--- /dev/null
+++ b/ddev/src/ddev/cli/env/shell.py
@@ -0,0 +1,38 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command('shell', short_help='Enter a shell alongside the Agent')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.pass_obj
+def shell(app: Application, *, intg_name: str, environment: str):
+    """
+    Enter a shell alongside the Agent.
+    """
+    from ddev.e2e.agent import get_agent_interface
+    from ddev.e2e.config import EnvDataStorage
+
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+
+    if not env_data.exists():
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+
+    metadata = env_data.read_metadata()
+    agent_type = metadata['env_type']
+    agent = get_agent_interface(agent_type)(app.platform, integration, environment, metadata, env_data.config_file)
+
+    try:
+        agent.enter_shell()
+    except Exception as e:
+        app.abort(str(e))
diff --git a/ddev/src/ddev/cli/env/show.py b/ddev/src/ddev/cli/env/show.py
new file mode 100644
index 00000000000000..4025f2f8d2b7ee
--- /dev/null
+++ b/ddev/src/ddev/cli/env/show.py
@@ -0,0 +1,87 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command('show', short_help='Show active or available environments')
+@click.argument('intg_name', required=False, metavar='INTEGRATION')
+@click.argument('environment', required=False)
+@click.option('--ascii', 'force_ascii', is_flag=True, help='Whether or not to only use ASCII characters')
+@click.pass_obj
+def show(app: Application, *, intg_name: str | None, environment: str | None, force_ascii: bool):
+    """
+    Show active or available environments.
+    """
+    from ddev.e2e.config import EnvDataStorage
+
+    storage = EnvDataStorage(app.data_dir)
+
+    # Display all active environments
+    if not intg_name:
+        for active_integration in storage.get_integrations():
+            integration = app.repo.integrations.get(active_integration)
+            envs = storage.get_environments(integration.name)
+
+            columns: dict[str, dict[int, str]] = {'Name': {}, 'Agent type': {}}
+            for i, name in enumerate(envs):
+                env_data = storage.get(integration.name, name)
+                metadata = env_data.read_metadata()
+
+                columns['Name'][i] = name
+                columns['Agent type'][i] = metadata['env_type']
+
+            app.display_table(active_integration, columns, show_lines=True, force_ascii=force_ascii)
+    # Display active and available environments for a specific integration
+    elif not environment:
+        import json
+        import sys
+
+        integration = app.repo.integrations.get(intg_name)
+        active_envs = storage.get_environments(integration.name)
+
+        active_columns: dict[str, dict[int, str]] = {'Name': {}, 'Agent type': {}}
+        for i, name in enumerate(active_envs):
+            env_data = storage.get(integration.name, name)
+            metadata = env_data.read_metadata()
+
+            active_columns['Name'][i] = name
+            active_columns['Agent type'][i] = metadata['env_type']
+
+        with integration.path.as_cwd():
+            environments = json.loads(
+                app.platform.check_command_output([sys.executable, '-m', 'hatch', 'env', 'show', '--json'])
+            )
+
+        available_columns: dict[str, dict[int, str]] = {'Name': {}}
+        for i, (name, data) in enumerate(environments.items()):
+            if not data.get('e2e-env') or name in active_envs:
+                continue
+
+            available_columns['Name'][i] = name
+
+        app.display_table('Active', active_columns, show_lines=True, force_ascii=force_ascii)
+        app.display_table('Available', available_columns, show_lines=True, force_ascii=force_ascii)
+    # Display information about a specific environment
+    else:
+        from ddev.e2e.agent import get_agent_interface
+
+        integration = app.repo.integrations.get(intg_name)
+        env_data = storage.get(integration.name, environment)
+
+        if not env_data.exists():
+            app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+
+        metadata = env_data.read_metadata()
+        agent_type = metadata['env_type']
+        agent = get_agent_interface(agent_type)(app.platform, integration, environment, metadata, env_data.config_file)
+
+        app.display_pair('Agent type', agent_type)
+        app.display_pair('Agent ID', agent.get_id())
diff --git a/ddev/src/ddev/cli/env/start.py b/ddev/src/ddev/cli/env/start.py
new file mode 100644
index 00000000000000..adc0876b31012c
--- /dev/null
+++ b/ddev/src/ddev/cli/env/start.py
@@ -0,0 +1,194 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command(short_help='Start an environment')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.option('--dev', 'local_dev', is_flag=True, help='Install the local version of the integration')
+@click.option(
+    '--base',
+    'local_base',
+    is_flag=True,
+    help='Install the local version of the base package, implicitly enabling the `--dev` option',
+)
+@click.option(
+    '--agent',
+    '-a',
+    'agent_build',
+    help=(
+        'The Agent build to use e.g. a Docker image like `datadog/agent:latest`. You can '
+        'also use the name of an Agent defined in the `agents` configuration section.'
+    ),
+)
+@click.option(
+    '-e',
+    'extra_env_vars',
+    multiple=True,
+    help='Environment variables to pass to the Agent e.g. -e DD_URL=app.datadoghq.com -e DD_API_KEY=foobar',
+)
+@click.option('--dogstatsd', is_flag=True, hidden=True)
+@click.option('--hide-help', is_flag=True, hidden=True)
+@click.option('--ignore-state', is_flag=True, hidden=True)
+@click.pass_context
+def start(
+    ctx: click.Context,
+    *,
+    intg_name: str,
+    environment: str,
+    local_dev: bool,
+    local_base: bool,
+    agent_build: str | None,
+    extra_env_vars: tuple[str, ...],
+    dogstatsd: bool,
+    hide_help: bool,
+    ignore_state: bool,
+):
+    """
+    Start an environment.
+    """
+    import json
+    import os
+    from contextlib import suppress
+
+    from ddev.e2e.agent import get_agent_interface
+    from ddev.e2e.config import EnvDataStorage
+    from ddev.e2e.constants import E2EEnvVars
+    from ddev.e2e.run import E2EEnvironmentRunner
+    from ddev.utils.fs import Path, temp_directory
+
+    app: Application = ctx.obj
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+    runner = E2EEnvironmentRunner(environment)
+
+    # Do as much error checking as possible before attempting to start the environment
+    if env_data.exists():
+        if ignore_state:
+            return
+
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is already running')
+
+    local_packages: dict[Path, str] = {}
+    if local_base:
+        features = '[db,deps,http,json,kube]'
+        if (base_package_path := (app.repo.path / 'datadog_checks_base')).is_dir():
+            local_packages[base_package_path] = features
+        else:
+            for repo_name, repo_path in app.config.repos.items():
+                if repo_name == app.repo.name:
+                    continue
+                elif (base_package_path := (Path(repo_path).expand() / 'datadog_checks_base')).is_dir():
+                    local_packages[base_package_path] = features
+                    break
+            else:
+                app.abort('Unable to find a local version of the base package')
+
+        # When using multiple namespaced packages it is required that they are all
+        # installed in the same way (normal versus editable)
+        local_dev = True
+
+    # Install the integration after the base package for the following reasons:
+    # 1. We want the dependencies of the integration in question to take precedence
+    # 2. The rare, ephemeral situation where a new version of the base package is
+    #    required but is not yet released
+    if local_dev:
+        local_packages[integration.path] = '[deps]'
+
+    app.display_header(f'Starting: {environment}')
+
+    with temp_directory() as temp_dir:
+        result_file = temp_dir / 'result.json'
+        env_vars = {E2EEnvVars.RESULT_FILE: str(result_file)}
+
+        with integration.path.as_cwd(env_vars=env_vars), runner.start() as command:
+            process = app.platform.run_command(command)
+            if process.returncode:
+                app.abort(code=process.returncode)
+
+        if not result_file.is_file():  # no cov
+            app.abort(f'No E2E result file found at {result_file}')
+
+        result = json.loads(result_file.read_text())
+
+    metadata = result['metadata']
+    env_data.write_metadata(metadata)
+
+    config = result['config']
+    env_data.write_config(config)
+
+    agent_type = metadata['env_type']
+    agent = get_agent_interface(agent_type)(app.platform, integration, environment, metadata, env_data.config_file)
+
+    if not agent_build:
+        agent_build = (
+            os.getenv(E2EEnvVars.AGENT_BUILD_PY2 if agent.python_version[0] == 2 else E2EEnvVars.AGENT_BUILD)
+            or app.config.agent.config.get(agent_type)
+            or ''
+        )
+
+    agent_env_vars = _get_agent_env_vars(app.config.org.config, metadata, extra_env_vars, dogstatsd)
+
+    try:
+        agent.start(agent_build=agent_build, local_packages=local_packages, env_vars=agent_env_vars)
+    except Exception as e:
+        from ddev.cli.env.stop import stop
+
+        app.display_critical(f'Unable to start the Agent: {e}')
+        with suppress(Exception):
+            ctx.invoke(stop, intg_name=intg_name, environment=environment)
+
+        app.abort()
+
+    if not hide_help:
+        app.output()
+        app.display_pair('Stop environment', app.style_info(f'ddev env stop {intg_name} {environment}'))
+        app.display_pair('Execute tests', app.style_info(f'ddev env test {intg_name} {environment}'))
+        app.display_pair('Check status', app.style_info(f'ddev env agent {intg_name} {environment} status'))
+        app.display_pair('Trigger run', app.style_info(f'ddev env agent {intg_name} {environment} check'))
+        app.display_pair('Reload config', app.style_info(f'ddev env reload {intg_name} {environment}'))
+        app.display_pair('Manage config', app.style_info('ddev env config'))
+        app.display_pair('Config file', f'[link={env_data.config_file}]{env_data.config_file}[/]')
+
+
+def _get_agent_env_vars(org_config, metadata, extra_env_vars, dogstatsd):
+    from ddev.e2e.constants import DEFAULT_DOGSTATSD_PORT, E2EEnvVars
+
+    # Use the environment variables defined by tests as defaults so tooling can override them
+    env_vars: dict[str, str] = metadata.get('env_vars', {}).copy()
+    env_vars.update(ev.split('=', maxsplit=1) for ev in extra_env_vars)
+
+    if api_key := org_config.get('api_key'):
+        env_vars['DD_API_KEY'] = api_key
+
+    if site := org_config.get('site'):
+        env_vars['DD_SITE'] = site
+
+    # Custom core Agent intake
+    if dd_url := org_config.get('dd_url'):
+        env_vars['DD_DD_URL'] = dd_url
+
+    # Custom logs Agent intake
+    if log_url := org_config.get('log_url'):
+        env_vars['DD_LOGS_CONFIG_DD_URL'] = log_url
+
+    # TODO: remove the CLI flag and exclusively rely on the metadata flag
+    if metadata.get('dogstatsd') or dogstatsd:
+        env_vars['DD_DOGSTATSD_PORT'] = str(DEFAULT_DOGSTATSD_PORT)
+        env_vars['DD_DOGSTATSD_NON_LOCAL_TRAFFIC'] = 'true'
+        env_vars['DD_DOGSTATSD_METRICS_STATS_ENABLE'] = 'true'
+
+    # Enable logs Agent by default if the environment is mounting logs
+    if any(ev.startswith(E2EEnvVars.LOGS_DIR_PREFIX) for ev in env_vars):
+        env_vars.setdefault('DD_LOGS_ENABLED', 'true')
+
+    return env_vars
diff --git a/ddev/src/ddev/cli/env/stop.py b/ddev/src/ddev/cli/env/stop.py
new file mode 100644
index 00000000000000..4b24eb5a1519eb
--- /dev/null
+++ b/ddev/src/ddev/cli/env/stop.py
@@ -0,0 +1,59 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command(short_help='Stop an environment')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.option('--ignore-state', is_flag=True, hidden=True)
+@click.pass_obj
+def stop(app: Application, *, intg_name: str, environment: str, ignore_state: bool):
+    """
+    Stop an environment.
+    """
+    from ddev.e2e.agent import get_agent_interface
+    from ddev.e2e.config import EnvDataStorage
+    from ddev.e2e.constants import E2EEnvVars
+    from ddev.e2e.run import E2EEnvironmentRunner
+    from ddev.utils.fs import temp_directory
+
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+    runner = E2EEnvironmentRunner(environment)
+
+    if not env_data.exists():
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+    elif ignore_state:
+        return
+
+    app.display_header(f'Stopping: {environment}')
+
+    # TODO: remove this required result file indicator once the E2E migration is complete
+    with temp_directory() as temp_dir:
+        result_file = temp_dir / 'result.json'
+        env_vars = {E2EEnvVars.RESULT_FILE: str(result_file)}
+
+        metadata = env_data.read_metadata()
+        env_vars.update(metadata['env_vars'])
+
+        agent_type = metadata['env_type']
+        agent = get_agent_interface(agent_type)(app.platform, integration, environment, metadata, env_data.config_file)
+
+        try:
+            agent.stop()
+        finally:
+            env_data.remove()
+
+            with integration.path.as_cwd(env_vars=env_vars), runner.stop() as command:
+                process = app.platform.run_command(command)
+                if process.returncode:
+                    app.abort(code=process.returncode)
diff --git a/ddev/src/ddev/cli/env/test.py b/ddev/src/ddev/cli/env/test.py
new file mode 100644
index 00000000000000..da0e1e1e959c49
--- /dev/null
+++ b/ddev/src/ddev/cli/env/test.py
@@ -0,0 +1,131 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command('test', short_help='Test environments')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment', required=False)
+@click.argument('args', nargs=-1)
+@click.option('--dev', 'local_dev', is_flag=True, help='Install the local version of the integration')
+@click.option(
+    '--base',
+    'local_base',
+    is_flag=True,
+    help='Install the local version of the base package, implicitly enabling the `--dev` option',
+)
+@click.option(
+    '--agent',
+    '-a',
+    'agent_build',
+    help=(
+        'The Agent build to use e.g. a Docker image like `datadog/agent:latest`. You can '
+        'also use the name of an Agent defined in the `agents` configuration section.'
+    ),
+)
+@click.option(
+    '-e',
+    'extra_env_vars',
+    multiple=True,
+    help='Environment variables to pass to the Agent e.g. -e DD_URL=app.datadoghq.com -e DD_API_KEY=foobar',
+)
+@click.option('--junit', is_flag=True, hidden=True)
+@click.option('--python-filter', envvar='PYTHON_FILTER', hidden=True)
+@click.option('--new-env', is_flag=True, hidden=True)
+@click.pass_context
+def test_command(
+    ctx: click.Context,
+    *,
+    intg_name: str,
+    environment: str | None,
+    args: tuple[str, ...],
+    local_dev: bool,
+    local_base: bool,
+    agent_build: str | None,
+    extra_env_vars: tuple[str, ...],
+    junit: bool,
+    python_filter: str | None,
+    new_env: bool,
+):
+    """
+    Test environments.
+
+    If no environment is specified, `active` is selected which will test all environments
+    that are currently running. You may choose `all` to test all environments whether or not
+    they are running.
+
+    Testing active environments will not stop them after tests complete. Testing environments
+    that are not running will start and stop them automatically.
+    """
+    from ddev.cli.env.start import start
+    from ddev.cli.env.stop import stop
+    from ddev.cli.test import test
+    from ddev.e2e.config import EnvDataStorage
+    from ddev.utils.ci import running_in_ci
+    from ddev.utils.structures import EnvVars
+
+    app: Application = ctx.obj
+    storage = EnvDataStorage(app.data_dir)
+    integration = app.repo.integrations.get(intg_name)
+    active_envs = storage.get_environments(integration.name)
+
+    if environment is None:
+        environment = 'all' if (not active_envs or running_in_ci()) else 'active'
+
+    if environment == 'all':
+        import json
+        import sys
+
+        with integration.path.as_cwd():
+            env_data_output = app.platform.check_command_output(
+                [sys.executable, '-m', 'hatch', '--no-color', '--no-interactive', 'env', 'show', '--json']
+            )
+            try:
+                environments = json.loads(env_data_output)
+            except json.JSONDecodeError:
+                app.abort(f'Failed to parse environments for `{integration.name}`:\n{repr(env_data_output)}')
+
+        env_names = [
+            name
+            for name, data in environments.items()
+            if data.get('e2e-env') and (python_filter is None or data.get('python') == python_filter)
+        ]
+    elif environment == 'active':
+        env_names = active_envs
+    else:
+        env_names = [environment]
+
+    if not env_names:
+        return
+
+    app.display_header(integration.display_name)
+
+    active = set(active_envs)
+    for env_name in env_names:
+        env_active = env_name in active
+        ctx.invoke(
+            start,
+            intg_name=intg_name,
+            environment=env_name,
+            local_dev=local_dev,
+            local_base=local_base,
+            agent_build=agent_build,
+            extra_env_vars=extra_env_vars,
+            hide_help=True,
+            ignore_state=env_active,
+        )
+
+        env_data = storage.get(integration.name, env_name)
+        metadata = env_data.read_metadata()
+        with EnvVars(metadata['env_vars']):
+            ctx.invoke(test, target_spec=f'{intg_name}:{env_name}', args=args, junit=True, hide_header=True, e2e=True)
+
+        ctx.invoke(stop, intg_name=intg_name, environment=env_name, ignore_state=env_active)
diff --git a/ddev/src/ddev/cli/terminal.py b/ddev/src/ddev/cli/terminal.py
index 20feaa064df6a6..bc0b8557fb6342 100644
--- a/ddev/src/ddev/cli/terminal.py
+++ b/ddev/src/ddev/cli/terminal.py
@@ -134,6 +134,7 @@ def __init__(self, verbosity: int, enable_color: bool, interactive: bool):
         self.console = Console(
             force_terminal=enable_color,
             no_color=enable_color is False,
+            highlight=False,
             # Force consistent output for test assertions
             legacy_windows=False if 'DDEV_SELF_TESTING' in os.environ else None,
         )
@@ -150,6 +151,28 @@ def __init__(self, verbosity: int, enable_color: bool, interactive: bool):
         # Chosen as the default since it's compatible everywhere and looks nice
         self._style_spinner = 'simpleDotsScrolling'
 
+    @cached_property
+    def kv_separator(self) -> Style:
+        return self.style_warning('->')
+
+    def style_success(self, text: str) -> Text:
+        return Text(text, style=self._style_level_success)
+
+    def style_error(self, text: str) -> Text:
+        return Text(text, style=self._style_level_error)
+
+    def style_warning(self, text: str) -> Text:
+        return Text(text, style=self._style_level_warning)
+
+    def style_waiting(self, text: str) -> Text:
+        return Text(text, style=self._style_level_waiting)
+
+    def style_info(self, text: str) -> Text:
+        return Text(text, style=self._style_level_info)
+
+    def style_debug(self, text: str) -> Text:
+        return Text(text, style=self._style_level_debug)
+
     def initialize_styles(self, styles: dict):  # no cov
         # Lazily display errors so that they use the correct style
         errors = []
@@ -187,37 +210,37 @@ def display_error(self, text='', stderr=True, indent=None, link=None, **kwargs):
         if self.verbosity < VerbosityLevels.ERROR:
             return
 
-        self.output(text, self._style_level_error, stderr=stderr, indent=indent, link=link, **kwargs)
+        self._output(text, self._style_level_error, stderr=stderr, indent=indent, link=link, **kwargs)
 
     def display_warning(self, text='', stderr=True, indent=None, link=None, **kwargs):
         if self.verbosity < VerbosityLevels.WARNING:
             return
 
-        self.output(text, self._style_level_warning, stderr=stderr, indent=indent, link=link, **kwargs)
+        self._output(text, self._style_level_warning, stderr=stderr, indent=indent, link=link, **kwargs)
 
     def display_info(self, text='', stderr=True, indent=None, link=None, **kwargs):
         if self.verbosity < VerbosityLevels.INFO:
             return
 
-        self.output(text, self._style_level_info, stderr=stderr, indent=indent, link=link, **kwargs)
+        self._output(text, self._style_level_info, stderr=stderr, indent=indent, link=link, **kwargs)
 
     def display_success(self, text='', stderr=True, indent=None, link=None, **kwargs):
         if self.verbosity < VerbosityLevels.INFO:
             return
 
-        self.output(text, self._style_level_success, stderr=stderr, indent=indent, link=link, **kwargs)
+        self._output(text, self._style_level_success, stderr=stderr, indent=indent, link=link, **kwargs)
 
     def display_waiting(self, text='', stderr=True, indent=None, link=None, **kwargs):
         if self.verbosity < VerbosityLevels.INFO:
             return
 
-        self.output(text, self._style_level_waiting, stderr=stderr, indent=indent, link=link, **kwargs)
+        self._output(text, self._style_level_waiting, stderr=stderr, indent=indent, link=link, **kwargs)
 
     def display_debug(self, text='', stderr=True, indent=None, link=None, **kwargs):
         if self.verbosity < VerbosityLevels.DEBUG:
             return
 
-        self.output(f'DEBUG: {text}', None, stderr=stderr, indent=indent, link=link, **kwargs)
+        self._output(f'DEBUG: {text}', None, stderr=stderr, indent=indent, link=link, **kwargs)
 
     def display_header(self, title=''):
         self.console.rule(Text(title, self._style_level_success))
@@ -225,7 +248,45 @@ def display_header(self, title=''):
     def display_markdown(self, text, stderr=False, **kwargs):
         from rich.markdown import Markdown
 
-        self.display_raw(Markdown(text), stderr=stderr, **kwargs)
+        self.output(Markdown(text), stderr=stderr, **kwargs)
+
+    def display_pair(self, key, value):
+        self.output(self.style_success(key), self.kv_separator, value)
+
+    def display_table(self, title, columns, *, show_lines=False, column_options=None, force_ascii=False, num_rows=0):
+        from rich.table import Table
+
+        if column_options is None:
+            column_options = {}
+
+        table_options = {}
+        if force_ascii:
+            from rich.box import ASCII_DOUBLE_HEAD
+
+            table_options['box'] = ASCII_DOUBLE_HEAD
+            table_options['safe_box'] = True
+
+        table = Table(title=title, show_lines=show_lines, title_style='', **table_options)
+        columns = dict(columns)
+
+        for title, indices in list(columns.items()):
+            if indices:
+                table.add_column(title, style='bold', **column_options.get(title, {}))
+            else:
+                columns.pop(title)
+
+        if not columns:
+            return
+
+        for i in range(num_rows or max(map(max, columns.values())) + 1):
+            row = []
+            for indices in columns.values():
+                row.append(indices.get(i, ''))
+
+            if any(row):
+                table.add_row(*row)
+
+        self.output(table)
 
     def create_validation_tracker(self, label: str):
         from rich.tree import Tree
@@ -253,30 +314,29 @@ def status(self):
             finalizer=lambda: setattr(self.platform, 'displaying_status', False),
         )
 
-    def output(self, text='', style=None, *, stderr=False, indent=None, link=None, **kwargs):
-        kwargs.setdefault('overflow', 'ignore')
-        kwargs.setdefault('no_wrap', True)
-        kwargs.setdefault('crop', False)
-
+    def _output(self, text='', style=None, *, stderr=False, indent=None, link=None, **kwargs):
         if indent:
             text = indent_text(text, indent)
 
         if link:
             style = style.update_link(self.platform.format_file_uri(link))
 
+        self.output(text, stderr=stderr, style=style, **kwargs)
+
+    def output(self, *args, stderr=False, **kwargs):
+        kwargs.setdefault('overflow', 'ignore')
+        kwargs.setdefault('no_wrap', True)
+        kwargs.setdefault('crop', False)
+
         if not stderr:
-            self.console.print(text, style=style, **kwargs)
+            self.console.print(*args, **kwargs)
         else:
             self.console.stderr = True
             try:
-                self.console.print(text, style=style, **kwargs)
+                self.console.print(*args, **kwargs)
             finally:
                 self.console.stderr = False
 
-    def display_raw(self, text, **kwargs):
-        # No styling
-        self.console.print(text, overflow='ignore', no_wrap=True, crop=False, **kwargs)
-
     @staticmethod
     def prompt(text, **kwargs):
         return click.prompt(text, **kwargs)
diff --git a/ddev/src/ddev/cli/test/__init__.py b/ddev/src/ddev/cli/test/__init__.py
index f0c02c5a5e3bc6..87c88c60045e55 100644
--- a/ddev/src/ddev/cli/test/__init__.py
+++ b/ddev/src/ddev/cli/test/__init__.py
@@ -38,6 +38,7 @@ def fix_coverage_report(report_file: Path):
 @click.option('--list', '-l', 'list_envs', is_flag=True, help='Show available test environments')
 @click.option('--python-filter', envvar='PYTHON_FILTER', hidden=True)
 @click.option('--junit', is_flag=True, hidden=True)
+@click.option('--hide-header', is_flag=True, hidden=True)
 @click.option('--e2e', is_flag=True, hidden=True)
 @click.pass_obj
 def test(
@@ -56,6 +57,7 @@ def test(
     list_envs: bool,
     python_filter: str | None,
     junit: bool,
+    hide_header: bool,
     e2e: bool,
 ):
     """
@@ -173,7 +175,8 @@ def test(
 
     app.display_debug(f'Targets: {", ".join(targets)}')
     for target in targets.values():
-        app.display_header(target.display_name)
+        if not hide_header:
+            app.display_header(target.display_name)
 
         command = base_command.copy()
         env_vars = global_env_vars.copy()
diff --git a/ddev/src/ddev/config/constants.py b/ddev/src/ddev/config/constants.py
index 556bede831fcbb..aa95b3d49c40a0 100644
--- a/ddev/src/ddev/config/constants.py
+++ b/ddev/src/ddev/config/constants.py
@@ -12,6 +12,8 @@ class AppEnvVars:
 
 
 class ConfigEnvVars:
+    DATA = 'HATCH_DATA_DIR'
+    CACHE = 'HATCH_CACHE_DIR'
     CONFIG = 'DDEV_CONFIG'
 
 
diff --git a/ddev/src/ddev/e2e/__init__.py b/ddev/src/ddev/e2e/__init__.py
new file mode 100644
index 00000000000000..e0cc3d0a7662c2
--- /dev/null
+++ b/ddev/src/ddev/e2e/__init__.py
@@ -0,0 +1,3 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
diff --git a/ddev/src/ddev/e2e/agent/__init__.py b/ddev/src/ddev/e2e/agent/__init__.py
new file mode 100644
index 00000000000000..531b78f4b63470
--- /dev/null
+++ b/ddev/src/ddev/e2e/agent/__init__.py
@@ -0,0 +1,18 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ddev.e2e.agent.interface import AgentInterface
+
+
+def get_agent_interface(agent_type: str) -> type[AgentInterface]:
+    if agent_type == 'docker':
+        from ddev.e2e.agent.docker import DockerAgent
+
+        return DockerAgent
+
+    raise NotImplementedError(f'Unsupported Agent type: {agent_type}')
diff --git a/ddev/src/ddev/e2e/agent/docker.py b/ddev/src/ddev/e2e/agent/docker.py
new file mode 100644
index 00000000000000..63dc4b2ad87f9d
--- /dev/null
+++ b/ddev/src/ddev/e2e/agent/docker.py
@@ -0,0 +1,293 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+import os
+import sys
+from functools import cached_property
+from typing import TYPE_CHECKING, Callable
+
+from ddev.e2e.agent.interface import AgentInterface
+from ddev.utils.structures import EnvVars
+
+if TYPE_CHECKING:
+    import subprocess
+
+    from ddev.utils.fs import Path
+
+
+class DockerAgent(AgentInterface):
+    @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:
+                pass
+
+        return False
+
+    @cached_property
+    def _container_name(self) -> str:
+        return f'dd_{super().get_id()}'
+
+    @cached_property
+    def _is_windows_container(self) -> bool:
+        return self.metadata.get('docker_platform') == 'windows'
+
+    @cached_property
+    def _package_mount_dir(self) -> str:
+        return 'C:\\Users\\ContainerAdministrator\\' if self._is_windows_container else '/home/'
+
+    @cached_property
+    def _config_mount_dir(self) -> str:
+        return (
+            f'C:\\ProgramData\\Datadog\\conf.d\\{self.integration.name}.d'
+            if self._is_windows_container
+            else f'/etc/datadog-agent/conf.d/{self.integration.name}.d'
+        )
+
+    @cached_property
+    def _python_path(self) -> str:
+        return (
+            f'C:\\Program Files\\Datadog\\Datadog Agent\\embedded{self.python_version[0]}\\python.exe'
+            if self._is_windows_container
+            else f'/opt/datadog-agent/embedded/bin/python{self.python_version[0]}'
+        )
+
+    def _format_command(self, command: list[str]) -> list[str]:
+        cmd = ['docker', 'exec']
+        if self._isatty:
+            cmd.append('-it')
+        cmd.append(self._container_name)
+
+        if command[0] == 'pip':
+            command = command[1:]
+            cmd.extend([self._python_path, '-m', 'pip'])
+
+        cmd.extend(command)
+        return cmd
+
+    def _captured_process(self, command: list[str]) -> subprocess.CompletedProcess:
+        return self._run_command(
+            command, stdout=self.platform.modules.subprocess.PIPE, stderr=self.platform.modules.subprocess.STDOUT
+        )
+
+    def _run_command(self, command: list[str], **kwargs) -> subprocess.CompletedProcess:
+        with EnvVars({'DOCKER_CLI_HINTS': 'false'}):
+            return self.platform.run_command(command, **kwargs)
+
+    def get_id(self) -> str:
+        return self._container_name
+
+    def start(self, *, agent_build: str, local_packages: dict[Path, str], env_vars: dict[str, str]) -> None:
+        if not agent_build:
+            agent_build = 'datadog/agent-dev:master'
+
+        # Add a potentially missing `py` suffix for default non-RC builds
+        if (
+            'rc' not in agent_build
+            and 'py' not in agent_build
+            and agent_build != 'datadog/agent:6'
+            and agent_build != 'datadog/agent:7'
+        ):
+            agent_build = f'{agent_build}-py{self.python_version[0]}'
+
+        if self.metadata.get('use_jmx') and not agent_build.endswith('-jmx'):
+            agent_build += '-jmx'
+
+        env_vars = env_vars.copy()
+
+        # Containerized agents require an API key to start
+        if 'DD_API_KEY' not in env_vars:
+            # This fake key must be the proper length
+            env_vars['DD_API_KEY'] = 'a' * 32
+
+        # Set Agent hostname for CI
+        env_vars['DD_HOSTNAME'] = get_hostname()
+
+        # Run API on a random free port
+        env_vars['DD_API_PORT'] = str(find_free_port(get_ip()))
+
+        # Disable trace Agent
+        env_vars['DD_APM_ENABLED'] = 'false'
+
+        # Set up telemetry
+        env_vars['DD_TELEMETRY_ENABLED'] = '1'
+        env_vars['DD_EXPVAR_PORT'] = '5000'
+
+        # TODO: Remove this when Python 2 support is removed
+        #
+        # Don't write .pyc, needed to fix this issue (only Python 2):
+        # More info: https://github.com/DataDog/integrations-core/pull/5454
+        # When reinstalling a package, .pyc are not cleaned correctly. The issue is fixed by not writing them
+        # in the first place.
+        env_vars['PYTHONDONTWRITEBYTECODE'] = '1'
+
+        volumes = []
+
+        # Mount the config directory, not the file, to ensure updates are propagated:
+        # https://github.com/moby/moby/issues/15793#issuecomment-135411504
+        volumes.append(f'{self.config_file.parent}:{self._config_mount_dir}')
+
+        # It is safe to assume that the directory name is unique across all repos
+        for local_package in local_packages:
+            volumes.append(f'{local_package}:{self._package_mount_dir}{local_package.name}')
+
+        if not self._is_windows_container:
+            volumes.append('/proc:/host/proc')
+
+        if not self.platform.windows:
+            volumes.extend(self.metadata.get('docker_volumes', []))
+        elif not self._is_windows_container:
+            for volume in self.metadata.get('docker_volumes', []):
+                parts = volume.split(':')
+                possible_file = ':'.join(parts[:2])
+                if not os.path.isfile(possible_file):
+                    volumes.append(volume)
+                else:
+                    # Workaround for https://github.com/moby/moby/issues/30555
+                    vm_file = possible_file.replace(':', '/', 1).replace('\\', '/')
+                    remaining = ':'.join(parts[2:])
+                    volumes.append(f'/{vm_file}:{remaining}')
+
+        process = self._run_command(['docker', 'pull', agent_build])
+        if process.returncode:
+            raise RuntimeError(f'Could not pull image {agent_build}')
+
+        command = [
+            'docker',
+            'run',
+            # Keep it up
+            '-d',
+            # Ensure consistent naming
+            '--name',
+            self._container_name,
+        ]
+
+        # Ensure access to host network
+        #
+        # Windows containers accessing the host network must use `docker.for.win.localhost` or `host.docker.internal`:
+        # https://docs.docker.com/docker-for-windows/networking/#use-cases-and-workarounds
+        if not self._is_windows_container:
+            command.extend(['--network', 'host'])
+
+        for volume in volumes:
+            command.extend(['-v', volume])
+
+        # Any environment variables passed to the start command in addition to the default ones
+        for key, value in sorted(env_vars.items()):
+            command.extend(['-e', f'{key}={value}'])
+
+        # The docker `--add-host` command will reliably create entries in the `/etc/hosts` file,
+        # otherwise, edits to that file will be overwritten on container restarts
+        for host, ip in self.metadata.get('custom_hosts', []):
+            command.extend(['--add-host', f'{host}:{ip}'])
+
+        if dogstatsd_port := env_vars.get('DD_DOGSTATSD_PORT'):
+            command.extend(['-p', f'{dogstatsd_port}:{dogstatsd_port}/udp'])
+
+        if 'proxy' in self.metadata:
+            if 'http' in self.metadata['proxy']:
+                command.extend(['-e', f'DD_PROXY_HTTP={self.metadata["proxy"]["http"]}'])
+            if 'https' in self.metadata['proxy']:
+                command.extend(['-e', f'DD_PROXY_HTTPS={self.metadata["proxy"]["https"]}'])
+
+        # The chosen tag
+        command.append(agent_build)
+
+        process = self._captured_process(command)
+        if process.returncode:
+            raise RuntimeError(
+                f'Unable to start Agent container `{self._container_name}`: {process.stdout.decode("utf-8")}'
+            )
+
+        start_commands = self.metadata.get('start_commands', [])
+        if start_commands:
+            for start_command in start_commands:
+                formatted_command = self._format_command(self.platform.modules.shlex.split(start_command))
+                process = self._run_command(formatted_command)
+                if process.returncode:
+                    raise RuntimeError(
+                        f'Unable to run start-up command in Agent container `{self._container_name}`: '
+                        f'{process.stdout.decode("utf-8")}'
+                    )
+
+        if local_packages:
+            base_pip_command = self._format_command(['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}'
+                process = self._run_command([*base_pip_command, package_mount])
+                if process.returncode:
+                    raise RuntimeError(
+                        f'Unable to install package `{local_package.name}` in Agent container '
+                        f'`{self._container_name}`: {process.stdout.decode("utf-8")}'
+                    )
+
+        post_install_commands = self.metadata.get('post_install_commands', [])
+        if post_install_commands:
+            for post_install_command in post_install_commands:
+                formatted_command = self._format_command(self.platform.modules.shlex.split(post_install_command))
+                process = self._run_command(formatted_command)
+                if process.returncode:
+                    raise RuntimeError(
+                        f'Unable to run post-install command in Agent container `{self._container_name}`: '
+                        f'{process.stdout.decode("utf-8")}'
+                    )
+
+        if local_packages or start_commands or post_install_commands:
+            self.restart()
+
+    def stop(self) -> None:
+        for command in (
+            ['docker', 'stop', '-t', '0', self._container_name],
+            ['docker', 'rm', self._container_name],
+        ):
+            process = self._captured_process(command)
+            if process.returncode:
+                raise RuntimeError(
+                    f'Unable to stop Agent container `{self._container_name}`: {process.stdout.decode("utf-8")}'
+                )
+
+    def restart(self) -> None:
+        process = self._captured_process(['docker', 'restart', self._container_name])
+        if process.returncode:
+            raise RuntimeError(
+                f'Unable to restart Agent container `{self._container_name}`: {process.stdout.decode("utf-8")}'
+            )
+
+    def invoke(self, args: list[str]) -> None:
+        self._run_command(self._format_command(['agent', *args]), check=True)
+
+    def enter_shell(self) -> None:
+        self.platform.exit_with_command(self._format_command(['cmd' if self._is_windows_container else '/bin/bash']))
+
+
+def get_hostname():
+    import socket
+
+    return socket.gethostname()
+
+
+def find_free_port(ip):
+    """Return a port available for listening on the given `ip`."""
+    import socket
+    from contextlib import closing
+
+    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+        s.bind((ip, 0))
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        return s.getsockname()[1]
+
+
+def get_ip():
+    """Return the IP address used to connect to external networks."""
+    import socket
+    from contextlib import closing
+
+    with closing(socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) as s:
+        # doesn't have to be reachable
+        s.connect(('10.255.255.255', 1))
+        return s.getsockname()[0]
diff --git a/ddev/src/ddev/e2e/agent/interface.py b/ddev/src/ddev/e2e/agent/interface.py
new file mode 100644
index 00000000000000..5aca0fbd74c1e1
--- /dev/null
+++ b/ddev/src/ddev/e2e/agent/interface.py
@@ -0,0 +1,79 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from functools import cached_property
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from ddev.integration.core import Integration
+    from ddev.utils.fs import Path
+    from ddev.utils.platform import Platform
+
+
+class AgentInterface(ABC):
+    def __init__(
+        self, platform: Platform, integration: Integration, env: str, metadata: dict[str, Any], config_file: Path
+    ) -> None:
+        self.__platform = platform
+        self.__integration = integration
+        self.__env = env
+        self.__metadata = metadata
+        self.__config_file = config_file
+
+    @property
+    def platform(self) -> Platform:
+        return self.__platform
+
+    @property
+    def integration(self) -> Integration:
+        return self.__integration
+
+    @property
+    def env(self) -> str:
+        return self.__env
+
+    @property
+    def metadata(self) -> dict[str, Any]:
+        return self.__metadata
+
+    @property
+    def config_file(self) -> Path:
+        return self.__config_file
+
+    @cached_property
+    def python_version(self) -> tuple[int, int]:
+        import re
+
+        if match := re.search(r'^py(\d)\.(\d+)', self.env):
+            return int(match.group(1)), int(match.group(2))
+
+        from ddev.repo.constants import PYTHON_VERSION
+
+        major, minor = PYTHON_VERSION.split('.')
+        return int(major), int(minor)
+
+    def get_id(self) -> str:
+        return f'{self.integration.name}_{self.env}'
+
+    @abstractmethod
+    def start(self, *, agent_build: str, local_packages: dict[Path, str], env_vars: dict[str, str]) -> None:
+        ...
+
+    @abstractmethod
+    def stop(self) -> None:
+        ...
+
+    @abstractmethod
+    def restart(self) -> None:
+        ...
+
+    @abstractmethod
+    def invoke(self, args: list[str]) -> None:
+        ...
+
+    @abstractmethod
+    def enter_shell(self) -> None:
+        ...
diff --git a/ddev/src/ddev/e2e/config.py b/ddev/src/ddev/e2e/config.py
new file mode 100644
index 00000000000000..07e49e9e6b8bc8
--- /dev/null
+++ b/ddev/src/ddev/e2e/config.py
@@ -0,0 +1,99 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+import os
+from contextlib import suppress
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from ddev.utils.fs import Path
+
+
+class EnvData:
+    def __init__(self, storage_dir: Path):
+        self.__storage_dir = storage_dir
+
+    @property
+    def storage_dir(self) -> Path:
+        return self.__storage_dir
+
+    @property
+    def metadata_file(self) -> Path:
+        return self.storage_dir / 'metadata.json'
+
+    @property
+    def config_dir(self) -> Path:
+        return self.storage_dir / 'config'
+
+    @property
+    def config_file(self) -> Path:
+        return self.config_dir / f'{self.storage_dir.parent.name}.yaml'
+
+    def read_config(self) -> dict[str, Any]:
+        import yaml
+
+        return yaml.safe_load(self.config_file.read_text())
+
+    def write_config(self, config: dict[str, Any]) -> None:
+        import yaml
+
+        if 'instances' not in config:
+            config = {'instances': [config]}
+
+        self.config_file.parent.ensure_dir_exists()
+        self.config_file.write_text(yaml.safe_dump(config, default_flow_style=False))
+
+    def read_metadata(self) -> dict[str, Any]:
+        import json
+
+        return json.loads(self.metadata_file.read_text())
+
+    def write_metadata(self, metadata: dict[str, Any]) -> None:
+        import json
+
+        self.metadata_file.parent.ensure_dir_exists()
+        self.metadata_file.write_text(json.dumps(metadata, indent=2))
+
+    def exists(self) -> bool:
+        return self.storage_dir.is_dir()
+
+    def remove(self) -> None:
+        if self.config_file.is_file():
+            # Containers sometimes change the permissions of the directory,
+            # so we need to make sure we can remove it
+            self.config_file.parent.chmod(0o777)
+            self.config_file.chmod(0o777)
+
+            with suppress(Exception):
+                from getpass import getuser
+
+                os.chown(self.config_file, getuser(), -1)
+
+        self.storage_dir.remove()
+
+
+class EnvDataStorage:
+    def __init__(self, data_dir: Path):
+        self.__path = data_dir / 'env'
+
+    @property
+    def path(self) -> Path:
+        return self.__path
+
+    def get_integrations(self) -> list[str]:
+        if not self.path.is_dir():
+            return []
+
+        return sorted(path.name for path in self.path.iterdir() if path.is_dir())
+
+    def get_environments(self, integration: str) -> list[str]:
+        intg_path = self.path / integration
+        if not intg_path.is_dir():
+            return []
+
+        return sorted(path.name for path in intg_path.iterdir() if path.is_dir())
+
+    def get(self, integration: str, env: str) -> EnvData:
+        return EnvData(self.path / integration / env)
diff --git a/ddev/src/ddev/e2e/constants.py b/ddev/src/ddev/e2e/constants.py
new file mode 100644
index 00000000000000..92c875f13cf5f1
--- /dev/null
+++ b/ddev/src/ddev/e2e/constants.py
@@ -0,0 +1,14 @@
+ENV_VAR_PREFIX = 'DDEV_E2E_ENV_'
+
+
+class E2EEnvVars:
+    AGENT_BUILD = 'DDEV_E2E_AGENT'
+    AGENT_BUILD_PY2 = 'DDEV_E2E_AGENT_PY2'
+    SET_UP = 'DDEV_E2E_UP'
+    TEAR_DOWN = 'DDEV_E2E_DOWN'
+    PARENT_PYTHON = 'DDEV_E2E_PYTHON_PATH'
+    RESULT_FILE = 'DDEV_E2E_RESULT_FILE'
+    LOGS_DIR_PREFIX = 'DDEV_E2E_ENV_TEMP_DIR_DD_LOG_'
+
+
+DEFAULT_DOGSTATSD_PORT = 8125
diff --git a/ddev/src/ddev/e2e/run.py b/ddev/src/ddev/e2e/run.py
new file mode 100644
index 00000000000000..98284ea140d60f
--- /dev/null
+++ b/ddev/src/ddev/e2e/run.py
@@ -0,0 +1,49 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+import sys
+from contextlib import contextmanager
+from typing import Generator
+
+from ddev.e2e.constants import E2EEnvVars
+from ddev.utils.structures import EnvVars
+
+
+class E2EEnvironmentRunner:
+    def __init__(self, env: str):
+        self.__env = env
+
+    @contextmanager
+    def start(self) -> Generator[list[str], None, None]:
+        with EnvVars({E2EEnvVars.TEAR_DOWN: 'false'}):
+            yield self.base_command()
+
+    @contextmanager
+    def stop(self) -> Generator[list[str], None, None]:
+        with EnvVars({E2EEnvVars.SET_UP: 'false'}):
+            yield self.base_command()
+
+    def base_command(self) -> list[str]:
+        command = [
+            sys.executable,
+            '-m',
+            'hatch',
+            'env',
+            'run',
+            '--env',
+            self.__env,
+            '--',
+            'test',
+            '--capture=no',
+            '--disable-warnings',
+            '--exitfirst',
+            # We need -2 verbosity and by default the test command sets the verbosity to +2
+            '-qqqq',
+        ]
+        # TODO: always use this flag when we drop support for Python 2
+        if not self.__env.startswith('py2'):
+            command.append('--no-header')
+
+        return command
diff --git a/ddev/src/ddev/utils/platform.py b/ddev/src/ddev/utils/platform.py
index a20ad558961f7b..4f0dd0a896a3f4 100644
--- a/ddev/src/ddev/utils/platform.py
+++ b/ddev/src/ddev/utils/platform.py
@@ -129,6 +129,16 @@ def stream_process_output(process):
         for line in iter(process.stdout.readline, b''):
             yield line.decode('utf-8')
 
+    @property
+    def format_file_uri(self):
+        if self.__format_file_uri is None:
+            if self.windows:
+                self.__format_file_uri = lambda p: f'file:///{p}'.replace('\\', '/')
+            else:
+                self.__format_file_uri = lambda p: f'file://{p}'
+
+        return self.__format_file_uri
+
     @property
     def windows(self):
         """