diff --git a/cloud_pipelines_backend/instrumentation/execution_annotations.py b/cloud_pipelines_backend/instrumentation/execution_annotations.py
index 7e9cc73..154a3d2 100644
--- a/cloud_pipelines_backend/instrumentation/execution_annotations.py
+++ b/cloud_pipelines_backend/instrumentation/execution_annotations.py
@@ -6,6 +6,7 @@
 load would put a SELECT in the commit path.
 """
 
+import enum
 import json
 import re
 from typing import NamedTuple
@@ -182,3 +183,148 @@ def accelerators(*, execution: "bts.ExecutionNode") -> tuple[str, int] | None:
         raise ValueError(f"Multiple accelerator types were specified: {requested=}")
     model, quantity = next(iter(requested.items()))
     return model, int(quantity)
+
+
+class PendingReason(str, enum.Enum):
+    """Why a task is not running yet, in our words rather than Kubernetes'.
+
+    A closed set. A Kubernetes release that invents a new waiting reason widens OTHER
+    rather than adding a label value, so the metric cannot grow a series behind us.
+    """
+
+    # What:    nothing has placed this pod on a machine yet.
+    # Fault:   the cluster's capacity, not the task's configuration.
+    # K8s:     a Pending pod with no containerStatuses at all -- there is nothing to
+    #          report on until something schedules it, so this one has no entry in
+    #          the mapping below.
+    # Example: a job asking for an accelerator (e.g. a B300 GPU) while every one
+    #          of them is already taken.
+    WAITING_FOR_NODE = "waiting_for_node"
+
+    # What:    the pod has a machine and the kubelet is bringing it up.
+    # Fault:   nobody's, until it lasts.
+    # Example: a very large image, still pulling ten minutes in.
+    STARTING_CONTAINER = "starting_container"
+
+    # What:    the image cannot be fetched.
+    # Fault:   the task's -- a tag that does not exist, or credentials the registry
+    #          rejected. Kubernetes does not separate a typo from a permissions
+    #          failure here, so neither can we.
+    # Example: a task pinned to an image tag that was never pushed.
+    IMAGE_UNAVAILABLE = "image_unavailable"
+
+    # What:    the pod spec names something missing or unreadable.
+    # Fault:   the task's own configuration or permissions, not the cluster's.
+    # Example: a renamed Secret that one pipeline still references.
+    BAD_CONFIG = "bad_config"
+
+    # What:    a waiting reason we have not mapped.
+    # Fault:   unknown, which is the point -- it stays out of the label set until
+    #          someone decides which bucket it belongs in.
+    # Example: a string a future Kubernetes release invents.
+    OTHER = "other"
+
+
+# Kubernetes' open vocabulary on the left, ours on the right. What each of ours means
+# is on the enum above; these are the strings the kubelet actually reports.
+#
+# The pod lifecycle they come from, top to bottom:
+#
+#   pod created
+#     |
+#     +-- nothing schedules it ................... (no containerStatuses)
+#     v
+#   scheduled to a node
+#     |
+#     +-- image reference malformed .............. InvalidImageName
+#     v
+#   init containers running .................... PodInitializing
+#     |
+#     v
+#   main container being created ............... ContainerCreating
+#     |
+#     +-- pull failed ............................ ErrImagePull
+#     |     +-- and kept failing ................. ImagePullBackOff
+#     +-- manifest unreadable .................... ImageInspectError
+#     +-- registry silent ........................ RegistryUnavailable
+#     +-- Secret or ConfigMap missing ............ CreateContainerConfigError
+#     +-- runtime refused the spec ............... CreateContainerError
+#     |
+#     v
+#   RUNNING
+#
+# The common path, not a state machine: ContainerCreating covers the image pull too,
+# so the kubelet may report either it or ErrImagePull for the same moment.
+_PENDING_REASONS = {
+    # The kubelet is bringing the pod up, in this order:
+    #   PodInitializing ──then──► ContainerCreating
+    #   init containers           the main container
+    #
+    # Init containers are still running, so the main container has not started.
+    "PodInitializing": PendingReason.STARTING_CONTAINER,
+    # The main container is being created: pulling layers, mounting volumes.
+    "ContainerCreating": PendingReason.STARTING_CONTAINER,
+    # The image did not arrive.
+    #   ErrImagePull ──retries──► ImagePullBackOff     one incident, two names
+    #
+    # The first pull attempt failed.
+    "ErrImagePull": PendingReason.IMAGE_UNAVAILABLE,
+    # The back-off loop the above decays into after repeated failures.
+    "ImagePullBackOff": PendingReason.IMAGE_UNAVAILABLE,
+    # The image was fetched but its manifest could not be read.
+    "ImageInspectError": PendingReason.IMAGE_UNAVAILABLE,
+    # The registry did not answer at all.
+    "RegistryUnavailable": PendingReason.IMAGE_UNAVAILABLE,
+    # The spec is wrong or not permitted, caught at two different moments:
+    #   InvalidImageName                                before any pull
+    #   CreateContainerConfigError, CreateContainerError at container creation
+    #
+    # The image reference is malformed, so nothing is ever pulled.
+    "InvalidImageName": PendingReason.BAD_CONFIG,
+    # A Secret, ConfigMap or token the spec names is missing or unreadable.
+    "CreateContainerConfigError": PendingReason.BAD_CONFIG,
+    # The runtime refused to create the container from this spec.
+    "CreateContainerError": PendingReason.BAD_CONFIG,
+}
+
+
+# Written onto ContainerExecution.extra_data while a task waits, read back when it stops
+# waiting -- by then the pod is running and the reason is gone from it.
+PENDING_REASON_KEY = "last_pending_reason"
+
+
+def pending_reason(*, launcher_data: dict) -> PendingReason | None:
+    """Why a task is not running yet, in our words rather than Kubernetes'.
+
+        a pod with no containerStatuses yet  → "waiting_for_node"      (a)
+        waiting.reason "ContainerCreating"   → "starting_container"
+        waiting.reason "ImagePullBackOff"    → "image_unavailable"
+        waiting.reason "CreateContainerConfigError"  → "bad_config"
+        waiting.reason "SomeNewK8sString"    → "other"                 (b)
+        a pod whose phase is not Pending     → None                    (c)
+
+        (a) scheduling has not placed it, so there is no container to report on
+        (b) a Kubernetes release adding a reason must not add a label value
+        (c) nothing is waiting, so there is nothing to attribute
+
+    Reads the serialised pod already persisted on ContainerExecution.launcher_data, so
+    no Kubernetes call and no database read happen here.
+    """
+    root = launcher_data.get("kubernetes_job") or launcher_data.get("kubernetes") or {}
+    pods = root.get("debug_pods")
+    if pods is None:
+        single = root.get("debug_pod")
+        pods = {"0": single} if single else {}
+    for pod in pods.values():
+        status = (pod or {}).get("status") or {}
+        if status.get("phase") != "Pending":
+            continue
+        container_statuses = status.get("containerStatuses")
+        if not container_statuses:
+            return PendingReason.WAITING_FOR_NODE
+        waiting = (container_statuses[0].get("state") or {}).get("waiting") or {}
+        reason = waiting.get("reason")
+        if reason:
+            return _PENDING_REASONS.get(reason, PendingReason.OTHER)
+        return PendingReason.WAITING_FOR_NODE
+    return None
diff --git a/cloud_pipelines_backend/instrumentation/metrics.py b/cloud_pipelines_backend/instrumentation/metrics.py
index 6f4446d..5d96990 100644
--- a/cloud_pipelines_backend/instrumentation/metrics.py
+++ b/cloud_pipelines_backend/instrumentation/metrics.py
@@ -324,7 +324,8 @@ def _count_slow_transitions_by_owner(session: orm.Session) -> None:
     try:
         with orm.Session(bind=session.get_bind()) as lookup_session:
             owners = _owners_of(
-                session=lookup_session, node_ids={node_id for node_id, _, _ in stashed}
+                session=lookup_session,
+                node_ids={node_id for node_id, _, _, _ in stashed},
             )
     except Exception:
         _logger.warning(
@@ -332,7 +333,7 @@ def _count_slow_transitions_by_owner(session: orm.Session) -> None:
             len(stashed),
             exc_info=True,
         )
-    for node_id, status_from, status_to in stashed:
+    for node_id, status_from, status_to, pending_reason in stashed:
         try:
             execution_slow_transitions.add(
                 1,
@@ -340,6 +341,7 @@ def _count_slow_transitions_by_owner(session: orm.Session) -> None:
                     "execution.status.from": status_from,
                     "execution.status.to": status_to,
                     "owner": owners.get(node_id, _UNSET),
+                    "execution.pending_reason": pending_reason,
                 },
             )
         except Exception:
@@ -349,6 +351,63 @@ def _count_slow_transitions_by_owner(session: orm.Session) -> None:
             )
 
 
+@sql_event.listens_for(backend_types_sql.ContainerExecution.launcher_data, "set")
+def _persist_pending_reason_on_extra_data(
+    container_execution: "backend_types_sql.ContainerExecution",
+    value: dict | None,
+    _old_value: object,
+    _initiator: object,
+) -> None:
+    """Save why a task is waiting onto its row, while it is still waiting.
+
+        poll, pod Pending, no containers   → extra_data = {"last_pending_reason": "waiting_for_node"}
+        poll, pod Pending, ImagePullBackOff → extra_data = {"last_pending_reason": "image_unavailable"}
+        poll, pod Running                   → left alone, the last reason stands   (a)
+
+        (a) by the time a task stops waiting its pod is running and the reason is gone,
+            so the transition can only read what an earlier poll wrote
+
+    The orchestrator assigns launcher_data on every sweep that sees a change, which is
+    the only thing in the system that updates while the status does not. The serialised
+    pod arrives as this event's own argument, so nothing is read from the database.
+    """
+    if not value:
+        return
+    reason = execution_annotations.pending_reason(launcher_data=value)
+    if reason is None:
+        return
+    extra_data = container_execution.extra_data or {}
+    if extra_data.get(execution_annotations.PENDING_REASON_KEY) == reason.value:
+        return
+    container_execution.extra_data = {
+        **extra_data,
+        # .value, not the member: extra_data is a JSON column and an attribute is a
+        # string, so the enum stops at this boundary.
+        execution_annotations.PENDING_REASON_KEY: reason.value,
+    }
+
+
+def _pending_reason(*, execution: "backend_types_sql.ExecutionNode") -> str:
+    """Why this task was waiting, as stamped by the most recent poll.
+
+        the ContainerExecution is already in the session  → "image_unavailable"
+        it would have to be fetched                       → "none"   (a)
+        nothing ever stamped one                          → "none"
+
+        (a) this runs in before_commit, where a lazy load would put a SELECT inside the
+            flush in progress; an unlabelled point beats a re-entrant query
+
+    reads: ContainerExecution.extra_data["last_pending_reason"]
+    """
+    if "container_execution" in sql.inspect(execution).unloaded:
+        return _UNSET
+    container_execution = execution.container_execution
+    if container_execution is None:
+        return _UNSET
+    extra_data = container_execution.extra_data or {}
+    return extra_data.get(execution_annotations.PENDING_REASON_KEY) or _UNSET
+
+
 @sql_event.listens_for(orm.Session, "before_commit")
 def _handle_before_commit(session: orm.Session) -> None:
     for obj in list(session.new) + list(session.dirty):
@@ -365,6 +424,7 @@ def _handle_before_commit(session: orm.Session) -> None:
             accelerator_type, accelerator_count = _accelerator(execution=obj)
             cluster = _cluster(execution=obj)
             namespace = _namespace(execution=obj)
+            pending_reason = _pending_reason(execution=obj)
             try:
                 execution_status_transition_duration.record(
                     (curr_time - prev_time).total_seconds(),
@@ -377,12 +437,13 @@ def _handle_before_commit(session: orm.Session) -> None:
                         "execution.accelerator_count": accelerator_count,
                         "execution.cluster": cluster,
                         "k8s.namespace.name": namespace,
+                        "execution.pending_reason": pending_reason,
                     },
                 )
                 waited = (curr_time - prev_time).total_seconds()
                 if prev["status"] != _RUNNING and waited >= _SLOW_TRANSITION_SECONDS:
                     session.info.setdefault(_SLOW_KEY, []).append(
-                        (obj.id, prev["status"], curr["status"])
+                        (obj.id, prev["status"], curr["status"], pending_reason)
                     )
             except Exception:
                 _logger.warning(
diff --git a/tests/test_sql_event_listeners.py b/tests/test_sql_event_listeners.py
index 28b5f36..18220ef 100644
--- a/tests/test_sql_event_listeners.py
+++ b/tests/test_sql_event_listeners.py
@@ -76,6 +76,7 @@ def test_second_status_change_records_duration_metric(
             "execution.accelerator_count": "0",
             "execution.cluster": "none",
             "k8s.namespace.name": "none",
+            "execution.pending_reason": "none",
         }
 
 
@@ -659,3 +660,206 @@ def _snapshot(**_kwargs: object) -> dict:
             for s in issued_before_the_owner_read
             if s.lstrip().upper().startswith("SELECT")
         ]
+
+
+class TestPendingReason:
+    """Why a task is waiting, stamped while it waits and read when it stops."""
+
+    @staticmethod
+    def _launcher_data(*, phase: str, reason: str | None, has_container: bool = True):
+        """The shape the Kubernetes launcher serialises onto launcher_data."""
+        status: dict = {"phase": phase}
+        if has_container:
+            state = {"waiting": {"reason": reason}} if reason else {}
+            status["containerStatuses"] = [{"state": state}]
+        return {"kubernetes_job": {"debug_pods": {"0": {"status": status}}}}
+
+    @pytest.mark.parametrize(
+        "reason, expected",
+        [
+            ("ContainerCreating", "starting_container"),
+            ("PodInitializing", "starting_container"),
+            ("ImagePullBackOff", "image_unavailable"),
+            ("ErrImagePull", "image_unavailable"),
+            ("CreateContainerConfigError", "bad_config"),
+            ("InvalidImageName", "bad_config"),
+        ],
+    )
+    def test_a_kubernetes_reason_becomes_one_of_ours(self, reason, expected):
+        data = self._launcher_data(phase="Pending", reason=reason)
+        assert execution_annotations.pending_reason(launcher_data=data) == expected
+
+    def test_an_unknown_kubernetes_reason_does_not_add_a_label_value(self):
+        data = self._launcher_data(phase="Pending", reason="SomeFutureK8sReason")
+        assert execution_annotations.pending_reason(launcher_data=data) == "other"
+
+    def test_a_pod_with_no_containers_yet_is_waiting_for_a_node(self):
+        data = self._launcher_data(phase="Pending", reason=None, has_container=False)
+        assert execution_annotations.pending_reason(launcher_data=data) == (
+            "waiting_for_node"
+        )
+
+    def test_a_running_pod_is_not_waiting_for_anything(self):
+        data = self._launcher_data(phase="Running", reason=None)
+        assert execution_annotations.pending_reason(launcher_data=data) is None
+
+    def test_a_launcher_that_is_not_kubernetes_yields_no_reason(self):
+        assert execution_annotations.pending_reason(launcher_data={"skypilot": {}}) is (
+            None
+        )
+
+    def test_a_poll_stamps_the_reason_onto_the_container_execution(
+        self, session: orm.Session
+    ):
+        container_execution = bts.ContainerExecution(
+            status=bts.ContainerExecutionStatus.PENDING
+        )
+        session.add(container_execution)
+        container_execution.launcher_data = self._launcher_data(
+            phase="Pending", reason="ImagePullBackOff"
+        )
+        session.commit()
+        assert container_execution.extra_data == {
+            execution_annotations.PENDING_REASON_KEY: "image_unavailable"
+        }
+
+    def test_what_lands_in_the_json_column_is_a_plain_string(
+        self, session: orm.Session
+    ):
+        """A str-valued enum passes == either way, so pin the stored type itself."""
+        container_execution = bts.ContainerExecution(
+            status=bts.ContainerExecutionStatus.PENDING
+        )
+        session.add(container_execution)
+        container_execution.launcher_data = self._launcher_data(
+            phase="Pending", reason="ImagePullBackOff"
+        )
+        # Before the commit: afterwards SQLite round-trips the JSON and hands back a
+        # plain string whatever was put in, which would hide the difference.
+        stored = container_execution.extra_data[
+            execution_annotations.PENDING_REASON_KEY
+        ]
+        assert type(stored) is str
+
+    def test_the_reason_survives_the_pod_starting_to_run(self, session: orm.Session):
+        """The flip is exactly when the pod no longer says why it was waiting."""
+        container_execution = bts.ContainerExecution(
+            status=bts.ContainerExecutionStatus.PENDING
+        )
+        session.add(container_execution)
+        container_execution.launcher_data = self._launcher_data(
+            phase="Pending", reason="ImagePullBackOff"
+        )
+        session.commit()
+        container_execution.launcher_data = self._launcher_data(
+            phase="Running", reason=None
+        )
+        session.commit()
+        assert container_execution.extra_data == {
+            execution_annotations.PENDING_REASON_KEY: "image_unavailable"
+        }
+
+    def test_a_node_reads_the_reason_off_its_container_execution(
+        self, session: orm.Session
+    ):
+        container_execution = bts.ContainerExecution(
+            status=bts.ContainerExecutionStatus.PENDING
+        )
+        node = bts.ExecutionNode(task_spec={})
+        node.container_execution = container_execution
+        session.add(node)
+        container_execution.launcher_data = self._launcher_data(
+            phase="Pending", reason="CreateContainerConfigError"
+        )
+        session.commit()
+        # The orchestrator holds both objects when it flips a status, which is the only
+        # time this reader runs; commit expires them, so reload as that caller would.
+        node.container_execution
+        assert metrics._pending_reason(execution=node) == "bad_config"
+
+    def test_a_reason_behind_a_lazy_load_is_skipped_rather_than_fetched(
+        self, session: orm.Session
+    ):
+        """A SELECT here would land inside the flush that before_commit runs in."""
+        container_execution = bts.ContainerExecution(
+            status=bts.ContainerExecutionStatus.PENDING
+        )
+        node = bts.ExecutionNode(task_spec={})
+        node.container_execution = container_execution
+        session.add(node)
+        container_execution.launcher_data = self._launcher_data(
+            phase="Pending", reason="ImagePullBackOff"
+        )
+        session.commit()
+        session.expire(node)
+        assert metrics._pending_reason(execution=node) == "none"
+
+    def test_a_node_with_no_container_execution_is_unlabelled(
+        self, session: orm.Session
+    ):
+        node = bts.ExecutionNode(task_spec={})
+        session.add(node)
+        session.commit()
+        assert metrics._pending_reason(execution=node) == "none"
+
+
+class TestPendingReasonOnTheInstruments:
+    """The reason rides both instruments, so an alert and its link agree."""
+
+    @staticmethod
+    def _waiting(reason: str) -> dict:
+        return {
+            "kubernetes_job": {
+                "debug_pods": {
+                    "0": {
+                        "status": {
+                            "phase": "Pending",
+                            "containerStatuses": [
+                                {"state": {"waiting": {"reason": reason}}}
+                            ],
+                        }
+                    }
+                }
+            }
+        }
+
+    def _wait_then_run(self, session: orm.Session, *, gap: datetime.timedelta):
+        """A task that waits on an image pull, then starts -- the case T5 exists for."""
+        container_execution = bts.ContainerExecution(
+            status=bts.ContainerExecutionStatus.PENDING
+        )
+        node = bts.ExecutionNode(task_spec={})
+        node.container_execution = container_execution
+        session.add(node)
+        node.container_execution_status = bts.ContainerExecutionStatus.PENDING
+        container_execution.launcher_data = self._waiting("ImagePullBackOff")
+        session.commit()
+        node.container_execution
+        history = node.extra_data[bts.EXECUTION_NODE_EXTRA_DATA_STATUS_HISTORY_KEY]
+        history[-1]["first_observed_at"] = (
+            datetime.datetime.now(datetime.timezone.utc) - gap
+        ).isoformat()
+        node.extra_data = {**node.extra_data}
+        node.container_execution_status = bts.ContainerExecutionStatus.RUNNING
+        return node
+
+    def test_the_histogram_carries_why_the_task_was_waiting(self, session: orm.Session):
+        with unittest.mock.patch.object(
+            metrics.execution_status_transition_duration, "record"
+        ) as record:
+            self._wait_then_run(session, gap=datetime.timedelta(minutes=3))
+            session.commit()
+        attributes = record.call_args.kwargs["attributes"]
+        assert attributes["execution.pending_reason"] == "image_unavailable"
+
+    def test_the_slow_counter_carries_why_the_task_was_waiting(
+        self, session: orm.Session
+    ):
+        with unittest.mock.patch.object(
+            metrics.execution_slow_transitions, "add"
+        ) as add:
+            self._wait_then_run(session, gap=datetime.timedelta(hours=4))
+            session.commit()
+        attributes = add.call_args.kwargs["attributes"]
+        assert attributes["execution.pending_reason"] == "image_unavailable"
+        assert attributes["execution.status.to"] == "RUNNING"