diff --git a/.github/workflows/pull_request.yml b/.github/workflows/pull_request.yml
index 2bb0b0a7..5819f97a 100644
--- a/.github/workflows/pull_request.yml
+++ b/.github/workflows/pull_request.yml
@@ -1,9 +1,19 @@
 name: Continuous Integration
 
-on: [pull_request, workflow_dispatch]
+on:
+  push:
+    branches:
+      - main
+      # Temporary: run CI on this branch's pushes until it's exercised via a
+      # real PR targeting main. Remove once that PR validates the workflow.
+      - integ-test-on-pr
+  pull_request:
+    branches:
+      - main
+  workflow_dispatch:
 
 concurrency:
-  group: ${{ github.workflow }}-${{ github.ref }}
+  group: ${{ github.workflow }}-${{ github.head_ref || github.ref }}
   cancel-in-progress: true
 
 jobs:
@@ -35,9 +45,6 @@ jobs:
         id: unit_tests
         continue-on-error: true
         env:
-          CONDUCTOR_AUTH_KEY: ${{ secrets.CONDUCTOR_AUTH_KEY }}
-          CONDUCTOR_AUTH_SECRET: ${{ secrets.CONDUCTOR_AUTH_SECRET }}
-          CONDUCTOR_SERVER_URL: ${{ secrets.CONDUCTOR_SERVER_URL }}
           COVERAGE_FILE: ${{ env.COVERAGE_DIR }}/.coverage.unit
         run: |
           coverage run -m pytest tests/unit -v
@@ -46,9 +53,6 @@ jobs:
         id: bc_tests
         continue-on-error: true
         env:
-          CONDUCTOR_AUTH_KEY: ${{ secrets.CONDUCTOR_AUTH_KEY }}
-          CONDUCTOR_AUTH_SECRET: ${{ secrets.CONDUCTOR_AUTH_SECRET }}
-          CONDUCTOR_SERVER_URL: ${{ secrets.CONDUCTOR_SERVER_URL }}
           COVERAGE_FILE: ${{ env.COVERAGE_DIR }}/.coverage.bc
         run: |
           coverage run -m pytest tests/backwardcompatibility -v
@@ -57,9 +61,6 @@ jobs:
         id: serdeser_tests
         continue-on-error: true
         env:
-          CONDUCTOR_AUTH_KEY: ${{ secrets.CONDUCTOR_AUTH_KEY }}
-          CONDUCTOR_AUTH_SECRET: ${{ secrets.CONDUCTOR_AUTH_SECRET }}
-          CONDUCTOR_SERVER_URL: ${{ secrets.CONDUCTOR_SERVER_URL }}
           COVERAGE_FILE: ${{ env.COVERAGE_DIR }}/.coverage.serdeser
         run: |
           coverage run -m pytest tests/serdesertest -v
@@ -92,4 +93,35 @@ jobs:
 
       - name: Check test results
         if: steps.unit_tests.outcome == 'failure' || steps.bc_tests.outcome == 'failure' || steps.serdeser_tests.outcome == 'failure'
-        run: exit 1
\ No newline at end of file
+        run: exit 1
+
+  integration-test:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    strategy:
+      fail-fast: false
+      matrix:
+        bucket: [test-all, long-sync, long-async, core]
+    name: integration-test (${{ matrix.bucket }})
+    env:
+      CONDUCTOR_SERVER_URL: ${{ vars.SDKDEV_V5_SERVER_URL }}
+      CONDUCTOR_AUTH_KEY: ${{ vars.SDKDEV_V5_AUTH_KEY }}
+      CONDUCTOR_AUTH_SECRET: ${{ secrets.SDKDEV_V5_AUTH_SECRET }}
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+          cache: 'pip'
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e .
+          pip install pytest
+
+      - name: Run integration tests
+        run: bash scripts/run_integration_tests.sh --bucket=${{ matrix.bucket }}
\ No newline at end of file
diff --git a/pyproject.toml b/pyproject.toml
index 6a699ca7..14a16d4e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -159,6 +159,11 @@ line-ending = "auto"
 
 [tool.pytest.ini_options]
 pythonpath = ["src"]
+markers = [
+    "slow_sync: long-running sync lease-extension tests (~90s)",
+    "slow_async: long-running async lease-extension tests (~90s)",
+    "slow_test_all: long-running aggregate workflow-client test_all (~83s)",
+]
 
 [tool.coverage.run]
 source = ["src/conductor"]
diff --git a/scripts/run_integration_tests.sh b/scripts/run_integration_tests.sh
new file mode 100755
index 00000000..cce260e3
--- /dev/null
+++ b/scripts/run_integration_tests.sh
@@ -0,0 +1,130 @@
+#!/usr/bin/env bash
+#
+# Run the SDK integration test suite locally, mirroring the `integration-test`
+# job in .github/workflows/pull_request.yml.
+#
+# The AI/agentic integration tests are excluded on purpose: they target a
+# dedicated AI-enabled server (test_ai_task_types.py and test_ai_examples.py
+# hardcode http://localhost:7001/api, and test_agentic_workflows.py needs an
+# `openai` LLM provider configured on the server), so they don't run against
+# the standard test server this suite targets.
+#
+# The performance test (test_update_task_v2_perf.py) is also excluded by
+# default: it submits ~1000 workflows and takes several minutes. Pass
+# --with-perf to include it.
+#
+# Server connection is read from the environment (see
+# src/conductor/client/configuration/configuration.py):
+#   CONDUCTOR_SERVER_URL   (required; defaults to http://localhost:8080/api)
+#   CONDUCTOR_AUTH_KEY     (optional; needed for Orkes/authenticated servers)
+#   CONDUCTOR_AUTH_SECRET  (optional; needed for Orkes/authenticated servers)
+#
+# Usage:
+#   export CONDUCTOR_SERVER_URL="http://localhost:8080/api"
+#   ./scripts/run_integration_tests.sh
+#   ./scripts/run_integration_tests.sh --with-perf   # also run the perf test
+#
+# Buckets (--bucket=<name>): the slowest tests are split into their own buckets
+# so they can run as separate parallel CI jobs and are skipped by default
+# locally. Each slow bucket is path-scoped so it only imports its own module.
+#   core        (default) everything except the slow buckets below
+#   long-sync   sync lease-extension tests   (test_lease_extension.py, ~90s)
+#   long-async  async lease-extension tests  (test_async_lease_extension.py, ~90s)
+#   test-all    aggregate workflow-client test_all (test_workflow_client_intg.py, ~83s)
+#   all         the full suite (no bucket filtering) — for a complete local run
+#
+#   ./scripts/run_integration_tests.sh                  # fast: skips slow buckets
+#   ./scripts/run_integration_tests.sh --bucket=long-sync
+#   ./scripts/run_integration_tests.sh --bucket=all     # run everything
+#
+# Any other arguments are passed straight through to pytest, which is handy for
+# targeting a subset of tests or getting more detail on failures:
+#
+#   # run a subset (by keyword or path) with live output
+#   ./scripts/run_integration_tests.sh -s -k lease
+#   ./scripts/run_integration_tests.sh tests/integration/test_lease_extension.py
+#
+#   # short tracebacks + a one-line reason for every failure/skip, with live logs
+#   ./scripts/run_integration_tests.sh -ra --tb=short --log-cli-level=INFO
+#
+#   # stop at the first failure instead of waiting for the whole suite
+#   ./scripts/run_integration_tests.sh -x --tb=long
+#
+#   # re-run just specific tests, e.g. the ones that failed
+#   ./scripts/run_integration_tests.sh -s -k "upsert_group or create_role"
+#
+#   # show the 15 slowest tests (setup/call/teardown timings) after the run
+#   ./scripts/run_integration_tests.sh --durations=15
+
+set -euo pipefail
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"
+
+# Slow-test files that get their own buckets / CI jobs (see --bucket above).
+LEASE_SYNC=tests/integration/test_lease_extension.py
+LEASE_ASYNC=tests/integration/test_async_lease_extension.py
+TEST_ALL_FILE=tests/integration/test_workflow_client_intg.py
+
+# The perf test is skipped by default; --with-perf opts back in.
+perf_ignore=(--ignore=tests/integration/test_update_task_v2_perf.py)
+bucket="core"
+pytest_args=()
+for arg in "$@"; do
+  case "$arg" in
+    --with-perf) perf_ignore=() ;;
+    --bucket=*)  bucket="${arg#*=}" ;;
+    *)           pytest_args+=("$arg") ;;
+  esac
+done
+
+# The AI/agentic tests always target a dedicated server (see header) and are
+# never part of these buckets.
+ai_ignore=(
+  --ignore=tests/integration/test_ai_task_types.py
+  --ignore=tests/integration/test_ai_examples.py
+  --ignore=tests/integration/test_agentic_workflows.py
+)
+
+# Build the target paths + selection for the chosen bucket. The slow buckets are
+# path-scoped so each job only imports its own module: this keeps
+# scan_for_annotated_workers from starting unrelated workers and lets the four
+# buckets run in parallel against one server without stealing each other's tasks.
+case "$bucket" in
+  core)
+    targets=(tests/integration)
+    select=("${ai_ignore[@]}" ${perf_ignore[@]+"${perf_ignore[@]}"} \
+      --ignore="$LEASE_SYNC" --ignore="$LEASE_ASYNC" --ignore="$TEST_ALL_FILE")
+    ;;
+  all)
+    targets=(tests/integration)
+    select=("${ai_ignore[@]}" ${perf_ignore[@]+"${perf_ignore[@]}"})
+    ;;
+  # The slow buckets below target a single specific file, so the perf test is
+  # never in scope and perf_ignore is unnecessary here. This also means
+  # --with-perf has no effect on these buckets; the perf test is only reachable
+  # via the core/all buckets.
+  long-sync)
+    targets=("$LEASE_SYNC")
+    select=(-m slow_sync)
+    ;;
+  long-async)
+    targets=("$LEASE_ASYNC")
+    select=(-m slow_async)
+    ;;
+  test-all)
+    targets=("$TEST_ALL_FILE")
+    select=(-m slow_test_all)
+    ;;
+  *)
+    echo "Unknown --bucket='$bucket' (expected: core, long-sync, long-async, test-all, all)" >&2
+    exit 2
+    ;;
+esac
+
+# Note: the "${arr[@]+"${arr[@]}"}" form is required so empty arrays don't trip
+# "unbound variable" under `set -u` on bash 3.2 (the default macOS bash).
+exec python3 -m pytest -v \
+  "${targets[@]}" \
+  ${select[@]+"${select[@]}"} \
+  ${pytest_args[@]+"${pytest_args[@]}"}
diff --git a/src/conductor/client/http/models/service_method.py b/src/conductor/client/http/models/service_method.py
index df03f550..3f6f4b23 100644
--- a/src/conductor/client/http/models/service_method.py
+++ b/src/conductor/client/http/models/service_method.py
@@ -15,7 +15,7 @@ class ServiceMethod:
         'input_type': 'str',
         'output_type': 'str',
         'request_params': 'list[RequestParam]',
