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/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

@@ -568,6 +568,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):

metaflow/plugins/aws/step_functions/step_functions.py

@@ -240,12 +240,14 @@ def get_existing_deployment(cls, name):
         workflow = StepFunctionsClient().get(name)
         if workflow is not None:
             try:
-                start = json.loads(workflow["definition"])["States"]["start"]
+                definition = json.loads(workflow["definition"])
+                start_state_name = definition.get("StartAt", "start")
+                start = definition["States"][start_state_name]
                 parameters = start["Parameters"]["Parameters"]
                 return parameters.get("metaflow.owner"), parameters.get(
                     "metaflow.production_token"
                 )
-            except KeyError:
+            except (KeyError, TypeError, AttributeError):
                 raise StepFunctionsException(
                     "An existing non-metaflow "
                     "workflow with the same name as "
@@ -271,17 +273,21 @@ def get_execution(cls, state_machine_name, name):
             )
         try:
             state_machine_arn = state_machine.get("stateMachineArn")
-            environment_vars = (
-                json.loads(state_machine.get("definition"))
-                .get("States")
-                .get("start")
-                .get("Parameters")
-                .get("ContainerOverrides")
-                .get("Environment")
-            )
-            parameters = {
-                item.get("Name"): item.get("Value") for item in environment_vars
-            }
+            definition = json.loads(state_machine.get("definition"))
+            start_state_name = definition.get("StartAt", "start")
+            try:
+                start = definition["States"][start_state_name]
+                environment_vars = start["Parameters"]["ContainerOverrides"][
+                    "Environment"
+                ]
+                parameters = {
+                    item.get("Name"): item.get("Value") for item in environment_vars
+                }
+            except (KeyError, TypeError, AttributeError):
+                raise StepFunctionsException(
+                    "A non-metaflow workflow *%s* already exists in AWS Step Functions."
+                    % state_machine_name
+                )
             executions = client.list_executions(state_machine_arn, states=["RUNNING"])
             for execution in executions:
                 if execution.get("name") == name:
@@ -295,9 +301,11 @@ def get_execution(cls, state_machine_name, name):
                     except KeyError:
                         raise StepFunctionsException(
                             "A non-metaflow workflow *%s* already exists in AWS Step Functions."
-                            % name
+                            % state_machine_name
                         )
             return None
+        except StepFunctionsException:
+            raise
         except Exception as e:
             raise StepFunctionsException(repr(e))
 
@@ -795,6 +803,10 @@ def _batch(self, node):
         metaflow_version["production_token"] = self.production_token
         env["METAFLOW_VERSION"] = json.dumps(metaflow_version)
 
+        multiflow_name = os.environ.get("METAFLOW_MULTIFLOW_NAME")
+        if multiflow_name:
+            env["METAFLOW_MULTIFLOW_NAME"] = multiflow_name
+
         # map config values
         cfg_env = {param["name"]: param["kv_name"] for param in self.config_parameters}
         if cfg_env:
@@ -926,6 +938,9 @@ def _step_cli(self, node, paths, code_package_url, user_code_retries):
             entrypoint = [R.entrypoint()]
         else:
             entrypoint = [executable, script_name]
+            multiflow_name = os.environ.get("METAFLOW_MULTIFLOW_NAME")
+            if multiflow_name:
+                entrypoint.append(multiflow_name)
 
         # Use AWS Batch job identifier as the globally unique task identifier.
         task_id = "${AWS_BATCH_JOB_ID}"

metaflow/plugins/cards/card_cli.py

@@ -649,6 +649,11 @@ def create(
     full_pathspec = "/".join([flowname, pathspec])
 
     graph_dict, _ = ctx.obj.graph.output_steps()
+    graph_payload = {
+        "steps": graph_dict,
+        "start_step": ctx.obj.graph.start_step,
+        "end_step": ctx.obj.graph.end_step,
+    }
 
     if card_uuid is None:
         card_uuid = str(uuid.uuid4()).replace("-", "")
@@ -702,12 +707,12 @@ def create(
                 mf_card = filtered_card(
                     options=options,
                     components=component_arr,
-                    graph=graph_dict,
+                    graph=graph_payload,
                     flow=ctx.obj.flow,
                 )
             else:
                 mf_card = filtered_card(
-                    components=component_arr, graph=graph_dict, flow=ctx.obj.flow
+                    components=component_arr, graph=graph_payload, flow=ctx.obj.flow
                 )
         except TypeError as e:
             if render_error_card:

metaflow/plugins/cards/card_modules/basic.py

@@ -12,7 +12,22 @@
 CSS_PATH = os.path.join(ABS_DIR_PATH, "bundle.css")
 
 
-def transform_flow_graph(step_info):
+def transform_flow_graph(graph):
+    if (
+        isinstance(graph, dict)
+        and "steps" in graph
+        and "start_step" in graph
+        and "end_step" in graph
+        and isinstance(graph["steps"], dict)
+    ):
+        step_info = graph["steps"]
+        start_step = graph.get("start_step")
+        end_step = graph.get("end_step")
+    else:
+        step_info = graph
+        start_step = "start" if "start" in graph else None
+        end_step = "end" if "end" in graph else None
+
     def node_to_type(node_type):
         if node_type in ["linear", "start", "end", "join"]:
             return node_type
@@ -47,7 +62,11 @@ def node_to_type(node_type):
 
         graph_dict[stepname] = node_info
 
-    return graph_dict
+    return {
+        "steps": graph_dict,
+        "start_step": start_step,
+        "end_step": end_step,
+    }
 
 
 def read_file(path):

metaflow/plugins/cards/ui/src/components/dag/dag.svelte

@@ -11,7 +11,7 @@
 
   export let componentData: DagComponent;
 
-  const { data: steps } = componentData;
+  const { steps, start_step: startStep, end_step: endStep } = componentData.data;
   let el: HTMLElement;
   let dagStructure: DagStructure = {};
 
@@ -42,8 +42,14 @@
   style="position: relative; line-height: 1"
   data-component="dag"
 >
-  {#if steps?.start}
-    <StepWrapper {steps} stepName="start" bind:dagStructure />
+  {#if startStep && steps?.[startStep]}
+    <StepWrapper
+      {steps}
+      stepName={startStep}
+      {startStep}
+      {endStep}
+      bind:dagStructure
+    />
   {:else}
     <p>No start step</p>
   {/if}