Feature/ngs360 workflow pin version

CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Changed
+
+- `NGS360Platform.copy_workflow` now resolves the workflow version the project
+  uses and returns `<uuid>:<version>`, instead of returning the workflow id
+  unchanged. A project therefore keeps running one version of a workflow even
+  after the `prod` alias moves on. An id that already names an alias or version
+  (`<uuid>:3`) is still returned untouched. Requires an NGS360 API server with
+  `/projects/{id}/workflow-versions`. SevenBridges and Arvados are unaffected,
+  and callers need no change.
+
 ## [v0.6.1](https://github.com/NGS360/PAML/releases/tag/v0.6.1) - 2026-09-21
 
 ### Changed

src/cwl_platform/ngs360_platform.py

@@ -13,6 +13,9 @@
 
 from .base_platform import Platform
 
+OMICS_ENGINE = "AWSHealthOmics (us-east)"
+PROD_ALIAS = "prod"
+
 
 class WESTask:
     """
@@ -362,13 +365,81 @@ def upload_file(
         return None
 
     # Task/Workflow methods
+    def _resolve_workflow_version(self, project_id, workflow_id):
+        """
+        Ask NGS360 which version of a workflow this project uses, recording the
+        version the `prod` alias points at if the project has none yet.
+
+        :param project_id: NGS360 project business key, e.g. P-00000000-0001
+        :param workflow_id: NGS360 workflow UUID
+        :return: The resolved record as a dict, including `version`
+        :raises RuntimeError: If NGS360 will not resolve a version
+        """
+        response = requests.post(
+            f"{self.ngs360_endpoint}/api/v1/projects/{project_id}"
+            f"/workflow-versions/{workflow_id}/resolve",
+            json={"engine": OMICS_ENGINE, "alias": PROD_ALIAS},
+            headers=self._auth_headers(),
+            timeout=30
+        )
+        if response.status_code not in (200, 201):
+            # Deliberately fatal. Raising error instead of returning the bare
+            # workflow id, which would leave WES to pick the highest version.
+            try:
+                detail = response.json().get("detail", response.text)
+            except ValueError:
+                detail = response.text
+            self.logger.error(
+                "NGS360 would not resolve a version of workflow %s for project %s: %s",
+                workflow_id, project_id, detail
+            )
+            raise RuntimeError(
+                f"Could not resolve a version of workflow {workflow_id} for "
+                f"project {project_id} on {OMICS_ENGINE}: {detail}"
+            )
+        return response.json()
+
     def copy_workflow(self, src_workflow, destination_project):
         """
-        Copy a workflow from one project to another
+        Return the workflow identifier this project should submit with.
+
+        Nothing is copied: NGS360 has no per-project copy of a workflow. Instead
+        the project's recorded version is resolved and returned as
+        "<uuid>:<version>", so every run of a project uses the same version even
+        after the `prod` alias moves on.
+
+        An identifier that already carries an alias or version ("<uuid>:prod",
+        "<uuid>:3") is returned untouched, and the project's recorded version is
+        neither read nor written -- a one-off run must not change what the
+        project uses from then on.
 
-        Note: WES API doesn't have a direct concept of projects, so this returns the workflow
+        :param src_workflow: NGS360 workflow UUID, optionally ":<alias-or-version>"
+        :param destination_project: The project the workflow will run for
+        :return: The workflow identifier to submit
+        :raises RuntimeError: If NGS360 will not resolve a version
         """
-        return src_workflow
+        if ":" in src_workflow:
+            self.logger.info(
+                "Workflow %s names a version explicitly; using it unchanged",
+                src_workflow
+            )
+            return src_workflow
+
+        project_id = (destination_project or {}).get("project_id")
+        if not project_id:
+            raise RuntimeError(
+                f"A project with a project_id is required to resolve a version of "
+                f"workflow {src_workflow}; got {destination_project!r}"
+            )
+
+        record = self._resolve_workflow_version(project_id, src_workflow)
+        version = record["version"]
+        self.logger.info(
+            "Workflow %s resolved to version %s for project %s (%s)",
+            src_workflow, version, project_id,
+            "newly recorded" if record.get("created") else "already recorded"
+        )
+        return f"{src_workflow}:{version}"
 
     def copy_workflows(self, reference_project, destination_project):
         """

tests/test_ngs360_ga4gh_platform.py