-        'example_input': 'dict'
+        'example_input': 'object'
     }
 
     attribute_map = {
diff --git a/src/conductor/client/http/models/upsert_user_request.py b/src/conductor/client/http/models/upsert_user_request.py
index 9d455be0..89305c7f 100644
--- a/src/conductor/client/http/models/upsert_user_request.py
+++ b/src/conductor/client/http/models/upsert_user_request.py
@@ -2,7 +2,7 @@
 import re  # noqa: F401
 import six
 from dataclasses import dataclass, field, InitVar
-from typing import List, Optional
+from typing import Dict, List, Optional
 from enum import Enum
 
 
@@ -30,41 +30,50 @@ class UpsertUserRequest:
     name: InitVar[Optional[str]] = None
     roles: InitVar[Optional[List[str]]] = None
     groups: InitVar[Optional[List[str]]] = None
+    contact_information: InitVar[Optional[Dict[str, str]]] = None
     
     _name: str = field(default=None, init=False)
     _roles: List[str] = field(default=None, init=False)
     _groups: List[str] = field(default=None, init=False)
+    _contact_information: Dict[str, str] = field(default=None, init=False)
     
     swagger_types = {
         'name': 'str',
         'roles': 'list[str]',
-        'groups': 'list[str]'
+        'groups': 'list[str]',
+        'contact_information': 'dict(str, str)'
     }
 
     attribute_map = {
         'name': 'name',
         'roles': 'roles',
-        'groups': 'groups'
+        'groups': 'groups',
+        'contact_information': 'contactInformation'
     }
 
-    def __init__(self, name=None, roles=None, groups=None):  # noqa: E501
+    def __init__(self, name=None, roles=None, groups=None, contact_information=None):  # noqa: E501
         """UpsertUserRequest - a model defined in Swagger"""  # noqa: E501
         self._name = None
         self._roles = None
         self._groups = None
+        self._contact_information = None
         self.discriminator = None
         self.name = name
         if roles is not None:
             self.roles = roles
         if groups is not None:
             self.groups = groups
+        if contact_information is not None:
+            self.contact_information = contact_information
 
-    def __post_init__(self, name, roles, groups):
+    def __post_init__(self, name, roles, groups, contact_information):
         self.name = name
         if roles is not None:
             self.roles = roles
         if groups is not None:
             self.groups = groups
+        if contact_information is not None:
+            self.contact_information = contact_information
 
     @property
     def name(self):
@@ -139,6 +148,29 @@ def groups(self, groups):
 
         self._groups = groups
 
+    @property
+    def contact_information(self):
+        """Gets the contact_information of this UpsertUserRequest.  # noqa: E501
+
+        User's contact information, e.g. {"email": "user@example.com"}  # noqa: E501
+
+        :return: The contact_information of this UpsertUserRequest.  # noqa: E501
+        :rtype: dict(str, str)
+        """
+        return self._contact_information
+
+    @contact_information.setter
+    def contact_information(self, contact_information):
+        """Sets the contact_information of this UpsertUserRequest.
+
+        User's contact information, e.g. {"email": "user@example.com"}  # noqa: E501
+
+        :param contact_information: The contact_information of this UpsertUserRequest.  # noqa: E501
+        :type: dict(str, str)
+        """
+
+        self._contact_information = contact_information
+
     def to_dict(self):
         """Returns the model properties as a dict"""
         result = {}
diff --git a/tests/integration/README.md b/tests/integration/README.md
index ac00666e..bc99d312 100644
--- a/tests/integration/README.md
+++ b/tests/integration/README.md
@@ -38,8 +38,55 @@ export CONDUCTOR_AUTH_SECRET="your-secret"
 
 ## Running Tests
 
+### Run the CI suite locally (recommended)
+
+Use the helper script to run exactly what CI runs (the `integration-test` job in
+[`.github/workflows/pull_request.yml`](../../.github/workflows/pull_request.yml)).
+It excludes the AI/agentic tests (which need a dedicated AI-enabled server) and
+the slow performance test, so you don't have to remember the `--ignore` flags:
+
+```bash
+export CONDUCTOR_SERVER_URL="http://localhost:8080/api"
+# For Orkes / authenticated servers also set:
+# export CONDUCTOR_AUTH_KEY="your-key"
+# export CONDUCTOR_AUTH_SECRET="your-secret"
+
+./scripts/run_integration_tests.sh
+
+# Also run the performance test (test_update_task_v2_perf.py, ~1000 workflows,
+# several minutes):
+./scripts/run_integration_tests.sh --with-perf
+```
+
+By default this runs the fast `core` bucket and **skips the slowest tests**
+(a few tests deliberately sleep ~50-90s to exercise lease-extension timeouts and
+one aggregate `test_all`). Those live in their own buckets, each of which runs as
+a separate parallel CI job. Select one with `--bucket=<name>`:
+
+| Bucket | What it runs |
+| --- | --- |
+| `core` (default) | everything except the slow buckets below |
+| `long-sync` | sync lease-extension tests (`test_lease_extension.py`, ~90s) |
+| `long-async` | async lease-extension tests (`test_async_lease_extension.py`, ~90s) |
+| `test-all` | aggregate workflow-client `test_all` (`test_workflow_client_intg.py`, ~83s) |
+| `all` | the full suite (no bucket filtering) — for a complete local run |
+
+```bash
+./scripts/run_integration_tests.sh                  # fast: skips the slow buckets
+./scripts/run_integration_tests.sh --bucket=long-sync
+./scripts/run_integration_tests.sh --bucket=all     # run everything
+```
+
+Any extra arguments pass straight through to pytest, which is handy for
+targeting a subset of tests or getting more detail on failures. See additional
+options and examples in the comments at the top of
+[`scripts/run_integration_tests.sh`](../../scripts/run_integration_tests.sh).
+
 ### Run All Integration Tests
 
+This includes the AI/agentic tests, which require an AI-enabled server (see
+[Tests excluded by default](#tests-excluded-by-default) below):
+
 ```bash
 python3 -m pytest tests/integration/ -v -s
 ```
@@ -442,46 +489,42 @@ To add more test scenarios:
 
 ---
 
-## CI/CD Integration
+## Tests excluded by default
+
+`scripts/run_integration_tests.sh` (and CI) skip a few tests by default.
 
-### GitHub Actions Example
+**AI/agentic tests** need a dedicated AI-enabled server:
 
-```yaml
-name: Integration Tests
+- `test_ai_task_types.py` and `test_ai_examples.py` hardcode
+  `http://localhost:7001/api`.
+- `test_agentic_workflows.py` needs an `openai` LLM provider (model
+  `gpt-4o-mini`) configured on the server.
 
-on: [push, pull_request]
+Run them only against a suitably configured AI-enabled server, e.g.:
 
-jobs:
-  integration:
-    runs-on: ubuntu-latest
+```bash
+python3 -m pytest tests/integration/test_ai_task_types.py -v -s
+```
 
-    services:
-      conductor:
-        image: conductoross/conductor-standalone:3.15.0
-        ports:
-          - 8080:8080
-          - 5000:5000
+**Performance test** (`test_update_task_v2_perf.py`) submits ~1000 workflows and
+takes several minutes. Include it with the `--with-perf` flag:
 
-    steps:
-      - uses: actions/checkout@v2
+```bash
+./scripts/run_integration_tests.sh --with-perf
+```
 
-      - name: Set up Python
-        uses: actions/setup-python@v2
-        with:
-          python-version: '3.9'
+---
 
-      - name: Install dependencies
-        run: pip install -e .
+## CI/CD Integration
 
-      - name: Wait for Conductor
-        run: |
-          timeout 60 bash -c 'until curl -f http://localhost:8080/api/health; do sleep 2; done'
+Integration tests run in CI via the `integration-test` job in
+[`.github/workflows/pull_request.yml`](../../.github/workflows/pull_request.yml)
+on pushes to `main`, PRs targeting `main`, and manual dispatch. The job invokes
+`scripts/run_integration_tests.sh` (excluding the AI tests and the performance
+test) and reads the server from the `SDKDEV_V5_*` repository variables/secret.
 
-      - name: Run integration tests
-        env:
-          CONDUCTOR_SERVER_URL: http://localhost:8080/api
-        run: python3 -m pytest tests/integration/ -v -s
-```
+To reproduce the CI run locally, use the script documented in
+[Running Tests](#running-tests) above.
 
 ---
 
diff --git a/tests/integration/client/orkes/test_orkes_clients.py b/tests/integration/client/orkes/test_orkes_clients.py
index 56e22ae4..2060da25 100644
--- a/tests/integration/client/orkes/test_orkes_clients.py
+++ b/tests/integration/client/orkes/test_orkes_clients.py
@@ -1,4 +1,5 @@
 import json
+import time
 
 from shortuuid import uuid
 
@@ -38,6 +39,19 @@
 TEST_IP_JSON = 'tests/integration/resources/test_data/loan_workflow_input.json'
 
 
+def _retry_on_404(func, *args, retries=5, **kwargs):
+    # Updating a task by ref name can transiently 404 while the server is still
+    # scheduling the referenced task. Retry with backoff to tolerate that race.
+    for attempt in range(retries):
+        try:
+            return func(*args, **kwargs)
+        except ApiException as e:
+            if e.status == 404 and attempt < retries - 1:
+                time.sleep(1 << attempt)
+                continue
+            raise
+
+
 class TestOrkesClients:
     def __init__(self, configuration: Configuration):
         self.api_client = ApiClient(configuration)
@@ -561,7 +575,7 @@ def __test_task_execution_lifecycle(self):
         assert self.task_client.get_queue_size_for_task(TASK_TYPE) == 1
 
         taskResult = TaskResult(
-            workflow_instance_id=workflow_uuid,
+            workflow_instance_id=polledTask.workflow_instance_id,
             task_id=polledTask.task_id,
             status=TaskResultStatus.COMPLETED
         )
@@ -571,31 +585,41 @@ def __test_task_execution_lifecycle(self):
         task = self.task_client.get_task(polledTask.task_id)
         assert task.status == TaskResultStatus.COMPLETED
 
-        batchPolledTasks = self.task_client.batch_poll_tasks(TASK_TYPE)
-        assert len(batchPolledTasks) == 1
-
-        polledTask = batchPolledTasks[0]
-        # Update first task of second workflow
-        self.task_client.update_task_by_ref_name(
-            workflow_uuid_2,
-            polledTask.reference_task_name,
-            "COMPLETED",
-            "task 2 op 2nd wf"
-        )
-
-        # Update second task of first workflow
-        self.task_client.update_task_by_ref_name(
-            workflow_uuid_2, "simple_task_ref_2", "COMPLETED", "task 2 op 1st wf"
-        )
-
-        # # Second task of second workflow is in the queue
-        # assert self.task_client.getQueueSizeForTask(TASK_TYPE) == 1
-        polledTask = self.task_client.poll_task(TASK_TYPE)
+        # The second task of the first workflow and both tasks of the second
+        # workflow all share TASK_TYPE and land in the same queue, so a poll can
+        # return a task from either workflow in a non-deterministic order. Drive
+        # every update from the polled task's own workflow id and reference name
+        # instead of assuming which workflow/ref we got back (the previous
+        # hardcoded workflow_uuid_2 / "simple_task_ref_2" pairing produced
+        # spurious 404s whenever the queue handed back the other workflow's
+        # task). We still exercise update_task_by_ref_name and update_task_sync.
 
-        # Update second task of second workflow
-        self.task_client.update_task_sync(
-            workflow_uuid, "simple_task_ref_2", "COMPLETED", "task 1 op 2nd wf"
-        )
+        # Three task executions remain (we completed one above); drain them all.
+        batchPolledTasks = self.task_client.batch_poll_tasks(TASK_TYPE)
+        remaining = 3
+        completed = 0
+        for polledTask in batchPolledTasks:
+            _retry_on_404(
+                self.task_client.update_task_by_ref_name,
+                polledTask.workflow_instance_id,
+                polledTask.reference_task_name,
+                "COMPLETED",
+                f"task op {completed + 1} (by ref name)"
+            )
+            completed += 1
+
+        while completed < remaining:
+            polledTask = self.task_client.poll_task(TASK_TYPE)
+            assert polledTask is not None, \
+                "expected a task to be available to poll while draining the queue"
+            _retry_on_404(
+                self.task_client.update_task_sync,
+                polledTask.workflow_instance_id,
+                polledTask.reference_task_name,
+                "COMPLETED",
+                f"task op {completed + 1} (sync)"
+            )
+            completed += 1
 
         queue_size = self.task_client.get_queue_size_for_task(TASK_TYPE)
         print(f'queue size for {TASK_TYPE} is {queue_size}')
diff --git a/tests/integration/client/test_async.py b/tests/integration/client/test_async.py
index 8efe4fc8..c83278a7 100644
--- a/tests/integration/client/test_async.py
+++ b/tests/integration/client/test_async.py
@@ -1,10 +1,18 @@
 from conductor.client.http.api.metadata_resource_api import MetadataResourceApi
 from conductor.client.http.api_client import ApiClient
+from conductor.client.http.models.task_def import TaskDef
+
+TASK_NAME = 'python_integration_test_task'
 
 
 def test_async_method(api_client: ApiClient):
     metadata_client = MetadataResourceApi(api_client)
+
+    # Ensure the task def exists so the async lookup has something to return,
+    # regardless of test ordering.
+    metadata_client.register_task_def(body=[TaskDef(name=TASK_NAME)])
+
     thread = metadata_client.get_task_def(
-        async_req=True, tasktype='python_integration_test_task')
+        async_req=True, tasktype=TASK_NAME)
     thread.wait()
     assert thread.get() is not None
diff --git a/tests/integration/metadata/test_workflow_definition.py b/tests/integration/metadata/test_workflow_definition.py
index f513973b..95f38827 100644
--- a/tests/integration/metadata/test_workflow_definition.py
+++ b/tests/integration/metadata/test_workflow_definition.py
@@ -1,6 +1,8 @@
+import time
 from typing import List
 
 from conductor.client.http.models import TaskDef
+from conductor.client.http.rest import ApiException
 from conductor.client.http.models.start_workflow_request import StartWorkflowRequest
 from conductor.client.workflow.conductor_workflow import ConductorWorkflow
 from conductor.client.workflow.executor.workflow_executor import WorkflowExecutor
@@ -68,7 +70,23 @@ def test_kitchensink_workflow_registration(workflow_executor: WorkflowExecutor)
     if type(workflow_id) != str or workflow_id == '':
         raise Exception(f'failed to start workflow, name: {WORKFLOW_NAME}')
 
-    workflow_executor.terminate(workflow_id=workflow_id, reason="End test")
+    _terminate_with_retry(workflow_executor, workflow_id, reason="End test")
+
+
+def _terminate_with_retry(workflow_executor: WorkflowExecutor, workflow_id: str,
+                          reason: str, retries: int = 5) -> None:
+    # Terminating a workflow that the server is still evaluating can transiently
+    # return 423 Locked ("Unable to acquire lock"). Retry with backoff so the
+    # test tolerates that race instead of failing.
+    for attempt in range(retries):
+        try:
+            workflow_executor.terminate(workflow_id=workflow_id, reason=reason)
+            return
+        except ApiException as e:
+            if e.status == 423 and attempt < retries - 1:
+                time.sleep(1 << attempt)
+                continue
+            raise
 
 
 def generate_simple_task(id: int) -> SimpleTask:
diff --git a/tests/integration/test_async_lease_extension.py b/tests/integration/test_async_lease_extension.py
index 9a73c88b..0c03fd32 100644
--- a/tests/integration/test_async_lease_extension.py
+++ b/tests/integration/test_async_lease_extension.py
@@ -24,12 +24,14 @@
 import os
 import sys
 import time
-import threading
 import unittest
+from uuid import uuid4
+
+import pytest
 
 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.worker.worker_task import worker_task
 from conductor.client.http.models.workflow_def import WorkflowDef
@@ -56,15 +58,25 @@
 # Number of fast tasks for performance comparison
 PERF_TASK_COUNT = 5
 
+# Per-run suffix so this suite's task/workflow names don't collide with other
+# runs (or other PRs/developers) on the shared dev server. With fixed names,
+# concurrent runs poll the same queues and steal/strand each other's tasks,
+# producing non-deterministic SCHEDULED/RUNNING failures.
+RUN_ID = uuid4().hex[:8]
+HEARTBEAT_TASK = f'async_lease_heartbeat_task_{RUN_ID}'
+NO_HEARTBEAT_TASK = f'async_lease_no_heartbeat_task_{RUN_ID}'
+FAST_HB_TASK = f'async_lease_fast_with_hb_{RUN_ID}'
+FAST_NO_HB_TASK = f'async_lease_fast_no_hb_{RUN_ID}'
+
 
 # -- Async Workers -----------------------------------------------------------
 
 @worker_task(
-    task_definition_name='async_lease_heartbeat_task',
+    task_definition_name=HEARTBEAT_TASK,
     lease_extend_enabled=True,
     register_task_def=True,
     task_def=TaskDef(
-        name='async_lease_heartbeat_task',
+        name=HEARTBEAT_TASK,
         response_timeout_seconds=RESPONSE_TIMEOUT_SECONDS,
         timeout_seconds=180,
         retry_count=0,
@@ -81,11 +93,11 @@ async def async_lease_heartbeat_task(job_id: str) -> dict:
 
 
 @worker_task(
-    task_definition_name='async_lease_no_heartbeat_task',
+    task_definition_name=NO_HEARTBEAT_TASK,
     lease_extend_enabled=False,
     register_task_def=True,
     task_def=TaskDef(
-        name='async_lease_no_heartbeat_task',
+        name=NO_HEARTBEAT_TASK,
         response_timeout_seconds=RESPONSE_TIMEOUT_SECONDS,
         timeout_seconds=120,
         retry_count=0,
@@ -102,11 +114,11 @@ async def async_lease_no_heartbeat_task(job_id: str) -> dict:
 
 
 @worker_task(
-    task_definition_name='async_lease_fast_with_hb',
+    task_definition_name=FAST_HB_TASK,
     lease_extend_enabled=True,
     register_task_def=True,
     task_def=TaskDef(
-        name='async_lease_fast_with_hb',
+        name=FAST_HB_TASK,
         response_timeout_seconds=60,
         timeout_seconds=120,
         retry_count=0,
@@ -120,11 +132,11 @@ async def async_lease_fast_with_hb(job_id: str) -> dict:
 
 
 @worker_task(
-    task_definition_name='async_lease_fast_no_hb',
+    task_definition_name=FAST_NO_HB_TASK,
     lease_extend_enabled=False,
     register_task_def=True,
     task_def=TaskDef(
-        name='async_lease_fast_no_hb',
+        name=FAST_NO_HB_TASK,
         response_timeout_seconds=60,
         timeout_seconds=120,
         retry_count=0,
@@ -139,8 +151,19 @@ async def async_lease_fast_no_hb(job_id: str) -> dict:
 
 # -- Test class --------------------------------------------------------------
 
+@pytest.mark.slow_async
 class TestAsyncLeaseExtension(unittest.TestCase):
 
+    # Only the workers defined in THIS module. Used to scope the TaskHandler so
+    # it doesn't spin up every @worker_task registered across the imported test
+    # suite (that was ~24 worker processes started/torn down per test).
+    WORKER_TASK_NAMES = {
+        HEARTBEAT_TASK,
+        NO_HEARTBEAT_TASK,
+        FAST_HB_TASK,
+        FAST_NO_HB_TASK,
+    }
+
     @classmethod
     def setUpClass(cls):
         from tests.integration.conftest import skip_if_server_unavailable
@@ -150,6 +173,41 @@ def setUpClass(cls):
         cls.metadata_client = OrkesMetadataClient(cls.config)
         cls.workflow_client = OrkesWorkflowClient(cls.config)
 
+        # Start workers ONCE for the whole class (not per test) and only the
+        # four workers this module needs. Both changes cut the repeated
+        # process-scan/start/stop overhead that dominated the runtime.
+        workers = [
+            w for w in get_registered_workers()
+            if w.get_task_definition_name() in cls.WORKER_TASK_NAMES
+        ]
+        cls._task_handler = TaskHandler(
+            workers=workers,
+            configuration=cls.config,
+            scan_for_annotated_workers=False,
+        )
+        cls._task_handler.start_processes()
+        time.sleep(3)  # let workers start (once, not per test)
+
+        # Confirm every worker this module needs was actually started with a
+        # live process. If one silently fails to start, its task sits in
+        # SCHEDULED forever and surfaces later as an opaque "workflow still
+        # RUNNING" assertion; fail loudly here naming the exact worker instead.
+        started = {}
+        for worker, process in zip(cls._task_handler.workers,
+                                   cls._task_handler.task_runner_processes):
+            started[worker.get_task_definition_name()] = process.is_alive()
+        logger.info("Started workers (name -> alive): %s", started)
+        missing = [n for n in cls.WORKER_TASK_NAMES
+                   if not started.get(n, False)]
+        assert not missing, (
+            f"workers not started/alive: {missing}; started={started}")
+
+    @classmethod
+    def tearDownClass(cls):
+        handler = getattr(cls, '_task_handler', None)
+        if handler is not None:
+            handler.stop_processes()
+
     def _register_workflow(self, wf_name, task_names):
         """Register a workflow with one or more tasks in sequence."""
         workflow = WorkflowDef(name=wf_name, version=1)
@@ -177,31 +235,56 @@ def _start_workflow(self, wf_name, job_id):
         logger.info("Started workflow %s: %s", wf_name, wf_id)
         return wf_id
 
+    TERMINAL_STATES = ('COMPLETED', 'FAILED', 'TIMED_OUT', 'TERMINATED')
+
     def _wait_for_workflow(self, wf_id, timeout_seconds=90):
-        """Poll until workflow reaches a terminal state."""
+        """Poll until workflow reaches a terminal state. If it doesn't within
+        the budget, dump server-side diagnostics so the ensuing assertion shows
+        *why* (e.g. a task stuck in SCHEDULED with no poller) rather than only a
+        bare status mismatch.
+        """
         for _ in range(timeout_seconds):
             wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
-            if wf.status in ('COMPLETED', 'FAILED', 'TIMED_OUT', 'TERMINATED'):
+            if wf.status in self.TERMINAL_STATES:
                 return wf
             time.sleep(1)
-        return self.workflow_client.get_workflow(wf_id, include_tasks=True)
-
-    def _run_workers_in_background(self, duration_seconds=90):
-        """Start workers in a background thread, return stop function."""
-        handler = TaskHandler(
-            configuration=self.config,
-            scan_for_annotated_workers=True,
-        )
-        handler.start_processes()
-
-        def stop():
-            handler.stop_processes()
-
-        timer = threading.Timer(duration_seconds, stop)
-        timer.daemon = True
-        timer.start()
-
-        return stop
+        wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
+        if wf.status not in self.TERMINAL_STATES:
+            print(f"  [diag] workflow {wf_id} still {wf.status} "
+                  f"after {timeout_seconds}s")
+            self._dump_workflow_diagnostics(wf)
+        return wf
+
+    def _dump_workflow_diagnostics(self, wf):
+        """Print task statuses plus server-side poll data / queue size so a
+        non-terminal workflow shows *why* (e.g. a task stuck in SCHEDULED with
+        no poller = the worker isn't consuming it), rather than only a bare
+        status assertion. Poll data is server-side, so it survives regardless
+        of worker child-process log capture.
+        """
+        from conductor.client.orkes.orkes_task_client import OrkesTaskClient
+        task_client = OrkesTaskClient(self.config)
+
+        # Worker processes can die *after* setUpClass; report current liveness
+        # so we can tell "worker crashed mid-suite" apart from "worker alive but
+        # not polling / server not timing out".
+        handler = getattr(self, '_task_handler', None)
+        if handler is not None:
+            alive = {}
+            for worker, process in zip(handler.workers, handler.task_runner_processes):
+                alive[worker.get_task_definition_name()] = process.is_alive()
+            print(f"  [diag] worker liveness: {alive}")
+
+        for task in (getattr(wf, 'tasks', None) or []):
+            print(f"  [diag] task {task.task_def_name}: {task.status}")
+            try:
+                queue_size = task_client.get_queue_size_for_task(task.task_def_name)
+                poll_data = task_client.get_task_poll_data(task.task_def_name) or []
+                pollers = [(p.worker_id, p.domain, p.last_poll_time) for p in poll_data]
+                print(f"  [diag] {task.task_def_name}: "
+                      f"queue_size={queue_size} pollers={pollers}")
+            except Exception as e:  # diagnostics must never mask the real failure
+                print(f"  [diag] {task.task_def_name}: failed to fetch poll data: {e!r}")
 
     # -- Tests ----------------------------------------------------------------
 
@@ -212,33 +295,27 @@ def test_01_async_with_heartbeat_completes(self):
         print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, task sleeps {TASK_SLEEP_SECONDS}s")
         print("=" * 80)
 
-        wf_name = 'test_async_lease_heartbeat'
-        self._register_workflow(wf_name, 'async_lease_heartbeat_task')
+        wf_name = f'test_async_lease_heartbeat_{RUN_ID}'
+        self._register_workflow(wf_name, HEARTBEAT_TASK)
 
-        stop_workers = self._run_workers_in_background(duration_seconds=90)
-        time.sleep(3)  # let workers start
+        wf_id = self._start_workflow(wf_name, 'ASYNC-HB-001')
+        wf = self._wait_for_workflow(wf_id, timeout_seconds=80)
 
-        try:
-            wf_id = self._start_workflow(wf_name, 'ASYNC-HB-001')
-            wf = self._wait_for_workflow(wf_id, timeout_seconds=80)
+        print(f"\n  Workflow ID: {wf_id}")
+        print(f"  Final status: {wf.status}")
+        for task in (wf.tasks or []):
+            print(f"  Task {task.task_def_name}: {task.status}")
 
-            print(f"\n  Workflow ID: {wf_id}")
-            print(f"  Final status: {wf.status}")
-            for task in (wf.tasks or []):
-                print(f"  Task {task.task_def_name}: {task.status}")
+        self.assertEqual(wf.status, 'COMPLETED',
+                         f"Workflow should COMPLETE with heartbeat, got {wf.status}")
 
-            self.assertEqual(wf.status, 'COMPLETED',
-                             f"Workflow should COMPLETE with heartbeat, got {wf.status}")
-
-            tasks_by_ref = {t.reference_task_name: t for t in wf.tasks}
-            task = tasks_by_ref.get('async_lease_heartbeat_task_ref')
-            self.assertIsNotNone(task)
-            self.assertEqual(task.status, 'COMPLETED')
-            self.assertEqual(task.output_data.get('job_id'), 'ASYNC-HB-001')
-            self.assertEqual(task.output_data.get('slept'), TASK_SLEEP_SECONDS)
-            print("\n  PASS: Async task completed with heartbeat keeping lease alive")
-        finally:
-            stop_workers()
+        tasks_by_ref = {t.reference_task_name: t for t in wf.tasks}
+        task = tasks_by_ref.get(f'{HEARTBEAT_TASK}_ref')
+        self.assertIsNotNone(task)
+        self.assertEqual(task.status, 'COMPLETED')
+        self.assertEqual(task.output_data.get('job_id'), 'ASYNC-HB-001')
+        self.assertEqual(task.output_data.get('slept'), TASK_SLEEP_SECONDS)
+        print("\n  PASS: Async task completed with heartbeat keeping lease alive")
 
     def test_02_async_without_heartbeat_times_out(self):
         """Async task WITHOUT lease_extend_enabled times out when sleep > responseTimeout."""
@@ -247,32 +324,26 @@ def test_02_async_without_heartbeat_times_out(self):
         print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, task sleeps {TASK_SLEEP_SECONDS}s")
         print("=" * 80)
 
-        wf_name = 'test_async_lease_no_heartbeat'
-        self._register_workflow(wf_name, 'async_lease_no_heartbeat_task')
+        wf_name = f'test_async_lease_no_heartbeat_{RUN_ID}'
+        self._register_workflow(wf_name, NO_HEARTBEAT_TASK)
 
-        stop_workers = self._run_workers_in_background(duration_seconds=90)
-        time.sleep(3)
+        wf_id = self._start_workflow(wf_name, 'ASYNC-NOHB-001')
+        wf = self._wait_for_workflow(wf_id, timeout_seconds=80)
 
-        try:
-            wf_id = self._start_workflow(wf_name, 'ASYNC-NOHB-001')
-            wf = self._wait_for_workflow(wf_id, timeout_seconds=80)
-
-            print(f"\n  Workflow ID: {wf_id}")
-            print(f"  Final status: {wf.status}")
-            for task in (wf.tasks or []):
-                print(f"  Task {task.task_def_name}: {task.status}")
-
-            self.assertIn(wf.status, ('FAILED', 'TIMED_OUT'),
-                          f"Workflow should FAIL/TIMEOUT without heartbeat, got {wf.status}")
-
-            tasks_by_ref = {t.reference_task_name: t for t in wf.tasks}
-            task = tasks_by_ref.get('async_lease_no_heartbeat_task_ref')
-            self.assertIsNotNone(task)
-            self.assertIn(task.status, ('TIMED_OUT', 'FAILED', 'CANCELED'),
-                          f"Task should be TIMED_OUT/FAILED, got {task.status}")
-            print("\n  PASS: Async task timed out as expected without heartbeat")
-        finally:
-            stop_workers()
+        print(f"\n  Workflow ID: {wf_id}")
+        print(f"  Final status: {wf.status}")
+        for task in (wf.tasks or []):
+            print(f"  Task {task.task_def_name}: {task.status}")
+
+        self.assertIn(wf.status, ('FAILED', 'TIMED_OUT'),
+                      f"Workflow should FAIL/TIMEOUT without heartbeat, got {wf.status}")
+
+        tasks_by_ref = {t.reference_task_name: t for t in wf.tasks}
+        task = tasks_by_ref.get(f'{NO_HEARTBEAT_TASK}_ref')
+        self.assertIsNotNone(task)
+        self.assertIn(task.status, ('TIMED_OUT', 'FAILED', 'CANCELED'),
+                      f"Task should be TIMED_OUT/FAILED, got {task.status}")
+        print("\n  PASS: Async task timed out as expected without heartbeat")
 
     def test_03_no_performance_overhead(self):
         """Heartbeat tracking adds no meaningful overhead to fast async tasks."""
@@ -281,63 +352,57 @@ def test_03_no_performance_overhead(self):
         print(f"  Running {PERF_TASK_COUNT} tasks each, sleep={FAST_TASK_SLEEP_SECONDS}s")
         print("=" * 80)
 
-        wf_with_hb = 'test_async_perf_with_hb'
-        wf_no_hb = 'test_async_perf_no_hb'
-        self._register_workflow(wf_with_hb, 'async_lease_fast_with_hb')
-        self._register_workflow(wf_no_hb, 'async_lease_fast_no_hb')
-
-        stop_workers = self._run_workers_in_background(duration_seconds=120)
-        time.sleep(3)
-
-        try:
-            # Run tasks WITH heartbeat tracking
-            hb_workflow_ids = []
-            for i in range(PERF_TASK_COUNT):
-                wf_id = self._start_workflow(wf_with_hb, f'PERF-HB-{i:03d}')
-                hb_workflow_ids.append(wf_id)
-
-            # Run tasks WITHOUT heartbeat tracking
-            no_hb_workflow_ids = []
-            for i in range(PERF_TASK_COUNT):
-                wf_id = self._start_workflow(wf_no_hb, f'PERF-NOHB-{i:03d}')
-                no_hb_workflow_ids.append(wf_id)
-
-            # Wait for all to complete
-            hb_times = []
-            for wf_id in hb_workflow_ids:
-                wf = self._wait_for_workflow(wf_id, timeout_seconds=30)
-                self.assertEqual(wf.status, 'COMPLETED',
-                                 f"Fast HB task should complete, got {wf.status}")
-                task = wf.tasks[0]
-                duration_ms = task.end_time - task.start_time
-                hb_times.append(duration_ms)
-
-            no_hb_times = []
-            for wf_id in no_hb_workflow_ids:
-                wf = self._wait_for_workflow(wf_id, timeout_seconds=30)
-                self.assertEqual(wf.status, 'COMPLETED',
-                                 f"Fast no-HB task should complete, got {wf.status}")
-                task = wf.tasks[0]
-                duration_ms = task.end_time - task.start_time
-                no_hb_times.append(duration_ms)
-
-            avg_hb = sum(hb_times) / len(hb_times)
-            avg_no_hb = sum(no_hb_times) / len(no_hb_times)
-            overhead_ms = avg_hb - avg_no_hb
-            overhead_pct = (overhead_ms / avg_no_hb * 100) if avg_no_hb > 0 else 0
-
-            print(f"\n  With heartbeat:    avg {avg_hb:.0f}ms  {hb_times}")
-            print(f"  Without heartbeat: avg {avg_no_hb:.0f}ms  {no_hb_times}")
-            print(f"  Overhead:          {overhead_ms:+.0f}ms ({overhead_pct:+.1f}%)")
-
-            # Heartbeat tracking should add < 500ms overhead per task
-            # (LeaseManager.track is just a dict insert + set add)
-            self.assertLess(overhead_ms, 500,
-                            f"Heartbeat overhead too high: {overhead_ms:.0f}ms")
-
-            print("\n  PASS: No meaningful performance overhead from heartbeat tracking")
-        finally:
-            stop_workers()
+        wf_with_hb = f'test_async_perf_with_hb_{RUN_ID}'
+        wf_no_hb = f'test_async_perf_no_hb_{RUN_ID}'
+        self._register_workflow(wf_with_hb, FAST_HB_TASK)
+        self._register_workflow(wf_no_hb, FAST_NO_HB_TASK)
+
+        # Run tasks WITH heartbeat tracking
+        hb_workflow_ids = []
+        for i in range(PERF_TASK_COUNT):
+            wf_id = self._start_workflow(wf_with_hb, f'PERF-HB-{i:03d}')
+            hb_workflow_ids.append(wf_id)
+
+        # Run tasks WITHOUT heartbeat tracking
+        no_hb_workflow_ids = []
+        for i in range(PERF_TASK_COUNT):
+            wf_id = self._start_workflow(wf_no_hb, f'PERF-NOHB-{i:03d}')
+            no_hb_workflow_ids.append(wf_id)
+
+        # Wait for all to complete
+        hb_times = []
+        for wf_id in hb_workflow_ids:
+            wf = self._wait_for_workflow(wf_id, timeout_seconds=30)
+            self.assertEqual(wf.status, 'COMPLETED',
+                             f"Fast HB task should complete, got {wf.status}")
+            task = wf.tasks[0]
+            duration_ms = task.end_time - task.start_time
+            hb_times.append(duration_ms)
+
+        no_hb_times = []
+        for wf_id in no_hb_workflow_ids:
+            wf = self._wait_for_workflow(wf_id, timeout_seconds=30)
+            self.assertEqual(wf.status, 'COMPLETED',
+                             f"Fast no-HB task should complete, got {wf.status}")
+            task = wf.tasks[0]
+            duration_ms = task.end_time - task.start_time
+            no_hb_times.append(duration_ms)
+
+        avg_hb = sum(hb_times) / len(hb_times)
+        avg_no_hb = sum(no_hb_times) / len(no_hb_times)
+        overhead_ms = avg_hb - avg_no_hb
+        overhead_pct = (overhead_ms / avg_no_hb * 100) if avg_no_hb > 0 else 0
+
+        print(f"\n  With heartbeat:    avg {avg_hb:.0f}ms  {hb_times}")
+        print(f"  Without heartbeat: avg {avg_no_hb:.0f}ms  {no_hb_times}")
+        print(f"  Overhead:          {overhead_ms:+.0f}ms ({overhead_pct:+.1f}%)")
+
+        # Heartbeat tracking should add < 500ms overhead per task
+        # (LeaseManager.track is just a dict insert + set add)
+        self.assertLess(overhead_ms, 500,
+                        f"Heartbeat overhead too high: {overhead_ms:.0f}ms")
+
+        print("\n  PASS: No meaningful performance overhead from heartbeat tracking")
 
 
 if __name__ == '__main__':
diff --git a/tests/integration/test_authorization_client_intg.py b/tests/integration/test_authorization_client_intg.py
index 8e9146a6..da0569c6 100644
--- a/tests/integration/test_authorization_client_intg.py
+++ b/tests/integration/test_authorization_client_intg.py
@@ -16,6 +16,7 @@
 from conductor.client.http.models.upsert_user_request import UpsertUserRequest
 from conductor.client.orkes.models.access_type import AccessType
 from conductor.client.orkes.models.metadata_tag import MetadataTag
+from conductor.client.http.rest import ApiException
 from conductor.client.orkes.orkes_authorization_client import OrkesAuthorizationClient
 
 logger = logging.getLogger(
@@ -50,7 +51,7 @@ def setUpClass(cls):
         cls.test_app_name = f"test_app_{cls.timestamp}"
         cls.test_user_id = f"test_user_{cls.timestamp}@example.com"
         cls.test_group_id = f"test_group_{cls.timestamp}"
-        cls.test_role_name = f"test_role_{cls.timestamp}"
+        cls.test_role_name = f"TEST_ROLE_{cls.timestamp}"
         cls.test_gateway_config_id = None
 
         # Store created resource IDs for cleanup
@@ -343,6 +344,7 @@ def test_19_upsert_group(self):
 
         request = UpsertGroupRequest()
         request.description = "Test Group"
+        request.roles = []
 
         group = self.client.upsert_group(request, self.test_group_id)
 
@@ -530,7 +532,9 @@ def test_38_create_role(self):
 
         request = CreateOrUpdateRoleRequest()
         request.name = self.test_role_name
-        request.permissions = ["workflow:read"]
+        # Permission strings must be in the server's ACCESS_RESOURCE form
+        # (e.g. READ_WORKFLOW_DEF); "workflow:read" is rejected as invalid.
+        request.permissions = ["READ_WORKFLOW_DEF"]
 
         result = self.client.create_role(request)
 
@@ -551,7 +555,8 @@ def test_40_update_role(self):
 
         request = CreateOrUpdateRoleRequest()
         request.name = self.test_role_name
-        request.permissions = ["workflow:read", "workflow:execute"]
+        # Valid ACCESS_RESOURCE permission strings (see test_38_create_role).
+        request.permissions = ["READ_WORKFLOW_DEF", "EXECUTE_WORKFLOW_DEF"]
 
         result = self.client.update_role(self.test_role_name, request)
 
@@ -585,7 +590,12 @@ def test_42_list_gateway_auth_configs(self):
         """Test: list_gateway_auth_configs"""
         logger.info('TEST: list_gateway_auth_configs')
 
-        configs = self.client.list_gateway_auth_configs()
+        try:
+            configs = self.client.list_gateway_auth_configs()
+        except ApiException as e:
+            if e.status == 404:
+                self.skipTest('API Gateway auth-config endpoint not available on this server')
+            raise
 
         self.assertIsNotNone(configs)
         self.assertIsInstance(configs, list)
diff --git a/tests/integration/test_authorization_complete.py b/tests/integration/test_authorization_complete.py
index 2ba4bcf0..1ede72ff 100644
--- a/tests/integration/test_authorization_complete.py
+++ b/tests/integration/test_authorization_complete.py
@@ -27,7 +27,7 @@
 from conductor.client.http.models.authentication_config import AuthenticationConfig
 from conductor.client.orkes.models.access_type import AccessType
 from conductor.client.orkes.models.metadata_tag import MetadataTag
-from conductor.client.http.rest import RestException
+from conductor.client.http.rest import ApiException, RestException
 
 
 @pytest.fixture(scope="module")
@@ -239,7 +239,7 @@ def test_access_key_lifecycle(self, auth_client, test_run_id, cleanup_tracker):
 
         # Get app by access key (Method 6)
         found_app = auth_client.get_app_by_access_key_id(created_key.id)
-        assert found_app == app.id
+        assert found_app['id'] == app.id
 
         # Delete access key (Method 15) - handled in cleanup
 
@@ -292,7 +292,9 @@ def test_user_lifecycle(self, auth_client, test_run_id, cleanup_tracker):
             target_type="WORKFLOW_DEF",
             target_id="test-workflow"
         )
-        assert isinstance(result, bool)
+        # The server returns a per-access permission map
+        # (e.g. {'CREATE': True, 'READ': True, ...}), not a single bool.
+        assert isinstance(result, dict)
 
 
 class TestGroupManagement:
@@ -469,7 +471,12 @@ def test_gateway_auth_config(self, auth_client, test_run_id, cleanup_tracker):
         auth_config.api_keys = ["test-key"]
         auth_config.fallback_to_default_auth = False
 
-        created = auth_client.create_gateway_auth_config(auth_config)
+        try:
+            created = auth_client.create_gateway_auth_config(auth_config)
+        except ApiException as e:
+            if e.status == 404:
+                pytest.skip('API Gateway auth-config endpoint not available on this server')
+            raise
         cleanup_tracker['auth_configs'].append(config_id)
 
         assert created.get('id') == config_id
diff --git a/tests/integration/test_comprehensive_e2e.py b/tests/integration/test_comprehensive_e2e.py
index f300b8c9..55c2f355 100644
--- a/tests/integration/test_comprehensive_e2e.py
+++ b/tests/integration/test_comprehensive_e2e.py
@@ -28,11 +28,11 @@
 import os
 import sys
 import time
-import threading
 import unittest
 from dataclasses import dataclass
 from typing import Optional, List, Dict, Union
 from collections import defaultdict
+from uuid import uuid4
 
 from conductor.client.worker.exception import NonRetryableException
 
@@ -71,15 +71,28 @@ def on_task_execution_failure(self, e): self.events['exec_failed'].append(e)
     def on_task_update_failure(self, e): self.events['update_failed'].append(e)
 
 
+# Per-run suffix so this suite's task/workflow names don't collide with other
+# runs (or other PRs/developers) on the shared dev server. With fixed names,
+# concurrent runs poll the same queues and steal/strand each other's tasks,
+# producing non-deterministic SCHEDULED failures.
+RUN_ID = uuid4().hex[:8]
+SYNC_BASIC = f'sync_basic_{RUN_ID}'
+ASYNC_BASIC = f'async_basic_{RUN_ID}'
+COMPLEX_SCHEMA = f'complex_schema_{RUN_ID}'
+TASK_IN_PROGRESS = f'task_in_progress_{RUN_ID}'
+FAILING_TASK = f'failing_task_{RUN_ID}'
+WF_NAME = f'e2e_comprehensive_test_{RUN_ID}'
+
+
 # Test workers covering all scenarios
-@worker_task(task_definition_name='sync_basic', thread_count=5, register_task_def=True)
+@worker_task(task_definition_name=SYNC_BASIC, thread_count=5, register_task_def=True)
 def sync_basic(value: str, count: int) -> dict:
     ctx = get_task_context()
     ctx.add_log(f"Processing {value}")
     return {'value': value, 'count': count, 'worker': 'sync'}
 
 
-@worker_task(task_definition_name='async_basic', thread_count=10, register_task_def=True)
+@worker_task(task_definition_name=ASYNC_BASIC, thread_count=10, register_task_def=True)
 async def async_basic(message: str) -> dict:
     await asyncio.sleep(0.1)
     return {'message': message, 'worker': 'async'}
@@ -93,17 +106,17 @@ class OrderData:
 
 
 @worker_task(
-    task_definition_name='complex_schema',
+    task_definition_name=COMPLEX_SCHEMA,
     register_task_def=True,
     strict_schema=True,
-    task_def=TaskDef(name='complex_schema', retry_count=2, timeout_policy='RETRY')
+    task_def=TaskDef(name=COMPLEX_SCHEMA, retry_count=2, timeout_policy='RETRY')
 )
 def complex_schema(data: OrderData, optional: Optional[str]) -> dict:
     assert data.id is not None
     return {'id': data.id, 'amount': data.amount, 'tag_count': len(data.tags)}
 
 
-@worker_task(task_definition_name='task_in_progress')
+@worker_task(task_definition_name=TASK_IN_PROGRESS)
 def task_in_progress(job_id: str) -> Union[dict, TaskInProgress]:
     ctx = get_task_context()
     polls = ctx.get_poll_count()
@@ -112,7 +125,7 @@ def task_in_progress(job_id: str) -> Union[dict, TaskInProgress]:
     return {'job_id': job_id, 'polls': polls}
 
 
-@worker_task(task_definition_name='failing_task')
+@worker_task(task_definition_name=FAILING_TASK)
 def failing_task(should_fail: bool) -> dict:
     if should_fail:
         raise NonRetryableException("Test failure")
@@ -122,6 +135,18 @@ def failing_task(should_fail: bool) -> dict:
 # Main test class
 class TestComprehensiveE2E(unittest.TestCase):
 
+    # Annotated workers this suite relies on. Every one must be discovered by
+    # the scan AND have a live process, otherwise its task silently stays
+    # SCHEDULED forever (which is exactly how a non-starting task_in_progress
+    # worker manifested as a "4 != 5 tasks" failure in CI).
+    EXPECTED_WORKERS = (
+        SYNC_BASIC,
+        ASYNC_BASIC,
+        COMPLEX_SCHEMA,
+        TASK_IN_PROGRESS,
+        FAILING_TASK,
+    )
+
     @classmethod
     def setUpClass(cls):
         from tests.integration.conftest import skip_if_server_unavailable
@@ -144,6 +169,7 @@ def setUpClass(cls):
         )
 
         cls.workers_started = False
+        cls.task_handler = None
 
     def test_01_create_workflow(self):
         """Create test workflow."""
@@ -151,18 +177,18 @@ def test_01_create_workflow(self):
         
         metadata_client = OrkesMetadataClient(self.config)
         
-        workflow = WorkflowDef(name='e2e_comprehensive_test', version=1)
+        workflow = WorkflowDef(name=WF_NAME, version=1)
         tasks = [
-            WorkflowTask(name='sync_basic', task_reference_name='sync_1', 
+            WorkflowTask(name=SYNC_BASIC, task_reference_name='sync_1', 
                         input_parameters={'value': 'test', 'count': 1}),
-            WorkflowTask(name='async_basic', task_reference_name='async_1',
+            WorkflowTask(name=ASYNC_BASIC, task_reference_name='async_1',
                         input_parameters={'message': 'hello'}),
-            WorkflowTask(name='complex_schema', task_reference_name='complex_1',
+            WorkflowTask(name=COMPLEX_SCHEMA, task_reference_name='complex_1',
                         input_parameters={'data': {'id': '123', 'amount': 99.99, 'tags': ['a', 'b']}, 
                                         'optional': None}),
-            WorkflowTask(name='task_in_progress', task_reference_name='tip_1',
+            WorkflowTask(name=TASK_IN_PROGRESS, task_reference_name='tip_1',
                         input_parameters={'job_id': 'JOB1'}),
-            WorkflowTask(name='failing_task', task_reference_name='fail_1',
+            WorkflowTask(name=FAILING_TASK, task_reference_name='fail_1',
                         input_parameters={'should_fail': True}),
         ]
         workflow._tasks = tasks
@@ -174,56 +200,131 @@ def test_01_create_workflow(self):
     def test_02_start_workers(self):
         """Start workers and verify they initialize."""
         print("\n" + "="*80 + "\nTEST 2: Start Workers\n" + "="*80)
-        
-        def run_workers():
-            with TaskHandler(
-                configuration=self.config,
-                metrics_settings=self.metrics_settings,
-                scan_for_annotated_workers=True,
-                event_listeners=[self.event_collector]
-            ) as handler:
-                handler.start_processes()
-                time.sleep(90)  # Run for test duration
-                handler.stop_processes()
-        
-        thread = threading.Thread(target=run_workers, daemon=True)
-        thread.start()
+
+        # Start the workers and keep the handler on the class so it stays alive
+        # for the remaining tests and is stopped deterministically in
+        # tearDownClass. (A previous version ran the handler in a daemon thread
+        # with a fixed 90s sleep; if the suite finished sooner, the worker
+        # processes were never stopped and the interpreter hung at exit joining
+        # them.)
+        handler = TaskHandler(
+            configuration=self.config,
+            metrics_settings=self.metrics_settings,
+            scan_for_annotated_workers=True,
+            event_listeners=[self.event_collector]
+        )
+        handler.start_processes()
+        self.__class__.task_handler = handler
         time.sleep(5)  # Wait for startup
-        
+
+        # Confirm every expected annotated worker was discovered by the scan AND
+        # its child process is actually alive. Assert each one individually so a
+        # failure names the exact worker that didn't come up, instead of the
+        # failure surfacing later (and opaquely) as a task stuck in SCHEDULED.
+        started = {}
+        for worker, process in zip(handler.workers, handler.task_runner_processes):
+            started[worker.get_task_definition_name()] = process.is_alive()
+        print(f"Discovered {len(started)} annotated worker(s): {sorted(started)}")
+
+        for name in self.EXPECTED_WORKERS:
+            self.assertIn(
+                name, started,
+                f"worker '{name}' was not discovered by the annotated-worker scan; "
+                f"discovered={sorted(started)}")
+            self.assertTrue(
+                started[name],
+                f"worker '{name}' was discovered but its process is not alive")
+
         print("✓ Workers started")
         self.__class__.workers_started = True
         self.assertTrue(self.workers_started)
 
-    def test_03_execute_workflow(self):
-        """Execute workflow and verify completion."""
-        print("\n" + "="*80 + "\nTEST 3: Execute Workflow\n" + "="*80)
-        
-        self.assertTrue(self.workers_started, "Workers must be started first")
-        
-        workflow_client = OrkesWorkflowClient(self.config)
+    EXPECTED_TASK_COUNT = 5
+
+    def _dump_stuck_task_diagnostics(self, wf):
+        """When a run fails to complete, print server-side poll data and queue
+        sizes so CI shows *why* a task is stuck (e.g. no worker polling its
+        queue) rather than just a bare task-count assertion. Poll data is
+        server-side, so it survives regardless of worker child-process log
+        capture.
+        """
+        from conductor.client.orkes.orkes_task_client import OrkesTaskClient
+        task_client = OrkesTaskClient(self.config)
+
+        stuck = [t.task_def_name for t in (getattr(wf, 'tasks', None) or [])
+                 if t.status not in ('COMPLETED', 'FAILED', 'FAILED_WITH_TERMINAL_ERROR')]
+        print(f"  [diag] non-terminal tasks: {stuck}")
+
+        # Include the expected workers so we can compare a stuck queue against a
+        # known-good one (e.g. task_in_progress vs sync_basic).
+        for task_type in sorted(set(stuck) | set(self.EXPECTED_WORKERS)):
+            try:
+                queue_size = task_client.get_queue_size_for_task(task_type)
+                poll_data = task_client.get_task_poll_data(task_type) or []
+                pollers = [(p.worker_id, p.domain, p.last_poll_time) for p in poll_data]
+                print(f"  [diag] {task_type}: queue_size={queue_size} pollers={pollers}")
+            except Exception as e:  # diagnostics must never mask the real failure
+                print(f"  [diag] {task_type}: failed to fetch poll data: {e!r}")
+
+    def _run_workflow_to_terminal(self, workflow_client, timeout_s=90):
+        """Start the e2e workflow and wait until it is genuinely terminal with
+        all expected tasks present. Returns (wf_id, workflow_or_None). A None
+        workflow (or a non-terminal / short task list) signals a timed-out run
+        the caller can retry, rather than asserting against a half-scheduled
+        workflow (which is what produced the flaky "4 != 5 tasks" failure).
+        """
         req = StartWorkflowRequest()
-        req.name = 'e2e_comprehensive_test'
+        req.name = WF_NAME
         req.version = 1
         req.input = {}
-        
+
         wf_id = workflow_client.start_workflow(start_workflow_request=req)
         print(f"✓ Started workflow: {wf_id}")
-        
-        # Wait for completion
-        for i in range(30):
+
+        deadline = time.time() + timeout_s
+        wf = None
+        while time.time() < deadline:
             wf = workflow_client.get_workflow(wf_id, include_tasks=True)
-            print(f"  Status: {wf.status} - ({i*2}s)")
-            if wf.status in ['COMPLETED', 'FAILED']:
-                break
-            for task in wf.tasks:
+            print(f"  Status: {wf.status} - tasks={len(wf.tasks or [])}")
+            if wf.status in ('COMPLETED', 'FAILED') \
+                    and len(wf.tasks or []) == self.EXPECTED_TASK_COUNT:
+                return wf_id, wf
+            for task in (wf.tasks or []):
                 print(f'task {task.task_def_name} : {task.status}')
             time.sleep(1)
+
+        return wf_id, wf
+
+    def test_03_execute_workflow(self):
+        """Execute workflow and verify completion."""
+        print("\n" + "="*80 + "\nTEST 3: Execute Workflow\n" + "="*80)
         
-        final_wf = workflow_client.get_workflow(wf_id, include_tasks=True)
+        self.assertTrue(self.workers_started, "Workers must be started first")
         
+        workflow_client = OrkesWorkflowClient(self.config)
+
+        # Each attempt starts a fresh workflow, so retrying a pathologically
+        # slow run (cold workers / slow CI) is safe and self-contained. This
+        # keeps a one-off timeout from forcing a manual CI job re-run.
+        final_wf = None
+        for attempt in range(3):
+            wf_id, wf = self._run_workflow_to_terminal(workflow_client, timeout_s=90)
+            if wf is not None and wf.status in ('COMPLETED', 'FAILED') \
+                    and len(wf.tasks or []) == self.EXPECTED_TASK_COUNT:
+                final_wf = wf
+                break
+            print(f"attempt {attempt + 1}: wf={wf_id} "
+                  f"status={getattr(wf, 'status', None)} "
+                  f"tasks={len(getattr(wf, 'tasks', []) or [])}; retrying")
+            self._dump_stuck_task_diagnostics(wf)
+
         # Assertions
-        self.assertIsNotNone(final_wf)
-        self.assertEqual(len(final_wf.tasks), 5, "Should have 5 tasks")
+        self.assertIsNotNone(
+            final_wf,
+            "workflow never reached a terminal state with all "
+            f"{self.EXPECTED_TASK_COUNT} tasks after retries")
+        self.assertEqual(len(final_wf.tasks), self.EXPECTED_TASK_COUNT,
+                         f"Should have {self.EXPECTED_TASK_COUNT} tasks")
         
         # Verify each task
         tasks_by_ref = {t.reference_task_name: t for t in final_wf.tasks}
@@ -288,7 +389,7 @@ def test_05_verify_task_definitions(self):
         
         metadata_client = OrkesMetadataClient(self.config)
         
-        tasks_to_check = ['sync_basic', 'async_basic', 'complex_schema']
+        tasks_to_check = [SYNC_BASIC, ASYNC_BASIC, COMPLEX_SCHEMA]
         
         for task_name in tasks_to_check:
             task_def = metadata_client.get_task_def(task_name)
@@ -297,7 +398,7 @@ def test_05_verify_task_definitions(self):
             print(f"✓ Task definition exists: {task_name}")
             
             # Check schemas if they should exist
-            if task_name in ['sync_basic', 'async_basic', 'complex_schema']:
+            if task_name in tasks_to_check:
                 # These have type hints, should have schemas
                 if hasattr(task_def, 'input_schema') and task_def.input_schema:
                     print(f"  ✓ Has input schema")
@@ -305,7 +406,7 @@ def test_05_verify_task_definitions(self):
                     print(f"  ✓ Has output schema")
 
         # Check complex_schema has TaskDef configuration
-        complex_def = metadata_client.get_task_def('complex_schema')
+        complex_def = metadata_client.get_task_def(COMPLEX_SCHEMA)
         self.assertEqual(complex_def.retry_count, 2, "Retry count from task_def should be applied")
         self.assertEqual(complex_def.timeout_policy, 'RETRY', "Timeout policy should be set")
         
@@ -422,6 +523,10 @@ def test_08_summary_assertions(self):
 
     @classmethod
     def tearDownClass(cls):
+        handler = getattr(cls, 'task_handler', None)
+        if handler is not None:
+            handler.stop_processes()
+            cls.task_handler = None
         if os.path.exists(cls.metrics_dir):
             import shutil
             shutil.rmtree(cls.metrics_dir)
diff --git a/tests/integration/test_lease_extension.py b/tests/integration/test_lease_extension.py
index 0bb72a56..ba15d468 100644
--- a/tests/integration/test_lease_extension.py
+++ b/tests/integration/test_lease_extension.py
@@ -23,6 +23,9 @@
 import time
 import threading
 import unittest
+from uuid import uuid4
+
+import pytest
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
 
@@ -47,16 +50,24 @@
 # ~30s) catches the expired task BEFORE the worker completes.
 TASK_SLEEP_SECONDS = 50
 
+# Per-run suffix so this suite's task/workflow names don't collide with other
+# runs (or other PRs/developers) on the shared dev server. With fixed names,
+# concurrent runs poll the same queues and steal/strand each other's tasks,
+# producing non-deterministic SCHEDULED/RUNNING failures.
+RUN_ID = uuid4().hex[:8]
+HEARTBEAT_TASK = f'lease_heartbeat_task_{RUN_ID}'
+NO_HEARTBEAT_TASK = f'lease_no_heartbeat_task_{RUN_ID}'
+
 
 # -- Workers -----------------------------------------------------------------
 
 # Worker WITH lease extension enabled — heartbeats keep it alive
 @worker_task(
-    task_definition_name='lease_heartbeat_task',
+    task_definition_name=HEARTBEAT_TASK,
     lease_extend_enabled=True,
     register_task_def=True,
     task_def=TaskDef(
-        name='lease_heartbeat_task',
+        name=HEARTBEAT_TASK,
         response_timeout_seconds=RESPONSE_TIMEOUT_SECONDS,
         timeout_seconds=180,
         retry_count=0,
@@ -74,11 +85,11 @@ def lease_heartbeat_task(job_id: str) -> dict:
 
 # Worker WITHOUT lease extension — will time out
 @worker_task(
-    task_definition_name='lease_no_heartbeat_task',
+    task_definition_name=NO_HEARTBEAT_TASK,
     lease_extend_enabled=False,
     register_task_def=True,
     task_def=TaskDef(
-        name='lease_no_heartbeat_task',
+        name=NO_HEARTBEAT_TASK,
         response_timeout_seconds=RESPONSE_TIMEOUT_SECONDS,
         timeout_seconds=120,
         retry_count=0,
@@ -96,6 +107,7 @@ def lease_no_heartbeat_task(job_id: str) -> dict:
 
 # -- Test class --------------------------------------------------------------
 
+@pytest.mark.slow_sync
 class TestLeaseExtension(unittest.TestCase):
 
     @classmethod
@@ -132,15 +144,52 @@ def _start_workflow(self, wf_name, job_id):
         logger.info("Started workflow %s: %s", wf_name, wf_id)
         return wf_id
 
+    TERMINAL_STATES = ('COMPLETED', 'FAILED', 'TIMED_OUT', 'TERMINATED')
+
     def _wait_for_workflow(self, wf_id, timeout_seconds=60):
-        """Poll until workflow reaches a terminal state."""
+        """Poll until workflow reaches a terminal state. If it doesn't within
+        the budget, dump server-side diagnostics so the ensuing assertion shows
+        *why* (e.g. a task stuck in SCHEDULED with no poller) rather than only a
+        bare status mismatch.
+        """
         for i in range(timeout_seconds):
             wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
-            if wf.status in ('COMPLETED', 'FAILED', 'TIMED_OUT', 'TERMINATED'):
+            if wf.status in self.TERMINAL_STATES:
                 return wf
             time.sleep(1)
-        # Return whatever state it's in after timeout
-        return self.workflow_client.get_workflow(wf_id, include_tasks=True)
+        wf = self.workflow_client.get_workflow(wf_id, include_tasks=True)
+        if wf.status not in self.TERMINAL_STATES:
+            print(f"  [diag] workflow {wf_id} still {wf.status} "
+                  f"after {timeout_seconds}s")
+            self._dump_workflow_diagnostics(wf)
+        return wf
+
+    def _dump_workflow_diagnostics(self, wf):
+        """Print task statuses plus server-side poll data / queue size so a
+        non-terminal workflow shows *why* (e.g. a task stuck in SCHEDULED with
+        no poller = the worker isn't consuming it). Poll data is server-side,
+        so it survives regardless of worker child-process log capture.
+        """
+        from conductor.client.orkes.orkes_task_client import OrkesTaskClient
+        task_client = OrkesTaskClient(self.config)
+
+        handler = getattr(self, '_active_handler', None)
+        if handler is not None:
+            alive = {}
+            for worker, process in zip(handler.workers, handler.task_runner_processes):
+                alive[worker.get_task_definition_name()] = process.is_alive()
+            print(f"  [diag] worker liveness: {alive}")
+
+        for task in (getattr(wf, 'tasks', None) or []):
+            print(f"  [diag] task {task.task_def_name}: {task.status}")
+            try:
+                queue_size = task_client.get_queue_size_for_task(task.task_def_name)
+                poll_data = task_client.get_task_poll_data(task.task_def_name) or []
+                pollers = [(p.worker_id, p.domain, p.last_poll_time) for p in poll_data]
+                print(f"  [diag] {task.task_def_name}: "
+                      f"queue_size={queue_size} pollers={pollers}")
+            except Exception as e:  # diagnostics must never mask the real failure
+                print(f"  [diag] {task.task_def_name}: failed to fetch poll data: {e!r}")
 
     def _run_workers_in_background(self, duration_seconds=60):
         """Start workers in a background thread, return stop function."""
@@ -149,6 +198,8 @@ def _run_workers_in_background(self, duration_seconds=60):
             scan_for_annotated_workers=True,
         )
         handler.start_processes()
+        # Expose the handler so _dump_workflow_diagnostics can report liveness.
+        self.__class__._active_handler = handler
 
         def stop():
             handler.stop_processes()
@@ -167,8 +218,8 @@ def test_01_with_heartbeat_completes(self):
         print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, task sleeps {TASK_SLEEP_SECONDS}s")
         print("=" * 80)
 
-        wf_name = 'test_lease_heartbeat'
-        self._register_workflow(wf_name, 'lease_heartbeat_task')
+        wf_name = f'test_lease_heartbeat_{RUN_ID}'
+        self._register_workflow(wf_name, HEARTBEAT_TASK)
 
         stop_workers = self._run_workers_in_background(duration_seconds=90)
         time.sleep(3)  # let workers start
@@ -186,7 +237,7 @@ def test_01_with_heartbeat_completes(self):
 
             # Verify task output
             tasks_by_ref = {t.reference_task_name: t for t in wf.tasks}
-            task = tasks_by_ref.get('lease_heartbeat_task_ref')
+            task = tasks_by_ref.get(f'{HEARTBEAT_TASK}_ref')
             self.assertIsNotNone(task)
             self.assertEqual(task.status, 'COMPLETED')
             self.assertEqual(task.output_data.get('job_id'), 'HEARTBEAT-001')
@@ -202,8 +253,8 @@ def test_02_without_heartbeat_times_out(self):
         print(f"  responseTimeoutSeconds={RESPONSE_TIMEOUT_SECONDS}s, task sleeps {TASK_SLEEP_SECONDS}s")
         print("=" * 80)
 
-        wf_name = 'test_lease_no_heartbeat'
-        self._register_workflow(wf_name, 'lease_no_heartbeat_task')
+        wf_name = f'test_lease_no_heartbeat_{RUN_ID}'
+        self._register_workflow(wf_name, NO_HEARTBEAT_TASK)
 
         stop_workers = self._run_workers_in_background(duration_seconds=90)
         time.sleep(3)  # let workers start
@@ -221,7 +272,7 @@ def test_02_without_heartbeat_times_out(self):
                           f"Workflow should FAIL/TIMEOUT without heartbeat, got {wf.status}")
 
             tasks_by_ref = {t.reference_task_name: t for t in wf.tasks}
-            task = tasks_by_ref.get('lease_no_heartbeat_task_ref')
+            task = tasks_by_ref.get(f'{NO_HEARTBEAT_TASK}_ref')
             self.assertIsNotNone(task)
             self.assertIn(task.status, ('TIMED_OUT', 'FAILED', 'CANCELED'),
                           f"Task should be TIMED_OUT/FAILED, got {task.status}")
diff --git a/tests/integration/test_workflow_client_intg.py b/tests/integration/test_workflow_client_intg.py
index d38ca080..960c16f3 100644
--- a/tests/integration/test_workflow_client_intg.py
+++ b/tests/integration/test_workflow_client_intg.py
@@ -1,10 +1,14 @@
 import logging
 import os
+import time
 import unittest
 
+import pytest
+
 from tests.integration.client.orkes.test_orkes_clients import TestOrkesClients
 from conductor.client.configuration.configuration import Configuration
 from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
+from conductor.client.http.rest import ApiException
 from conductor.client.orkes.orkes_workflow_client import OrkesWorkflowClient
 from conductor.client.workflow.executor.workflow_executor import WorkflowExecutor
 from tests.integration.metadata.test_workflow_definition import run_workflow_definition_tests
@@ -29,6 +33,31 @@ def get_configuration():
     return configuration
 
 
+def _run_tolerating_transient(label, func, *args, retries=3, **kwargs):
+    """Run a sub-suite, retrying only on a transient (status 0) transport blip
+    against the shared dev server.
+
+    A `(0)` ApiException is a raw connection/protocol hiccup (stale keep-alive,
+    HTTP/2 GOAWAY race, client closed by a fork cleanup, etc.) — not a real
+    assertion or server failure — and was observed flaking the `test-all`
+    bucket. Retrying absorbs that network noise while still surfacing genuine
+    failures immediately (any non-transient error, or a transient one on the
+    final attempt, re-raises).
+    """
+    for attempt in range(retries):
+        try:
+            return func(*args, **kwargs)
+        except ApiException as e:
+            transient = getattr(e, 'transient', False) or e.status in (0, None)
+            if transient and attempt < retries - 1:
+                logger.warning(
+                    'transient (%s) API error in %s (attempt %d/%d): %s; retrying',
+                    e.status, label, attempt + 1, retries, e)
+                time.sleep(2 ** attempt)
+                continue
+            raise
+
+
 class TestOrkesWorkflowClientIntg(unittest.TestCase):
 
     @classmethod
@@ -40,13 +69,20 @@ def setUpClass(cls):
         cls.workflow_client = OrkesWorkflowClient(cls.config)
         logger.info(f'setting up TestOrkesWorkflowClientIntg with config {cls.config}')
 
+    @pytest.mark.slow_test_all
     def test_all(self):
         logger.info('START: integration tests')
         configuration = self.config
         workflow_executor = WorkflowExecutor(configuration)
 
         # test_async.test_async_method(api_client)
-        run_workflow_definition_tests(workflow_executor)
-        run_workflow_execution_tests(configuration, workflow_executor)
-        TestOrkesClients(configuration=configuration).run()
+        _run_tolerating_transient(
+            'workflow_definition_tests',
+            run_workflow_definition_tests, workflow_executor)
+        _run_tolerating_transient(
+            'workflow_execution_tests',
+            run_workflow_execution_tests, configuration, workflow_executor)
+        _run_tolerating_transient(
+            'orkes_clients',
+            TestOrkesClients(configuration=configuration).run)
         logger.info('END: integration tests')
diff --git a/tests/integration/workflow/test_workflow_execution.py b/tests/integration/workflow/test_workflow_execution.py
index 2583ca3f..951c88f1 100644
--- a/tests/integration/workflow/test_workflow_execution.py
+++ b/tests/integration/workflow/test_workflow_execution.py
@@ -46,21 +46,21 @@ def run_workflow_execution_tests(configuration: Configuration, workflow_executor
     )
     task_handler.start_processes()
     try:
-        test_get_workflow_by_correlation_ids(workflow_executor)
+        scenario_get_workflow_by_correlation_ids(workflow_executor)
         logger.debug('finished workflow correlation ids test')
-        test_workflow_registration(workflow_executor)
+        scenario_workflow_registration(workflow_executor)
         logger.debug('finished workflow registration tests')
-        test_workflow_execution(
+        scenario_workflow_execution(
             workflow_quantity=6,
             workflow_name=WORKFLOW_NAME,
             workflow_executor=workflow_executor,
             workflow_completion_timeout=5.0
         )
-        test_decorated_workers(workflow_executor)
+        scenario_decorated_workers(workflow_executor)
         logger.debug('finished decorated workers tests')
-        test_execute_workflow_async_features(workflow_executor)
+        scenario_execute_workflow_async_features(workflow_executor)
         logger.debug('finished execute_workflow reactive features tests')
-        test_execute_workflow_error_handling(workflow_executor)
+        scenario_execute_workflow_error_handling(workflow_executor)
         logger.debug('finished execute_workflow error handling tests')
         run_signal_tests(configuration, workflow_executor)
         logger.debug('finished signal API tests')
@@ -86,7 +86,7 @@ def generate_tasks_defs():
     return [python_simple_task_from_code]
 
 
-def test_get_workflow_by_correlation_ids(workflow_executor: WorkflowExecutor):
+def scenario_get_workflow_by_correlation_ids(workflow_executor: WorkflowExecutor):
     _run_with_retry_attempt(
         workflow_executor.get_by_correlation_ids,
         {
@@ -98,7 +98,7 @@ def test_get_workflow_by_correlation_ids(workflow_executor: WorkflowExecutor):
     )
 
 
-def test_workflow_registration(workflow_executor: WorkflowExecutor):
+def scenario_workflow_registration(workflow_executor: WorkflowExecutor):
     workflow = generate_workflow(workflow_executor)
     try:
         workflow_executor.metadata_client.unregister_workflow_def_with_http_info(
@@ -113,7 +113,7 @@ def test_workflow_registration(workflow_executor: WorkflowExecutor):
     )
 
 
-def test_decorated_workers(
+def scenario_decorated_workers(
         workflow_executor: WorkflowExecutor,
         workflow_name: str = 'TestPythonDecoratedWorkerWf',
 ) -> None:
@@ -154,7 +154,7 @@ def test_decorated_workers(
     workflow_executor.metadata_client.unregister_workflow_def(wf.name, wf.version)
 
 
-def test_workflow_execution(
+def scenario_workflow_execution(
         workflow_quantity: int,
         workflow_name: str,
         workflow_executor: WorkflowExecutor,
@@ -229,7 +229,7 @@ def _run_with_retry_attempt(f, params, retries=4) -> None:
                 raise e
             sleep(1 << attempt)
 
-def test_execute_workflow_async_features(workflow_executor: WorkflowExecutor):
+def scenario_execute_workflow_async_features(workflow_executor: WorkflowExecutor):
     """Test the execute_workflow method with reactive features (consistency and return_strategy)"""
     logger.debug('Starting execute_workflow reactive features tests')
 
@@ -367,7 +367,7 @@ def test_execute_workflow_async_features(workflow_executor: WorkflowExecutor):
     logger.debug('All execute_workflow reactive features tests passed!')
 
 
-def test_execute_workflow_error_handling(workflow_executor: WorkflowExecutor):
+def scenario_execute_workflow_error_handling(workflow_executor: WorkflowExecutor):
     """Test error handling in execute_workflow with invalid parameters"""
     logger.debug('Starting execute_workflow error handling tests')
 
@@ -433,19 +433,19 @@ def run_signal_tests(configuration: Configuration, workflow_executor: WorkflowEx
         _register_signal_test_workflows(workflow_executor)
 
         # Test sync signal with different return strategies
-        test_signal_target_workflow(workflow_executor)
-        test_signal_blocking_workflow(workflow_executor)
-        test_signal_blocking_task(workflow_executor)
-        test_signal_blocking_task_input(workflow_executor)
+        scenario_signal_target_workflow(workflow_executor)
+        scenario_signal_blocking_workflow(workflow_executor)
+        scenario_signal_blocking_task(workflow_executor)
+        scenario_signal_blocking_task_input(workflow_executor)
 
         # Test default return strategy
-        test_signal_default_strategy(workflow_executor)
+        scenario_signal_default_strategy(workflow_executor)
 
         # Test async signal
-        test_signal_async(workflow_executor)
+        scenario_signal_async(workflow_executor)
 
         # Test to_dict fix
-        test_signal_to_dict_fix(workflow_executor)
+        scenario_signal_to_dict_fix(workflow_executor)
 
         logger.info('All signal tests completed successfully')
 
@@ -569,7 +569,7 @@ def _complete_workflow(workflow_executor: WorkflowExecutor, workflow_id: str):
         raise
 
 
-def test_signal_target_workflow(workflow_executor: WorkflowExecutor):
+def scenario_signal_target_workflow(workflow_executor: WorkflowExecutor):
     """Test signal with TARGET_WORKFLOW return strategy"""
     logger.info('Testing signal with TARGET_WORKFLOW strategy...')
 
@@ -640,7 +640,7 @@ def test_signal_target_workflow(workflow_executor: WorkflowExecutor):
     logger.info('TARGET_WORKFLOW strategy test completed')
 
 
-def test_signal_blocking_workflow(workflow_executor: WorkflowExecutor):
+def scenario_signal_blocking_workflow(workflow_executor: WorkflowExecutor):
     """Test signal with BLOCKING_WORKFLOW return strategy"""
     logger.info('Testing signal with BLOCKING_WORKFLOW strategy...')
 
@@ -669,7 +669,7 @@ def test_signal_blocking_workflow(workflow_executor: WorkflowExecutor):
     logger.info('BLOCKING_WORKFLOW strategy test completed')
 
 
-def test_signal_blocking_task(workflow_executor: WorkflowExecutor):
+def scenario_signal_blocking_task(workflow_executor: WorkflowExecutor):
     """Test signal with BLOCKING_TASK return strategy"""
     logger.info('Testing signal with BLOCKING_TASK strategy...')
 
@@ -699,7 +699,7 @@ def test_signal_blocking_task(workflow_executor: WorkflowExecutor):
     logger.info('BLOCKING_TASK strategy test completed')
 
 
-def test_signal_blocking_task_input(workflow_executor: WorkflowExecutor):
+def scenario_signal_blocking_task_input(workflow_executor: WorkflowExecutor):
     """Test signal with BLOCKING_TASK_INPUT return strategy"""
     logger.info('Testing signal with BLOCKING_TASK_INPUT strategy...')
 
@@ -730,7 +730,7 @@ def test_signal_blocking_task_input(workflow_executor: WorkflowExecutor):
     logger.info('BLOCKING_TASK_INPUT strategy test completed')
 
 
-def test_signal_default_strategy(workflow_executor: WorkflowExecutor):
+def scenario_signal_default_strategy(workflow_executor: WorkflowExecutor):
     """Test signal with default return strategy"""
     logger.info('Testing signal with default strategy...')
 
@@ -754,7 +754,7 @@ def test_signal_default_strategy(workflow_executor: WorkflowExecutor):
     logger.info('Default strategy test completed')
 
 
-def test_signal_async(workflow_executor: WorkflowExecutor):
+def scenario_signal_async(workflow_executor: WorkflowExecutor):
     """Test async signal"""
     logger.info('Testing async signal...')
 
@@ -775,7 +775,7 @@ def test_signal_async(workflow_executor: WorkflowExecutor):
     logger.info('Async signal test completed')
 
 
-def test_signal_to_dict_fix(workflow_executor: WorkflowExecutor):
+def scenario_signal_to_dict_fix(workflow_executor: WorkflowExecutor):
     """Test that to_dict() returns actual values, not property objects"""
     logger.info('Testing to_dict() method fix...')