Fix WES state mapping for completed/materializing invocations

completed invocations reported WES UNKNOWN (unmapped); map to COMPLETE.
Add requires_materialization -> INITIALIZING.
Strengthen test_wes_state_mapping: wait for completion, assert COMPLETE.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lib/galaxy/webapps/galaxy/services/wes.py

@@ -71,8 +71,10 @@ log = logging.getLogger(__name__)
 # Map Galaxy workflow invocation states to WES states
 GALAXY_TO_WES_STATE = {
     "new": State.QUEUED,
+    "requires_materialization": State.INITIALIZING,
     "ready": State.INITIALIZING,
     "scheduled": State.RUNNING,
+    "completed": State.COMPLETE,
     "failed": State.EXECUTOR_ERROR,
     "cancelled": State.CANCELED,
     "cancelling": State.CANCELING,

lib/galaxy_test/api/test_wes.py

@@ -386,22 +386,14 @@ steps:
                 history_id=history_id,
             )
 
-            # Get status via WES API
-            status = self._get_run_status_validated(invocation_id)
+            # A successfully finished invocation (Galaxy state "completed") must map to
+            # the terminal WES state COMPLETE - not UNKNOWN - so clients can poll for done.
+            self.workflow_populator.wait_for_invocation_and_completion(invocation_id, assert_ok=True)
+            invocation = self.workflow_populator.get_invocation(invocation_id)
+            assert invocation["state"] == "completed"
 
-            # Verify state is properly mapped
-            # After waiting for completion, should be COMPLETE or similar
-            assert status["state"] in [
-                "QUEUED",
-                "INITIALIZING",
-                "RUNNING",
-                "PAUSED",
-                "COMPLETE",
-                "EXECUTOR_ERROR",
-                "SYSTEM_ERROR",
-                "CANCELED",
-                "CANCELING",
-            ]
+            status = self._get_run_status_validated(invocation_id)
+            assert status["state"] == "COMPLETE"
 
     def test_wes_error_handling_missing_workflow_type(self):
         """Test error handling when workflow_type is missing."""