Fix start/end step regressions: events, SFN, cards, argo, graph, lint

Core fixes:
- events.py: use end_task.parent.id instead of hardcoded run_obj["end"]
- step_functions.py: use StartAt from definition JSON (both get_existing_deployment
  and get_execution) instead of hardcoded ["States"]["start"]
- argo_workflows.py: _matching_conditional_join uses self.graph.end_step
  instead of hardcoded "end" fallback
- graph.py: None guard in output_steps() raises clear ValueError
- client/core.py: narrow _graph_endpoints exception caching to
  (KeyError, MetaflowNotFound); transient errors return uncached fallback
- runtime.py: guard metadata registration with is_cloned check + try/except

Cards:
- card_cli.py: wrap graph_dict in payload with start_step/end_step metadata
- basic.py: accept new payload format, pass through start/end
- dag.svelte, step-wrapper.svelte, types.ts: dynamic start/end props

Lint:
- Improved check_basic_steps to distinguish "no start step" from "multiple
  @step(start=True) annotations"
- New check_annotation_name_conflict: warns when @step(start=True) coexists
  with a step named "start" (and likewise for end)

Tests:
- Negative-path tests for malformed annotation patterns (8 cases)
- Card rendering with custom endpoints
- Trigger.from_runs() with custom terminal step
- Step Functions StartAt lookup

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: talsperre

metaflow/client/core.py

@@ -2192,8 +2192,13 @@ def _graph_endpoints(self):
                 params_meta = self["_parameters"].task.metadata_dict
                 start = params_meta.get("start_step", "start")
                 end = params_meta.get("end_step", "end")
-            except Exception:
+            except (KeyError, MetaflowNotFound):
+                # Expected for old runs without _parameters or metadata.
                 pass
+            except Exception:
+                # Transient error (network, metadata service) -- do NOT cache
+                # the fallback so a subsequent access can retry.
+                return (start, end)
             self._cached_endpoints = (start, end)
         return self._cached_endpoints
 

metaflow/decorators.py

@@ -591,10 +591,10 @@ def _base_step_decorator(decotype, *args, **kwargs):
 
         # Step decorator applied to a class with a synthesized step method.
         # Used by extensions that create a single synthetic @step on a
