SG-44787 Validate previous published pipeline step when building new scene

info.yml

@@ -131,6 +131,24 @@ configuration:
         default_value: {}
         allows_empty: true
 
+    pipeline_step_dependencies:
+        type: dict
+        description: "Maps a pipeline step to the upstream pipeline step whose workfile must
+                     be published before a new scene is built for it. Keys and values must
+                     match the Flow Production Tracking Step name exactly as it appears in
+                     the task's Pipeline Step, for example {Rig: Model, Texture: Model}
+                     (the names are project-specific - use whatever your site calls the
+                     modelling, surfacing and rigging steps). Used by the Flow Asset
+                     Management 'Build New Scene' action to warn the artist when the
+                     upstream step has nothing published to reference yet. Steps absent
+                     from this mapping have no upstream requirement and are never validated.
+                     Only Maya scenes are checked, so this setting has no effect in other
+                     DCCs."
+        default_value:
+            Rig: Model
+            Texture: Model
+        allows_empty: true
+
     entities:
         default_value:
             - caption: Project

python/tk_multi_loader/constants.py

@@ -103,3 +103,8 @@
 # FlowAM versions starts with 0 and FlowPT versions starts with 1.
 # This is used to identify a FlowAM draft version in the UI.
 DRAFT_VERSION_IDENTIFIER = -1
+
+# Schema type of the Maya workfile, mirroring FlowHost.WORKFILE_TYPE in tk-maya,
+# which this app cannot import. Only the Maya scene publish is validated when
+# building a new scene, so hosts building other workfile types skip that check.
+MAYA_WORKFILE_TYPE = "type.workfile.maya"

python/tk_multi_loader/flowam/flowam_actions.py

@@ -19,7 +19,7 @@
 
 from ..build_asset_dialog import BuildAssetDialog
 from ..build_template_dialog import BuildTemplateDialog
-from ..constants import DRAFT_VERSION_IDENTIFIER
+from ..constants import DRAFT_VERSION_IDENTIFIER, MAYA_WORKFILE_TYPE
 from .create import (
     CreateInputs,
     CreateTemplateInputs,
@@ -33,6 +33,7 @@
     open_draft,
 )
 from .reference import copy_reference_link, reference_revision
+from .step_validation import find_unpublished_upstream_step
 
 
 class FlowAMActions:
@@ -178,6 +179,13 @@ def _on_build_scene_dialog_accepted(
             prep_scene_callback=functools.partial(self._prep_scene, sg_publish_data),
         )
 
+        if not self._confirm_upstream_step_published(create_inputs):
+            self._app.log_info(
+                "Build new scene cancelled: the previous pipeline step has not "
+                "been published."
+            )
+            return
+
         try:
             asset = create_dcc_workfile(create_inputs)
             self._app.log_debug(f"Created a DCC workfile asset: {asset}")
@@ -190,6 +198,59 @@ def _on_build_scene_dialog_accepted(
                 str(exc),
             )
 
