diff --git a/src/rpdk/core/rqts/__init__.py b/src/rpdk/core/rqts/__init__.py
new file mode 100644
index 00000000..3e07489a
--- /dev/null
+++ b/src/rpdk/core/rqts/__init__.py
@@ -0,0 +1 @@
+"""RQTS (CTv2 local) contract test execution support for ``cfn test --v2``."""
diff --git a/src/rpdk/core/rqts/constants.py b/src/rpdk/core/rqts/constants.py
new file mode 100644
index 00000000..c35c2dc1
--- /dev/null
+++ b/src/rpdk/core/rqts/constants.py
@@ -0,0 +1,82 @@
+"""Constants for the RQTS (CTv2 local) contract test executor.
+
+These values are owned by the CLI and pin the behavior of ``cfn test --v2``:
+the RQTS Docker image reference, the executor extension selector, the image
+pull retry policy, the bind-mount root and output directory, the environment
+variables the executor reads, and the run summary messages.
+
+``cfn test --v2`` targets the published executor image's DirectJar handler
+mode, whose CLI contract is::
+
+    --extension contract-tests run-tests <artifact> --direct-jar -r <region> -o <output>
+
+Input subsetting (``-i``) is intentionally NOT emitted: the executor uses the
+inputs packaged inside the artifact zip. Scenario subsetting (``-s``) is NOT
+emitted either: scenario selection is owned by the executor image, which gates
+1P-oriented scenarios (``tagging-*``) by resource-type namespace, so the CLI
+passes no scenario list and no exclusions.
+"""
+
+# Fully-qualified, CLI-pinned RQTS image on the ECR Public Gallery.
+# ``s5r7m5i4`` is the permanent default registry alias of the publishing
+# account. The mutable ``latest`` tag is followed deliberately: the cloud
+# contract-test path always runs the latest published image, and the CLI keeps
+# parity by attempting a pull on every run (falling back to the local cache
+# when the registry is unreachable) instead of pinning per release.
+RQTS_IMAGE_REFERENCE = "public.ecr.aws/s5r7m5i4/cfn-rqts-executor-external:latest"
+
+# The executor top-level extension selector for contract tests. Passed as the
+# top-level ``--extension`` option BEFORE the ``run-tests`` subcommand.
+EXTENSION_CONTRACT_TESTS = "contract-tests"
+
+# Image pull retry policy.
+PULL_MAX_ATTEMPTS = 3
+PULL_ATTEMPT_TIMEOUT_SECONDS = 120
+
+# Credential environment variables the executor's ambient SDK credential chain
+# reads. Passed as ``key_names`` to ``get_temporary_credentials``, which zips
+# them positionally with (AccessKeyId, SecretAccessKey, SessionToken).
+ENV_CRED_KEYS = (
+    "AWS_ACCESS_KEY_ID",
+    "AWS_SECRET_ACCESS_KEY",
+    "AWS_SESSION_TOKEN",
+)
+
+# The same credential values under the names the executor repackages into the
+# handler request payload. Positionally aligned with ENV_CRED_KEYS.
+CALLER_ENV_CRED_KEYS = (
+    "CALLER_AWS_ACCESS_KEY_ID",
+    "CALLER_AWS_SECRET_ACCESS_KEY",
+    "CALLER_AWS_SESSION_TOKEN",
+)
+
+# The executor reads the type configuration as JSON content from this variable,
+# after the ``ct.typeConfiguration`` system property and before the
+# ``typeConfiguration`` packaged in the artifact's test inputs.
+ENV_TYPE_CONFIGURATION = "TYPE_CONFIGURATION"
+
+# No AWS_REGION: the executor takes the region from ``-r``, which configures
+# every SDK client it builds.
+
+# Container-internal bind-mount root.
+CONTAINER_WORKDIR = "/work"
+
+# Directory the executor writes its output to, relative to the project root on
+# the host. Created before the container runs so the bind-mounted path exists.
+HOST_OUTPUT_DIRNAME = "rqts-output"
+
+# The same directory addressed inside the container: under the bind mount, so
+# results are visible on the host.
+CONTAINER_OUTPUT_DIR = f"{CONTAINER_WORKDIR}/{HOST_OUTPUT_DIRNAME}"
+
+# Overall summary lines, phrased to match the pytest path's "One or more
+# contract tests failed" while naming the RQTS runner.
+# B105 false positive: a log summary string, not a password.
+PASS_MESSAGE = "RQTS contract tests passed"  # nosec B105
+FAIL_MESSAGE = "One or more RQTS contract tests failed"
+
+# NOTE: the CLI deliberately owns NO scenario set. The executor image decides
+# which scenarios run: capability conditions (taggable/creatable/...) plus
+# namespace gating that restricts the 1P-oriented ``tagging-*`` scenarios to
+# reserved first-party namespaces. 3P resources therefore get the non-tagging
+# subset automatically, with no scenario or exclusion flags from the CLI.
diff --git a/src/rpdk/core/rqts/image.py b/src/rpdk/core/rqts/image.py
new file mode 100644
index 00000000..3b7f89e5
--- /dev/null
+++ b/src/rpdk/core/rqts/image.py
@@ -0,0 +1,212 @@
+# have to skip B404, subprocess is required to drive the local Docker CLI
+# have to skip B603/B607, docker is invoked with a fixed, non-shell argv
+"""The RQTS (CTv2 local) executor image.
+
+:class:`RqtsImage` owns everything Docker-facing about ``cfn test --v2``: which
+image to run, making it available locally, and running the container.
+
+``cfn test --v2`` targets the executor's DirectJar handler mode::
+
+    --extension contract-tests run-tests /work/<artifact>.zip --direct-jar \
+        -r <region> -o <output-dir>
+
+DirectJar loads the handler JAR directly into the executor JVM, so there is no
+handler endpoint, no host networking, and no ``--handler-jar``/``-h``/``-tn``
+flags. Inputs are packaged in the artifact zip and resolved by the executor, so
+``-i`` is not emitted. Scenario selection is owned by the executor image
+(capability conditions plus namespace gating of the 1P-oriented ``tagging-*``
+scenarios), so no ``-s``/``--scenarios`` and no exclusion flags are emitted.
+
+Credential secrets never enter the command line: the ``-e`` flags are name-only
+and docker resolves each value from the client process environment, so the
+logged command is safe and invisible to ``ps``.
+"""
+import logging
+import os
+import subprocess  # nosec B404
+from pathlib import Path
+
+from rpdk.core.exceptions import SysExitRecommendedError
+
+from .constants import (
+    CALLER_ENV_CRED_KEYS,
+    CONTAINER_OUTPUT_DIR,
+    CONTAINER_WORKDIR,
+    ENV_CRED_KEYS,
+    ENV_TYPE_CONFIGURATION,
+    EXTENSION_CONTRACT_TESTS,
+    HOST_OUTPUT_DIRNAME,
+    PULL_ATTEMPT_TIMEOUT_SECONDS,
+    PULL_MAX_ATTEMPTS,
+    RQTS_IMAGE_REFERENCE,
+)
+
+LOG = logging.getLogger(__name__)
+
+
+class RqtsImage:
+    """The RQTS executor image: which one, whether it is here, and running it."""
+
+    def __init__(self, reference=None):
+        """Store the effective image reference.
+
+        :param reference: image reference override, or ``None`` for the
+            CLI-pinned :data:`RQTS_IMAGE_REFERENCE`.
+        """
+        self.reference = reference or RQTS_IMAGE_REFERENCE
+
+    @staticmethod
+    def _run_docker(args, timeout=None, env=None, capture=True):
+        """Run ``docker`` with a fixed, non-shell argv.
+
+        The single place a docker command line is assembled and invoked.
+
+        :param list args: arguments following ``docker``
+        :param timeout: optional timeout in seconds
+        :param env: complete environment for the child, or ``None`` to inherit
+            the ambient one. Left unset for pull and inspect so no credential
+            can reach them.
+        :param bool capture: capture stdout/stderr. ``False`` lets the child
+            inherit the parent's terminal and stream output live as it is
+            produced rather than buffering to completion (Requirement 5.1).
+        :rtype: subprocess.CompletedProcess
+        """
+        argv = ["docker", *args]
+        LOG.debug("Running: %s", " ".join(argv))
+        return subprocess.run(  # nosec B603 B607
+            argv,
+            check=False,
+            capture_output=capture,
+            timeout=timeout,
+            env=env,
+        )
+
+    def ensure(self):
+        """Pull the image, falling back to a locally cached copy.
+
+        A pull is attempted on every run so a moved mutable tag (``latest``) is
+        picked up without a manual ``docker pull``; an up-to-date image costs
+        only a manifest check. The pull is anonymous - the image is published to
+        the ECR Public Gallery, so no AWS credentials are ever supplied to it.
+
+        :raises SysExitRecommendedError: if every pull attempt fails and the
+            image is not in the local image store
+        """
+        for attempt in range(1, PULL_MAX_ATTEMPTS + 1):
+            LOG.info("Pulling RQTS image %s (attempt %d)", self.reference, attempt)
+            try:
+                completed = self._run_docker(
+                    ["pull", self.reference], timeout=PULL_ATTEMPT_TIMEOUT_SECONDS
+                )
+            except (OSError, subprocess.SubprocessError) as err:
+                error = err
+            else:
+                if completed.returncode == 0:
+                    return
+                error = (completed.stderr or b"").decode("utf-8", "replace").strip()
+            LOG.debug(
+                "Pull attempt %d for %s failed: %s", attempt, self.reference, error
+            )
+
+        try:
+            cached = (
+                self._run_docker(["image", "inspect", self.reference]).returncode == 0
+            )
+        except (OSError, subprocess.SubprocessError) as err:
+            LOG.debug("Local image inspect for %s failed: %s", self.reference, err)
+            cached = False
+
+        if cached:
+            LOG.warning(
+                "Could not pull the RQTS image using the locally cached "
+                "copy (%s), which may be stale",
+                self.reference,
+            )
+            return
+
+        raise SysExitRecommendedError(
+            f"Failed to pull the RQTS image '{self.reference}' after "
+            f"{PULL_MAX_ATTEMPTS} attempts: {error}"
+        )
+
+    def build_run_args(self, project, region, creds, type_configuration=None):
+        """Build the docker ``run`` arguments and the child environment.
+
+        Side-effect free: no subprocess, filesystem, network or AWS calls. The
+        ``-e`` flags are name-only, so the returned arguments carry no secret;
+        the values travel in the returned environment only.
+
+        Composition::
+
+            run --rm
+            -e <each environment variable, by name only>
+            -v <project.root>:/work
+            <reference>
+            --extension contract-tests run-tests /work/<artifact>.zip
+            --direct-jar -r <region> -o /work/rqts-output
+
+        :param project: the loaded :class:`rpdk.core.project.Project`. ``root``
+            is the bind-mount source and ``hypenated_name`` names the artifact.
+        :param str region: the effective AWS region, passed via ``-r``.
+        :param creds: minted temporary credentials keyed by
+            :data:`~rpdk.core.rqts.constants.ENV_CRED_KEYS`.
+        :param type_configuration: the type configuration as a JSON string, or
+            ``None`` to leave the variable unset so the executor falls back to
+            the ``typeConfiguration`` packaged in the artifact.
+        :returns: an ``(args, env)`` tuple.
+        """
+        # The executor reads the same credentials under both sets of names: the
+        # ambient SDK chain uses AWS_*, and CALLER_AWS_* is what it repackages
+        # into the handler request payload.
+        env = dict(creds)
+        env.update(zip(CALLER_ENV_CRED_KEYS, (creds[key] for key in ENV_CRED_KEYS)))
+        if type_configuration is not None:
+            env[ENV_TYPE_CONFIGURATION] = type_configuration
+
+        args = ["run", "--rm"]
+        for name in env:
+            args += ["-e", name]
+        # Bind mount the project root so the artifact zip is readable in-container.
+        args += ["-v", f"{project.root}:{CONTAINER_WORKDIR}", self.reference]
+        # The executor command: top-level --extension selector, then the
+        # run-tests subcommand with the positional artifact path and DirectJar.
+        args += [
+            "--extension",
+            EXTENSION_CONTRACT_TESTS,
+            "run-tests",
+            f"{CONTAINER_WORKDIR}/{project.hypenated_name}.zip",
+            "--direct-jar",
+            "-r",
+            region,
+            "-o",
+            CONTAINER_OUTPUT_DIR,
+        ]
+        return args, env
+
+    def run(self, project, region, creds, type_configuration=None):
+        """Ensure the image, run the container, and return its exit code.
+
+        Blocks until the container exits, streaming its output live.
+
+        :rtype: int
+        :raises SysExitRecommendedError: if the image is unavailable or the
+            container process cannot be spawned
+        """
+        self.ensure()
+
+        # Create the host side of the bind-mounted output directory so the
+        # container's -o path exists when docker mounts it.
+        (Path(project.root) / HOST_OUTPUT_DIRNAME).mkdir(parents=True, exist_ok=True)
+
+        args, env = self.build_run_args(
+            project, region, creds, type_configuration=type_configuration
+        )
+        try:
+            return self._run_docker(
+                args, env={**os.environ, **env}, capture=False
+            ).returncode
+        except OSError as err:
+            LOG.debug("Failed to start the RQTS container", exc_info=err)
+            raise SysExitRecommendedError(
+                f"the RQTS container could not be started: {err}"
+            ) from err
diff --git a/src/rpdk/core/rqts/preconditions.py b/src/rpdk/core/rqts/preconditions.py
new file mode 100644
index 00000000..1351e977
--- /dev/null
+++ b/src/rpdk/core/rqts/preconditions.py
@@ -0,0 +1,119 @@
+"""Precondition checks for the RQTS (CTv2 local) contract test runner.
+
+``cfn test --v2`` requires several runtime and project prerequisites before the
+RQTS container can run: a working Docker runtime, a built artifact package, and
+valid AWS credentials and region.
+
+The DirectJar handler mode loads the handler JAR directly into the executor
+JVM, so there is no SAM Local handler endpoint to probe and no separate input
+resolution: inputs are packaged inside the artifact zip and read from there by
+the executor.
+
+Each check in this module is independent and side-effect-free with respect to
+the others. A check appends a single human-readable message on failure and
+NEVER raises, so the caller (``RqtsRunner``) can aggregate every unmet
+precondition into one error rather than failing on the first problem
+(Requirement 3.7). ``check_preconditions`` returns the aggregated list of
+failure messages; an empty list means all preconditions are met.
+"""
+
+import logging
+import shutil
+import subprocess  # nosec B404
+
+from ..boto_helpers import create_sdk_session
+
+LOG = logging.getLogger(__name__)
+
+# Bounded timeout (seconds) for the Docker daemon ping so a hung daemon cannot
+# stall the precondition phase.
+_DOCKER_INFO_TIMEOUT_SECONDS = 10
+
+
+def _check_docker():
+    """Return a failure message if Docker is unavailable, else ``None``.
+
+    Docker is available only when the ``docker`` CLI is present on PATH AND the
+    Docker daemon is reachable (probed with ``docker info``) (Requirement 3.2).
+    """
+    if shutil.which("docker") is None:
+        return (
+            "Docker is required and must be running: the 'docker' CLI was not "
+            "found on PATH."
+        )
+
+    try:
+        result = subprocess.run(  # nosec B603, B607
+            ["docker", "info"],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            timeout=_DOCKER_INFO_TIMEOUT_SECONDS,
+            check=False,
+        )
+    except (OSError, subprocess.SubprocessError) as e:
+        LOG.debug("Docker daemon ping failed", exc_info=e)
+        return (
+            "Docker is required and must be running: the Docker daemon could "
+            "not be reached."
+        )
+
+    if result.returncode != 0:
+        return (
+            "Docker is required and must be running: the Docker daemon could "
+            "not be reached."
+        )
+
+    return None
+
+
+def _check_artifact(project):
+    """Return a failure message if the built artifact package is missing.
+
+    Mirrors ``Project._get_zip_file_path()``: the package lives at
+    ``project.root / f"{project.hypenated_name}.zip"`` (Requirement 3.3).
+    """
+    artifact_name = f"{project.hypenated_name}.zip"
+    artifact_path = project.root / artifact_name
+    if not artifact_path.is_file():
+        return (
+            f"artifact package '{artifact_name}' not found; build the project " "first."
+        )
+    return None
+
+
+def _check_credentials(args):
+    """Return a failure message if AWS credentials or region are unavailable.
+
+    Reuses ``boto_helpers.create_sdk_session`` which raises when the region or
+    credentials are missing; that exception is caught and converted to a failure
+    message so the check never raises (Requirement 3.5).
+    """
+    try:
+        create_sdk_session(args.region, args.profile)
+    except Exception as e:  # pylint: disable=broad-except
+        LOG.debug("AWS session could not be created", exc_info=e)
+        return "valid AWS credentials and a region are required."
+    return None
+
+
+def check_preconditions(args, project):
+    """Verify all ``cfn test --v2`` preconditions and aggregate any failures.
+
+    Runs each independent precondition check and collects the human-readable
+    failure message from every unmet one. No check raises; the caller turns a
+    non-empty list into a single ``SysExitRecommendedError`` (Requirements 3.1,
+    3.7).
+
+    :param args: parsed CLI arguments (uses ``args.region`` and ``args.profile``)
+    :param project: loaded ``rpdk.core.project.Project``
+    :returns: list of failure messages; empty means all preconditions are met
+    """
+    failures = []
+    for message in (
+        _check_docker(),
+        _check_artifact(project),
+        _check_credentials(args),
+    ):
+        if message is not None:
+            failures.append(message)
+    return failures
diff --git a/src/rpdk/core/rqts/runner.py b/src/rpdk/core/rqts/runner.py
new file mode 100644
index 00000000..0d36efa3
--- /dev/null
+++ b/src/rpdk/core/rqts/runner.py
@@ -0,0 +1,126 @@
+"""RQTS (CTv2 local) contract test runner.
+
+:class:`RqtsRunner` orchestrates the ``cfn test --v2`` pipeline: it guards the
+project artifact type, aggregates and enforces preconditions, mints temporary
+AWS credentials, resolves the type configuration, then hands off to
+:class:`~rpdk.core.rqts.image.RqtsImage`, which owns everything Docker-facing.
+
+A zero container exit code logs the pass message and returns so the CLI exits
+``0``; any non-zero code raises
+:class:`~rpdk.core.exceptions.SysExitRecommendedError`, which ``cli.py`` maps to
+``SystemExit(1)``. The per-scenario outcomes are streamed live by the container,
+so the summary neither re-parses nor duplicates them.
+"""
+
+import json
+import logging
+
+from rpdk.core.exceptions import SysExitRecommendedError
+
+from ..boto_helpers import create_sdk_session, get_temporary_credentials
+from ..contract.type_configuration import TypeConfiguration
+from ..project import ARTIFACT_TYPE_HOOK, ARTIFACT_TYPE_RESOURCE
+from .constants import ENV_CRED_KEYS, FAIL_MESSAGE, PASS_MESSAGE
+from .image import RqtsImage
+from .preconditions import check_preconditions
+
+LOG = logging.getLogger(__name__)
+
+
+class RqtsRunner:
+    """Orchestrates the ``cfn test --v2`` RQTS pipeline.
+
+    A single instance owns the parsed CLI ``args`` and the loaded
+    :class:`~rpdk.core.project.Project` and drives the fixed pipeline in
+    :meth:`run`: artifact-type guard, precondition aggregation, credential
+    minting, type configuration resolution, the container run (delegated to
+    :class:`~rpdk.core.rqts.image.RqtsImage`), and exit-code mapping.
+
+    Module projects are short-circuited upstream in ``test()`` before the runner
+    is constructed, so this class only handles resource (the supported case),
+    hook, and indeterminate artifact types.
+    """
+
+    def __init__(self, args, project):
+        """Store the parsed CLI arguments and the loaded project.
+
+        :param args: parsed CLI arguments (an argparse ``Namespace``). The
+            runner reads ``region``, ``profile``, ``role_arn``,
+            ``source_account``, ``source_arn``, ``typeconfig`` and
+            ``rqts_image``.
+        :param project: the loaded :class:`~rpdk.core.project.Project`.
+        """
+        self.args = args
+        self.project = project
+
+    def _guard_artifact_type(self):
+        """Fail fast unless the project is a supported resource type.
+
+        Hook projects are unsupported by the RQTS local runner (Requirement
+        7.2); any artifact type that is neither a resource nor a hook is treated
+        as indeterminate (Requirement 7.5). Module projects are handled upstream
+        in ``test()`` and never reach this method.
+
+        :raises SysExitRecommendedError: for hook or indeterminate artifact
+            types
+        """
+        artifact_type = self.project.artifact_type
+        if artifact_type == ARTIFACT_TYPE_HOOK:
+            raise SysExitRecommendedError(
+                "the RQTS local test runner supports resource types only"
+            )
+        if artifact_type != ARTIFACT_TYPE_RESOURCE:
+            raise SysExitRecommendedError(
+                "could not determine the project artifact type"
+            )
+
+    def run(self):
+        """Orchestrate the full ``--v2`` pipeline.
+
+        Raises :class:`~rpdk.core.exceptions.SysExitRecommendedError` on any
+        failure (guard, preconditions, image pull, container start, or a
+        non-zero container exit code); returns normally when every RQTS contract
+        test passes.
+        """
+        self._guard_artifact_type()
+
+        failures = check_preconditions(self.args, self.project)
+        if failures:
+            raise SysExitRecommendedError(
+                "cannot run 'cfn test --v2'; the following preconditions were "
+                "not met:\n" + "\n".join(f"  - {failure}" for failure in failures)
+            )
+
+        # Temporary credentials for the container, keyed by the environment
+        # variables the executor reads (Requirement 3.5).
+        session = create_sdk_session(self.args.region, self.args.profile)
+        creds = get_temporary_credentials(
+            session,
+            ENV_CRED_KEYS,
+            self.args.role_arn,
+            headers={
+                "account_id": self.args.source_account,
+                "source_arn": self.args.source_arn,
+            },
+        )
+
+        # TypeConfiguration memoizes the parsed file on a class attribute with no
+        # invalidation, so clear it to guarantee this run reads from disk. A
+        # missing file leaves the executor's variable unset, so it falls back to
+        # the typeConfiguration packaged in the artifact.
+        TypeConfiguration.TYPE_CONFIGURATION = None
+        type_configuration = TypeConfiguration.get_type_configuration(
+            self.args.typeconfig
+        )
+
+        exit_code = RqtsImage(self.args.rqts_image).run(
+            self.project,
+            self.args.region,
+            creds,
+            type_configuration=(
+                json.dumps(type_configuration) if type_configuration else None
+            ),
+        )
+        if exit_code != 0:
+            raise SysExitRecommendedError(FAIL_MESSAGE)
+        LOG.info(PASS_MESSAGE)
diff --git a/src/rpdk/core/test.py b/src/rpdk/core/test.py
index e0f909c3..83a23fe2 100644
--- a/src/rpdk/core/test.py
+++ b/src/rpdk/core/test.py
@@ -418,6 +418,13 @@ def test(args):
         LOG.warning("The test command is not supported in a module project")
         return
 