-        # FlowSpec subclass. _step_spec_step_name is set by the
-        # extension's metaclass.
-        if isinstance(func, type) and hasattr(func, "_step_spec_step_name"):
-            step_func = getattr(func, func._step_spec_step_name)
+        # FlowSpec subclass. _function_spec_step_name is set by the
+        # extension's metaclass (e.g. FunctionSpecMeta).
+        if isinstance(func, type) and hasattr(func, "_function_spec_step_name"):
+            step_func = getattr(func, func._function_spec_step_name)
             if hasattr(step_func, "is_step"):
                 step_func.decorators.append(
                     decotype(attributes=kwargs, statically_defined=True)

metaflow/events.py

@@ -57,18 +57,24 @@ def __init__(self, _meta=None):
     @classmethod
     def from_runs(cls, run_objs: List["metaflow.Run"]):
         run_objs.sort(key=lambda x: x.finished_at, reverse=True)
-        trigger = Trigger(
-            [
+        valid_runs = []
+        meta = []
+        for run_obj in run_objs:
+            end_task = run_obj.end_task
+            if end_task is None:
+                continue
+            valid_runs.append(run_obj)
+            meta.append(
                 {
                     "type": "run",
                     "timestamp": run_obj.finished_at,
-                    "name": "metaflow.%s.%s" % (run_obj.parent.id, run_obj["end"].id),
-                    "id": run_obj.end_task.pathspec,
+                    "name": "metaflow.%s.%s" % (run_obj.parent.id, end_task.parent.id),
+                    "id": end_task.pathspec,
                 }
-                for run_obj in run_objs
-            ]
-        )
-        trigger._runs = run_objs
+            )
+
+        trigger = Trigger(meta)
+        trigger._runs = valid_runs
         return trigger
 
     @property

metaflow/flowspec.py

@@ -164,21 +164,11 @@ def _merge_value(self, inherited_value, self_value):
 
 
 class FlowSpecMeta(type):
-    _registry = {}  # {class_name: class} for all FlowSpec/StepSpec subclasses
-
-    # Names of base classes that should NOT be registered or have their
-    # graph/attrs initialized.  Subclasses of FlowSpecMeta (like StepSpecMeta)
-    # can extend this set.
-    _base_class_names = frozenset({"FlowSpec"})
-
     def __init__(cls, name, bases, attrs):
         super().__init__(name, bases, attrs)
-        # Check against the metaclass's own _base_class_names — this
-        # allows StepSpecMeta to add "StepSpec" to the set.
-        if name in type(cls)._base_class_names:
+        if name == "FlowSpec":
             return
 
-        type(cls)._registry[name] = cls
         cls._init_attrs()
 
     def _init_attrs(cls):

metaflow/graph.py

@@ -59,12 +59,20 @@ def __init__(
         is_start_step=False,
         is_end_step=False,
         node_info=None,
+        name=None,
+        num_args=None,
     ):
-        self.name = func_ast.name
+        if func_ast is None and name is None:
+            raise ValueError("name is required when func_ast is None")
+
+        self.name = func_ast.name if func_ast is not None else name
         self.source_file = source_file
         # lineno is the start line of decorators in source_file
         # func_ast.lineno is lines from decorators start to def of function
-        self.func_lineno = lineno + func_ast.lineno - 1
+        if func_ast is not None:
+            self.func_lineno = lineno + func_ast.lineno - 1
+        else:
+            self.func_lineno = lineno
         self.decorators = decos
         self.wrappers = wrappers
         self.config_decorators = config_decorators
@@ -83,13 +91,14 @@ def __init__(
         self.out_funcs = []
         self.has_tail_next = False
         self.invalid_tail_next = False
-        self.num_args = 0
+        self.num_args = 0 if num_args is None else num_args
         self.switch_cases = {}
         self.condition = None
         self.foreach_param = None
         self.num_parallel = 0
         self.parallel_foreach = False
-        self._parse(func_ast, lineno)
+        if func_ast is not None:
+            self._parse(func_ast, lineno)
 
         # these attributes are populated by _traverse_graph
         self.in_funcs = set()
@@ -282,48 +291,34 @@ def _identify_start_end(self):
         """
         Determine the start and end steps.
 
-        Uses explicit ``@step(start=True)`` / ``@step(end=True)`` annotations
-        if present.  Falls back to looking for steps named ``"start"`` /
-        ``"end"`` for backward compatibility.
-
-        Sets ``self.start_step`` and ``self.end_step`` to step name strings,
-        or ``None`` if the graph is malformed (validated later by lint).
-        Also assigns the ``"start"`` and ``"end"`` node types.
+        Uses explicit @step(start=True) / @step(end=True) annotations if present.
+        Falls back to steps named "start" / "end" for backward compatibility.
+        Sets self.start_step and self.end_step to step name strings, or None
+        if the graph is malformed (validated later by lint).
         """
-        # 1. Look for explicit annotations
-        annotated_start = [
-            name
-            for name, node in self.nodes.items()
-            if node.is_start_step and not name.startswith("_")
-        ]
-        annotated_end = [
-            name
-            for name, node in self.nodes.items()
-            if node.is_end_step and not name.startswith("_")
-        ]
-
-        # 2. Determine start step (annotation first, then name fallback)
-        if len(annotated_start) == 1:
-            self.start_step = annotated_start[0]
-        elif len(annotated_start) == 0:
-            self.start_step = "start" if "start" in self.nodes else None
-        else:
-            self.start_step = None  # Multiple annotated — lint will catch
 
-        # 3. Determine end step (annotation first, then name fallback)
-        if len(annotated_end) == 1:
-            self.end_step = annotated_end[0]
-        elif len(annotated_end) == 0:
-            self.end_step = "end" if "end" in self.nodes else None
-        else:
-            self.end_step = None  # Multiple annotated — lint will catch
+        def _resolve(attr, fallback_name):
+            """Find the unique annotated step, or fall back to a named step."""
+            annotated = [
+                name
+                for name, node in self.nodes.items()
+                if getattr(node, attr) and not name.startswith("_")
+            ]
+            if len(annotated) == 1:
+                return annotated[0]
+            if len(annotated) == 0:
+                return fallback_name if fallback_name in self.nodes else None
+            return None  # Multiple annotated — lint will catch
+
+        self.start_step = _resolve("is_start_step", "start")
+        self.end_step = _resolve("is_end_step", "end")
 
-        # 4. Assign types based on identified start/end.
-        # Only upgrade "linear" → "start" for the entry point; do NOT override
+        # Assign node types for graph traversal.
+        # Only upgrade "linear" -> "start" for the entry point; do NOT override
         # "split", "foreach", etc. since those types are needed for
         # split/join balance checking.
         if self.start_step and self.start_step == self.end_step:
-            # Single-step flow: terminal node that is also the entry point
+            # Single-step flow: terminal node that is also the entry point.
             self.nodes[self.start_step].type = "end"
         else:
             if self.start_step:
@@ -333,29 +328,76 @@ def _identify_start_end(self):
             if self.end_step:
                 self.nodes[self.end_step].type = "end"
 
+    def _create_sourceless_single_step_node(
+        self, name, func, is_start_step, is_end_step
+    ):
+        """Create a DAGNode for a dynamically-generated single-step method.
+
+        When ``inspect.getsourcelines()`` fails (e.g. for steps synthesized
+        via ``compile()`` + ``exec()`` by extension metaclasses like
+        ``FunctionSpecMeta``), this method builds a DAGNode without AST
+        parsing.  This is safe because single-step flows (``start=True,
+        end=True``) have no ``self.next()`` transitions to analyze.
+        """
+        code = getattr(func, "__code__", None)
+        source_file = inspect.getsourcefile(func) or inspect.getfile(func)
+        lineno = getattr(code, "co_firstlineno", 0)
+
+        try:
+            num_args = len(inspect.signature(func).parameters)
+        except (TypeError, ValueError):
+            num_args = getattr(code, "co_argcount", 0)
+
+        return DAGNode(
+            None,
+            func.decorators,
+            func.wrappers,
+            func.config_decorators,
+            func.__doc__,
+            source_file,
+            lineno,
+            is_start_step=is_start_step,
+            is_end_step=is_end_step,
+            node_info=getattr(func, "node_info", None),
+            name=name,
+            num_args=num_args,
+        )
+
     def _create_nodes(self, flow):
         nodes = {}
         for element in dir(flow):
             func = getattr(flow, element)
             if callable(func) and hasattr(func, "is_step"):
-                source_file = inspect.getsourcefile(func)
-                source_lines, lineno = inspect.getsourcelines(func)
-                # This also works for code (strips out leading whitspace based on
-                # first line)
-                source_code = deindent_docstring("".join(source_lines))
-                function_ast = ast.parse(source_code).body[0]
-                node = DAGNode(
-                    function_ast,
-                    func.decorators,
-                    func.wrappers,
-                    func.config_decorators,
-                    func.__doc__,
-                    source_file,
-                    lineno,
-                    is_start_step=getattr(func, "is_start_step", False),
-                    is_end_step=getattr(func, "is_end_step", False),
-                    node_info=getattr(func, "node_info", None),
-                )
+                is_start = getattr(func, "is_start_step", False)
+                is_end = getattr(func, "is_end_step", False)
+
+                try:
+                    source_file = inspect.getsourcefile(func) or inspect.getfile(func)
+                    source_lines, lineno = inspect.getsourcelines(func)
+                except OSError:
+                    if is_start and is_end:
+                        node = self._create_sourceless_single_step_node(
+                            element, func, is_start, is_end
+                        )
+                    else:
+                        raise
+                else:
+                    # This also works for code (strips out leading whitespace based on
+                    # first line)
+                    source_code = deindent_docstring("".join(source_lines))
+                    function_ast = ast.parse(source_code).body[0]
+                    node = DAGNode(
+                        function_ast,
+                        func.decorators,
+                        func.wrappers,
+                        func.config_decorators,
+                        func.__doc__,
+                        source_file,
+                        lineno,
+                        is_start_step=is_start,
+                        is_end_step=is_end,
+                        node_info=getattr(func, "node_info", None),
+                    )
                 nodes[element] = node
         return nodes
 
@@ -568,6 +610,18 @@ def populate_block(start_name, end_name):
                         break
             return resulting_list
 
+        if self.start_step is None or self.end_step is None:
+            missing = []
+            if self.start_step is None:
+                missing.append("start")
+            if self.end_step is None:
+                missing.append("end")
+            raise ValueError(
+                "Cannot compute graph structure: no %s step identified. "
+                "Use @step(start=True)/@step(end=True) or name your steps "
+                "'start'/'end'." % " or ".join(missing)
+            )
+
         if self.start_step == self.end_step:
             # Single-step flow
             graph_structure = []

metaflow/lint.py

@@ -59,17 +59,60 @@ def check_reserved_words(graph):
 @linter.check
 def check_basic_steps(graph):
     if graph.start_step is None:
+        annotated = [
+            name
+            for name, node in graph.nodes.items()
+            if node.is_start_step and not name.startswith("_")
+        ]
+        if len(annotated) > 1:
+            raise LintWarn(
+                "Multiple steps annotated with @step(start=True): %s. "
+                "Exactly one is allowed." % ", ".join(sorted(annotated))
+            )
         raise LintWarn(
             "Your flow must have exactly one start step. Either name a step "
             "'start' or use @step(start=True)."
         )
     if graph.end_step is None:
+        annotated = [
+            name
+            for name, node in graph.nodes.items()
+            if node.is_end_step and not name.startswith("_")
+        ]
+        if len(annotated) > 1:
+            raise LintWarn(
+                "Multiple steps annotated with @step(end=True): %s. "
+                "Exactly one is allowed." % ", ".join(sorted(annotated))
+            )
         raise LintWarn(
             "Your flow must have exactly one end step. Either name a step "
             "'end' or use @step(end=True)."
         )
 
 
+@linter.ensure_fundamentals
+@linter.check
+def check_annotation_name_conflict(graph):
+    """Detect conflict between @step(start/end=True) and legacy step names."""
+    if (
+        graph.start_step is not None
+        and graph.start_step != "start"
+        and "start" in graph.nodes
+    ):
+        raise LintWarn(
+            "Ambiguous start step: step '%s' is annotated with @step(start=True) "
+            "but a step named 'start' also exists. Remove the 'start' name or "
+            "the @step(start=True) annotation." % graph.start_step
+        )
+
+    if graph.end_step is not None and graph.end_step != "end" and "end" in graph.nodes:
+        raise LintWarn(
+            "Ambiguous end step: step '%s' is annotated with @step(end=True) "
+            "but a step named 'end' also exists. Remove the 'end' name or "
+            "the @step(end=True) annotation." % graph.end_step
+        )
+
+
 @linter.ensure_static_graph
 @linter.check
 def check_start_end_degree(graph):

metaflow/plugins/argo/argo_workflows.py

@@ -1201,8 +1201,9 @@ def _is_recursive_node(self, node):
         return node.name in self.recursive_nodes
 
     def _matching_conditional_join(self, node):
-        # If no earlier conditional join step is found during parsing, then 'end' is always one.
-        return self.matching_conditional_join_dict.get(node.name, "end")
+        # If no earlier conditional join step is found during parsing,
+        # fall back to the graph's terminal step.
+        return self.matching_conditional_join_dict.get(node.name, self.graph.end_step)
 
     # Visit every node and yield the uber DAGTemplate(s).
     def _dag_templates(self):