diff --git a/backend/apps/workflows/cloud/client.py b/backend/apps/workflows/cloud/client.py
index 57cc82ab..ad4e0a32 100644
--- a/backend/apps/workflows/cloud/client.py
+++ b/backend/apps/workflows/cloud/client.py
@@ -15,7 +15,7 @@ from pydantic import BaseModel, ConfigDict, Field
 from typeguard import typechecked
 
 from backend.apps.settings.credentials import account_auth
-from backend.apps.workflows.cloud.schedule import CloudSchedule
+from backend.apps.workflows.cloud.schedule import CloudSchedule, wire
 
 # The cloud router is mounted at /api/workflows and a trailing slash 404s there, so the collection paths are the empty string, not "/".
 COLLECTION = ""
@@ -71,6 +71,9 @@ class HostedWorkflow(BaseModel):
     id: str
     enabled: bool = False
     next_run_at: Optional[int] = None
+    # Total fires, cloud plus the ones done here before the handover. None from a control plane that
+    # predates the count, and a "we were not told" must never be read as a zero.
+    runs_done: Optional[int] = None
 
 
 class CloudPreflight(BaseModel):
@@ -160,10 +163,12 @@ def p_hosted(raw: Any) -> Optional[HostedWorkflow]:
     if not isinstance(ident, str):
         return None
     nxt = raw.get("next_run_at")
+    done = raw.get("runs_done")
     return HostedWorkflow(
         id=ident,
         enabled=bool(raw.get("enabled")),
         next_run_at=nxt if isinstance(nxt, int) else None,
+        runs_done=done if isinstance(done, int) else None,
     )
 
 
@@ -202,16 +207,13 @@ async def preflight(definition: Dict[str, Any], hosted_id: Optional[str]) -> Clo
     if not isinstance(raw, dict):
         raise CloudUnreachable("the cloud sent a preflight we could not read")
     limits, usage = p_allowance(raw)
-    capability = raw.get("capability")
+    cap = raw.get("capability")
     return CloudPreflight(
         plan=raw.get("plan") if isinstance(raw.get("plan"), str) else None,
         limits=limits,
         usage=usage,
-        capability=(
-            CloudCapability(ok=bool(capability.get("ok")), reason=capability.get("reason"))
-            if isinstance(capability, dict)
-            else None
-        ),
+        capability=CloudCapability(ok=bool(cap.get("ok")), reason=cap.get("reason"))
+        if isinstance(cap, dict) else None,
         hosted=p_hosted(raw.get("hosted")),
     )
 
@@ -229,11 +231,13 @@ async def p_preflight_from_list(hosted_id: Optional[str]) -> CloudPreflight:
 
 @typechecked
 async def put_workflow(
-    *, hosted_id: Optional[str], name: str, definition: Dict[str, Any], schedule: CloudSchedule
+    *, hosted_id: Optional[str], name: str, definition: Dict[str, Any],
+    schedule: CloudSchedule, runs_before: int = 0,
 ) -> HostedWorkflow:
-    """Create the hosted copy, or re-push onto the existing row so an edited
-    workflow stops running last week's prose."""
-    body = {"name": name, "definition": definition, "schedule": schedule.model_dump()}
+    """Create the hosted copy, or re-push onto the existing row so an edited workflow stops running
+    last week's prose. runs_before rides only on the create: an edit that resent it would hand a
+    nearly-spent run cap its whole budget back."""
+    body: Dict[str, Any] = {"name": name, "definition": definition, "schedule": wire(schedule)}
     if hosted_id:
         try:
             raw = await p_call("POST", f"/{hosted_id}/update", body)
