Stop suites from starting and stopping each other's workers

scan_for_annotated_workers makes a TaskHandler start a worker for every
@worker_task/@WorkerTask in every imported test module -- both decorators
share one registry -- not just the calling suite's. In one CI run the
core bucket showed it plainly:

  Discovered 8 annotated worker(s): ['async_basic_475f576a', ...,
    'v2_fallback_task_a_84d0a6cf', 'v2_fallback_task_b_84d0a6cf']

Two different RUN_IDs: test_comprehensive_e2e's handler was running the
v2-fallback workers 13s before that suite started, and each suite's
teardown stopped the other's workers. A task whose worker disappears
sits in SCHEDULED until it ages out, which surfaces later as the opaque
"workflow expected to be COMPLETED, but received RUNNING" and as
"0 completed, 0 failed, 5 pending" -- the two failures that have been
taking down runs that are otherwise green.

Pass each suite the workers it actually drives, filtered from
get_registered_workers() by name, and turn scanning off. This is the
pattern test_async_lease_extension already uses, whose comment describes
this exact symptom.

test_workflow_execution keeps import_modules so the decorators still
register in spawned children; it just no longer adopts everything they
registered.

Author: manan164

tests/integration/test_v2_fallback_intg.py

@@ -18,7 +18,7 @@
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
 
-from conductor.client.automator.task_handler import TaskHandler
+from conductor.client.automator.task_handler import TaskHandler, get_registered_workers
 from conductor.client.configuration.configuration import Configuration
 from conductor.client.http.models.start_workflow_request import StartWorkflowRequest
 from conductor.client.http.models.workflow_def import WorkflowDef
@@ -129,9 +129,20 @@ def test_1_workflows_complete_with_v2_or_fallback(self):
         handler_ref = {}
 
         def _run_workers():
+            # Only this module's two workers. With scan_for_annotated_workers
+            # every TaskHandler in the session starts a worker for every
+            # @worker_task in every imported test module -- so this suite was
+            # running test_comprehensive_e2e's workers, and its handler was
+            # running ours, each stopping the other's on teardown. Same pattern
+            # as test_async_lease_extension.
+            workers = [
+                w for w in get_registered_workers()
+                if w.get_task_definition_name() in (TASK_A, TASK_B)
+            ]
             handler = TaskHandler(
+                workers=workers,
                 configuration=self.config,
-                scan_for_annotated_workers=True,
+                scan_for_annotated_workers=False,
             )
             handler_ref["h"] = handler
             handler.start_processes()

tests/integration/workflow/test_workflow_execution.py

@@ -2,7 +2,7 @@
 import time
 from time import sleep
 
-from conductor.client.automator.task_handler import TaskHandler
+from conductor.client.automator.task_handler import TaskHandler, get_registered_workers
 from conductor.client.configuration.configuration import Configuration
 from conductor.client.http.models import StartWorkflowRequest
 from conductor.client.http.models import TaskDef
@@ -19,6 +19,9 @@
 TASK_NAME = "python_integration_test_task"
 WORKFLOW_VERSION = 1234
 WORKFLOW_OWNER_EMAIL = "test@test"
+# Decorated workers this module drives, pulled from the shared decorator
+# registry by name instead of scanning everything that happens to be imported.
+DECORATED_TASK_NAMES = ('test_python_decorated_worker',)
 COMPLEX_WF_NAME = 'complex_wf_signal_test'
 SUB_WF_1_NAME = 'complex_wf_signal_test_subworkflow_1'
 SUB_WF_2_NAME = 'complex_wf_signal_test_subworkflow_2'
@@ -47,10 +50,20 @@ def run_workflow_execution_tests(configuration: Configuration, workflow_executor
         generate_worker(worker_with_task_input_and_generic_output),
         generate_worker(worker_with_task_input_and_task_result_output),
     ]
+    # Add only the decorated workers this module actually drives. Scanning would
+    # start one for every @worker_task/@WorkerTask in every imported test module
+    # (both decorators share one registry), so this handler was starting other
+    # suites' workers and stopping them again on teardown -- which strands their
+    # tasks in SCHEDULED and surfaces as "workflow expected to be COMPLETED, but
+    # received RUNNING". Same pattern as test_async_lease_extension.
+    workers += [
+        w for w in get_registered_workers()
+        if w.get_task_definition_name() in DECORATED_TASK_NAMES
+    ]
     task_handler = TaskHandler(
         workers=workers,
         configuration=configuration,
-        scan_for_annotated_workers=True,
+        scan_for_annotated_workers=False,
         import_modules=['tests.integration.resources.worker.python.python_worker']
     )
     task_handler.start_processes()