Add Argo workflow title & description support (#2628)

fixes #2626 

Adds support for Argo Workflow Title & description.

---------

Co-authored-by: Sakari Ikonen <sakari.a.ikonen@gmail.com>

Author: ShreehariVaasishta

metaflow/plugins/argo/argo_workflows.py

@@ -121,6 +121,8 @@ def __init__(
         incident_io_metadata: List[str] = None,
         enable_heartbeat_daemon=True,
         enable_error_msg_capture=False,
+        workflow_title=None,
+        workflow_description=None,
     ):
         # Some high-level notes -
         #
@@ -177,6 +179,8 @@ def __init__(
         )
         self.enable_heartbeat_daemon = enable_heartbeat_daemon
         self.enable_error_msg_capture = enable_error_msg_capture
+        self.workflow_title = workflow_title
+        self.workflow_description = workflow_description
         self.parameters = self._process_parameters()
         self.config_parameters = self._process_config_parameters()
         self.triggers, self.trigger_options = self._process_triggers()
@@ -430,6 +434,25 @@ def _base_kubernetes_annotations(self):
                     "metaflow/project_flow_name": current.project_flow_name,
                 }
             )
+
+        # Add Argo Workflows title and description annotations
+        # https://argo-workflows.readthedocs.io/en/latest/title-and-description/
+        # Use CLI-provided values or auto-populate from metadata
+        title = (
+            (self.workflow_title.strip() if self.workflow_title else None)
+            or current.get("project_flow_name")
+            or self.flow.name
+        )
+
+        description = (
+            self.workflow_description.strip() if self.workflow_description else None
+        ) or (self.flow.__doc__.strip() if self.flow.__doc__ else None)
+
+        if title:
+            annotations["workflows.argoproj.io/title"] = title
+        if description:
+            annotations["workflows.argoproj.io/description"] = description
+
         return annotations
 
     def _get_schedule(self):
@@ -894,7 +917,16 @@ def _compile_workflow_template(self):
                     .annotations(
                         {
                             **annotations,
-                            **self._base_annotations,
+                            **{
+                                k: v
+                                for k, v in self._base_annotations.items()
+                                if k
+                                # Skip custom title/description for workflows as this makes it harder to find specific runs.
+                                not in [
+                                    "workflows.argoproj.io/title",
+                                    "workflows.argoproj.io/description",
+                                ]
+                            },
                             **{"metaflow/run_id": "argo-{{workflow.name}}"},
                         }
                     )

metaflow/plugins/argo/argo_workflows_cli.py

@@ -227,6 +227,18 @@ def argo_workflows(obj, name=None):
     show_default=True,
     help="Capture stack trace of first failed task in exit hook.",
 )
+@click.option(
+    "--workflow-title",
+    default=None,
+    type=str,
+    help="Custom title for the workflow displayed in Argo Workflows UI. Defaults to `project_flow_name`. Supports markdown formatting.",
+)
+@click.option(
+    "--workflow-description",
+    default=None,
+    type=str,
+    help="Custom description for the workflow displayed in Argo Workflows UI. Defaults to the flow's docstring if available. Supports markdown formatting and multi-line text.",
+)
 @click.pass_obj
 def create(
     obj,
@@ -248,6 +260,8 @@ def create(
     incident_io_alert_source_config_id=None,
     incident_io_metadata=None,
     enable_heartbeat_daemon=True,
+    workflow_title=None,
+    workflow_description=None,
     deployer_attribute_file=None,
     enable_error_msg_capture=False,
 ):
@@ -312,6 +326,8 @@ def create(
         incident_io_metadata,
         enable_heartbeat_daemon,
         enable_error_msg_capture,
+        workflow_title,
+        workflow_description,
     )
 
     if only_json:
@@ -658,6 +674,8 @@ def make_flow(
     incident_io_metadata,
     enable_heartbeat_daemon,
     enable_error_msg_capture,
+    workflow_title,
+    workflow_description,
 ):
     # TODO: Make this check less specific to Amazon S3 as we introduce
     #       support for more cloud object stores.
@@ -750,6 +768,8 @@ def make_flow(
         incident_io_metadata=incident_io_metadata,
         enable_heartbeat_daemon=enable_heartbeat_daemon,
         enable_error_msg_capture=enable_error_msg_capture,
+        workflow_title=workflow_title,
+        workflow_description=workflow_description,
     )