Fix DataManagerTool.exec_after_process for async API submission

When a data manager job is submitted via POST /api/jobs (async path), the
Celery task reconstructs the tool from raw XML without a data_manager_id,
causing DataManagerJobAssociation to store the tool XML id instead of the
conf id when those differ. exec_after_process then fails with "Invalid data
manager requested" because DataManagers registry is keyed by conf id.

Fix by preferring self.data_manager_id (always set correctly at toolbox load
time) over job.data_manager_association.data_manager_id in exec_after_process.

Add a regression test using the "data_manager" test tool (tool XML id) which
is registered in sample_data_manager_conf.xml as id="test_data_manager",
replicating the exact id mismatch seen in production (e.g. bowtie2).

lib/galaxy/tools/__init__.py

@@ -3772,7 +3772,10 @@ class DataManagerTool(OutputParameterJSONTool):
             return
         super().exec_after_process(app, inp_data, out_data, param_dict, job=job, final_job_state=final_job_state)
         # process results of tool
-        data_manager_id = job.data_manager_association.data_manager_id
+        # Use self.data_manager_id (from data_manager_conf.xml, set at toolbox load time) because
+        # the async API reconstructs the tool from raw XML without data_manager_id, causing the
+        # DataManagerJobAssociation to store the tool XML id instead of the conf id when those differ.
+        data_manager_id = self.data_manager_id or job.data_manager_association.data_manager_id
         data_manager = self.app.data_managers.get_manager(data_manager_id)
         assert (
             data_manager is not None

lib/galaxy_test/api/test_jobs.py

@@ -1329,3 +1329,39 @@ steps:
         jobs = jobs_response.json()
         assert isinstance(jobs, list)
         return jobs
+
+
+class TestDataManagerJobsApi(ApiTestCase):
+    """API tests for data manager jobs submitted via the async POST /api/jobs endpoint."""
+
+    require_admin_user = True
+    dataset_populator: DatasetPopulator
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @skip_without_tool("data_manager")
+    def test_data_manager_async_submission_with_mismatched_conf_id(self):
+        # Regression for data manager jobs submitted via POST /api/jobs failing with
+        # "Invalid data manager requested" when the data_manager_conf.xml <data_manager id>
+        # differs from the tool XML <tool id>. The test tool "data_manager" (tool XML id)
+        # is registered in sample_data_manager_conf.xml as id="test_data_manager", which
+        # is exactly that mismatch. Before the fix, exec_after_process looked up the data
+        # manager using DataManagerJobAssociation.data_manager_id, which was set from the
+        # reconstructed tool's XML id ("data_manager") rather than the conf id
+        # ("test_data_manager"), causing the lookup to return None and the job to fail with
+        # exit code 0 but error state.
+        with self.dataset_populator.test_history() as history_id:
+            response = self.dataset_populator.tool_request_raw(
+                tool_id="data_manager",
+                inputs={"ignored_value": "test", "exit_code": 0},
+                history_id=history_id,
+                strict=False,
+            )
+            response.raise_for_status()
+            tool_request_id = response.json()["tool_request_id"]
+            submitted = self.dataset_populator.wait_on_tool_request(tool_request_id)
+            assert submitted, self.dataset_populator.get_tool_request(tool_request_id)
+            jobs = self.galaxy_interactor.jobs_for_tool_request(tool_request_id)
+            self.dataset_populator.wait_for_jobs(jobs, assert_ok=True)

test/functional/tools/data_manager.xml

@@ -1,10 +1,10 @@
 <tool id="data_manager" name="Test Data Manager" tool_type="manage_data" version="0.0.1">
     <configfiles>
-        <configfile name="static_test_data">{"data_tables": {"testbeta": [{"value": "newvalue", "path": "newvalue.txt"}]}}</configfile>
+        <configfile name="static_test_data">{"data_tables": {"testbeta": [{"value": "asyncvalue", "path": "asyncvalue.txt"}]}}</configfile>
     </configfiles>
     <command detect_errors="exit_code">
         mkdir $out_file.files_path ;
-        echo "A new value" > '$out_file.files_path/newvalue.txt';
+        echo "A new value" > '$out_file.files_path/asyncvalue.txt';
         cp '$static_test_data' '$out_file';
         exit $exit_code
     </command>