diff --git a/oracle/config.py b/oracle/config.py index 706c0c8..88fb664 100644 --- a/oracle/config.py +++ b/oracle/config.py @@ -68,6 +68,21 @@ "digest_narrate": False, # LLM prose costs a cloud call; opt-in "digest_notify_file": "", # "" = disabled; else JSONL sink path "digest_retention_days": 14, # prune stored digests + # ── Scheduled tasks (docs/scheduled-tasks.md) ── + "scheduler_enabled": True, + "scheduler_tick_seconds": 30, + "scheduler_default_timeout_s": 1800, + "scheduler_max_timeout_s": 21600, + "scheduler_approval_ttl_s": 3600, + "scheduler_misfire_grace_s": 900, + "scheduler_run_retention_days": 30, + "scheduler_default_permission_mode": "acceptEdits", + # bypassPermissions lets a headless run act without any permission check; + # a schedule may only ask for it once this is switched on. + "scheduler_allow_bypass_permissions": False, + "scheduler_default_max_budget_usd": 0, + "mobile_schedules_enabled": True, # read: schedules / runs / approvals + "mobile_schedules_write": True, # create / edit / run now / decide "auto_open_browser": True, "theme": "dark", "ui_last_tab": "chat", # persisted UI preference (hub parity) diff --git a/pyproject.toml b/pyproject.toml index 0cdb390..4b4890e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,6 +51,7 @@ dependencies = [ "pyyaml>=6.0.0", "keyring>=24.0", "cryptography>=41.0", + "tzdata>=2024.1; sys_platform == 'win32'", ] [project.optional-dependencies] diff --git a/services/cron_expr.py b/services/cron_expr.py new file mode 100644 index 0000000..4b03174 --- /dev/null +++ b/services/cron_expr.py @@ -0,0 +1,292 @@ +"""5-field cron expressions and the ``when`` shapes a schedule row carries. + +A ``when`` is one of:: + + {"type": "cron", "expr": "0 9 * * 1-5", "tz": "local"} + {"type": "every", "seconds": 3600} + {"type": "at", "when": "2026-10-14T09:00:00+00:00"} + {"type": "preset", "preset": "weekdays", "at": "09:00", "days": [], "tz": "local"} + +``next_fire`` turns any of them into the first aware UTC datetime strictly +after a reference instant, or ``None`` when nothing is left to fire. +""" +from __future__ import annotations + +import os +import re +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone, tzinfo +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +WHEN_TYPES = ("cron", "every", "at", "preset") +PRESETS = ("hourly", "daily", "weekdays", "weekly", "monthly") + +MIN_EVERY_SECONDS = 60 +SEARCH_LIMIT_DAYS = 366 + +_ALIASES = { + "@hourly": "0 * * * *", + "@daily": "0 0 * * *", + "@midnight": "0 0 * * *", + "@weekly": "0 0 * * 0", + "@monthly": "0 0 1 * *", + "@yearly": "0 0 1 1 *", + "@annually": "0 0 1 1 *", +} +_MONTHS = {m: i + 1 for i, m in enumerate( + ("jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec"))} +_DOWS = {d: i for i, d in enumerate(("sun", "mon", "tue", "wed", "thu", "fri", "sat"))} +_HHMM = re.compile(r"^([01]?\d|2[0-3]):([0-5]\d)$") + + +@dataclass(frozen=True) +class CronSpec: + minutes: frozenset + hours: frozenset + days: frozenset + months: frozenset + weekdays: frozenset + day_restricted: bool + weekday_restricted: bool + + def matches_day(self, dt: datetime) -> bool: + if dt.month not in self.months: + return False + dom = dt.day in self.days + dow = _cron_weekday(dt) in self.weekdays + if self.day_restricted and self.weekday_restricted: + return dom or dow + if self.day_restricted: + return dom + if self.weekday_restricted: + return dow + return True + + +def _cron_weekday(dt: datetime) -> int: + return (dt.weekday() + 1) % 7 + + +def _parse_field(text: str, lo: int, hi: int, names: dict, label: str) -> tuple[frozenset, bool]: + values: set[int] = set() + restricted = False + for item in text.split(","): + item = item.strip().lower() + if not item: + raise ValueError(f"{label}: empty list item in {text!r}") + step = 1 + if "/" in item: + item, step_text = item.split("/", 1) + if not step_text.isdigit() or int(step_text) < 1: + raise ValueError(f"{label}: bad step in {text!r}") + step = int(step_text) + if item == "*": + start, end = lo, hi + if step > 1: + restricted = True + elif "-" in item: + a, b = item.split("-", 1) + start, end = _atom(a, lo, hi, names, label), _atom(b, lo, hi, names, label) + if end < start: + raise ValueError(f"{label}: descending range {item!r}") + restricted = True + else: + start = _atom(item, lo, hi, names, label) + end = hi if step > 1 else start + restricted = True + values.update(range(start, end + 1, step)) + return frozenset(values), restricted + + +def _atom(text: str, lo: int, hi: int, names: dict, label: str) -> int: + if text in names: + return names[text] + if not text.isdigit(): + raise ValueError(f"{label}: {text!r} is not a number or name") + value = int(text) + if label == "weekday" and value == 7: + value = 0 + if value < lo or value > hi: + raise ValueError(f"{label}: {value} outside {lo}-{hi}") + return value + + +def parse(expr: str) -> CronSpec: + """Parse a 5-field cron expression (minute hour day month weekday). + + Supports ``*``, lists, ranges, ``/step``, month and weekday names, ``7`` + for Sunday and the ``@daily``-style aliases. Raises ``ValueError``. + """ + text = (expr or "").strip().lower() + text = _ALIASES.get(text, text) + fields = text.split() + if len(fields) != 5: + raise ValueError(f"cron needs 5 fields, got {len(fields)}: {expr!r}") + minutes, _ = _parse_field(fields[0], 0, 59, {}, "minute") + hours, _ = _parse_field(fields[1], 0, 23, {}, "hour") + days, day_r = _parse_field(fields[2], 1, 31, {}, "day") + months, _ = _parse_field(fields[3], 1, 12, _MONTHS, "month") + weekdays, dow_r = _parse_field(fields[4], 0, 7, _DOWS, "weekday") + return CronSpec(minutes, hours, days, months, weekdays, day_r, dow_r) + + +def resolve_tz(name: str) -> tzinfo: + """``tzinfo`` for a schedule's ``tz``: an IANA name, or ``local``/empty. + + ``local`` honours ``C3_SCHEDULER_TZ`` when set, else the machine's + current UTC offset. Callers recompute ``next_run_at`` at every fire, so a + DST change is picked up one fire late at worst. Raises ``ValueError`` for + an unknown IANA name. + """ + key = (name or "").strip() + if not key or key.lower() == "local": + key = os.environ.get("C3_SCHEDULER_TZ", "").strip() + if not key: + return datetime.now().astimezone().tzinfo or timezone.utc + if key.upper() == "UTC": + return timezone.utc + try: + return ZoneInfo(key) + except (ZoneInfoNotFoundError, ValueError) as exc: + raise ValueError(f"unknown timezone {name!r}") from exc + + +def next_after(spec: CronSpec, after: datetime, tz: tzinfo | None = None) -> datetime | None: + """First instant matching ``spec`` strictly after ``after``, as aware UTC. + + ``after`` must be aware. Wall-clock matching happens in ``tz`` (UTC when + omitted). A wall time that does not exist on a DST-gap day fires at the + shifted instant; a repeated wall time fires once. ``None`` when no match + falls within ``SEARCH_LIMIT_DAYS``. + """ + if after.tzinfo is None: + raise ValueError("after must be timezone-aware") + zone = tz or timezone.utc + local = after.astimezone(zone).replace(second=0, microsecond=0) + timedelta(minutes=1) + after_utc = after.astimezone(timezone.utc) + hours = sorted(spec.hours) + minutes = sorted(spec.minutes) + day = local.replace(hour=0, minute=0) + first_day = True + for _ in range(SEARCH_LIMIT_DAYS): + if spec.matches_day(day): + for hour in hours: + for minute in minutes: + if first_day and (hour, minute) < (local.hour, local.minute): + continue + candidate = datetime(day.year, day.month, day.day, hour, minute, tzinfo=zone) + candidate_utc = candidate.astimezone(timezone.utc) + if candidate_utc > after_utc: + return candidate_utc + first_day = False + day = (day + timedelta(days=1, hours=2)).replace(hour=0, minute=0) + return None + + +def preset_to_cron(preset: str, at: str = "09:00", days=()) -> str: + """Cron expression for a preset at ``HH:MM``; ``days`` names the weekly days.""" + name = (preset or "").strip().lower() + if name not in PRESETS: + raise ValueError(f"unknown preset {preset!r} (use one of {', '.join(PRESETS)})") + m = _HHMM.match((at or "09:00").strip()) + if not m: + raise ValueError(f"preset time must be HH:MM, got {at!r}") + hour, minute = int(m.group(1)), int(m.group(2)) + if name == "hourly": + return f"{minute} * * * *" + if name == "daily": + return f"{minute} {hour} * * *" + if name == "weekdays": + return f"{minute} {hour} * * 1-5" + if name == "monthly": + return f"{minute} {hour} 1 * *" + dows = [] + for d in days or ("mon",): + text = str(d).strip().lower() + if text in _DOWS: + dows.append(_DOWS[text]) + elif text.isdigit() and 0 <= int(text) <= 7: + dows.append(int(text) % 7) + else: + raise ValueError(f"weekly preset: unknown day {d!r}") + return f"{minute} {hour} * * {','.join(str(d) for d in sorted(set(dows)))}" + + +def validate_when(when: dict) -> dict: + """Normalised copy of ``when`` or ``ValueError`` naming the first defect.""" + if not isinstance(when, dict): + raise ValueError("when must be an object") + kind = str(when.get("type") or "").strip().lower() + if kind not in WHEN_TYPES: + raise ValueError(f"when.type must be one of {', '.join(WHEN_TYPES)}") + out: dict = {"type": kind} + if kind == "cron": + expr = str(when.get("expr") or "").strip() + parse(expr) + out["expr"] = expr + out["tz"] = _tz_field(when) + elif kind == "every": + try: + seconds = int(when.get("seconds") or 0) + except (TypeError, ValueError) as exc: + raise ValueError("when.seconds must be an integer") from exc + if seconds < MIN_EVERY_SECONDS: + raise ValueError(f"when.seconds must be at least {MIN_EVERY_SECONDS}") + out["seconds"] = seconds + elif kind == "at": + out["when"] = parse_iso(str(when.get("when") or "")).isoformat() + else: + preset = str(when.get("preset") or "").strip().lower() + at = str(when.get("at") or "09:00").strip() + days = list(when.get("days") or []) + out.update({"preset": preset, "at": at, "days": days, "tz": _tz_field(when), + "expr": preset_to_cron(preset, at, days)}) + return out + + +def _tz_field(when: dict) -> str: + tz = str(when.get("tz") or "local").strip() + resolve_tz(tz) + return tz + + +def parse_iso(text: str) -> datetime: + """Aware UTC datetime from ISO-8601 text; a naive value is taken as UTC.""" + value = (text or "").strip() + if not value: + raise ValueError("empty datetime") + if value.endswith("Z"): + value = value[:-1] + "+00:00" + try: + dt = datetime.fromisoformat(value) + except ValueError as exc: + raise ValueError(f"not an ISO-8601 datetime: {text!r}") from exc + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc) + + +def next_fire(when: dict, after: datetime) -> datetime | None: + """Next aware-UTC fire of a validated ``when`` strictly after ``after``.""" + kind = when.get("type") + if kind in ("cron", "preset"): + return next_after(parse(when["expr"]), after, resolve_tz(when.get("tz", "local"))) + if kind == "every": + return after.astimezone(timezone.utc) + timedelta(seconds=int(when["seconds"])) + if kind == "at": + at = parse_iso(when["when"]) + return at if at > after.astimezone(timezone.utc) else None + raise ValueError(f"unknown when.type {kind!r}") + + +def preview(when: dict, after: datetime, count: int = 5) -> list[datetime]: + """The next ``count`` fires of ``when`` after ``after``.""" + out: list[datetime] = [] + cursor = after + for _ in range(max(0, count)): + nxt = next_fire(when, cursor) + if nxt is None: + break + out.append(nxt) + cursor = nxt + return out diff --git a/services/schedule_store.py b/services/schedule_store.py new file mode 100644 index 0000000..93c6117 --- /dev/null +++ b/services/schedule_store.py @@ -0,0 +1,598 @@ +"""Machine-wide store for scheduled tasks, their runs and the audit trail. + +Lives under ``~/.c3/oracle`` because the runner is per machine: one Oracle +owns the lock, the due-claim transaction and the approval inbox. Every row +still names its ``project_path`` so notifications and runs attach to the +right project. + +Files: ``schedules.json`` (all schedule rows, ``rev``-versioned), +``schedule_runs/.json`` (one mutable file per run), +``schedule_events.jsonl`` (append-only audit), ``schedules.lock``. + +Concurrency: the Oracle tick, the Hub and an MCP server are separate +processes that may all mutate ``schedules.json``. There is no in-memory +cache; every mutation is load -> mutate -> atomic save under an in-process +``threading.Lock`` plus the OS byte lock in ``schedules.lock``. ``claim_due`` +advances ``next_run_at`` inside that same transaction, so a row stops being +due the moment it is claimed, whichever process claimed it. +""" +from __future__ import annotations + +import json +import os +import threading +import time +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from services.atomic_json import write_json_atomic +from services.cron_expr import next_fire, parse_iso, validate_when + +SCHEMA_VERSION = 1 +KINDS = ("skill", "prompt", "command") +PROVIDERS = ("claude", "codex", "gemini", "grok") +EFFORT_LEVELS = ("", "low", "medium", "high", "xhigh", "max") +AUTONOMY = ("autonomous", "approval") +OVERLAP = ("skip", "allow") +MISFIRE_POLICIES = ("grace", "skip") +PERMISSION_MODES = ("acceptEdits", "auto", "bypassPermissions", "manual", "dontAsk", "plan") +TRIGGERS = ("cron", "manual", "approval", "check") + +RUN_STATUSES = ( + "claimed", "skipped_overlap", "missed", "check_running", "check_skipped", + "awaiting_approval", "approved", "denied", "expired", + "dispatched", "running", "done", "failed", "timeout", "cancelled", "lost", +) +OPEN_RUN_STATUSES = frozenset( + {"claimed", "check_running", "awaiting_approval", "approved", "dispatched", "running"}) +TERMINAL_RUN_STATUSES = frozenset(RUN_STATUSES) - OPEN_RUN_STATUSES +EXECUTED_STATUSES = frozenset({"done", "failed", "timeout", "cancelled", "lost"}) + +MIN_TIMEOUT_S = 30 +MAX_TIMEOUT_S = 6 * 3600 +DEFAULT_TIMEOUT_S = 1800 +DEFAULT_APPROVAL_TTL_S = 3600 +DEFAULT_MISFIRE_GRACE_S = 900 +RESULT_EXCERPT_CAP = 2000 +_LOCK_TIMEOUT_S = 30.0 +_EVENTS_ROTATE_BYTES = 5_000_000 + +_SCHEDULE_FIELDS = ( + "id", "name", "kind", "target", "provider", "model", "effort", "permission_mode", + "allowed_tools", "max_budget_usd", "autonomy", "approval_ttl_s", "project_path", "cwd", + "enabled", "timeout_s", "when", "misfire", "overlap", "check_command", + "next_run_at", "last_run_at", "last_run_id", "last_status", "run_count", + "created_at", "updated_at", "created_by", +) +_EDITABLE = frozenset(_SCHEDULE_FIELDS) - { + "id", "next_run_at", "last_run_at", "last_run_id", "last_status", "run_count", + "created_at", "updated_at", "created_by"} +_RUN_FIELDS = ( + "run_id", "schedule_id", "schedule_name", "project_path", "trigger", "due_at", "claimed_at", + "dispatched_at", "finished_at", "status", "job_id", "approval_id", "exit_code", "duration_ms", + "output_id", "cost_usd", "tokens", "turns", "model", "result_excerpt", "error", +) + + +class ScheduleError(ValueError): + """A schedule or run argument is unusable; the message names the field.""" + + +class RevConflict(ScheduleError): + """``expected_rev`` no longer matches the document on disk.""" + + +def default_root() -> Path: + return Path.home() / ".c3" / "oracle" + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _iso(dt: datetime | None) -> str | None: + return dt.astimezone(timezone.utc).isoformat() if dt else None + + +def _new_schedule_id() -> str: + return "sch-" + uuid.uuid4().hex[:8] + + +def _new_run_id() -> str: + return "sr-" + uuid.uuid4().hex[:10] + + +class _FileLock: + """Advisory cross-process lock on a sidecar file, released by the OS on death.""" + + def __init__(self, path): + self.path = Path(path) + self._fh = None + + def acquire(self, timeout_s: float = _LOCK_TIMEOUT_S) -> bool: + self.path.parent.mkdir(parents=True, exist_ok=True) + self._fh = open(self.path, "ab") # the fd exists only to hold the lock + deadline = time.monotonic() + timeout_s + while True: + try: + if os.name == "nt": + import msvcrt + self._fh.seek(0) + msvcrt.locking(self._fh.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + fcntl.flock(self._fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except OSError: + if time.monotonic() >= deadline: + self._fh.close() + self._fh = None + return False + time.sleep(0.05) + + def release(self) -> None: + if self._fh is None: + return + try: + if os.name == "nt": + import msvcrt + self._fh.seek(0) + msvcrt.locking(self._fh.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + fcntl.flock(self._fh.fileno(), fcntl.LOCK_UN) + except OSError: + pass + finally: + self._fh.close() + self._fh = None + + def __enter__(self): + if not self.acquire(): + raise TimeoutError(f"could not acquire {self.path.name} within {_LOCK_TIMEOUT_S:.0f}s") + return self + + def __exit__(self, exc_type, exc, tb): + self.release() + return False + + +def normalize_schedule(fields: dict, *, defaults: dict | None = None, existing: dict | None = None) -> dict: + """Validated schedule row from ``fields`` (merged over ``existing`` on update). + + ``defaults`` supplies ``timeout_s``, ``max_timeout_s``, ``approval_ttl_s``, + ``misfire_grace_s``, ``permission_mode``, ``allow_bypass_permissions`` and + ``max_budget_usd`` when the caller omits them. Raises ``ScheduleError``. + """ + d = defaults or {} + row = dict(existing or {}) + unknown = set(fields) - _EDITABLE + if unknown: + raise ScheduleError(f"unknown field(s): {', '.join(sorted(unknown))}") + row.update(fields) + + name = str(row.get("name") or "").strip() + if not name: + raise ScheduleError("name is required") + kind = str(row.get("kind") or "skill").strip().lower() + if kind not in KINDS: + raise ScheduleError(f"kind must be one of {', '.join(KINDS)}") + target = row.get("target") + if not isinstance(target, dict): + raise ScheduleError("target must be an object") + target = _normalize_target(kind, target) + provider = str(row.get("provider") or "claude").strip().lower() + if provider not in PROVIDERS: + raise ScheduleError(f"provider must be one of {', '.join(PROVIDERS)}") + effort = str(row.get("effort") or "").strip().lower() + if effort not in EFFORT_LEVELS: + raise ScheduleError(f"effort must be one of {', '.join(e for e in EFFORT_LEVELS if e)}") + mode = str(row.get("permission_mode") or d.get("permission_mode") or "acceptEdits").strip() + if mode not in PERMISSION_MODES: + raise ScheduleError(f"permission_mode must be one of {', '.join(PERMISSION_MODES)}") + if mode == "bypassPermissions" and not d.get("allow_bypass_permissions", False): + raise ScheduleError("permission_mode bypassPermissions is disabled by scheduler_allow_bypass_permissions") + autonomy = str(row.get("autonomy") or "approval").strip().lower() + if autonomy not in AUTONOMY: + raise ScheduleError(f"autonomy must be one of {', '.join(AUTONOMY)}") + overlap = str(row.get("overlap") or "skip").strip().lower() + if overlap not in OVERLAP: + raise ScheduleError(f"overlap must be one of {', '.join(OVERLAP)}") + project_path = str(row.get("project_path") or "").strip() + if not project_path: + raise ScheduleError("project_path is required") + cwd = str(row.get("cwd") or project_path).strip() + + timeout_s = _int_in(row.get("timeout_s", d.get("timeout_s", DEFAULT_TIMEOUT_S)), + MIN_TIMEOUT_S, int(d.get("max_timeout_s", MAX_TIMEOUT_S)), "timeout_s") + approval_ttl_s = _int_in(row.get("approval_ttl_s", d.get("approval_ttl_s", DEFAULT_APPROVAL_TTL_S)), + 60, 7 * 86400, "approval_ttl_s") + misfire = row.get("misfire") + if misfire is None: + misfire = {} + if not isinstance(misfire, dict): + raise ScheduleError("misfire must be an object") + policy = str(misfire.get("policy") or "grace").strip().lower() + if policy not in MISFIRE_POLICIES: + raise ScheduleError(f"misfire.policy must be one of {', '.join(MISFIRE_POLICIES)}") + grace_s = _int_in(misfire.get("grace_s", d.get("misfire_grace_s", DEFAULT_MISFIRE_GRACE_S)), + 0, 7 * 86400, "misfire.grace_s") + try: + budget = float(row.get("max_budget_usd", d.get("max_budget_usd", 0)) or 0) + except (TypeError, ValueError) as exc: + raise ScheduleError("max_budget_usd must be a number") from exc + if budget < 0: + raise ScheduleError("max_budget_usd must be >= 0") + try: + when = validate_when(row.get("when")) + except ValueError as exc: + raise ScheduleError(str(exc)) from exc + + row.update({ + "name": name, "kind": kind, "target": target, "provider": provider, + "model": str(row.get("model") or "").strip(), "effort": effort, + "permission_mode": mode, "allowed_tools": str(row.get("allowed_tools") or "").strip(), + "max_budget_usd": budget, "autonomy": autonomy, "approval_ttl_s": approval_ttl_s, + "project_path": project_path, "cwd": cwd, "enabled": bool(row.get("enabled", True)), + "timeout_s": timeout_s, "when": when, + "misfire": {"policy": policy, "grace_s": grace_s}, "overlap": overlap, + "check_command": str(row.get("check_command") or "").strip(), + }) + return row + + +def _normalize_target(kind: str, target: dict) -> dict: + if kind == "skill": + skill = str(target.get("skill") or "").strip() + if not skill.startswith("/") or len(skill) < 2 or " " in skill: + raise ScheduleError("target.skill must look like /skill-name") + return {"skill": skill, "args": str(target.get("args") or "").strip()} + if kind == "prompt": + prompt = str(target.get("prompt") or "").strip() + if not prompt: + raise ScheduleError("target.prompt is required") + return {"prompt": prompt} + cmd = str(target.get("cmd") or "").strip() + if not cmd: + raise ScheduleError("target.cmd is required") + return {"cmd": cmd} + + +def _int_in(value, lo: int, hi: int, label: str) -> int: + try: + number = int(value) + except (TypeError, ValueError) as exc: + raise ScheduleError(f"{label} must be an integer") from exc + if number < lo or number > hi: + raise ScheduleError(f"{label} must be between {lo} and {hi}") + return number + + +def prompt_text(schedule: dict) -> str: + """The text handed to an LLM provider for a skill or prompt schedule.""" + target = schedule.get("target") or {} + if schedule.get("kind") == "skill": + args = target.get("args") or "" + return f"{target.get('skill', '')} {args}".strip() + return str(target.get("prompt") or "") + + +class ScheduleStore: + """Schedules, runs and events under one root. See the module docstring.""" + + def __init__(self, root: Path | str | None = None): + self.root = Path(root) if root else default_root() + self.schedules_file = self.root / "schedules.json" + self.runs_dir = self.root / "schedule_runs" + self.events_file = self.root / "schedule_events.jsonl" + self.lock_file = self.root / "schedules.lock" + self._lock = threading.Lock() + + def _empty_doc(self) -> dict: + return {"schema_version": SCHEMA_VERSION, "rev": 0, "schedules": []} + + def _load(self) -> dict: + if not self.schedules_file.exists(): + return self._empty_doc() + try: + doc = json.loads(self.schedules_file.read_text(encoding="utf-8")) + except (OSError, ValueError): + return self._empty_doc() + if not isinstance(doc, dict) or not isinstance(doc.get("schedules"), list): + return self._empty_doc() + doc.setdefault("schema_version", SCHEMA_VERSION) + doc.setdefault("rev", 0) + return doc + + def _save(self, doc: dict) -> None: + doc["rev"] = int(doc.get("rev", 0)) + 1 + write_json_atomic(self.schedules_file, doc, ensure_ascii=False) + + def _txn(self): + return _Txn(self) + + def _event(self, event: str, **data) -> None: + self.root.mkdir(parents=True, exist_ok=True) + try: + if self.events_file.exists() and self.events_file.stat().st_size > _EVENTS_ROTATE_BYTES: + os.replace(self.events_file, self.events_file.with_suffix(".jsonl.1")) + except OSError: + pass + row = {"ts": _iso(_now()), "event": event, **data} + with open(self.events_file, "a", encoding="utf-8") as fh: + fh.write(json.dumps(row, ensure_ascii=False) + "\n") + + def list(self, *, project_path: str | None = None, enabled: bool | None = None) -> list[dict]: + rows = self._load()["schedules"] + if project_path: + key = _path_key(project_path) + rows = [r for r in rows if _path_key(r.get("project_path", "")) == key] + if enabled is not None: + rows = [r for r in rows if bool(r.get("enabled")) == enabled] + return sorted(rows, key=lambda r: (r.get("next_run_at") or "~", r.get("name", ""))) + + def get(self, schedule_id: str) -> dict | None: + for row in self._load()["schedules"]: + if row.get("id") == schedule_id: + return row + return None + + def rev(self) -> int: + return int(self._load().get("rev", 0)) + + def create(self, fields: dict, *, defaults: dict | None = None, created_by: str = "") -> dict: + now = _now() + row = normalize_schedule(fields, defaults=defaults) + row.update({ + "id": _new_schedule_id(), "created_at": _iso(now), "updated_at": _iso(now), + "created_by": created_by, "last_run_at": None, "last_run_id": None, + "last_status": None, "run_count": 0, + }) + row["next_run_at"] = _iso(next_fire(row["when"], now)) if row["enabled"] else None + with self._txn() as doc: + doc["schedules"].append(row) + self._event("schedule.created", schedule_id=row["id"], name=row["name"], + project_path=row["project_path"], by=created_by) + return row + + def update(self, schedule_id: str, patch: dict, *, expected_rev: int | None = None, + defaults: dict | None = None, by: str = "") -> dict: + with self._txn() as doc: + if expected_rev is not None and int(doc.get("rev", 0)) != int(expected_rev): + raise RevConflict(f"rev is {doc.get('rev')}, expected {expected_rev}") + row = _find(doc, schedule_id) + if row is None: + raise ScheduleError(f"no schedule {schedule_id}") + updated = normalize_schedule(patch, defaults=defaults, existing=row) + now = _now() + updated["updated_at"] = _iso(now) + if not updated["enabled"]: + updated["next_run_at"] = None + elif "when" in patch or not row.get("enabled") or not row.get("next_run_at"): + updated["next_run_at"] = _iso(next_fire(updated["when"], now)) + row.clear() + row.update(updated) + result = dict(row) + self._event("schedule.updated", schedule_id=schedule_id, fields=sorted(patch), by=by) + return result + + def set_enabled(self, schedule_id: str, enabled: bool, *, by: str = "") -> dict: + return self.update(schedule_id, {"enabled": bool(enabled)}, by=by) + + def delete(self, schedule_id: str, *, by: str = "") -> bool: + with self._txn() as doc: + before = len(doc["schedules"]) + doc["schedules"] = [r for r in doc["schedules"] if r.get("id") != schedule_id] + removed = len(doc["schedules"]) != before + if removed: + self._event("schedule.deleted", schedule_id=schedule_id, by=by) + return removed + + def claim_due(self, now: datetime | None = None) -> list[tuple[dict, dict]]: + """Claim every enabled schedule whose ``next_run_at`` has passed. + + Inside one locked transaction each claimed row gets its ``next_run_at`` + advanced (a one-shot ``at`` is disabled instead) and a ``claimed`` run + row is written, so a second caller sees nothing due. Returns + ``(schedule, run)`` pairs; ``run.due_at`` keeps the slot that fired. + """ + now = now or _now() + claimed: list[tuple[dict, dict]] = [] + with self._txn() as doc: + for row in doc["schedules"]: + due = row.get("next_run_at") + if not row.get("enabled") or not due: + continue + if parse_iso(due) > now: + continue + nxt = next_fire(row["when"], now) + if row["when"].get("type") == "at" or nxt is None: + row["enabled"] = False + row["next_run_at"] = None + else: + row["next_run_at"] = _iso(nxt) + row["updated_at"] = _iso(now) + run = self._new_run(row, trigger="cron", due_at=due, now=now) + claimed.append((dict(row), run)) + for row, run in claimed: + self._event("run.claimed", run_id=run["run_id"], schedule_id=row["id"], + 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.""" + if trigger not in TRIGGERS: + raise ScheduleError(f"trigger must be one of {', '.join(TRIGGERS)}") + now = _now() + with self._txn() as doc: + 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) + self._event("run.claimed", run_id=run["run_id"], schedule_id=schedule_id, + due_at=run["due_at"], trigger=trigger, by=by) + return run + + def _new_run(self, row: dict, *, trigger: str, due_at: str, now: datetime) -> 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": "", + }) + self._write_run(run) + return run + + def _run_path(self, run_id: str) -> Path: + return self.runs_dir / f"{run_id}.json" + + def _write_run(self, run: dict) -> None: + write_json_atomic(self._run_path(run["run_id"]), run, ensure_ascii=False, fsync=False) + + def get_run(self, run_id: str) -> dict | None: + path = self._run_path(run_id) + if not path.exists(): + return None + try: + run = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + return run if isinstance(run, dict) else None + + def update_run(self, run_id: str, **patch) -> dict: + unknown = set(patch) - set(_RUN_FIELDS) + if unknown: + raise ScheduleError(f"unknown run field(s): {', '.join(sorted(unknown))}") + status = patch.get("status") + if status is not None and status not in RUN_STATUSES: + raise ScheduleError(f"unknown run status {status!r}") + with self._lock: + run = self.get_run(run_id) + if run is None: + raise ScheduleError(f"no run {run_id}") + run.update(patch) + if "result_excerpt" in patch and run.get("result_excerpt"): + run["result_excerpt"] = str(run["result_excerpt"])[:RESULT_EXCERPT_CAP] + self._write_run(run) + if status is not None: + self._event("run.status", run_id=run_id, schedule_id=run.get("schedule_id"), + status=status, job_id=run.get("job_id")) + return run + + def finish_run(self, run_id: str, status: str, **patch) -> dict: + """Terminal update of a run plus the owning schedule's ``last_*`` summary.""" + if status not in TERMINAL_RUN_STATUSES: + raise ScheduleError(f"{status!r} is not a terminal run status") + now = _now() + run = self.update_run(run_id, status=status, finished_at=_iso(now), **patch) + with self._txn() as doc: + row = _find(doc, run.get("schedule_id", "")) + if row is not None: + row["last_run_at"] = _iso(now) + row["last_run_id"] = run_id + row["last_status"] = status + if status in EXECUTED_STATUSES: + row["run_count"] = int(row.get("run_count") or 0) + 1 + return run + + def runs(self, *, schedule_id: str | None = None, status=None, project_path: str | None = None, + limit: int = 50) -> list[dict]: + statuses = None + if status: + statuses = {status} if isinstance(status, str) else set(status) + key = _path_key(project_path) if project_path else None + out = [] + for run in self._iter_runs(): + if schedule_id and run.get("schedule_id") != schedule_id: + continue + if statuses and run.get("status") not in statuses: + continue + if key and _path_key(run.get("project_path", "")) != key: + continue + out.append(run) + out.sort(key=lambda r: r.get("claimed_at") or "", reverse=True) + return out[:max(0, int(limit))] if limit else out + + def open_runs(self) -> list[dict]: + return self.runs(status=OPEN_RUN_STATUSES, limit=0) + + def open_run_for(self, schedule_id: str, *, exclude: str = "") -> dict | None: + for run in self.open_runs(): + if run.get("schedule_id") == schedule_id and run.get("run_id") != exclude: + return run + return None + + def _iter_runs(self): + if not self.runs_dir.exists(): + return + for path in self.runs_dir.glob("sr-*.json"): + try: + run = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + if isinstance(run, dict) and run.get("run_id"): + yield run + + def sweep_runs(self, retention_days: int, now: datetime | None = None) -> int: + """Delete terminal runs older than ``retention_days``; returns the count.""" + now = now or _now() + cutoff = now - timedelta(days=max(1, int(retention_days))) + removed = 0 + for run in list(self._iter_runs()): + if run.get("status") in OPEN_RUN_STATUSES: + continue + stamp = run.get("finished_at") or run.get("claimed_at") or "" + try: + when = parse_iso(stamp) + except ValueError: + continue + if when < cutoff: + try: + self._run_path(run["run_id"]).unlink() + removed += 1 + except OSError: + pass + return removed + + +class _Txn: + """``with store._txn() as doc:`` — locked load, mutate, save on success.""" + + def __init__(self, store: ScheduleStore): + self.store = store + self.flock = _FileLock(store.lock_file) + self.doc = None + + def __enter__(self) -> dict: + self.store._lock.acquire() + try: + self.flock.__enter__() + except Exception: + self.store._lock.release() + raise + self.doc = self.store._load() + return self.doc + + def __exit__(self, exc_type, exc, tb): + try: + if exc_type is None: + self.store._save(self.doc) + finally: + self.flock.release() + self.store._lock.release() + return False + + +def _find(doc: dict, schedule_id: str) -> dict | None: + for row in doc["schedules"]: + if row.get("id") == schedule_id: + return row + return None + + +def _path_key(path: str) -> str: + try: + return os.path.normcase(os.path.normpath(str(Path(path).resolve()))) + except OSError: + return os.path.normcase(os.path.normpath(str(path))) diff --git a/tests/test_cron_expr.py b/tests/test_cron_expr.py new file mode 100644 index 0000000..fa91b0a --- /dev/null +++ b/tests/test_cron_expr.py @@ -0,0 +1,145 @@ +import unittest +from datetime import datetime, timedelta, timezone +from zoneinfo import ZoneInfo + +from services import cron_expr as ce + +UTC = timezone.utc + + +def at(y, m, d, hh=0, mm=0, tz=UTC): + return datetime(y, m, d, hh, mm, tzinfo=tz) + + +class TestParse(unittest.TestCase): + def test_star_fields_are_unrestricted(self): + spec = ce.parse("* * * * *") + self.assertEqual(len(spec.minutes), 60) + self.assertEqual(len(spec.hours), 24) + self.assertFalse(spec.day_restricted) + self.assertFalse(spec.weekday_restricted) + + def test_lists_ranges_steps_and_names(self): + spec = ce.parse("0,30 9-17/4 1,15 jan,JUL mon-fri") + self.assertEqual(spec.minutes, frozenset({0, 30})) + self.assertEqual(spec.hours, frozenset({9, 13, 17})) + self.assertEqual(spec.days, frozenset({1, 15})) + self.assertEqual(spec.months, frozenset({1, 7})) + self.assertEqual(spec.weekdays, frozenset({1, 2, 3, 4, 5})) + + def test_seven_is_sunday_and_aliases_expand(self): + self.assertEqual(ce.parse("0 0 * * 7").weekdays, frozenset({0})) + self.assertEqual(ce.parse("@weekly").weekdays, frozenset({0})) + self.assertEqual(ce.parse("@hourly").minutes, frozenset({0})) + + def test_rejects_bad_expressions(self): + for expr in ("* * * *", "60 * * * *", "* 24 * * *", "* * 0 * *", "* * * 13 *", + "*/0 * * * *", "5-1 * * * *", "x * * * *", ""): + with self.assertRaises(ValueError, msg=expr): + ce.parse(expr) + + +class TestNextAfter(unittest.TestCase): + def test_strictly_after_and_minute_floor(self): + spec = ce.parse("*/15 * * * *") + base = at(2026, 10, 13, 9, 0) + timedelta(seconds=20) + self.assertEqual(ce.next_after(spec, base), at(2026, 10, 13, 9, 15)) + self.assertEqual(ce.next_after(spec, at(2026, 10, 13, 9, 15)), at(2026, 10, 13, 9, 30)) + + def test_weekdays_skip_the_weekend(self): + spec = ce.parse("0 9 * * 1-5") + friday = at(2026, 10, 16, 10, 0) + self.assertEqual(ce.next_after(spec, friday), at(2026, 10, 19, 9, 0)) + + def test_day_and_weekday_both_restricted_is_or(self): + spec = ce.parse("0 0 13 * 5") + self.assertEqual(ce.next_after(spec, at(2026, 10, 10)), at(2026, 10, 13)) + self.assertEqual(ce.next_after(spec, at(2026, 10, 13)), at(2026, 10, 16)) + + def test_month_end_rollover(self): + spec = ce.parse("0 0 31 * *") + self.assertEqual(ce.next_after(spec, at(2026, 2, 1)), at(2026, 3, 31)) + + def test_impossible_date_returns_none(self): + self.assertIsNone(ce.next_after(ce.parse("0 0 30 2 *"), at(2026, 1, 1))) + + def test_local_wall_clock_in_named_zone(self): + ny = ZoneInfo("America/New_York") + spec = ce.parse("0 9 * * *") + got = ce.next_after(spec, at(2026, 1, 5, 12, 0), tz=ny) + self.assertEqual(got, at(2026, 1, 5, 14, 0)) + self.assertEqual(got.tzinfo, UTC) + + def test_dst_gap_fires_once_at_shifted_instant(self): + ny = ZoneInfo("America/New_York") + spec = ce.parse("30 2 * * *") + first = ce.next_after(spec, at(2026, 3, 7, 12, 0), tz=ny) + second = ce.next_after(spec, first, tz=ny) + self.assertEqual(first, at(2026, 3, 8, 7, 30)) + self.assertEqual(second, at(2026, 3, 9, 6, 30)) + + def test_dst_overlap_fires_once(self): + ny = ZoneInfo("America/New_York") + spec = ce.parse("30 1 * * *") + first = ce.next_after(spec, at(2026, 10, 31, 12, 0), tz=ny) + second = ce.next_after(spec, first, tz=ny) + self.assertEqual(first, at(2026, 11, 1, 5, 30)) + self.assertEqual(second, at(2026, 11, 2, 6, 30)) + + def test_naive_after_is_rejected(self): + with self.assertRaises(ValueError): + ce.next_after(ce.parse("* * * * *"), datetime(2026, 1, 1)) + + +class TestWhenShapes(unittest.TestCase): + def test_presets_build_cron(self): + self.assertEqual(ce.preset_to_cron("daily", "09:05"), "5 9 * * *") + self.assertEqual(ce.preset_to_cron("weekdays", "18:00"), "0 18 * * 1-5") + self.assertEqual(ce.preset_to_cron("hourly", "00:15"), "15 * * * *") + self.assertEqual(ce.preset_to_cron("monthly", "08:00"), "0 8 1 * *") + self.assertEqual(ce.preset_to_cron("weekly", "07:00", ["fri", "Mon"]), "0 7 * * 1,5") + with self.assertRaises(ValueError): + ce.preset_to_cron("fortnightly") + with self.assertRaises(ValueError): + ce.preset_to_cron("daily", "9am") + + def test_validate_when_normalises_each_type(self): + self.assertEqual(ce.validate_when({"type": "cron", "expr": "0 9 * * *"}), + {"type": "cron", "expr": "0 9 * * *", "tz": "local"}) + self.assertEqual(ce.validate_when({"type": "every", "seconds": "600"}), + {"type": "every", "seconds": 600}) + self.assertEqual(ce.validate_when({"type": "at", "when": "2026-10-14T09:00Z"})["when"], + "2026-10-14T09:00:00+00:00") + preset = ce.validate_when({"type": "preset", "preset": "weekdays", "at": "09:00"}) + self.assertEqual(preset["expr"], "0 9 * * 1-5") + + def test_validate_when_rejects(self): + bad = [{"type": "weekly"}, {"type": "every", "seconds": 5}, {"type": "at", "when": "soon"}, + {"type": "cron", "expr": "0 9 * * *", "tz": "Mars/Olympus"}, "0 9 * * *"] + for when in bad: + with self.assertRaises(ValueError, msg=repr(when)): + ce.validate_when(when) + + def test_next_fire_by_type(self): + now = at(2026, 10, 13, 9, 0) + self.assertEqual(ce.next_fire({"type": "every", "seconds": 600}, now), at(2026, 10, 13, 9, 10)) + self.assertEqual(ce.next_fire({"type": "cron", "expr": "0 10 * * *", "tz": "UTC"}, now), + at(2026, 10, 13, 10, 0)) + self.assertEqual(ce.next_fire({"type": "at", "when": "2026-10-14T00:00:00+00:00"}, now), + at(2026, 10, 14)) + self.assertIsNone(ce.next_fire({"type": "at", "when": "2026-10-13T00:00:00+00:00"}, now)) + + def test_preview_counts_and_stops(self): + fires = ce.preview({"type": "cron", "expr": "0 * * * *", "tz": "UTC"}, at(2026, 1, 1, 0, 30), 3) + self.assertEqual([f.hour for f in fires], [1, 2, 3]) + once = ce.preview({"type": "at", "when": "2026-01-02T00:00:00+00:00"}, at(2026, 1, 1), 5) + self.assertEqual(len(once), 1) + + def test_resolve_tz(self): + self.assertEqual(ce.resolve_tz("UTC"), UTC) + self.assertIsNotNone(ce.resolve_tz("local").utcoffset(datetime.now())) + self.assertEqual(str(ce.resolve_tz("Europe/Berlin")), "Europe/Berlin") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_schedule_store.py b/tests/test_schedule_store.py new file mode 100644 index 0000000..8080e56 --- /dev/null +++ b/tests/test_schedule_store.py @@ -0,0 +1,250 @@ +import json +import os +import subprocess +import sys +import tempfile +import threading +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from services import schedule_store as ss + +REPO = Path(__file__).resolve().parents[1] + + +def fields(**over): + base = { + "name": "daily report", "kind": "skill", "target": {"skill": "/daily_report"}, + "project_path": str(REPO), "when": {"type": "every", "seconds": 3600}, + "autonomy": "autonomous", + } + base.update(over) + return base + + +class StoreBase(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name) / "oracle" + self.store = ss.ScheduleStore(self.root) + + def tearDown(self): + self._tmp.cleanup() + + +class TestNormalize(StoreBase): + def test_defaults_and_shape(self): + row = self.store.create(fields(), created_by="test") + self.assertTrue(row["id"].startswith("sch-")) + self.assertEqual(row["provider"], "claude") + self.assertEqual(row["permission_mode"], "acceptEdits") + self.assertEqual(row["timeout_s"], ss.DEFAULT_TIMEOUT_S) + self.assertEqual(row["misfire"], {"policy": "grace", "grace_s": ss.DEFAULT_MISFIRE_GRACE_S}) + self.assertEqual(row["cwd"], str(REPO)) + self.assertIsNotNone(row["next_run_at"]) + self.assertEqual(row["run_count"], 0) + self.assertEqual(self.store.rev(), 1) + + def test_agent_default_is_approval(self): + row = self.store.create(fields(autonomy=None)) + self.assertEqual(row["autonomy"], "approval") + + def test_rejections(self): + cases = [ + fields(name=""), fields(kind="cron"), fields(target={"skill": "daily"}), + fields(target={"skill": "/a b"}), fields(kind="prompt", target={}), + fields(kind="command", target={"cmd": ""}), fields(provider="gpt"), + fields(effort="ultra"), fields(permission_mode="yolo"), + fields(permission_mode="bypassPermissions"), fields(autonomy="maybe"), + fields(overlap="queue"), fields(project_path=""), fields(timeout_s=5), + fields(misfire={"policy": "retry"}), fields(max_budget_usd=-1), + fields(when={"type": "cron", "expr": "bad"}), fields(nonsense=1), + ] + for case in cases: + with self.assertRaises(ss.ScheduleError, msg=json.dumps(case)): + self.store.create(case) + + def test_bypass_needs_the_switch(self): + row = self.store.create(fields(permission_mode="bypassPermissions"), + defaults={"allow_bypass_permissions": True}) + self.assertEqual(row["permission_mode"], "bypassPermissions") + + def test_defaults_argument_fills_omitted_fields(self): + row = self.store.create(fields(), defaults={"timeout_s": 300, "misfire_grace_s": 60, + "approval_ttl_s": 120, "max_budget_usd": 1.5}) + self.assertEqual((row["timeout_s"], row["misfire"]["grace_s"], row["approval_ttl_s"], + row["max_budget_usd"]), (300, 60, 120, 1.5)) + + def test_prompt_text(self): + self.assertEqual(ss.prompt_text({"kind": "skill", "target": {"skill": "/x", "args": "a b"}}), "/x a b") + self.assertEqual(ss.prompt_text({"kind": "prompt", "target": {"prompt": "hi"}}), "hi") + + +class TestCrud(StoreBase): + def test_update_recomputes_next_run_only_when_needed(self): + row = self.store.create(fields()) + first = row["next_run_at"] + same = self.store.update(row["id"], {"name": "renamed"}) + self.assertEqual(same["next_run_at"], first) + moved = self.store.update(row["id"], {"when": {"type": "every", "seconds": 120}}) + self.assertNotEqual(moved["next_run_at"], first) + off = self.store.set_enabled(row["id"], False) + self.assertIsNone(off["next_run_at"]) + on = self.store.set_enabled(row["id"], True) + self.assertIsNotNone(on["next_run_at"]) + + def test_rev_conflict(self): + row = self.store.create(fields()) + rev = self.store.rev() + self.store.update(row["id"], {"name": "a"}) + with self.assertRaises(ss.RevConflict): + self.store.update(row["id"], {"name": "b"}, expected_rev=rev) + self.assertEqual(self.store.get(row["id"])["name"], "a") + + def test_delete_and_list_filters(self): + a = self.store.create(fields(name="a")) + self.store.create(fields(name="b", enabled=False, project_path=str(REPO / "tests"))) + self.assertEqual(len(self.store.list()), 2) + self.assertEqual([r["name"] for r in self.store.list(enabled=True)], ["a"]) + self.assertEqual([r["name"] for r in self.store.list(project_path=str(REPO / "tests"))], ["b"]) + self.assertTrue(self.store.delete(a["id"])) + self.assertFalse(self.store.delete(a["id"])) + self.assertIsNone(self.store.get(a["id"])) + + def test_corrupt_document_reads_as_empty(self): + self.store.create(fields()) + self.store.schedules_file.write_text("{not json", encoding="utf-8") + self.assertEqual(self.store.list(), []) + + def test_events_are_appended(self): + row = self.store.create(fields()) + self.store.delete(row["id"]) + lines = self.store.events_file.read_text(encoding="utf-8").splitlines() + self.assertEqual([json.loads(l)["event"] for l in lines], ["schedule.created", "schedule.deleted"]) + + +class TestClaim(StoreBase): + def _due(self, **over): + row = self.store.create(fields(**over)) + past = (datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat() + with self.store._txn() as doc: + ss._find(doc, row["id"])["next_run_at"] = past + return self.store.get(row["id"]) + + def test_claim_marks_next_run_before_returning(self): + row = self._due() + claimed = self.store.claim_due() + self.assertEqual(len(claimed), 1) + sched, run = claimed[0] + self.assertEqual(run["status"], "claimed") + self.assertEqual(run["due_at"], row["next_run_at"]) + self.assertGreater(sched["next_run_at"], datetime.now(timezone.utc).isoformat()) + self.assertEqual(self.store.claim_due(), []) + self.assertEqual(self.store.get_run(run["run_id"])["schedule_id"], row["id"]) + + def test_disabled_and_future_rows_are_not_claimed(self): + self._due(enabled=False) + self.store.create(fields()) + self.assertEqual(self.store.claim_due(), []) + + def test_one_shot_disables_at_claim(self): + when = {"type": "at", "when": (datetime.now(timezone.utc) + timedelta(days=1)).isoformat()} + row = self.store.create(fields(when=when)) + with self.store._txn() as doc: + ss._find(doc, row["id"])["next_run_at"] = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat() + self.assertEqual(len(self.store.claim_due()), 1) + after = self.store.get(row["id"]) + self.assertFalse(after["enabled"]) + self.assertIsNone(after["next_run_at"]) + + def test_concurrent_threads_claim_once(self): + self._due() + results = [] + barrier = threading.Barrier(4) + + def worker(): + store = ss.ScheduleStore(self.root) + barrier.wait() + results.append(len(store.claim_due())) + + threads = [threading.Thread(target=worker) for _ in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(sum(results), 1) + self.assertEqual(len(self.store.runs()), 1) + + def test_concurrent_processes_claim_once(self): + self._due() + code = ("import sys; from services.schedule_store import ScheduleStore; " + f"print(len(ScheduleStore({str(self.root)!r}).claim_due()))") + env = dict(os.environ, PYTHONPATH=str(REPO)) + procs = [subprocess.Popen([sys.executable, "-c", code], cwd=str(REPO), env=env, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + for _ in range(3)] + outs = [p.communicate(timeout=60) for p in procs] + for p, (out, err) in zip(procs, outs): + self.assertEqual(p.returncode, 0, err) + self.assertEqual(sum(int(out.strip()) for out, _ in outs), 1) + + def test_request_run_and_open_run_queries(self): + row = self.store.create(fields()) + run = self.store.request_run(row["id"], by="hub") + self.assertEqual(run["trigger"], "manual") + self.assertEqual(self.store.open_run_for(row["id"])["run_id"], run["run_id"]) + self.assertIsNone(self.store.open_run_for(row["id"], exclude=run["run_id"])) + with self.assertRaises(ss.ScheduleError): + self.store.request_run("sch-nope") + + +class TestRuns(StoreBase): + def test_update_and_finish(self): + row = self.store.create(fields()) + run = self.store.request_run(row["id"]) + self.store.update_run(run["run_id"], status="dispatched", job_id="j1") + with self.assertRaises(ss.ScheduleError): + self.store.update_run(run["run_id"], status="exploded") + with self.assertRaises(ss.ScheduleError): + self.store.update_run(run["run_id"], bogus=1) + with self.assertRaises(ss.ScheduleError): + self.store.finish_run(run["run_id"], "running") + done = self.store.finish_run(run["run_id"], "done", exit_code=0, cost_usd=0.12, + result_excerpt="x" * 5000) + self.assertEqual(len(done["result_excerpt"]), ss.RESULT_EXCERPT_CAP) + sched = self.store.get(row["id"]) + self.assertEqual((sched["last_status"], sched["last_run_id"], sched["run_count"]), + ("done", run["run_id"], 1)) + skipped = self.store.request_run(row["id"]) + self.store.finish_run(skipped["run_id"], "missed") + self.assertEqual(self.store.get(row["id"])["run_count"], 1) + + def test_runs_filters_and_order(self): + a = self.store.create(fields(name="a")) + b = self.store.create(fields(name="b", project_path=str(REPO / "tests"))) + r1 = self.store.request_run(a["id"]) + r2 = self.store.request_run(b["id"]) + self.store.finish_run(r1["run_id"], "done") + self.assertEqual([r["run_id"] for r in self.store.runs(schedule_id=b["id"])], [r2["run_id"]]) + self.assertEqual([r["run_id"] for r in self.store.runs(status="done")], [r1["run_id"]]) + self.assertEqual(len(self.store.runs(project_path=str(REPO / "tests"))), 1) + self.assertEqual(len(self.store.open_runs()), 1) + self.assertEqual(len(self.store.runs(limit=1)), 1) + + def test_sweep_keeps_open_and_recent(self): + row = self.store.create(fields()) + old = self.store.request_run(row["id"]) + self.store.finish_run(old["run_id"], "done") + self.store.update_run(old["run_id"], finished_at=(datetime.now(timezone.utc) - timedelta(days=40)).isoformat()) + fresh = self.store.request_run(row["id"]) + self.store.finish_run(fresh["run_id"], "failed") + stuck = self.store.request_run(row["id"]) + self.store.update_run(stuck["run_id"], claimed_at=(datetime.now(timezone.utc) - timedelta(days=90)).isoformat()) + self.assertEqual(self.store.sweep_runs(30), 1) + left = {r["run_id"] for r in self.store.runs()} + self.assertEqual(left, {fresh["run_id"], stuck["run_id"]}) + + +if __name__ == "__main__": + unittest.main()