diff --git a/CHANGELOG.md b/CHANGELOG.md
index b4752eb97..aab47160e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,6 +18,8 @@ All notable changes to this project will be documented in this file.
 - Liveness for stateful runs: a `SCHEDULED` task with zero polls beyond `livenessStallSeconds` surfaces as `WorkerStallError` from the handle's wait instead of burning the full timeout
 - Swarm hand-offs: transfer tools echo the hand-off `message`; `check_transfer` is first-wins with `transfer_message` and surfaces non-winning transfers in `dropped_transfers` (with a warning) instead of silently discarding them
 - `agent-e2e` GitHub workflow runs the e2e suites as a two-server matrix: the Conductor OSS server boot JAR `3.32.0-rc.8` (Maven Central) and the released `agentspan-server-0.4.4.jar`
+- `SchedulerClient.pauseSchedule(name, reason)` — an optional pause reason, persisted as `pausedReason`
+- `WorkflowSchedule`/`SaveScheduleRequest` gain `nextRunTime` (read) and `cronSchedules` (a new `CronSchedule` list — multi-cron schedules) — additive, null-omitting serialization keeps existing save bodies unchanged when unused; `cronSchedules` is sent as-is and silently ignored on servers whose create DTO doesn't expose it
 
 ### Removed
 
@@ -26,6 +28,8 @@ All notable changes to this project will be documented in this file.
 ### Changed
 
 - `useEnvVariables(true)` (and the `new ApiClient()` no-arg constructor) no longer throws when `CONDUCTOR_SERVER_URL` is unset — it falls back to `AGENTSPAN_SERVER_URL`, then `http://localhost:8080/api`; the resolved URL is normalized to end in `/api`, and credentials gain the `AGENTSPAN_AUTH_KEY`/`AGENTSPAN_AUTH_SECRET` fallback
