From a7880577a7b46392677a6bfaecd7af001a8a3d37 Mon Sep 17 00:00:00 2001 From: drknowhow Date: Tue, 13 Oct 2026 09:54:12 -0400 Subject: [PATCH 1/4] feat(schedule): approval inbox and the scheduler tick loop services/schedule_approvals.py holds pending approvals for approval-mode runs (one pending row per schedule, human-only decide, TTL sweep, kind= schedule notifications). oracle/services/scheduler.py is the tick: lapse approvals, fold finished jobs (JobStore.find), claim_due, then route each run through the misfire window, overlap rule, check_command and autonomy gate, dispatching as detached jobs so the tick never blocks. A kernel-held scheduler.lock keeps a second Oracle in standby. Not started anywhere yet. Verified: 21 tests incl. non-blocking dispatch and two-instance standby. Claude-Session: https://claude.ai/code/session_01JUcvmQq81aLo9ewD9Xc4qY --- oracle/services/scheduler.py | 322 +++++++++++++++++++++++++++++++ services/notifications.py | 4 +- services/schedule_approvals.py | 202 +++++++++++++++++++ services/shell_jobs.py | 5 + tests/test_schedule_approvals.py | 86 +++++++++ tests/test_scheduler_tick.py | 257 ++++++++++++++++++++++++ 6 files changed, 875 insertions(+), 1 deletion(-) create mode 100644 oracle/services/scheduler.py create mode 100644 services/schedule_approvals.py create mode 100644 tests/test_schedule_approvals.py create mode 100644 tests/test_scheduler_tick.py diff --git a/oracle/services/scheduler.py b/oracle/services/scheduler.py new file mode 100644 index 0000000..b6c5ce5 --- /dev/null +++ b/oracle/services/scheduler.py @@ -0,0 +1,322 @@ +"""The Oracle's scheduler: one tick thread that claims due schedules and fires them. + +Each tick: lapse stale approvals, fold finished jobs into their runs, claim +what is due, route each claimed run through the overlap, misfire, check and +autonomy gates, dispatch approved runs, sweep old run files. A fire is a +detached ``services.shell_jobs`` job, so the tick never waits on one. + +Only one scheduler per machine may dispatch: ``start`` takes +``scheduler.lock`` and a loser stays in ``standby``, retrying every tick, +because two Oracles can share a port on Windows and both would otherwise +fire every schedule. +""" +from __future__ import annotations + +import logging +import threading +from datetime import datetime, timezone + +from oracle.config import load_config +from services import schedule_runner as runner +from services.activity_log import ActivityLog +from services.notifications import notify +from services.schedule_approvals import ApprovalStore +from services.schedule_store import EXECUTED_STATUSES, ScheduleError, ScheduleStore, _FileLock, parse_iso + +log = logging.getLogger("oracle.scheduler") + +INITIAL_DELAY_S = 5 +SKIP_POLICY_LATENESS_S = 60 +RUN_SEVERITY = {"done": "info", "check_skipped": "info", "skipped_overlap": "info", + "denied": "info", "expired": "info", "missed": "warning"} + + +def defaults_from_config(cfg: dict) -> dict: + """The ``normalize_schedule`` defaults a config dict implies.""" + return { + "timeout_s": int(cfg.get("scheduler_default_timeout_s", 1800)), + "max_timeout_s": int(cfg.get("scheduler_max_timeout_s", 21600)), + "approval_ttl_s": int(cfg.get("scheduler_approval_ttl_s", 3600)), + "misfire_grace_s": int(cfg.get("scheduler_misfire_grace_s", 900)), + "permission_mode": str(cfg.get("scheduler_default_permission_mode", "acceptEdits")), + "allow_bypass_permissions": bool(cfg.get("scheduler_allow_bypass_permissions", False)), + "max_budget_usd": float(cfg.get("scheduler_default_max_budget_usd", 0) or 0), + } + + +class Scheduler: + """Tick loop over a ``ScheduleStore``; see the module docstring.""" + + def __init__(self, *, store: ScheduleStore, approvals: ApprovalStore, jobstore, + config=load_config, interval: int = 30, lock_path=None): + self.store = store + self.approvals = approvals + self.jobstore = jobstore + self.config = config + self.interval = max(5, int(interval)) + self._lock = _FileLock(lock_path or (store.root / "scheduler.lock")) + self._lock_held = False + self._stop = threading.Event() + self._wake = threading.Event() + self._thread: threading.Thread | None = None + self._tick_guard = threading.Lock() + self._running = False + self._last_tick_at: str | None = None + self._last_summary: dict = {} + + @property + def status(self) -> dict: + schedules = self.store.list(enabled=True) + due = sorted(r["next_run_at"] for r in schedules if r.get("next_run_at")) + return { + "running": self._running, + "standby": self._running and not self._lock_held, + "lock_held": self._lock_held, + "interval_seconds": self.interval, + "last_tick_at": self._last_tick_at, + "last_tick": self._last_summary, + "enabled_schedules": len(schedules), + "next_due": due[0] if due else None, + "open_runs": len(self.store.open_runs()), + "pending_approvals": len(self.approvals.list(status="pending")), + } + + def start(self) -> None: + if self._thread and self._thread.is_alive(): + return + self._stop.clear() + self._thread = threading.Thread(target=self._loop, daemon=True, name="oracle-scheduler") + self._thread.start() + self._running = True + log.info("Scheduler started (interval=%ds)", self.interval) + + def stop(self) -> None: + self._stop.set() + self._wake.set() + self._running = False + if self._thread: + self._thread.join(timeout=5) + self._release_lock() + log.info("Scheduler stopped") + + def kick(self) -> None: + self._wake.set() + + 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") + self.kick() + return run + + def _loop(self) -> None: + self._wake.wait(INITIAL_DELAY_S) + while not self._stop.is_set(): + self._wake.clear() + try: + self.tick() + except Exception as exc: + log.error("Scheduler tick failed: %s", exc) + self._wake.wait(self.interval) + + def _acquire_lock(self) -> bool: + if self._lock_held: + return True + self._lock_held = self._lock.acquire(timeout_s=0.2) + if not self._lock_held: + log.warning("Scheduler standby: another scheduler holds %s", self._lock.path) + return self._lock_held + + def _release_lock(self) -> None: + if self._lock_held: + self._lock.release() + self._lock_held = False + + def tick(self, now: datetime | None = None) -> dict: + """One pass; returns counts. Safe to call from a request thread.""" + now = now or datetime.now(timezone.utc) + cfg = self.config() or {} + summary = {"standby": False, "expired": 0, "finished": 0, "claimed": 0, "dispatched": 0, + "awaiting_approval": 0, "skipped": 0, "swept": 0} + if not bool(cfg.get("scheduler_enabled", True)): + summary["disabled"] = True + return summary + if not self._acquire_lock(): + summary["standby"] = True + return summary + with self._tick_guard: + self._last_tick_at = now.isoformat() + self._step(summary, "expired", lambda: self._expire_approvals(now)) + self._step(summary, "finished", lambda: self._reconcile_jobs()) + self._step(summary, "claimed", lambda: len(self.store.claim_due(now))) + self._step(summary, None, lambda: self._route_claimed(now, cfg, summary)) + self._step(summary, None, lambda: self._route_decided(summary)) + self._step(summary, "swept", lambda: self.store.sweep_runs( + int(cfg.get("scheduler_run_retention_days", 30)), now)) + self._last_summary = summary + return summary + + @staticmethod + def _step(summary: dict, key, fn) -> None: + try: + result = fn() + except Exception as exc: + log.error("Scheduler step %s failed: %s", key or fn.__name__, exc) + summary.setdefault("errors", []).append(f"{key or 'route'}: {exc}") + return + if key is not None and isinstance(result, int): + summary[key] += result + + def _expire_approvals(self, now: datetime) -> int: + expired = self.approvals.sweep_expired(now) + for row in expired: + run = self.store.get_run(row.get("run_id") or "") + if run and run.get("status") == "awaiting_approval": + self._finish(run, "expired") + return len(expired) + + def _reconcile_jobs(self) -> int: + try: + self.jobstore.reap(session_id=runner.SESSION_ID) + except Exception as exc: + log.warning("job reap failed: %s", exc) + finished = 0 + for run in self.store.runs(status=("dispatched", "running", "check_running"), limit=0): + job = self.jobstore.find(run.get("job_id") or "") if run.get("job_id") else None + if job is None: + self._finish(run, "lost", error="job record missing") + finished += 1 + continue + if not job.terminal: + if run.get("status") == "dispatched" and job.status == "running": + self.store.update_run(run["run_id"], status="running") + continue + schedule = self.store.get(run.get("schedule_id", "")) or {} + if run.get("status") == "check_running": + self._settle_check(run, job, schedule) + else: + parser = "command" if schedule.get("kind") == "command" else str(schedule.get("provider") or "claude") + done = runner.finalize(self.store, self.jobstore, run, job, parser=parser) + self._announce(done) + finished += 1 + return finished + + def _settle_check(self, run: dict, job, schedule: dict) -> None: + if job.status != "done" and job.exit_code != 1: + self._finish(run, "failed", job_id=job.id, exit_code=job.exit_code, + error=f"check_command ended {job.status}: {job.error or ''}".strip()) + return + if job.exit_code == 0: + self._finish(run, "check_skipped", job_id=job.id, exit_code=0, duration_ms=job.duration_ms) + return + if not schedule: + self._finish(run, "failed", error="schedule deleted while its check ran") + return + run = self.store.update_run(run["run_id"], status="claimed", job_id=None, exit_code=None) + self._gate(schedule, run, self.config() or {}) + + def _route_claimed(self, now: datetime, cfg: dict, summary: dict) -> None: + for run in self.store.runs(status="claimed", limit=0): + schedule = self.store.get(run.get("schedule_id", "")) + if schedule is None: + self._finish(run, "failed", error="schedule deleted before it ran") + continue + if run.get("trigger") == "cron" and self._misfired(schedule, run, now): + self._finish(run, "missed", error="fired outside the misfire window") + summary["skipped"] += 1 + continue + if schedule.get("overlap", "skip") == "skip" and self.store.open_run_for( + schedule["id"], exclude=run["run_id"]): + self._finish(run, "skipped_overlap", error="previous run still open") + summary["skipped"] += 1 + continue + if schedule.get("check_command"): + runner.dispatch(self.store, self.jobstore, schedule, run, check=True) + continue + self._gate(schedule, run, cfg, summary) + + @staticmethod + def _misfired(schedule: dict, run: dict, now: datetime) -> bool: + try: + late = (now - parse_iso(run.get("due_at") or "")).total_seconds() + except ValueError: + return False + misfire = schedule.get("misfire") or {} + if misfire.get("policy") == "skip": + return late > SKIP_POLICY_LATENESS_S + return late > int(misfire.get("grace_s") or 0) + + def _gate(self, schedule: dict, run: dict, cfg: dict, summary: dict | None = None) -> None: + if schedule.get("autonomy") == "approval": + ttl = int(schedule.get("approval_ttl_s") or cfg.get("scheduler_approval_ttl_s", 3600)) + row = self.approvals.create(schedule, run, ttl_s=ttl) + if row.get("duplicate"): + self._finish(run, "skipped_overlap", error=f"approval {row['id']} already pending") + return + self.store.update_run(run["run_id"], status="awaiting_approval", approval_id=row["id"]) + if summary is not None: + summary["awaiting_approval"] += 1 + return + self._dispatch(schedule, run, summary) + + def _dispatch(self, schedule: dict, run: dict, summary: dict | None) -> None: + done = runner.dispatch(self.store, self.jobstore, schedule, run) + if done.get("status") in EXECUTED_STATUSES: + self._announce(done) + elif summary is not None: + summary["dispatched"] += 1 + + def _route_decided(self, summary: dict) -> None: + for run in self.store.runs(status=("awaiting_approval", "approved"), limit=0): + schedule = self.store.get(run.get("schedule_id", "")) + if schedule is None: + self._finish(run, "failed", error="schedule deleted while awaiting approval") + continue + if run.get("status") == "approved": + self._dispatch(schedule, run, summary) + continue + row = self.approvals.get(run.get("approval_id") or "") + state = (row or {}).get("status") + if state == "approved": + self.store.update_run(run["run_id"], status="approved") + self._dispatch(schedule, dict(run, status="approved"), summary) + elif state == "denied": + self._finish(run, "denied", error=f"denied by {row.get('decided_by') or 'unknown'}") + elif state in (None, "expired"): + self._finish(run, "expired") + + def _finish(self, run: dict, status: str, **extra) -> dict: + try: + done = self.store.finish_run(run["run_id"], status, **extra) + except ScheduleError as exc: + log.warning("finish %s as %s failed: %s", run.get("run_id"), status, exc) + return run + self._announce(done) + return done + + def _announce(self, run: dict) -> None: + status = str(run.get("status") or "") + name = run.get("schedule_name") or run.get("schedule_id") or "schedule" + parts = [f"{status}"] + if run.get("exit_code") is not None: + parts.append(f"exit {run['exit_code']}") + if run.get("duration_ms") is not None: + parts.append(f"{(run['duration_ms'] or 0) / 1000:.1f}s") + if run.get("cost_usd") is not None: + parts.append(f"${float(run['cost_usd']):.3f}") + if run.get("error"): + parts.append(str(run["error"])[:200]) + severity = RUN_SEVERITY.get(status, "warning") + notify(run.get("project_path") or "", agent="scheduler", severity=severity, + title=f"Scheduled run {status}: {name}", message=" — ".join(parts), + kind="schedule", ref_id=run["run_id"]) + try: + ActivityLog(run.get("project_path") or "").log("schedule_run", { + "run_id": run["run_id"], "schedule_id": run.get("schedule_id"), "name": name, + "status": status, "trigger": run.get("trigger"), "job_id": run.get("job_id"), + "exit_code": run.get("exit_code"), "duration_ms": run.get("duration_ms"), + "cost_usd": run.get("cost_usd"), "output_id": run.get("output_id"), + }) + except (OSError, ValueError) as exc: + log.debug("activity log skipped for %s: %s", run["run_id"], exc) diff --git a/services/notifications.py b/services/notifications.py index a3a2396..caf6c36 100644 --- a/services/notifications.py +++ b/services/notifications.py @@ -26,7 +26,9 @@ # session ref_id = host session (cli/hook_session_open.py, hook_session_end.py) # mcp ref_id = C3 session id (cli/mcp_server.py, runtime ready) # override ref_id = request id (services/override_requests.py) -EVENT_KINDS = ("shell_job", "ci", "session", "mcp", "override") +# schedule ref_id = run id, or approval id for a pending approval +# (oracle/services/scheduler.py) +EVENT_KINDS = ("shell_job", "ci", "session", "mcp", "override", "schedule") def notify(project_path, agent: str, severity: str, title: str, message: str, diff --git a/services/schedule_approvals.py b/services/schedule_approvals.py new file mode 100644 index 0000000..61edee6 --- /dev/null +++ b/services/schedule_approvals.py @@ -0,0 +1,202 @@ +"""Pending approvals for scheduled runs whose schedule says ``autonomy: approval``. + +A row is created when such a run is claimed, lands in the project's +notification feed (``kind="schedule"``) so the Hub, the Desk and the phone can +show it, and is decided by a human only. The scheduler polls ``approved`` rows +and dispatches them; ``sweep_expired`` lapses the rest at their TTL. Spawning +never happens here. +""" +from __future__ import annotations + +import json +import threading +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from services.atomic_json import write_json_atomic +from services.notifications import notify +from services.schedule_store import _FileLock, _path_key, default_root, parse_iso + +STATUS_PENDING = "pending" +STATUS_APPROVED = "approved" +STATUS_DENIED = "denied" +STATUS_EXPIRED = "expired" +STATUSES = (STATUS_PENDING, STATUS_APPROVED, STATUS_DENIED, STATUS_EXPIRED) +DECISIONS = ("approve", "deny") +NOTE_CAP = 400 +RETAIN_DECIDED_DAYS = 30 + + +class ApprovalError(ValueError): + """The approval does not exist or is no longer pending.""" + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _iso(dt: datetime) -> str: + return dt.astimezone(timezone.utc).isoformat() + + +class ApprovalStore: + """``schedule_approvals.json`` under the Oracle root; see the module docstring.""" + + def __init__(self, root: Path | str | None = None): + self.root = Path(root) if root else default_root() + self.file = self.root / "schedule_approvals.json" + self.lock_file = self.root / "schedule_approvals.lock" + self._lock = threading.Lock() + + def _load(self) -> list[dict]: + if not self.file.exists(): + return [] + try: + rows = json.loads(self.file.read_text(encoding="utf-8")) + except (OSError, ValueError): + return [] + return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else [] + + def _save(self, rows: list[dict]) -> None: + write_json_atomic(self.file, rows, ensure_ascii=False) + + def _txn(self): + return _Txn(self) + + def list(self, *, status: str | None = None, project_path: str | None = None, + limit: int = 100) -> list[dict]: + rows = self._load() + if status: + rows = [r for r in rows if r.get("status") == status] + if project_path: + key = _path_key(project_path) + rows = [r for r in rows if _path_key(r.get("project_path", "")) == key] + rows.sort(key=lambda r: r.get("created_at") or "", reverse=True) + return rows[:limit] if limit else rows + + def get(self, approval_id: str) -> dict | None: + for row in self._load(): + if row.get("id") == approval_id: + return row + return None + + def pending_for(self, schedule_id: str) -> dict | None: + for row in self._load(): + if row.get("schedule_id") == schedule_id and row.get("status") == STATUS_PENDING: + return row + return None + + def create(self, schedule: dict, run: dict, *, ttl_s: int) -> dict: + """A pending row for ``run``; an existing pending row for the same schedule is returned instead.""" + now = _now() + with self._txn() as rows: + for row in rows: + if row.get("schedule_id") == schedule.get("id") and row.get("status") == STATUS_PENDING: + return dict(row, duplicate=True) + row = { + "id": "sa-" + uuid.uuid4().hex[:10], + "schedule_id": schedule.get("id"), "schedule_name": schedule.get("name", ""), + "run_id": run.get("run_id"), "project_path": schedule.get("project_path", ""), + "summary": _summary(schedule), "status": STATUS_PENDING, + "created_at": _iso(now), "expires_at": _iso(now + timedelta(seconds=max(60, int(ttl_s)))), + "resolved_at": None, "decided_by": "", "note": "", + } + rows.append(row) + notify(row["project_path"], agent="scheduler", severity="warning", + title=f"Approve scheduled run: {row['schedule_name']}?", + message=f"{row['summary']} — expires {row['expires_at']}", + kind="schedule", ref_id=row["id"]) + return row + + def decide(self, approval_id: str, decision: str, *, decided_by: str, note: str = "") -> dict: + """Human decision on a pending row. Raises ``ApprovalError`` when it is not pending.""" + if decision not in DECISIONS: + raise ApprovalError(f"decision must be one of {', '.join(DECISIONS)}") + now = _now() + with self._txn() as rows: + row = next((r for r in rows if r.get("id") == approval_id), None) + if row is None: + raise ApprovalError(f"no approval {approval_id}") + if row.get("status") != STATUS_PENDING: + raise ApprovalError(f"approval {approval_id} is already {row.get('status')}") + if parse_iso(row["expires_at"]) <= now: + row["status"] = STATUS_EXPIRED + row["resolved_at"] = _iso(now) + raise ApprovalError(f"approval {approval_id} expired at {row['expires_at']}") + row["status"] = STATUS_APPROVED if decision == "approve" else STATUS_DENIED + row["resolved_at"] = _iso(now) + row["decided_by"] = str(decided_by or "")[:80] + row["note"] = str(note or "")[:NOTE_CAP] + result = dict(row) + notify(result["project_path"], agent="scheduler", severity="info", + title=f"Scheduled run {result['status']}: {result['schedule_name']}", + message=f"by {result['decided_by'] or 'unknown'}" + (f" — {result['note']}" if result["note"] else ""), + kind="schedule", ref_id=result["id"]) + return result + + def sweep_expired(self, now: datetime | None = None) -> list[dict]: + """Lapse pending rows past ``expires_at`` and drop decided rows older than the retention.""" + now = now or _now() + cutoff = now - timedelta(days=RETAIN_DECIDED_DAYS) + expired: list[dict] = [] + with self._txn() as rows: + keep = [] + for row in rows: + if row.get("status") == STATUS_PENDING and parse_iso(row["expires_at"]) <= now: + row["status"] = STATUS_EXPIRED + row["resolved_at"] = _iso(now) + expired.append(dict(row)) + stamp = row.get("resolved_at") or row.get("created_at") or "" + if row.get("status") != STATUS_PENDING and stamp and parse_iso(stamp) < cutoff: + continue + keep.append(row) + rows[:] = keep + for row in expired: + notify(row["project_path"], agent="scheduler", severity="info", + title=f"Scheduled run expired unapproved: {row['schedule_name']}", + message=row["summary"], kind="schedule", ref_id=row["id"]) + return expired + + +class _Txn: + def __init__(self, store: ApprovalStore): + self.store = store + self.flock = _FileLock(store.lock_file) + self.rows: list[dict] = [] + + def __enter__(self) -> list[dict]: + self.store._lock.acquire() + try: + self.flock.__enter__() + except Exception: + self.store._lock.release() + raise + self.rows = self.store._load() + return self.rows + + def __exit__(self, exc_type, exc, tb): + try: + if exc_type is None or isinstance(exc, ApprovalError): + self.store._save(self.rows) + finally: + self.flock.release() + self.store._lock.release() + return False + + +def _summary(schedule: dict) -> str: + target = schedule.get("target") or {} + if schedule.get("kind") == "skill": + what = f"{target.get('skill', '')} {target.get('args') or ''}".strip() + elif schedule.get("kind") == "prompt": + what = str(target.get("prompt") or "")[:120] + else: + what = str(target.get("cmd") or "")[:120] + parts = [what] + if schedule.get("kind") != "command": + parts.append(schedule.get("provider", "claude") + (f"/{schedule['model']}" if schedule.get("model") else "")) + if schedule.get("effort"): + parts.append(f"effort {schedule['effort']}") + parts.append(f"in {schedule.get('cwd') or schedule.get('project_path', '')}") + return " · ".join(p for p in parts if p) diff --git a/services/shell_jobs.py b/services/shell_jobs.py index 54b8e2d..9f8558c 100644 --- a/services/shell_jobs.py +++ b/services/shell_jobs.py @@ -514,6 +514,11 @@ def locate(self, job_id: str) -> Path | None: def load(self, job: JobState) -> JobState | None: return _load_job(job.path) + def find(self, job_id: str) -> JobState | None: + """The current record for ``job_id`` under this root, or None.""" + path = self.locate(job_id) + return _load_job(path) if path is not None else None + def save(self, job: JobState) -> None: _mkdir_private(Path(job.dir)) _write_json_atomic(job.path, job.to_dict()) diff --git a/tests/test_schedule_approvals.py b/tests/test_schedule_approvals.py new file mode 100644 index 0000000..e442977 --- /dev/null +++ b/tests/test_schedule_approvals.py @@ -0,0 +1,86 @@ +import json +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch + +from services import schedule_approvals as sa +from services.notifications import EVENT_KINDS + +REPO = Path(__file__).resolve().parents[1] +SCHED = {"id": "sch-1", "name": "report", "kind": "skill", "target": {"skill": "/daily_report", "args": ""}, + "provider": "claude", "model": "sonnet", "effort": "medium", "project_path": str(REPO), "cwd": str(REPO)} +RUN = {"run_id": "sr-1"} + + +class TestApprovals(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.store = sa.ApprovalStore(Path(self._tmp.name)) + self.notify = patch("services.schedule_approvals.notify") + self.notified = self.notify.start() + + def tearDown(self): + self.notify.stop() + self._tmp.cleanup() + + def test_kind_is_registered(self): + self.assertIn("schedule", EVENT_KINDS) + + def test_create_notifies_and_dedupes(self): + row = self.store.create(SCHED, RUN, ttl_s=600) + self.assertEqual(row["status"], "pending") + self.assertIn("/daily_report", row["summary"]) + self.assertIn("claude/sonnet", row["summary"]) + kw = self.notified.call_args.kwargs + self.assertEqual((kw["kind"], kw["ref_id"], kw["severity"]), ("schedule", row["id"], "warning")) + again = self.store.create(SCHED, {"run_id": "sr-2"}, ttl_s=600) + self.assertEqual(again["id"], row["id"]) + self.assertTrue(again["duplicate"]) + self.assertEqual(len(self.store.list()), 1) + self.assertEqual(self.store.pending_for("sch-1")["id"], row["id"]) + + def test_decide_approve_and_deny(self): + row = self.store.create(SCHED, RUN, ttl_s=600) + done = self.store.decide(row["id"], "approve", decided_by="desk", note="go") + self.assertEqual((done["status"], done["decided_by"], done["note"]), ("approved", "desk", "go")) + with self.assertRaises(sa.ApprovalError): + self.store.decide(row["id"], "deny", decided_by="desk") + other = self.store.create(dict(SCHED, id="sch-2"), {"run_id": "sr-3"}, ttl_s=600) + self.assertEqual(self.store.decide(other["id"], "deny", decided_by="hub")["status"], "denied") + with self.assertRaises(sa.ApprovalError): + self.store.decide(other["id"], "maybe", decided_by="hub") + with self.assertRaises(sa.ApprovalError): + self.store.decide("sa-missing", "approve", decided_by="hub") + self.assertEqual(self.notified.call_args.kwargs["severity"], "info") + + def test_decide_on_lapsed_row_expires_it(self): + row = self.store.create(SCHED, RUN, ttl_s=60) + rows = json.loads(self.store.file.read_text(encoding="utf-8")) + rows[0]["expires_at"] = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat() + self.store.file.write_text(json.dumps(rows), encoding="utf-8") + with self.assertRaises(sa.ApprovalError): + self.store.decide(row["id"], "approve", decided_by="desk") + self.assertEqual(self.store.get(row["id"])["status"], "expired") + + def test_sweep_expires_and_prunes(self): + row = self.store.create(SCHED, RUN, ttl_s=60) + later = datetime.now(timezone.utc) + timedelta(seconds=120) + expired = self.store.sweep_expired(later) + self.assertEqual([r["id"] for r in expired], [row["id"]]) + self.assertEqual(self.store.get(row["id"])["status"], "expired") + self.assertEqual(self.store.list(status="pending"), []) + self.assertEqual(self.store.sweep_expired(later + timedelta(days=sa.RETAIN_DECIDED_DAYS + 1)), []) + self.assertIsNone(self.store.get(row["id"])) + + def test_list_filters(self): + self.store.create(SCHED, RUN, ttl_s=600) + self.store.create(dict(SCHED, id="sch-2", project_path=str(REPO / "tests")), {"run_id": "sr-9"}, ttl_s=600) + self.assertEqual(len(self.store.list(project_path=str(REPO / "tests"))), 1) + self.assertEqual(len(self.store.list(status="pending")), 2) + self.assertEqual(len(self.store.list(limit=1)), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_scheduler_tick.py b/tests/test_scheduler_tick.py new file mode 100644 index 0000000..0518565 --- /dev/null +++ b/tests/test_scheduler_tick.py @@ -0,0 +1,257 @@ +import tempfile +import time +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from oracle.services import scheduler as sch +from services import schedule_approvals as sa +from services import schedule_store as ss + +REPO = Path(__file__).resolve().parents[1] +CFG = {"scheduler_enabled": True, "scheduler_run_retention_days": 30, "scheduler_approval_ttl_s": 600} + + +class FakeJob(SimpleNamespace): + @property + def terminal(self): + return self.status in ("done", "failed", "timeout", "cancelled", "lost") + + +class FakeJobStore: + """Jobs finish only when the test says so; start() returns immediately.""" + + def __init__(self): + self.jobs: dict[str, FakeJob] = {} + self.started: list[dict] = [] + self.output_store = self + self.block_start = False + + def start(self, **kw): + if self.block_start: + time.sleep(5) + jid = f"j-{len(self.jobs) + 1}" + job = FakeJob(id=jid, status="running", exit_code=None, error="", created_at="2026-10-13T09:00:00+00:00", + output_id="", duration_ms=None, project_path=kw["project_path"], session_id=kw["session_id"], + payload=kw["payload"]) + self.jobs[jid] = job + self.started.append(kw) + return job + + def finish(self, jid, status="done", exit_code=0): + job = self.jobs[jid] + job.status, job.exit_code, job.duration_ms = status, exit_code, 42 + + def find(self, job_id): + return self.jobs.get(job_id) + + def reap(self, **kw): + return [] + + +def fields(**over): + base = {"name": "echo", "kind": "command", "target": {"cmd": "echo hi"}, "project_path": str(REPO), + "when": {"type": "every", "seconds": 3600}, "autonomy": "autonomous"} + base.update(over) + return base + + +class SchedulerBase(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name) + self.store = ss.ScheduleStore(self.root) + self.approvals = sa.ApprovalStore(self.root) + self.jobs = FakeJobStore() + self.sched = sch.Scheduler(store=self.store, approvals=self.approvals, jobstore=self.jobs, + config=lambda: dict(CFG), interval=30) + self._patches = [patch("oracle.services.scheduler.notify"), patch("services.schedule_approvals.notify"), + patch("oracle.services.scheduler.ActivityLog")] + self.notify, _, _ = [p.start() for p in self._patches] + + def tearDown(self): + for p in self._patches: + p.stop() + self.sched.stop() + self._tmp.cleanup() + + def due(self, **over): + row = self.store.create(fields(**over)) + with self.store._txn() as doc: + ss._find(doc, row["id"])["next_run_at"] = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat() + return self.store.get(row["id"]) + + def only_run(self): + runs = self.store.runs(limit=0) + self.assertEqual(len(runs), 1) + return runs[0] + + +class TestTick(SchedulerBase): + def test_autonomous_claim_dispatches_without_waiting(self): + self.due() + self.jobs.block_start = False + t0 = time.monotonic() + summary = self.sched.tick() + self.assertLess(time.monotonic() - t0, 2.0) + self.assertEqual((summary["claimed"], summary["dispatched"]), (1, 1)) + run = self.only_run() + self.assertEqual((run["status"], run["job_id"]), ("dispatched", "j-1")) + self.assertEqual(self.jobs.started[0]["payload"]["cmd"], "echo hi") + self.assertEqual(self.sched.tick()["claimed"], 0) + + def test_finished_job_is_folded_and_announced(self): + self.due() + self.sched.tick() + self.jobs.finish("j-1") + summary = self.sched.tick() + self.assertEqual(summary["finished"], 1) + run = self.only_run() + self.assertEqual((run["status"], run["exit_code"], run["duration_ms"]), ("done", 0, 42)) + kw = self.notify.call_args.kwargs + self.assertEqual((kw["kind"], kw["ref_id"], kw["severity"]), ("schedule", run["run_id"], "info")) + + def test_running_job_updates_status_and_missing_job_is_lost(self): + self.due() + self.sched.tick() + self.sched.tick() + self.assertEqual(self.only_run()["status"], "running") + self.jobs.jobs.clear() + self.sched.tick() + self.assertEqual(self.only_run()["status"], "lost") + + def test_overlap_skip(self): + row = self.due() + self.sched.tick() + self.store.request_run(row["id"]) + self.sched.tick() + statuses = sorted(r["status"] for r in self.store.runs(limit=0)) + self.assertEqual(statuses, ["running", "skipped_overlap"]) + + def test_overlap_allow_dispatches_both(self): + row = self.due(overlap="allow") + self.sched.tick() + self.store.request_run(row["id"]) + self.sched.tick() + self.assertEqual(sorted(r["status"] for r in self.store.runs(limit=0)), ["dispatched", "running"]) + + def test_misfire_grace_and_skip(self): + late = self.due(misfire={"policy": "grace", "grace_s": 60}) + with self.store._txn() as doc: + ss._find(doc, late["id"])["next_run_at"] = (datetime.now(timezone.utc) - timedelta(minutes=10)).isoformat() + self.sched.tick() + self.assertEqual(self.only_run()["status"], "missed") + self.assertEqual(self.store.get(late["id"])["run_count"], 0) + within = self.due(name="fresh", misfire={"policy": "skip"}) + self.sched.tick() + fresh = [r for r in self.store.runs(limit=0) if r["schedule_id"] == within["id"]][0] + self.assertEqual(fresh["status"], "dispatched") + + def test_check_command_gates_the_run(self): + self.due(check_command="test -s inbox") + self.sched.tick() + run = self.only_run() + self.assertEqual(run["status"], "check_running") + self.assertEqual(self.jobs.started[0]["payload"]["cmd"], "test -s inbox") + self.jobs.finish("j-1", exit_code=0) + self.sched.tick() + self.assertEqual(self.only_run()["status"], "check_skipped") + self.due(name="armed", check_command="test -s inbox") + self.sched.tick() + armed = [r for r in self.store.runs(limit=0) if r["schedule_name"] == "armed"][0] + self.jobs.finish(armed["job_id"], status="failed", exit_code=1) + self.sched.tick() + armed = self.store.get_run(armed["run_id"]) + self.assertEqual((armed["status"], armed["job_id"]), ("dispatched", "j-3")) + self.assertEqual(self.jobs.started[-1]["payload"]["cmd"], "echo hi") + + def test_check_command_failure(self): + self.due(check_command="boom") + self.sched.tick() + self.jobs.finish("j-1", status="failed", exit_code=2) + self.sched.tick() + run = self.only_run() + self.assertEqual(run["status"], "failed") + self.assertIn("check_command", run["error"]) + + def test_disabled_config_does_nothing(self): + self.due() + self.sched.config = lambda: {"scheduler_enabled": False} + self.assertTrue(self.sched.tick().get("disabled")) + self.assertEqual(self.store.runs(limit=0), []) + + +class TestApprovalFlow(SchedulerBase): + def test_approval_then_approve_dispatches(self): + row = self.due(autonomy="approval", approval_ttl_s=600) + summary = self.sched.tick() + self.assertEqual(summary["awaiting_approval"], 1) + run = self.only_run() + self.assertEqual(run["status"], "awaiting_approval") + pending = self.approvals.pending_for(row["id"]) + self.assertEqual(pending["run_id"], run["run_id"]) + self.sched.tick() + self.assertEqual(self.only_run()["status"], "awaiting_approval") + self.approvals.decide(pending["id"], "approve", decided_by="desk") + self.sched.tick() + run = self.only_run() + self.assertEqual((run["status"], run["job_id"]), ("dispatched", "j-1")) + + def test_deny_and_expire(self): + row = self.due(autonomy="approval") + self.sched.tick() + pending = self.approvals.pending_for(row["id"]) + self.approvals.decide(pending["id"], "deny", decided_by="hub") + self.sched.tick() + self.assertEqual(self.only_run()["status"], "denied") + self.assertEqual(self.jobs.started, []) + again = self.store.request_run(row["id"]) + self.sched.tick() + self.assertEqual(self.store.get_run(again["run_id"])["status"], "awaiting_approval") + later = datetime.now(timezone.utc) + timedelta(days=1) + summary = self.sched.tick(now=later) + self.assertEqual(summary["expired"], 1) + self.assertEqual(self.store.get_run(again["run_id"])["status"], "expired") + + def test_second_pending_for_same_schedule_is_not_queued(self): + row = self.due(autonomy="approval", overlap="allow") + self.sched.tick() + self.store.request_run(row["id"]) + self.sched.tick() + self.assertEqual(sorted(r["status"] for r in self.store.runs(limit=0)), + ["awaiting_approval", "skipped_overlap"]) + + def test_run_now_bypass(self): + row = self.store.create(fields(autonomy="approval")) + self.sched.run_now(row["id"], bypass_approval=True) + self.sched.tick() + self.assertEqual(self.only_run()["status"], "dispatched") + self.assertEqual(self.approvals.list(), []) + + +class TestSingleInstance(SchedulerBase): + def test_second_scheduler_stays_in_standby(self): + self.due() + other = sch.Scheduler(store=ss.ScheduleStore(self.root), approvals=sa.ApprovalStore(self.root), + jobstore=FakeJobStore(), config=lambda: dict(CFG)) + self.assertFalse(self.sched.tick()["standby"]) + summary = other.tick() + self.assertTrue(summary["standby"]) + self.assertEqual(len(self.store.runs(limit=0)), 1) + self.assertTrue(self.sched.status["lock_held"]) + self.sched.stop() + self.assertFalse(other.tick()["standby"]) + other.stop() + + def test_status_shape(self): + self.due() + status = self.sched.status + self.assertEqual(status["enabled_schedules"], 1) + self.assertIsNotNone(status["next_due"]) + self.assertFalse(status["running"]) + + +if __name__ == "__main__": + unittest.main() From f6bc5b68e27ed827fa41f30a6d3a59c8e50c8d15 Mon Sep 17 00:00:00 2001 From: drknowhow Date: Tue, 13 Oct 2026 09:54:12 -0400 Subject: [PATCH 2/4] feat(schedule): Oracle starts the scheduler and serves /api/schedules The Oracle builds the schedule, approval and job stores in _init_services, starts the Scheduler beside the review agent when scheduler_enabled, and registers oracle/services/schedule_api.py: CRUD with expected_rev, run now, enable/disable, runs, approvals decide (409 once lapsed), preview, status, kick. GETs stay open on loopback like the rest of the local API; mutations sit behind _local_write_guard. The blueprint's plain functions are what the mobile gateway and the Hub will call next. docs/scheduled-tasks.md describes the model and loop. Verified: 11 Flask tests against the real app. Claude-Session: https://claude.ai/code/session_01JUcvmQq81aLo9ewD9Xc4qY --- docs/scheduled-tasks.md | 90 ++++++++++++ oracle/oracle_server.py | 24 +++ oracle/services/schedule_api.py | 252 ++++++++++++++++++++++++++++++++ tests/test_schedule_api.py | 165 +++++++++++++++++++++ 4 files changed, 531 insertions(+) create mode 100644 docs/scheduled-tasks.md create mode 100644 oracle/services/schedule_api.py create mode 100644 tests/test_schedule_api.py diff --git a/docs/scheduled-tasks.md b/docs/scheduled-tasks.md new file mode 100644 index 0000000..676e36f --- /dev/null +++ b/docs/scheduled-tasks.md @@ -0,0 +1,90 @@ +# Scheduled tasks + +A scheduled task runs a Claude Code skill, a free prompt, or a plain command +at a time or cadence, with the model, effort and permission mode you choose, +either autonomously or only after a human approves each run. The Oracle hosts +the loop; the Hub, the Desk app and the phone show and decide; `c3_schedule` +lets an agent propose one. + +## Model + +A **schedule** (`~/.c3/oracle/schedules.json`): + +| field | meaning | +|---|---| +| `kind` | `skill` (`target.skill` like `/daily_report`, optional `target.args`), `prompt` (`target.prompt`) or `command` (`target.cmd`, no LLM) | +| `provider` | `claude` (default), `codex`, `gemini`, `grok`; ignored for `command` | +| `model`, `effort` | passed through (`--model`, `--effort low..max`); empty = the CLI's default | +| `permission_mode` | Claude's `--permission-mode`; `bypassPermissions` only when `scheduler_allow_bypass_permissions` is on | +| `allowed_tools`, `max_budget_usd` | `--allowedTools`, `--max-budget-usd` | +| `autonomy` | `autonomous` fires; `approval` creates a pending approval first and expires it after `approval_ttl_s` (expire-and-skip) | +| `when` | `{type: cron, expr, tz}`, `{type: every, seconds}`, `{type: at, when}` or `{type: preset, preset: hourly\|daily\|weekdays\|weekly\|monthly, at: "09:00", days}` | +| `misfire` | `{policy: grace, grace_s}` runs a late slot inside the window, else logs `missed`; `skip` runs only slots caught within a minute | +| `overlap` | `skip` (default) logs `skipped_overlap` while a previous run is open; `allow` runs anyway | +| `check_command` | runs first; exit 0 = nothing to do (`check_skipped`, no LLM cost), exit 1 arms the run, anything else fails it | +| `project_path`, `cwd`, `timeout_s`, `enabled` | where and how long; `project_path` must be a registered project | + +`tz: local` is the machine's current UTC offset (or `C3_SCHEDULER_TZ`); name an +IANA zone for DST-exact wall clocks. + +A **run** (`~/.c3/oracle/schedule_runs/.json`) moves through +`claimed → awaiting_approval | check_running | dispatched → running → done | +failed | timeout | cancelled | lost`, or ends early as `missed`, +`skipped_overlap`, `check_skipped`, `denied` or `expired`. It records the +shell job id, exit code, duration, output id, and for LLM providers the cost, +tokens, turns, model and a result excerpt. + +## Loop + +`oracle/services/scheduler.py` ticks every `scheduler_tick_seconds` (30): + +1. lapse pending approvals past their TTL; +2. fold finished shell jobs into their runs (`JobStore.find`); a run whose job vanished is `lost`; +3. `ScheduleStore.claim_due` — one locked transaction advances `next_run_at` and writes the run row, so a claimed slot is never claimed again by any process; one-shots are disabled here; +4. each claimed run passes the misfire window, the overlap rule, the check command and the autonomy gate; autonomous runs are dispatched as detached `services/shell_jobs.py` jobs (tree-kill on timeout, no console window, output kept, feed notification); +5. approved runs are dispatched, denied or lapsed ones closed; +6. old run files are swept after `scheduler_run_retention_days`. + +The tick never waits on a job. `start()` takes `~/.c3/oracle/scheduler.lock`; +a second Oracle stays in `standby` and retries each tick, so two processes on +one machine cannot both fire. + +Every terminal run writes a `kind=schedule` notification into the project's +`.c3/notifications.jsonl` (which wakes the Desk and phone feeds) and an +`activity_log` row. + +## Headless Claude + +A skill or prompt run is `claude -p --output-format json --no-session-persistence +--permission-mode --permission-prompts none --name sched: [--model] +[--effort] [--max-budget-usd] [--allowedTools]` with the prompt on stdin, run in +`cwd` with `CLAUDECODE` and the other nesting markers removed so Claude does not +treat it as a child session. Skills, CLAUDE.md, hooks and MCP servers load as +in an interactive session. Cost is read from the JSON result. + +## REST + +Oracle (`/api/schedules`, GETs open on loopback, mutations behind the dashboard +cookie or Discovery Bearer): + +``` +GET /api/schedules?project= POST /api/schedules +GET /api/schedules/ PATCH /api/schedules/ (expected_rev → 409) +DELETE /api/schedules/ POST /api/schedules//run {bypass_approval} +POST /api/schedules//enable|disable GET /api/schedules//runs +GET /api/schedules/runs?project&status&schedule_id&limit +GET /api/schedules/runs/ +GET /api/schedules/approvals?status&project +POST /api/schedules/approvals/ {decision: approve|deny, note} (lapsed → 409) +POST /api/schedules/preview {when, count} → next fire times +GET /api/schedules/status POST /api/schedules/kick +``` + +## Config (`~/.c3/oracle/config.json`) + +`scheduler_enabled`, `scheduler_tick_seconds`, `scheduler_default_timeout_s`, +`scheduler_max_timeout_s`, `scheduler_approval_ttl_s`, +`scheduler_misfire_grace_s`, `scheduler_run_retention_days`, +`scheduler_default_permission_mode`, `scheduler_allow_bypass_permissions`, +`scheduler_default_max_budget_usd`, `mobile_schedules_enabled`, +`mobile_schedules_write`. diff --git a/oracle/oracle_server.py b/oracle/oracle_server.py index a9e0ba3..e651b54 100644 --- a/oracle/oracle_server.py +++ b/oracle/oracle_server.py @@ -36,6 +36,7 @@ from oracle.services.memory_writer import MemoryWriter # noqa: E402 from oracle.services.project_scanner import ProjectScanner # noqa: E402 from oracle.services.review_agent import ReviewAgent # noqa: E402 +from oracle.services.scheduler import Scheduler # noqa: E402 from oracle.services.tool_executor import ToolExecutor # noqa: E402 from oracle.services.tool_registry import ToolRegistry, _c3_version # noqa: E402 from services.ollama_bridge import OllamaBridge # noqa: E402 @@ -60,10 +61,14 @@ _federated: FederatedGraph | None = None _tool_registry: ToolRegistry | None = None _activity_reporter: ActivityReporter | None = None +_scheduler: Scheduler | None = None +_schedule_store = None +_approval_store = None def _init_services(): global _cfg, _bridge, _scanner, _reader, _checker, _writer, _cross_memory, _engine, _agent, _model_verified, _chat_store, _chat_engine, _c3_bridge, _federated, _tool_registry, _activity_reporter + global _scheduler, _schedule_store, _approval_store _cfg = load_config() _bridge = OllamaBridge( base_url=_cfg.get("ollama_base_url", "https://ollama.com"), @@ -125,6 +130,13 @@ def _verify(): ) mobile_api.init_services(scanner=_scanner, checker=_checker, reporter=_activity_reporter) + from services.schedule_approvals import ApprovalStore + from services.schedule_store import ScheduleStore + from services.shell_jobs import JobStore + _schedule_store = ScheduleStore() + _approval_store = ApprovalStore() + _scheduler = Scheduler(store=_schedule_store, approvals=_approval_store, jobstore=JobStore(), + interval=int(_cfg.get("scheduler_tick_seconds", 30))) atexit.register(lambda: _c3_bridge.shutdown() if _c3_bridge else None) @@ -268,6 +280,15 @@ def _local_write_guard(): get_cfg=lambda: _cfg) app.register_blueprint(chat_poll.bp) +# Scheduled tasks (/api/schedules/*): GETs open like the rest of the local +# API, mutations behind _local_write_guard. See docs/scheduled-tasks.md. +from oracle.services import schedule_api # noqa: E402 + +schedule_api.configure(get_scheduler=lambda: _scheduler, get_store=lambda: _schedule_store, + get_approvals=lambda: _approval_store, get_scanner=lambda: _scanner, + get_cfg=lambda: _cfg) +app.register_blueprint(schedule_api.bp) + # ── Static ──────────────────────────────────────────────── @@ -1252,6 +1273,9 @@ def run_oracle(port: int = None, open_browser: bool = None): if cfg.get("review_enabled", True) and _agent: _agent.start() atexit.register(_agent.stop) + if cfg.get("scheduler_enabled", True) and _scheduler: + _scheduler.start() + atexit.register(_scheduler.stop) # Start the discovery MCP server on its own loopback port if enabled. if cfg.get("mcp_enabled", True) and _tool_registry is not None: diff --git a/oracle/services/schedule_api.py b/oracle/services/schedule_api.py new file mode 100644 index 0000000..1c75cff --- /dev/null +++ b/oracle/services/schedule_api.py @@ -0,0 +1,252 @@ +"""``/api/schedules/*`` — the Oracle's local surface over the scheduler. + +GETs are open like the rest of the local Oracle API; mutations go through +``_local_write_guard`` (dashboard cookie or Discovery Bearer). The plain +functions below do the work on dicts so the mobile gateway and the Hub can +serve the same operations with their own auth and ``decided_by``. +""" +from __future__ import annotations + +import os +from datetime import datetime, timezone +from pathlib import Path + +from flask import Blueprint, jsonify, request + +from services.cron_expr import preview, validate_when +from services.schedule_approvals import ApprovalError +from services.schedule_store import RUN_STATUSES, RevConflict, ScheduleError + +bp = Blueprint("schedule_api", __name__, url_prefix="/api/schedules") + +_get_scheduler = lambda: None # noqa: E731 +_get_store = lambda: None # noqa: E731 +_get_approvals = lambda: None # noqa: E731 +_get_scanner = lambda: None # noqa: E731 +_get_cfg = lambda: {} # noqa: E731 + + +def configure(*, get_scheduler, get_store, get_approvals, get_scanner, get_cfg) -> None: + global _get_scheduler, _get_store, _get_approvals, _get_scanner, _get_cfg + _get_scheduler, _get_store, _get_approvals = get_scheduler, get_store, get_approvals + _get_scanner, _get_cfg = get_scanner, get_cfg + + +class ApiError(Exception): + def __init__(self, message: str, status: int = 400): + super().__init__(message) + self.status = status + + +def registered_project(raw: str, scanner) -> str: + """The registered project path matching ``raw``; ``ApiError`` 404 otherwise.""" + raw = (raw or "").strip() + if not raw: + raise ApiError("project_path is required", 400) + if scanner is None: + return raw + try: + target = os.path.normcase(str(Path(raw).resolve())) + except (OSError, ValueError) as exc: + raise ApiError("unknown project", 404) from exc + for proj in scanner.discover(): + if not proj.get("has_c3"): + continue + try: + if os.path.normcase(str(Path(proj["path"]).resolve())) == target: + return str(Path(proj["path"]).resolve()) + except (OSError, ValueError): + continue + raise ApiError("unknown project", 404) + + +def defaults() -> dict: + from oracle.services.scheduler import defaults_from_config + return defaults_from_config(_get_cfg() or {}) + + +def create_schedule(body: dict, *, scanner, by: str) -> dict: + fields = dict(body or {}) + fields.pop("expected_rev", None) + fields["project_path"] = registered_project(str(fields.get("project_path") or fields.pop("project", "")), scanner) + try: + return _get_store().create(fields, defaults=defaults(), created_by=by) + except ScheduleError as exc: + raise ApiError(str(exc), 400) from exc + + +def update_schedule(schedule_id: str, body: dict, *, scanner, by: str) -> dict: + patch = dict(body or {}) + expected_rev = patch.pop("expected_rev", None) + if "project_path" in patch or "project" in patch: + patch["project_path"] = registered_project(str(patch.get("project_path") or patch.pop("project", "")), scanner) + store = _get_store() + if store.get(schedule_id) is None: + raise ApiError(f"no schedule {schedule_id}", 404) + try: + return store.update(schedule_id, patch, expected_rev=expected_rev, defaults=defaults(), by=by) + except RevConflict as exc: + raise ApiError(str(exc), 409) from exc + except ScheduleError as exc: + raise ApiError(str(exc), 400) from exc + + +def run_schedule(schedule_id: str, body: dict, *, by: str) -> dict: + if _get_store().get(schedule_id) is None: + raise ApiError(f"no schedule {schedule_id}", 404) + bypass = bool((body or {}).get("bypass_approval", False)) + scheduler = _get_scheduler() + if scheduler is None: + raise ApiError("scheduler not running", 503) + return scheduler.run_now(schedule_id, bypass_approval=bypass, by=by) + + +def decide_approval(approval_id: str, body: dict, *, by: str) -> dict: + decision = str((body or {}).get("decision") or "").strip().lower() + note = str((body or {}).get("note") or "") + approvals = _get_approvals() + if approvals.get(approval_id) is None: + raise ApiError(f"no approval {approval_id}", 404) + try: + row = approvals.decide(approval_id, decision, decided_by=by, note=note) + except ApprovalError as exc: + status = 400 if "decision must be" in str(exc) else 409 + raise ApiError(str(exc), status) from exc + scheduler = _get_scheduler() + if scheduler is not None: + scheduler.kick() + return row + + +def preview_when(body: dict) -> dict: + try: + when = validate_when((body or {}).get("when")) + except ValueError as exc: + raise ApiError(str(exc), 400) from exc + count = max(1, min(int((body or {}).get("count") or 5), 20)) + fires = preview(when, datetime.now(timezone.utc), count) + return {"when": when, "next": [f.isoformat() for f in fires]} + + +def _limit(default: int = 50, cap: int = 500) -> int: + try: + return max(1, min(int(request.args.get("limit") or default), cap)) + except ValueError: + return default + + +def _status_arg(): + status = (request.args.get("status") or "").strip() + if status and status not in RUN_STATUSES: + raise ApiError(f"unknown status {status}", 400) + return status or None + + +@bp.errorhandler(ApiError) +def _api_error(exc: ApiError): + return jsonify({"error": str(exc)}), exc.status + + +@bp.route("", methods=["GET"]) +def list_schedules(): + project = (request.args.get("project") or "").strip() or None + rows = _get_store().list(project_path=project) + return jsonify({"schedules": rows, "count": len(rows), "rev": _get_store().rev()}) + + +@bp.route("", methods=["POST"]) +def create(): + row = create_schedule(request.get_json(silent=True) or {}, scanner=_get_scanner(), by="oracle") + return jsonify(row), 201 + + +@bp.route("/status", methods=["GET"]) +def status(): + scheduler = _get_scheduler() + return jsonify(scheduler.status if scheduler else {"running": False}) + + +@bp.route("/kick", methods=["POST"]) +def kick(): + scheduler = _get_scheduler() + if scheduler is None: + raise ApiError("scheduler not running", 503) + scheduler.kick() + return jsonify({"kicked": True}) + + +@bp.route("/preview", methods=["POST"]) +def preview_route(): + return jsonify(preview_when(request.get_json(silent=True) or {})) + + +@bp.route("/runs", methods=["GET"]) +def runs(): + rows = _get_store().runs(schedule_id=(request.args.get("schedule_id") or None), status=_status_arg(), + project_path=(request.args.get("project") or None), limit=_limit()) + return jsonify({"runs": rows, "count": len(rows)}) + + +@bp.route("/runs/", methods=["GET"]) +def one_run(run_id: str): + run = _get_store().get_run(run_id) + if run is None: + raise ApiError(f"no run {run_id}", 404) + return jsonify(run) + + +@bp.route("/approvals", methods=["GET"]) +def approvals(): + status = (request.args.get("status") or "").strip() or None + rows = _get_approvals().list(status=status, project_path=(request.args.get("project") or None), + limit=_limit(100)) + return jsonify({"approvals": rows, "count": len(rows)}) + + +@bp.route("/approvals/", methods=["POST"]) +def decide(approval_id: str): + return jsonify(decide_approval(approval_id, request.get_json(silent=True) or {}, by="oracle")) + + +@bp.route("/", methods=["GET"]) +def one(schedule_id: str): + row = _get_store().get(schedule_id) + if row is None: + raise ApiError(f"no schedule {schedule_id}", 404) + return jsonify(dict(row, rev=_get_store().rev())) + + +@bp.route("/", methods=["PATCH"]) +def update(schedule_id: str): + return jsonify(update_schedule(schedule_id, request.get_json(silent=True) or {}, scanner=_get_scanner(), + by="oracle")) + + +@bp.route("/", methods=["DELETE"]) +def delete(schedule_id: str): + if not _get_store().delete(schedule_id, by="oracle"): + raise ApiError(f"no schedule {schedule_id}", 404) + return jsonify({"deleted": schedule_id}) + + +@bp.route("//run", methods=["POST"]) +def run_now(schedule_id: str): + return jsonify(run_schedule(schedule_id, request.get_json(silent=True) or {}, by="oracle")), 202 + + +@bp.route("//enable", methods=["POST"]) +def enable(schedule_id: str): + return jsonify(update_schedule(schedule_id, {"enabled": True}, scanner=_get_scanner(), by="oracle")) + + +@bp.route("//disable", methods=["POST"]) +def disable(schedule_id: str): + return jsonify(update_schedule(schedule_id, {"enabled": False}, scanner=_get_scanner(), by="oracle")) + + +@bp.route("//runs", methods=["GET"]) +def schedule_runs(schedule_id: str): + if _get_store().get(schedule_id) is None: + raise ApiError(f"no schedule {schedule_id}", 404) + rows = _get_store().runs(schedule_id=schedule_id, status=_status_arg(), limit=_limit()) + return jsonify({"runs": rows, "count": len(rows)}) diff --git a/tests/test_schedule_api.py b/tests/test_schedule_api.py new file mode 100644 index 0000000..f143da2 --- /dev/null +++ b/tests/test_schedule_api.py @@ -0,0 +1,165 @@ +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)) + +os.environ["C3_ORACLE_API_KEY"] = "schedule-api-key" + +import oracle.oracle_server as srv # noqa: E402 +from oracle.services import scheduler as sch # noqa: E402 +from services import schedule_approvals as sa # noqa: E402 +from services import schedule_store as ss # noqa: E402 + +AUTH = {"Authorization": "Bearer schedule-api-key"} + + +class _StubScanner: + def __init__(self, paths): + self.paths = paths + + def discover(self, force=False): + return [{"path": str(p), "name": p.name, "has_c3": True} for p in self.paths] + + +class _NoJobs: + def start(self, **kw): + raise AssertionError("tick must not run inside a request") + + def find(self, job_id): + return None + + def reap(self, **kw): + return [] + + +class TestScheduleApi(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.tmp = Path(tempfile.mkdtemp(prefix="c3-sched-api-")) + cls.alpha = cls.tmp / "alpha" + (cls.alpha / ".c3").mkdir(parents=True) + cls.outsider = cls.tmp / "outsider" + (cls.outsider / ".c3").mkdir(parents=True) + cls._env = mock.patch.dict(os.environ, {"C3_ORACLE_API_KEY": "schedule-api-key"}) + cls._env.start() + cls._prior = (srv._cfg, srv._scanner, srv._schedule_store, srv._approval_store, srv._scheduler) + srv._cfg = {"scheduler_approval_ttl_s": 600} + srv._scanner = _StubScanner([cls.alpha]) + srv._schedule_store = ss.ScheduleStore(cls.tmp / "oracle") + srv._approval_store = sa.ApprovalStore(cls.tmp / "oracle") + srv._scheduler = sch.Scheduler(store=srv._schedule_store, approvals=srv._approval_store, jobstore=_NoJobs(), + config=lambda: srv._cfg) + cls.client = srv.app.test_client() + + @classmethod + def tearDownClass(cls): + srv._cfg, srv._scanner, srv._schedule_store, srv._approval_store, srv._scheduler = cls._prior + 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 create(self, **over): + resp = self.client.post("/api/schedules", json=self.body(**over), headers=AUTH) + self.assertEqual(resp.status_code, 201, resp.get_json()) + return resp.get_json() + + def test_mutations_need_auth_and_reads_do_not(self): + self.assertEqual(self.client.post("/api/schedules", json=self.body()).status_code, 401) + self.assertEqual(self.client.get("/api/schedules").status_code, 200) + self.assertEqual(self.client.get("/api/schedules/status").status_code, 200) + + def test_crud_roundtrip(self): + row = self.create(name="crud") + self.assertEqual(row["project_path"], str(self.alpha.resolve())) + listed = self.client.get(f"/api/schedules?project={self.alpha}").get_json() + self.assertIn(row["id"], [r["id"] for r in listed["schedules"]]) + one = self.client.get(f"/api/schedules/{row['id']}").get_json() + self.assertEqual(one["name"], "crud") + rev = one["rev"] + patched = self.client.patch(f"/api/schedules/{row['id']}", json={"name": "crud2", "expected_rev": rev}, + headers=AUTH) + self.assertEqual(patched.status_code, 200) + stale = self.client.patch(f"/api/schedules/{row['id']}", json={"name": "x", "expected_rev": rev}, headers=AUTH) + self.assertEqual(stale.status_code, 409) + bad = self.client.patch(f"/api/schedules/{row['id']}", json={"effort": "ultra"}, headers=AUTH) + self.assertEqual(bad.status_code, 400) + off = self.client.post(f"/api/schedules/{row['id']}/disable", headers=AUTH).get_json() + self.assertFalse(off["enabled"]) + self.assertIsNone(off["next_run_at"]) + on = self.client.post(f"/api/schedules/{row['id']}/enable", headers=AUTH).get_json() + self.assertIsNotNone(on["next_run_at"]) + self.assertEqual(self.client.delete(f"/api/schedules/{row['id']}", headers=AUTH).status_code, 200) + self.assertEqual(self.client.get(f"/api/schedules/{row['id']}").status_code, 404) + self.assertEqual(self.client.delete(f"/api/schedules/{row['id']}", headers=AUTH).status_code, 404) + + def test_unregistered_project_is_404_and_validation_400(self): + resp = self.client.post("/api/schedules", json=self.body(project_path=str(self.outsider)), headers=AUTH) + self.assertEqual(resp.status_code, 404) + resp = self.client.post("/api/schedules", json=self.body(when={"type": "cron", "expr": "nope"}), headers=AUTH) + self.assertEqual(resp.status_code, 400) + self.assertIn("cron", resp.get_json()["error"]) + + def test_run_now_and_runs(self): + row = self.create(name="runme") + resp = self.client.post(f"/api/schedules/{row['id']}/run", json={}, headers=AUTH) + self.assertEqual(resp.status_code, 202) + run = resp.get_json() + self.assertEqual((run["status"], run["trigger"]), ("claimed", "manual")) + runs = self.client.get(f"/api/schedules/{row['id']}/runs").get_json() + self.assertEqual([r["run_id"] for r in runs["runs"]], [run["run_id"]]) + self.assertEqual(self.client.get(f"/api/schedules/runs/{run['run_id']}").get_json()["schedule_id"], row["id"]) + self.assertEqual(self.client.get("/api/schedules/runs?status=bogus").status_code, 400) + self.assertEqual(self.client.post("/api/schedules/sch-none/run", json={}, headers=AUTH).status_code, 404) + bypass = self.client.post(f"/api/schedules/{row['id']}/run", json={"bypass_approval": True}, headers=AUTH) + self.assertEqual(bypass.get_json()["status"], "approved") + + def test_approvals_decide(self): + row = self.create(name="gated", autonomy="approval") + run = srv._schedule_store.request_run(row["id"]) + with mock.patch("services.schedule_approvals.notify"): + pending = srv._approval_store.create(row, run, ttl_s=600) + listed = self.client.get("/api/schedules/approvals?status=pending").get_json() + self.assertIn(pending["id"], [a["id"] for a in listed["approvals"]]) + self.assertEqual(self.client.post(f"/api/schedules/approvals/{pending['id']}", json={"decision": "later"}, + headers=AUTH).status_code, 400) + ok = self.client.post(f"/api/schedules/approvals/{pending['id']}", json={"decision": "approve"}, + headers=AUTH) + self.assertEqual(ok.status_code, 200) + self.assertEqual((ok.get_json()["status"], ok.get_json()["decided_by"]), ("approved", "oracle")) + again = self.client.post(f"/api/schedules/approvals/{pending['id']}", json={"decision": "deny"}, + headers=AUTH) + self.assertEqual(again.status_code, 409) + self.assertEqual(self.client.post("/api/schedules/approvals/sa-none", json={"decision": "approve"}, + headers=AUTH).status_code, 404) + + def test_preview(self): + resp = self.client.post("/api/schedules/preview", + json={"when": {"type": "cron", "expr": "0 9 * * *", "tz": "UTC"}, "count": 3}, + headers=AUTH) + self.assertEqual(resp.status_code, 200) + data = resp.get_json() + self.assertEqual(len(data["next"]), 3) + self.assertTrue(all(t.endswith("09:00:00+00:00") for t in data["next"])) + self.assertEqual(self.client.post("/api/schedules/preview", json={"when": {"type": "x"}}, + headers=AUTH).status_code, 400) + + def test_status_and_kick(self): + status = self.client.get("/api/schedules/status").get_json() + self.assertIn("pending_approvals", status) + self.assertEqual(self.client.post("/api/schedules/kick", headers=AUTH).get_json(), {"kicked": True}) + self.assertEqual(json.loads(json.dumps(status))["running"], False) + + +if __name__ == "__main__": + unittest.main() From d34a0cbc85c09cf3215063ff5914ef9c544ee4b4 Mon Sep 17 00:00:00 2001 From: drknowhow Date: Tue, 13 Oct 2026 09:59:49 -0400 Subject: [PATCH 3/4] test: schedule is a documented notification kind Claude-Session: https://claude.ai/code/session_01JUcvmQq81aLo9ewD9Xc4qY --- tests/test_notification_kinds.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_notification_kinds.py b/tests/test_notification_kinds.py index 75a1ba5..2b82623 100644 --- a/tests/test_notification_kinds.py +++ b/tests/test_notification_kinds.py @@ -101,7 +101,7 @@ def test_notify_never_raises_on_an_unwritable_project(self): def test_event_kinds_are_the_documented_set(self): self.assertEqual(set(ntf.EVENT_KINDS), - {"shell_job", "ci", "session", "mcp", "override"}) + {"shell_job", "ci", "session", "mcp", "override", "schedule"}) class TestShellJobProducer(_Base): From fbb0499b1e343c1104cb9c83c2c6c00c2d6fa0e1 Mon Sep 17 00:00:00 2001 From: drknowhow Date: Tue, 13 Oct 2026 10:18:58 -0400 Subject: [PATCH 4/4] test: keep the Discovery key override inside the test class Set at import, the key outlived the module and made every later test_oracle_discovery_api call answer 401. Claude-Session: https://claude.ai/code/session_01JUcvmQq81aLo9ewD9Xc4qY --- tests/test_schedule_api.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/test_schedule_api.py b/tests/test_schedule_api.py index f143da2..2443e56 100644 --- a/tests/test_schedule_api.py +++ b/tests/test_schedule_api.py @@ -9,8 +9,6 @@ REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT)) -os.environ["C3_ORACLE_API_KEY"] = "schedule-api-key" - import oracle.oracle_server as srv # noqa: E402 from oracle.services import scheduler as sch # noqa: E402 from services import schedule_approvals as sa # noqa: E402