diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
index 9ffbd9081..04234cc85 100644
--- a/.github/workflows/integration-tests.yml
+++ b/.github/workflows/integration-tests.yml
@@ -5,6 +5,26 @@ on:
     workflows: ["Java Client Build"]
     types:
       - completed
+  workflow_dispatch:
+    inputs:
+      oss_conductor_version:
+        description: 'OSS Conductor image tag (falls back to E2E_TEST_OSS_CONDUCTOR_VERSION org var)'
+        required: false
+        type: string
+
+  # ---------------------------------------------------------------------------
+  # TEMPORARY -- REMOVE BEFORE MERGE.
+  #
+  # A workflow_run-triggered run always executes the copy of this file on the
+  # default branch, so the integration-tests-oss job below cannot be exercised
+  # from a pull request. push runs the branch's own copy, so this trigger is
+  # what lets the job actually run while it is being developed.
+  #
+  # Deleting this block is sufficient to revert; nothing else below depends on it.
+  # ---------------------------------------------------------------------------
+  push:
+    branches:
+      - e2e-against-conductor-with-local-script
 
 # allow this workflow to update the status of the PR that triggered it
 permissions:
@@ -12,7 +32,9 @@ permissions:
   checks: write
 
 concurrency:
-  group: ${{ github.workflow }}-${{ github.event.workflow_run.head_branch }}
+  # head_branch is only set for workflow_run; fall back to the ref so runs from
+  # other triggers do not all collapse into one unnamed concurrency group.
+  group: ${{ github.workflow }}-${{ github.event.workflow_run.head_branch || github.ref }}
   cancel-in-progress: true
 
 jobs:
@@ -44,7 +66,11 @@ jobs:
         if: always()
         with:
           report_paths: '**/build/test-results/test/TEST-*.xml'
-          
+          # Distinct per suite: the action defaults to 'JUnit Test Report' for every caller, so
+          # this report, the OSS one below, and ci.yml's unit report all landed on a single
+          # check run and overwrote each other.
+          check_name: Integration Test Report
+
       - name: Update PR Status
         if: always()
         uses: actions/github-script@v8
@@ -60,4 +86,90 @@ jobs:
               description: 'Integration tests ${{ job.status }}'
             });
 
+  # Runs against a throwaway Conductor OSS stack rather than the Orkes deployment the job above
+  # targets, so it needs no secrets and no `environment`. It is deliberately a sibling of that
+  # job, not a dependent: an OSS failure must not suppress the enterprise suite, or vice versa.
+  integration-tests-oss:
+    runs-on: ubuntu-latest
+    name: Integration Tests (OSS)
+    timeout-minutes: 30
+    # The build-succeeded gate only makes sense for workflow_run, which is the only trigger that
+    # has a triggering run to inspect. Any other trigger is a direct request to run this suite.
+    if: ${{ github.event_name != 'workflow_run' || github.event.workflow_run.conclusion == 'success' }}
+    env:
+      CONDUCTOR_SERVER_URL: http://localhost:8080/api
+      CONDUCTOR_SERVER_TYPE: oss
+      OSS_CONDUCTOR_VERSION: ${{ inputs.oss_conductor_version || vars.E2E_TEST_OSS_CONDUCTOR_VERSION }}
+
+    steps:
+      - name: Verify OSS Conductor version is set
+        run: |
+          if [ -z "$OSS_CONDUCTOR_VERSION" ]; then
+            echo "::error::No Conductor OSS image tag resolved. Set the E2E_TEST_OSS_CONDUCTOR_VERSION organization variable (and ensure its repository access policy includes this repo), or pass the oss_conductor_version input via workflow_dispatch."
+            exit 1
+          fi
+          echo "Using conductoross/conductor:$OSS_CONDUCTOR_VERSION"
+
+      - name: Checkout
+        uses: actions/checkout@v6
+        with:
+          # workflow_run carries the triggering run's head; workflow_dispatch has neither, and
+          # falls back to the ref the dispatch was made against.
+          ref: ${{ github.event.workflow_run.head_sha || github.sha }}
+          repository: ${{ github.event.workflow_run.repository.full_name || github.repository }}
+
+      - name: Set up Zulu JDK 21
+        uses: actions/setup-java@v5
+        with:
+          distribution: "zulu"
+          java-version: "21"
+
+      - name: Start Conductor OSS stack
+        run: docker compose -f scripts/docker-compose-oss.yaml up -d
+
+      - name: Wait for Conductor to be healthy
+        # Matches HEALTH_TIMEOUT in scripts/run-integration-oss.sh, and stays under the compose
+        # healthcheck's own ~200s budget.
+        run: timeout 180 bash -c 'until curl -sf http://localhost:8080/health; do sleep 5; done'
+
+      - name: Run integration tests (OSS)
+        id: integration_tests
+        continue-on-error: true
+        run: ./gradlew :tests:test -PIntegrationTests
+
+      - name: Dump Conductor logs
+        if: failure() || steps.integration_tests.outcome == 'failure'
+        run: docker compose -f scripts/docker-compose-oss.yaml logs conductor-server
+
+      - name: Publish Test Report
+        if: always()
+        uses: mikepenz/action-junit-report@v6
+        with:
+          report_paths: 'tests/build/test-results/test/TEST-*.xml'
+          check_name: OSS Integration Test Report
+
+      # Before Update PR Status, not after: the test step is continue-on-error, so job.status is
+      # still 'success' until this step's exit 1 flips it.
+      - name: Check Integration Tests Status
+        if: steps.integration_tests.outcome == 'failure'
+        run: |
+          echo "::error::Integration tests (OSS) failed. See the 'Run integration tests (OSS)' step above for the Gradle/test output, the 'Dump Conductor logs' step for server-side logs, and the 'Publish Test Report' step's JUnit summary for which test(s) failed."
+          exit 1
+
+      - name: Update PR Status
+        # Skipped on workflow_dispatch: there is no triggering run, so there is no PR head to
+        # report against and context.payload.workflow_run would be undefined.
+        if: ${{ always() && github.event_name == 'workflow_run' }}
+        uses: actions/github-script@v8
+        with:
+          script: |
+            const { owner, repo } = context.repo;
+            const sha = context.payload.workflow_run.head_sha;
+            await github.rest.repos.createCommitStatus({
+              owner, repo, sha,
+              state: '${{ job.status }}' === 'success' ? 'success' : 'failure',
+              context: 'Integration Tests (OSS)',
+              target_url: `${context.serverUrl}/${owner}/${repo}/actions/runs/${context.runId}`,
+              description: 'OSS integration tests ${{ job.status }}'
+            });
 
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 7bfa1458c..7876321c5 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -23,6 +23,47 @@ Run the SDK test suite:
 ./gradlew test jacocoTestReport
 ```
 
+### Running the OSS integration suite locally
+
+The `tests` module also has an integration suite (`-PIntegrationTests`) that runs against a
+real Conductor server, separate from the unit suite above. `scripts/run-integration-oss.sh`
+mirrors the `integration-tests-oss` job in `integration-tests.yml`: it starts a local Conductor OSS +
+Postgres stack (defined in `scripts/docker-compose-oss.yaml`), waits for `/health`, runs the
+integration suite, and tears the stack down on exit.
+
+```shell
+scripts/run-integration-oss.sh                        # against `latest`
+scripts/run-integration-oss.sh --version 3.32.0-rc18
+scripts/run-integration-oss.sh --keep-up               # leave the stack running afterwards
+scripts/run-integration-oss.sh --include-gated         # also run tests normally skipped as Orkes-only
+```
+
+The script always prints the resolved `conductoross/conductor` tag and pulls it before
+starting the stack, since `latest` (the local default) is a mutable tag — without an
+explicit pull, `docker compose up` would silently reuse a stale cached image instead of
+fetching the current one. It also always runs Gradle with `--rerun-tasks`, since the `test`
+task's up-to-date check doesn't account for env vars like `CONDUCTOR_SERVER_TYPE` or the
+state of the live server underneath — without it, a rerun after changing gating or switching
+server versions could silently report a stale cached result instead of executing anything.
+
+The script doesn't pin a JDK itself, but CI runs on Zulu 21. If your local default JDK is
+newer (e.g. 23) you may hit `Unsupported class file major version` errors compiling tests —
+set `JAVA_HOME` explicitly to match CI:
+
+```shell
+JAVA_HOME=/Library/Java/JavaVirtualMachines/zulu-21.jdk/Contents/Home ./scripts/run-integration-oss.sh
+```
+
+Tests annotated `@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches =
+"oss")` skip themselves against plain OSS because they exercise Orkes-managed-only features
+(e.g. the Service Registry, Authorization, Prompts/Integrations, Environment Variables, and
+Secrets APIs, plus a handful of task/workflow endpoints OSS doesn't implement or that hit
+known Postgres-persistence bugs). Each annotation's `disabledReason` documents the specific,
+empirically-confirmed gap — treat those as the source of truth rather than a list here, since
+they can drift as OSS gains features. If you add or remove that annotation, re-verify against
+a freshly-pulled image first: a test that fails against a stale local image may pass against
+current OSS, and vice versa.
+
 Compile the maintained agent examples when changing their APIs or documentation:
 
 ```shell