+    def _confirm_upstream_step_published(self, create_inputs: CreateInputs) -> bool:
+        """
+        Warn the artist when the previous pipeline step has nothing published yet.
+
+        The upstream publish is what gets referenced into the new scene, so
+        building without it leaves an empty scene. The artist is given the choice
+        to continue regardless.
+
+        Only the Maya scene publish is in scope. Nuke and Houdini reach this
+        method through their own build actions, but a downstream step references
+        the Maya scene rather than the current host's workfile type, so those
+        hosts are left alone.
+
+        :param create_inputs: Inputs describing the scene about to be built.
+        :returns: True when the build should go ahead.
+        """
+        host = getattr(sgtk.platform.current_engine(), "flow_host", None)
+        workfile_type = getattr(host, "WORKFILE_TYPE", "")
+        if workfile_type != MAYA_WORKFILE_TYPE:
+            return True
+
+        upstream_step = find_unpublished_upstream_step(
+            am_project_id=create_inputs.am_project_id,
+            sg_entity_name=create_inputs.sg_entity_name,
+            sg_entity_type=create_inputs.sg_entity_type,
+            sg_pipeline_step=create_inputs.sg_pipeline_step,
+            step_dependencies=self._app.get_setting("pipeline_step_dependencies", {}),
+            workfile_type=workfile_type,
+        )
+        if not upstream_step:
+            return True
+
+        message = (
+            f'The "{upstream_step}" step has no published scene for '
+            f'"{create_inputs.sg_entity_name}" yet, so there is nothing to '
+            f'reference into a new "{create_inputs.sg_pipeline_step}" scene.'
+            "\n\nBuild an empty scene anyway?"
+        )
+        self._app.log_warning(message)
+
+        if not host:
+            return True
+
+        response = host.dialog(
+            "Previous step not published",
+            message,
+            buttons=["Build empty scene", "Cancel"],
+            default=1,
+            cancel=1,
+            no_ui_option=1,
+        )
+        return response == 0
+
     def _prep_scene(self, sg_publish_data: dict) -> None:
         """
         Let clients run set-up scripts when building a new scene/asset.

python/tk_multi_loader/flowam/step_validation.py

@@ -0,0 +1,167 @@
+# Copyright (c) 2026 Shotgun Software Inc.
+#
+# CONFIDENTIAL AND PROPRIETARY
+#
+# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
+# Source Code License included in this distribution package. See LICENSE.
+# By accessing, using, copying or modifying this work you indicate your
+# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
+# not expressly granted therein are reserved by Shotgun Software Inc.
+
+"""Pipeline step validation helpers for Flow Asset Management.
+
+Building a new scene for a downstream department only makes sense once the
+upstream department has published its workfile, because that publish is what
+gets referenced into the new scene.
+"""
+
+from __future__ import annotations  # needed for Houdini support
+
+from typing import Dict, Optional
+
+import sgtk
+from sgtk.flowam.create import ASSET_FOLDER, ASSET_TYPE, SHOT_TYPE
+from tank_vendor.flow_integration_sdk import exceptions, objects, schema
+
+logger = sgtk.platform.get_logger(__name__)
+
+
+def get_upstream_step(
+    pipeline_step: str, step_dependencies: Dict[str, str]
+) -> Optional[str]:
+    """Return the pipeline step that must be published before *pipeline_step*.
+
+    :param pipeline_step: Name of the step a new scene is being built for.
+    :param step_dependencies: Mapping of step name to upstream step name, as
+        provided by the ``pipeline_step_dependencies`` app setting.
+    :returns: Upstream step name, or ``None`` when the step has no configured
+        upstream requirement.
+    """
+    if not pipeline_step or not step_dependencies:
+        return None
+
+    return step_dependencies.get(pipeline_step) or None
+
+
+def find_unpublished_upstream_step(
+    am_project_id: str,
+    sg_entity_type: str,
+    sg_entity_name: str,
+    sg_pipeline_step: str,
+    workfile_type: str,
+    step_dependencies: Dict[str, str],
+) -> Optional[str]:
+    """Return the upstream pipeline step that still needs to be published.
+
+    :param am_project_id: Id of the Flow AM project holding the asset.
+    :param sg_entity_type: FPTR entity type of the asset, e.g. ``"Asset"``.
+    :param sg_entity_name: FPTR entity name of the asset.
+    :param sg_pipeline_step: Step the new scene is being built for.
+    :param workfile_type: Schema type name of the workfile to look for, e.g.
+        ``"type.workfile.maya"`` from ``FlowHost.WORKFILE_TYPE``.
+    :param step_dependencies: Mapping of step name to upstream step name.
+    :returns: Name of the upstream step when it is configured but has no
+        published workfile, otherwise ``None``.
+    """
+    upstream_step = get_upstream_step(sg_pipeline_step, step_dependencies)
+    if not upstream_step:
+        return None
+
+    if has_published_workfile(
+        am_project_id=am_project_id,
+        pipeline_step=upstream_step,
+        sg_entity_name=sg_entity_name,
+        sg_entity_type=sg_entity_type,
+        workfile_type=workfile_type,
+    ):
+        return None
+
+    return upstream_step
+
+
+def has_published_workfile(
+    am_project_id: str,
+    pipeline_step: str,
+    sg_entity_name: str,
+    sg_entity_type: str,
+    workfile_type: str,
+) -> bool:
+    """Return ``True`` when *pipeline_step* has a published workfile for the asset.
+
+    A workfile asset only exists in Flow AM once it has been published:
+    ``sandbox.create_asset_in_sandbox()`` writes a local draft and defers the
+    Flow AM asset creation to publish time. Finding a workfile-typed child is
+    therefore enough to prove the step was published, whereas the hierarchy
+    enclosing it may well exist for a step nobody has published yet.
+
+    When the answer cannot be determined this returns ``True``, so a transient
+    Flow AM error never blocks a build behind a misleading "not published"
+    message.
+
+    :param am_project_id: Id of the Flow AM project holding the asset.
+    :param pipeline_step: Step to look for a published workfile under.
+    :param sg_entity_name: FPTR entity name of the asset.
+    :param sg_entity_type: FPTR entity type of the asset, e.g. ``"Asset"``.
+    :param workfile_type: Schema type name of the workfile to look for.
+    :returns: ``True`` when a published workfile exists or cannot be ruled out.
+    """
+    root_folder_name = _get_root_folder_name(sg_entity_type)
+    if not root_folder_name:
+        logger.warning(
+            f'Cannot locate Flow AM assets for entity type "{sg_entity_type}". '
+            f'Skipping the publish check for pipeline step "{pipeline_step}".'
+        )
+        return True
+
+    workfile_type_id = schema.get_schema_id(workfile_type)
+    if not workfile_type_id:
+        # An unresolved type id would disable the type filter in find_children()
+        # and match every child, so skip the check rather than trust it.
+        logger.warning(
+            f'Could not resolve the schema id for workfile type "{workfile_type}". '
+            f'Skipping the publish check for pipeline step "{pipeline_step}".'
+        )
+        return True
+
+    try:
+        node = objects.FlowProject(am_project_id)
+        # Walk down to the "root asset" grouping the workfiles of this step:
+        # Assets/<entity>/<step>/<entity>. See get_or_create_workfile_parent()
+        # in tk-core's tank/flowam/create.py for the hierarchy this mirrors.
+        for name in (
+            root_folder_name,
+            sg_entity_name,
+            pipeline_step,
+            sg_entity_name,
+        ):
+            node = node.find_child(name)
+            if node is None:
+                return False
+
+        return bool(node.find_children(type_id=workfile_type_id))
+    except exceptions.FlowError as exc:
+        logger.warning(
+            f'Could not verify whether pipeline step "{pipeline_step}" has a '
+            f'published workfile for "{sg_entity_name}". Allowing the build to '
+            f"proceed. ({exc})"
+        )
+        return True
+
+
+def _get_root_folder_name(sg_entity_type: str) -> Optional[str]:
+    """Return the name of the top-level folder holding assets of *sg_entity_type*.
+
+    Mirrors ``get_or_create_root_folder()`` in tk-core's ``tank/flowam/create.py``,
+    where the two folder names are asymmetric: assets live under ``ASSET_FOLDER``
+    ("Assets") while shots live under a folder named after ``SHOT_TYPE`` ("Shot").
+
+    :param sg_entity_type: FPTR entity type of the asset.
+    :returns: Folder name, or ``None`` when the entity type has no such folder.
+    """
+    if sg_entity_type == ASSET_TYPE:
+        return ASSET_FOLDER
+
+    if sg_entity_type == SHOT_TYPE:
+        return SHOT_TYPE
+
+    return None