S3: SchedulerClient is the get/save/list source of truth; AgentScheduleClient becomes an inert deprecated shim

The scoped breaking change (user-accepted): schedule client objects no longer
expose the mapped get/save/list_for_agent wrappers — the native methods are the
one API for reads/writes/lists:

  get(wire) -> ScheduleInfo        =>  get_schedule(wire) -> WorkflowSchedule|None
  save(schedule, agent)            =>  save_schedule(SaveScheduleRequest)
  list_for_agent(agent)            =>  get_all_schedules(workflow_name=agent)

Everything else is byte-frozen: the six lifecycle methods, the module-level
schedules.* API (all eight functions keep their signatures and ScheduleInfo
returns — internals rewired onto _get_info/_list_infos over the raw methods),
symbols/constructors/accessors, and all examples.

- AgentScheduleClient rewritten as a pure delegation shim: subclasses
  SchedulerClient, 14 endpoint delegations to the wrapped client,
  _start_workflow via the wrapped workflow client, lifecycle methods inherited,
  zero logic of its own. The raw requests.put pause/resume workaround dies here
  (S1's fixed transport + 405 fallback covers both server dialects). DEPRECATED;
  removal planned for the next major release (get_agent_schedule_client too).
- tests/unit/ai/test_schedule.py: byte-identical except the run_now-wait
  fixture's read stub (get -> get_schedule returning a typed WorkflowSchedule) —
  direct collateral of the accepted break; all 32 tests pass, now exercising the
  shipped SchedulerClient code paths through the shim.
- e2e suite21 (13 call sites) and suite24 (3 call sites) migrate to the
  source-of-truth methods with WorkflowSchedule-field assertions; fixtures and
  constructions unchanged.
- New guard: no schedule client may grow get/save/list_for_agent back.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Kowser

e2e/test_suite21_scheduling.py

@@ -31,9 +31,9 @@
 from conductor.ai.agents.schedule import (
     Schedule,
     ScheduleNameConflict,
-    ScheduleNotFound,
 )
 from conductor.ai.agents.schedule.client import ScheduleClient
+from conductor.client.ai.schedule import _from_workflow_schedule
 
 pytestmark = [pytest.mark.e2e]
 
@@ -152,13 +152,13 @@ def test_creates_schedules(self, schedule_client, agent_name):
                 Schedule(name="weekly", cron="0 0 9 * * MON"),
             ],
         )
-        infos = schedule_client.list_for_agent(agent_name)
-        by_short = {i.short_name: i for i in infos}
-        assert set(by_short) == {"daily", "weekly"}
-        assert by_short["daily"].name == f"{agent_name}-daily"
-        assert by_short["daily"].cron == "0 0 9 * * ?"
-        assert by_short["daily"].input == {"k": 1}
-        assert by_short["daily"].agent == agent_name
+        scheds = schedule_client.get_all_schedules(workflow_name=agent_name)
+        by_wire = {s.name: s for s in scheds}
+        assert set(by_wire) == {f"{agent_name}-daily", f"{agent_name}-weekly"}
+        daily = by_wire[f"{agent_name}-daily"]
+        assert daily.cron_expression == "0 0 9 * * ?"
+        assert daily.start_workflow_request.input == {"k": 1}
+        assert daily.start_workflow_request.name == agent_name
 
     def test_upsert_and_prune(self, schedule_client, agent_name):
         schedule_client.reconcile(
@@ -176,21 +176,21 @@ def test_upsert_and_prune(self, schedule_client, agent_name):
                 Schedule(name="c", cron="0 0 17 * * ?"),
             ],
         )
-        infos = {i.short_name: i for i in schedule_client.list_for_agent(agent_name)}
-        assert set(infos) == {"a", "c"}
-        assert infos["a"].cron == "0 0 9 * * ?"
+        scheds = {s.name: s for s in schedule_client.get_all_schedules(workflow_name=agent_name)}
+        assert set(scheds) == {f"{agent_name}-a", f"{agent_name}-c"}
+        assert scheds[f"{agent_name}-a"].cron_expression == "0 0 9 * * ?"
 
     def test_empty_list_purges(self, schedule_client, agent_name):
         schedule_client.reconcile(agent_name, [Schedule(name="x", cron="0 * * * * ?")])