@@ -915,10 +915,6 @@ def test_workflow_methods(self):
         '''
         project = {'project_id': 'test_project'}
 
-        # Test copy_workflow
-        result = self.platform.copy_workflow('src_workflow', project)
-        self.assertEqual(result, 'src_workflow')
-
         # Test copy_workflows
         result = self.platform.copy_workflows('ref_project', project)
         self.assertEqual(result, [])
@@ -1044,6 +1040,112 @@ def test_current_task_and_users(self):
         result = self.platform.get_project_users(project)
         self.assertEqual(result, [])
 
+    # --- copy_workflow: resolving the version a project uses -----------------
+
+    PROJECT = {'project_id': 'P-00000000-0001'}
+    WORKFLOW_ID = 'b6fd2ade-0000-0000-0000-000000000001'
+    RESOLVE_URL = (
+        'https://ngs360.example.com/api/v1/projects/P-00000000-0001'
+        '/workflow-versions/b6fd2ade-0000-0000-0000-000000000001/resolve'
+    )
+
+    @staticmethod
+    def _resolve_response(status_code=200, **body):
+        '''Build a mock NGS360 resolve response'''
+        response = MagicMock()
+        response.status_code = status_code
+        response.json.return_value = body
+        return response
+
+    @patch('requests.post')
+    def test_copy_workflow_records_version_on_first_use(self, mock_post):
+        '''
+        A bare workflow id resolves to the project's version, and is returned
+        suffixed with it so WES runs that exact version
+        '''
+        mock_post.return_value = self._resolve_response(201, version=2, created=True)
+
+        result = self.platform.copy_workflow(self.WORKFLOW_ID, self.PROJECT)
+
+        self.assertEqual(result, f'{self.WORKFLOW_ID}:2')
+        mock_post.assert_called_once_with(
+            self.RESOLVE_URL,
+            json={'engine': 'AWSHealthOmics (us-east)', 'alias': 'prod'},
+            headers={'Authorization': 'Bearer test_token'},
+            timeout=30
+        )
+
+    @patch('requests.post')
+    def test_copy_workflow_reuses_recorded_version(self, mock_post):
+        '''
+        The recorded version is used even once the alias has moved past it,
+        which is the whole point of recording it
+        '''
+        mock_post.return_value = self._resolve_response(
+            200, version=2, created=False,
+            message='Project already records version 2; returned it instead of alias \'prod\'.'
+        )
+
+        result = self.platform.copy_workflow(self.WORKFLOW_ID, self.PROJECT)
+
+        self.assertEqual(result, f'{self.WORKFLOW_ID}:2')
+
+    @patch('requests.post')
+    def test_copy_workflow_leaves_explicit_versions_alone(self, mock_post):
+        '''
+        An id that already names an alias or version is submitted as given, and
+        the project's recorded version is neither read nor written
+        '''
+        for explicit in (f'{self.WORKFLOW_ID}:3', f'{self.WORKFLOW_ID}:prod'):
+            result = self.platform.copy_workflow(explicit, self.PROJECT)
+            self.assertEqual(result, explicit)
+
+        mock_post.assert_not_called()
+
+    @patch('requests.post')
+    def test_copy_workflow_raises_when_no_version_can_be_resolved(self, mock_post):
+        '''
+        A refusal from NGS360 stops the caller rather than falling back to the
+        bare id, which WES would resolve to the highest version
+        '''
+        for status_code in (404, 422):
+            mock_post.return_value = self._resolve_response(
+                status_code, detail='No deployment on that engine'
+            )
+            with self.assertRaises(RuntimeError) as raised:
+                self.platform.copy_workflow(self.WORKFLOW_ID, self.PROJECT)
+            self.assertIn('No deployment on that engine', str(raised.exception))
+
+    @patch('requests.post')
+    def test_copy_workflow_requires_a_project(self, mock_post):
+        '''
+        Without a project there is nothing to resolve a version for, and the
+        error should say so rather than failing on a missing key
+        '''
+        for no_project in (None, {}, {'name': 'Test Project'}):
+            with self.assertRaises(RuntimeError) as raised:
+                self.platform.copy_workflow(self.WORKFLOW_ID, no_project)
+            self.assertIn('project_id', str(raised.exception))
+
+        mock_post.assert_not_called()
+
+    @patch('requests.request')
+    def test_task_reuse_ignores_the_version_suffix(self, mock_request):
+        '''
+        Tasks are matched on tags, so the change from "<uuid>" to "<uuid>:N"
+        does not stop a rerun reusing completed tasks
+        '''
+        mock_response = MagicMock()
+        mock_response.content = b'{"runs": []}'
+        mock_response.json.return_value = {'runs': []}
+        mock_request.return_value = mock_response
+
+        self.platform.get_tasks_by_name(self.PROJECT, task_name='Sample 1 - alignment')
+
+        _, kwargs = mock_request.call_args
+        filters = json.loads(kwargs['params']['filters'])
+        self.assertNotIn('workflow_url', filters)
+
     def test_detect_method(self):
         '''
         Test detect class method