+    if args.v2:
+        # local import keeps the pytest path import-light
+        from .rqts.runner import RqtsRunner  # pylint: disable=import-outside-toplevel
+
+        RqtsRunner(args, project).run()
+        return
+
     if project.artifact_type == ARTIFACT_TYPE_HOOK:
         overrides = get_hook_overrides(
             project.root,
@@ -545,6 +552,25 @@ def setup_subparser(subparsers, parents):
         help="Source Type Version Arn key used for Assume Role to Run Contract Tests",
     )
 
+    parser.add_argument(
+        "--v2",
+        action="store_true",
+        default=False,
+        help=(
+            "Opt-in: run the RQTS local test runner (CTv2) in a Docker container "
+            "instead of the default pytest-based contract tests."
+        ),
+    )
+
+    parser.add_argument(
+        "--rqts-image",
+        default=None,
+        help=(
+            "Override the RQTS container image reference (for testing or pre-release "
+            "images). Defaults to the CLI-pinned public.ecr.aws image."
+        ),
+    )
+
 
 def _sam_arguments(parser):
     parser.add_argument(
diff --git a/tests/rqts/__init__.py b/tests/rqts/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/rqts/test_image.py b/tests/rqts/test_image.py
new file mode 100644
index 00000000..8003b799
--- /dev/null
+++ b/tests/rqts/test_image.py
@@ -0,0 +1,715 @@
+"""Tests for :class:`rpdk.core.rqts.image.RqtsImage`.
+
+Covers image resolution (Property 1), the bounded anonymous pull with its
+cached-image fallback (Property 2), the ``docker run`` argument/environment
+construction (Properties 4-12), and running the container.
+
+Docker is never actually invoked: the single ``_run_docker`` seam is patched so
+the retry policy and the container run can be driven deterministically, and the
+one test that reaches ``subprocess.run`` patches it.
+
+Library: Hypothesis (the standard Python property-based testing library). Each
+property test runs at least 100 generated examples via
+``@settings(max_examples=100)`` and is tagged with a comment referencing the
+design property it validates.
+"""
+
+import json
+import subprocess
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+from hypothesis import given, settings, strategies as st
+
+from rpdk.core.exceptions import SysExitRecommendedError
+from rpdk.core.rqts import image as image_module
+from rpdk.core.rqts.constants import (
+    CALLER_ENV_CRED_KEYS,
+    CONTAINER_OUTPUT_DIR,
+    CONTAINER_WORKDIR,
+    ENV_CRED_KEYS,
+    ENV_TYPE_CONFIGURATION,
+    EXTENSION_CONTRACT_TESTS,
+    HOST_OUTPUT_DIRNAME,
+    PULL_MAX_ATTEMPTS,
+    RQTS_IMAGE_REFERENCE,
+)
+from rpdk.core.rqts.image import RqtsImage
+
+# The CLI default region: ``cfn test`` defines ``--region`` with this default, so
+# the effective region is always populated even when the user omits ``--region``.
+CLI_DEFAULT_REGION = "us-east-1"
+
+# A marker used to build distinctive, searchable fake credential values, made of
+# characters that never appear in generated image references, regions or names,
+# so a secret can be confirmed absent anywhere in the docker arguments.
+_SECRET_MARKER = "AWSSECRETMARKER"
+
+# Credential environment variable names that must never be threaded into an
+# anonymous ``docker pull``.
+_AWS_CRED_TOKENS = ENV_CRED_KEYS + CALLER_ENV_CRED_KEYS + (_SECRET_MARKER,)
+
+# Flags the DirectJar contract must NEVER emit (they belong to the old SAM Local
+# / remote-lambda shapes).
+FORBIDDEN_FLAGS = ("--handler-jar", "-h", "-tn", "--sam-local", "--remote-lambda")
+
+# The tagging scenarios that ``cfn test --v2`` must NEVER run.
+FORBIDDEN_TAGGING_SCENARIOS = (
+    "tagging-oob",
+    "tagging-permission",
+    "tagging-stack",
+    "tagging-system",
+)
+
+# Fixed credentials for the tests that do not exercise credential handling.
+CREDS = dict(zip(ENV_CRED_KEYS, ("AKID", "SECRET", "TOKEN")))
+
+_IMAGE_ALPHABET = "abcdefghijklmnopqrstuvwxyz0123456789./:-"
+_SEGMENT_ALPHABET = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+
+# Image references, including the empty string, so resolution covers both the
+# override and the pinned-default cases.
+image_refs = st.text(alphabet=_IMAGE_ALPHABET, min_size=1, max_size=60)
+
+_segments = st.text(alphabet=_SEGMENT_ALPHABET, min_size=1, max_size=12)
+
+_KNOWN_REGIONS = [
+    CLI_DEFAULT_REGION,
+    "us-west-2",
+    "eu-west-1",
+    "ap-southeast-2",
+    "eu-central-1",
+]
+regions = st.one_of(
+    st.sampled_from(_KNOWN_REGIONS),
+    st.text(alphabet=_SEGMENT_ALPHABET + "-", min_size=1, max_size=20),
+)
+
+# Distinctive, searchable secret values.
+_secret_values = st.text(alphabet=_SEGMENT_ALPHABET, min_size=6, max_size=24).map(
+    lambda body: _SECRET_MARKER + body
+)
+
+# Project root paths for the bind mount.
+roots = st.text(alphabet=_SEGMENT_ALPHABET + "/", min_size=1, max_size=40).map(
+    lambda p: "/" + p.strip("/")
+)
+
+
+@st.composite
+def hyphenated_names(draw):
+    """Generate realistic ``org-service-resource`` style hyphenated names."""
+    parts = draw(st.lists(_segments, min_size=3, max_size=3))
+    return "-".join(parts).lower()
+
+
+@st.composite
+def credentials(draw):
+    """Generate a credentials dict keyed by the executor's env var names."""
+    return {key: draw(_secret_values) for key in ENV_CRED_KEYS}
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_project(hypenated_name, root="/project/root"):
+    """Lightweight stand-in for a loaded Project (only .hypenated_name/.root used)."""
+    return SimpleNamespace(hypenated_name=hypenated_name, root=root)
+
+
+def _completed(args, returncode):
+    return subprocess.CompletedProcess(
+        args=["docker", *args],
+        returncode=returncode,
+        stdout=b"",
+        stderr=b"" if returncode == 0 else b"boom",
+    )
+
+
+class _RecordingDocker:
+    """A fake ``_run_docker`` that records calls and replays fixed results.
+
+    ``pull_results`` queues one outcome per ``docker pull``: a ``returncode``
+    int, or an exception instance to raise (modelling a timeout or a spawn
+    failure). ``inspect_result`` is the outcome of the cached-image check, and
+    ``run_returncode`` the container's exit code.
+
+    Patched in as a plain instance rather than a function, so attribute lookup
+    on the class does not bind ``self``.
+    """
+
+    def __init__(self, pull_results=(), inspect_result=1, run_returncode=0):
+        self._pull_results = list(pull_results)
+        self._inspect_result = inspect_result
+        self._run_returncode = run_returncode
+        self.calls = []
+
+    def __call__(self, args, timeout=None, env=None, capture=True):
+        self.calls.append(
+            SimpleNamespace(args=list(args), timeout=timeout, env=env, capture=capture)
+        )
+        if args[0] == "pull":
+            result = self._pull_results[len(self.pull_calls) - 1]
+            if isinstance(result, BaseException):
+                raise result
+            return _completed(args, result)
+        if args[:2] == ["image", "inspect"]:
+            if isinstance(self._inspect_result, BaseException):
+                raise self._inspect_result
+            return _completed(args, self._inspect_result)
+        return _completed(args, self._run_returncode)
+
+    @property
+    def pull_calls(self):
+        return [call for call in self.calls if call.args[0] == "pull"]
+
+    @property
+    def run_calls(self):
+        return [call for call in self.calls if call.args[0] == "run"]
+
+
+def _patch_docker(fake):
+    return mock.patch.object(RqtsImage, "_run_docker", fake)
+
+
+# ===========================================================================
+# Property 1: image resolution
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 1: Image resolution honors override, else pinned default
+@settings(max_examples=100)
+@given(override=st.one_of(st.none(), st.just(""), image_refs))
+def test_property_1_image_resolution_override_else_default(override):
+    """The reference is the override when non-empty, else the pinned default.
+
+    Validates: Requirements 2.1, 2.2
+    """
+    reference = RqtsImage(override).reference
+
+    if override:
+        assert reference == override
+    else:
+        assert reference == RQTS_IMAGE_REFERENCE
+
+
+def test_no_reference_uses_default():
+    """Constructed with no argument at all, the pinned default is used."""
+    assert RqtsImage().reference == RQTS_IMAGE_REFERENCE
+
+
+# ===========================================================================
+# Property 2: bounded, anonymous pull
+# ===========================================================================
+def _expected_attempts(results):
+    """Model the bounded retry policy: attempt until first success, capped."""
+    for index, result in enumerate(results[:PULL_MAX_ATTEMPTS]):
+        if result == 0:
+            return index + 1, True
+    return min(len(results), PULL_MAX_ATTEMPTS), False
+
+
+@st.composite
+def pull_result_sequences(draw):
+    """Generate a sequence of per-attempt pull outcomes.
+
+    Each element is a ``0`` (success), a non-zero return code (failure), or an
+    exception instance (timeout / spawn error). The sequence is padded to at
+    least ``PULL_MAX_ATTEMPTS`` so ``ensure`` always has an outcome to consume
+    even when every attempt fails.
+    """
+    outcome = st.one_of(
+        st.just(0),
+        st.integers(min_value=1, max_value=255),
+        st.just(subprocess.TimeoutExpired(cmd="docker pull", timeout=1)),
+        st.just(OSError("docker not found")),
+    )
+    return draw(
+        st.lists(outcome, min_size=PULL_MAX_ATTEMPTS, max_size=PULL_MAX_ATTEMPTS + 4)
+    )
+
+
+# Feature: cfn-test-v2-flag, Property 2: Image pull is bounded and anonymous
+@settings(max_examples=100)
+@given(results=pull_result_sequences(), image_ref=image_refs, cached=st.booleans())
+def test_property_2_pull_is_bounded_and_anonymous(results, image_ref, cached):
+    """``ensure`` pulls at most PULL_MAX_ATTEMPTS regardless of the local cache
+    state, stops on first success, never supplies AWS credentials to the pull,
+    falls back to a cached image on exhaustion, and raises only when no cached
+    copy exists.
+
+    Validates: Requirements 2.3, 2.4, 2.5, 2.6
+    """
+    fake_docker = _RecordingDocker(results, inspect_result=0 if cached else 1)
+    expected_attempts, should_succeed = _expected_attempts(results)
+
+    # ``mock.patch`` context managers are used (rather than the monkeypatch
+    # fixture) so the patches are applied and reset for every generated example.
+    with _patch_docker(fake_docker):
+        if should_succeed or cached:
+            RqtsImage(image_ref).ensure()
+        else:
+            with pytest.raises(SysExitRecommendedError):
+                RqtsImage(image_ref).ensure()
+
+    # Bounded: pull is attempted at most PULL_MAX_ATTEMPTS times, and exactly the
+    # number of attempts the retry policy predicts (stopping on first success).
+    assert len(fake_docker.pull_calls) <= PULL_MAX_ATTEMPTS
+    assert len(fake_docker.pull_calls) == expected_attempts
+
+    for call in fake_docker.pull_calls:
+        # Every invocation is an anonymous ``docker pull <ref>`` for this image.
+        assert call.args == ["pull", image_ref]
+        # The pull inherits the ambient environment: no credential is threaded
+        # in, positionally or by keyword.
+        assert call.env is None
+        for token in call.args:
+            for cred in _AWS_CRED_TOKENS:
+                assert cred not in token
+
+
+def test_pull_exhaustion_without_cached_image_raises():
+    """All attempts fail with no cached image -> SysExitRecommendedError after
+    exactly PULL_MAX_ATTEMPTS attempts.
+
+    Validates: Requirements 2.6
+    """
+    fake_docker = _RecordingDocker([1] * (PULL_MAX_ATTEMPTS + 2), inspect_result=1)
+
+    with _patch_docker(fake_docker):
+        with pytest.raises(SysExitRecommendedError) as excinfo:
+            RqtsImage("some/image:tag").ensure()
+
+    assert len(fake_docker.pull_calls) == PULL_MAX_ATTEMPTS
+    assert "some/image:tag" in str(excinfo.value)
+
+
+def test_pull_exhaustion_with_cached_image_falls_back_with_warning(caplog):
+    """All attempts fail but the image is cached locally -> warn and run from the
+    local store instead of raising.
+
+    Validates: Requirements 2.5
+    """
+    fake_docker = _RecordingDocker([1] * PULL_MAX_ATTEMPTS, inspect_result=0)
+
+    with _patch_docker(fake_docker):
+        with caplog.at_level("WARNING", logger=image_module.__name__):
+            RqtsImage("cached/image:tag").ensure()
+
+    assert len(fake_docker.pull_calls) == PULL_MAX_ATTEMPTS
+    assert fake_docker.calls[-1].args == ["image", "inspect", "cached/image:tag"]
+    assert any(
+        "cached" in record.getMessage() and "cached/image:tag" in record.getMessage()
+        for record in caplog.records
+    )
+
+
+def test_pull_exhaustion_on_repeated_timeouts():
+    """Repeated per-attempt timeouts also exhaust after exactly PULL_MAX_ATTEMPTS."""
+    timeouts = [
+        subprocess.TimeoutExpired(cmd="docker pull", timeout=1)
+        for _ in range(PULL_MAX_ATTEMPTS + 1)
+    ]
+    fake_docker = _RecordingDocker(timeouts, inspect_result=1)
+
+    with _patch_docker(fake_docker):
+        with pytest.raises(SysExitRecommendedError):
+            RqtsImage("some/image:tag").ensure()
+
+    assert len(fake_docker.pull_calls) == PULL_MAX_ATTEMPTS
+
+
+def test_ensure_pulls_even_when_present_locally():
+    """``ensure`` always attempts the pull, so a moved mutable tag (e.g. latest)
+    is refreshed without a manual docker pull, and no inspect is needed on
+    success.
+
+    Validates: Requirements 2.3
+    """
+    fake_docker = _RecordingDocker([0], inspect_result=0)
+
+    with _patch_docker(fake_docker):
+        RqtsImage("present/image:tag").ensure()
+
+    assert fake_docker.calls == [
+        SimpleNamespace(
+            args=["pull", "present/image:tag"],
+            timeout=fake_docker.calls[0].timeout,
+            env=None,
+            capture=True,
+        )
+    ]
+    # The per-attempt timeout is bounded so a hung pull cannot stall the run.
+    assert fake_docker.pull_calls[0].timeout is not None
+
+
+@pytest.mark.parametrize(
+    "docker_error", [OSError("docker not found"), subprocess.SubprocessError("boom")]
+)
+def test_inspect_that_cannot_run_is_treated_as_absent(docker_error):
+    """An inspect that cannot even run is treated as 'not cached'.
+
+    Keeps the fallback decision safe when the docker CLI is missing or unusable:
+    absent, rather than assumed cached.
+    """
+    fake_docker = _RecordingDocker([1] * PULL_MAX_ATTEMPTS, inspect_result=docker_error)
+
+    with _patch_docker(fake_docker):
+        with pytest.raises(SysExitRecommendedError):
+            RqtsImage("ref:tag").ensure()
+
+
+# ===========================================================================
+# The docker seam
+# ===========================================================================
+def test_run_docker_invokes_docker_cli_with_fixed_argv():
+    """``_run_docker`` drives the docker CLI with a fixed, non-shell argv.
+
+    This is the one place that actually reaches ``subprocess.run``; it is
+    patched here so no docker process is spawned. A fixed argv list (never a
+    shell string) is what makes the image reference safe to pass through.
+    """
+    completed = subprocess.CompletedProcess(args=["docker", "info"], returncode=0)
+
+    with mock.patch.object(
+        image_module.subprocess, "run", return_value=completed
+    ) as run:
+        result = RqtsImage._run_docker(  # pylint: disable=protected-access
+            ["image", "inspect", "ref:tag"], timeout=7
+        )
+
+    assert result is completed
+    run.assert_called_once()
+    assert run.call_args[0][0] == ["docker", "image", "inspect", "ref:tag"]
+    assert run.call_args[1]["check"] is False
+    assert run.call_args[1]["capture_output"] is True
+    assert run.call_args[1]["timeout"] == 7
+    # No env by default: the child inherits the ambient environment.
+    assert run.call_args[1]["env"] is None
+
+
+def test_run_docker_streams_when_capture_disabled():
+    """``capture=False`` leaves stdout/stderr uncaptured so the child inherits
+    the terminal and output streams live.
+
+    Validates: Requirements 5.1
+    """
+    completed = subprocess.CompletedProcess(args=["docker", "run"], returncode=0)
+
+    with mock.patch.object(
+        image_module.subprocess, "run", return_value=completed
+    ) as run:
+        RqtsImage._run_docker(  # pylint: disable=protected-access
+            ["run", "--rm", "img"], env={"A": "B"}, capture=False
+        )
+
+    assert run.call_args[1]["capture_output"] is False
+    assert run.call_args[1]["env"] == {"A": "B"}
+
+
+# ===========================================================================
+# Property 4
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 4: docker run bind-mounts the project root onto /work
+@settings(max_examples=100)
+@given(name=hyphenated_names(), region=regions, root=roots)
+def test_property_4_docker_run_bind_mounts_project_root(name, region, root):
+    args, _env = RqtsImage("img:ref").build_run_args(
+        _make_project(name, root=root), region, CREDS
+    )
+
+    assert args[:2] == ["run", "--rm"]
+    assert "-v" in args
+    assert args[args.index("-v") + 1] == f"{root}:{CONTAINER_WORKDIR}"
+
+
+# ===========================================================================
+# Property 5
+# ===========================================================================
+def _env_entries(args):
+    """Return the token following each ``-e`` flag."""
+    return [args[i + 1] for i, token in enumerate(args) if token == "-e"]
+
+
+# Feature: cfn-test-v2-flag, Property 5: credential values never enter the command
+# line; they travel only through the returned env mapping
+@settings(max_examples=100)
+@given(name=hyphenated_names(), region=regions, creds=credentials())
+def test_property_5_credentials_only_in_env_never_in_args(name, region, creds):
+    args, env = RqtsImage("img:ref").build_run_args(_make_project(name), region, creds)
+
+    # The -e flags are NAME-ONLY: the variable names are referenced so docker
+    # inherits them from the client process environment; no `=` and no values.
+    assert _env_entries(args) == list(ENV_CRED_KEYS) + list(CALLER_ENV_CRED_KEYS)
+
+    # No credential secret value appears ANYWHERE in the arguments (this is what
+    # makes the logged command line and `ps` output safe).
+    for token in args:
+        for secret in creds.values():
+            assert secret not in token
+
+    # The same credentials are exported under both name sets: the ambient SDK
+    # chain reads AWS_*, and the executor repackages CALLER_AWS_* into the
+    # handler request payload.
+    for key, caller_key in zip(ENV_CRED_KEYS, CALLER_ENV_CRED_KEYS):
+        assert env[key] == creds[key]
+        assert env[caller_key] == creds[key]
+
+    # The region reaches the executor through -r only, never as AWS_REGION.
+    assert "AWS_REGION" not in env
+
+
+# ===========================================================================
+# Property 6
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 6: run-tests receives the positional artifact path inside the mount
+@settings(max_examples=100)
+@given(name=hyphenated_names(), region=regions)
+def test_property_6_extension_run_tests_and_artifact_path(name, region):
+    args, _env = RqtsImage("img:ref").build_run_args(_make_project(name), region, CREDS)
+
+    # The top-level extension selector precedes the run-tests subcommand.
+    assert "--extension" in args
+    ext_index = args.index("--extension")
+    assert args[ext_index + 1] == EXTENSION_CONTRACT_TESTS
+    assert args[ext_index + 2] == "run-tests"
+
+    # The positional artifact path immediately follows run-tests and is addressed
+    # under the container working directory.
+    artifact_path = args[ext_index + 3]
+    assert artifact_path == f"{CONTAINER_WORKDIR}/{name}.zip"
+
+
+# ===========================================================================
+# Property 7
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 7: DirectJar handler mode is always selected
+@settings(max_examples=100)
+@given(name=hyphenated_names(), region=regions)
+def test_property_7_direct_jar_mode_always_present(name, region):
+    args, _env = RqtsImage("img:ref").build_run_args(_make_project(name), region, CREDS)
+
+    assert args.count("--direct-jar") == 1
+
+
+# ===========================================================================
+# Property 8
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 8: region argument always present with the effective region
+@settings(max_examples=100)
+@given(name=hyphenated_names(), use_default=st.booleans(), region=regions)
+def test_property_8_region_argument_always_present(name, use_default, region):
+    effective_region = CLI_DEFAULT_REGION if use_default else region
+    args, _env = RqtsImage("img:ref").build_run_args(
+        _make_project(name), effective_region, CREDS
+    )
+
+    assert "-r" in args
+    assert args[args.index("-r") + 1] == effective_region
+
+
+# ===========================================================================
+# Property 9
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 9: the old SAM Local / remote-lambda flags are never emitted
+@settings(max_examples=100)
+@given(name=hyphenated_names(), region=regions)
+def test_property_9_forbidden_flags_never_emitted(name, region):
+    args, _env = RqtsImage("img:ref").build_run_args(_make_project(name), region, CREDS)
+
+    for flag in FORBIDDEN_FLAGS:
+        assert flag not in args
+
+
+# ===========================================================================
+# Property 10
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 10: no host networking is configured (DirectJar needs none)
+@settings(max_examples=100)
+@given(name=hyphenated_names(), region=regions)
+def test_property_10_no_host_networking(name, region):
+    args, _env = RqtsImage("img:ref").build_run_args(_make_project(name), region, CREDS)
+
+    assert "--network" not in args
+    assert "--add-host" not in args
+    assert "host.docker.internal" not in " ".join(args)
+
+
+# ===========================================================================
+# Property 11
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 11: output directory is emitted via -o under the mount
+@settings(max_examples=100)
+@given(name=hyphenated_names(), region=regions)
+def test_property_11_output_dir_emitted(name, region):
+    args, _env = RqtsImage("img:ref").build_run_args(_make_project(name), region, CREDS)
+
+    assert "-o" in args
+    assert args[args.index("-o") + 1] == CONTAINER_OUTPUT_DIR
+    # The output dir lives under the container working directory mount.
+    assert CONTAINER_OUTPUT_DIR.startswith(CONTAINER_WORKDIR)
+
+
+# ===========================================================================
+# Property 12
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 12: no scenario-selection or exclusion flags
+# are ever emitted; scenario selection is owned by the executor image
+@settings(max_examples=100)
+@given(name=hyphenated_names(), region=regions)
+def test_property_12_no_scenario_selection_or_exclusion_flags(name, region):
+    args, _env = RqtsImage("img:ref").build_run_args(_make_project(name), region, CREDS)
+
+    assert "-s" not in args
+    assert "--scenarios" not in args
+    for token in args:
+        assert not token.startswith("--exclude-")
+    for tagging_scenario in FORBIDDEN_TAGGING_SCENARIOS:
+        assert tagging_scenario not in args
+
+    # The executor command therefore ends at the output directory.
+    assert args[-2:] == ["-o", CONTAINER_OUTPUT_DIR]
+
+
+# ===========================================================================
+# Type configuration
+# ===========================================================================
+def test_type_configuration_referenced_by_name_and_valued_in_env():
+    """A supplied type configuration is referenced by name in the arguments and
+    its JSON content travels only in the env mapping."""
+    type_configuration = json.dumps({"Credentials": {"ApiKey": f"{_SECRET_MARKER}key"}})
+
+    args, env = RqtsImage("img:ref").build_run_args(
+        _make_project("aws-foo-bar"),
+        "us-east-1",
+        CREDS,
+        type_configuration=type_configuration,
+    )
+
+    assert ENV_TYPE_CONFIGURATION in _env_entries(args)
+    assert env[ENV_TYPE_CONFIGURATION] == type_configuration
+    for token in args:
+        assert f"{_SECRET_MARKER}key" not in token
+
+
+def test_type_configuration_omitted_leaves_variable_unset():
+    """Without a type configuration the variable is neither referenced nor set,
+    so the executor falls back to the one packaged in the artifact."""
+    args, env = RqtsImage("img:ref").build_run_args(
+        _make_project("aws-foo-bar"), "us-east-1", CREDS
+    )
+
+    assert ENV_TYPE_CONFIGURATION not in _env_entries(args)
+    assert ENV_TYPE_CONFIGURATION not in env
+
+
+def test_image_reference_precedes_the_executor_command():
+    """The image is the last docker option and the executor command follows it."""
+    args, _env = RqtsImage("img:ref").build_run_args(
+        _make_project("aws-foo-bar", root="/my/project/root"), "us-east-1", CREDS
+    )
+
+    assert args[args.index("img:ref") + 1] == "--extension"
+    assert args[args.index("-v") + 1] == f"/my/project/root:{CONTAINER_WORKDIR}"
+
+
+# ===========================================================================
+# Running the container
+# ===========================================================================
+def test_run_ensures_image_creates_output_dir_and_returns_exit_code(tmp_path):
+    """``run`` pulls the image, creates the host output directory, spawns the
+    container with the credential env merged over the ambient environment, and
+    returns the container's exit code.
+    """
+    fake_docker = _RecordingDocker([0], run_returncode=0)
+    project = _make_project("aws-foo-bar", root=str(tmp_path))
+
+    with _patch_docker(fake_docker):
+        with mock.patch.dict(image_module.os.environ, {"AMBIENT": "yes"}, clear=False):
+            exit_code = RqtsImage("img:ref").run(project, "us-east-1", CREDS)
+
+    assert exit_code == 0
+    # The pull happened before the container ran.
+    assert [call.args[0] for call in fake_docker.calls] == ["pull", "run"]
+    # The host side of the bind-mounted output directory exists.
+    assert (tmp_path / HOST_OUTPUT_DIRNAME).is_dir()
+
+    run_call = fake_docker.run_calls[0]
+    # Output streams live, and the ambient environment is preserved alongside
+    # the credential values.
+    assert run_call.capture is False
+    assert run_call.env["AMBIENT"] == "yes"
+    for key in ENV_CRED_KEYS + CALLER_ENV_CRED_KEYS:
+        assert run_call.env[key] == CREDS[key.replace("CALLER_", "")]
+    # No credential value appears in the arguments themselves.
+    for token in run_call.args:
+        assert token not in CREDS.values()
+
+
+def test_run_returns_nonzero_container_exit_code(tmp_path):
+    """A failing container's exit code is returned unchanged (no raise here)."""
+    fake_docker = _RecordingDocker([0], run_returncode=7)
+
+    with _patch_docker(fake_docker):
+        exit_code = RqtsImage("img:ref").run(
+            _make_project("aws-foo-bar", root=str(tmp_path)), "us-east-1", CREDS
+        )
+
+    assert exit_code == 7
+
+
+def test_run_does_not_start_container_when_image_unavailable(tmp_path):
+    """A fatal pull failure halts before the container is spawned."""
+    fake_docker = _RecordingDocker([1] * PULL_MAX_ATTEMPTS, inspect_result=1)
+
+    with _patch_docker(fake_docker):
+        with pytest.raises(SysExitRecommendedError):
+            RqtsImage("img:ref").run(
+                _make_project("aws-foo-bar", root=str(tmp_path)), "us-east-1", CREDS
+            )
+
+    assert fake_docker.run_calls == []
+
+
+@pytest.mark.parametrize(
+    "spawn_error", [FileNotFoundError("no docker"), OSError("boom")]
+)
+def test_run_spawn_failure_raises(tmp_path, spawn_error):
+    """A container that cannot be spawned surfaces as SysExitRecommendedError.
+
+    Validates: Requirements 5.4
+    """
+    project = _make_project("aws-foo-bar", root=str(tmp_path))
+
+    def docker(args, timeout=None, env=None, capture=True):
+        # pylint: disable=unused-argument
+        if args[0] == "pull":
+            return _completed(args, 0)
+        raise spawn_error
+
+    # staticmethod: a plain function patched onto the class would bind ``self``.
+    with _patch_docker(staticmethod(docker)):
+        with pytest.raises(SysExitRecommendedError) as excinfo:
+            RqtsImage("img:ref").run(project, "us-east-1", CREDS)
+
+    assert "could not be started" in str(excinfo.value)
+
+
+def test_run_passes_type_configuration_through_to_the_container(tmp_path):
+    """The resolved type configuration reaches the container env, by name only in
+    the arguments."""
+    fake_docker = _RecordingDocker([0])
+    type_configuration = json.dumps({"Credentials": {"ApiKey": "abc"}})
+
+    with _patch_docker(fake_docker):
+        RqtsImage("img:ref").run(
+            _make_project("aws-foo-bar", root=str(tmp_path)),
+            "us-east-1",
+            CREDS,
+            type_configuration=type_configuration,
+        )
+
+    run_call = fake_docker.run_calls[0]
+    assert run_call.env[ENV_TYPE_CONFIGURATION] == type_configuration
+    assert ENV_TYPE_CONFIGURATION in _env_entries(run_call.args)
diff --git a/tests/rqts/test_preconditions.py b/tests/rqts/test_preconditions.py
new file mode 100644
index 00000000..069d3996
--- /dev/null
+++ b/tests/rqts/test_preconditions.py
@@ -0,0 +1,262 @@
+"""Tests for ``rpdk.core.rqts.preconditions``.
+
+Covers:
+- Property 3: precondition failures are aggregated exactly (over the three
+  DirectJar checks: Docker runtime, artifact package, credentials+region).
+- Each precondition failing in isolation produces its own message and prevents
+  any container run.
+
+The DirectJar handler mode has no SAM Local endpoint to probe and reads inputs
+from the packaged artifact zip, so there is no endpoint or inputs check.
+
+Every check is controlled independently by patching within
+``rpdk.core.rqts.preconditions`` (``shutil.which`` + ``subprocess.run`` for
+Docker, ``create_sdk_session`` for credentials) and by toggling the artifact
+package file under a ``tmp_path`` working directory.
+"""
+
+import contextlib
+import subprocess
+from unittest import mock
+
+import pytest
+from hypothesis import HealthCheck, given, settings, strategies as st
+
+from rpdk.core.exceptions import CLIMisconfiguredError
+from rpdk.core.rqts.preconditions import check_preconditions
+
+PRECONDITIONS_MODULE = "rpdk.core.rqts.preconditions"
+
+# The three checks, in the order check_preconditions runs them, keyed to the
+# distinctive substring of the failure message each one emits.
+CHECK_NAMES = ("docker", "artifact", "credentials")
+MESSAGE_SUBSTRINGS = {
+    "docker": "Docker is required",
+    "artifact": "artifact package",
+    "credentials": "valid AWS credentials and a region",
+}
+
+
+class FakeProject:
+    """Minimal stand-in for ``rpdk.core.project.Project``.
+
+    Only the attributes the precondition checks read are provided: ``root`` and
+    ``hypenated_name``.
+    """
+
+    def __init__(self, root):
+        self.root = root
+        self.hypenated_name = "aws-foo-bar"
+
+
+def _make_args():
+    args = mock.Mock()
+    args.region = "us-east-1"
+    args.profile = None
+    return args
+
+
+@contextlib.contextmanager
+def configured_env(work_dir, states):
+    """Force each precondition to pass/fail per ``states``.
+
+    ``states`` maps check name -> bool, where ``True`` means the check should
+    PASS and ``False`` means it should FAIL. Yields ``(args, project)`` wired so
+    that ``check_preconditions`` observes exactly those outcomes.
+    """
+    project = FakeProject(work_dir)
+
+    # Artifact package presence (real filesystem toggle).
+    artifact_path = work_dir / f"{project.hypenated_name}.zip"
+    if states["artifact"]:
+        artifact_path.write_bytes(b"zip")
+    elif artifact_path.exists():
+        artifact_path.unlink()
+
+    # Docker: which() present + `docker info` returncode 0 => pass.
+    which_return = "/usr/bin/docker" if states["docker"] else None
+    docker_info_result = mock.Mock(returncode=0 if states["docker"] else 1)
+
+    # Credentials: create_sdk_session succeeds => pass, raises => fail.
+    session_side_effect = (
+        None if states["credentials"] else CLIMisconfiguredError("no creds")
+    )
+
+    with contextlib.ExitStack() as stack:
+        stack.enter_context(
+            mock.patch(
+                f"{PRECONDITIONS_MODULE}.shutil.which", return_value=which_return
+            )
+        )
+        stack.enter_context(
+            mock.patch(
+                f"{PRECONDITIONS_MODULE}.subprocess.run",
+                return_value=docker_info_result,
+            )
+        )
+        stack.enter_context(
+            mock.patch(
+                f"{PRECONDITIONS_MODULE}.create_sdk_session",
+                return_value=mock.Mock(),
+                side_effect=session_side_effect,
+            )
+        )
+        yield _make_args(), project
+
+
+def _matched_checks(failures):
+    """Return the set of check names whose message substring appears in ``failures``."""
+    matched = set()
+    for name, substring in MESSAGE_SUBSTRINGS.items():
+        if any(substring in message for message in failures):
+            matched.add(name)
+    return matched
+
+
+# Feature: cfn-test-v2-flag, Property 3: For any subset of the precondition
+# checks (Docker runtime, artifact package, credentials+region) forced to fail,
+# check_preconditions returns a failure list whose messages correspond to
+# exactly that subset - every unmet precondition is named and every met
+# precondition is absent.
+@settings(max_examples=200, suppress_health_check=[HealthCheck.function_scoped_fixture])
+@given(pass_flags=st.fixed_dictionaries({name: st.booleans() for name in CHECK_NAMES}))
+def test_precondition_failures_aggregated_exactly(tmp_path, pass_flags):
+    """Validates: Requirements 3.7"""
+    expected_failures = {name for name, passed in pass_flags.items() if not passed}
+
+    with configured_env(tmp_path, pass_flags) as (args, project):
+        failures = check_preconditions(args, project)
+
+    # Every unmet precondition is named exactly once, and no met precondition is.
+    assert _matched_checks(failures) == expected_failures
+    assert len(failures) == len(expected_failures)
+
+
+# ---------------------------------------------------------------------------
+# Each precondition failing in isolation.
+# ---------------------------------------------------------------------------
+
+ALL_PASS = dict.fromkeys(CHECK_NAMES, True)
+
+
+def _states_with_only_failing(check):
+    states = dict(ALL_PASS)
+    states[check] = False
+    return states
+
+
+def test_all_preconditions_met_returns_empty(tmp_path):
+    """Sanity baseline: when every check passes, no failures are returned."""
+    with configured_env(tmp_path, dict(ALL_PASS)) as (args, project):
+        assert not check_preconditions(args, project)
+
+
+def test_docker_unavailable_in_isolation(tmp_path):
+    """Requirement 3.2: Docker unavailable yields its message and blocks the run."""
+    with configured_env(tmp_path, _states_with_only_failing("docker")) as (
+        args,
+        project,
+    ):
+        failures = check_preconditions(args, project)
+
+    assert len(failures) == 1
+    assert MESSAGE_SUBSTRINGS["docker"] in failures[0]
+    # A non-empty failure list prevents the caller from ever running a container.
+    assert failures
+
+
+def test_artifact_missing_in_isolation(tmp_path):
+    """Requirement 3.3: missing artifact package yields its message and blocks the run."""
+    with configured_env(tmp_path, _states_with_only_failing("artifact")) as (
+        args,
+        project,
+    ):
+        failures = check_preconditions(args, project)
+
+    assert len(failures) == 1
+    assert MESSAGE_SUBSTRINGS["artifact"] in failures[0]
+    assert failures
+
+
+def test_credentials_unavailable_in_isolation(tmp_path):
+    """Requirement 3.5: missing credentials/region yields the message and blocks the run."""
+    with configured_env(tmp_path, _states_with_only_failing("credentials")) as (
+        args,
+        project,
+    ):
+        failures = check_preconditions(args, project)
+
+    assert len(failures) == 1
+    assert MESSAGE_SUBSTRINGS["credentials"] in failures[0]
+    assert failures
+
+
+# ---------------------------------------------------------------------------
+# Docker daemon ping failure modes (docker CLI present, daemon unreachable).
+# ---------------------------------------------------------------------------
+
+
+@contextlib.contextmanager
+def docker_ping_env(work_dir, **run_kwargs):
+    """Yield ``(args, project)`` with docker on PATH and ``docker info`` stubbed.
+
+    The artifact and credential checks are forced to pass, so any failure the
+    caller observes comes from the Docker daemon ping alone. ``run_kwargs`` is
+    forwarded to ``mock.patch`` for ``subprocess.run`` (``return_value`` for an
+    exit status, ``side_effect`` to raise).
+    """
+    project = FakeProject(work_dir)
+    (work_dir / f"{project.hypenated_name}.zip").write_bytes(b"zip")
+
+    with contextlib.ExitStack() as stack:
+        stack.enter_context(
+            mock.patch(
+                f"{PRECONDITIONS_MODULE}.shutil.which", return_value="/usr/bin/docker"
+            )
+        )
+        stack.enter_context(
+            mock.patch(f"{PRECONDITIONS_MODULE}.subprocess.run", **run_kwargs)
+        )
+        stack.enter_context(
+            mock.patch(
+                f"{PRECONDITIONS_MODULE}.create_sdk_session", return_value=mock.Mock()
+            )
+        )
+        yield _make_args(), project
+
+
+def test_docker_daemon_ping_nonzero_exit_reports_unreachable(tmp_path):
+    """docker CLI present but ``docker info`` exits non-zero -> unreachable daemon.
+
+    Distinct from the missing-CLI case: the binary exists, so the ping itself is
+    what fails.
+
+    Validates: Requirements 3.2
+    """
+    with docker_ping_env(tmp_path, return_value=mock.Mock(returncode=1)) as (
+        args,
+        project,
+    ):
+        failures = check_preconditions(args, project)
+
+    assert len(failures) == 1
+    assert MESSAGE_SUBSTRINGS["docker"] in failures[0]
+
+
+@pytest.mark.parametrize(
+    "ping_error",
+    [OSError("cannot exec"), subprocess.TimeoutExpired(cmd="docker info", timeout=10)],
+)
+def test_docker_daemon_ping_error_reports_unreachable(tmp_path, ping_error):
+    """A ping that raises (spawn failure or timeout) -> unreachable daemon.
+
+    The check converts the exception into a message rather than propagating it,
+    so a hung or broken daemon still aggregates with other failures.
+
+    Validates: Requirements 3.2
+    """
+    with docker_ping_env(tmp_path, side_effect=ping_error) as (args, project):
+        failures = check_preconditions(args, project)
+
+    assert len(failures) == 1
+    assert MESSAGE_SUBSTRINGS["docker"] in failures[0]
diff --git a/tests/rqts/test_runner.py b/tests/rqts/test_runner.py
new file mode 100644
index 00000000..c28d1628
--- /dev/null
+++ b/tests/rqts/test_runner.py
@@ -0,0 +1,333 @@
+"""Tests for ``rpdk.core.rqts.runner``.
+
+Covers the orchestration the runner still owns after everything Docker-facing
+moved to :class:`rpdk.core.rqts.image.RqtsImage`: the artifact-type guards,
+precondition aggregation, credential minting, type configuration resolution,
+and the exit-code contract.
+
+Docker and AWS are never invoked: ``RqtsImage`` and the ``boto_helpers``
+functions imported into ``runner`` are patched at ``rpdk.core.rqts.runner``.
+
+Library: Hypothesis (the standard Python property-based testing library). The
+property test runs at least 100 generated examples via
+``@settings(max_examples=100)`` and is tagged with a comment referencing the
+design property it validates.
+"""
+
+import json
+import logging
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+from hypothesis import given, settings, strategies as st
+
+from rpdk.core.contract.type_configuration import TypeConfiguration
+from rpdk.core.exceptions import InvalidProjectError, SysExitRecommendedError
+from rpdk.core.project import ARTIFACT_TYPE_HOOK, ARTIFACT_TYPE_RESOURCE
+from rpdk.core.rqts.constants import ENV_CRED_KEYS, FAIL_MESSAGE, PASS_MESSAGE
+from rpdk.core.rqts.runner import RqtsRunner
+
+RUNNER_MODULE = "rpdk.core.rqts.runner"
+
+CREDS = dict(zip(ENV_CRED_KEYS, ("AKID", "SECRET", "TOKEN")))
+
+
+@pytest.fixture(autouse=True)
+def _reset_type_configuration():
+    """Keep the ``TypeConfiguration`` class-level cache from leaking between tests.
+
+    The cache has no invalidation, so a value parsed by one test would otherwise
+    be reused by the next.
+    """
+    TypeConfiguration.TYPE_CONFIGURATION = None
+    yield
+    TypeConfiguration.TYPE_CONFIGURATION = None
+
+
+def _make_resource_project(root):
+    """Build a minimal resource project the happy-path runner can drive."""
+    return SimpleNamespace(
+        artifact_type=ARTIFACT_TYPE_RESOURCE,
+        type_name="AWS::Foo::Bar",
+        hypenated_name="aws-foo-bar",
+        root=root,
+    )
+
+
+def _make_args(**overrides):
+    args = SimpleNamespace(
+        region="us-east-1",
+        profile=None,
+        role_arn=None,
+        source_account=None,
+        source_arn=None,
+        typeconfig=None,
+        rqts_image=None,
+    )
+    for key, value in overrides.items():
+        setattr(args, key, value)
+    return args
+
+
+class _Pipeline:
+    """Context manager patching everything ``RqtsRunner.run`` depends on."""
+
+    def __init__(self, exit_code=0, type_configuration=None, failures=()):
+        self._exit_code = exit_code
+        self._patches = {
+            "check": mock.patch(
+                f"{RUNNER_MODULE}.check_preconditions", return_value=list(failures)
+            ),
+            "session": mock.patch(f"{RUNNER_MODULE}.create_sdk_session"),
+            "creds": mock.patch(
+                f"{RUNNER_MODULE}.get_temporary_credentials", return_value=CREDS
+            ),
+            "typeconfig": mock.patch.object(
+                TypeConfiguration,
+                "get_type_configuration",
+                return_value=type_configuration,
+            ),
+            "image": mock.patch(f"{RUNNER_MODULE}.RqtsImage", autospec=True),
+        }
+        self.mocks = {}
+
+    def __enter__(self):
+        self.mocks = {name: patch.start() for name, patch in self._patches.items()}
+        self.image.return_value.run.return_value = self._exit_code
+        return self
+
+    def __exit__(self, *_exc):
+        for patch in reversed(list(self._patches.values())):
+            patch.stop()
+        return False
+
+    @property
+    def image(self):
+        return self.mocks["image"]
+
+    @property
+    def container_run(self):
+        """The mocked ``RqtsImage.run`` the runner delegates to."""
+        return self.mocks["image"].return_value.run
+
+    @property
+    def get_type_configuration(self):
+        return self.mocks["typeconfig"]
+
+    @property
+    def get_temporary_credentials(self):
+        return self.mocks["creds"]
+
+
+# ===========================================================================
+# Property 12: exit-code mapping
+# ===========================================================================
+# Feature: cfn-test-v2-flag, Property 12: Exit code mapping
+@settings(max_examples=100, deadline=None)
+@given(
+    code=st.one_of(
+        st.just(0),
+        st.integers(min_value=1, max_value=255),
+        st.integers(min_value=-255, max_value=-1),
+        st.integers(),
+    )
+)
+def test_property_12_exit_code_mapping(code):
+    """A zero container exit code returns normally; any non-zero code raises
+    SysExitRecommendedError with the fail message.
+
+    Validates: Requirements 5.2, 5.3, 6.3
+    """
+    TypeConfiguration.TYPE_CONFIGURATION = None
+    runner = RqtsRunner(_make_args(), _make_resource_project("/tmp/project"))
+
+    with _Pipeline(exit_code=code):
+        if code == 0:
+            assert runner.run() is None
+        else:
+            with pytest.raises(SysExitRecommendedError) as excinfo:
+                runner.run()
+            assert str(excinfo.value) == FAIL_MESSAGE
+
+
+# ===========================================================================
+# Artifact-type guards
+# ===========================================================================
+def test_guard_artifact_type_hook_rejected_and_does_not_proceed():
+    """Hook project -> SysExitRecommendedError 'resources only'; pipeline halts.
+
+    Validates: Requirements 7.2
+    """
+    runner = RqtsRunner(_make_args(), SimpleNamespace(artifact_type=ARTIFACT_TYPE_HOOK))
+
+    with mock.patch(f"{RUNNER_MODULE}.check_preconditions") as check, mock.patch(
+        f"{RUNNER_MODULE}.RqtsImage"
+    ) as image:
+        with pytest.raises(SysExitRecommendedError) as excinfo:
+            runner.run()
+
+    assert "resource types only" in str(excinfo.value)
+    # The guard fails fast: no downstream stage runs.
+    check.assert_not_called()
+    image.assert_not_called()
+
+
+def test_guard_artifact_type_indeterminate_rejected_and_does_not_proceed():
+    """Indeterminate artifact type -> 'could not determine' error; pipeline halts.
+
+    Validates: Requirements 7.5
+    """
+    runner = RqtsRunner(_make_args(), SimpleNamespace(artifact_type=None))
+
+    with mock.patch(f"{RUNNER_MODULE}.check_preconditions") as check, mock.patch(
+        f"{RUNNER_MODULE}.RqtsImage"
+    ) as image:
+        with pytest.raises(SysExitRecommendedError) as excinfo:
+            runner.run()
+
+    assert "could not determine the project artifact type" in str(excinfo.value)
+    check.assert_not_called()
+    image.assert_not_called()
+
+
+# ===========================================================================
+# Precondition enforcement
+# ===========================================================================
+def test_run_unmet_preconditions_aggregated_and_halts():
+    """Unmet preconditions -> a single error naming every failure; nothing runs.
+
+    The runner turns the aggregated list from ``check_preconditions`` into one
+    ``SysExitRecommendedError`` instead of failing on the first problem, and
+    halts before any container is started.
+
+    Validates: Requirements 3.1, 3.5, 3.7
+    """
+    runner = RqtsRunner(_make_args(), _make_resource_project("/tmp/project"))
+    failures = [
+        "Docker is required and must be running: the Docker daemon could not "
+        "be reached.",
+        "artifact package 'aws-foo-bar.zip' not found; build the project first.",
+    ]
+
+    with _Pipeline(failures=failures) as pipeline:
+        with pytest.raises(SysExitRecommendedError) as excinfo:
+            runner.run()
+
+    message = str(excinfo.value)
+    assert "preconditions were not met" in message
+    for failure in failures:
+        assert failure in message
+    pipeline.image.assert_not_called()
+
+
+# ===========================================================================
+# Credentials and type configuration
+# ===========================================================================
+def test_run_mints_credentials_keyed_for_the_executor():
+    """Credentials are requested with the executor's env var names as key names,
+    with the confused-deputy headers, and handed to the image unchanged.
+
+    Validates: Requirements 3.5
+    """
+    args = _make_args(
+        role_arn="arn:aws:iam::1:role/r", source_account="1", source_arn="arn:x"
+    )
+    project = _make_resource_project("/tmp/project")
+    runner = RqtsRunner(args, project)
+
+    with _Pipeline() as pipeline:
+        runner.run()
+
+    session = pipeline.get_temporary_credentials.call_args[0][0]
+    assert pipeline.get_temporary_credentials.call_args[0][1] is ENV_CRED_KEYS
+    assert pipeline.get_temporary_credentials.call_args[0][2] == args.role_arn
+    assert pipeline.get_temporary_credentials.call_args[1]["headers"] == {
+        "account_id": "1",
+        "source_arn": "arn:x",
+    }
+    assert session is not None
+
+    pipeline.image.assert_called_once_with(args.rqts_image)
+    pipeline.container_run.assert_called_once_with(
+        project, args.region, CREDS, type_configuration=None
+    )
+
+
+def test_run_serializes_type_configuration_to_json():
+    """A resolved type configuration is passed to the image as a JSON string."""
+    config = {"Credentials": {"ApiKey": "123", "ApplicationKey": "456"}}
+    project = _make_resource_project("/tmp/project")
+    runner = RqtsRunner(_make_args(typeconfig="./tc.json"), project)
+
+    with _Pipeline(type_configuration=config) as pipeline:
+        runner.run()
+
+    pipeline.get_type_configuration.assert_called_once_with("./tc.json")
+    passed = pipeline.container_run.call_args[1]["type_configuration"]
+    assert json.loads(passed) == config
+
+
+@pytest.mark.parametrize("config", [None, {}])
+def test_run_passes_no_type_configuration_when_absent_or_empty(config):
+    """A missing or empty type configuration leaves the variable unset so the
+    executor falls back to the one packaged in the artifact."""
+    runner = RqtsRunner(_make_args(), _make_resource_project("/tmp/project"))
+
+    with _Pipeline(type_configuration=config) as pipeline:
+        runner.run()
+
+    assert pipeline.container_run.call_args[1]["type_configuration"] is None
+
+
+def test_run_clears_the_type_configuration_cache_before_reading():
+    """The class-level cache is cleared so each run reads the file from disk
+    rather than reusing a value parsed earlier in the process."""
+    TypeConfiguration.TYPE_CONFIGURATION = {"stale": True}
+    observed = {}
+
+    def record(typeconfigloc):  # pylint: disable=unused-argument
+        observed["cache"] = TypeConfiguration.TYPE_CONFIGURATION
+
+    runner = RqtsRunner(_make_args(), _make_resource_project("/tmp/project"))
+
+    with _Pipeline() as pipeline:
+        pipeline.get_type_configuration.side_effect = record
+        runner.run()
+
+    assert observed["cache"] is None
+
+
+def test_run_surfaces_invalid_type_configuration():
+    """An invalid type configuration file fails the run with the CLI's own
+    InvalidProjectError (a SysExitRecommendedError), before the container runs."""
+    runner = RqtsRunner(
+        _make_args(typeconfig="./bad.json"), _make_resource_project("/x")
+    )
+
+    with _Pipeline() as pipeline:
+        pipeline.get_type_configuration.side_effect = InvalidProjectError(
+            "Type configuration file './bad.json' is invalid"
+        )
+        with pytest.raises(SysExitRecommendedError) as excinfo:
+            runner.run()
+
+    assert "is invalid" in str(excinfo.value)
+    pipeline.container_run.assert_not_called()
+
+
+# ===========================================================================
+# Reporting
+# ===========================================================================
+def test_run_logs_pass_message_on_success(caplog):
+    """A passing run logs the pass message at INFO and returns.
+
+    Validates: Requirements 6.1
+    """
+    runner = RqtsRunner(_make_args(), _make_resource_project("/tmp/project"))
+
+    with _Pipeline(exit_code=0):
+        with caplog.at_level(logging.INFO, logger=RUNNER_MODULE):
+            assert runner.run() is None
+
+    assert PASS_MESSAGE in caplog.text
diff --git a/tests/test_test.py b/tests/test_test.py
index 031feb83..be1ea927 100644
--- a/tests/test_test.py
+++ b/tests/test_test.py
@@ -1,5 +1,6 @@
 # fixture and parameter have the same name
-# pylint: disable=protected-access,redefined-outer-name
+# pylint: disable=protected-access,redefined-outer-name,too-many-lines
+import argparse
 import json
 import os
 from contextlib import contextmanager
@@ -31,6 +32,7 @@
     get_marker_options,
     get_overrides,
     get_type,
+    setup_subparser,
     temporary_ini_file,
 )
 from rpdk.core.utils.handler_utils import generate_handler_name