@@ -244,7 +248,7 @@ async def put_workflow(
             # 404 is the row being gone (deleted elsewhere, or a control plane with no update route); make a fresh one.
             if exc.status != 404:
                 raise
-    raw = await p_call("POST", COLLECTION, body)
+    raw = await p_call("POST", COLLECTION, {**body, "runs_before": max(0, runs_before)})
     hosted = p_hosted(raw)
     if not hosted:
         raise CloudUnreachable("the cloud accepted the workflow but did not say which one")
@@ -281,16 +285,14 @@ async def list_runs(hosted_id: str) -> List[CloudRun]:
         if not isinstance(row, dict):
             continue
         notices = row.get("notices")
-        out.append(
-            CloudRun(
-                id=str(row.get("id") or ""),
-                status=str(row.get("status") or "unknown"),
-                started_at=row.get("started_at") if isinstance(row.get("started_at"), int) else None,
-                finished_at=row.get("finished_at") if isinstance(row.get("finished_at"), int) else None,
-                error=row.get("error") if isinstance(row.get("error"), str) else None,
-                answer=row.get("answer") if isinstance(row.get("answer"), str) else None,
-                notices=[n for n in notices if isinstance(n, str)] if isinstance(notices, list) else [],
-                cost_usd=row.get("cost_usd") if isinstance(row.get("cost_usd"), (int, float)) else None,
-            )
-        )
+        out.append(CloudRun(
+            id=str(row.get("id") or ""),
+            status=str(row.get("status") or "unknown"),
+            started_at=row.get("started_at") if isinstance(row.get("started_at"), int) else None,
+            finished_at=row.get("finished_at") if isinstance(row.get("finished_at"), int) else None,
+            error=row.get("error") if isinstance(row.get("error"), str) else None,
+            answer=row.get("answer") if isinstance(row.get("answer"), str) else None,
+            notices=[n for n in notices if isinstance(n, str)] if isinstance(notices, list) else [],
+            cost_usd=row.get("cost_usd") if isinstance(row.get("cost_usd"), (int, float)) else None,
+        ))
     return out
diff --git a/backend/apps/workflows/cloud/handover.py b/backend/apps/workflows/cloud/handover.py
index ef41fc4b..fff48142 100644
--- a/backend/apps/workflows/cloud/handover.py
+++ b/backend/apps/workflows/cloud/handover.py
@@ -18,7 +18,7 @@ from typeguard import typechecked
 from backend.apps.workflows import scheduler, storage
 from backend.apps.workflows.cloud import client as cloud
 from backend.apps.workflows.cloud.definition import cloud_definition, definition_signature
-from backend.apps.workflows.cloud.schedule import ScheduleSupported, to_cloud_schedule
+from backend.apps.workflows.cloud.schedule import ScheduleSupported, to_cloud_schedule, wire
 from backend.apps.workflows.cloud.status import epoch_to_datetime
 from backend.apps.workflows.models import Workflow
 
@@ -57,6 +57,7 @@ async def hand_to_cloud(wf: Workflow, enabled: bool) -> TargetOutcome:
             name=wf.title or "Workflow",
             definition=definition,
             schedule=mapping.schedule,
+            runs_before=wf.schedule.runs_count,
         )
         if hosted.enabled != enabled:
             hosted = await cloud.set_enabled(hosted.id, enabled)
@@ -70,7 +71,7 @@ async def hand_to_cloud(wf: Workflow, enabled: bool) -> TargetOutcome:
 
     wf.execution_target = "cloud"
     wf.cloud_workflow_id = hosted.id
-    wf.cloud_definition_signature = definition_signature(definition, mapping.schedule.model_dump())
+    wf.cloud_definition_signature = definition_signature(definition, wire(mapping.schedule))
     wf.schedule.enabled = enabled
     wf.next_run_at = epoch_to_datetime(hosted.next_run_at) if enabled else None
     wf.updated_at = datetime.now()
diff --git a/backend/apps/workflows/cloud/schedule.py b/backend/apps/workflows/cloud/schedule.py
index 35ca89e7..88cd5f09 100644
--- a/backend/apps/workflows/cloud/schedule.py
+++ b/backend/apps/workflows/cloud/schedule.py
@@ -1,16 +1,22 @@
-"""Map a local schedule onto the cloud scheduler's much smaller vocabulary.
+"""Map a local schedule onto the cloud scheduler's smaller vocabulary.
 
