Gabriel Vidal
d5cfc97467
A scheduled job has no composer to pick a harness, model or effort in, so it now says so itself: the leading YAML block of `.claude/agents/<job>.md` carries `harness`, `model`, `effort` and `thinking`. `cron.run_config()` reads and normalizes them and `_fire_cron_job` hands them to the same `_send_message` path the composer uses, so a firing runs exactly like a hand-spawned session. The keys are a subset of Claude Code's own agent frontmatter, so a job's prompt file stays a usable `.claude/agents` subagent. Invalid values are dropped rather than raised — a firing on the defaults beats a job that stops firing — and the resolved config comes back on the job as `runConfig`, rendered as chips in Settings → Cron so an unapplied key is visible. New jobs get a stub carrying the default frontmatter. The store's per-job `model` is now only the fallback. All four homelab jobs are set to Claude Code + sonnet + no effort. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
458 lines
18 KiB
Python
458 lines
18 KiB
Python
"""Cron jobs: scheduled `claude -p` sessions that work toward goals.
|
|
|
|
A job is a 5-field cron schedule plus a **prompt file** (a markdown file under
|
|
``.claude/agents/`` in the workspace). When the schedule fires, the backend
|
|
spawns a new agent session (via the host sidecar, exactly like the composer's
|
|
spawn) with the file's content as the prompt, and records the run in the job's
|
|
``history`` — a JSON mapping ``{ISO timestamp: sessionId}`` — so every run
|
|
links back to its conversation in the viewer.
|
|
|
|
**How a job runs is declared in the prompt file's frontmatter**, not in the
|
|
store: ``harness``, ``model``, ``effort`` and ``thinking`` are read out of the
|
|
leading YAML block of ``.claude/agents/<job>.md`` (see ``run_config``) and
|
|
passed to the same spawn path the composer uses. The prompt file is the single
|
|
place a job is authored — schedule in the store, everything about the run next
|
|
to the prompt it runs.
|
|
|
|
Storage is a single JSON sidecar (``/data/cron-jobs.json``, same pattern as
|
|
``meta.py``) keyed by job id:
|
|
|
|
{ "<jobId>": {
|
|
"id": "goal-keeper",
|
|
"name": "Goal keeper",
|
|
"schedule": "0 */5 * * *", # min hour dom mon dow
|
|
"promptFile": ".claude/agents/goal-keeper.md",
|
|
"enabled": true,
|
|
"model": null, # legacy fallback for frontmatter
|
|
"createdAt": iso,
|
|
"lastFired": "YYYY-MM-DDTHH:MM", # minute claim (double-fire guard)
|
|
"lastStatus": "ok" | "error: …",
|
|
"history": { iso: sessionId, … },
|
|
} }
|
|
|
|
The scheduler is a daemon thread that wakes every ~20s and fires each enabled
|
|
job at most once per matching minute. The minute claim is re-read from disk
|
|
right before firing, so the brief blue-green deploy window where two backends
|
|
share ``/data`` doesn't double-spawn a job.
|
|
|
|
The cron matcher is dependency-free: standard 5-field expressions with ``*``,
|
|
lists, ranges and ``/step`` (names and @shortcuts are not supported). Schedules
|
|
are evaluated in the container's local time (``TZ`` in docker-compose).
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
|
|
import yaml
|
|
|
|
# (low, high) bounds per field: minute, hour, day-of-month, month, day-of-week.
|
|
_BOUNDS = [(0, 59), (0, 23), (1, 31), (1, 12), (0, 6)]
|
|
|
|
_ID_RE = re.compile(r"[^a-z0-9-]+")
|
|
|
|
# Prompt files must live here (relative to the workspace root).
|
|
PROMPT_DIR = ".claude/agents"
|
|
|
|
_FRONTMATTER_RE = re.compile(r"\A---\s*\n(?P<body>.*?)\n---\s*\n", re.DOTALL)
|
|
|
|
# The run knobs a job's frontmatter can set — the same closed sets the sidecar
|
|
# validates (``sidecar.py``: ``_valid_harness`` / ``EFFORT_LEVELS``), checked
|
|
# here too so a typo shows up as an unapplied chip in the UI instead of a
|
|
# rejected spawn.
|
|
HARNESSES = ("claude", "pi")
|
|
EFFORT_LEVELS = ("low", "medium", "high", "xhigh", "max")
|
|
# Spellings of "run with no --effort flag at all", distinct from an absent key
|
|
# only in that the UI can show it was a deliberate choice.
|
|
_EFFORT_NONE = ("none", "off", "default", "")
|
|
_MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/:-]{0,127}$")
|
|
|
|
# Frontmatter written into a freshly created job's prompt-file stub. Every job
|
|
# should say out loud what it runs on; these are the lab's defaults.
|
|
DEFAULT_RUN_CONFIG = {"harness": "claude", "model": "sonnet", "effort": ""}
|
|
|
|
|
|
# ── cron expression parsing ──────────────────────────────────────────────────
|
|
def _parse_field(spec: str, lo: int, hi: int) -> set[int]:
|
|
vals: set[int] = set()
|
|
for part in spec.split(","):
|
|
part = part.strip()
|
|
step = 1
|
|
if "/" in part:
|
|
part, step_s = part.split("/", 1)
|
|
step = int(step_s)
|
|
if part in ("*", ""):
|
|
a, b = lo, hi
|
|
elif "-" in part:
|
|
a_s, b_s = part.split("-", 1)
|
|
a, b = int(a_s), int(b_s)
|
|
else:
|
|
a = b = int(part)
|
|
if not (lo <= a <= b <= hi) or step < 1:
|
|
raise ValueError(f"cron field out of range: {spec!r}")
|
|
vals.update(range(a, b + 1, step))
|
|
return vals
|
|
|
|
|
|
def parse_schedule(expr: str) -> list[set[int]]:
|
|
"""Parse a 5-field cron expression; raises ValueError on anything else."""
|
|
fields = (expr or "").split()
|
|
if len(fields) != 5:
|
|
raise ValueError("schedule must have 5 fields (min hour dom mon dow)")
|
|
return [_parse_field(f, lo, hi)
|
|
for f, (lo, hi) in zip(fields, _BOUNDS)]
|
|
|
|
|
|
def _day_matches(parsed: list[set[int]], dt: datetime) -> bool:
|
|
"""Standard cron semantics: dom and dow are OR'd when both are restricted,
|
|
AND'd (trivially) when either is ``*``."""
|
|
_, _, dom, mon, dow = parsed
|
|
if dt.month not in mon:
|
|
return False
|
|
cron_dow = (dt.weekday() + 1) % 7 # cron: 0 = Sunday
|
|
dom_any = dom == set(range(1, 32))
|
|
dow_any = dow == set(range(0, 7))
|
|
if dom_any and dow_any:
|
|
return True
|
|
if dom_any:
|
|
return cron_dow in dow
|
|
if dow_any:
|
|
return dt.day in dom
|
|
return dt.day in dom or cron_dow in dow
|
|
|
|
|
|
def due(parsed: list[set[int]], dt: datetime) -> bool:
|
|
"""Does this minute match the schedule?"""
|
|
return (dt.minute in parsed[0] and dt.hour in parsed[1]
|
|
and _day_matches(parsed, dt))
|
|
|
|
|
|
def next_run(expr: str, after: datetime | None = None) -> str | None:
|
|
"""The next fire time strictly after ``after`` (default: now), as an ISO
|
|
minute string **with the local UTC offset** (a viewer in any timezone
|
|
parses it to the right instant) — or None for an invalid/never-matching
|
|
schedule. Day-level skipping keeps this fast for any real schedule."""
|
|
try:
|
|
parsed = parse_schedule(expr)
|
|
except ValueError:
|
|
return None
|
|
base = (after or datetime.now()).replace(second=0, microsecond=0)
|
|
minutes, hours = sorted(parsed[0]), sorted(parsed[1])
|
|
day = base
|
|
for _ in range(366 * 2):
|
|
if not _day_matches(parsed, day):
|
|
day = (day + timedelta(days=1)).replace(hour=0, minute=0)
|
|
continue
|
|
for h in hours:
|
|
if h < day.hour:
|
|
continue
|
|
for m in minutes:
|
|
cand = day.replace(hour=h, minute=m)
|
|
if cand > base:
|
|
return cand.astimezone().isoformat(timespec="minutes")
|
|
day = (day + timedelta(days=1)).replace(hour=0, minute=0)
|
|
return None
|
|
|
|
|
|
# ── prompt files ─────────────────────────────────────────────────────────────
|
|
def valid_prompt_file(rel: str) -> str:
|
|
"""Normalize + validate a prompt-file path: a ``.md`` directly under
|
|
``.claude/agents/``. Raises ValueError otherwise."""
|
|
rel = (rel or "").strip().lstrip("/")
|
|
p = pathlib.PurePosixPath(rel)
|
|
if ".." in p.parts:
|
|
raise ValueError("promptFile must not contain '..'")
|
|
if p.suffix.lower() != ".md":
|
|
raise ValueError("promptFile must be a .md file")
|
|
if str(p.parent) != PROMPT_DIR:
|
|
raise ValueError(f"promptFile must live in {PROMPT_DIR}/")
|
|
return str(p)
|
|
|
|
|
|
def strip_frontmatter(text: str) -> str:
|
|
"""Drop a leading YAML frontmatter block (``.claude/agents`` files double
|
|
as Claude Code agent definitions; the frontmatter isn't prompt text)."""
|
|
return _FRONTMATTER_RE.sub("", text, count=1)
|
|
|
|
|
|
def parse_frontmatter(text: str) -> dict:
|
|
"""The leading YAML frontmatter block as a mapping — ``{}`` when there is
|
|
none, or when it isn't parseable/isn't a mapping. Never raises: a prompt
|
|
file with a broken header still runs, just on the defaults."""
|
|
m = _FRONTMATTER_RE.match(text or "")
|
|
if not m:
|
|
return {}
|
|
try:
|
|
data = yaml.safe_load(m.group("body"))
|
|
except yaml.YAMLError:
|
|
return {}
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def _norm_thinking(v) -> bool | None:
|
|
if isinstance(v, bool):
|
|
return v
|
|
if isinstance(v, str):
|
|
s = v.strip().lower()
|
|
if s in ("true", "on", "yes", "enabled"):
|
|
return True
|
|
if s in ("false", "off", "no", "disabled"):
|
|
return False
|
|
return None
|
|
|
|
|
|
def run_config(text: str) -> dict:
|
|
"""How a job's prompt file says its runs should be launched.
|
|
|
|
Reads ``harness`` / ``model`` / ``effort`` / ``thinking`` out of the
|
|
frontmatter and normalizes them to the shape ``_send_message`` takes.
|
|
``None`` means "not set" (the spawn path's own default applies); an
|
|
``effort`` of ``""`` means the file explicitly asked for *no* ``--effort``
|
|
flag. Invalid values are dropped rather than raised — a firing on the
|
|
defaults beats a job that never fires, and the resolved config is echoed
|
|
back to the UI so an unapplied key is visible.
|
|
|
|
``model`` and ``effort`` are also valid Claude Code agent-definition keys,
|
|
so a job's prompt file stays a working ``.claude/agents`` subagent."""
|
|
fm = parse_frontmatter(text)
|
|
out: dict = {"harness": None, "model": None, "effort": None,
|
|
"thinking": None}
|
|
|
|
h = fm.get("harness")
|
|
if isinstance(h, str) and h.strip().lower() in HARNESSES:
|
|
out["harness"] = h.strip().lower()
|
|
|
|
m = fm.get("model")
|
|
if isinstance(m, str) and _MODEL_RE.match(m.strip()):
|
|
out["model"] = m.strip()
|
|
|
|
if "effort" in fm:
|
|
e = fm["effort"]
|
|
e = "" if e is None else str(e).strip().lower()
|
|
if e in EFFORT_LEVELS:
|
|
out["effort"] = e
|
|
elif e in _EFFORT_NONE:
|
|
out["effort"] = ""
|
|
|
|
out["thinking"] = _norm_thinking(fm.get("thinking"))
|
|
return out
|
|
|
|
|
|
def frontmatter_block(cfg: dict) -> str:
|
|
"""A ``---``-delimited YAML header for ``cfg`` — what a new job's prompt
|
|
stub is seeded with."""
|
|
lines = ["---"]
|
|
for k, v in cfg.items():
|
|
if isinstance(v, bool):
|
|
v = "true" if v else "false"
|
|
lines.append(f"{k}: {v if v != '' else 'none'}")
|
|
lines.append("---")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def slugify(name: str) -> str:
|
|
slug = _ID_RE.sub("-", (name or "").lower()).strip("-")
|
|
return slug or uuid.uuid4().hex[:8]
|
|
|
|
|
|
# ── the store ────────────────────────────────────────────────────────────────
|
|
# Seeded on first run (no store file yet): the goal-keeper job — every 5 hours,
|
|
# read the repo's GOAL.md files and push one of them forward.
|
|
DEFAULT_JOBS = {
|
|
"goal-keeper": {
|
|
"id": "goal-keeper",
|
|
"name": "Goal keeper",
|
|
"schedule": "0 */5 * * *",
|
|
"promptFile": f"{PROMPT_DIR}/goal-keeper.md",
|
|
"enabled": True,
|
|
"model": None,
|
|
"createdAt": None, # stamped at seed time
|
|
"lastFired": None,
|
|
"lastStatus": None,
|
|
"history": {},
|
|
},
|
|
}
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now().astimezone().strftime("%Y-%m-%dT%H:%M:%S%z")
|
|
|
|
|
|
class CronStore:
|
|
"""The JSON store. Mirrors MetaStore's discipline: one lock, atomic saves,
|
|
mtime-based reload so out-of-band edits (or the other container during a
|
|
deploy cutover) are picked up."""
|
|
|
|
def __init__(self, path: str):
|
|
self.path = pathlib.Path(path)
|
|
self._lock = threading.Lock()
|
|
self._mtime: float | None = None
|
|
self._data: dict[str, dict] = {}
|
|
self._load()
|
|
if self._mtime is None: # no file yet → seed the defaults
|
|
with self._lock:
|
|
self._data = json.loads(json.dumps(DEFAULT_JOBS))
|
|
for job in self._data.values():
|
|
job["createdAt"] = _now_iso()
|
|
self._save()
|
|
|
|
def _load(self) -> None:
|
|
try:
|
|
self._data = json.loads(self.path.read_text(encoding="utf-8")) or {}
|
|
self._mtime = self.path.stat().st_mtime
|
|
except (OSError, ValueError):
|
|
self._data = {}
|
|
self._mtime = None
|
|
|
|
def _save(self) -> None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = self.path.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(self._data, indent=2, sort_keys=True),
|
|
encoding="utf-8")
|
|
os.replace(tmp, self.path)
|
|
try:
|
|
self._mtime = self.path.stat().st_mtime
|
|
except OSError:
|
|
pass
|
|
|
|
def _reload_if_changed(self) -> None:
|
|
try:
|
|
m = self.path.stat().st_mtime
|
|
except OSError:
|
|
return
|
|
if m != self._mtime:
|
|
self._load()
|
|
|
|
def list(self) -> list[dict]:
|
|
with self._lock:
|
|
self._reload_if_changed()
|
|
return json.loads(json.dumps(list(self._data.values())))
|
|
|
|
def get(self, job_id: str) -> dict | None:
|
|
with self._lock:
|
|
self._reload_if_changed()
|
|
job = self._data.get(job_id)
|
|
return json.loads(json.dumps(job)) if job else None
|
|
|
|
def create(self, name: str, schedule: str, prompt_file: str,
|
|
enabled: bool = True, model: str | None = None) -> dict:
|
|
parse_schedule(schedule) # raises on invalid
|
|
prompt_file = valid_prompt_file(prompt_file)
|
|
with self._lock:
|
|
self._reload_if_changed()
|
|
base = slugify(name)
|
|
job_id, n = base, 2
|
|
while job_id in self._data:
|
|
job_id, n = f"{base}-{n}", n + 1
|
|
job = {"id": job_id, "name": name or job_id, "schedule": schedule,
|
|
"promptFile": prompt_file, "enabled": bool(enabled),
|
|
"model": model, "createdAt": _now_iso(),
|
|
"lastFired": None, "lastStatus": None, "history": {}}
|
|
self._data[job_id] = job
|
|
self._save()
|
|
return json.loads(json.dumps(job))
|
|
|
|
def update(self, job_id: str, patch: dict) -> dict | None:
|
|
if "schedule" in patch and patch["schedule"] is not None:
|
|
parse_schedule(patch["schedule"])
|
|
if "promptFile" in patch and patch["promptFile"] is not None:
|
|
patch["promptFile"] = valid_prompt_file(patch["promptFile"])
|
|
with self._lock:
|
|
self._reload_if_changed()
|
|
job = self._data.get(job_id)
|
|
if not job:
|
|
return None
|
|
for k in ("name", "schedule", "promptFile", "enabled", "model"):
|
|
if k in patch and patch[k] is not None:
|
|
job[k] = patch[k]
|
|
if patch.get("model") == "": # explicit clear
|
|
job["model"] = None
|
|
self._save()
|
|
return json.loads(json.dumps(job))
|
|
|
|
def delete(self, job_id: str) -> bool:
|
|
with self._lock:
|
|
self._reload_if_changed()
|
|
if job_id not in self._data:
|
|
return False
|
|
del self._data[job_id]
|
|
self._save()
|
|
return True
|
|
|
|
def claim_minute(self, job_id: str, minute_key: str) -> bool:
|
|
"""Atomically claim this job's fire for ``minute_key`` (a local
|
|
``YYYY-MM-DDTHH:MM``). Re-reads the file first so a parallel backend's
|
|
claim (deploy cutover) is honored. False = already fired."""
|
|
with self._lock:
|
|
self._reload_if_changed()
|
|
job = self._data.get(job_id)
|
|
if not job or job.get("lastFired") == minute_key:
|
|
return False
|
|
job["lastFired"] = minute_key
|
|
self._save()
|
|
return True
|
|
|
|
def record_run(self, job_id: str, session_id: str,
|
|
status: str = "ok") -> None:
|
|
"""Append ``{now: sessionId}`` to the job's history (the JSON mapping
|
|
the viewer links back to conversations) and stamp the outcome."""
|
|
with self._lock:
|
|
self._reload_if_changed()
|
|
job = self._data.get(job_id)
|
|
if not job:
|
|
return
|
|
if session_id:
|
|
hist = dict(job.get("history") or {})
|
|
hist[_now_iso()] = session_id
|
|
# Keep the most recent 200 runs (dict order = insertion).
|
|
if len(hist) > 200:
|
|
for k in sorted(hist)[:len(hist) - 200]:
|
|
del hist[k]
|
|
job["history"] = hist
|
|
job["lastStatus"] = status
|
|
self._save()
|
|
|
|
|
|
# ── the scheduler ────────────────────────────────────────────────────────────
|
|
class CronScheduler(threading.Thread):
|
|
"""Fires enabled jobs on their schedule. ``fire(job)`` is injected by
|
|
main.py (it needs the sidecar spawn path + the meta store)."""
|
|
|
|
def __init__(self, store: CronStore, fire, interval: float = 20.0):
|
|
super().__init__(daemon=True, name="cron-scheduler")
|
|
self.store = store
|
|
self.fire = fire
|
|
self.interval = interval
|
|
|
|
def run(self) -> None:
|
|
while True:
|
|
time.sleep(self.interval)
|
|
try:
|
|
self._tick()
|
|
except Exception:
|
|
pass
|
|
|
|
def _tick(self) -> None:
|
|
now = datetime.now()
|
|
minute_key = now.strftime("%Y-%m-%dT%H:%M")
|
|
for job in self.store.list():
|
|
if not job.get("enabled"):
|
|
continue
|
|
try:
|
|
parsed = parse_schedule(job.get("schedule") or "")
|
|
except ValueError:
|
|
continue
|
|
if not due(parsed, now):
|
|
continue
|
|
if not self.store.claim_minute(job["id"], minute_key):
|
|
continue # already fired this minute (or a parallel backend won)
|
|
try:
|
|
self.fire(job)
|
|
except Exception as e: # never kill the loop on one bad job
|
|
self.store.record_run(job["id"], "", status=f"error: {e}")
|