diff --git a/cli/hub_server.py b/cli/hub_server.py
index e699ced..3f0bd17 100644
--- a/cli/hub_server.py
+++ b/cli/hub_server.py
@@ -3249,6 +3249,180 @@ def _hub_override_audit(action: str, row: dict, *, confirmed: bool = False,
         pass
 
 
+# ── Scheduled tasks (docs/scheduled-tasks.md) ────────────────────────────
+# The Oracle owns the loop; the hub reads and writes the same stores on this
+# machine, so a Hub edit or decision is picked up on the Oracle's next tick.
+
+def _schedule_api():
+    from oracle.config import load_config as _oracle_config
+    from oracle.services import schedule_api
+    from services.schedule_approvals import ApprovalStore
+    from services.schedule_store import ScheduleStore
+    if schedule_api._get_store() is None:
+        store, approvals = ScheduleStore(), ApprovalStore()
+        schedule_api.configure(get_scheduler=lambda: None, get_store=lambda: store,
+                               get_approvals=lambda: approvals, get_scanner=lambda: None,
+                               get_cfg=_oracle_config)
+    return schedule_api
+
+
+def _sched_call(fn, *args, **kwargs):
+    api = _schedule_api()
+    try:
+        return fn(*args, **kwargs), None
+    except api.ApiError as exc:
+        return None, (jsonify({"error": str(exc)}), exc.status)
+    except ValueError as exc:
+        return None, (jsonify({"error": str(exc)}), 400)
+
+
+def _sched_project_arg(key: str = "path") -> tuple[str, object]:
+    raw = (request.args.get(key) or "").strip()
+    if not raw:
+        return "", None
+    try:
+        return str(_resolve_project_path(raw)), None
+    except ValueError as exc:
+        return "", (jsonify({"error": str(exc)}), 404)
+
+
+def _sched_body_with_project(body: dict) -> tuple[dict, object]:
+    body = dict(body or {})
+    raw = str(body.pop("project", "") or body.get("project_path") or "").strip()
+    if raw:
+        try:
+            body["project_path"] = str(_resolve_project_path(raw))
+        except ValueError as exc:
+            return body, (jsonify({"error": str(exc)}), 404)
+    return body, None
+
+
+def _sched_with_names(rows: list) -> list:
+    return [dict(r, project_name=Path(str(r.get("project_path") or "")).name) for r in rows]
+
+
+@app.route("/api/hub/schedules", methods=["GET"])
+def api_hub_schedules():
+    project_path, err = _sched_project_arg()
+    if err:
+        return err
+    store = _schedule_api()._get_store()
+    rows = _sched_with_names(store.list(project_path=project_path or None))
+    return jsonify({"schedules": rows, "count": len(rows), "rev": store.rev(), "project": project_path})
+
+
+@app.route("/api/hub/schedules", methods=["POST"])
+def api_hub_schedules_create():
+    body, err = _sched_body_with_project(request.get_json(silent=True) or {})
+    if err:
+        return err
+    row, err = _sched_call(_schedule_api().create_schedule, body, scanner=None, by="desktop")
+    return err or (jsonify(row), 201)
+
+
+@app.route("/api/hub/schedules/status", methods=["GET"])
+def api_hub_schedules_status():
+    """The Oracle's scheduler status, or ``reachable: false`` when it is down."""
+    base = (_read_hub_config().get("oracle_url") or "").strip() or _oracle_service().url()
+    try:
+        with urllib.request.urlopen(f"{base.rstrip('/')}/api/schedules/status", timeout=2) as resp:
+            data = json.loads(resp.read().decode("utf-8"))
+    except (OSError, ValueError) as exc:
+        return jsonify({"running": False, "reachable": False, "error": str(exc), "oracle_url": base})
+    data["reachable"] = True
+    data["oracle_url"] = base
+    return jsonify(data)
+
+
+@app.route("/api/hub/schedules/preview", methods=["POST"])
+def api_hub_schedules_preview():
+    data, err = _sched_call(_schedule_api().preview_when, request.get_json(silent=True) or {})
+    return err or jsonify(data)
+
+
+@app.route("/api/hub/schedules/runs", methods=["GET"])
+def api_hub_schedule_runs():
+    project_path, err = _sched_project_arg()
+    if err:
+        return err
+    try:
+        limit = max(1, min(int(request.args.get("limit") or 50), 500))
+    except (TypeError, ValueError):
+        limit = 50
+    rows = _schedule_api()._get_store().runs(
+        schedule_id=(request.args.get("schedule_id") or None),
+        status=((request.args.get("status") or "").strip() or None),
+        project_path=project_path or None, limit=limit)
+    return jsonify({"runs": _sched_with_names(rows), "count": len(rows), "project": project_path})
+
+
+@app.route("/api/hub/schedules/approvals", methods=["GET"])
+def api_hub_schedule_approvals():
+    project_path, err = _sched_project_arg()
+    if err:
+        return err
+    rows = _schedule_api()._get_approvals().list(
+        status=((request.args.get("status") or "").strip() or None), project_path=project_path or None)
+    return jsonify({"approvals": _sched_with_names(rows), "count": len(rows)})
+
+
+@app.route("/api/hub/schedules/approvals/<approval_id>", methods=["POST"])
+def api_hub_schedule_decide(approval_id):
+    row, err = _sched_call(_schedule_api().decide_approval, str(approval_id),
+                           request.get_json(silent=True) or {}, by="desktop")
+    return err or jsonify(row)
+
+
+@app.route("/api/hub/schedules/<schedule_id>", methods=["GET"])
+def api_hub_schedule_one(schedule_id):
+    store = _schedule_api()._get_store()
+    row = store.get(schedule_id)
+    if row is None:
+        return jsonify({"error": f"no schedule {schedule_id}"}), 404
+    return jsonify(dict(row, rev=store.rev(), project_name=Path(str(row.get("project_path") or "")).name))
+
+
+@app.route("/api/hub/schedules/<schedule_id>", methods=["PATCH"])
+def api_hub_schedule_update(schedule_id):
+    body, err = _sched_body_with_project(request.get_json(silent=True) or {})
+    if err:
+        return err
+    row, err = _sched_call(_schedule_api().update_schedule, schedule_id, body, scanner=None, by="desktop")
+    return err or jsonify(row)
+
+
+@app.route("/api/hub/schedules/<schedule_id>", methods=["DELETE"])
+def api_hub_schedule_delete(schedule_id):
+    if not _schedule_api()._get_store().delete(schedule_id, by="desktop"):
+        return jsonify({"error": f"no schedule {schedule_id}"}), 404
+    return jsonify({"deleted": schedule_id})
+
+
+@app.route("/api/hub/schedules/<schedule_id>/run", methods=["POST"])
+def api_hub_schedule_run(schedule_id):
+    run, err = _sched_call(_schedule_api().run_schedule, schedule_id, request.get_json(silent=True) or {},
+                           by="desktop")
+    return err or (jsonify(run), 202)
+
+
+@app.route("/api/hub/schedules/<schedule_id>/<action>", methods=["POST"])
+def api_hub_schedule_toggle(schedule_id, action):
+    if action not in ("enable", "disable"):
+        return jsonify({"error": "unknown action"}), 404
+    row, err = _sched_call(_schedule_api().update_schedule, schedule_id, {"enabled": action == "enable"},
+                           scanner=None, by="desktop")
+    return err or jsonify(row)
+
+
+@app.route("/api/hub/schedules/<schedule_id>/runs", methods=["GET"])
+def api_hub_schedule_runs_one(schedule_id):
+    store = _schedule_api()._get_store()
+    if store.get(schedule_id) is None:
+        return jsonify({"error": f"no schedule {schedule_id}"}), 404
+    rows = store.runs(schedule_id=schedule_id, status=((request.args.get("status") or "").strip() or None))
+    return jsonify({"runs": rows, "count": len(rows)})
+
+
 @app.route("/api/hub/overrides", methods=["GET"])
 def api_hub_overrides():
     """Pending + recently-decided requests across every project, newest first.
diff --git a/docs/scheduled-tasks.md b/docs/scheduled-tasks.md
index 676e36f..f7943e9 100644
--- a/docs/scheduled-tasks.md
+++ b/docs/scheduled-tasks.md
@@ -80,6 +80,18 @@ POST   /api/schedules/preview             {when, count} → next fire times
 GET    /api/schedules/status              POST /api/schedules/kick
 ```
 
+Desk and phone (`/api/mobile/schedules…`, Bearer on every method, capabilities
+`schedules` / `schedules_write` switched by `mobile_schedules_enabled` /
+`mobile_schedules_write`): the same list, one, runs, approvals, create, PATCH,
+DELETE, `/run`, `/enable`, `/disable` and `/approvals/<id>/decide`, every row
+filtered to the projects the gateway serves; `decided_by` is the client kind.
+
+Hub (`/api/hub/schedules…`, loopback): the same operations over the same files
+plus `?path=` project filters, `project_name` on rows, `/preview`, and
+`/status`, which proxies the Oracle's scheduler status and reports
+`reachable: false` when the Oracle is down. The Hub never dispatches; a Hub
+write is picked up on the Oracle's next tick.
+
 ## Config (`~/.c3/oracle/config.json`)
 
 `scheduler_enabled`, `scheduler_tick_seconds`, `scheduler_default_timeout_s`,
diff --git a/oracle/services/mobile_api.py b/oracle/services/mobile_api.py
index 4d46643..b86f94b 100644
--- a/oracle/services/mobile_api.py
+++ b/oracle/services/mobile_api.py
@@ -76,7 +76,7 @@
 #    trailing window, the "this rule is costing you" nudge (D3a). The
 #    /overrides/<id> detail also carries `allow_session_grants`,
 #    `max_ttl_s` and `session_confirm` at the top level.
-API_VERSION = 5
+API_VERSION = 6
 
 CAPABILITIES = [
     "feed", "feed_wait", "projects", "health", "pm", "pm_events", "digest",
@@ -92,6 +92,9 @@
     # Background shell jobs (/jobs) and local CI runs (/ci/runs, /ci/run):
     # read-only views over the job store and the CI index (v2.126.0, D0b).
     "jobs", "ci",
+    # Scheduled tasks (/schedules): read the schedules, their runs and pending
+    # approvals; write = create/edit/run-now/decide (docs/scheduled-tasks.md).
+    "schedules", "schedules_write",
     # Poll-based chat transport, served by services/chat_poll.py under
     # /api/mobile/chat/*. Advertised here because /info is the single place a
     # client probes; the routes themselves live in that blueprint.
@@ -117,6 +120,8 @@
     # that mutate are switchable.
     "insights_write": "mobile_insights_write",
     "suggestions_write": "mobile_suggestions_write",
+    "schedules": "mobile_schedules_enabled",
+    "schedules_write": "mobile_schedules_write",
 }
 
 FEED_TYPES = ("notification", "activity", "edit", "session_stat")
@@ -966,6 +971,200 @@ def mobile_jobs():
                     "status": status, "limit": limit})
 
 
+# ── Scheduled tasks ───────────────────────────────────────
+# The operations live in oracle/services/schedule_api.py; this surface adds
+# the Bearer gate, the capability switches, the registered-project filter and
+# the per-client decided_by.
+
+def _schedule_write_gate():
+    gate = _feature_or_404("schedules") or _feature_or_404("schedules_write")
+    if gate:
+        return gate
+    return _security_gate()
+
+
+def _schedule_api():
+    from oracle.services import schedule_api
+    return schedule_api
+
+
+def _schedule_row_or_404(schedule_id: str):
+    row = _schedule_api()._get_store().get(schedule_id)
+    if row is None or _resolve_registered(str(row.get("project_path") or "")) is None:
+        return None, (jsonify({"error": "unknown schedule"}), 404)
+    return row, None
+
+
+def _registered_rows(rows: list) -> list:
+    return [r for r in rows if _resolve_registered(str(r.get("project_path") or "")) is not None]
+
+
+def _schedule_call(fn, *args, **kwargs):
+    api = _schedule_api()
+    try:
+        return fn(*args, **kwargs), None
+    except api.ApiError as exc:
+        return None, (jsonify({"error": str(exc)}), exc.status)
+    except Exception as exc:
+        return None, _svc_error(exc)
+
+
+@bp.route("/schedules", methods=["GET"])
+def mobile_schedules_list():
+    """Every schedule on registered projects (``project`` narrows to one)."""
+    gate = _feature_or_404("schedules")
+    if gate:
+        return gate
+    project_path = ""
+    raw_project = (request.args.get("project") or "").strip()
+    if raw_project:
+        resolved, err = _project_or_404(raw_project)
+        if err:
+            return err
+        project_path = str(resolved)
+    store = _schedule_api()._get_store()
+    rows = _registered_rows(store.list(project_path=project_path or None))
+    return jsonify({"schedules": rows, "count": len(rows), "rev": store.rev(), "project": project_path})
+
+
+@bp.route("/schedules", methods=["POST"])
+def mobile_schedules_create():
+    gate = _schedule_write_gate()
+    if gate:
+        return gate
+    row, err = _schedule_call(_schedule_api().create_schedule, request.get_json(silent=True) or {},
+                              scanner=_scanner, by=_decided_by())
+    if err:
+        return err
+    _gw_audit("mobile.schedule.create", {"schedule_id": row["id"], "kind": row["kind"]})
+    return jsonify(row), 201
+
+
+@bp.route("/schedules/runs", methods=["GET"])
+def mobile_schedule_runs():
+    """Runs newest first. Query: project, schedule_id, status, limit."""
+    gate = _feature_or_404("schedules")
+    if gate:
+        return gate
+    project_path = ""
+    raw_project = (request.args.get("project") or "").strip()
+    if raw_project:
+        resolved, err = _project_or_404(raw_project)
+        if err:
+            return err
+        project_path = str(resolved)
+    status = (request.args.get("status") or "").strip() or None
+    store = _schedule_api()._get_store()
+    rows = store.runs(schedule_id=(request.args.get("schedule_id") or None), status=status,
+                      project_path=project_path or None, limit=_SOURCE_CAP)
+    rows = _registered_rows(rows)[:_limit_arg()]
+    return jsonify({"runs": rows, "count": len(rows), "project": project_path})
+
+
+@bp.route("/schedules/approvals", methods=["GET"])
+def mobile_schedule_approvals():
+    gate = _feature_or_404("schedules")
+    if gate:
+        return gate
+    status = (request.args.get("status") or "").strip() or None
+    rows = _schedule_api()._get_approvals().list(status=status, limit=_SOURCE_CAP)
+    rows = _registered_rows(rows)[:_limit_arg()]
+    return jsonify({"approvals": rows, "count": len(rows)})
+
+
+@bp.route("/schedules/approvals/<approval_id>/decide", methods=["POST"])
+def mobile_schedule_decide(approval_id):
+    """Approve or deny one pending scheduled run; 409 once it lapsed."""
+    gate = _schedule_write_gate()
+    if gate:
+        return gate
+    api = _schedule_api()
+    row = api._get_approvals().get(str(approval_id))
+    if row is None or _resolve_registered(str(row.get("project_path") or "")) is None:
+        return jsonify({"error": "unknown approval"}), 404
+    result, err = _schedule_call(api.decide_approval, str(approval_id), request.get_json(silent=True) or {},
+                                 by=_decided_by())
+    if err:
+        return err
+    _gw_audit(f"mobile.schedule.{result['status']}", {"approval_id": str(approval_id),
+                                                      "schedule_id": result.get("schedule_id")})
+    return jsonify(result)
+
+
+@bp.route("/schedules/<schedule_id>", methods=["GET"])
+def mobile_schedule_one(schedule_id):
+    gate = _feature_or_404("schedules")
+    if gate:
+        return gate
+    row, err = _schedule_row_or_404(schedule_id)
+    if err:
+        return err
+    return jsonify(dict(row, rev=_schedule_api()._get_store().rev()))
+
+
+@bp.route("/schedules/<schedule_id>", methods=["PATCH"])
+def mobile_schedule_update(schedule_id):
+    gate = _schedule_write_gate()
+    if gate:
+        return gate
+    _, err = _schedule_row_or_404(schedule_id)
+    if err:
+        return err
+    row, err = _schedule_call(_schedule_api().update_schedule, schedule_id, request.get_json(silent=True) or {},
+                              scanner=_scanner, by=_decided_by())
+    if err:
+        return err
+    _gw_audit("mobile.schedule.update", {"schedule_id": schedule_id})
+    return jsonify(row)
+
+
+@bp.route("/schedules/<schedule_id>", methods=["DELETE"])
+def mobile_schedule_delete(schedule_id):
+    gate = _schedule_write_gate()
+    if gate:
+        return gate
+    _, err = _schedule_row_or_404(schedule_id)
+    if err:
+        return err
+    _schedule_api()._get_store().delete(schedule_id, by=_decided_by())
+    _gw_audit("mobile.schedule.delete", {"schedule_id": schedule_id})
+    return jsonify({"deleted": schedule_id})
+
+
+@bp.route("/schedules/<schedule_id>/run", methods=["POST"])
+def mobile_schedule_run(schedule_id):
+    gate = _schedule_write_gate()
+    if gate:
+        return gate
+    _, err = _schedule_row_or_404(schedule_id)
+    if err:
+        return err
+    run, err = _schedule_call(_schedule_api().run_schedule, schedule_id, request.get_json(silent=True) or {},
+                              by=_decided_by())
+    if err:
+        return err
+    _gw_audit("mobile.schedule.run", {"schedule_id": schedule_id, "run_id": run["run_id"]})
+    return jsonify(run), 202
+
+
+@bp.route("/schedules/<schedule_id>/<action>", methods=["POST"])
+def mobile_schedule_toggle(schedule_id, action):
+    if action not in ("enable", "disable"):
+        return jsonify({"error": "unknown action"}), 404
+    gate = _schedule_write_gate()
+    if gate:
+        return gate
+    _, err = _schedule_row_or_404(schedule_id)
+    if err:
+        return err
+    row, err = _schedule_call(_schedule_api().update_schedule, schedule_id, {"enabled": action == "enable"},
+                              scanner=_scanner, by=_decided_by())
+    if err:
+        return err
+    _gw_audit(f"mobile.schedule.{action}", {"schedule_id": schedule_id})
+    return jsonify(row)
+
+
 @bp.route("/ci/runs", methods=["GET"])
 def mobile_ci_runs():
     """Local CI run index for one project, newest first.
diff --git a/oracle/services/schedule_api.py b/oracle/services/schedule_api.py
index 1c75cff..2b45aa6 100644
--- a/oracle/services/schedule_api.py
+++ b/oracle/services/schedule_api.py
@@ -92,13 +92,16 @@ def update_schedule(schedule_id: str, body: dict, *, scanner, by: str) -> dict:
 
 
 def run_schedule(schedule_id: str, body: dict, *, by: str) -> dict:
-    if _get_store().get(schedule_id) is None:
+    """Queue a manual run; the next tick (whichever process holds the lock) dispatches it."""
+    store = _get_store()
+    if store.get(schedule_id) is None:
         raise ApiError(f"no schedule {schedule_id}", 404)
     bypass = bool((body or {}).get("bypass_approval", False))
+    run = store.request_run(schedule_id, trigger="manual", by=by, approved=bypass)
     scheduler = _get_scheduler()
-    if scheduler is None:
-        raise ApiError("scheduler not running", 503)
-    return scheduler.run_now(schedule_id, bypass_approval=bypass, by=by)
+    if scheduler is not None:
+        scheduler.kick()
+    return run
 
 
 def decide_approval(approval_id: str, body: dict, *, by: str) -> dict:
diff --git a/oracle/services/scheduler.py b/oracle/services/scheduler.py
index b6c5ce5..4140555 100644
--- a/oracle/services/scheduler.py
+++ b/oracle/services/scheduler.py
@@ -104,9 +104,7 @@ def kick(self) -> None:
 
     def run_now(self, schedule_id: str, *, bypass_approval: bool = False, by: str = "") -> dict:
         """Queue a manual run; ``bypass_approval`` skips the approval gate once."""
-        run = self.store.request_run(schedule_id, trigger="manual", by=by)
-        if bypass_approval:
-            run = self.store.update_run(run["run_id"], status="approved")
+        run = self.store.request_run(schedule_id, trigger="manual", by=by, approved=bypass_approval)
         self.kick()
         return run
 
diff --git a/services/schedule_store.py b/services/schedule_store.py
index 93c6117..bab53dd 100644
--- a/services/schedule_store.py
+++ b/services/schedule_store.py
@@ -420,8 +420,9 @@ def claim_due(self, now: datetime | None = None) -> list[tuple[dict, dict]]:
                         due_at=run["due_at"], trigger="cron")
         return claimed
 
