[GSoC] Add failure-investigation accessors to the core client (stacked on #3348)

metaflow/client/core.py

@@ -4,10 +4,12 @@
 import os
 import tarfile
 from collections import namedtuple
+from collections.abc import Mapping
+from dataclasses import dataclass
 from datetime import datetime
 from tempfile import TemporaryDirectory
 from io import BytesIO
-from itertools import chain
+from itertools import chain, islice
 from typing import (
     Any,
     Dict,
@@ -61,6 +63,48 @@
 current_metadata = False
 
 
+@dataclass(frozen=True)
+class FailureSummary:
+    """
+    Normalized description of the exception that failed a `Task`.
+
+    Returned by `Task.failure_summary`. Any field may be None when the stored
+    exception artifact does not carry that particular detail.
+    """
+
+    exception_type: Optional[str]
+    message: Optional[str]
+    stacktrace: Optional[str]
+    attempt: Optional[int]
+
+
+def _normalize_exception(data: Any) -> Dict[str, Optional[str]]:
+    """
+    Flatten a task's ``_exception`` artifact into type/message/stacktrace strings.
+
+    Handles both mapping-shaped exception records and exception-like objects,
+    falling back to ``str(data)`` for the message so there is always something
+    human (or agent) readable.
+    """
+    if isinstance(data, Mapping):
+        exception_type = data.get("type")
+        message = data.get("message") or data.get("exception")
+        stacktrace = data.get("stacktrace")
+    else:
+        exception_type = getattr(data, "type", None)
+        message = getattr(data, "message", None) or getattr(data, "exception", None)
+        stacktrace = getattr(data, "stacktrace", None)
+        if exception_type is None:
+            exception_type = "%s.%s" % (type(data).__module__, type(data).__name__)
+    if message is None:
+        message = str(data)
+    return {
+        "type": str(exception_type) if exception_type is not None else None,
+        "message": str(message),
+        "stacktrace": str(stacktrace) if stacktrace is not None else None,
+    }
+
+
 def metadata(ms: str) -> str:
     """
     Switch Metadata provider.
@@ -435,6 +479,37 @@ def _filtered_children(self, *tags):
             if all(tag in child.tags for tag in tags):
                 yield child
 
+    def _iter_children(self, query_filters=None, page_size=None, required_tags=()):
+        """Stream child records through the active metadata provider."""
+        namespace_filter = {}
+        if self._namespace_check and self._current_namespace:
+            namespace_filter = {"any_tags": self._current_namespace}
+
+        objects = self._metaflow.metadata.iter_objects(
+            self._NAME,
+            _CLASSES[self._CHILD_CLASS]._NAME,
+            namespace_filter,
+            self._attempt,
+            *self.path_components,
+            query_filters=query_filters,
+            page_size=page_size,
+        )
+        for obj in objects:
+            child = _CLASSES[self._CHILD_CLASS](
+                attempt=self._attempt,
+                _object=obj,
+                _parent=self,
+                _metaflow=self._metaflow,
+                _namespace_check=self._namespace_check,
+                _current_namespace=(
+                    self._current_namespace if self._namespace_check else None
+                ),
+            )
+            if self._iter_filter(child) and all(
+                tag in child.tags for tag in required_tags
+            ):
+                yield child
+
     def _ipython_key_completions_(self):
         """Returns available options for ipython auto-complete."""
         return [child.id for child in self._filtered_children()]
@@ -1603,6 +1678,36 @@ def exception(self) -> Optional[Any]:
         except KeyError:
             return None
 
+    @property
+    def failure_summary(self) -> Optional[FailureSummary]:
+        """
+        Returns a normalized summary of the exception that failed this task.
+
+        This is a convenience over `exception` that flattens the stored
+        exception artifact into ``exception_type``, ``message``, ``stacktrace``,
+        and ``attempt`` fields, so callers (including agents summarizing a
+        failure) do not need to know how the exception was serialized.
+
+        Returns None when the task recorded no exception (for example a task
+        that succeeded or has not failed). Errors while reading the underlying
+        exception or attempt are allowed to propagate.
+
+        Returns
+        -------
+        FailureSummary, optional
+            Normalized failure details, or None if the task has no exception.
+        """
+        exception = self.exception
+        if exception is None:
+            return None
+        normalized = _normalize_exception(exception)
+        return FailureSummary(
+            exception_type=normalized["type"],
+            message=normalized["message"],
+            stacktrace=normalized["stacktrace"],
+            attempt=self.current_attempt,
+        )
+
     @property
     def finished_at(self) -> Optional[datetime]:
         """
@@ -2305,6 +2410,39 @@ def successful(self) -> bool:
         else:
             return False
 
+    @property
+    def failed_task(self) -> Optional[Task]:
+        """
+        Returns the latest failed (unsuccessful) task in this run, if any.
+
+        Steps and their tasks are scanned in iteration order, which is
+        newest-created first (see `MetaflowObject.__iter__`), so the first
+        match is the latest failed task; for a failed run this is typically
+        the task that caused the failure. Returns None when every task in
+        the run is successful.
+
+        Together with `Flow.failed_runs` and `Task.failure_summary` this lets
+        you investigate failures through the regular client:
+
+        ```
+        for run in Flow("MyFlow").failed_runs(max_runs=10):
+            task = run.failed_task
+            if task is not None:
+                summary = task.failure_summary
+                print(task.pathspec, summary and summary.exception_type)
+        ```
+
+        Returns
+        -------
+        Task, optional
+            The latest unsuccessful task, or None if the run has none.
+        """
+        for step in self:
+            for task in step:
+                if not task.successful:
+                    return task
+        return None
+
     @property
     def finished(self) -> bool:
         """
@@ -2585,7 +2723,13 @@ def latest_successful_run(self) -> Optional[Run]:
             if run.successful:
                 return run
 
-    def runs(self, *tags: str) -> Iterator[Run]:
+    def runs(
+        self,
+        *tags: str,
+        filters: Optional[Dict[str, Any]] = None,
+        page_size: Optional[int] = None,
+        max_runs: Optional[int] = None,
+    ) -> Iterator[Run]:
         """
         Returns an iterator over all `Run`s of this flow.
 
@@ -2596,14 +2740,76 @@ def runs(self, *tags: str) -> Iterator[Run]:
         Parameters
         ----------
         tags : str
-            Tags to match.
+            Tags to match. Applied locally after listing, so this works with
+            local metadata as well as the metadata service.
+        filters : dict, optional
+            Server-side run filters using the metadata service's ``field:operator``
+            grammar, for example ``{"status:eq": "failed"}``. Requires a metadata
+            service with pagination and filtering support.
+        page_size : int, optional
+            Number of records requested from the metadata service per page.
+        max_runs : int, optional
+            Maximum number of runs to yield, newest first.
 
         Yields
         ------
         Run
             `Run` objects in this flow.
         """
-        return self._filtered_children(*tags)
+        if filters is not None and not hasattr(filters, "items"):
+            raise TypeError("filters must be a mapping")
+        if max_runs is not None:
+            if isinstance(max_runs, bool) or not isinstance(max_runs, int):
+                raise TypeError("max_runs must be an integer")
+            if max_runs < 0:
+                raise ValueError("max_runs must be non-negative")
+            if max_runs == 0:
+                return iter(())
+
+        if filters is None and page_size is None and max_runs is None:
+            return self._filtered_children(*tags)
+
+        runs = self._iter_children(
+            query_filters=dict(filters) if filters else None,
+            page_size=page_size,
+            required_tags=tags,
+        )
+        if max_runs is None:
+            return runs
+        return islice(runs, max_runs)
+
+    def failed_runs(
+        self,
+        *,
+        since: Optional[int] = None,
+        max_runs: Optional[int] = None,
+    ) -> Iterator[Run]:
+        """
+        Returns an iterator over the failed `Run`s of this flow, newest first.
+
+        This is a convenience wrapper over `runs` that applies the metadata
+        service's ``status:eq`` = ``failed`` filter server-side. Because it
+        relies on server-side filtering, it requires a metadata service with
+        pagination and filtering support; against the local metadata provider
+        or an older service it raises, exactly like ``runs(filters=...)``.
+
+        Parameters
+        ----------
+        since : int, optional
+            Inclusive lower bound on run start time as epoch milliseconds
+            (``ts_epoch``). Only runs at or after this time are returned.
+        max_runs : int, optional
+            Maximum number of failed runs to yield, newest first.
+
+        Yields
+        ------
+        Run
+            Failed `Run` objects in this flow, newest first.
+        """
+        filters = {"status:eq": "failed"}
+        if since is not None:
+            filters["ts_epoch:ge"] = int(since)
+        return self.runs(filters=filters, max_runs=max_runs)
 
     def __iter__(self) -> Iterator[Task]:
         """

metaflow/metadata_provider/metadata.py

@@ -6,7 +6,11 @@
 from itertools import chain
 
 from typing import List
-from metaflow.exception import MetaflowInternalError, MetaflowTaggingError
+from metaflow.exception import (
+    MetaflowException,
+    MetaflowInternalError,
+    MetaflowTaggingError,
+)
 from metaflow.tagging_util import validate_tag
 from metaflow.util import get_username, resolve_identity_as_tuple, is_stringish
 
@@ -348,6 +352,37 @@ def add_sticky_tags(self, tags=None, sys_tags=None):
         if sys_tags:
             self.sticky_sys_tags.update(sys_tags)
 
+    @classmethod
+    def _validate_object_query(cls, obj_type, sub_type):
+        """Reject nonsensical obj_type/sub_type combinations.
+
+        Shared by get_object and the streaming listing paths so every access,
+        materialized or paginated, enforces the same rules. Returns the
+        (type_order, sub_order) pair for callers that need it.
+        """
+        type_order = ObjectOrder.type_to_order(obj_type)
+        sub_order = ObjectOrder.type_to_order(sub_type)
+
+        if type_order is None:
+            raise MetaflowInternalError(msg="Cannot find type %s" % obj_type)
+        if type_order >= ObjectOrder.type_to_order("metadata"):
+            raise MetaflowInternalError(msg="Type %s is not allowed" % obj_type)
+
+        if sub_order is None:
+            raise MetaflowInternalError(msg="Cannot find subtype %s" % sub_type)
+
+        if type_order >= sub_order:
+            raise MetaflowInternalError(
+                msg="Subtype %s not allowed for %s" % (sub_type, obj_type)
+            )
+
+        # Metadata is always only at the task level
+        if sub_type == "metadata" and obj_type != "task":
+            raise MetaflowInternalError(
+                msg="Metadata can only be retrieved at the task level"
+            )
+        return type_order, sub_order
+
     @classmethod
     def get_object(cls, obj_type, sub_type, filters, attempt, *args):
         """Returns the requested object depending on obj_type and sub_type
@@ -397,27 +432,7 @@ def get_object(cls, obj_type, sub_type, filters, attempt, *args):
             object or list :
                 Depending on the call, the type of object return varies
         """
-        type_order = ObjectOrder.type_to_order(obj_type)
-        sub_order = ObjectOrder.type_to_order(sub_type)
-
-        if type_order is None:
-            raise MetaflowInternalError(msg="Cannot find type %s" % obj_type)
-        if type_order >= ObjectOrder.type_to_order("metadata"):
-            raise MetaflowInternalError(msg="Type %s is not allowed" % obj_type)
-
-        if sub_order is None:
-            raise MetaflowInternalError(msg="Cannot find subtype %s" % sub_type)
-
-        if type_order >= sub_order:
-            raise MetaflowInternalError(
-                msg="Subtype %s not allowed for %s" % (sub_type, obj_type)
-            )
-
-        # Metadata is always only at the task level
-        if sub_type == "metadata" and obj_type != "task":
-            raise MetaflowInternalError(
-                msg="Metadata can only be retrieved at the task level"
-            )
+        type_order, sub_order = cls._validate_object_query(obj_type, sub_type)
 
         if attempt is not None:
             try:
@@ -439,6 +454,45 @@ def get_object(cls, obj_type, sub_type, filters, attempt, *args):
             pre_filter, attempt_int
         )
 
+    @classmethod
+    def iter_objects(cls, obj_type, sub_type, filters, attempt, *args, **kwargs):
+        """Iterate over a collection returned by ``get_object``.
+
+        Providers can override this method to stream records without materializing the
+        complete collection. ``query_filters`` and ``page_size`` are optional provider
+        hints; the default implementation supports only unfiltered iteration.
+        """
+        query_filters = kwargs.pop("query_filters", None)
+        page_size = kwargs.pop("page_size", None)
+        if kwargs:
+            raise TypeError("Unexpected iterator options: %s" % ", ".join(kwargs))
+        if query_filters:
+            raise MetaflowException(
+                "Server-side metadata filters are not supported by the %s provider"
+                % cls.TYPE
+            )
+        if page_size is not None:
+            if isinstance(page_size, bool) or not isinstance(page_size, int):
+                raise TypeError("page_size must be an integer")
+            if page_size <= 0:
+                raise ValueError("page_size must be positive")
+
+        objects = cls.get_object(obj_type, sub_type, filters, attempt, *args)
+        if isinstance(objects, dict):
+            objects = [objects]
+        # Yield newest-first (descending ts_epoch). This mirrors the order the
+        # metadata service returns for paginated listings, so callers see a
+        # consistent order whether records came from the service or a legacy /
+        # local provider. ts_epoch may be missing on some records, so fall back
+        # to 0 to keep the sort total.
+        objects = sorted(
+            objects or [],
+            key=lambda obj: obj.get("ts_epoch") or 0,
+            reverse=True,
+        )
+        for obj in objects:
+            yield obj
+
     @classmethod
     def mutate_user_tags_for_run(
         cls, flow_id, run_id, tags_to_remove=None, tags_to_add=None