+- `SchedulerClient.pauseSchedule`/`resumeSchedule` now send `PUT` first (OSS Conductor's verb) and fall back to `GET` — caching that verdict for the client's lifetime — only on a `405`; any other error status propagates immediately. This makes the same client portable across OSS Conductor, Orkes Conductor (`GET`-only today), and agentspan-embedded servers with no configuration
+- `SchedulerClient.getSchedule` now throws consistently on a missing schedule
 
 ## [5.1.0]
 
diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/schedule/Schedules.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/schedule/Schedules.java
index ea1100a1d..3526ca643 100644
--- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/schedule/Schedules.java
+++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/schedule/Schedules.java
@@ -12,37 +12,30 @@
  */
 package org.conductoross.conductor.ai.schedule;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
+import java.util.function.Supplier;
 
 import org.conductoross.conductor.ai.model.AgentHandle;
 import org.conductoross.conductor.ai.model.AgentResult;
 
 import com.netflix.conductor.client.exception.ConductorClientException;
 import com.netflix.conductor.client.http.ConductorClient;
-import com.netflix.conductor.client.http.ConductorClientRequest;
-import com.netflix.conductor.client.http.ConductorClientRequest.Method;
 import com.netflix.conductor.client.http.WorkflowClient;
 import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest;
 import com.netflix.conductor.common.run.Workflow;
 
+import io.orkes.conductor.client.SchedulerClient;
 import io.orkes.conductor.client.exceptions.AgentAPIException;
-
-import com.fasterxml.jackson.core.type.TypeReference;
+import io.orkes.conductor.client.http.OrkesSchedulerClient;
+import io.orkes.conductor.client.model.SaveScheduleRequest;
+import io.orkes.conductor.client.model.WorkflowSchedule;
 
 /**
  * Lifecycle API for cron-based agent schedules. Obtained via {@code runtime.schedules()}.
  *
- * <p>All requests ride the shared native Conductor {@link ConductorClient}/ApiClient
- * (same HTTP + token-auth backend as every other client) — the scheduler CRUD via
- * {@link ConductorClientRequest}/{@link ConductorClient#execute} ({@code /api/scheduler/*},
- * for which the Conductor client ships no typed {@code SchedulerClient}), and
- * {@code runNow} via the typed {@link WorkflowClient}.
+ * <p>This class contributes only agent-scoped naming (prefix/unprefix),
+ * {@link Schedule}/{@link ScheduleInfo} DTO mapping, declarative
+ * {@link #reconcile}, and {@code runNow} sugar (via the typed {@link WorkflowClient}).
  *
  * <p>Operations are keyed by the <strong>wire name</strong> (prefixed with
  * {@code agent-}) returned by {@link #list(String)}. Use {@link Schedule} to
@@ -50,65 +43,43 @@
  */
 public class Schedules {
 
-    private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<Map<String, Object>>() {};
-    private static final TypeReference<List<Map<String, Object>>> LIST_MAP_TYPE =
-            new TypeReference<List<Map<String, Object>>>() {};
-    private static final TypeReference<List<Long>> LIST_LONG_TYPE = new TypeReference<List<Long>>() {};
-
-    private final ConductorClient client;
+    private final SchedulerClient schedulerClient;
     /** Shared native Conductor client for starting workflows (runNow). */
     private final WorkflowClient workflowClient;
 
     public Schedules(ConductorClient conductorClient) {
-        this.client = conductorClient;
+        this.schedulerClient = new OrkesSchedulerClient(conductorClient);
         this.workflowClient = new WorkflowClient(conductorClient);
     }
 
     /** Test seam: inject a {@link WorkflowClient} so {@code runNow}/{@code runNowAndWait} can be unit-tested. */
     Schedules(ConductorClient conductorClient, WorkflowClient workflowClient) {
-        this.client = conductorClient;
+        this.schedulerClient = new OrkesSchedulerClient(conductorClient);
         this.workflowClient = workflowClient;
     }
 
     // ── CRUD ────────────────────────────────────────────────────────────
 
     public void save(Schedule schedule, String agentName) {
-        Map<String, Object> body = toSaveRequest(schedule, agentName);
-        execVoid(ConductorClientRequest.builder()
-                .method(Method.POST)
-                .path("/scheduler/schedules")
-                .body(body)
-                .build());
+        SaveScheduleRequest request = toSaveRequest(schedule, agentName);
+        translate(() -> {
+            schedulerClient.saveSchedule(request);
+            return null;
+        });
     }
 
     public ScheduleInfo get(String wireName) {
-        Map<String, Object> resp = exec(
-                ConductorClientRequest.builder()
-                        .method(Method.GET)
-                        .path("/scheduler/schedules/{name}")
-                        .addPathParam("name", wireName)
-                        .build(),
-                MAP_TYPE);
-        if (resp == null || resp.isEmpty() || resp.get("name") == null) {
-            throw new ScheduleException.NotFound("Schedule '" + wireName + "' not found");
-        }
-        return fromWorkflowSchedule(resp, null);
+        WorkflowSchedule ws = translate(() -> schedulerClient.getSchedule(wireName));
+        return fromWorkflowSchedule(ws, null);
     }
 
     public List<ScheduleInfo> list(String agentName) {
-        List<Map<String, Object>> resp = exec(
-                ConductorClientRequest.builder()
-                        .method(Method.GET)
-                        .path("/scheduler/schedules")
-                        .addQueryParam("workflowName", agentName)
-                        .build(),
-                LIST_MAP_TYPE);
+        List<WorkflowSchedule> resp = translate(() -> schedulerClient.getAllSchedules(agentName));
         if (resp == null) return new ArrayList<>();
-        List<ScheduleInfo> out = new ArrayList<>();
-        for (Map<String, Object> item : resp) {
-            if (item != null) out.add(fromWorkflowSchedule(item, agentName));
-        }
-        return out;
+        return resp.stream()
+                .filter(Objects::nonNull)
+                .map(item -> fromWorkflowSchedule(item, agentName))
+                .toList();
     }
 
     public void pause(String wireName) {
@@ -116,28 +87,24 @@ public void pause(String wireName) {
     }
 
     public void pause(String wireName, String reason) {
-        ConductorClientRequest.Builder b = ConductorClientRequest.builder()
-                .method(Method.PUT)
-                .path("/scheduler/schedules/{name}/pause")
-                .addPathParam("name", wireName);
-        if (reason != null) b.addQueryParam("reason", reason);
-        execVoid(b.build());
+        translate(() -> {
+            schedulerClient.pauseSchedule(wireName, reason);
+            return null;
+        });
     }
 
     public void resume(String wireName) {
-        execVoid(ConductorClientRequest.builder()
-                .method(Method.PUT)
-                .path("/scheduler/schedules/{name}/resume")
-                .addPathParam("name", wireName)
-                .build());
+        translate(() -> {
+            schedulerClient.resumeSchedule(wireName);
+            return null;
+        });
     }
 
     public void delete(String wireName) {
-        execVoid(ConductorClientRequest.builder()
-                .method(Method.DELETE)
-                .path("/scheduler/schedules/{name}")
-                .addPathParam("name", wireName)
-                .build());
+        translate(() -> {
+            schedulerClient.deleteSchedule(wireName);
+            return null;
+        });
     }
 
     /**
@@ -241,14 +208,7 @@ static boolean isTerminal(Workflow wf) {
     }
 
     public List<Long> previewNext(String cron, int n) {
-        List<Long> resp = exec(
-                ConductorClientRequest.builder()
-                        .method(Method.GET)
-                        .path("/scheduler/nextFewSchedules")
-                        .addQueryParam("cronExpression", cron)
-                        .addQueryParam("limit", Integer.valueOf(n))
-                        .build(),
-                LIST_LONG_TYPE);
+        List<Long> resp = translate(() -> schedulerClient.getNextFewSchedules(cron, null, null, n));
         return resp != null ? resp : new ArrayList<>();
     }
 
@@ -304,68 +264,52 @@ static void checkUniqueNames(List<Schedule> schedules) {
         }
     }
 
-    static Map<String, Object> toSaveRequest(Schedule s, String agentName) {
-        Map<String, Object> swr = new LinkedHashMap<>();
-        swr.put("name", agentName);
-        swr.put("input", s.getInput() == null ? new LinkedHashMap<>() : new LinkedHashMap<>(s.getInput()));
-
-        Map<String, Object> req = new LinkedHashMap<>();
-        req.put("name", prefix(agentName, s.getName()));
-        req.put("cronExpression", s.getCron());
-        req.put("zoneId", s.getTimezone());
-        req.put("runCatchupScheduleInstances", s.isCatchup());
-        req.put("paused", s.isPaused());
-        if (s.getStartAt() != null) req.put("scheduleStartTime", s.getStartAt());
-        if (s.getEndAt() != null) req.put("scheduleEndTime", s.getEndAt());
-        if (s.getDescription() != null) req.put("description", s.getDescription());
-        req.put("startWorkflowRequest", swr);
-        return req;
+    static SaveScheduleRequest toSaveRequest(Schedule s, String agentName) {
+        StartWorkflowRequest startWorkflowRequest = new StartWorkflowRequest()
+                .withName(agentName)
+                .withInput(s.getInput() == null ? Map.of() : new LinkedHashMap<>(s.getInput()));
+        return new SaveScheduleRequest()
+                .name(prefix(agentName, s.getName()))
+                .cronExpression(s.getCron())
+                .zoneId(s.getTimezone())
+                .runCatchupScheduleInstances(s.isCatchup())
+                .paused(s.isPaused())
+                .scheduleStartTime(s.getStartAt())
+                .scheduleEndTime(s.getEndAt())
+                .startWorkflowRequest(startWorkflowRequest)
+                .description(s.getDescription());
     }
 
-    @SuppressWarnings("unchecked")
-    static ScheduleInfo fromWorkflowSchedule(Map<String, Object> ws, String agentHint) {
-        Map<String, Object> swr = (Map<String, Object>) ws.getOrDefault("startWorkflowRequest", new HashMap<>());
-        String wireName = (String) ws.getOrDefault("name", "");
-        String swrName = (String) swr.getOrDefault("name", "");
+    static ScheduleInfo fromWorkflowSchedule(WorkflowSchedule ws, String agentHint) {
+        StartWorkflowRequest swr = ws.getStartWorkflowRequest();
+        String wireName = ws.getName() != null ? ws.getName() : "";
+        String swrName = swr != null && swr.getName() != null ? swr.getName() : "";
         String agent = agentHint != null ? agentHint : (swrName.isEmpty() ? "" : swrName);
 
         return new ScheduleInfo(
                 wireName,
                 unprefix(agent, wireName),
                 swrName,
-                (String) ws.getOrDefault("cronExpression", ""),
-                (String) ws.getOrDefault("zoneId", "UTC"),
-                (Map<String, Object>) swr.getOrDefault("input", new HashMap<>()),
-                Boolean.TRUE.equals(ws.get("paused")),
-                (String) ws.get("pausedReason"),
-                Boolean.TRUE.equals(ws.get("runCatchupScheduleInstances")),
-                longOrNull(ws.get("scheduleStartTime")),
-                longOrNull(ws.get("scheduleEndTime")),
-                (String) ws.get("description"),
-                longOrNull(ws.get("nextRunTime")),
-                longOrNull(ws.get("createTime")),
-                longOrNull(ws.get("updatedTime")),
-                (String) ws.get("createdBy"),
-                (String) ws.get("updatedBy"));
-    }
-
-    private static Long longOrNull(Object o) {
-        return o instanceof Number ? ((Number) o).longValue() : null;
-    }
-
-    /** Execute a scheduler request returning a typed body via the native Conductor client. */
-    private <T> T exec(ConductorClientRequest req, TypeReference<T> type) {
-        try {
-            return client.execute(req, type).getData();
-        } catch (ConductorClientException e) {
-            throw mapException(e);
-        }
+                ws.getCronExpression() != null ? ws.getCronExpression() : "",
+                ws.getZoneId() != null ? ws.getZoneId() : "UTC",
+                swr != null && swr.getInput() != null ? swr.getInput() : new HashMap<>(),
+                Boolean.TRUE.equals(ws.isPaused()),
+                ws.getPausedReason(),
+                Boolean.TRUE.equals(ws.isRunCatchupScheduleInstances()),
+                ws.getScheduleStartTime(),
+                ws.getScheduleEndTime(),
+                ws.getDescription(),
+                ws.getNextRunTime(),
+                ws.getCreateTime(),
+                ws.getUpdatedTime(),
+                ws.getCreatedBy(),
+                ws.getUpdatedBy());
     }
 
-    /** Execute a scheduler request that returns no body. */
-    private void execVoid(ConductorClientRequest req) {
+    /** Run a {@link SchedulerClient} call, translating transport errors to the scheduler's typed exceptions. */
+    private static <T> T translate(Supplier<T> call) {
         try {
-            client.execute(req);
+            return call.get();
         } catch (ConductorClientException e) {
             throw mapException(e);
         }
diff --git a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/schedule/ScheduleTest.java b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/schedule/ScheduleTest.java
index eb03e4bcd..905b2b83f 100644
--- a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/schedule/ScheduleTest.java
+++ b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/schedule/ScheduleTest.java
@@ -19,6 +19,11 @@
 
 import org.junit.jupiter.api.Test;
 
+import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest;
+
+import io.orkes.conductor.client.model.SaveScheduleRequest;
+import io.orkes.conductor.client.model.WorkflowSchedule;
+
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -121,16 +126,15 @@ void agentNameWithHyphen() {
     @Test
     void toSaveRequestMinimal() {
         Schedule s = Schedule.builder().name("daily").cron("0 0 9 * * ?").build();
-        Map<String, Object> req = Schedules.toSaveRequest(s, "digest");
-        assertEquals("digest-daily", req.get("name"));
-        assertEquals("0 0 9 * * ?", req.get("cronExpression"));
-        assertEquals("UTC", req.get("zoneId"));
-        assertEquals(false, req.get("paused"));
-        assertEquals(false, req.get("runCatchupScheduleInstances"));
-        @SuppressWarnings("unchecked")
-        Map<String, Object> swr = (Map<String, Object>) req.get("startWorkflowRequest");
-        assertEquals("digest", swr.get("name"));
-        assertTrue(((Map<?, ?>) swr.get("input")).isEmpty());
+        SaveScheduleRequest req = Schedules.toSaveRequest(s, "digest");
+        assertEquals("digest-daily", req.getName());
+        assertEquals("0 0 9 * * ?", req.getCronExpression());
+        assertEquals("UTC", req.getZoneId());
+        assertEquals(false, req.isPaused());
+        assertEquals(false, req.isRunCatchupScheduleInstances());
+        StartWorkflowRequest swr = req.getStartWorkflowRequest();
+        assertEquals("digest", swr.getName());
+        assertTrue(swr.getInput().isEmpty());
     }
 
     @Test
@@ -149,13 +153,13 @@ void toSaveRequestFull() {
                 .endAt(2000L)
                 .description("desc")
                 .build();
-        Map<String, Object> req = Schedules.toSaveRequest(s, "digest");
-        assertEquals("America/Los_Angeles", req.get("zoneId"));
-        assertEquals(true, req.get("paused"));
-        assertEquals(true, req.get("runCatchupScheduleInstances"));
-        assertEquals(1000L, req.get("scheduleStartTime"));
-        assertEquals(2000L, req.get("scheduleEndTime"));
-        assertEquals("desc", req.get("description"));
+        SaveScheduleRequest req = Schedules.toSaveRequest(s, "digest");
+        assertEquals("America/Los_Angeles", req.getZoneId());
+        assertEquals(true, req.isPaused());
+        assertEquals(true, req.isRunCatchupScheduleInstances());
+        assertEquals(1000L, req.getScheduleStartTime());
+        assertEquals(2000L, req.getScheduleEndTime());
+        assertEquals("desc", req.getDescription());
     }
 
     @Test
@@ -164,29 +168,29 @@ void inputCopiedNotShared() {
         original.put("a", 1);
         Schedule s =
                 Schedule.builder().name("x").cron("* * * * * ?").input(original).build();
-        Map<String, Object> req = Schedules.toSaveRequest(s, "agent");
-        @SuppressWarnings("unchecked")
-        Map<String, Object> swrInput =
-                (Map<String, Object>) ((Map<String, Object>) req.get("startWorkflowRequest")).get("input");
+        SaveScheduleRequest req = Schedules.toSaveRequest(s, "agent");
+        Map<String, Object> swrInput = req.getStartWorkflowRequest().getInput();
         swrInput.put("mutated", true);
         assertNull(original.get("mutated"));
     }
 
     @Test
     void fromWorkflowScheduleBasic() {
-        Map<String, Object> ws = new LinkedHashMap<>();
-        ws.put("name", "digest-daily");
-        ws.put("cronExpression", "0 0 9 * * ?");
-        ws.put("zoneId", "UTC");
-        ws.put("paused", false);
-        Map<String, Object> swr = new LinkedHashMap<>();
-        swr.put("name", "digest");
         Map<String, Object> input = new LinkedHashMap<>();
         input.put("c", "#eng");
-        swr.put("input", input);
-        ws.put("startWorkflowRequest", swr);
-        ws.put("createTime", 111L);
-        ws.put("createdBy", "alice");
+        StartWorkflowRequest swr = new StartWorkflowRequest();
+        swr.setName("digest");
+        swr.setInput(input);
+
+        WorkflowSchedule ws = WorkflowSchedule.builder()
+                .name("digest-daily")
+                .cronExpression("0 0 9 * * ?")
+                .zoneId("UTC")
+                .paused(false)
+                .startWorkflowRequest(swr)
+                .createTime(111L)
+                .createdBy("alice")
+                .build();
 
         ScheduleInfo info = Schedules.fromWorkflowSchedule(ws, "digest");
         assertEquals("digest-daily", info.getName());
@@ -201,11 +205,14 @@ void fromWorkflowScheduleBasic() {
 
     @Test
     void fromWorkflowScheduleDerivesAgentWhenOmitted() {
-        Map<String, Object> ws = new LinkedHashMap<>();
-        ws.put("name", "digest-daily");
-        Map<String, Object> swr = new LinkedHashMap<>();
-        swr.put("name", "digest");
-        ws.put("startWorkflowRequest", swr);
+        StartWorkflowRequest swr = new StartWorkflowRequest();
+        swr.setName("digest");
+
+        WorkflowSchedule ws = WorkflowSchedule.builder()
+                .name("digest-daily")
+                .startWorkflowRequest(swr)
+                .build();
+
         ScheduleInfo info = Schedules.fromWorkflowSchedule(ws, null);
         assertEquals("digest", info.getAgent());
         assertEquals("daily", info.getShortName());
diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/SchedulerClient.java b/conductor-client/src/main/java/io/orkes/conductor/client/SchedulerClient.java
index 878176fd6..94f38d91d 100644
--- a/conductor-client/src/main/java/io/orkes/conductor/client/SchedulerClient.java
+++ b/conductor-client/src/main/java/io/orkes/conductor/client/SchedulerClient.java
@@ -92,6 +92,21 @@ public interface SchedulerClient {
      */
     void pauseSchedule(String name);
 
+    /**
+     * Pauses a specific schedule, recording an optional reason.
+     * <p>
+     * The reason is persisted as {@code pausedReason} where the server supports it (e.g. OSS
+     * Conductor); it is silently ignored otherwise (e.g. Orkes Conductor today). The default
+     * implementation ignores the reason and delegates to {@link #pauseSchedule(String)} so
+     * existing {@link SchedulerClient} implementations remain source- and binary-compatible.
+     *
+     * @param name   the name of the schedule to pause
+     * @param reason optional reason for the pause, or {@code null}
+     */
+    default void pauseSchedule(String name, String reason) {
+        pauseSchedule(name);
+    }
+
     /**
      * Resumes a paused schedule, allowing new workflow executions.
      *
diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/http/OrkesSchedulerClient.java b/conductor-client/src/main/java/io/orkes/conductor/client/http/OrkesSchedulerClient.java
index bacbbdaaf..dfc7de1dd 100644
--- a/conductor-client/src/main/java/io/orkes/conductor/client/http/OrkesSchedulerClient.java
+++ b/conductor-client/src/main/java/io/orkes/conductor/client/http/OrkesSchedulerClient.java
@@ -67,6 +67,11 @@ public void pauseSchedule(String name) {
         schedulerResource.pauseSchedule(name);
     }
 
+    @Override
+    public void pauseSchedule(String name, String reason) {
+        schedulerResource.pauseSchedule(name, reason);
+    }
+
     @Override
     public void requeueAllExecutionRecords() {
         schedulerResource.requeueAllExecutionRecords();
diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java b/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java
index 2506c0df1..ca595cca8 100644
--- a/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java
+++ b/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java
@@ -15,6 +15,7 @@
 import java.util.List;
 import java.util.Map;
 
+import com.netflix.conductor.client.exception.ConductorClientException;
 import com.netflix.conductor.client.http.ConductorClient;
 import com.netflix.conductor.client.http.ConductorClientRequest;
 import com.netflix.conductor.client.http.ConductorClientRequest.Method;
@@ -32,6 +33,13 @@ public class SchedulerResource {
 
     private final ConductorClient client;
 
+    /**
+     * Once a server answers {@code 405} to a PUT pause/resume request (OSS Conductor's original
+     * verb; Orkes Conductor historically accepted only {@code GET}), fall back to {@code GET} for
+     * the remaining lifetime of this instance rather than probing on every call.
+     */
+    private volatile boolean useGetForScheduleStateChange = false;
+
     public SchedulerResource(ConductorClient client) {
         this.client = client;
     }
@@ -88,7 +96,14 @@ public WorkflowSchedule getSchedule(String name) {
         ConductorClientResponse<WorkflowSchedule> resp = client.execute(request, new TypeReference<>() {
         });
 
-        return resp.getData();
+        WorkflowSchedule schedule = resp.getData();
+        if (schedule == null || schedule.getName() == null) {
+            // OSS Conductor returns 200 with an empty/null body on a miss (Orkes returns 404,
+            // which already surfaces as a ConductorClientException below); normalize both server
+            // families to the same not-found contract.
+            throw new ConductorClientException(404, "Schedule '" + name + "' not found");
+        }
+        return schedule;
     }
 
     public Map<String, Object> pauseAllSchedules() {
@@ -104,13 +119,15 @@ public Map<String, Object> pauseAllSchedules() {
     }
 
     public void pauseSchedule(String name) {
-        ConductorClientRequest request = ConductorClientRequest.builder()
-                .method(Method.GET)
-                .path("/scheduler/schedules/{name}/pause")
-                .addPathParam("name", name)
-                .build();
+        pauseSchedule(name, null);
+    }
 
-        client.execute(request);
+    /**
+     * Pauses a schedule, recording an optional reason (persisted as {@code pausedReason} where
+     * the server supports it; silently ignored otherwise — e.g. Orkes Conductor today).
+     */
+    public void pauseSchedule(String name, String reason) {
+        executeStateChange("/scheduler/schedules/{name}/pause", name, reason);
     }
 
     public Map<String, Object> requeueAllExecutionRecords() {
@@ -138,13 +155,40 @@ public Map<String, Object> resumeAllSchedules() {
     }
 
     public void resumeSchedule(String name) {
-        ConductorClientRequest request = ConductorClientRequest.builder()
-                .method(Method.GET)
-                .path("/scheduler/schedules/{name}/resume")
-                .addPathParam("name", name)
-                .build();
+        executeStateChange("/scheduler/schedules/{name}/resume", name, null);
+    }
 
-        client.execute(request);
+    /**
+     * Sends a schedule state-change (pause/resume) as {@code PUT} first; if the server answers
+     * {@code 405 Method Not Allowed} (Orkes Conductor, historically), falls back to {@code GET}
+     * and remembers that verdict for the rest of this instance's lifetime. Any other error
+     * status (403, 404, 5xx, ...) propagates immediately — only a routing mismatch (405)
+     * triggers the fallback.
+     */
+    private void executeStateChange(String path, String name, String reason) {
+        if (!useGetForScheduleStateChange) {
+            try {
+                client.execute(stateChangeRequest(Method.PUT, path, name, reason));
+                return;
+            } catch (ConductorClientException e) {
+                if (e.getStatus() != 405) {
+                    throw e;
+                }
+                useGetForScheduleStateChange = true;
+            }
+        }
+        client.execute(stateChangeRequest(Method.GET, path, name, reason));
+    }
+
+    private ConductorClientRequest stateChangeRequest(Method method, String path, String name, String reason) {
+        ConductorClientRequest.Builder builder = ConductorClientRequest.builder()
+                .method(method)
+                .path(path)
+                .addPathParam("name", name);
+        if (reason != null) {
+            builder.addQueryParam("reason", reason);
+        }
+        return builder.build();
     }
 
     public void saveSchedule(SaveScheduleRequest saveScheduleRequest) {
diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/model/CronSchedule.java b/conductor-client/src/main/java/io/orkes/conductor/client/model/CronSchedule.java
new file mode 100644
index 000000000..8dbc63689
--- /dev/null
+++ b/conductor-client/src/main/java/io/orkes/conductor/client/model/CronSchedule.java
@@ -0,0 +1,26 @@
+/*
+ * 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 io.orkes.conductor.client.model;
+
+import lombok.*;
+
+/** One entry of a multi-cron schedule: its own cron expression and timezone. */
+@Data
+@NoArgsConstructor
+@Builder
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public class CronSchedule {
+
+    private String cronExpression;
+    private String zoneId;
+}
diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/model/SaveScheduleRequest.java b/conductor-client/src/main/java/io/orkes/conductor/client/model/SaveScheduleRequest.java
index c1190ce99..f8c602435 100644
--- a/conductor-client/src/main/java/io/orkes/conductor/client/model/SaveScheduleRequest.java
+++ b/conductor-client/src/main/java/io/orkes/conductor/client/model/SaveScheduleRequest.java
@@ -12,6 +12,8 @@
  */
 package io.orkes.conductor.client.model;
 
+import java.util.List;
+
 import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest;
 
 import lombok.*;
@@ -46,6 +48,8 @@ public class SaveScheduleRequest {
 
     private String description;
 
+    private List<CronSchedule> cronSchedules;
+
     public SaveScheduleRequest createdBy(String createdBy) {
         this.createdBy = createdBy;
         return this;
@@ -104,4 +108,13 @@ public SaveScheduleRequest zoneId(String zoneId) {
         return this;
     }
 
+    public SaveScheduleRequest cronSchedules(List<CronSchedule> cronSchedules) {
+        this.cronSchedules = cronSchedules;
+        return this;
+    }
+
+    public SaveScheduleRequest description(String description) {
+        this.description = description;
+        return this;
+    }
 }
\ No newline at end of file
diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/model/WorkflowSchedule.java b/conductor-client/src/main/java/io/orkes/conductor/client/model/WorkflowSchedule.java
index f29e1c958..dc36e13fa 100644
--- a/conductor-client/src/main/java/io/orkes/conductor/client/model/WorkflowSchedule.java
+++ b/conductor-client/src/main/java/io/orkes/conductor/client/model/WorkflowSchedule.java
@@ -56,6 +56,11 @@ public class WorkflowSchedule {
 
     private String description;
 
+    /** Server-owned next-scheduled execution time, in epoch millis. Can be null if unavailable. **/
+    private Long nextRunTime;
+
+    private List<CronSchedule> cronSchedules;
+
     public WorkflowSchedule createTime(Long createTime) {
         this.createTime = createTime;
         return this;
@@ -123,4 +128,14 @@ public WorkflowSchedule zoneId(String zoneId) {
         this.zoneId = zoneId;
         return this;
     }
+
+    public WorkflowSchedule nextRunTime(Long nextRunTime) {
+        this.nextRunTime = nextRunTime;
+        return this;
+    }
+
+    public WorkflowSchedule cronSchedules(List<CronSchedule> cronSchedules) {
+        this.cronSchedules = cronSchedules;
+        return this;
+    }
 }
\ No newline at end of file
diff --git a/conductor-client/src/test/java/io/orkes/conductor/client/http/SchedulerResourceTest.java b/conductor-client/src/test/java/io/orkes/conductor/client/http/SchedulerResourceTest.java
new file mode 100644
index 000000000..bf42e4e23
--- /dev/null
+++ b/conductor-client/src/test/java/io/orkes/conductor/client/http/SchedulerResourceTest.java
@@ -0,0 +1,348 @@
+/*
+ * 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 io.orkes.conductor.client.http;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import com.netflix.conductor.client.exception.ConductorClientException;
+import com.netflix.conductor.client.http.ConductorClient;
+
+import io.orkes.conductor.client.SchedulerClient;
+import io.orkes.conductor.client.model.CronSchedule;
+import io.orkes.conductor.client.model.SaveScheduleRequest;
+import io.orkes.conductor.client.model.SearchResultWorkflowScheduleExecution;
+import io.orkes.conductor.client.model.TagObject;
+import io.orkes.conductor.client.model.WorkflowSchedule;
+
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Covers the two behaviour changes to {@link SchedulerResource} that make it portable across
+ * OSS Conductor (PUT pause/resume, 200-empty get-miss) and Orkes Conductor (GET pause/resume,
+ * 404 get-miss): the cached PUT-then-405-fallback state machine, and the normalized get-miss
+ * contract.
+ */
+class SchedulerResourceTest {
+
+    private MockWebServer server;
+    private SchedulerResource resource;
+
+    @BeforeEach
+    void setUp() throws IOException {
+        server = new MockWebServer();
+        server.start();
+        resource = new SchedulerResource(new ConductorClient(server.url("/api").toString()));
+    }
+
+    @AfterEach
+    void tearDown() throws IOException {
+        server.shutdown();
+    }
+
+    private static MockResponse json(String body) {
+        return new MockResponse().setHeader("Content-Type", "application/json").setBody(body);
+    }
+
+    private RecordedRequest takeRequest() throws InterruptedException {
+        RecordedRequest request = server.takeRequest(5, TimeUnit.SECONDS);
+        assertNotNull(request, "expected a request to reach the stub server");
+        return request;
+    }
+
+    // ── pause/resume verb fallback ──────────────────────────────────────
+
+    @Test
+    void pauseSendsPutFirst() throws InterruptedException {
+        server.enqueue(json("{}"));
+
+        resource.pauseSchedule("s1");
+
+        RecordedRequest request = takeRequest();
+        assertEquals("PUT", request.getMethod());
+        assertTrue(request.getPath().endsWith("/scheduler/schedules/s1/pause"));
+    }
+
+    @Test
+    void pauseFallsBackToGetOn405AndCachesVerdictAcrossPauseAndResume() throws InterruptedException {
+        server.enqueue(new MockResponse().setResponseCode(405));
+        server.enqueue(json("{}"));
+
+        resource.pauseSchedule("s1");
+
+        RecordedRequest first = takeRequest();
+        assertEquals("PUT", first.getMethod());
+        RecordedRequest second = takeRequest();
+        assertEquals("GET", second.getMethod());
+        assertTrue(second.getPath().endsWith("/scheduler/schedules/s1/pause"));
+
+        // Verdict is cached: resume (a different method) goes straight to GET, no PUT probe.
+        server.enqueue(json("{}"));
+        resource.resumeSchedule("s1");
+
+        RecordedRequest third = takeRequest();
+        assertEquals("GET", third.getMethod());
+        assertTrue(third.getPath().endsWith("/scheduler/schedules/s1/resume"));
+    }
+
+    @Test
+    void pauseRethrowsOn403WithoutFallingBackToGet() throws InterruptedException {
+        server.enqueue(new MockResponse().setResponseCode(403));
+
+        ConductorClientException ex = assertThrows(ConductorClientException.class, () -> resource.pauseSchedule("s1"));
+        assertEquals(403, ex.getStatus());
+
+        assertEquals(1, server.getRequestCount());
+    }
+
+    @Test
+    void pauseRethrowsOn404WithoutFallingBackToGet() throws InterruptedException {
+        server.enqueue(new MockResponse().setResponseCode(404));
+
+        ConductorClientException ex = assertThrows(ConductorClientException.class, () -> resource.pauseSchedule("s1"));
+        assertEquals(404, ex.getStatus());
+
+        assertEquals(1, server.getRequestCount());
+    }
+
+    @Test
+    void resumeSharesTheSameCachedFallbackVerdictAsPause() throws InterruptedException {
+        server.enqueue(new MockResponse().setResponseCode(405));
+        server.enqueue(json("{}"));
+
+        resource.resumeSchedule("s1");
+        takeRequest(); // PUT (405)
+        RecordedRequest fallback = takeRequest();
+        assertEquals("GET", fallback.getMethod());
+
+        server.enqueue(json("{}"));
+        resource.pauseSchedule("s2");
+
+        RecordedRequest thirdCall = takeRequest();
+        assertEquals("GET", thirdCall.getMethod());
+    }
+
+    // ── reason query param ───────────────────────────────────────────────
+
+    @Test
+    void pauseSendsReasonQueryParamWhenProvided() throws InterruptedException {
+        server.enqueue(json("{}"));
+
+        resource.pauseSchedule("s1", "maintenance window");
+
+        RecordedRequest request = takeRequest();
+        assertTrue(request.getPath().contains("reason=maintenance"));
+    }
+
+    @Test
+    void pauseOmitsReasonQueryParamWhenNull() throws InterruptedException {
+        server.enqueue(json("{}"));
+
+        resource.pauseSchedule("s1");
+
+        RecordedRequest request = takeRequest();
+        assertFalse(request.getPath().contains("reason"));
+    }
+
+    @Test
+    void reasonSurvivesTheGetFallback() throws InterruptedException {
+        server.enqueue(new MockResponse().setResponseCode(405));
+        server.enqueue(json("{}"));
+
+        resource.pauseSchedule("s1", "maintenance window");
+
+        takeRequest(); // PUT (405)
+        RecordedRequest fallback = takeRequest();
+        assertEquals("GET", fallback.getMethod());
+        assertTrue(fallback.getPath().contains("reason=maintenance"));
+    }
+
+    // ── get-miss contract ────────────────────────────────────────────────
+
+    @Test
+    void getScheduleThrowsOn404() {
+        server.enqueue(new MockResponse().setResponseCode(404));
+
+        ConductorClientException ex = assertThrows(ConductorClientException.class, () -> resource.getSchedule("missing"));
+        assertEquals(404, ex.getStatus());
+    }
+
+    @Test
+    void getScheduleThrowsOnOssStyleEmptyBody() {
+        server.enqueue(json(""));
+
+        assertThrows(ConductorClientException.class, () -> resource.getSchedule("missing"));
+    }
+
+    @Test
+    void getScheduleThrowsOnEmptyJsonObject() {
+        server.enqueue(json("{}"));
+
+        assertThrows(ConductorClientException.class, () -> resource.getSchedule("missing"));
+    }
+
+    @Test
+    void getScheduleReturnsScheduleOnHitIncludingNextRunTimeAndCronSchedules() {
+        server.enqueue(json("{"
+                + "\"name\":\"s1\","
+                + "\"cronExpression\":\"0 */5 * * * *\","
+                + "\"nextRunTime\":1700000000000,"
+                + "\"cronSchedules\":[{\"cronExpression\":\"0 0 * * * *\",\"zoneId\":\"UTC\"}]"
+                + "}"));
+
+        WorkflowSchedule schedule = resource.getSchedule("s1");
+
+        assertEquals("s1", schedule.getName());
+        assertEquals(1700000000000L, schedule.getNextRunTime());
+        assertEquals(1, schedule.getCronSchedules().size());
+        assertEquals("0 0 * * * *", schedule.getCronSchedules().get(0).getCronExpression());
+    }
+
+    @Test
+    void getScheduleToleratesMissingNextRunTime() {
+        server.enqueue(json("{\"name\":\"s1\",\"cronExpression\":\"0 */5 * * * *\"}"));
+
+        WorkflowSchedule schedule = resource.getSchedule("s1");
+
+        assertEquals("s1", schedule.getName());
+        assertEquals(null, schedule.getNextRunTime());
+    }
+
+    // ── save body wire shape ─────────────────────────────────────────────
+
+    @Test
+    void saveScheduleOmitsCronSchedulesWhenNotSet() throws InterruptedException {
+        server.enqueue(json("{}"));
+
+        SaveScheduleRequest request = SaveScheduleRequest.builder()
+                .name("s1")
+                .cronExpression("0 */5 * * * *")
+                .build();
+        resource.saveSchedule(request);
+
+        RecordedRequest recorded = takeRequest();
+        assertFalse(recorded.getBody().readUtf8().contains("cronSchedules"));
+    }
+
+    @Test
+    void saveScheduleSendsCronSchedulesWhenSet() throws InterruptedException {
+        server.enqueue(json("{}"));
+
+        SaveScheduleRequest request = SaveScheduleRequest.builder()
+                .name("s1")
+                .cronSchedules(List.of(CronSchedule.builder().cronExpression("0 0 * * * *").zoneId("UTC").build()))
+                .build();
+        resource.saveSchedule(request);
+
+        RecordedRequest recorded = takeRequest();
+        String body = recorded.getBody().readUtf8();
+        assertTrue(body.contains("\"cronSchedules\""));
+        assertTrue(body.contains("0 0 * * * *"));
+    }
+
+    // ── default-method backward compatibility ───────────────────────────
+
+    /** A minimal {@link SchedulerClient} implementation predating the reason overload — proves
+     * the interface addition doesn't break existing implementers. */
+    private static class LegacySchedulerClient implements SchedulerClient {
+        String lastPausedName;
+
+        @Override
+        public void pauseSchedule(String name) {
+            lastPausedName = name;
+        }
+
+        @Override
+        public void saveSchedule(SaveScheduleRequest saveScheduleRequest) {}
+
+        @Override
+        public WorkflowSchedule getSchedule(String name) {
+            return null;
+        }
+
+        @Override
+        public List<WorkflowSchedule> getAllSchedules(String workflowName) {
+            return null;
+        }
+
+        @Override
+        public void deleteSchedule(String name) {}
+
+        @Override
+        public void resumeSchedule(String name) {}
+
+        @Override
+        public void pauseAllSchedules() {}
+
+        @Override
+        public void resumeAllSchedules() {}
+
+        @Override
+        public com.netflix.conductor.common.model.BulkResponse pauseSchedulers(List<String> schedulerIds) {
+            return null;
+        }
+
+        @Override
+        public com.netflix.conductor.common.model.BulkResponse resumeSchedulers(List<String> schedulerIds) {
+            return null;
+        }
+
+        @Override
+        public List<Long> getNextFewSchedules(String cronExpression, Long scheduleStartTime, Long scheduleEndTime,
+                Integer limit) {
+            return null;
+        }
+
+        @Override
+        public void requeueAllExecutionRecords() {}
+
+        @Override
+        public SearchResultWorkflowScheduleExecution search(Integer start, Integer size, String sort,
+                String freeText, String query) {
+            return null;
+        }
+
+        @Override
+        public void setSchedulerTags(List<TagObject> body, String name) {}
+
+        @Override
+        public void deleteSchedulerTags(List<TagObject> body, String name) {}
+
+        @Override
+        public List<TagObject> getSchedulerTags(String name) {
+            return null;
+        }
+    }
+
+    @Test
+    void defaultReasonOverloadDelegatesToOneArgOverloadWhenNotImplemented() {
+        LegacySchedulerClient client = new LegacySchedulerClient();
+
+        client.pauseSchedule("s1", "some reason");
+
+        assertEquals("s1", client.lastPausedName);
+    }
+}
diff --git a/conductor-client/src/test/resources/ser_deser_json_string.json b/conductor-client/src/test/resources/ser_deser_json_string.json
index fe566189f..c7aa59e6f 100644
--- a/conductor-client/src/test/resources/ser_deser_json_string.json
+++ b/conductor-client/src/test/resources/ser_deser_json_string.json
@@ -222,12 +222,23 @@
         "zoneId": "sample_zoneId",
         "runCatchupScheduleInstances": true,
         "scheduleStartTime": 123,
-        "scheduleEndTime": 123
+        "scheduleEndTime": 123,
+        "cronSchedules": [
+          "${CronSchedule}"
+        ]
       },
       "dependencies": [
-        "StartWorkflowRequest"
+        "StartWorkflowRequest",
+        "CronSchedule"
       ]
     },
+    "CronSchedule": {
+      "content": {
+        "cronExpression": "sample_cronExpression",
+        "zoneId": "sample_zoneId"
+      },
+      "dependencies": []
+    },
     "GrantedAccess": {
       "content": {
         "access": [
@@ -1993,11 +2004,16 @@
         "createdBy": "sample_createdBy",
         "name": "sample_name",
         "zoneId": "sample_zoneId",
-        "scheduleEndTime": 123
+        "scheduleEndTime": 123,
+        "nextRunTime": 123,
+        "cronSchedules": [
+          "${CronSchedule}"
+        ]
       },
       "dependencies": [
         "Tag",
-        "StartWorkflowRequest"
+        "StartWorkflowRequest",
+        "CronSchedule"
       ]
     },
     "T": {