-        assert len(schedule_client.list_for_agent(agent_name)) == 1
+        assert len(schedule_client.get_all_schedules(workflow_name=agent_name)) == 1
         schedule_client.reconcile(agent_name, [])
-        assert schedule_client.list_for_agent(agent_name) == []
+        assert not schedule_client.get_all_schedules(workflow_name=agent_name)
 
     def test_none_preserves(self, schedule_client, agent_name):
         schedule_client.reconcile(agent_name, [Schedule(name="x", cron="0 * * * * ?")])
         schedule_client.reconcile(agent_name, None)
-        infos = schedule_client.list_for_agent(agent_name)
-        assert [i.short_name for i in infos] == ["x"]
+        scheds = schedule_client.get_all_schedules(workflow_name=agent_name)
+        assert [s.name for s in scheds] == [f"{agent_name}-x"]
 
     def test_duplicate_name_raises_before_io(self, schedule_client, agent_name):
         with pytest.raises(ScheduleNameConflict):
@@ -202,45 +202,44 @@ def test_duplicate_name_raises_before_io(self, schedule_client, agent_name):
                 ],
             )
         # And nothing landed on the server.
-        assert schedule_client.list_for_agent(agent_name) == []
+        assert not schedule_client.get_all_schedules(workflow_name=agent_name)
 
 
 class TestPauseResume:
     def test_pause_then_resume(self, schedule_client, agent_name):
         schedule_client.reconcile(agent_name, [Schedule(name="p", cron="0 0 9 * * ?")])
         wire = f"{agent_name}-p"
 
-        info = schedule_client.get(wire)
-        assert info.paused is False
+        ws = schedule_client.get_schedule(wire)
+        assert not ws.paused
 
         schedule_client.pause(wire)
-        assert schedule_client.get(wire).paused is True
+        assert schedule_client.get_schedule(wire).paused is True
 
         schedule_client.resume(wire)
-        assert schedule_client.get(wire).paused is False
+        assert not schedule_client.get_schedule(wire).paused
 
     def test_paused_on_create_preserves_state(self, schedule_client, agent_name):
         """Spec §10 Q3: paused-on-create still records the schedule cleanly."""
         schedule_client.reconcile(
             agent_name, [Schedule(name="silent", cron="0 0 9 * * ?", paused=True)]
         )
-        info = schedule_client.get(f"{agent_name}-silent")
-        assert info.paused is True
+        ws = schedule_client.get_schedule(f"{agent_name}-silent")
+        assert ws.paused is True
 
 
 class TestDelete:
     def test_delete_removes(self, schedule_client, agent_name):
         schedule_client.reconcile(agent_name, [Schedule(name="d", cron="0 * * * * ?")])
         wire = f"{agent_name}-d"
         schedule_client.delete(wire)
-        assert schedule_client.list_for_agent(agent_name) == []
+        assert not schedule_client.get_all_schedules(workflow_name=agent_name)
 
-    def test_get_after_delete_raises(self, schedule_client, agent_name):
+    def test_get_after_delete_returns_none(self, schedule_client, agent_name):
         schedule_client.reconcile(agent_name, [Schedule(name="g", cron="0 * * * * ?")])
         wire = f"{agent_name}-g"
         schedule_client.delete(wire)
-        with pytest.raises(ScheduleNotFound):
-            schedule_client.get(wire)
+        assert schedule_client.get_schedule(wire) is None
 
 
 class TestPreviewNext:
@@ -257,7 +256,7 @@ def test_returns_execution_id_immediately(self, schedule_client, agent_name):
         schedule_client.reconcile(
             agent_name, [Schedule(name="r", cron="0 0 9 * * ?", input={"trigger": "manual"})]
         )