@@ -838,3 +840,193 @@ def test_input_files_read_safely(self, base):
 
         result = get_inputs(base, DEFAULT_REGION, DEFAULT_ENDPOINT, 1, None, None, {})
         assert result is not None and "CREATE" in result
+
+
+# --v2 flag registration and RQTS routing tests
+# (Task 8.3: flag registration; Task 8.4: routing + backward compatibility)
+
+
+def _build_test_parser():
+    """Build a parser wired up exactly the way cli.py does for the test command.
+
+    Mirrors rpdk.core.cli.main: a root parser with a shared base subparser
+    (providing -v/--verbose) as the only parent, and the test subcommand
+    registered via setup_subparser.
+    """
+    parser = argparse.ArgumentParser()
+    base_subparser = argparse.ArgumentParser(add_help=False)
+    base_subparser.add_argument("-v", "--verbose", action="count", default=0)
+    subparsers = parser.add_subparsers(dest="subparser_name")
+    setup_subparser(subparsers, [base_subparser])
+    return parser, subparsers
+
+
+# Task 8.3 - Requirements 1.1, 1.5, 4.11
+
+
+def test_v2_flag_defaults_to_false():
+    parser, _subparsers = _build_test_parser()
+    args = parser.parse_args(["test"])
+    assert args.v2 is False
+
+
+def test_v2_flag_present_sets_true():
+    parser, _subparsers = _build_test_parser()
+    args = parser.parse_args(["test", "--v2"])
+    assert args.v2 is True
+
+
+def test_rqts_image_defaults_to_none():
+    parser, _subparsers = _build_test_parser()
+    args = parser.parse_args(["test"])
+    assert args.rqts_image is None
+
+
+def test_rqts_image_can_be_overridden():
+    parser, _subparsers = _build_test_parser()
+    args = parser.parse_args(["test", "--rqts-image", "foo:bar"])
+    assert args.rqts_image == "foo:bar"
+
+
+def test_v2_help_text_identifies_opt_in_rqts_runner():
+    _parser, subparsers = _build_test_parser()
+    help_text = subparsers.choices["test"].format_help()
+    lowered = help_text.lower()
+    assert "opt-in" in lowered
+    assert "rqts local test runner" in lowered
+
+
+def test_no_scenario_selection_option_on_parser():
+    parser, _subparsers = _build_test_parser()
+    # Scenario selection is owned by the executor image and is not
+    # user-selectable through the CLI (Req 4.11); a scenario-selection option
+    # is not a recognized argument.
+    with pytest.raises(SystemExit):
+        parser.parse_args(["test", "--scenarios", "x"])
+
+
+# Task 8.4 - Requirements 1.2, 1.3, 1.4, 7.1, 7.3, 7.4
+
+
+def test_v2_absent_uses_pytest_path_and_does_not_construct_runner(base):
+    create_input_file(base, '{"a": 1}', '{"a": 2}', '{"b": 1}')
+    mock_project = Mock(spec=Project)
+    mock_project.schema = RESOURCE_SCHEMA
+    mock_project.root = base
+    mock_project.executable_entrypoint = None
+    mock_project.artifact_type = ARTIFACT_TYPE_RESOURCE
+
+    patch_project = patch(
+        "rpdk.core.test.Project", autospec=True, return_value=mock_project
+    )
+    patch_plugin = patch("rpdk.core.test.ContractPlugin", autospec=True)
+    patch_resource_client = patch("rpdk.core.test.ResourceClient", autospec=True)
+    patch_pytest = patch("rpdk.core.test.pytest.main", autospec=True, return_value=0)
+    patch_ini = patch(
+        "rpdk.core.test.temporary_ini_file", side_effect=mock_temporary_ini_file
+    )
+    patch_runner = patch("rpdk.core.rqts.runner.RqtsRunner", autospec=True)
+    # fmt: off
+    with patch_project, \
+            patch_plugin, \
+            patch_resource_client, \
+            patch_ini, \
+            patch_pytest as mock_pytest, \
+            patch_runner as mock_runner:
+        main(args_in=["test"])
+    # fmt: on
+
+    # The pytest path is exercised and the RQTS runner is never constructed.
+    mock_pytest.assert_called_once()
+    mock_runner.assert_not_called()
+
+
+def test_v2_absent_nonzero_pytest_return_still_raises(base):
+    create_input_file(base, '{"a": 1}', '{"a": 2}', '{"b": 1}')
+    mock_project = Mock(spec=Project)
+    mock_project.schema = RESOURCE_SCHEMA
+    mock_project.root = base
+    mock_project.executable_entrypoint = None
+    mock_project.artifact_type = ARTIFACT_TYPE_RESOURCE
+
+    patch_project = patch(
+        "rpdk.core.test.Project", autospec=True, return_value=mock_project
+    )
+    patch_plugin = patch("rpdk.core.test.ContractPlugin", autospec=True)
+    patch_resource_client = patch("rpdk.core.test.ResourceClient", autospec=True)
+    patch_pytest = patch("rpdk.core.test.pytest.main", autospec=True, return_value=1)
+    patch_ini = patch(
+        "rpdk.core.test.temporary_ini_file", side_effect=mock_temporary_ini_file
+    )
+    patch_runner = patch("rpdk.core.rqts.runner.RqtsRunner", autospec=True)
+    # fmt: off
+    with patch_project, \
+            patch_plugin, \
+            patch_resource_client, \
+            patch_ini, \
+            patch_pytest as mock_pytest, \
+            patch_runner as mock_runner:
+        with pytest.raises(SystemExit) as excinfo:
+            main(args_in=["test"])
+    # fmt: on
+
+    # Existing backward-compatible behavior: a non-zero pytest return raises
+    # SysExitRecommendedError, mapped by cli.py to a non-unhandled SystemExit.
+    assert excinfo.value.code != EXIT_UNHANDLED_EXCEPTION
+    # The --v2 branch did not alter forwarding to the pytest path.
+    mock_pytest.assert_called_once()
+    mock_runner.assert_not_called()
+
+
+def test_v2_resource_project_constructs_and_runs_runner(base):
+    mock_project = Mock(spec=Project)
+    mock_project.schema = RESOURCE_SCHEMA
+    mock_project.root = base
+    mock_project.executable_entrypoint = None
+    mock_project.artifact_type = ARTIFACT_TYPE_RESOURCE
+
+    patch_project = patch(
+        "rpdk.core.test.Project", autospec=True, return_value=mock_project
+    )
+    patch_pytest = patch("rpdk.core.test.pytest.main", autospec=True, return_value=0)
+    # test() imports RqtsRunner locally (from .rqts.runner import RqtsRunner),
+    # so patch it at its source module.
+    patch_runner = patch("rpdk.core.rqts.runner.RqtsRunner", autospec=True)
+    # fmt: off
+    with patch_project, \
+            patch_pytest as mock_pytest, \
+            patch_runner as mock_runner:
+        main(args_in=["test", "--v2"])
+    # fmt: on
+
+    # The runner is constructed with the parsed args and loaded project, and
+    # run() is invoked exactly once; the pytest path is not exercised.
+    mock_runner.assert_called_once()
+    called_args = mock_runner.call_args[0]
+    assert called_args[1] is mock_project
+    mock_runner.return_value.run.assert_called_once_with()
+    mock_pytest.assert_not_called()
+
+
+def test_v2_module_project_warns_and_does_not_construct_runner(capsys):
+    mock_project = Mock(spec=Project)
+    mock_project.artifact_type = ARTIFACT_TYPE_MODULE
+
+    patch_project = patch(
+        "rpdk.core.test.Project", autospec=True, return_value=mock_project
+    )
+    patch_pytest = patch("rpdk.core.test.pytest.main", autospec=True, return_value=0)
+    patch_runner = patch("rpdk.core.rqts.runner.RqtsRunner", autospec=True)
+    # fmt: off
+    with patch_project, \
+            patch_pytest as mock_pytest, \
+            patch_runner as mock_runner:
+        # The module short-circuit precedes the --v2 branch: clean return, no
+        # SystemExit raised.
+        main(args_in=["test", "--v2"])
+    # fmt: on
+
+    mock_runner.assert_not_called()
+    mock_pytest.assert_not_called()
+    out, err = capsys.readouterr()
+    assert "module" in (out + err).lower()