-The cloud speaks two cadences: repeat every N minutes, or once a day at a UTC
-time. Everything else this app can express (set weekdays, monthly, every third
-day, stop after N runs) has no cloud equivalent, and silently rounding one of
-them off would fire a workflow on days the user never picked. So anything that
-does not map exactly is refused here, in the user's own words, and stays on
-their machine where it already works.
+The cloud speaks three cadences: repeat every N minutes, once a day, or on the
+weekdays you picked. Each can carry an end date and a run cap. What it does not
+speak is a cadence with a phase longer than one period (every third day, every
+other week, monthly), because the phase is anchored to the workflow's creation
+on this machine and there is nowhere on the wire to put that anchor. Silently
+rounding one of those off would fire a workflow on days the user never picked,
+so it is refused here in the user's own words and stays on their machine.
+
+The wall-clock kinds carry the IANA zone rather than a UTC hour. A UTC hour is a
+schedule that moves by an hour twice a year for everyone outside UTC: "9am" set
+in July quietly becomes 8am in November. The cloud does its recurrence maths in
+the zone for the same reason scheduler._next_fire_after does.
 """
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Literal, Optional, Union
+from typing import Any, Dict, List, Literal, Optional, Union
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from pydantic import BaseModel, ConfigDict
@@ -19,25 +25,40 @@ from typeguard import typechecked
 from backend.apps.workflows.models import ScheduleConfig
 from backend.apps.workflows.scheduler import host_timezone_name
 
-CADENCE_PREFIX = "Cloud runs repeat on an interval or once a day."
+CADENCE_PREFIX = "Cloud runs repeat on an interval, once a day, or on the weekdays you pick."
 
 
-class CloudIntervalSchedule(BaseModel):
+class CloudScheduleBase(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
 
+    # Both optional, both meaning "and then it is finished". Sent only when set: the cloud's schema
+    # takes an absent field, not a null one.
+    ends_at: Optional[int] = None
+    max_runs: Optional[int] = None
+
+
+class CloudIntervalSchedule(CloudScheduleBase):
     kind: Literal["interval"] = "interval"
     minutes: int
 
 
-class CloudDailySchedule(BaseModel):
-    model_config = ConfigDict(validate_assignment=True)
-
+class CloudDailySchedule(CloudScheduleBase):
     kind: Literal["daily"] = "daily"
-    hour_utc: int
-    minute_utc: int
+    hour: int
+    minute: int
+    timezone: str
 
 
-CloudSchedule = Union[CloudIntervalSchedule, CloudDailySchedule]
+class CloudWeeklySchedule(CloudScheduleBase):
+    kind: Literal["weekly"] = "weekly"
+    # Sunday=0, the same convention as ScheduleConfig.on_days.
+    days: List[int]
+    hour: int
+    minute: int
+    timezone: str
+
+
+CloudSchedule = Union[CloudIntervalSchedule, CloudDailySchedule, CloudWeeklySchedule]
 
 
 class ScheduleSupported(BaseModel):
@@ -58,62 +79,87 @@ ScheduleMapping = Union[ScheduleSupported, ScheduleUnsupported]
 
 
 @typechecked
-def p_zone(name: str) -> ZoneInfo:
-    if not name or name == "local":
-        name = host_timezone_name()
-    try:
-        return ZoneInfo(name)
-    except ZoneInfoNotFoundError:
-        return ZoneInfo("UTC")
+def wire(sched: CloudSchedule) -> Dict[str, Any]:
+    """The JSON body shape. Unset bounds are dropped rather than sent as null, which is what the
+    cloud's schema expects and what keeps the definition fingerprint stable across versions."""
+    return sched.model_dump(exclude_none=True)
 
 
 @typechecked
-def p_utc_time_of_day(sched: ScheduleConfig, ref: Optional[datetime] = None) -> CloudDailySchedule:
-    """The user's wall-clock time expressed in UTC, using today's offset."""
-    zone = p_zone(sched.timezone)
-    local = (ref or datetime.now(zone)).astimezone(zone)
-    at = local.replace(hour=sched.hour, minute=sched.minute, second=0, microsecond=0)
-    utc = at.astimezone(ZoneInfo("UTC"))
-    return CloudDailySchedule(hour_utc=utc.hour, minute_utc=utc.minute)
+def p_zone_name(name: str) -> str:
+    """A concrete IANA name the cloud can hand to its own tz database. "local" and anything
+    unresolvable fall back to this host's zone, which is what the local scheduler already does."""
+    if not name or name == "local":
+        return host_timezone_name()
+    try:
+        ZoneInfo(name)
+    except ZoneInfoNotFoundError:
+        return host_timezone_name()
+    return name
+
+
+@typechecked
+def p_epoch_ms(when: datetime) -> int:
+    # Naive datetimes are host-local, matching how the local scheduler reads its own stored dates.
+    aware = when if when.tzinfo is not None else when.replace(tzinfo=ZoneInfo(host_timezone_name()))
+    return int(aware.timestamp() * 1000)
+
+
+@typechecked
+def p_bounds(sched: ScheduleConfig) -> Dict[str, Any]:
+    out: Dict[str, Any] = {}
+    if sched.ends_at is not None:
+        out["ends_at"] = p_epoch_ms(sched.ends_at)
+    if sched.max_runs is not None:
+        out["max_runs"] = sched.max_runs
+    return out
 
 
 @typechecked
 def to_cloud_schedule(sched: ScheduleConfig) -> ScheduleMapping:
-    if sched.max_runs is not None:
-        return ScheduleUnsupported(
-            reason=(
-                f"This schedule stops itself after {sched.max_runs} "
-                f"run{'' if sched.max_runs == 1 else 's'}, and the cloud scheduler cannot count down "
-                "to a stop. Remove the limit to run it in the cloud."
-            ),
-        )
-    if sched.ends_at is not None:
-        return ScheduleUnsupported(
-            reason=(
-                "This schedule has an end date, and the cloud scheduler cannot honour one. "
-                "Remove the end date to run it in the cloud."
-            ),
-        )
+    bounds = p_bounds(sched)
     if sched.repeat_unit == "minute":
-        return ScheduleSupported(schedule=CloudIntervalSchedule(minutes=max(5, sched.repeat_every)))
+        return ScheduleSupported(schedule=CloudIntervalSchedule(minutes=max(5, sched.repeat_every), **bounds))
     if sched.repeat_unit == "hour":
-        return ScheduleSupported(schedule=CloudIntervalSchedule(minutes=max(5, sched.repeat_every * 60)))
-    if sched.repeat_unit == "day" and sched.repeat_every == 1:
-        return ScheduleSupported(schedule=p_utc_time_of_day(sched))
+        return ScheduleSupported(
+            schedule=CloudIntervalSchedule(minutes=max(5, sched.repeat_every * 60), **bounds)
+        )
+
+    zone = p_zone_name(sched.timezone)
     if sched.repeat_unit == "day":
+        if sched.repeat_every == 1:
+            return ScheduleSupported(
+                schedule=CloudDailySchedule(hour=sched.hour, minute=sched.minute, timezone=zone, **bounds)
+            )
         return ScheduleUnsupported(
             reason=(
                 f"{CADENCE_PREFIX} This one runs every {sched.repeat_every} days at a set time, "
                 "which the cloud scheduler cannot do yet, so it stays on this device."
             ),
         )
+
     if sched.repeat_unit == "week":
+        if not sched.on_days:
+            return ScheduleUnsupported(
+                reason="Pick the days this should run on before choosing where it runs.",
+            )
+        if sched.repeat_every == 1:
+            return ScheduleSupported(
+                schedule=CloudWeeklySchedule(
+                    days=sorted(sched.on_days),
+                    hour=sched.hour,
+                    minute=sched.minute,
+                    timezone=zone,
+                    **bounds,
+                )
+            )
         return ScheduleUnsupported(
             reason=(
-                f"{CADENCE_PREFIX} This one runs on the weekdays you picked, "
+                f"{CADENCE_PREFIX} This one runs every {sched.repeat_every} weeks, "
                 "which the cloud scheduler cannot do yet, so it stays on this device."
             ),
         )
