diff --git a/docs/skills/test-authoring/behave/SKILL.md b/docs/skills/test-authoring/behave/SKILL.md
index 167f4ef8..0c081fe3 100644
--- a/docs/skills/test-authoring/behave/SKILL.md
+++ b/docs/skills/test-authoring/behave/SKILL.md
@@ -252,6 +252,22 @@ never matches and passes falsely.
 `@flatpak_cli`. Tag CLI-only, image-agnostic software scenarios with `@flatpak_cli`
 so they still run on gnomeos and other non-Bluefin images.
 
+## `@requires_cached_image` gates scenarios on a pre-pulled OCI image
+
+A scenario must never pull the container image it needs — a cold
+`distrobox create` pulls inside the scenario and eats the CI timeout (#501).
+Tag it `@requires_cached_image`; `skip_when_image_not_cached()` in
+`tests/shared/image_cache.py`, called from the suite's `before_scenario` right
+after `skip_quarantine`, reads the image refs out of the scenario's own step
+text, probes each with `podman image exists` on the DUT, and skips while any is
+absent. The scenario then activates on its own once the image is cached.
+
+It is a **runtime capability gate** like `@requires_bctl`, not a non-runnable
+tag: keep it out of `_SKIP_TAGS` / `NON_RUNNABLE_TAGS` / `BEHAVE_TAG_ARGS`, and
+never pair it with `@pending` or `@future` — `skip_quarantine` returns first and
+the gate goes inert. See
+[the cached-image gate reference](references/cached-image-gate.md).
+
 ## Feature scaffolding with @future
 
 
@@ -326,6 +342,8 @@ Each suite loads only its own `steps/*.py` files plus `qecore.common_steps`. A s
 
 Lesson surfaced 2026-05-30: `No journal entries match "{pattern}"` was added to `software/steps.py` but `ptyxis.feature` (developer suite) also used it — causing `UndefinedStep` at runtime.
 
+Isolation cuts the other way too: an `environment.py` hook that imports a module containing `@step` decorators registers those phrases into the suite it runs in. `tests/shared/ssh_steps.py` collides with the DX suite's own `SSH command return code is "{code}"`, so hooks that need to run a command on the DUT resolve connection details from `tests/shared/ssh_config.py` and call `subprocess` directly rather than importing the step library (#501).
+
 ## behave rerun output can contain non-path noise
 
 
@@ -473,6 +491,7 @@ non-dependent scenarios to a separate feature.
 - [Mocking interactive CLI tools (gum, fzf, gh) in ujust coverage.](references/mocking-interactive-cli.md)
 - [Driving bluefinctl devmode non-interactively, and the assertion traps around it.](references/bctl-devmode.md)
 - [Which ujust recipes can be driven non-interactively, and why the rest stay @pending.](references/ujust-noninteractive.md)
+- [Gating scenarios on a pre-pulled OCI image with @requires_cached_image.](references/cached-image-gate.md)
 
 ## Sources
 
diff --git a/docs/skills/test-authoring/behave/references/cached-image-gate.md b/docs/skills/test-authoring/behave/references/cached-image-gate.md
new file mode 100644
index 00000000..76b2e2e5
--- /dev/null
+++ b/docs/skills/test-authoring/behave/references/cached-image-gate.md
@@ -0,0 +1,104 @@
+---
+name: cached-image-gate
+description: "Gating scenarios on a pre-pulled OCI image with @requires_cached_image, and the masking trap that makes the tag inert."
+metadata:
+  type: reference
+  audience: agents
+  maturity: stable
+---
+# Cached Image Gate
+
+## Why the tag exists
+
+A scenario that needs a container image on the device under test must never pull
+that image itself. `distrobox create --image registry.fedoraproject.org/fedora-toolbox:latest`
+against a cold podman store pulls hundreds of megabytes inside the scenario and
+eats the CI timeout, so the run reports "distrobox is broken" when the real
+finding is "the runner had no image". That is the blocker recorded on
+`projectbluefin/testsuite#501` and tracked lab-side as `projectbluefin/lab#621`.
+
+`@requires_cached_image` turns that blocker into a runtime condition instead of a
+hand-maintained `@pending` marker.
+
+## How it works
+
+`tests/shared/image_cache.py` exposes `skip_when_image_not_cached(context, scenario)`.
+A suite calls it from `before_scenario`, immediately after `skip_quarantine`:
+
+```python
+from tests.shared.image_cache import skip_when_image_not_cached
+
+if skip_when_image_not_cached(context, scenario):
+    return
+```
+
+For a tagged scenario it:
+
+1. reads the image references out of the **scenario's own step text** — the
+   feature file stays the single source of truth for what is under test, so the
+   probe can never drift from the image the steps actually use;
+2. runs `podman image exists <ref>` on the DUT for each distinct reference;
+3. skips with the missing references named when any is absent.
+
+`podman image exists` is local-only and never contacts a registry. That is
+deliberate and load-bearing: a probe that could pull would trigger the very
+timeout the gate exists to prevent. `tests/unit/test_image_cache.py` asserts the
+probe command shape for exactly this reason.
+
+An unreachable DUT counts as "not cached" — a failed SSH probe is not evidence
+that an image is present, and skipping is the safe reading.
+
+## Why the probe does not use `run_ssh`
+
+`tests/shared/ssh_steps.py` looks like the obvious way to run the probe, but
+importing it registers its `@step` phrases into behave's global registry. Two of
+them — `SSH command return code is "{code}"` and `Last command output contains
+"{text}"` — are also defined by `tests/dx/features/steps/steps.py`. A DX
+`before_scenario` that imported the module would raise `AmbiguousStep` and take
+the entire suite down, and the DX suite is the first consumer of this gate.
+
+So the probe resolves connection details from `tests/shared/ssh_config.py` — the
+helper explicitly documented for "suite `environment.py` hooks that probe the VM
+directly", carrying no step definitions — and shells out itself.
+`test_gate_does_not_import_the_shared_ssh_step_library` pins this.
+
+The probe also leaves `context.command_stdout` / `context.ssh_rc` alone, unlike
+a step. It runs before `before_scenario` resets per-scenario state, so writing
+to those would leak a probe result into the scenario's first assertion.
+
+## Three rules that are easy to get wrong
+
+**It is not a non-runnable tag.** Do not add it to `_SKIP_TAGS` in
+`tests/shared/quarantine.py`, to `NON_RUNNABLE_TAGS` in
+`tests/shared/behave_retry.py`, or to `BEHAVE_TAG_ARGS` in `e2e.yml`. It is a
+runtime capability gate in the family of `@requires_bctl` and
+`@requires_toggle_action`: the scenario must begin running the moment the image
+is cached, with no feature-file edit and no follow-up PR.
+
+**Never pair it with `@pending`, `@future`, `@quarantine`, or `@hardware_blocked`.**
+`skip_quarantine` runs first and returns early, so the gate never executes and the
+tag is inert — the masking trap documented in
+`docs/skills/test-authoring/suite-map/SKILL.md`. This is not hypothetical: the
+three distrobox scenarios carried `@pending @requires_cached_image` from the day
+they landed, and the tag did nothing at all until #501 was finished.
+`test_tagged_scenarios_are_not_masked_by_a_non_runnable_tag` fails the build if it
+recurs.
+
+**Name the image with a registry and a tag.** `registry.fedoraproject.org/fedora-toolbox:latest`
+is recognised; bare `fedora:latest` is not, because what it resolves to depends on
+the DUT's `registries.conf` search list — the probe could then disagree with the
+pull the scenario would perform. A tagged scenario whose steps name no qualified
+image is an authoring error, and
+`test_every_tagged_scenario_names_an_image` fails on it rather than letting it
+skip forever while looking like an infra gap.
+
+## Verifying a gated scenario
+
+`behave --dry-run` cannot exercise the gate: it never calls `before_scenario`.
+Check the gate with the unit tests, and check the scenario body against a DUT
+that has the image pre-pulled:
+
+```bash
+python3 -m pytest tests/unit/test_image_cache.py -q
+podman pull registry.fedoraproject.org/fedora-toolbox:latest   # on the DUT
+```
diff --git a/docs/skills/test-authoring/suite-map/SKILL.md b/docs/skills/test-authoring/suite-map/SKILL.md
index 797ea064..1f1fcfbe 100644
--- a/docs/skills/test-authoring/suite-map/SKILL.md
+++ b/docs/skills/test-authoring/suite-map/SKILL.md
@@ -149,6 +149,7 @@ Set `chunked_enabled: true` once `ghcr.io/projectbluefin/bluefin:latest` ships z
 | `@regression` | Anchors a known incident regression guard; must remain active indefinitely |
 | `@kde_smoke` | KDE Plasma smoke-suite identifier; used by `e2e.yml` suite registration (#645) |
 | `@informational` | Bake-period tier; scenario runs and reports results but does not gate promotion until promoted to `@critical` |
+| `@requires_cached_image` | Scenario needs the OCI image named in its own steps to be pre-pulled on the DUT; `tests/shared/image_cache.py` probes `podman image exists` from `before_scenario` and skips while it is absent. A **runtime capability gate** like `@requires_bctl`, not a non-runnable tag — never pair it with `@pending`/`@future`, which mask it (#501) |
 
 ## Coverage snapshot
 
@@ -159,14 +160,14 @@ Set `chunked_enabled: true` once `ghcr.io/projectbluefin/bluefin:latest` ships z
 
 <!-- coverage-snapshot:start -->
 
-526 scenarios across 72 feature files: 412 active, 0 quarantined, 114 `@future`/`@pending`/`@hardware_blocked`
+526 scenarios across 72 feature files: 415 active, 0 quarantined, 111 `@future`/`@pending`/`@hardware_blocked`
 
 | Suite | Scenarios | Active | Quarantined | Pending/Future | Notes |
 |---|---|---|---|---|---|
 | bazzite | 20 | 20 | 0 | 0 | Extension presence + shell behaviour |
 | common | 121 | 101 | 0 | 20 | Signing assertions `@future` pending the ublue-os→projectbluefin policy migration; flatpak model/state, dconf defaults, immutability and portal socket checks `@pending` on CI infra; Flatpak model + state; XDG portal health + integration; container runtime (podman); polkit rules; shell env + sourcing; system scripts; ujust recipes; devmode via bctl (non-interactive contract + idempotent state-check gated `@requires_bctl`, group mutation `@pending` on CI polkit); GSettings/dconf defaults; immutable OS integrity; desktop entries; signing assertions; Dakota `ujust --choose` regression guard active (`@dakota_only`); `ujust report` is `@pending` on #706 until a Dakota lab run validates the mocked submit flow |
 | developer | 23 | 7 | 0 | 16 | 6 brew + 6 ptyxis + 4 bctl now `@pending`: `brew-setup.service` masked in CI (#487) and the ptyxis AT-SPI restart issue (#368) |
-| dx | 18 | 10 | 0 | 8 | distrobox enter/create/install/export, JupyterLab, brew, mise — infra gaps, all `@pending` |
+| dx | 18 | 13 | 0 | 5 | distrobox create/install/export are active behind the `@requires_cached_image` runtime gate — they skip until `fedora-toolbox:latest` is pre-pulled on the VM (#501 / projectbluefin/lab#621) and activate without a feature-file edit; distrobox enter, JupyterLab, brew, mise remain `@pending` on infra gaps |
 | flatcar | 13 | 12 | 0 | 1 | boot (7 active) + lifecycle (5 active); 1 `@future` (boot from installed target disk — needs KubeVirt boot-order support in `projectbluefin/lab`) |
 | hardware | 13 | 13 | 0 | 0 | udev rules syntax validation (ZSA, Apple SuperDrive, Framework 16, AMD s2idle, Wooting, VIIA); emulated peripherals driven by shared SSH steps |
 | installer | 3 | 3 | 0 | 0 | post-boot assertions for installer-driven installs (UEFI, Flatpak exclusion, LUKS cmdline) |
@@ -255,7 +256,7 @@ skipped-coverage table above.
 | ptyxis: `@brew` (×1) | developer | `@pending` | brew must be initialized first (#487) |
 | ptyxis: `@input`, `@podman`, `@regression`, `@new_tab`, `@close` (×5) | developer | `@pending` | AT-SPI restart issue in CI (#368) — ptyxis reopens between scenarios but the new process isn't reliably accessible |
 | distrobox enter (×1) | dx | `@pending` | pulls `fedora:latest`; no pre-pull in CI, times out |
-| distrobox create/install/export (×3) | dx | `@pending @requires_cached_image` | no cached `fedora-toolbox:latest` on the VM; lab-side OCI image pre-pull required (#501, tracked in projectbluefin/lab#621) |
+| distrobox create/install/export (×3) | dx | `@requires_cached_image` | Active, gated at runtime, **not** `@pending`. `tests/shared/image_cache.py` probes `podman image exists` for the image each scenario names and skips while `fedora-toolbox:latest` is absent from the VM's podman store. Self-activating once the lab-side OCI image pre-pull lands (#501, tracked in projectbluefin/lab#621) — no feature-file edit needed |
 | JupyterLab (×1) | dx | `@pending` | not preinstalled in DX image |
 | brew + mise (×3) | dx | `@pending` | `brew-setup.service` masked (#487) — mise uses brew-installed shims |
 | ujust report confirm validation (×1) | smoke | `@pending` | `just` template change not in the booted image; awaiting rebuild |
diff --git a/scripts/update_coverage_snapshot.py b/scripts/update_coverage_snapshot.py
index c7d14e39..77babaa1 100644
--- a/scripts/update_coverage_snapshot.py
+++ b/scripts/update_coverage_snapshot.py
@@ -46,7 +46,7 @@
     "hardware": "udev rules syntax validation (ZSA, Apple SuperDrive, Framework 16, AMD s2idle, Wooting, VIIA); emulated peripherals driven by shared SSH steps",
     "security": "cosign verify: projectbluefin (bluefin, lts, dakota) + ublue-os (latest, LTS, DX, nvidia, GTS, DX-nvidia, negative)",
     "bazzite": "Extension presence + shell behaviour",
-    "dx": "distrobox enter/create/install/export, JupyterLab, brew, mise — infra gaps, all `@pending`",
+    "dx": "distrobox create/install/export are active behind the `@requires_cached_image` runtime gate — they skip until `fedora-toolbox:latest` is pre-pulled on the VM (#501 / projectbluefin/lab#621) and activate without a feature-file edit; distrobox enter, JupyterLab, brew, mise remain `@pending` on infra gaps",
     "nvidia": "`@future` / `@hardware_blocked` until GPU passthrough exists in the lab",
     "flatcar": "boot (7 active) + lifecycle (5 active); 1 `@future` (boot from installed target disk — needs KubeVirt boot-order support in `projectbluefin/lab`)",
     "kde-smoke": "Plasma session, D-Bus services, AT-SPI tree, KWin output, one KCM, Dolphin, Konsole, Kickoff; all `@informational`",
diff --git a/tests/dx/features/dx_tools.feature b/tests/dx/features/dx_tools.feature
index 2ca97208..8628d17a 100644
--- a/tests/dx/features/dx_tools.feature
+++ b/tests/dx/features/dx_tools.feature
@@ -41,19 +41,24 @@ Feature: Bluefin DX variant smoke tests
   # create a container, install an app inside it, export it to the host.
   # All carry @requires_cached_image: they need a pre-pulled
   # registry.fedoraproject.org/fedora-toolbox:latest on the VM, which no
-  # caching infrastructure provides yet (see projectbluefin/testsuite#501).
-  # They are @pending until projectbluefin/lab provisions that image cache;
-  # tests/shared/quarantine.py skips @pending at runtime.
-  @pending @dx @distrobox @plain_ssh @requires_cached_image
+  # caching infrastructure provides yet (see projectbluefin/testsuite#501,
+  # blocked on projectbluefin/lab#621).
+  #
+  # @requires_cached_image is a runtime capability gate, not a pending marker:
+  # environment.py probes `podman image exists` for the image each scenario
+  # names and skips with an explicit reason while it is absent, exactly as
+  # @requires_bctl does for bluefinctl. Once lab#621 pre-pulls the image these
+  # scenarios activate on their own — no edit to this file.
+  @dx @distrobox @plain_ssh @requires_cached_image
   Scenario: distrobox container can be created from fedora-toolbox
     * DX distrobox "test-box" can be created from "registry.fedoraproject.org/fedora-toolbox:latest"
 
-  @pending @dx @distrobox @plain_ssh @requires_cached_image
+  @dx @distrobox @plain_ssh @requires_cached_image
   Scenario: package can be installed inside a distrobox container
     * DX distrobox "test-box" can be created from "registry.fedoraproject.org/fedora-toolbox:latest"
     * DX distrobox "test-box" installs package "htop"
 
-  @pending @dx @distrobox @plain_ssh @requires_cached_image
+  @dx @distrobox @plain_ssh @requires_cached_image
   Scenario: app inside a distrobox container can be exported to the host
     * DX distrobox "test-box" can be created from "registry.fedoraproject.org/fedora-toolbox:latest"
     * DX distrobox "test-box" installs package "htop"
diff --git a/tests/dx/features/environment.py b/tests/dx/features/environment.py
index 3107dbac..c76371ad 100644
--- a/tests/dx/features/environment.py
+++ b/tests/dx/features/environment.py
@@ -88,10 +88,16 @@ def before_all(context):
 
 
 def before_scenario(context, scenario):
+    from tests.shared.image_cache import skip_when_image_not_cached
     from tests.shared.quarantine import skip_quarantine
 
     if skip_quarantine(scenario):
         return
+    # @requires_cached_image scenarios assume their image is already in the
+    # VM's podman store; running one without it would pull from the registry
+    # mid-scenario and hit the CI timeout instead of reporting a result.
+    if skip_when_image_not_cached(context, scenario):
+        return
     context.scenario = scenario
     context.command_stdout = ""
     context.last_command_output = ""
diff --git a/tests/shared/image_cache.py b/tests/shared/image_cache.py
new file mode 100644
index 00000000..1201d863
--- /dev/null
+++ b/tests/shared/image_cache.py
@@ -0,0 +1,167 @@
+"""Runtime gate for scenarios that need a pre-cached OCI image on the DUT.
+
+Scenarios tagged ``@requires_cached_image`` pull no image themselves: they
+assume the image they name is already in the device-under-test's local podman
+store.  When it is not, a live registry pull would run inside the scenario and
+blow the CI timeout instead of reporting a useful result, so the scenario is
+skipped with an explicit reason.
+
+This is the same "skip until the capability exists, then activate
+automatically" contract as ``@requires_bctl`` and ``@requires_toggle_action``
+in ``tests/common/features/environment.py``.  It is deliberately NOT a
+non-runnable tag: it does not belong in ``tests/shared/quarantine.py`` or in
+the CI tag filters, because the scenario must run the moment the image is
+cached — no feature-file edit required.
+
+The images a scenario needs are read from its own step text rather than from a
+constant here, so the feature file stays the single source of truth for which
+image is under test.
+"""
+
+from __future__ import annotations
+
+import re
+import shlex
+import subprocess
+
+
+REQUIRES_CACHED_IMAGE_TAG = "requires_cached_image"
+
+# Quoted arguments in a behave step name, e.g. the "…fedora-toolbox:latest" in
+#   * DX distrobox "test-box" can be created from "registry.fedoraproject.org/fedora-toolbox:latest"
+_QUOTED = re.compile(r'"([^"]*)"')
+
+
+def _looks_like_image_ref(token: str) -> bool:
+    """Return True when ``token`` is an OCI image reference.
+
+    Deliberately strict so ordinary quoted step arguments (container names,
+    package names, absolute paths like ``/usr/bin/htop``) are never mistaken
+    for images:
+
+    * a registry host — the segment before the first ``/`` must contain a dot
+      or a port, or be ``localhost``;
+    * a tag or digest on the final path segment.
+
+    Bare references such as ``fedora:latest`` are rejected: without an explicit
+    registry, what ``podman image exists`` resolves depends on the DUT's
+    ``registries.conf`` search list, so the probe could disagree with the pull
+    the scenario would perform.
+    """
+    if not token or any(char.isspace() for char in token):
+        return False
+    registry, _, remainder = token.partition("/")
+    if not remainder:
+        return False
+    if "." not in registry and ":" not in registry and registry != "localhost":
+        return False
+    final_segment = remainder.rsplit("/", 1)[-1]
+    return ":" in final_segment or "@" in final_segment
+
+
+def images_required_by(scenario) -> list[str]:
+    """Return the image references named in ``scenario``'s steps, in order.
+
+    Duplicates are collapsed: several steps in one scenario normally name the
+    same image, and probing it once is enough.
+    """
+    images: list[str] = []
+    for step in getattr(scenario, "steps", []) or []:
+        for token in _QUOTED.findall(getattr(step, "name", "") or ""):
+            if _looks_like_image_ref(token) and token not in images:
+                images.append(token)
+    return images
+
+
+def _ssh_returncode(context, command: str, timeout: int) -> int:
+    """Run ``command`` on the DUT over SSH and return its exit status.
+
+    This deliberately does NOT reuse ``run_ssh`` from
+    ``tests/shared/ssh_steps.py``. Importing that module registers its ``@step``
+    phrases into behave's global registry, and several of them
+    (``SSH command return code is "{code}"``,
+    ``Last command output contains "{text}"``) are also defined by
+    ``tests/dx/features/steps/steps.py``. Importing it from a DX
+    ``before_scenario`` hook would raise ``AmbiguousStep`` and take down the
+    whole suite. ``tests/shared/ssh_config.py`` is the connection-detail helper
+    that carries no step definitions, which is exactly what a hook needs.
+
+    Unlike a step, this probe leaves ``context`` untouched: it runs before the
+    per-scenario state reset and must not smear command output across scenarios.
+    """
+    from tests.shared.ssh_config import resolve_ssh_details
+
+    details = resolve_ssh_details(context)
+    argv = [
+        "ssh",
+        "-i", details["ssh_key"],
+        "-o", "StrictHostKeyChecking=no",
+        "-o", "UserKnownHostsFile=/dev/null",
+        "-o", "ConnectTimeout=10",
+        "-o", "LogLevel=ERROR",
+    ]
+    if details.get("ssh_port"):
+        argv += ["-p", str(details["ssh_port"])]
+    argv += [f"{details['ssh_user']}@{details['vm_ip']}", command]
+    return subprocess.run(argv, capture_output=True, text=True, timeout=timeout).returncode
+
+
+def image_is_cached(context, image: str, timeout: int = 30) -> bool:
+    """Return True when ``image`` is already in the DUT's local podman store.
+
+    ``podman image exists`` exits 0 for a locally present image and 1
+    otherwise, and never contacts a registry — which is the whole point: the
+    probe itself must not be able to trigger the pull it is guarding against.
+    """
+    try:
+        returncode = _ssh_returncode(
+            context, f"podman image exists {shlex.quote(image)}", timeout
+        )
+    except (subprocess.TimeoutExpired, OSError):
+        # An unreachable DUT is not evidence that the image is cached.
+        return False
+    return returncode == 0
+
+
+def _skip(scenario, reason: str) -> None:
+    """Skip ``scenario``, tolerating behave versions whose skip() takes no reason."""
+    try:
+        scenario.skip(reason)
+    except TypeError:
+        scenario.skip()
+
+
+def skip_when_image_not_cached(context, scenario) -> bool:
+    """Skip ``scenario`` unless every image it names is cached on the DUT.
+
+    Returns True when the scenario was skipped, so ``before_scenario`` hooks
+    can mirror the ``skip_quarantine`` call shape::
+
+        if skip_when_image_not_cached(context, scenario):
+            return
+    """
+    tags = set(getattr(scenario, "effective_tags", scenario.tags))
+    if REQUIRES_CACHED_IMAGE_TAG not in tags:
+        return False
+
+    images = images_required_by(scenario)
+    if not images:
+        # Authoring error, not an environment condition: the tag promises an
+        # image the steps never name.  tests/unit/test_image_cache.py fails on
+        # this against the real feature files, so it cannot reach CI silently.
+        _skip(
+            scenario,
+            f"@{REQUIRES_CACHED_IMAGE_TAG} — no image reference found in the "
+            "scenario's steps; tag it on a scenario that names its image",
+        )
+        return True
+
+    missing = [image for image in images if not image_is_cached(context, image)]
+    if missing:
+        _skip(
+            scenario,
+            f"@{REQUIRES_CACHED_IMAGE_TAG} — not in the DUT's local podman "
+            f"store: {', '.join(missing)}",
+        )
+        return True
+    return False
diff --git a/tests/unit/test_image_cache.py b/tests/unit/test_image_cache.py
new file mode 100644
index 00000000..48956fa6
--- /dev/null
+++ b/tests/unit/test_image_cache.py
@@ -0,0 +1,368 @@
+"""Unit tests for tests/shared/image_cache.py (#501)."""
+
+from __future__ import annotations
+
+import re
+import subprocess
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from tests.shared import image_cache
+from tests.shared.image_cache import (
+    REQUIRES_CACHED_IMAGE_TAG,
+    _looks_like_image_ref,
+    images_required_by,
+    skip_when_image_not_cached,
+)
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+TOOLBOX_IMAGE = "registry.fedoraproject.org/fedora-toolbox:latest"
+
+
+# ── helpers ───────────────────────────────────────────────────────────────────
+
+
+class _FakeStep:
+    def __init__(self, name: str):
+        self.name = name
+
+
+class _FakeScenario:
+    """Minimal behave Scenario stand-in for unit testing."""
+
+    def __init__(self, tags=None, steps=None, *, skip_raises=False):
+        self.tags = list(tags or [])
+        self.effective_tags = self.tags
+        self.steps = [_FakeStep(text) for text in (steps or [])]
+        self.skip_raises = skip_raises
+        self.skip_message: str | None = None
+        self.skipped = False
+
+    def skip(self, message: str | None = None) -> None:
+        if self.skip_raises and message is not None:
+            raise TypeError("skip() takes no arguments")
+        self.skipped = True
+        self.skip_message = message
+
+
+class _FakeSSH:
+    """Stand-in for image_cache._ssh_returncode, recording probe commands."""
+
+    def __init__(self, cached=(), raises=None):
+        self.cached = set(cached)
+        self.raises = raises
+        self.commands: list[str] = []
+
+    def __call__(self, context, command, timeout=30):
+        self.commands.append(command)
+        if self.raises is not None:
+            raise self.raises
+        image = command.rsplit(" ", 1)[-1].strip("'\"")
+        return 0 if image in self.cached else 1
+
+
+@pytest.fixture
+def fake_ssh(monkeypatch):
+    """Install a fake SSH probe; returns a factory the test configures."""
+
+    def _install(cached=(), raises=None):
+        ssh = _FakeSSH(cached=cached, raises=raises)
+        monkeypatch.setattr(image_cache, "_ssh_returncode", ssh)
+        return ssh
+
+    return _install
+
+
+def _distrobox_scenario(tags=("requires_cached_image",), image=TOOLBOX_IMAGE):
+    return _FakeScenario(
+        tags=list(tags),
+        steps=[
+            f'DX distrobox "test-box" can be created from "{image}"',
+            'DX distrobox "test-box" installs package "htop"',
+            'DX distrobox "test-box" exports "/usr/bin/htop" to the host',
+        ],
+    )
+
+
+# ── _looks_like_image_ref ─────────────────────────────────────────────────────
+
+
+class TestLooksLikeImageRef:
+    @pytest.mark.parametrize(
+        "token",
+        [
+            TOOLBOX_IMAGE,
+            "ghcr.io/ublue-os/bluefin-dx:latest",
+            "localhost/my-image:dev",
+            "registry.example.com:5000/team/app:1.2.3",
+            "quay.io/fedora/fedora@sha256:abc123",
+        ],
+    )
+    def test_accepts_registry_qualified_refs(self, token):
+        assert _looks_like_image_ref(token) is True
+
+    @pytest.mark.parametrize(
+        "token",
+        [
+            "",
+            "test-box",
+            "htop",
+            "/usr/bin/htop",
+            "~/.local/bin",
+            "fedora:latest",  # no registry: resolution depends on registries.conf
+            "ghcr.io/ublue-os/bluefin-dx",  # no tag or digest
+            "registry.example.com/some image:latest",  # whitespace
+        ],
+    )
+    def test_rejects_everything_else(self, token):
+        assert _looks_like_image_ref(token) is False
+
+
+# ── images_required_by ────────────────────────────────────────────────────────
+
+
+class TestImagesRequiredBy:
+    def test_extracts_the_image_from_step_text(self):
+        assert images_required_by(_distrobox_scenario()) == [TOOLBOX_IMAGE]
+
+    def test_collapses_duplicate_references(self):
+        scenario = _FakeScenario(
+            steps=[
+                f'DX distrobox "a" can be created from "{TOOLBOX_IMAGE}"',
+                f'DX distrobox "b" can be created from "{TOOLBOX_IMAGE}"',
+            ]
+        )
+        assert images_required_by(scenario) == [TOOLBOX_IMAGE]
+
+    def test_preserves_order_of_distinct_images(self):
+        other = "ghcr.io/ublue-os/bluefin-dx:latest"
+        scenario = _FakeScenario(
+            steps=[
+                f'DX distrobox "a" can be created from "{TOOLBOX_IMAGE}"',
+                f'DX distrobox "b" can be created from "{other}"',
+            ]
+        )
+        assert images_required_by(scenario) == [TOOLBOX_IMAGE, other]
+
+    def test_ignores_non_image_arguments(self):
+        scenario = _FakeScenario(
+            steps=['DX distrobox "test-box" exports "/usr/bin/htop" to the host']
+        )
+        assert images_required_by(scenario) == []
+
+    def test_tolerates_a_scenario_without_steps(self):
+        assert images_required_by(_FakeScenario()) == []
+
+
+# ── skip_when_image_not_cached ────────────────────────────────────────────────
+
+
+class TestSkipWhenImageNotCached:
+    def test_untagged_scenario_is_never_probed(self, fake_ssh):
+        ssh = fake_ssh(cached=())
+        scenario = _distrobox_scenario(tags=("dx", "distrobox"))
+
+        assert skip_when_image_not_cached(None, scenario) is False
+        assert not scenario.skipped
+        assert ssh.commands == []
+
+    def test_runs_when_the_image_is_cached(self, fake_ssh):
+        fake_ssh(cached=(TOOLBOX_IMAGE,))
+        scenario = _distrobox_scenario()
+
+        assert skip_when_image_not_cached(None, scenario) is False
+        assert not scenario.skipped
+
+    def test_skips_when_the_image_is_absent(self, fake_ssh):
+        fake_ssh(cached=())
+        scenario = _distrobox_scenario()
+
+        assert skip_when_image_not_cached(None, scenario) is True
+        assert scenario.skipped
+        assert TOOLBOX_IMAGE in scenario.skip_message
+        assert REQUIRES_CACHED_IMAGE_TAG in scenario.skip_message
+
+    def test_probes_each_distinct_image_once(self, fake_ssh):
+        ssh = fake_ssh(cached=(TOOLBOX_IMAGE,))
+        skip_when_image_not_cached(None, _distrobox_scenario())
+
+        assert ssh.commands == [f"podman image exists {TOOLBOX_IMAGE}"]
+
+    def test_skips_when_the_tag_names_no_image(self, fake_ssh):
+        fake_ssh(cached=(TOOLBOX_IMAGE,))
+        scenario = _FakeScenario(
+            tags=[REQUIRES_CACHED_IMAGE_TAG],
+            steps=['DX distrobox "test-box" installs package "htop"'],
+        )
+
+        assert skip_when_image_not_cached(None, scenario) is True
+        assert "no image reference found" in scenario.skip_message
+
+    def test_skip_message_is_optional_for_older_behave(self, fake_ssh):
+        fake_ssh(cached=())
+        scenario = _distrobox_scenario()
+        scenario.skip_raises = True
+
+        assert skip_when_image_not_cached(None, scenario) is True
+        assert scenario.skipped
+
+
+class _FakeCompletedProcess:
+    def __init__(self, returncode):
+        self.returncode = returncode
+        self.stdout = ""
+        self.stderr = ""
+
+
+def _ssh_context():
+    """Context carrying the connection details resolve_ssh_details() reads."""
+    context = MagicMock()
+    context.ssh_key = "/tmp/key"
+    context.vm_ip = "192.0.2.10"
+    context.ssh_user = "bluefin-test"
+    context.ssh_port = "2222"
+    context.config.userdata = {}
+    return context
+
+
+class TestImageIsCached:
+    def _patch_subprocess(self, monkeypatch, result):
+        calls: list[list[str]] = []
+
+        def fake_run(argv, **kwargs):
+            calls.append(argv)
+            if isinstance(result, BaseException):
+                raise result
+            return _FakeCompletedProcess(result)
+
+        monkeypatch.setattr(image_cache.subprocess, "run", fake_run)
+        return calls
+
+    def test_true_when_podman_image_exists_succeeds(self, monkeypatch):
+        self._patch_subprocess(monkeypatch, 0)
+        assert image_cache.image_is_cached(_ssh_context(), TOOLBOX_IMAGE) is True
+
+    def test_false_when_podman_image_exists_fails(self, monkeypatch):
+        self._patch_subprocess(monkeypatch, 1)
+        assert image_cache.image_is_cached(_ssh_context(), TOOLBOX_IMAGE) is False
+
+    def test_false_when_the_dut_is_unreachable(self, monkeypatch):
+        self._patch_subprocess(monkeypatch, OSError("no route to host"))
+        assert image_cache.image_is_cached(_ssh_context(), TOOLBOX_IMAGE) is False
+
+    def test_false_when_the_probe_times_out(self, monkeypatch):
+        self._patch_subprocess(monkeypatch, subprocess.TimeoutExpired("ssh", 30))
+        assert image_cache.image_is_cached(_ssh_context(), TOOLBOX_IMAGE) is False
+
+    def test_probe_never_contacts_a_registry(self, monkeypatch):
+        """`podman image exists` is local-only — a pulling probe would defeat the gate."""
+        calls = self._patch_subprocess(monkeypatch, 0)
+        image_cache.image_is_cached(_ssh_context(), TOOLBOX_IMAGE)
+
+        remote_command = calls[0][-1]
+        assert remote_command == f"podman image exists {TOOLBOX_IMAGE}"
+        assert "pull" not in remote_command
+
+    def test_probe_uses_the_resolved_connection_details(self, monkeypatch):
+        calls = self._patch_subprocess(monkeypatch, 0)
+        image_cache.image_is_cached(_ssh_context(), TOOLBOX_IMAGE)
+
+        argv = calls[0]
+        assert argv[0] == "ssh"
+        assert "bluefin-test@192.0.2.10" in argv
+        assert argv[argv.index("-i") + 1] == "/tmp/key"
+        assert argv[argv.index("-p") + 1] == "2222"
+
+    def test_probe_does_not_mutate_context_command_state(self, monkeypatch):
+        """The gate runs before before_scenario resets state; it must not smear output."""
+        self._patch_subprocess(monkeypatch, 0)
+        context = _ssh_context()
+        context.command_stdout = "sentinel"
+        context.ssh_rc = 99
+
+        image_cache.image_is_cached(context, TOOLBOX_IMAGE)
+
+        assert context.command_stdout == "sentinel"
+        assert context.ssh_rc == 99
+
+
+def test_gate_does_not_import_the_shared_ssh_step_library():
+    """Importing tests/shared/ssh_steps.py from a hook would raise AmbiguousStep.
+
+    It registers `SSH command return code is "{code}"` and
+    `Last command output contains "{text}"`, which tests/dx/features/steps/steps.py
+    also defines. A DX before_scenario hook that pulled it in would take the
+    whole suite down, so the probe must stay on tests/shared/ssh_config.py.
+    """
+    source = (REPO_ROOT / "tests/shared/image_cache.py").read_text(encoding="utf-8")
+    importing_lines = [
+        line
+        for line in source.splitlines()
+        if line.strip().startswith(("import ", "from ")) and "ssh_steps" in line
+    ]
+    assert not importing_lines, (
+        f"image_cache.py must not import ssh_steps: {importing_lines}"
+    )
+
+
+# ── the tag must never be inert in the real feature files ─────────────────────
+
+
+_SCENARIO_BLOCK = re.compile(
+    r"^[ \t]*(@[^\n]*\n[ \t]*)*@[^\n]*\brequires_cached_image\b[^\n]*\n"
+    r"(?P<body>(?:[ \t]*(?:Scenario|Scenario Outline):[^\n]*\n)(?:(?![ \t]*@|[ \t]*Scenario).*\n)*)",
+    re.MULTILINE,
+)
+
+
+def _tagged_scenario_bodies():
+    for feature in sorted(REPO_ROOT.glob("tests/*/features/**/*.feature")):
+        text = feature.read_text(encoding="utf-8")
+        if REQUIRES_CACHED_IMAGE_TAG not in text:
+            continue
+        for match in _SCENARIO_BLOCK.finditer(text):
+            yield feature, match.group("body")
+
+
+def test_every_tagged_scenario_names_an_image():
+    """A @requires_cached_image scenario whose steps name no image would skip forever.
+
+    The runtime gate reads the image out of the scenario's own steps, so a tag
+    applied to a scenario that never names one is an authoring error. Catching
+    it here keeps it out of CI, where it would look like an infra skip.
+    """
+    bodies = list(_tagged_scenario_bodies())
+    assert bodies, "expected at least one @requires_cached_image scenario"
+
+    for feature, body in bodies:
+        scenario = _FakeScenario(steps=body.splitlines())
+        assert images_required_by(scenario), (
+            f"{feature.relative_to(REPO_ROOT)}: @{REQUIRES_CACHED_IMAGE_TAG} scenario "
+            f"names no registry-qualified image:\n{body}"
+        )
+
+
+def test_tagged_scenarios_are_not_masked_by_a_non_runnable_tag():
+    """The gate only runs for scenarios the runtime actually reaches.
+
+    tests/shared/quarantine.py skips @quarantine/@hardware_blocked/@future/@pending
+    before this gate ever sees the scenario, so pairing @requires_cached_image
+    with one of them makes it inert — the masking trap documented in
+    docs/skills/test-authoring/suite-map/SKILL.md.
+    """
+    from tests.shared.quarantine import _SKIP_TAGS
+
+    for feature in sorted(REPO_ROOT.glob("tests/*/features/**/*.feature")):
+        for line in feature.read_text(encoding="utf-8").splitlines():
+            stripped = line.strip()
+            if not stripped.startswith("@") or REQUIRES_CACHED_IMAGE_TAG not in stripped:
+                continue
+            tags = {tag.lstrip("@") for tag in stripped.split()}
+            masking = tags.intersection(_SKIP_TAGS)
+            assert not masking, (
+                f"{feature.relative_to(REPO_ROOT)}: @{REQUIRES_CACHED_IMAGE_TAG} is masked "
+                f"by {sorted(masking)} — the gate never runs for this scenario"
+            )