-    def request_run(self, schedule_id: str, *, trigger: str = "manual", by: str = "") -> dict:
-        """A ``claimed`` run outside the cadence (Run now). The tick dispatches it."""
+    def request_run(self, schedule_id: str, *, trigger: str = "manual", by: str = "",
+                    approved: bool = False) -> dict:
+        """A run outside the cadence (Run now); ``approved`` skips the approval gate once."""
         if trigger not in TRIGGERS:
             raise ScheduleError(f"trigger must be one of {', '.join(TRIGGERS)}")
         now = _now()
@@ -429,17 +430,19 @@ def request_run(self, schedule_id: str, *, trigger: str = "manual", by: str = ""
             row = _find(doc, schedule_id)
             if row is None:
                 raise ScheduleError(f"no schedule {schedule_id}")
-            run = self._new_run(row, trigger=trigger, due_at=_iso(now), now=now)
+            run = self._new_run(row, trigger=trigger, due_at=_iso(now), now=now,
+                                status="approved" if approved else "claimed")
         self._event("run.claimed", run_id=run["run_id"], schedule_id=schedule_id,
-                    due_at=run["due_at"], trigger=trigger, by=by)
+                    due_at=run["due_at"], trigger=trigger, by=by, status=run["status"])
         return run
 
-    def _new_run(self, row: dict, *, trigger: str, due_at: str, now: datetime) -> dict:
+    def _new_run(self, row: dict, *, trigger: str, due_at: str, now: datetime,
+                 status: str = "claimed") -> dict:
         run = {k: None for k in _RUN_FIELDS}
         run.update({
             "run_id": _new_run_id(), "schedule_id": row["id"], "schedule_name": row.get("name", ""),
             "project_path": row.get("project_path", ""), "trigger": trigger, "due_at": due_at,
-            "claimed_at": _iso(now), "status": "claimed", "tokens": {}, "error": "",
+            "claimed_at": _iso(now), "status": status, "tokens": {}, "error": "",
         })
         self._write_run(run)
         return run