+
     return ScheduleUnsupported(
         reason=(
             f"{CADENCE_PREFIX} This one runs monthly, "
diff --git a/backend/apps/workflows/cloud/status.py b/backend/apps/workflows/cloud/status.py
index 96677731..e665aec0 100644
--- a/backend/apps/workflows/cloud/status.py
+++ b/backend/apps/workflows/cloud/status.py
@@ -16,7 +16,7 @@ from typeguard import typechecked
 from backend.apps.workflows import storage
 from backend.apps.workflows.cloud import client as cloud
 from backend.apps.workflows.cloud.definition import cloud_definition, definition_signature
-from backend.apps.workflows.cloud.schedule import ScheduleSupported, to_cloud_schedule
+from backend.apps.workflows.cloud.schedule import ScheduleSupported, to_cloud_schedule, wire
 from backend.apps.workflows.models import Workflow
 
 
@@ -74,7 +74,7 @@ def current_signature(wf: Workflow) -> Optional[str]:
     mapping = to_cloud_schedule(wf.schedule)
     if not isinstance(mapping, ScheduleSupported):
         return None
-    return definition_signature(cloud_definition(wf), mapping.schedule.model_dump())
+    return definition_signature(cloud_definition(wf), wire(mapping.schedule))
 
 
 @typechecked
@@ -91,6 +91,11 @@ def p_mirror_cloud_state(wf: Workflow, hosted: Optional[cloud.HostedWorkflow]) -
     if hosted and wf.schedule.enabled != hosted.enabled:
         wf.schedule.enabled = hosted.enabled
         changed = True
+    # Runs the cloud performed count against a max_runs cap the same as ours do, so copy its total
+    # back. Without this, taking a nearly-spent schedule off the cloud would restart it at zero.
+    if hosted and hosted.runs_done is not None and wf.schedule.runs_count != hosted.runs_done:
+        wf.schedule.runs_count = hosted.runs_done
+        changed = True
     if changed:
         storage.save_workflow(wf)
 
diff --git a/backend/tests/test_cloud_schedule_agreement.py b/backend/tests/test_cloud_schedule_agreement.py
new file mode 100644
index 00000000..b3f91e24
--- /dev/null
+++ b/backend/tests/test_cloud_schedule_agreement.py
@@ -0,0 +1,121 @@
+"""The two schedulers have to agree on when "9am Monday" is.
+
+A workflow can sit on this machine or on our servers, and the user is told a "next run" either way.
+Two recurrence engines in two languages means two chances to be wrong, and the place they would
+diverge is exactly the place wall-clock scheduling is hard: the morning a clock jumps forward and
+2:30am never happens, and the morning it falls back and 1:30am happens twice.
+
+The vector table below is duplicated verbatim in the cloud service's tests/workflow-schedule.test.ts.
+Each row asserts the same answer on both sides, so if either engine drifts one of the suites fails.
+Nothing here talks to the network; both halves are pure functions of a schedule and a moment.
+"""
+from datetime import datetime, timezone
+from zoneinfo import ZoneInfo
+
+from backend.apps.workflows.cloud.schedule import (
+    CloudDailySchedule,
+    CloudWeeklySchedule,
+    ScheduleSupported,
+    to_cloud_schedule,
+    wire,
+)
+from backend.apps.workflows.models import ScheduleConfig, Workflow
+from backend.apps.workflows.scheduler import compute_next_fire
+
+LA = "America/Los_Angeles"
+
+# (label, ScheduleConfig kwargs, asked at, fires at). Times are UTC because that is what both
+# engines return; the point of each row is the LOCAL clock it corresponds to, named in the label.
+VECTORS = [
+    ("LA daily 9am, the Saturday before the clocks go forward",
+     dict(repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone=LA),
+     "2026-03-06T20:00:00Z", "2026-03-07T17:00:00Z"),
+    ("LA daily 9am, the day the clocks go forward",
+     dict(repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone=LA),
+     "2026-03-07T20:00:00Z", "2026-03-08T16:00:00Z"),
+    ("LA daily 9am, the day the clocks go back",
+     dict(repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone=LA),
+     "2026-10-31T20:00:00Z", "2026-11-01T17:00:00Z"),
+    ("LA weekdays 9am, asked Friday after the slot, over a spring-forward weekend",
+     dict(repeat_unit="week", repeat_every=1, on_days=[1, 2, 3, 4, 5], hour=9, minute=0, timezone=LA),
+     "2026-03-06T18:00:00Z", "2026-03-09T16:00:00Z"),
+    ("LA weekdays 9am, asked Friday after the slot, over a fall-back weekend",
+     dict(repeat_unit="week", repeat_every=1, on_days=[1, 2, 3, 4, 5], hour=9, minute=0, timezone=LA),
+     "2026-10-30T18:00:00Z", "2026-11-02T17:00:00Z"),
+    ("LA weekends 9am, asked on a Wednesday",
+     dict(repeat_unit="week", repeat_every=1, on_days=[0, 6], hour=9, minute=0, timezone=LA),
+     "2026-06-10T18:00:00Z", "2026-06-13T16:00:00Z"),
+    ("LA daily 2:30am, an hour the clocks skip",
+     dict(repeat_unit="day", repeat_every=1, hour=2, minute=30, timezone=LA),
+     "2026-03-08T09:59:00Z", "2026-03-08T10:30:00Z"),
+    ("LA daily 1:30am, first time through the repeated hour",
+     dict(repeat_unit="day", repeat_every=1, hour=1, minute=30, timezone=LA),
+     "2026-11-01T08:00:00Z", "2026-11-01T08:30:00Z"),
+    ("LA daily 1:30am, asked at 1:45 the first time round, so today's slot has gone",
+     dict(repeat_unit="day", repeat_every=1, hour=1, minute=30, timezone=LA),
+     "2026-11-01T08:45:00Z", "2026-11-02T09:30:00Z"),
+    ("LA daily 1:30am, asked at 1:15 the second time round, so it fires again this hour",
+     dict(repeat_unit="day", repeat_every=1, hour=1, minute=30, timezone=LA),
+     "2026-11-01T09:15:00Z", "2026-11-01T09:30:00Z"),
+    ("LA daily 1:30am, asked at exactly 1:30, so this one has just gone",
+     dict(repeat_unit="day", repeat_every=1, hour=1, minute=30, timezone=LA),
+     "2026-11-01T08:30:00Z", "2026-11-02T09:30:00Z"),
+    ("Berlin daily 2:30am, an hour the clocks skip",
+     dict(repeat_unit="day", repeat_every=1, hour=2, minute=30, timezone="Europe/Berlin"),
+     "2026-03-29T00:30:00Z", "2026-03-29T01:30:00Z"),
+    ("Berlin daily 2:30am, the second time through the repeated hour",
+     dict(repeat_unit="day", repeat_every=1, hour=2, minute=30, timezone="Europe/Berlin"),
+     "2026-10-25T01:17:33Z", "2026-10-25T01:30:00Z"),
+    ("Sydney Sundays 11:45pm, over their spring-forward",
+     dict(repeat_unit="week", repeat_every=1, on_days=[0], hour=23, minute=45, timezone="Australia/Sydney"),
+     "2026-10-02T05:00:00Z", "2026-10-04T12:45:00Z"),
+    ("Kolkata daily 9am, a half-hour offset with no DST",
+     dict(repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone="Asia/Kolkata"),
+     "2026-06-15T12:00:00Z", "2026-06-16T03:30:00Z"),
+    ("UTC daily midnight, asked one second after it fired",
+     dict(repeat_unit="day", repeat_every=1, hour=0, minute=0, timezone="UTC"),
+     "2026-06-15T00:00:01Z", "2026-06-16T00:00:00Z"),
+]
+
+
+def p_moment(text: str) -> datetime:
+    return datetime.fromisoformat(text.replace("Z", "+00:00"))
+
+
+def p_workflow(config: dict) -> Workflow:
+    # created_at is the phase anchor for multi-period cadences. Every schedule here repeats once per
+    # period, so it cannot move an answer; it is pinned only so nothing about these rows floats.
+    return Workflow(
+        title="agreement",
+        created_at=datetime(2020, 1, 1, tzinfo=timezone.utc),
+        schedule=ScheduleConfig(enabled=True, **config),
+    )
+
+
+def test_the_local_scheduler_hits_every_vector():
+    for label, config, asked, fires in VECTORS:
+        assert compute_next_fire(p_workflow(config), p_moment(asked)) == p_moment(fires), label
+
+
+def test_every_vector_maps_onto_a_schedule_the_cloud_can_hold():
+    """A vector the cloud would refuse proves nothing about the two engines agreeing."""
+    for label, config, *_ in VECTORS:
+        mapping = to_cloud_schedule(ScheduleConfig(enabled=True, **config))
+        assert isinstance(mapping, ScheduleSupported), label
+        assert isinstance(mapping.schedule, (CloudDailySchedule, CloudWeeklySchedule)), label
+        # The zone has to survive the trip, or the cloud does its maths in the wrong one.
+        assert wire(mapping.schedule)["timezone"] == config["timezone"], label
+
+
+def test_a_daily_nine_am_never_drifts_off_nine_am():
+    """The reason the zone travels with the schedule at all: stored as a UTC hour, every one of
+    these fires would move by an hour twice a year for anyone outside UTC."""
+    wf = p_workflow(dict(repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone=LA))
+    cursor = datetime(2026, 3, 5, tzinfo=timezone.utc)
+    days = set()
+    for _ in range(250):
+        cursor = compute_next_fire(wf, cursor)
+        local = cursor.astimezone(ZoneInfo(LA))
+        assert (local.hour, local.minute) == (9, 0), cursor.isoformat()
+        days.add(local.date())
+    assert len(days) == 250, "one fire per calendar day, none doubled and none skipped"
diff --git a/backend/tests/test_cloud_workflow_definition.py b/backend/tests/test_cloud_workflow_definition.py
index 8625e911..ff4c713e 100644
--- a/backend/tests/test_cloud_workflow_definition.py
+++ b/backend/tests/test_cloud_workflow_definition.py
@@ -4,8 +4,16 @@ Both are pure functions, and both decide something a user reads: a schedule the
 express has to be refused in words rather than rounded off, and the copy we push must not carry
 anything local (session ids, a phone number) off this machine.
 """
+from datetime import datetime, timezone
+
+from backend.apps.workflows.cloud import schedule as cloud_schedule
 from backend.apps.workflows.cloud.definition import cloud_definition, definition_signature
-from backend.apps.workflows.cloud.schedule import ScheduleSupported, ScheduleUnsupported, to_cloud_schedule
+from backend.apps.workflows.cloud.schedule import (
+    ScheduleSupported,
+    ScheduleUnsupported,
+    to_cloud_schedule,
+    wire,
+)
 from backend.apps.workflows.models import PermissionTier, ScheduleConfig, Workflow, WorkflowStep
 
 
@@ -21,15 +29,25 @@ def p_wf(**overrides) -> Workflow:
     return Workflow(**base)
 
 
-def test_only_daily_and_interval_schedules_map_to_the_cloud():
-    assert isinstance(to_cloud_schedule(p_sched()), ScheduleSupported)
-    assert isinstance(to_cloud_schedule(p_sched(repeat_unit="minute", repeat_every=30)), ScheduleSupported)
-    assert isinstance(to_cloud_schedule(p_sched(repeat_unit="hour", repeat_every=6)), ScheduleSupported)
+def test_interval_daily_and_weekday_schedules_map_to_the_cloud():
+    for supported in (
+        p_sched(),
+        p_sched(repeat_unit="minute", repeat_every=30),
+        p_sched(repeat_unit="hour", repeat_every=6),
+        p_sched(repeat_unit="week", on_days=[1, 2, 3, 4, 5]),
+        p_sched(max_runs=5),
+        p_sched(ends_at=datetime(2027, 1, 1, tzinfo=timezone.utc)),
+    ):
+        assert isinstance(to_cloud_schedule(supported), ScheduleSupported)
+
+
+def test_a_cadence_with_a_phase_the_wire_cannot_carry_is_refused_in_words():
     for unsupported in (
-        p_sched(repeat_unit="week", on_days=[1]),
         p_sched(repeat_unit="month", day_of_month=1),
         p_sched(repeat_unit="day", repeat_every=3),
-        p_sched(max_runs=5),
+        p_sched(repeat_unit="week", repeat_every=2, on_days=[1]),
+        # A weekly schedule with no days chosen is not a cadence yet, and saying "monthly" here would be a lie.
+        p_sched(repeat_unit="week", on_days=[]),
     ):
         mapping = to_cloud_schedule(unsupported)
         assert isinstance(mapping, ScheduleUnsupported)
@@ -37,10 +55,47 @@ def test_only_daily_and_interval_schedules_map_to_the_cloud():
         assert mapping.reason.endswith(".") and " " in mapping.reason
 
 
-def test_a_9am_wall_clock_becomes_the_right_utc_time():
+def test_a_wall_clock_time_travels_with_its_zone_and_not_as_a_utc_hour():
+    """Storing 9am Tokyo as its UTC hour is how a schedule silently moves an hour twice a year."""
     mapping = to_cloud_schedule(p_sched(hour=9, minute=30, timezone="Asia/Tokyo"))
     assert isinstance(mapping, ScheduleSupported)
-    assert mapping.schedule.model_dump() == {"kind": "daily", "hour_utc": 0, "minute_utc": 30}
+    assert wire(mapping.schedule) == {
+        "kind": "daily", "hour": 9, "minute": 30, "timezone": "Asia/Tokyo",
+    }
+
+
+def test_weekdays_travel_sorted_and_end_conditions_ride_along():
+    mapping = to_cloud_schedule(
+        p_sched(
+            repeat_unit="week",
+            on_days=[5, 1, 3],
+            max_runs=4,
+            ends_at=datetime(2027, 3, 1, 12, 0, tzinfo=timezone.utc),
+        )
+    )
+    assert isinstance(mapping, ScheduleSupported)
+    assert wire(mapping.schedule) == {
+        "kind": "weekly", "days": [1, 3, 5], "hour": 9, "minute": 0, "timezone": "UTC",
+        "max_runs": 4, "ends_at": 1803902400000,  # 2027-03-01T12:00Z
+    }
+
+
+def test_a_legacy_local_zone_becomes_this_machines_real_zone(monkeypatch):
+    """Records written before schedules carried a zone say "local". Sending that word, or flattening
+    it to UTC, moves every one of their fires by this machine's whole offset."""
+    monkeypatch.setattr(cloud_schedule, "host_timezone_name", lambda: "America/New_York")
+    for stored in ("local", "", "Mars/Olympus_Mons"):
+        mapping = to_cloud_schedule(p_sched(timezone=stored))
+        assert isinstance(mapping, ScheduleSupported)
+        assert wire(mapping.schedule)["timezone"] == "America/New_York", stored
+
+
+def test_unset_end_conditions_are_absent_rather_than_null():
+    """The cloud's schema takes an absent field, not a null one, so a null would 400 every push."""
+    mapping = to_cloud_schedule(p_sched())
+    assert isinstance(mapping, ScheduleSupported)
+    assert "ends_at" not in wire(mapping.schedule)
+    assert "max_runs" not in wire(mapping.schedule)
 
 
 def test_the_cloud_copy_carries_no_local_secrets_and_no_live_timer():
@@ -59,7 +114,7 @@ def test_the_cloud_copy_carries_no_local_secrets_and_no_live_timer():
 
 def test_a_signature_tracks_edits_and_ignores_the_clock():
     wf = p_wf()
-    schedule = {"kind": "daily", "hour_utc": 9, "minute_utc": 0}
+    schedule = {"kind": "daily", "hour": 9, "minute": 0, "timezone": "UTC"}
     first = definition_signature(cloud_definition(wf), schedule)
 
     wf.title = wf.title
diff --git a/backend/tests/test_cloud_workflow_target.py b/backend/tests/test_cloud_workflow_target.py
index 0a39eada..9d56949f 100644
--- a/backend/tests/test_cloud_workflow_target.py
+++ b/backend/tests/test_cloud_workflow_target.py
@@ -119,7 +119,7 @@ async def test_a_refused_flip_leaves_the_workflow_on_this_device(monkeypatch):
 
 @pytest.mark.asyncio
 async def test_an_unsupported_schedule_never_reaches_the_network(monkeypatch):
-    wf = p_wf(schedule=p_sched(repeat_unit="week", on_days=[1]))
+    wf = p_wf(schedule=p_sched(repeat_unit="month", day_of_month=1))
     seen = p_answer(monkeypatch, lambda method, path, body: p_preflight_body())
     outcome = await set_workflow_target(wf.id, TargetRequest(target="cloud", enabled=True))
     assert outcome.ok is False
@@ -127,6 +127,40 @@ async def test_an_unsupported_schedule_never_reaches_the_network(monkeypatch):
     assert storage.get_workflow(wf.id).execution_target == "device"
 
 
+@pytest.mark.asyncio
+async def test_weekdays_go_up_with_the_days_the_user_picked(monkeypatch):
+    """"Every weekday at 9am" is the schedule people actually write, and it used to be refused."""
+    wf = p_wf(schedule=p_sched(repeat_unit="week", on_days=[5, 1, 2, 3, 4], timezone="America/Los_Angeles"))
+    seen = p_answer(monkeypatch, lambda method, path, body: p_hosted())
+    outcome = await set_workflow_target(wf.id, TargetRequest(target="cloud", enabled=True))
+    assert outcome.ok is True
+    sent = seen[-1][2]["schedule"]
+    assert sent == {"kind": "weekly", "days": [1, 2, 3, 4, 5], "hour": 9, "minute": 0,
+                    "timezone": "America/Los_Angeles"}
+    assert storage.get_workflow(wf.id).execution_target == "cloud"
+
+
+@pytest.mark.asyncio
+async def test_a_capped_schedule_hands_over_its_cap_and_what_it_has_already_spent(monkeypatch):
+    wf = p_wf(schedule=p_sched(max_runs=5, runs_count=2))
+    seen = p_answer(monkeypatch, lambda method, path, body: p_hosted())
+    assert (await set_workflow_target(wf.id, TargetRequest(target="cloud", enabled=True))).ok is True
+    body = seen[-1][2]
+    assert body["schedule"]["max_runs"] == 5
+    # Without this the cloud would give a schedule with 3 runs left a fresh 5.
+    assert body["runs_before"] == 2
+
+
+@pytest.mark.asyncio
+async def test_runs_the_cloud_performed_come_back_onto_our_own_counter(monkeypatch):
+    wf = p_wf(schedule=p_sched(max_runs=5, runs_count=2), execution_target="cloud",
+              cloud_workflow_id="cloud-1")
+    storage.save_workflow(wf)
+    p_answer(monkeypatch, lambda method, path, body: p_preflight_body(hosted=p_hosted(runs_done=4)))
+    await compute_status(wf)
+    assert storage.get_workflow(wf.id).schedule.runs_count == 4
+
+
 @pytest.mark.asyncio
 async def test_an_accepted_flip_records_which_copy_is_up_there(monkeypatch):
     wf = p_wf()