Poll to terminal instead of fixed sleeps in the last two flaky spots

Task-level diagnostics from a red run showed these are two different
problems, not the one I assumed:

scenario_decorated_workers: task SCHEDULED, pollCount=0, workerId=None
after sleep(15), while both decorated workers were demonstrably active
and polling every 100ms. The server simply had not handed the task out
inside that window. This is the same false negative the batch-completion
budget in this file was already raised to fix, so use that budget here
and poll to terminal.

test_v2_fallback_intg: 4 of a workflow's 5 tasks COMPLETED with the last
IN_PROGRESS on a live worker (pollCount=1, workerId set) when the 60s
budget expired -- it was progressing, not stuck. Raised to 120s. A task
still IN_PROGRESS after that is a real lost update rather than slowness,
and the diagnostic prints it.

Author: manan164

tests/integration/test_v2_fallback_intg.py

@@ -142,8 +142,13 @@ def _run_workers():
 
             print(f"\n  Submitted {len(workflow_ids)} workflows")
 
-            # Wait for completion
-            deadline = time.time() + 60  # 60s timeout
+            # Wait for completion. 60s was marginal: a red run showed 4 of a
+            # workflow's 5 tasks COMPLETED and the last one still IN_PROGRESS on
+            # a live worker, i.e. the run was progressing when the budget ran
+            # out. Give the shared server the same headroom the other suites
+            # use. A task still IN_PROGRESS after this is a genuine stuck
+            # update, and the diagnostic below prints it.
+            deadline = time.time() + 120
             pending = set(workflow_ids)
             completed = 0
             failed = 0

tests/integration/workflow/test_workflow_execution.py

@@ -156,8 +156,20 @@ def scenario_decorated_workers(
     start_wf_req = StartWorkflowRequest(name=workflow_name, task_to_domain=td_map)
     workflow_id_2 = workflow_executor.start_workflow(start_wf_req)
 
-    logger.debug(f'started TestPythonDecoratedWorkerWf with domain:cool and id: {workflow_id_2}')
-    sleep(15)
+    logger.info('started TestPythonDecoratedWorkerWf %s (no domain) and %s (domain:cool)',
+                workflow_id, workflow_id_2)
+
+    # Poll to terminal instead of sleeping a fixed 15s. Both of these run a
+    # single decorated-worker task, and on a loaded shared server the task can
+    # sit SCHEDULED well past 15s before the server hands it to a poller --
+    # observed as status=SCHEDULED pollCount=0 while the worker was demonstrably
+    # alive and polling every 100ms. Same false-negative the batch-completion
+    # budget above was raised to fix; use that budget here too.
+    for wf_id in (workflow_id, workflow_id_2):
+        wait_for_workflow_terminal(
+            workflow_executor, wf_id,
+            timeout_seconds=WORKFLOW_COMPLETION_MAX_WAIT_SECONDS,
+        )
 
     _run_with_retry_attempt(
         validate_workflow_status,