diff --git a/tests/test_hub_schedules_routes.py b/tests/test_hub_schedules_routes.py
new file mode 100644
index 0000000..ac651d5
--- /dev/null
+++ b/tests/test_hub_schedules_routes.py
@@ -0,0 +1,111 @@
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+import cli.hub_server as hub_server  # noqa: E402
+from oracle.services import schedule_api  # noqa: E402
+from services import schedule_approvals as sa  # noqa: E402
+from services import schedule_store as ss  # noqa: E402
+
+
+class TestHubScheduleRoutes(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.proj = Path(self._tmp.name) / "proj"
+        (self.proj / ".c3").mkdir(parents=True)
+        self.store = ss.ScheduleStore(Path(self._tmp.name) / "oracle")
+        self.approvals = sa.ApprovalStore(Path(self._tmp.name) / "oracle")
+        self._prior_api = dict(get_scheduler=schedule_api._get_scheduler, get_store=schedule_api._get_store,
+                               get_approvals=schedule_api._get_approvals, get_scanner=schedule_api._get_scanner,
+                               get_cfg=schedule_api._get_cfg)
+        schedule_api.configure(get_scheduler=lambda: None, get_store=lambda: self.store,
+                               get_approvals=lambda: self.approvals, get_scanner=lambda: None,
+                               get_cfg=lambda: {})
+        hub_server.app.config["TESTING"] = True
+        self.client = hub_server.app.test_client()
+
+        def resolve(raw):
+            if Path(raw) == self.proj:
+                return str(self.proj)
+            raise ValueError(f"unknown project {raw}")
+
+        self._resolve = mock.patch.object(hub_server, "_resolve_project_path", side_effect=resolve)
+        self._resolve.start()
+        self._notify = mock.patch("services.schedule_approvals.notify")
+        self._notify.start()
+
+    def tearDown(self):
+        self._notify.stop()
+        self._resolve.stop()
+        schedule_api.configure(**self._prior_api)
+        self._tmp.cleanup()
+
+    def body(self, **over):
+        base = {"name": "report", "kind": "command", "target": {"cmd": "echo hi"}, "project": str(self.proj),
+                "when": {"type": "preset", "preset": "weekdays", "at": "09:00", "tz": "UTC"},
+                "autonomy": "approval"}
+        base.update(over)
+        return base
+
+    def test_create_list_and_names(self):
+        resp = self.client.post("/api/hub/schedules", json=self.body())
+        self.assertEqual(resp.status_code, 201, resp.get_json())
+        row = resp.get_json()
+        self.assertEqual(row["when"]["expr"], "0 9 * * 1-5")
+        listed = self.client.get("/api/hub/schedules").get_json()
+        self.assertEqual(listed["schedules"][0]["project_name"], "proj")
+        self.assertEqual(self.client.get(f"/api/hub/schedules?path={self.proj}").get_json()["count"], 1)
+        self.assertEqual(self.client.get("/api/hub/schedules?path=/nowhere").status_code, 404)
+        self.assertEqual(self.client.post("/api/hub/schedules", json=self.body(project="/nowhere")).status_code, 404)
+        self.assertEqual(self.client.post("/api/hub/schedules", json=self.body(effort="ultra")).status_code, 400)
+
+    def test_update_toggle_run_delete(self):
+        row = self.client.post("/api/hub/schedules", json=self.body()).get_json()
+        one = self.client.get(f"/api/hub/schedules/{row['id']}").get_json()
+        patched = self.client.patch(f"/api/hub/schedules/{row['id']}",
+                                    json={"name": "renamed", "expected_rev": one["rev"]})
+        self.assertEqual(patched.get_json()["name"], "renamed")
+        stale = self.client.patch(f"/api/hub/schedules/{row['id']}", json={"name": "x", "expected_rev": one["rev"]})
+        self.assertEqual(stale.status_code, 409)
+        self.assertFalse(self.client.post(f"/api/hub/schedules/{row['id']}/disable").get_json()["enabled"])
+        self.assertTrue(self.client.post(f"/api/hub/schedules/{row['id']}/enable").get_json()["enabled"])
+        self.assertEqual(self.client.post(f"/api/hub/schedules/{row['id']}/bogus").status_code, 404)
+        run = self.client.post(f"/api/hub/schedules/{row['id']}/run", json={"bypass_approval": True})
+        self.assertEqual(run.status_code, 202)
+        self.assertEqual(run.get_json()["status"], "approved")
+        runs = self.client.get(f"/api/hub/schedules/{row['id']}/runs").get_json()
+        self.assertEqual(runs["count"], 1)
+        self.assertEqual(self.client.get("/api/hub/schedules/runs?status=approved").get_json()["count"], 1)
+        self.assertEqual(self.client.delete(f"/api/hub/schedules/{row['id']}").status_code, 200)
+        self.assertEqual(self.client.get(f"/api/hub/schedules/{row['id']}").status_code, 404)
+        self.assertEqual(self.client.post("/api/hub/schedules/sch-none/run", json={}).status_code, 404)
+
+    def test_approvals_decide_as_desktop(self):
+        row = self.client.post("/api/hub/schedules", json=self.body()).get_json()
+        run = self.store.request_run(row["id"])
+        pending = self.approvals.create(row, run, ttl_s=600)
+        listed = self.client.get("/api/hub/schedules/approvals?status=pending").get_json()
+        self.assertEqual(listed["approvals"][0]["project_name"], "proj")
+        ok = self.client.post(f"/api/hub/schedules/approvals/{pending['id']}", json={"decision": "deny", "note": "no"})
+        self.assertEqual(ok.status_code, 200)
+        self.assertEqual((ok.get_json()["status"], ok.get_json()["decided_by"]), ("denied", "desktop"))
+        self.assertEqual(self.client.post(f"/api/hub/schedules/approvals/{pending['id']}",
+                                          json={"decision": "approve"}).status_code, 409)
+
+    def test_preview_and_status_unreachable(self):
+        resp = self.client.post("/api/hub/schedules/preview",
+                                json={"when": {"type": "cron", "expr": "*/30 * * * *", "tz": "UTC"}, "count": 2})
+        self.assertEqual(len(resp.get_json()["next"]), 2)
+        with mock.patch.object(hub_server, "_read_hub_config", return_value={"oracle_url": "http://127.0.0.1:9"}):
+            status = self.client.get("/api/hub/schedules/status").get_json()
+        self.assertFalse(status["reachable"])
+        self.assertFalse(status["running"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_mobile_clients.py b/tests/test_mobile_clients.py
index 17cd9f9..8fc35d5 100644
--- a/tests/test_mobile_clients.py
+++ b/tests/test_mobile_clients.py
@@ -352,7 +352,7 @@ def test_client_token_authenticates_and_info_names_the_principal(self):
         r = self.client.get("/api/mobile/info", headers=self.bearer(tok))
         self.assertEqual(r.status_code, 200)
         body = r.get_json()
-        self.assertEqual(body["api_version"], 5)
+        self.assertEqual(body["api_version"], 6)
         self.assertIn("clients", body["capabilities"])
         self.assertEqual(body["client"], {"kind": "desk", "client_id": cid})
 
diff --git a/tests/test_mobile_override_costs.py b/tests/test_mobile_override_costs.py
index 26344b9..174d365 100644
--- a/tests/test_mobile_override_costs.py
+++ b/tests/test_mobile_override_costs.py
@@ -112,7 +112,7 @@ def test_capability_is_advertised_and_follows_the_override_switch(self):
 
     def test_api_version_is_not_bumped(self):
         # Additive: an older client must not be told the API moved.
-        self.assertEqual(mobile_api.API_VERSION, 5)
+        self.assertEqual(mobile_api.API_VERSION, 6)
 
     def test_switch_off_is_404(self):
         srv._cfg["mobile_override_enabled"] = False
diff --git a/tests/test_mobile_schedules.py b/tests/test_mobile_schedules.py
new file mode 100644
index 0000000..80eb95a
--- /dev/null
+++ b/tests/test_mobile_schedules.py
@@ -0,0 +1,144 @@
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+import oracle.oracle_server as srv  # noqa: E402
+from oracle.services import mobile_api, schedule_api  # noqa: E402
+from services import schedule_approvals as sa  # noqa: E402
+from services import schedule_store as ss  # noqa: E402
+
+AUTH = {"Authorization": "Bearer mobile-sched-key"}
+
+
+class _StubScanner:
+    def __init__(self, paths):
+        self.paths = paths
+
+    def discover(self, force=False):
+        return [{"path": str(p), "name": p.name, "tags": [], "active": False, "has_c3": True, "fact_count": 0}
+                for p in self.paths]
+
+
+class TestMobileSchedules(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.tmp = Path(tempfile.mkdtemp(prefix="c3-mobile-sched-"))
+        cls.alpha = cls.tmp / "alpha"
+        cls.outsider = cls.tmp / "outsider"
+        for p in (cls.alpha, cls.outsider):
+            (p / ".c3").mkdir(parents=True)
+        cls._env = mock.patch.dict(os.environ, {"C3_ORACLE_API_KEY": "mobile-sched-key"})
+        cls._env.start()
+        cls._prior_cfg = srv._cfg
+        srv._cfg = {"mobile_api_enabled": True, "api_rate_limit_per_min": 0,
+                    "mobile_security_rate_limit_per_min": 0}
+        mobile_api.init_services(scanner=_StubScanner([cls.alpha]))
+        cls.store = ss.ScheduleStore(cls.tmp / "oracle")
+        cls.approvals = sa.ApprovalStore(cls.tmp / "oracle")
+        cls._prior_api = dict(get_scheduler=schedule_api._get_scheduler, get_store=schedule_api._get_store,
+                              get_approvals=schedule_api._get_approvals, get_scanner=schedule_api._get_scanner,
+                              get_cfg=schedule_api._get_cfg)
+        schedule_api.configure(get_scheduler=lambda: None, get_store=lambda: cls.store,
+                               get_approvals=lambda: cls.approvals, get_scanner=lambda: mobile_api._scanner,
+                               get_cfg=lambda: srv._cfg)
+        cls.client = srv.app.test_client()
+        cls.notify = mock.patch("services.schedule_approvals.notify")
+        cls.notify.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.notify.stop()
+        schedule_api.configure(**cls._prior_api)
+        srv._cfg = cls._prior_cfg
+        cls._env.stop()
+
+    def body(self, **over):
+        base = {"name": "report", "kind": "skill", "target": {"skill": "/daily_report"},
+                "project_path": str(self.alpha), "when": {"type": "every", "seconds": 3600},
+                "autonomy": "autonomous"}
+        base.update(over)
+        return base
+
+    def test_info_advertises_and_switch_hides(self):
+        info = self.client.get("/api/mobile/info", headers=AUTH).get_json()
+        self.assertGreaterEqual(info["api_version"], 6)
+        self.assertIn("schedules", info["capabilities"])
+        self.assertIn("schedules_write", info["capabilities"])
+        with mock.patch.dict(srv._cfg, {"mobile_schedules_enabled": False}):
+            info = self.client.get("/api/mobile/info", headers=AUTH).get_json()
+            self.assertNotIn("schedules", info["capabilities"])
+            self.assertEqual(self.client.get("/api/mobile/schedules", headers=AUTH).status_code, 404)
+        with mock.patch.dict(srv._cfg, {"mobile_schedules_write": False}):
+            self.assertEqual(self.client.post("/api/mobile/schedules", json=self.body(), headers=AUTH).status_code, 404)
+            self.assertEqual(self.client.get("/api/mobile/schedules", headers=AUTH).status_code, 200)
+
+    def test_bearer_required_on_get(self):
+        self.assertEqual(self.client.get("/api/mobile/schedules").status_code, 401)
+
+    def test_create_list_update_run_delete(self):
+        resp = self.client.post("/api/mobile/schedules", json=self.body(name="m1"), headers=AUTH)
+        self.assertEqual(resp.status_code, 201, resp.get_json())
+        row = resp.get_json()
+        listed = self.client.get("/api/mobile/schedules", headers=AUTH).get_json()
+        self.assertIn(row["id"], [r["id"] for r in listed["schedules"]])
+        one = self.client.get(f"/api/mobile/schedules/{row['id']}", headers=AUTH).get_json()
+        self.assertEqual(one["name"], "m1")
+        patched = self.client.patch(f"/api/mobile/schedules/{row['id']}", json={"name": "m2"}, headers=AUTH)
+        self.assertEqual(patched.get_json()["name"], "m2")
+        off = self.client.post(f"/api/mobile/schedules/{row['id']}/disable", headers=AUTH).get_json()
+        self.assertFalse(off["enabled"])
+        self.assertEqual(self.client.post(f"/api/mobile/schedules/{row['id']}/explode", headers=AUTH).status_code, 404)
+        run = self.client.post(f"/api/mobile/schedules/{row['id']}/run", json={}, headers=AUTH)
+        self.assertEqual(run.status_code, 202)
+        runs = self.client.get(f"/api/mobile/schedules/runs?schedule_id={row['id']}", headers=AUTH).get_json()
+        self.assertEqual([r["run_id"] for r in runs["runs"]], [run.get_json()["run_id"]])
+        self.assertEqual(self.client.delete(f"/api/mobile/schedules/{row['id']}", headers=AUTH).status_code, 200)
+        self.assertEqual(self.client.get(f"/api/mobile/schedules/{row['id']}", headers=AUTH).status_code, 404)
+
+    def test_unregistered_project_rows_are_hidden(self):
+        resp = self.client.post("/api/mobile/schedules", json=self.body(project_path=str(self.outsider)), headers=AUTH)
+        self.assertEqual(resp.status_code, 404)
+        hidden = self.store.create(self.body(name="hidden", project_path=str(self.outsider)))
+        ids = [r["id"] for r in self.client.get("/api/mobile/schedules", headers=AUTH).get_json()["schedules"]]
+        self.assertNotIn(hidden["id"], ids)
+        self.assertEqual(self.client.get(f"/api/mobile/schedules/{hidden['id']}", headers=AUTH).status_code, 404)
+        self.assertEqual(self.client.post(f"/api/mobile/schedules/{hidden['id']}/run", json={},
+                                          headers=AUTH).status_code, 404)
+        run = self.store.request_run(hidden["id"])
+        runs = self.client.get("/api/mobile/schedules/runs", headers=AUTH).get_json()
+        self.assertNotIn(run["run_id"], [r["run_id"] for r in runs["runs"]])
+        self.assertEqual(self.client.get("/api/mobile/schedules?project=" + str(self.outsider),
+                                         headers=AUTH).status_code, 404)
+
+    def test_approvals_decide_and_409(self):
+        row = self.store.create(self.body(name="gated", autonomy="approval"))
+        run = self.store.request_run(row["id"])
+        pending = self.approvals.create(row, run, ttl_s=600)
+        listed = self.client.get("/api/mobile/schedules/approvals?status=pending", headers=AUTH).get_json()
+        self.assertIn(pending["id"], [a["id"] for a in listed["approvals"]])
+        ok = self.client.post(f"/api/mobile/schedules/approvals/{pending['id']}/decide",
+                              json={"decision": "approve"}, headers=AUTH)
+        self.assertEqual(ok.status_code, 200, ok.get_json())
+        self.assertEqual((ok.get_json()["status"], ok.get_json()["decided_by"]), ("approved", "mobile"))
+        again = self.client.post(f"/api/mobile/schedules/approvals/{pending['id']}/decide",
+                                 json={"decision": "deny"}, headers=AUTH)
+        self.assertEqual(again.status_code, 409)
+        self.assertEqual(self.client.post("/api/mobile/schedules/approvals/sa-none/decide",
+                                          json={"decision": "approve"}, headers=AUTH).status_code, 404)
+        foreign = self.store.create(self.body(name="foreign", autonomy="approval", project_path=str(self.outsider)))
+        frun = self.store.request_run(foreign["id"])
+        fpending = self.approvals.create(foreign, frun, ttl_s=600)
+        self.assertEqual(self.client.post(f"/api/mobile/schedules/approvals/{fpending['id']}/decide",
+                                          json={"decision": "approve"}, headers=AUTH).status_code, 404)
+        self.assertEqual(json.loads(json.dumps(self.approvals.get(fpending["id"])))["status"], "pending")
+
+
+if __name__ == "__main__":
+    unittest.main()