-        info = schedule_client.get(f"{agent_name}-r")
+        info = _from_workflow_schedule(schedule_client.get_schedule(f"{agent_name}-r"))
 
         t0 = time.monotonic()
         execution_id = schedule_client.run_now(info)

e2e/test_suite24_agent_client.py

@@ -129,21 +129,20 @@ def test_schedule_then_list_then_purge(self, runtime, noop_agent_name):
 
         # Clean slate.
         schedules.reconcile(noop_agent_name, [])
-        assert schedules.list_for_agent(noop_agent_name) == []
+        assert not schedules.get_all_schedules(workflow_name=noop_agent_name)
 
         # Reconcile a single schedule via the client's schedule surface.
         schedules.reconcile(
             noop_agent_name,
             [Schedule(name="daily", cron="0 0 9 * * ?", input={"k": 1})],
         )
-        infos = {i.short_name: i for i in schedules.list_for_agent(noop_agent_name)}
-        assert set(infos) == {"daily"}
-        assert infos["daily"].name == f"{noop_agent_name}-daily"
-        assert infos["daily"].cron == "0 0 9 * * ?"
+        by_wire = {s.name: s for s in schedules.get_all_schedules(workflow_name=noop_agent_name)}
+        assert set(by_wire) == {f"{noop_agent_name}-daily"}
+        assert by_wire[f"{noop_agent_name}-daily"].cron_expression == "0 0 9 * * ?"
 
         # Counterfactual: reconcile with an empty list purges it.
         schedules.reconcile(noop_agent_name, [])
-        assert schedules.list_for_agent(noop_agent_name) == []
+        assert not schedules.get_all_schedules(workflow_name=noop_agent_name)
 
 
 # ── structural consistency: runtime + client share one schedule surface ──

src/conductor/ai/agents/schedule/api.py

@@ -18,6 +18,12 @@
 from typing import Any, List, Optional
 
 from conductor.ai.agents.schedule.schedule import Schedule, ScheduleInfo
+from conductor.client.ai.schedule import (
+    _get_info,
+    _list_infos,
+    _to_save_request,
+    _translate,
+)
 
 
 def _client(runtime: Optional[Any]) -> Any:
@@ -32,7 +38,7 @@ def list(  # noqa: A001 — module-level API mirrors the spec
     agent: str, *, runtime: Optional[Any] = None
 ) -> List[ScheduleInfo]:
     """List all schedules attached to ``agent`` (workflow name)."""
-    return _client(runtime).list_for_agent(agent)
+    return _list_infos(_client(runtime), agent)
 
 
 def get(name: str, *, runtime: Optional[Any] = None) -> ScheduleInfo:
@@ -41,7 +47,7 @@ def get(name: str, *, runtime: Optional[Any] = None) -> ScheduleInfo:
     The agent is recovered from the schedule's ``startWorkflowRequest.name``;
     ``short_name`` on the returned :class:`ScheduleInfo` is the user's original.
     """
-    return _client(runtime).get(name)
+    return _get_info(_client(runtime), name)
 
 
 def pause(name: str, reason: Optional[str] = None, *, runtime: Optional[Any] = None) -> None:
@@ -73,7 +79,7 @@ def run_now(
     seconds) — consistent with ``run()``'s completed-workflow result.
     """
     client = _client(runtime)
-    info = client.get(name)
+    info = _get_info(client, name)
     execution_id = client.run_now(info)
     if not wait:
         return execution_id
@@ -119,4 +125,8 @@ def save(schedule: Schedule, agent: str, *, runtime: Optional[Any] = None) -> No
     Useful for ad-hoc creation from the UI / scripts. Most users should
     prefer the declarative ``deploy(agent, schedules=[...])`` flow.
     """
-    _client(runtime).save(schedule, agent)
+    client = _client(runtime)
+    try:
+        client.save_schedule(_to_save_request(schedule, agent))
+    except Exception as exc:  # noqa: BLE001
+        raise _translate(exc) from exc