diff --git a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java
index 33654d78d..f185e6e6d 100644
--- a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java
+++ b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java
@@ -174,10 +174,21 @@ private void initMonitor() {
                     for (Map.Entry<String, CompletableFuture<Workflow>> entry : runningWorkflowFutures.entrySet()) {
                         String workflowId = entry.getKey();
                         CompletableFuture<Workflow> future = entry.getValue();
-                        Workflow workflow = workflowClient.getWorkflow(workflowId, true);
-                        if (workflow.getStatus().isTerminal()) {
-                            future.complete(workflow);
+                        try {
+                            Workflow workflow = workflowClient.getWorkflow(workflowId, true);
+                            if (workflow.getStatus().isTerminal()) {
+                                future.complete(workflow);
+                                runningWorkflowFutures.remove(workflowId);
+                            }
+                        } catch (Exception e) {
+                            // scheduleAtFixedRate silently kills all future ticks on any uncaught
+                            // exception, so this must not escape: one failed poll would otherwise
+                            // stop completion tracking for every workflow, forever. Stop tracking
+                            // this one and let its caller see the failure.
+                            LOGGER.error("Error polling workflow {} for completion; completing its "
+                                    + "future exceptionally", workflowId, e);
                             runningWorkflowFutures.remove(workflowId);
+                            future.completeExceptionally(e);
                         }
                     }
                 },
diff --git a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java
index e2ed6d107..634f9399f 100644
--- a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java
+++ b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java
@@ -52,6 +52,12 @@ public class AnnotatedWorkerExecutor {
 
     private final Set<String> scannedPackages = new HashSet<>();
 
+    /**
+     * Set whenever a worker is added, cleared whenever {@link #startPolling()} builds a task runner.
+     * Lets startPolling() distinguish a real (re)start from a redundant duplicate call.
+     */
+    private boolean workersChanged = false;
+
     private final WorkerConfiguration workerConfiguration;
 
     public AnnotatedWorkerExecutor(TaskClient taskClient) {
@@ -77,6 +83,9 @@ public AnnotatedWorkerExecutor(TaskClient taskClient, WorkerConfiguration worker
      */
     public synchronized void initWorkers(String... basePackages) {
         scanWorkers(basePackages);
+        // scanWorkers() -> initWorkersFromClasses() -> initWorkersFromInstances() already reaches
+        // startPolling(). This second call is therefore redundant, but startPolling() is idempotent
+        // while the worker set is unchanged, so it is a no-op rather than a task runner restart.
         startPolling();
     }
 
@@ -109,10 +118,18 @@ public synchronized void initWorkersFromClasses(List<? extends Class<?>> classes
     }
 
 
-    /** Shuts down the workers */
-    public void shutdown() {
+    /**
+     * Shuts down the workers.
+     *
+     * <p>Clears the task runner as well as shutting it down, so a later {@link #startPolling()}
+     * builds a fresh one and resumes. A {@link TaskRunnerConfigurer} is single-use — its
+     * shutdown closes the executor backing it — so leaving the field set would make
+     * startPolling() take its already-polling fast path and never poll again.
+     */
+    public synchronized void shutdown() {
         if (taskRunner != null) {
             taskRunner.shutdown();
+            taskRunner = null;
         }
     }
 
@@ -168,7 +185,15 @@ private boolean classBelongsToPackage(List<String> packagesToScan, String classN
         return false;
     }
 
-    public void addBean(Object bean) {
+    /**
+     * Registers every {@link WorkerTask}-annotated method on the bean as a worker.
+     *
+     * <p>Synchronized because it is the public entry point that mutates the worker set and the
+     * {@code workersChanged} flag {@link #startPolling()} reads to decide whether a restart is
+     * needed. Without a common lock, a worker added on another thread could be missed and never
+     * polled.
+     */
+    public synchronized void addBean(Object bean) {
         Class<?> clazz = bean.getClass();
         for (Method method : clazz.getMethods()) {
             WorkerTask annotation = method.getAnnotation(WorkerTask.class);
@@ -222,6 +247,7 @@ private void addMethod(WorkerTask annotation, Method method, Object bean) {
         for (int i = 0; i < pollerCount; i++) {
             workers.add(executor);
         }
+        workersChanged = true;
 
         LOGGER.info(
                 "Adding worker for task {}, method {} with threadCount {} and polling interval set to {} ms",
@@ -231,11 +257,28 @@ private void addMethod(WorkerTask annotation, Method method, Object bean) {
                 pollingInterval);
     }
 
-    public void startPolling() {
+    /**
+     * Builds a {@link TaskRunnerConfigurer} over the currently registered workers and starts polling.
+     *
+     * <p>Idempotent: if a task runner is already polling and no worker has been added since it was
+     * built, this returns without doing anything. Restarting unnecessarily would stand up a second
+     * runner polling the same task types and only shut the first one down afterwards, so a task
+     * already leased by the outgoing runner could be left in-progress until its response timeout
+     * expired. Callers that add workers and call this again still get the intended restart.
+     */
+    public synchronized void startPolling() {
         if (workers.isEmpty()) {
             return;
         }
 
+        if (taskRunner != null && !workersChanged) {
+            LOGGER.debug(
+                    "Task runner is already polling {} workers and the worker set is unchanged; "
+                            + "skipping redundant restart.",
+                    workers.size());
+            return;
+        }
+
         LOGGER.info("Starting {} with threadCount {}", workers.stream().map(Worker::getTaskDefName).toList(), workerToThreadCount);
         LOGGER.info("Worker domains {}", workerDomains);
         LOGGER.info("Worker workerToPollTimeout (in millis) {}", workerToPollTimeout);
@@ -253,6 +296,8 @@ public void startPolling() {
         taskRunner = builder.build();
         taskRunner.init();
 
+        workersChanged = false;
+
         oldTaskRunner.ifPresent(taskRunner -> {
             LOGGER.trace("Shutting down previous task runner with {} workers.", taskRunner.getWorkerCount());
             taskRunner.shutdown();
diff --git a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java
new file mode 100644
index 000000000..56f684ac4
--- /dev/null
+++ b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java
@@ -0,0 +1,110 @@
+/*
+ * Copyright 2026 Conductor Authors.
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package com.netflix.conductor.sdk.workflow.executor;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import com.netflix.conductor.client.http.MetadataClient;
+import com.netflix.conductor.client.http.TaskClient;
+import com.netflix.conductor.client.http.WorkflowClient;
+import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest;
+import com.netflix.conductor.common.run.Workflow;
+import com.netflix.conductor.sdk.workflow.executor.task.AnnotatedWorkerExecutor;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Covers the completion-tracking monitor started by {@link WorkflowExecutor}'s constructors. The
+ * monitor runs under {@code scheduleAtFixedRate}, which cancels all future ticks on any uncaught
+ * exception, so a single failed {@code getWorkflow} call must not be allowed to escape.
+ */
+public class WorkflowExecutorMonitorTests {
+
+    private static final String WORKFLOW_ID = "test-workflow-id";
+
+    private static final String OTHER_WORKFLOW_ID = "other-test-workflow-id";
+
+    private WorkflowExecutor executorFor(WorkflowClient workflowClient) {
+        return new WorkflowExecutor(
+                mock(TaskClient.class),
+                workflowClient,
+                mock(MetadataClient.class),
+                mock(AnnotatedWorkerExecutor.class));
+    }
+
+    @Test
+    @DisplayName("a failed poll should complete that workflow's future exceptionally")
+    void monitorFailsTheFutureOnPollFailure() {
+        WorkflowClient workflowClient = mock(WorkflowClient.class);
+        when(workflowClient.startWorkflow(any(StartWorkflowRequest.class))).thenReturn(WORKFLOW_ID);
+        when(workflowClient.getWorkflow(anyString(), anyBoolean()))
+                .thenThrow(new RuntimeException("poll failure"));
+
+        WorkflowExecutor executor = executorFor(workflowClient);
+        try {
+            CompletableFuture<Workflow> future = executor.executeWorkflow("wf", 1, Map.of());
+
+            ExecutionException thrown = assertThrows(
+                    ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS));
+
+            assertInstanceOf(RuntimeException.class, thrown.getCause());
+            assertEquals("poll failure", thrown.getCause().getMessage());
+        } finally {
+            executor.shutdown();
+        }
+    }
+
+    @Test
+    @DisplayName("one workflow's poll failure should not stop the monitor tracking the rest")
+    void monitorKeepsTrackingOtherWorkflowsAfterAFailure() throws Exception {
+        Workflow completed = new Workflow();
+        completed.setStatus(Workflow.WorkflowStatus.COMPLETED);
+
+        WorkflowClient workflowClient = mock(WorkflowClient.class);
+        when(workflowClient.startWorkflow(any(StartWorkflowRequest.class)))
+                .thenReturn(WORKFLOW_ID)
+                .thenReturn(OTHER_WORKFLOW_ID);
+        when(workflowClient.getWorkflow(eq(WORKFLOW_ID), anyBoolean()))
+                .thenThrow(new RuntimeException("poll failure"));
+        when(workflowClient.getWorkflow(eq(OTHER_WORKFLOW_ID), anyBoolean()))
+                .thenReturn(completed);
+
+        WorkflowExecutor executor = executorFor(workflowClient);
+        try {
+            CompletableFuture<Workflow> failing = executor.executeWorkflow("wf", 1, Map.of());
+            assertThrows(ExecutionException.class, () -> failing.get(5, TimeUnit.SECONDS));
+
+            // Registered only after the failure has already happened, so it can complete at all
+            // only if the tick loop survived it -- scheduleAtFixedRate would have cancelled every
+            // future tick had the exception been allowed to escape.
+            CompletableFuture<Workflow> healthy = executor.executeWorkflow("wf", 1, Map.of());
+            assertEquals(Workflow.WorkflowStatus.COMPLETED, healthy.get(5, TimeUnit.SECONDS).getStatus());
+        } finally {
+            executor.shutdown();
+        }
+    }
+}
diff --git a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java
index abc4a19fd..37c0b4c7c 100644
--- a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java
+++ b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java
@@ -84,6 +84,84 @@ void directInstanceSupply() {
         )));
     }
 
+    @Test
+    @DisplayName("initWorkers should leave exactly one task runner polling")
+    void initWorkersStartsASingleTaskRunner() {
+        var executor = new AnnotatedWorkerExecutor(mock(TaskClient.class));
+        executor.initWorkers("com.netflix.conductor.sdk.workflow.executor.task.workers1");
+
+        var runner = executor.getTaskRunner();
+        assertNotNull(runner);
+
+        // The shape used by examples/old/.../taskdomains/Main.java and by callers following the
+        // docs: initWorkers() followed by an explicit startPolling(). This must not stand up a
+        // replacement runner alongside the live one.
+        executor.startPolling();
+        assertSame(runner, executor.getTaskRunner());
+
+        executor.shutdown();
+    }
+
+    @Test
+    @DisplayName("startPolling should be a no-op while the worker set is unchanged")
+    void startPollingIsIdempotent() {
+        var executor = new AnnotatedWorkerExecutor(mock(TaskClient.class));
+        executor.addBean(new MultipleInputParams());
+
+        executor.startPolling();
+        var first = executor.getTaskRunner();
+        assertNotNull(first);
+
+        executor.startPolling();
+        executor.startPolling();
+        assertSame(first, executor.getTaskRunner());
+
+        executor.shutdown();
+    }
+
+    @Test
+    @DisplayName("startPolling should still rebuild the task runner once new workers are added")
+    void startPollingRebuildsWhenWorkersAreAdded() {
+        var executor = new AnnotatedWorkerExecutor(mock(TaskClient.class));
+        executor.addBean(new MultipleInputParams());
+        executor.startPolling();
+        var first = executor.getTaskRunner();
+        assertEquals(1, first.getWorkerCount());
+
+        executor.addBean(new AnotherAnnotationInput());
+        executor.startPolling();
+        var second = executor.getTaskRunner();
+
+        assertNotSame(first, second);
+        assertEquals(2, second.getWorkerCount());
+
+        executor.shutdown();
+    }
+
+    @Test
+    @DisplayName("startPolling should build a fresh task runner after a shutdown")
+    void startPollingRestartsAfterShutdown() {
+        var executor = new AnnotatedWorkerExecutor(mock(TaskClient.class));
+        executor.addBean(new MultipleInputParams());
+        executor.startPolling();
+        var first = executor.getTaskRunner();
+        assertNotNull(first);
+
+        // A TaskRunnerConfigurer is single-use, so shutdown() has to clear the field as well as
+        // shut the runner down. Otherwise startPolling()'s idempotence check sees a non-null
+        // runner with an unchanged worker set and silently declines to poll ever again.
+        executor.shutdown();
+        assertNull(executor.getTaskRunner());
+
+        executor.startPolling();
+        var second = executor.getTaskRunner();
+        assertNotNull(second);
+        assertNotSame(first, second);
+        assertEquals(1, second.getWorkerCount());
+
+        executor.shutdown();
+    }
+
     @Test
     @DisplayName("it should handle null values when InputParam is a List")
     void nullListAsInputParam() throws NoSuchMethodException {
diff --git a/scripts/docker-compose-oss.yaml b/scripts/docker-compose-oss.yaml
new file mode 100644
index 000000000..cb1f36f9b
--- /dev/null
+++ b/scripts/docker-compose-oss.yaml
@@ -0,0 +1,37 @@
+# Conductor OSS stack used to run the SDK integration tests against open-source Conductor.
+# Shared by scripts/run-integration-oss.sh and the integration-tests-oss job in
+# .github/workflows/integration-tests.yml.
+#
+# OSS_CONDUCTOR_VERSION defaults to `latest` for local runs. CI resolves it from the
+# E2E_TEST_OSS_CONDUCTOR_VERSION org variable (or a workflow_dispatch input); that variable is
+# currently set to `latest` too, so CI tracks whatever `latest` resolves to at run time rather
+# than a fixed version. Set the org variable to a real tag if the job needs to be deterministic.
+
+services:
+  conductor-server:
+    image: conductoross/conductor:${OSS_CONDUCTOR_VERSION:-latest}
+    environment:
+      - CONFIG_PROP=config-postgres.properties
+    ports:
+      - "8080:8080"
+    healthcheck:
+      test: ["CMD", "curl", "-I", "-XGET", "http://localhost:8080/health"]
+      interval: 10s
+      timeout: 10s
+      retries: 20
+    links:
+      - conductor-postgres:postgresdb
+    depends_on:
+      conductor-postgres:
+        condition: service_healthy
+
+  conductor-postgres:
+    image: postgres:16
+    environment:
+      - POSTGRES_USER=conductor
+      - POSTGRES_PASSWORD=conductor
+    healthcheck:
+      test: timeout 5 bash -c 'cat < /dev/null > /dev/tcp/localhost/5432'
+      interval: 5s
+      timeout: 5s
+      retries: 12
diff --git a/scripts/run-integration-oss.sh b/scripts/run-integration-oss.sh
new file mode 100755
index 000000000..1d0d27ac7
--- /dev/null
+++ b/scripts/run-integration-oss.sh
@@ -0,0 +1,113 @@
+#!/usr/bin/env bash
+#
+# Spin up a local Conductor OSS stack and run the `tests` module's
+# integration suite against it, mirroring the `integration-tests-oss` job in
+# .github/workflows/integration-tests.yml. Orkes-Enterprise-only test
+# classes are annotated with
+# @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss")
+# so they skip themselves when it's set (see the individual test files for
+# the empirically-confirmed gaps).
+#
+# The stack (Conductor OSS + Postgres) is defined in
+# scripts/docker-compose-oss.yaml and is torn down automatically on exit. The
+# image is always pulled before starting, since `latest` (the local default)
+# is a mutable tag and a cached copy would otherwise go stale silently.
+#
+# Usage:
+#   scripts/run-integration-oss.sh [--keep-up] [--version <tag>] [--include-gated] [-- gradle args]
+# Examples:
+#   scripts/run-integration-oss.sh
+#   scripts/run-integration-oss.sh --version 3.32.0-rc18
+#   scripts/run-integration-oss.sh --keep-up
+#   scripts/run-integration-oss.sh --include-gated       # also run tests normally skipped as Orkes-only
+#   scripts/run-integration-oss.sh -- --tests "*WorkflowClientTests"
+set -euo pipefail
+
+KEEP_UP=0
+INCLUDE_GATED=0
+extra=()
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --keep-up) KEEP_UP=1; shift ;;
+    --version) OSS_CONDUCTOR_VERSION="${2:?--version needs a tag}"; shift 2 ;;
+    --include-gated) INCLUDE_GATED=1; shift ;;
+    -h|--help)
+      echo "Usage: $0 [--keep-up] [--version <tag>] [--include-gated] [-- gradle args]"
+      exit 0
+      ;;
+    --) shift; extra=("$@"); break ;;
+    *) echo "Unknown argument: $1" >&2; exit 1 ;;
+  esac
+done
+
+export OSS_CONDUCTOR_VERSION="${OSS_CONDUCTOR_VERSION:-latest}"
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
+COMPOSE_FILE="${SCRIPT_DIR}/docker-compose-oss.yaml"
+cd "${REPO_ROOT}"
+
+compose() { docker compose -f "${COMPOSE_FILE}" "$@"; }
+
+cleanup() {
+  if [[ "${KEEP_UP}" == "1" ]]; then
+    echo "--keep-up set: leaving the OSS stack running. Tear down with:"
+    echo "  docker compose -f ${COMPOSE_FILE} down -v"
+    return
+  fi
+  echo "Tearing down Conductor OSS stack..."
+  compose down -v || true
+}
+trap cleanup EXIT
+
+echo "Using conductoross/conductor:${OSS_CONDUCTOR_VERSION}"
+
+# `docker compose up` only pulls an image when it is missing locally, so a
+# previously-cached `latest` (or any other mutable tag) would silently be
+# reused instead of getting the current version. Pull unconditionally so the
+# stack always reflects the tag we just printed.
+echo "Pulling conductoross/conductor:${OSS_CONDUCTOR_VERSION} to ensure it's current..."
+compose pull conductor-server
+
+echo "Starting Conductor OSS stack..."
+compose up -d
+
+echo "Waiting for Conductor to be healthy..."
+HEALTH_TIMEOUT="${HEALTH_TIMEOUT:-180}"
+deadline=$(( SECONDS + HEALTH_TIMEOUT ))
+until curl -sf http://localhost:8080/health >/dev/null 2>&1; do
+  if (( SECONDS >= deadline )); then
+    echo "Error: Conductor did not become healthy within ${HEALTH_TIMEOUT}s." >&2
+    compose logs conductor-server || true
+    exit 1
+  fi
+  sleep 5
+done
+echo "Conductor is up."
+
+export CONDUCTOR_SERVER_URL="http://localhost:8080/api"
+
+# Plain OSS Conductor has no authentication layer and no /token endpoint. ClientTestUtil builds
+# its client with useEnvVariables(true), and ApiClient.applyEnvVariables() attaches credentials
+# whenever both of these are present -- so a shell that still has them exported for the Orkes
+# suite would send the whole run through an auth flow the local server cannot serve.
+unset CONDUCTOR_AUTH_KEY CONDUCTOR_AUTH_SECRET
+unset CONDUCTOR_SERVER_AUTH_KEY CONDUCTOR_SERVER_AUTH_SECRET
+
+if [[ "${INCLUDE_GATED}" == "1" ]]; then
+  echo "--include-gated set: leaving CONDUCTOR_SERVER_TYPE unset, so tests normally" \
+       "skipped as Orkes-only will run against OSS too."
+  unset CONDUCTOR_SERVER_TYPE || true
+else
+  export CONDUCTOR_SERVER_TYPE="oss"
+fi
+
+
+# --rerun-tasks: the `test` task's up-to-date check only considers the compiled
+# test classpath, not env vars like CONDUCTOR_SERVER_URL/CONDUCTOR_SERVER_TYPE
+# or the state of the live server underneath. Without this, Gradle can report
+# BUILD SUCCESSFUL while silently reusing a stale cached result from a
+# previous run against a different server/tag/gating state instead of
+# actually executing anything.
+./gradlew :tests:test -PIntegrationTests --rerun-tasks ${extra[@]+"${extra[@]}"}
diff --git a/tests/src/test/java/io/orkes/conductor/client/ServiceRegistryClientTest.java b/tests/src/test/java/io/orkes/conductor/client/ServiceRegistryClientTest.java
index 76922097e..64fc95d0f 100644
--- a/tests/src/test/java/io/orkes/conductor/client/ServiceRegistryClientTest.java
+++ b/tests/src/test/java/io/orkes/conductor/client/ServiceRegistryClientTest.java
@@ -19,6 +19,7 @@
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.common.model.OrkesCircuitBreakerConfig;
 import com.netflix.conductor.common.model.ServiceMethod;
@@ -30,6 +31,8 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+        disabledReason = "the Service Registry API (/registry/service) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/registry/service')")
 public class ServiceRegistryClientTest {
 
     private static final String PROTO_FILENAME = "compiled.bin";
diff --git a/tests/src/test/java/io/orkes/conductor/client/WorkflowRetryTest.java b/tests/src/test/java/io/orkes/conductor/client/WorkflowRetryTest.java
index e3793c077..739367075 100644
--- a/tests/src/test/java/io/orkes/conductor/client/WorkflowRetryTest.java
+++ b/tests/src/test/java/io/orkes/conductor/client/WorkflowRetryTest.java
@@ -18,6 +18,7 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
 import com.netflix.conductor.common.metadata.tasks.TaskResult;
@@ -36,6 +37,8 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
+@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+        disabledReason = "workflowClient.uploadCompletedWorkflows() (/workflow/document-store/upload) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/workflow/document-store/upload')")
 public class WorkflowRetryTest {
     private final OrkesMetadataClient metadataClient;
     private final OrkesWorkflowClient workflowClient;
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/AuthorizationClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/AuthorizationClientTests.java
index d40e829bc..fa896a45d 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/AuthorizationClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/AuthorizationClientTests.java
@@ -25,6 +25,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.client.exception.ConductorClientException;
 import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
@@ -49,6 +50,8 @@
 import io.orkes.conductor.client.util.ClientTestUtil;
 import io.orkes.conductor.client.util.Commons;
 
+@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+        disabledReason = "the Authorization APIs (applications/users/groups/roles/permissions) are not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/applications|users|groups|...')")
 public class AuthorizationClientTests {
     private static AuthorizationClient authorizationClient;
     private static String applicationId;
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/EnvironmentClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/EnvironmentClientTests.java
index 6dd0e3359..d1de9535b 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/EnvironmentClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/EnvironmentClientTests.java
@@ -19,11 +19,14 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
 
+@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+        disabledReason = "environment variable writes are not supported by plain OSS Conductor, confirmed empirically: OSS added read-only GET /environment in 3.32.0-rc.9 but PUT /environment/{key} still 405s ('Request method 'PUT' is not supported')")
 public class EnvironmentClientTests {
 
     private static EnvironmentClient envClient;
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java
index c1e21d90b..104e13ca5 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java
@@ -24,6 +24,7 @@
 
 import io.orkes.conductor.client.util.ClientTestUtil;
 import io.orkes.conductor.client.util.Commons;
+import io.orkes.conductor.client.util.TestUtil;
 
 public class EventClientTests {
     private static final String EVENT_NAME = "test_sdk_java_event_name";
@@ -35,9 +36,9 @@ void testEventHandler() {
         try {
             eventClient.unregisterEventHandler(EVENT_NAME);
         } catch (ConductorClientException e) {
-            if (e.getStatus() != 404) {
-                throw e;
-            }
+            // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server
+            // we're running against actually reports it.
+            TestUtil.tolerateNotFound(e, "EventHandler with name");
         }
         EventHandler eventHandler = getEventHandler();
         eventClient.registerEventHandler(eventHandler);
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java
index 1e91169d9..8338aa67e 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java
@@ -16,6 +16,7 @@
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.client.exception.ConductorClientException;
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
@@ -38,9 +39,9 @@ void taskDefinition() {
         try {
             metadataClient.unregisterTaskDef(Commons.TASK_NAME);
         } catch (ConductorClientException e) {
-            if (e.getStatus() != 404) {
-                throw e;
-            }
+            // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server
+            // we're running against actually reports it.
+            TestUtil.tolerateNotFound(e, "No such task definition");
         }
         TaskDef taskDef = Commons.getTaskDef();
         metadataClient.registerTaskDefs(List.of(taskDef));
@@ -54,9 +55,9 @@ void workflow() {
         try {
             metadataClient.unregisterWorkflowDef(Commons.WORKFLOW_NAME, Commons.WORKFLOW_VERSION);
         } catch (ConductorClientException e) {
-            if (e.getStatus() != 404) {
-                throw e;
-            }
+            // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server
+            // we're running against actually reports it.
+            TestUtil.tolerateNotFound(e, "No such workflow definition");
         }
         metadataClient.registerTaskDefs(List.of(Commons.getTaskDef()));
         WorkflowDef workflowDef = WorkflowUtil.getWorkflowDef();
@@ -73,6 +74,8 @@ void workflow() {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "task tagging (/metadata/task/{name}/tags) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/metadata/task/{name}/tags')")
     void tagTask() throws Exception {
         metadataClient.registerTaskDefs(List.of(Commons.getTaskDef()));
         try {
@@ -98,6 +101,8 @@ void tagTask() throws Exception {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "workflow tagging (/metadata/workflow/{name}/tags) is not implemented by plain OSS Conductor, confirmed empirically (the {version} path segment ends up matching the literal string \"tags\" instead, a server-side routing collision)")
     void tagWorkflow() {
         TagObject tagObject = Commons.getTagObject();
         try {
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/PromptClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/PromptClientTests.java
index 53d427232..2ce260590 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/PromptClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/PromptClientTests.java
@@ -20,6 +20,7 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.client.exception.ConductorClientException;
 
@@ -32,6 +33,8 @@
 import org.conductoross.conductor.client.model.ai.PromptTemplate;
 import io.orkes.conductor.client.util.ClientTestUtil;
 
+@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+        disabledReason = "the Prompts and Integrations APIs (/prompts, /integrations) are not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/prompts|integrations/...')")
 public class PromptClientTests {
     private static final String PROMPT_NAME = "test-sdk-java-prompt";
     private static final String PROMPT_DESCRIPTION = "Test prompt for Java SDK";
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/SchedulerClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/SchedulerClientTests.java
index f86c332d2..9bbc030a7 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/SchedulerClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/SchedulerClientTests.java
@@ -16,6 +16,7 @@
 import java.util.UUID;
 
 import org.junit.jupiter.api.*;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.common.model.BulkResponse;
 
@@ -51,6 +52,10 @@ void afterEach() {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "GET /scheduler/search does not 404 on plain OSS Conductor, but confirmed empirically to " +
+                    "always return zero results (even after polling for 30s) -- schedules aren't surfaced via search " +
+                    "on plain OSS the way they are on Orkes Enterprise")
     void testMethods() {
         schedulerClient.deleteSchedule(SCHEDULE_1);
         Assertions.assertTrue(schedulerClient.getNextFewSchedules(CRON_EXPRESSION_1, 0L, 0L, 0).isEmpty());
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/SchemaClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/SchemaClientTests.java
index b74b318fc..8dd145160 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/SchemaClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/SchemaClientTests.java
@@ -17,6 +17,7 @@
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.client.exception.ConductorClientException;
 import com.netflix.conductor.common.metadata.SchemaDef;
@@ -24,6 +25,8 @@
 import io.orkes.conductor.client.SchemaClient;
 import io.orkes.conductor.client.util.ClientTestUtil;
 
+@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+        disabledReason = "the Schema API (/schema) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/schema')")
 public class SchemaClientTests {
 
     private static final String SCHEMA_NAME = "test-sdk-java-schema";
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/SecretClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/SecretClientTests.java
index 1a8daecf8..33ce31ee7 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/SecretClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/SecretClientTests.java
@@ -16,6 +16,7 @@
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.client.exception.ConductorClientException;
 
@@ -24,6 +25,8 @@
 import io.orkes.conductor.client.util.ClientTestUtil;
 
 
+@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+        disabledReason = "secret writes are not supported by plain OSS Conductor, confirmed empirically: its env-var-backed secrets DAO is read-only, so putSecret() 501s ('env-backed secrets are read-only')")
 public class SecretClientTests {
     private final String SECRET_NAME = "test-sdk-java-secret_name";
     private final String SECRET_KEY = "test-sdk-java-secret_key";
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/ServiceRegistryClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/ServiceRegistryClientTests.java
index d5253fd8a..a09ba6c85 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/ServiceRegistryClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/ServiceRegistryClientTests.java
@@ -18,6 +18,7 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.client.exception.ConductorClientException;
 import com.netflix.conductor.common.model.CircuitBreakerTransitionResponse;
@@ -32,6 +33,8 @@
 
 import static org.junit.jupiter.api.Assertions.assertNull;
 
+@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+        disabledReason = "the Service Registry API (/registry/service) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/registry/service')")
 public class ServiceRegistryClientTests {
     private static final String SERVICE_NAME = "test-sdk-java-service";
     private static final String SERVICE_URI = "localhost:50051";
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java
index 4efb61f59..ef13c06d6 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java
@@ -30,6 +30,7 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
 import org.testcontainers.shaded.com.google.common.util.concurrent.Uninterruptibles;
 
@@ -146,6 +147,8 @@ public void testUpdateByRefName() {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the sync task-update endpoint (POST /tasks/{workflowId}/{taskRefName}/{status}/sync) never returns the updated workflow on plain OSS Conductor, confirmed empirically (caller times out waiting for terminal status)")
     public void testUpdateByRefNameSync() {
         StartWorkflowRequest request = new StartWorkflowRequest();
         request.setName(workflowName);
@@ -331,6 +334,8 @@ private void completeWorkflow(String workflowId) throws Exception {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically")
     void testSyncTargetWorkflow() throws Exception {
         String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.TARGET_WORKFLOW);
 
@@ -345,6 +350,8 @@ void testSyncTargetWorkflow() throws Exception {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically")
     void testSyncBlockingWorkflow() throws Exception {
         String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.BLOCKING_WORKFLOW);
 
@@ -359,6 +366,8 @@ void testSyncBlockingWorkflow() throws Exception {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically")
     void testSyncBlockingTask() throws Exception {
         String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.BLOCKING_TASK);
 
@@ -373,6 +382,8 @@ void testSyncBlockingTask() throws Exception {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically")
     void testSyncBlockingTaskInput() throws Exception {
         String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.BLOCKING_TASK_INPUT);
 
@@ -390,7 +401,10 @@ void testSyncBlockingTaskInput() throws Exception {
     private static final String REGION_DURABLE_ENABLED = "CONDUCTOR_REGION_DURABLE_ENABLED";
 
     @Test
-    @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true")
+    @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true",
+            disabledReason = "target server has no region replication configured")
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically")
     void testDurableTargetWorkflow() throws Exception {
         String workflowId = startComplexWorkflow(Consistency.REGION_DURABLE, ReturnStrategy.TARGET_WORKFLOW);
 
@@ -405,7 +419,10 @@ void testDurableTargetWorkflow() throws Exception {
     }
 
     @Test
-    @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true")
+    @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true",
+            disabledReason = "target server has no region replication configured")
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically")
     void testDurableBlockingWorkflow() throws Exception {
         String workflowId = startComplexWorkflow(Consistency.REGION_DURABLE, ReturnStrategy.BLOCKING_WORKFLOW);
 
@@ -420,6 +437,8 @@ void testDurableBlockingWorkflow() throws Exception {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically")
     void testDurableBlockingTask() throws Exception {
         String workflowId = startComplexWorkflow(Consistency.DURABLE, ReturnStrategy.BLOCKING_TASK);
 
@@ -434,7 +453,10 @@ void testDurableBlockingTask() throws Exception {
     }
 
     @Test
-    @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true")
+    @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true",
+            disabledReason = "target server has no region replication configured")
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically")
     void testDurableBlockingTaskInput() throws Exception {
         String workflowId = startComplexWorkflow(Consistency.REGION_DURABLE, ReturnStrategy.BLOCKING_TASK_INPUT);
 
@@ -449,6 +471,8 @@ void testDurableBlockingTaskInput() throws Exception {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically")
     void testDefaultReturnStrategy() throws Exception {
         String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.TARGET_WORKFLOW);
 
@@ -727,6 +751,8 @@ void testRequeuePendingTasksByTaskType() {
     // ==================== Search Tests ====================
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "GET /tasks/search fails on plain OSS Conductor with a Postgres persistence layer, confirmed empirically (ERROR: column \"workflow_id\" does not exist)")
     void testSearchTasks() {
         StartWorkflowRequest request = new StartWorkflowRequest();
         request.setName(workflowName);
@@ -763,6 +789,8 @@ void testSearchV2Tasks() {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "GET /tasks/search fails on plain OSS Conductor with a Postgres persistence layer, confirmed empirically (ERROR: column \"workflow_id\" does not exist)")
     void testPaginatedSearchTasks() {
         StartWorkflowRequest request = new StartWorkflowRequest();
         request.setName(workflowName);
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/TokenClientTest.java b/tests/src/test/java/io/orkes/conductor/client/http/TokenClientTest.java
index 06b2f25dc..7f6ed98db 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/TokenClientTest.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/TokenClientTest.java
@@ -16,11 +16,14 @@
 
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import io.orkes.conductor.client.model.GenerateTokenRequest;
 import io.orkes.conductor.client.model.TokenResponse;
 import io.orkes.conductor.client.util.ClientTestUtil;
 
+@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+        disabledReason = "the auth Token API (/token) is not implemented by plain OSS Conductor (which has no authentication layer), confirmed empirically (404 'No static resource api/token')")
 public class TokenClientTest {
 
     public static OrkesTokenClient tokenClient;
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/WorkflowClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/WorkflowClientTests.java
index a8d906940..c4c5e8e13 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/WorkflowClientTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/WorkflowClientTests.java
@@ -22,6 +22,7 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
 import com.netflix.conductor.common.metadata.tasks.TaskResult;
@@ -102,6 +103,8 @@ public void startWorkflow() {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "batch correlation-id search (POST /workflow/correlated/batch) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/workflow/correlated/batch')")
     public void testSearchByCorrelationIds() {
         List<String> correlationIds = new ArrayList<>();
         Set<String> workflowNames = new HashSet<>();
@@ -188,6 +191,8 @@ public void testSkipTaskFromWorkflow() throws Exception {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "POST /workflow/{workflowId}/variables is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/workflow/{id}/variables')")
     public void testUpdateVariables() {
         ConductorWorkflow<Object> workflow = new ConductorWorkflow<>(workflowExecutor);
         workflow.add(new SimpleTask("simple_task", "simple_task_ref"));
diff --git a/tests/src/test/java/io/orkes/conductor/client/http/WorkflowStateUpdateTests.java b/tests/src/test/java/io/orkes/conductor/client/http/WorkflowStateUpdateTests.java
index 286d428a5..0faad3086 100644
--- a/tests/src/test/java/io/orkes/conductor/client/http/WorkflowStateUpdateTests.java
+++ b/tests/src/test/java/io/orkes/conductor/client/http/WorkflowStateUpdateTests.java
@@ -22,6 +22,7 @@
 import org.conductoross.conductor.common.model.WorkflowRun;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 
 import com.netflix.conductor.client.exception.ConductorClientException;
 import com.netflix.conductor.common.metadata.tasks.Task;
@@ -91,6 +92,8 @@ public String startWorkflow() {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "POST /workflow/{workflowId}/state (updateWorkflow) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/workflow/{id}/state')")
     public void test() {
         String workflowId = startWorkflow();
         System.out.println(workflowId);
@@ -135,6 +138,8 @@ public void test() {
     }
 
     @Test
+    @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss",
+            disabledReason = "workflow start idempotency keys are not honored by plain OSS Conductor, confirmed empirically (RETURN_EXISTING starts a brand-new run instead of returning the original workflowId)")
     public void testIdempotency() {
         StartWorkflowRequest startWorkflowRequest = new StartWorkflowRequest();
         startWorkflowRequest.setName("sync_task_variable_updates");
diff --git a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java
index 65b13f72e..bbb77c76d 100644
--- a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java
+++ b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java
@@ -19,6 +19,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeoutException;
 
+import com.netflix.conductor.client.exception.ConductorClientException;
 import com.netflix.conductor.common.config.ObjectMapperProvider;
 import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
 import com.netflix.conductor.common.run.Workflow;
@@ -160,4 +161,26 @@ private static boolean isTerminalFailure(Workflow workflow) {
         return workflow.getStatus() == Workflow.WorkflowStatus.FAILED
                 || workflow.getStatus() == Workflow.WorkflowStatus.TERMINATED;
     }
+
+    /**
+     * Tolerates a caught exception that represents "resource doesn't exist", in either shape a
+     * Conductor server reports it in: a proper 404, or -- on plain OSS Conductor, for the
+     * endpoints where this has been empirically confirmed -- a 500 whose message contains
+     * {@code ossMessageSubstring}. Anything else is rethrown, since it isn't the "doesn't exist"
+     * case this is meant to tolerate.
+     *
+     * <p>Both shapes are accepted regardless of {@code CONDUCTOR_SERVER_TYPE}. Keying off that
+     * variable would break two ways: {@code run-integration-oss.sh --include-gated} runs against
+     * OSS with it deliberately unset, and OSS returning a correct 404 for one of these endpoints
+     * should not start failing the suite.
+     */
+    public static void tolerateNotFound(ConductorClientException e, String ossMessageSubstring) {
+        if (e.getStatus() == 404) {
+            return;
+        }
+        if (e.getStatus() == 500 && e.getMessage() != null && e.getMessage().contains(ossMessageSubstring)) {
+            return;
+        }
+        throw e;
+    }
 }
diff --git a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java
index 452da6872..391cfd926 100644
--- a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java
+++ b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java
@@ -14,9 +14,7 @@
 
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -38,11 +36,15 @@
 public class WorkflowSDKTests {
 
     @Test
-    public void testCreateWorkflow() {
+    public void testCreateWorkflow() throws Exception {
         ConductorClient client = ClientTestUtil.getClient();
 
         AnnotatedWorkerExecutor workerExecutor = new AnnotatedWorkerExecutor(new TaskClient(client), new WorkerConfiguration());
         workerExecutor.initWorkers("io.orkes.conductor.sdk");
+        // Redundant -- initWorkers() already reaches startPolling() -- but kept deliberately: this is
+        // the shape the docs and examples use, and startPolling() is now idempotent, so it must not
+        // restart the runner out from under an in-flight poll. See
+        // AnnotatedWorkerTests#initWorkersStartsASingleTaskRunner.
         workerExecutor.startPolling();
 
         WorkflowExecutor executor = new WorkflowExecutor(client, workerExecutor);
@@ -57,11 +59,14 @@ public void testCreateWorkflow() {
         CompletableFuture<Workflow> result = workflow.execute(Map.of("name", "orkes"));
         Assertions.assertNotNull(result);
         try {
-            Workflow executedWorkflow = result.get(3, TimeUnit.SECONDS);
+            // WorkflowExecutor's monitor thread polls every 100ms (see initMonitor()), so 10s is a generous margin.
+            Workflow executedWorkflow = result.get(10, TimeUnit.SECONDS);
             Assertions.assertNotNull(executedWorkflow);
             Assertions.assertEquals(Workflow.WorkflowStatus.COMPLETED, executedWorkflow.getStatus());
-        } catch (InterruptedException | ExecutionException | TimeoutException e) {
-            Assertions.fail(e.getMessage());
+        } catch (Exception e) {
+            // fail(e), not fail(e.getMessage()): a TimeoutException carries a null message, which
+            // previously surfaced in CI as a bare AssertionFailedError with nothing to diagnose.
+            Assertions.fail(e);
         }
     }