diff --git a/.gitignore b/.gitignore index 33c4a94..4027d21 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,9 @@ state.json repos/ logs/ +# Worktrees +.worktrees/ + # Python .venv/ __pycache__/ @@ -21,3 +24,6 @@ docker-compose.override.yml # IDE .idea + +# SDD scratch +.superpowers/ diff --git a/CLAUDE.md b/CLAUDE.md index e341739..cd0696b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -69,15 +69,30 @@ src/clayde/ __init__.py work.py # run(issue_url) — unified: Claude decides next action # (ask, plan, implement, open PR, or address review) - webhook/ + service/ # job-execution core, shared by the Pebble webhook and + # the scheduler + __init__.py + queue.py # Job (origin: "pebble" | "scheduler"), JobQueue + # (in-memory asyncio.Queue), QueueFullError + runner.py # invoke_claude_job — async CLI subprocess (auto + # permission mode), extract_notification_payload + skills.py # Skill model, /skills/ discovery (SKILL.md + flat + # builtin), origin-aware system/user prompt builders + notify.py # send_ntfy + NotificationPayload model + worker.py # worker_loop, process_job — pop jobs, OTel process + # span; skips the success-notify for origin=scheduler + webhook/ # thin HTTP layer — Pebble ingress only __init__.py app.py # FastAPI app, /webhook/pebble, /health, OTel enqueue span auth.py # constant-time bearer-token verification - notify.py # send_ntfy + NotificationPayload model - queue.py # PebbleJob, JobQueue (in-memory asyncio.Queue), QueueFullError - runner.py # invoke_claude_pebble — async CLI subprocess, fresh session - skills.py # Skill model, /skills/ discovery, system + user prompt builders - worker.py # worker_loop, process_job — pop jobs, OTel process span + scheduler/ + __init__.py + tasks.py # ScheduledTask, parse_task_file(), discover_tasks() + # — cron/at frontmatter under CLAYDE_SCHEDULER_DIR + state.py # load_state()/save_state(), get_last_fired()/ + # set_last_fired() — /data/scheduler_state.json + loop.py # scheduler_loop()/run_tick() — due-ness, lateness + # annotation, enqueue, done/ move for fired one-offs skills_builtin/ ping.md # built-in health-check skill (baked into image) @@ -124,6 +139,11 @@ Plain `KEY=VALUE` file (no shell quoting). All keys use `CLAYDE_` prefix and are | `CLAYDE_DISK_ALERT_THRESHOLD_PCT` | Usage % that triggers an ntfy alert (default `85`) | | `CLAYDE_DISK_ALERT_PATH` | Path whose partition is checked — same volume as host root (default `/data`) | | `CLAYDE_DISK_ALERT_COOLDOWN_S` | Min seconds between repeat alerts while over threshold (default `21600`) | +| `CLAYDE_SCHEDULER_ENABLED` | Set to `true` to enable the scheduled-task loop (default `false`) | +| `CLAYDE_SCHEDULER_DIR` | In-container dir scanned for task markdown files (default `/tasks`) | +| `CLAYDE_SCHEDULER_INTERVAL_S` | Scheduler poll interval in seconds (default `30`) | +| `CLAYDE_SCHEDULER_TZ` | Default IANA timezone for task frontmatter without its own `tz` (default `Europe/Berlin`) | +| `CLAYDE_SCHEDULER_TIMEOUT` | Per-task CLI timeout in seconds (default `300`) | Config is loaded via `get_settings()` (singleton). `GH_TOKEN` is exported at startup for the `gh` CLI. diff --git a/README.md b/README.md index 2556226..bd7ffda 100644 --- a/README.md +++ b/README.md @@ -236,3 +236,97 @@ spawns a fresh Claude CLI session (no context carries between requests) with `cwd` set to the knowledge-base mount. Claude is free to use any number of skills per request; every terminal outcome (success, failure, timeout, usage limit, queue full, etc.) emits an ntfy notification. + +--- + +## Scheduler + +Clayde can also run tasks on a schedule — recurring (cron) or one-off (a +single future time) — instead of waiting for a Pebble request. Scheduled +runs feed the same job queue and worker as the Pebble webhook, so they use +the same Claude CLI backend and permission mode; only the prompt framing and +notification behaviour differ (below). + +To enable: + +1. Create the task directory on the host: `mkdir -p ~/clayde-tasks`. It's + mounted **read-write** at `/tasks` (already wired in + `docker-compose.yml`) — read-write because fired one-off tasks are moved + into a `done/` subdirectory, not deleted. +2. Set `CLAYDE_SCHEDULER_ENABLED=true` in `data/config.env`. Other + `CLAYDE_SCHEDULER_*` keys (poll interval `INTERVAL_S`, default timezone + `TZ`, in-container task dir `DIR`, per-task CLI timeout `TIMEOUT`) have + working defaults — see `config.env.template` if you need to change them. +3. Drop one markdown file per task into `~/clayde-tasks/`: + + ```markdown + --- + cron: "0 8 * * *" # recurring, 5-field cron + # at: 2026-09-21T08:00 # one-off, ISO-8601 local datetime (mutually exclusive with cron) + tz: Europe/Berlin # optional; default CLAYDE_SCHEDULER_TZ + enabled: true # optional; default true + title: keep-warm # optional; label for logs only + timeout: 4h # optional; default CLAYDE_SCHEDULER_TIMEOUT (300s) + --- + Run a trivial health check and confirm you are alive. + ``` + + Exactly one of `cron` or `at` is required — `cron` is a standard 5-field + expression; `at` is a local datetime interpreted in `tz`. Everything + after the frontmatter is the prompt sent to Claude. Malformed files + (missing/both schedule keys, bad cron, unterminated frontmatter) are + logged and skipped, not ntfy'd — that would spam every poll tick. + + `timeout` sets this task's own CLI timeout, overriding + `CLAYDE_SCHEDULER_TIMEOUT` for that one job — useful for a long overnight + deep-research run that needs more than the default budget. It accepts a + duration (`4h`, `90m`, `45s`) or a bare number of seconds, and is + hard-capped at 4 hours; a requested value above the cap is clamped and + logged, not rejected. A malformed `timeout` value makes the whole file + malformed, same as a bad `cron`. + + Long-running tasks should have their prompt instruct the agent to persist + progress periodically (e.g. write interim findings to the KB inbox as it + goes), not just at the end. A run that hits a usage limit or its timeout + is a single unattended attempt with no auto-resume, so whatever interim + state it wrote is all that survives. + +A fired one-off task is moved to `~/clayde-tasks/done/-.md` +rather than deleted, so it stays as a record of what ran and when. Recurring +tasks are never moved; their last-fired time is tracked in the container's +own `/data/scheduler_state.json`, keyed by filename. + +### Notifications + +Unlike Pebble requests, a scheduler job stays **silent on success** — the +framework emits no ntfy for a clean run. The framework still ntfy's on +**failure** (timeout, usage limit, CLI error, auth error, worker crash), +because a run that didn't finish can't self-report. If a task should notify +on success (e.g. "call the dentist" or a genuine reminder), say so in the +task's own prompt and let the agent send it itself via the `ntfy-ping` +skill — there is no per-task notify field. + +### Skill library mount + +`docker-compose.yml` mounts the whole personal skill library read-only at +`/skills/kb`, alongside the existing `/skills/personal` and `/skills/shared` +Pebble skill dirs, so a scheduled task (or a Pebble request) can use any +skill from the knowledge base, including `ntfy-ping`. + +### Permission mode + +Both scheduled and Pebble jobs run the Claude CLI with +`--permission-mode auto --permission-prompts none` — Claude proceeds without +interactive approval, since nobody is watching an unattended run to answer a +prompt. + +### Bootstrapping caveat + +The scheduler presumes the Claude CLI login (see [Option B: Claude Code +CLI](#option-b-claude-code-cli-cli) above) is already established. A recurring +task that runs the CLI regularly keeps that login's OAuth refresh lineage +alive once it's ticking — but the login has to be created once, by hand, +*before* the first tick, and must not be left to lapse in the meantime. A +scheduler enabled against a login that was never created, or that expired +before its first run, fails with an auth error on every tick (which does +ntfy, per the failure behaviour above). diff --git a/config.env.template b/config.env.template index 42f6209..c269208 100644 --- a/config.env.template +++ b/config.env.template @@ -44,3 +44,17 @@ CLAYDE_NTFY_TIMEOUT_S=10 # --- Knowledge base (default cwd for Pebble runs) --- # Mounted from host ~/knowledge_base/. Synced by Syncthing — no git in container. CLAYDE_KB_PATH=/home/clayde/knowledge_base + +# --- Scheduler --- +# Set to true to enable the scheduled-task loop (see README "Scheduler"). +CLAYDE_SCHEDULER_ENABLED=false +# In-container directory scanned for task markdown files (default /tasks; +# mounted from host ~/clayde-tasks in docker-compose.yml). +CLAYDE_SCHEDULER_DIR=/tasks +# How often the scheduler checks for due tasks, in seconds (default 30). +CLAYDE_SCHEDULER_INTERVAL_S=30 +# Default timezone for task frontmatter that omits its own tz (default Europe/Berlin). +CLAYDE_SCHEDULER_TZ=Europe/Berlin +# Default wall-clock budget (seconds) for a scheduled task. +# Override per task with `timeout:` frontmatter field; hard-capped at 4h (14400s). +CLAYDE_SCHEDULER_TIMEOUT=300 diff --git a/docker-compose.yml b/docker-compose.yml index 4aa10eb..1ac054e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -47,9 +47,16 @@ services: # under /skills/. Subdirectory layout is free; discovery is recursive. - ~/skills/personal:/skills/personal:ro - ~/skills/shared:/skills/shared:ro + # Whole personal skill library, read-only — lets scheduled tasks (and + # Pebble requests) use any skill from the knowledge base, e.g. ntfy-ping + # for an intentional success notification. + - ~/knowledge_base/skills:/skills/kb:ro # Pebble knowledge-base working directory — Syncthing on the host # handles cross-device sync; container performs no git on the KB. - ~/knowledge_base:/home/clayde/knowledge_base + # Scheduled-task markdown files (cron/at frontmatter). Read-write: the + # scheduler moves fired one-off files into a done/ subdirectory. + - ~/clayde-tasks:/tasks labels: - "traefik.enable=true" - "traefik.http.routers.clayde.rule=Host(`${CLAYDE_PEBBLE_HOST}`) && PathPrefix(`/webhook`)" diff --git a/docs/superpowers/plans/2026-09-21-scheduled-tasks.md b/docs/superpowers/plans/2026-09-21-scheduled-tasks.md new file mode 100644 index 0000000..c9bbc51 --- /dev/null +++ b/docs/superpowers/plans/2026-09-21-scheduled-tasks.md @@ -0,0 +1,1226 @@ +# Scheduled Tasks Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add a file-driven scheduler to Clayde that runs prompt tasks (one-off timestamp or recurring cron) from a host-mounted directory, independent of any interactive session, through the existing job pipeline. + +**Architecture:** Lift the shared job-execution core out of `webhook/` into a neutral `service/` package. Add a `scheduler/` package whose `scheduler_loop()` coroutine joins the existing `asyncio.gather` in the orchestrator, scans `/tasks/*.md`, and enqueues due tasks as `Job(origin="scheduler")` into the shared `JobQueue`. The existing worker runs them via the Claude CLI; notification is by origin (scheduler = silent on success, framework ntfy on failure). Also mounts the whole KB skill library and switches the runner to auto permission mode. + +**Tech Stack:** Python ≥3.12, `uv`, FastAPI/uvicorn, asyncio, pydantic-settings, `croniter`, stdlib `zoneinfo`, pytest + pytest-asyncio. + +**Spec:** `docs/superpowers/specs/2026-09-20-scheduled-tasks-design.md` + +## Global Constraints + +- Python ≥3.12, managed with `uv` (`~/.local/bin/uv`); run tests with `uv run pytest`. +- Commit style: Scoped Commits — `: `, scope = subsystem (e.g. `service:`, `scheduler:`). No change-type prefixes like `fix(...)`. +- No AI-authorship footer or `Co-Authored-By` trailer on commits (ClaydeCode repo rule). +- All settings use the `CLAYDE_` env prefix, loaded by pydantic-settings from `data/config.env`. +- The whole suite must pass (`uv run pytest`) at the end of every task before committing. +- New dependency floor: `croniter>=2.0`. +- Behaviour-preserving moves must not change existing test assertions except import paths and the `PebbleJob`→`Job` / `invoke_claude_pebble`→`invoke_claude_job` renames. + +--- + +### Task 1: Move execution core to `service/`, de-Pebble the names + +Behaviour-preserving refactor. Moves the shared execution modules out of `webhook/`, renames `PebbleJob`→`Job` and `invoke_claude_pebble`→`invoke_claude_job`, relocates their tests. No logic changes. + +**Files:** +- Move: `src/clayde/webhook/{queue,worker,runner,notify,skills}.py` → `src/clayde/service/` +- Create: `src/clayde/service/__init__.py` +- Modify: `src/clayde/webhook/__init__.py`, `src/clayde/orchestrator.py`, `src/clayde/webhook/app.py` +- Move tests: `tests/test_webhook_{queue,worker,runner,runner_parse,skills,notify}.py` → `tests/service/`; keep `tests/test_webhook_{app,auth}.py` → `tests/webhook/` +- Create: `tests/service/__init__.py`, `tests/webhook/__init__.py` + +**Interfaces:** +- Produces: `clayde.service.queue.Job(id: str, text: str, timestamp: int)` (frozen dataclass; `origin` added in Task 2), `JobQueue`, `QueueFullError`; `clayde.service.worker.worker_loop`, `process_job`; `clayde.service.runner.invoke_claude_job`, `extract_notification_payload`; `clayde.service.notify.send_ntfy`, `NotificationPayload`; `clayde.service.skills.discover_skills`, `build_system_prompt`, `build_user_prompt`, `SKILLS_ROOT`, `Skill`. +- `clayde.service.__init__` re-exports `Job`, `JobQueue`, `QueueFullError`, `worker_loop`. + +- [ ] **Step 1: Move the modules with git** + +```bash +cd $(git rev-parse --show-toplevel) +mkdir -p src/clayde/service +git mv src/clayde/webhook/queue.py src/clayde/service/queue.py +git mv src/clayde/webhook/worker.py src/clayde/service/worker.py +git mv src/clayde/webhook/runner.py src/clayde/service/runner.py +git mv src/clayde/webhook/notify.py src/clayde/service/notify.py +git mv src/clayde/webhook/skills.py src/clayde/service/skills.py +: > src/clayde/service/__init__.py +``` + +- [ ] **Step 2: Rename symbols and fix intra-package imports** + +Global rename across `src/` and `tests/`: +- `PebbleJob` → `Job` +- `invoke_claude_pebble` → `invoke_claude_job` +- import paths `clayde.webhook.queue` → `clayde.service.queue` (and `worker`, `runner`, `notify`, `skills`). + +```bash +grep -rl 'PebbleJob\|invoke_claude_pebble\|clayde\.webhook\.\(queue\|worker\|runner\|notify\|skills\)' src tests \ + | xargs sed -i \ + -e 's/PebbleJob/Job/g' \ + -e 's/invoke_claude_pebble/invoke_claude_job/g' \ + -e 's/clayde\.webhook\.queue/clayde.service.queue/g' \ + -e 's/clayde\.webhook\.worker/clayde.service.worker/g' \ + -e 's/clayde\.webhook\.runner/clayde.service.runner/g' \ + -e 's/clayde\.webhook\.notify/clayde.service.notify/g' \ + -e 's/clayde\.webhook\.skills/clayde.service.skills/g' +``` + +- [ ] **Step 3: Populate `service/__init__.py`** + +```python +from clayde.service.queue import Job, JobQueue, QueueFullError +from clayde.service.worker import worker_loop + +__all__ = ["Job", "JobQueue", "QueueFullError", "worker_loop"] +``` + +- [ ] **Step 4: Update `webhook/__init__.py` and orchestrator imports** + +`webhook/__init__.py` currently re-exports `JobQueue, create_app, worker_loop`. Make it: + +```python +from clayde.service import Job, JobQueue, QueueFullError, worker_loop +from clayde.webhook.app import create_app + +__all__ = ["Job", "JobQueue", "QueueFullError", "worker_loop", "create_app"] +``` + +`orchestrator.py` keeps `from clayde.webhook import JobQueue, create_app, worker_loop` — still valid via the re-export. Leave it. + +- [ ] **Step 5: Relocate tests into packages** + +```bash +mkdir -p tests/service tests/webhook +: > tests/service/__init__.py +: > tests/webhook/__init__.py +git mv tests/test_webhook_queue.py tests/service/test_queue.py +git mv tests/test_webhook_worker.py tests/service/test_worker.py +git mv tests/test_webhook_runner.py tests/service/test_runner.py +git mv tests/test_webhook_runner_parse.py tests/service/test_runner_parse.py +git mv tests/test_webhook_skills.py tests/service/test_skills.py +git mv tests/test_webhook_notify.py tests/service/test_notify.py +git mv tests/test_webhook_app.py tests/webhook/test_app.py +git mv tests/test_webhook_auth.py tests/webhook/test_auth.py +``` + +- [ ] **Step 6: Run the full suite** + +Run: `uv run pytest -q` +Expected: PASS, 364 tests, 0 failures (same count as baseline — only paths/names changed). + +- [ ] **Step 7: Commit** + +```bash +git add -A +git commit -m "service: move job-execution core out of webhook, rename PebbleJob to Job" +``` + +--- + +### Task 2: Add `Job.origin` and generalise the process span + +**Files:** +- Modify: `src/clayde/service/queue.py` (add field) +- Modify: `src/clayde/service/worker.py` (span name + attribute) +- Test: `tests/service/test_queue.py`, `tests/service/test_worker.py` + +**Interfaces:** +- Produces: `Job(id, text, timestamp, origin="pebble")` where `origin ∈ {"pebble","scheduler"}`. + +- [ ] **Step 1: Write failing test for the default and field** + +Add to `tests/service/test_queue.py`: + +```python +from clayde.service.queue import Job + +def test_job_origin_defaults_to_pebble(): + job = Job(id="1", text="hi", timestamp=0) + assert job.origin == "pebble" + +def test_job_origin_can_be_scheduler(): + job = Job(id="1", text="hi", timestamp=0, origin="scheduler") + assert job.origin == "scheduler" +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `uv run pytest tests/service/test_queue.py -q` +Expected: FAIL (`Job() got an unexpected keyword argument 'origin'`). + +- [ ] **Step 3: Add the field** + +In `src/clayde/service/queue.py`: + +```python +@dataclass(frozen=True) +class Job: + id: str + text: str + timestamp: int + origin: str = "pebble" +``` + +- [ ] **Step 4: Generalise the worker span** + +In `src/clayde/service/worker.py`, `process_job`, rename the span and add the attribute: + +```python +with tracer.start_as_current_span("clayde.job.process") as span: + span.set_attribute("job.origin", job.origin) + span.set_attribute("pebble.job_id", job.id) + # ... existing attributes unchanged +``` + +Update any assertion in `tests/service/test_worker.py` referencing `clayde.pebble.process` to `clayde.job.process`. + +- [ ] **Step 5: Run tests** + +Run: `uv run pytest tests/service -q` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add -A +git commit -m "service: add Job.origin and rename process span to clayde.job.process" +``` + +--- + +### Task 3: Switch the runner to auto permission mode + +**Files:** +- Modify: `src/clayde/service/runner.py` (`invoke_claude_job` argv) +- Test: `tests/service/test_runner.py` + +- [ ] **Step 1: Write failing test asserting the argv** + +The runner builds `cmd` before `create_subprocess_exec`. Add a test that inspects the command by monkeypatching `asyncio.create_subprocess_exec` to capture args. Add to `tests/service/test_runner.py`: + +```python +import asyncio +import clayde.service.runner as runner + +async def test_invoke_uses_auto_permission_mode(monkeypatch): + captured = {} + + class FakeProc: + returncode = 0 + async def communicate(self): + return (b'{"result": "ok", "is_error": false}', b"") + def kill(self): pass + async def wait(self): return 0 + + async def fake_exec(*args, **kwargs): + captured["args"] = args + return FakeProc() + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec) + await runner.invoke_claude_job( + system_prompt="s", user_text="u", cwd="/tmp", timeout_s=5, + ) + args = captured["args"] + assert "--permission-mode" in args + assert "auto" in args + assert "--permission-prompts" in args + assert "none" in args + assert "--dangerously-skip-permissions" not in args +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `uv run pytest tests/service/test_runner.py::test_invoke_uses_auto_permission_mode -q` +Expected: FAIL (`--dangerously-skip-permissions` still present). + +- [ ] **Step 3: Change the argv** + +In `src/clayde/service/runner.py`, replace the `--dangerously-skip-permissions` element: + +```python +cmd = [ + cli_bin, + "-p", user_text, + "--append-system-prompt", system_prompt, + "--output-format", "json", + "--permission-mode", "auto", + "--permission-prompts", "none", +] +``` + +- [ ] **Step 4: Run tests** + +Run: `uv run pytest tests/service -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -m "service: run the CLI under auto permission mode instead of skip-permissions" +``` + +--- + +### Task 4: Notification by origin (scheduler success is silent) + +**Files:** +- Modify: `src/clayde/service/worker.py` (`process_job` success branch) +- Test: `tests/service/test_worker.py` + +**Interfaces:** +- Consumes: `Job.origin` (Task 2). + +- [ ] **Step 1: Write failing tests** + +Add to `tests/service/test_worker.py` (follow the file's existing style for stubbing `invoke_claude_job` and `send_ntfy`; assert on whether `_notify`/`send_ntfy` was called): + +```python +async def test_scheduler_success_does_not_notify(monkeypatch): + calls = _stub_success_run(monkeypatch) # existing helper pattern; returns notify-call recorder + job = Job(id="1", text="t", timestamp=0, origin="scheduler") + await process_job(job, timeout_s=5, kb_path="/tmp") + assert calls.notify_count == 0 + +async def test_scheduler_failure_notifies(monkeypatch): + calls = _stub_timeout_run(monkeypatch) + job = Job(id="1", text="t", timestamp=0, origin="scheduler") + await process_job(job, timeout_s=5, kb_path="/tmp") + assert calls.notify_count == 1 + +async def test_pebble_success_still_notifies(monkeypatch): + calls = _stub_success_run(monkeypatch) + job = Job(id="1", text="t", timestamp=0, origin="pebble") + await process_job(job, timeout_s=5, kb_path="/tmp") + assert calls.notify_count == 1 +``` + +If the test file has no such helpers, write the two stubs inline using `monkeypatch.setattr` on `clayde.service.worker.invoke_claude_job` (return a JSON-tail string for success; raise `InvocationTimeoutError` for timeout) and on `clayde.service.worker.send_ntfy` (record calls). + +- [ ] **Step 2: Run to verify failure** + +Run: `uv run pytest tests/service/test_worker.py -k scheduler -q` +Expected: FAIL (scheduler success currently notifies). + +- [ ] **Step 3: Gate the success notify on origin** + +In `process_job`, the success path currently calls `await _notify(...)` after parsing the payload. Wrap only that success call: + +```python +if job.origin != "scheduler": + await _notify(title=payload.title, body=payload.body, success=payload.success) +log.info("[%s] processed outcome=%s", job.id, outcome) +``` + +Leave every failure/except branch's `_notify` untouched. + +- [ ] **Step 4: Run tests** + +Run: `uv run pytest tests/service/test_worker.py -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -m "service: suppress success notification for scheduler-origin jobs" +``` + +--- + +### Task 5: Origin-aware prompt framing + +**Files:** +- Modify: `src/clayde/service/skills.py` (`build_system_prompt`, `build_user_prompt`) +- Modify: `src/clayde/service/worker.py` (pass `job.origin` into the builders) +- Test: `tests/service/test_skills.py`, `tests/service/test_worker.py` + +**Interfaces:** +- Produces: `build_system_prompt(skills, timeout_s=300, origin="pebble")`, `build_user_prompt(text, timestamp, origin="pebble")`. + +- [ ] **Step 1: Write failing tests** + +Add to `tests/service/test_skills.py`: + +```python +from clayde.service.skills import build_system_prompt, build_user_prompt + +def test_system_prompt_scheduler_framing(): + p = build_system_prompt([], timeout_s=300, origin="scheduler") + assert "scheduled task" in p.lower() + assert "pebble watch" not in p.lower() + +def test_system_prompt_pebble_framing_unchanged(): + p = build_system_prompt([], timeout_s=300, origin="pebble") + assert "pebble watch" in p.lower() + +def test_user_prompt_scheduler_has_no_timestamp_prefix(): + assert build_user_prompt("do it", 123, origin="scheduler") == "do it" + +def test_user_prompt_pebble_unchanged(): + assert build_user_prompt("do it", 123, origin="pebble") == "(timestamp 123)\ndo it" +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `uv run pytest tests/service/test_skills.py -k framing -q` +Expected: FAIL (`origin` kwarg unknown). + +- [ ] **Step 3: Parametrise the builders** + +In `src/clayde/service/skills.py`, split the opening line by origin. Replace the hardcoded first line of `_SYSTEM_PROMPT_TEMPLATE` with a `{intro}` placeholder and choose it in `build_system_prompt`: + +```python +_INTRO = { + "pebble": "You are Clayde, executing a request from the user via a Pebble watch.", + "scheduler": "You are Clayde, executing a scheduled task.", +} + +def build_system_prompt(skills, timeout_s: int = 300, origin: str = "pebble") -> str: + ... + return _SYSTEM_PROMPT_TEMPLATE.format( + intro=_INTRO.get(origin, _INTRO["pebble"]), + skill_section=skill_section, timeout_s=timeout_s, + ) + +def build_user_prompt(text: str, timestamp: int, origin: str = "pebble") -> str: + if origin == "scheduler": + return text + return f"(timestamp {timestamp})\n{text}" +``` + +(Adjust `_SYSTEM_PROMPT_TEMPLATE` so it begins with `{intro}\n` in place of the current literal first sentence.) + +- [ ] **Step 4: Thread origin through the worker** + +In `src/clayde/service/worker.py`, `process_job`: + +```python +system_prompt = build_system_prompt(skills, timeout_s=timeout_s, origin=job.origin) +user_text = build_user_prompt(job.text, job.timestamp, origin=job.origin) +``` + +- [ ] **Step 5: Run tests** + +Run: `uv run pytest tests/service -q` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add -A +git commit -m "service: origin-aware system and user prompt framing" +``` + +--- + +### Task 6: `SKILL.md`-aware skill discovery + +**Files:** +- Modify: `src/clayde/service/skills.py` (`discover_skills`) +- Test: `tests/service/test_skills.py` + +**Interfaces:** +- Unchanged public signature `discover_skills(root=SKILLS_ROOT) -> list[Skill]`. + +- [ ] **Step 1: Write failing tests** + +Add to `tests/service/test_skills.py` (uses `tmp_path`): + +```python +from clayde.service.skills import discover_skills + +def _write(p, name, desc): + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(f"---\nname: {name}\ndescription: {desc}\n---\nbody\n") + +def test_directory_skill_matched(tmp_path): + _write(tmp_path / "kb" / "ntfy-ping" / "SKILL.md", "ntfy-ping", "send a push") + names = {s.name for s in discover_skills(tmp_path)} + assert "ntfy-ping" in names + +def test_reference_md_ignored_without_warning(tmp_path, caplog): + _write(tmp_path / "kb" / "foo" / "SKILL.md", "foo", "the foo skill") + (tmp_path / "kb" / "foo" / "references").mkdir(parents=True) + (tmp_path / "kb" / "foo" / "references" / "notes.md").write_text("# just notes\n") + names = {s.name for s in discover_skills(tmp_path)} + assert names == {"foo"} + assert "Failed to parse skill" not in caplog.text + +def test_flat_builtin_md_matched(tmp_path): + _write(tmp_path / "builtin" / "ping.md", "ping", "health check") + names = {s.name for s in discover_skills(tmp_path)} + assert "ping" in names +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `uv run pytest tests/service/test_skills.py -k "matched or ignored" -q` +Expected: FAIL (reference file logged/parsed; or builtin flat file handling differs). + +- [ ] **Step 3: Filter candidates before parsing** + +In `discover_skills`, replace the `all_files = sorted(root.rglob("*.md"))` line with a candidate filter: + +```python +def _is_skill_candidate(p: Path) -> bool: + # Directory skills use SKILL.md; the flat builtin format lives under builtin/. + return p.name == "SKILL.md" or p.parent.name == "builtin" + +all_files = sorted(p for p in root.rglob("*.md") if _is_skill_candidate(p)) +``` + +Keep the existing non-builtin-first ordering and name de-duplication. `_is_builtin` and the rest are unchanged. + +- [ ] **Step 4: Run tests** + +Run: `uv run pytest tests/service/test_skills.py -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -m "service: discover SKILL.md directory skills, ignore reference markdown" +``` + +--- + +### Task 7: Dependency + scheduler settings + +**Files:** +- Modify: `pyproject.toml` (add `croniter`) +- Modify: `src/clayde/config.py` (new settings) +- Test: `tests/test_config.py` + +**Interfaces:** +- Produces on `Settings`: `scheduler_enabled: bool`, `scheduler_dir: str`, `scheduler_interval_s: int`, `scheduler_tz: str`, `scheduler_timeout: int`. + +- [ ] **Step 1: Add croniter and sync** + +In `pyproject.toml`, add to `[project.dependencies]`: `"croniter>=2.0"`. Then: + +```bash +uv sync --extra dev +``` + +- [ ] **Step 2: Write failing test for defaults** + +Add to `tests/test_config.py`: + +```python +def test_scheduler_settings_defaults(monkeypatch): + from clayde.config import _reset_settings, get_settings + _reset_settings() + s = get_settings() + assert s.scheduler_enabled is False + assert s.scheduler_dir == "/tasks" + assert s.scheduler_interval_s == 30 + assert s.scheduler_tz == "Europe/Berlin" + assert s.scheduler_timeout == 300 +``` + +- [ ] **Step 3: Run to verify failure** + +Run: `uv run pytest tests/test_config.py::test_scheduler_settings_defaults -q` +Expected: FAIL (attributes missing). + +- [ ] **Step 4: Add the settings** + +In `src/clayde/config.py`, in `Settings`, after the Pebble block: + +```python + # Scheduler + scheduler_enabled: bool = False + scheduler_dir: str = "/tasks" + scheduler_interval_s: int = 30 + scheduler_tz: str = "Europe/Berlin" + scheduler_timeout: int = 300 +``` + +- [ ] **Step 5: Run tests** + +Run: `uv run pytest tests/test_config.py -q` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add -A +git commit -m "scheduler: add croniter dependency and scheduler settings" +``` + +--- + +### Task 8: Task file parsing (`scheduler/tasks.py`) + +**Files:** +- Create: `src/clayde/scheduler/__init__.py`, `src/clayde/scheduler/tasks.py` +- Test: `tests/scheduler/__init__.py`, `tests/scheduler/test_tasks.py` + +**Interfaces:** +- Produces: `ScheduledTask(path: Path, prompt: str, cron: str | None, at: datetime | None, tz: ZoneInfo, enabled: bool, title: str | None)`; `parse_task_file(path: Path, default_tz: str) -> ScheduledTask` (raises `ValueError` on malformed); `discover_tasks(root: Path, default_tz: str) -> list[ScheduledTask]` (skips `done/`, logs+skips malformed). + +- [ ] **Step 1: Write failing tests** + +Create `tests/scheduler/test_tasks.py`: + +```python +from datetime import datetime +from pathlib import Path +import pytest +from clayde.scheduler.tasks import parse_task_file, discover_tasks, ScheduledTask + +def _w(p: Path, fm: str, body: str = "do the thing"): + p.write_text(f"---\n{fm}\n---\n{body}\n") + +def test_parse_cron(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"') + t = parse_task_file(f, "Europe/Berlin") + assert t.cron == "0 8 * * *" and t.at is None and t.enabled is True + assert t.prompt.strip() == "do the thing" + +def test_parse_at(tmp_path): + f = tmp_path / "k.md"; _w(f, "at: 2026-09-21T08:00") + t = parse_task_file(f, "Europe/Berlin") + assert t.at == datetime(2026, 9, 21, 8, 0, tzinfo=t.tz) and t.cron is None + +def test_both_keys_rejected(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\nat: 2026-09-21T08:00') + with pytest.raises(ValueError): + parse_task_file(f, "Europe/Berlin") + +def test_neither_key_rejected(tmp_path): + f = tmp_path / "k.md"; _w(f, "title: x") + with pytest.raises(ValueError): + parse_task_file(f, "Europe/Berlin") + +def test_bad_cron_rejected(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "not a cron"') + with pytest.raises(ValueError): + parse_task_file(f, "Europe/Berlin") + +def test_bad_tz_rejected(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\ntz: Mars/Phobos') + with pytest.raises(ValueError): + parse_task_file(f, "Europe/Berlin") + +def test_enabled_false(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\nenabled: false') + assert parse_task_file(f, "Europe/Berlin").enabled is False + +def test_discover_skips_done_and_malformed(tmp_path, caplog): + _w(tmp_path / "good.md", 'cron: "0 8 * * *"') + (tmp_path / "bad.md").write_text("no frontmatter") + (tmp_path / "done").mkdir() + _w(tmp_path / "done" / "old.md", 'cron: "0 8 * * *"') + tasks = discover_tasks(tmp_path, "Europe/Berlin") + assert [t.path.name for t in tasks] == ["good.md"] +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `uv run pytest tests/scheduler/test_tasks.py -q` +Expected: FAIL (module missing). + +- [ ] **Step 3: Implement `tasks.py`** + +```python +"""Scheduled-task markdown files: model, parsing, discovery.""" +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +import yaml +from croniter import croniter + +log = logging.getLogger("clayde.scheduler") + + +@dataclass(frozen=True) +class ScheduledTask: + path: Path + prompt: str + cron: str | None + at: datetime | None + tz: ZoneInfo + enabled: bool + title: str | None + + +def _split_frontmatter(text: str) -> tuple[dict, str]: + if not text.startswith("---\n"): + raise ValueError("missing frontmatter") + end = text.find("\n---", 4) + if end == -1: + raise ValueError("unterminated frontmatter") + data = yaml.safe_load(text[4:end]) or {} + body = text[end + 4:].lstrip("\n") + if not isinstance(data, dict): + raise ValueError("frontmatter is not a mapping") + return data, body + + +def parse_task_file(path: Path, default_tz: str) -> ScheduledTask: + data, body = _split_frontmatter(path.read_text()) + + cron = data.get("cron") + at_raw = data.get("at") + if (cron is None) == (at_raw is None): + raise ValueError("exactly one of 'cron' or 'at' is required") + + tz_name = data.get("tz", default_tz) + try: + tz = ZoneInfo(str(tz_name)) + except (ZoneInfoNotFoundError, ValueError) as e: + raise ValueError(f"bad tz {tz_name!r}") from e + + if cron is not None: + cron = str(cron) + if not croniter.is_valid(cron): + raise ValueError(f"bad cron {cron!r}") + at = None + else: + cron = None + base = at_raw if isinstance(at_raw, datetime) else datetime.fromisoformat(str(at_raw)) + at = base.replace(tzinfo=tz) if base.tzinfo is None else base + + enabled = bool(data.get("enabled", True)) + title = data.get("title") + return ScheduledTask( + path=path, prompt=body, cron=cron, at=at, tz=tz, + enabled=enabled, title=str(title) if title is not None else None, + ) + + +def discover_tasks(root: Path, default_tz: str) -> list[ScheduledTask]: + if not root.exists(): + return [] + tasks: list[ScheduledTask] = [] + for p in sorted(root.glob("*.md")): + try: + tasks.append(parse_task_file(p, default_tz)) + except Exception as e: + log.warning("Skipping malformed task file %s: %s", p, e) + return tasks +``` + +Note: `root.glob("*.md")` is non-recursive, so `done/` is skipped automatically. + +- [ ] **Step 4: Run tests** + +Run: `uv run pytest tests/scheduler/test_tasks.py -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -m "scheduler: task-file model, parsing, and discovery" +``` + +--- + +### Task 9: Scheduler state (`scheduler/state.py`) + +**Files:** +- Create: `src/clayde/scheduler/state.py` +- Test: `tests/scheduler/test_state.py` + +**Interfaces:** +- Produces: `load_state(path: Path) -> dict`; `save_state(path: Path, state: dict) -> None`; `get_last_fired(state: dict, key: str) -> datetime | None`; `set_last_fired(state: dict, key: str, dt: datetime) -> None`. + +- [ ] **Step 1: Write failing tests** + +Create `tests/scheduler/test_state.py`: + +```python +from datetime import datetime, timezone +from clayde.scheduler.state import load_state, save_state, get_last_fired, set_last_fired + +def test_missing_file_is_empty(tmp_path): + assert load_state(tmp_path / "none.json") == {"recurring": {}} + +def test_roundtrip(tmp_path): + p = tmp_path / "s.json" + state = load_state(p) + dt = datetime(2026, 9, 20, 8, 0, tzinfo=timezone.utc) + set_last_fired(state, "keep-warm.md", dt) + save_state(p, state) + again = load_state(p) + assert get_last_fired(again, "keep-warm.md") == dt + +def test_get_missing_key_is_none(tmp_path): + assert get_last_fired(load_state(tmp_path / "s.json"), "x") is None +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `uv run pytest tests/scheduler/test_state.py -q` +Expected: FAIL (module missing). + +- [ ] **Step 3: Implement `state.py`** + +```python +"""Container-owned scheduler run-state (recurring dedup).""" +from __future__ import annotations + +import json +from datetime import datetime +from pathlib import Path + + +def load_state(path: Path) -> dict: + try: + data = json.loads(path.read_text()) + except (FileNotFoundError, json.JSONDecodeError): + data = {} + data.setdefault("recurring", {}) + return data + + +def save_state(path: Path, state: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(state, indent=2)) + tmp.replace(path) + + +def get_last_fired(state: dict, key: str) -> datetime | None: + entry = state.get("recurring", {}).get(key) + if not entry or "last_fired_at" not in entry: + return None + return datetime.fromisoformat(entry["last_fired_at"]) + + +def set_last_fired(state: dict, key: str, dt: datetime) -> None: + state.setdefault("recurring", {})[key] = {"last_fired_at": dt.isoformat()} +``` + +- [ ] **Step 4: Run tests** + +Run: `uv run pytest tests/scheduler/test_state.py -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -m "scheduler: recurring-task run-state persistence" +``` + +--- + +### Task 10: Due-ness and lateness (pure functions in `scheduler/loop.py`) + +**Files:** +- Create: `src/clayde/scheduler/loop.py` (pure helpers first; the async loop is Task 11) +- Test: `tests/scheduler/test_schedule.py` + +**Interfaces:** +- Produces: `baseline_recurring(cron: str, tz, now: datetime) -> datetime`; `recurring_due(cron: str, tz, now: datetime, last_fired: datetime) -> datetime | None`; `oneoff_due(at: datetime, now: datetime) -> bool`; `lateness_note(scheduled: datetime, now: datetime) -> str | None`. + +- [ ] **Step 1: Write failing tests** + +Create `tests/scheduler/test_schedule.py`: + +```python +from datetime import datetime, timedelta +from zoneinfo import ZoneInfo +from clayde.scheduler.loop import ( + baseline_recurring, recurring_due, oneoff_due, lateness_note, +) + +TZ = ZoneInfo("Europe/Berlin") + +def _at(y, m, d, hh, mm): + return datetime(y, m, d, hh, mm, tzinfo=TZ) + +def test_baseline_is_last_past_occurrence(tmp=None): + now = _at(2026, 9, 21, 15, 0) + assert baseline_recurring("0 8 * * *", TZ, now) == _at(2026, 9, 21, 8, 0) + +def test_recurring_fires_once_after_occurrence(): + now = _at(2026, 9, 21, 8, 0) + last = _at(2026, 9, 20, 8, 0) + assert recurring_due("0 8 * * *", TZ, now, last) == _at(2026, 9, 21, 8, 0) + +def test_recurring_not_due_when_already_fired(): + now = _at(2026, 9, 21, 8, 30) + last = _at(2026, 9, 21, 8, 0) + assert recurring_due("0 8 * * *", TZ, now, last) is None + +def test_recurring_single_fire_after_downtime(): + # down for two days; only the most recent occurrence fires, once + now = _at(2026, 9, 23, 9, 0) + last = _at(2026, 9, 20, 8, 0) + assert recurring_due("0 8 * * *", TZ, now, last) == _at(2026, 9, 23, 8, 0) + +def test_oneoff_due(): + assert oneoff_due(_at(2026, 9, 21, 8, 0), _at(2026, 9, 21, 8, 1)) is True + assert oneoff_due(_at(2026, 9, 21, 8, 0), _at(2026, 9, 21, 7, 59)) is False + +def test_lateness_note_present_when_late(): + note = lateness_note(_at(2026, 9, 21, 8, 0), _at(2026, 9, 21, 9, 0)) + assert note is not None and "late" in note.lower() + +def test_lateness_note_absent_when_on_time(): + assert lateness_note(_at(2026, 9, 21, 8, 0), _at(2026, 9, 21, 8, 0, 30)) is None +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `uv run pytest tests/scheduler/test_schedule.py -q` +Expected: FAIL (module/functions missing). + +- [ ] **Step 3: Implement the pure helpers** + +Create `src/clayde/scheduler/loop.py` with (async loop added in Task 11): + +```python +"""Scheduler tick loop and its pure scheduling helpers.""" +from __future__ import annotations + +import logging +from datetime import datetime, timedelta + +from croniter import croniter + +log = logging.getLogger("clayde.scheduler") + +_LATE_THRESHOLD = timedelta(seconds=60) + + +def baseline_recurring(cron: str, tz, now: datetime) -> datetime: + return croniter(cron, now.astimezone(tz)).get_prev(datetime) + + +def recurring_due(cron: str, tz, now: datetime, last_fired: datetime) -> datetime | None: + prev = croniter(cron, now.astimezone(tz)).get_prev(datetime) + return prev if prev > last_fired else None + + +def oneoff_due(at: datetime, now: datetime) -> bool: + return now >= at + + +def lateness_note(scheduled: datetime, now: datetime) -> str | None: + delta = now - scheduled + if delta <= _LATE_THRESHOLD: + return None + mins = int(delta.total_seconds() // 60) + when = scheduled.strftime("%Y-%m-%d %H:%M %Z") + dur = f"{mins} min" if mins else f"{int(delta.total_seconds())} s" + return f"[This task was scheduled for {when} and is running {dur} late.]" +``` + +- [ ] **Step 4: Run tests** + +Run: `uv run pytest tests/scheduler/test_schedule.py -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -m "scheduler: pure due-ness and lateness helpers" +``` + +--- + +### Task 11: Scheduler loop wiring (`scheduler/loop.py`) + +**Files:** +- Modify: `src/clayde/scheduler/loop.py` (add `scheduler_loop` + one-off move) +- Test: `tests/scheduler/test_loop.py` + +**Interfaces:** +- Consumes: `JobQueue`/`Job`/`QueueFullError` (service), `discover_tasks` (Task 8), state helpers (Task 9), due-ness helpers (Task 10). +- Produces: `async scheduler_loop(queue, *, tasks_dir: Path, state_path: Path, default_tz: str, interval_s: int) -> None`; `run_tick(queue, *, tasks_dir: Path, state_path: Path, default_tz: str, now: datetime) -> None` (single tick, the testable unit). + +- [ ] **Step 1: Write failing tests** + +Create `tests/scheduler/test_loop.py`: + +```python +from datetime import datetime, timedelta +from pathlib import Path +from zoneinfo import ZoneInfo +import pytest +from clayde.service.queue import JobQueue +from clayde.scheduler.loop import run_tick + +TZ = "Europe/Berlin" + +def _w(d: Path, name: str, fm: str, body="do it"): + (d / name).write_text(f"---\n{fm}\n---\n{body}\n") + +async def _drain(q: JobQueue): + out = [] + while not q._q.empty(): + out.append(await q.get()) + return out + +async def test_first_encounter_baselines_without_firing(tmp_path): + _w(tmp_path, "k.md", 'cron: "0 8 * * *"') + q = JobQueue(maxsize=10) + now = datetime(2026, 9, 21, 15, 0, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=tmp_path / "s.json", default_tz=TZ, now=now) + assert await _drain(q) == [] + assert (tmp_path / "s.json").exists() + +async def test_recurring_fires_and_dedups(tmp_path): + _w(tmp_path, "k.md", 'cron: "0 8 * * *"') + q = JobQueue(maxsize=10) + sp = tmp_path / "s.json" + # seed state so it's not first-encounter + from clayde.scheduler.state import load_state, save_state, set_last_fired + st = load_state(sp); set_last_fired(st, "k.md", datetime(2026, 9, 20, 8, 0, tzinfo=ZoneInfo(TZ))); save_state(sp, st) + now = datetime(2026, 9, 21, 8, 0, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=sp, default_tz=TZ, now=now) + jobs = await _drain(q) + assert len(jobs) == 1 and jobs[0].origin == "scheduler" + # second tick same minute: no duplicate + run_tick(q, tasks_dir=tmp_path, state_path=sp, default_tz=TZ, now=now) + assert await _drain(q) == [] + +async def test_oneoff_fires_and_moves_to_done(tmp_path): + _w(tmp_path, "call.md", "at: 2026-09-21T08:00") + q = JobQueue(maxsize=10) + now = datetime(2026, 9, 21, 8, 1, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=tmp_path / "s.json", default_tz=TZ, now=now) + jobs = await _drain(q) + assert len(jobs) == 1 + assert not (tmp_path / "call.md").exists() + assert list((tmp_path / "done").glob("*call.md")) + +async def test_late_oneoff_prepends_note(tmp_path): + _w(tmp_path, "call.md", "at: 2026-09-21T08:00", body="ring the bell") + q = JobQueue(maxsize=10) + now = datetime(2026, 9, 21, 9, 0, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=tmp_path / "s.json", default_tz=TZ, now=now) + jobs = await _drain(q) + assert "late" in jobs[0].text.lower() and "ring the bell" in jobs[0].text +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `uv run pytest tests/scheduler/test_loop.py -q` +Expected: FAIL (`run_tick` missing). + +- [ ] **Step 3: Implement `run_tick`, `scheduler_loop`, and the move helper** + +Append to `src/clayde/scheduler/loop.py`: + +```python +import asyncio +import shutil +import uuid +from pathlib import Path + +from clayde.service.queue import Job, JobQueue, QueueFullError +from clayde.scheduler.state import ( + load_state, save_state, get_last_fired, set_last_fired, +) +from clayde.scheduler.tasks import discover_tasks + + +def _move_to_done(path: Path, now: datetime) -> None: + done = path.parent / "done" + done.mkdir(exist_ok=True) + shutil.move(str(path), str(done / f"{int(now.timestamp())}-{path.name}")) + + +def run_tick(queue: JobQueue, *, tasks_dir: Path, state_path: Path, + default_tz: str, now: datetime) -> None: + state = load_state(state_path) + for task in discover_tasks(tasks_dir, default_tz): + if not task.enabled: + continue + key = task.path.name + scheduled: datetime | None = None + + if task.cron is not None: + last = get_last_fired(state, key) + if last is None: + set_last_fired(state, key, baseline_recurring(task.cron, task.tz, now)) + continue + scheduled = recurring_due(task.cron, task.tz, now, last) + elif oneoff_due(task.at, now): + scheduled = task.at + + if scheduled is None: + continue + + note = lateness_note(scheduled, now) + text = f"{note}\n{task.prompt}" if note else task.prompt + job = Job(id=str(uuid.uuid4()), text=text, + timestamp=int(now.timestamp()), origin="scheduler") + try: + queue.enqueue(job) + except QueueFullError: + log.warning("Queue full — deferring task %s", key) + continue + if task.cron is not None: + set_last_fired(state, key, scheduled) + else: + _move_to_done(task.path, now) + + save_state(state_path, state) + + +async def scheduler_loop(queue: JobQueue, *, tasks_dir: str, state_path: str, + default_tz: str, interval_s: int) -> None: + log.info("Scheduler loop started (dir=%s, interval=%ds)", tasks_dir, interval_s) + from datetime import timezone + while True: + try: + run_tick(queue, tasks_dir=Path(tasks_dir), state_path=Path(state_path), + default_tz=default_tz, now=datetime.now(timezone.utc)) + except Exception: + log.exception("Scheduler tick failed — continuing") + await asyncio.sleep(interval_s) +``` + +- [ ] **Step 4: Run tests** + +Run: `uv run pytest tests/scheduler -q` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -m "scheduler: tick loop — enqueue due tasks, dedup, move one-offs" +``` + +--- + +### Task 12: Wire the scheduler into the orchestrator + +**Files:** +- Modify: `src/clayde/orchestrator.py` +- Test: `tests/test_orchestrator.py` + +**Interfaces:** +- Consumes: `scheduler_loop` (Task 11), scheduler settings (Task 7). + +- [ ] **Step 1: Write failing test** + +The orchestrator builds a task list in `_run_with_pebble`. Extract the decision into a testable helper `_scheduler_enabled(settings) -> bool` (returns `settings.scheduler_enabled`) and assert wiring via the state path. Add to `tests/test_orchestrator.py`: + +```python +def test_scheduler_state_path_under_data(): + from clayde.orchestrator import _scheduler_state_path + assert _scheduler_state_path().endswith("/scheduler_state.json") +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `uv run pytest tests/test_orchestrator.py::test_scheduler_state_path_under_data -q` +Expected: FAIL (helper missing). + +- [ ] **Step 3: Add the helper and gather the loop** + +In `src/clayde/orchestrator.py`: + +```python +from clayde.config import DATA_DIR +from clayde.scheduler.loop import scheduler_loop + +def _scheduler_state_path() -> str: + return str(DATA_DIR / "scheduler_state.json") +``` + +In `_run_with_pebble`, after the `worker_task` definition and before `tasks = [...]`: + +```python + async def scheduler_task() -> None: + await scheduler_loop( + queue, + tasks_dir=settings.scheduler_dir, + state_path=_scheduler_state_path(), + default_tz=settings.scheduler_tz, + interval_s=settings.scheduler_interval_s, + ) + + tasks = [server.serve(), worker_task()] + if settings.scheduler_enabled: + log.info("Scheduler loop enabled") + tasks.append(scheduler_task()) + else: + log.info("Scheduler loop disabled (CLAYDE_SCHEDULER_ENABLED not set)") + if settings.fs_enabled: + ... # existing freeshard block unchanged +``` + +Note: scheduler jobs use `settings.scheduler_timeout`; the worker currently takes a single `timeout_s`. For v1 the shared worker uses `pebble_timeout` for all jobs. Deferring per-origin timeout keeps the worker unchanged; record it as a follow-up in the commit body. + +- [ ] **Step 4: Run tests** + +Run: `uv run pytest -q` +Expected: PASS (full suite). + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -m "orchestrator: run the scheduler loop alongside the webhook when enabled" +``` + +--- + +### Task 13: Deployment config and docs + +**Files:** +- Modify: `docker-compose.yml`, `config.env.template`, `README.md`, `CLAUDE.md` +- No tests (config/docs). + +- [ ] **Step 1: Add mounts to docker-compose** + +In the `clayde` service `volumes:`, add: + +```yaml + - ~/clayde-tasks:/tasks + - ~/knowledge_base/skills:/skills/kb:ro +``` + +- [ ] **Step 2: Document the new settings** + +In `config.env.template`, add commented entries for `CLAYDE_SCHEDULER_ENABLED`, `CLAYDE_SCHEDULER_DIR`, `CLAYDE_SCHEDULER_INTERVAL_S`, `CLAYDE_SCHEDULER_TZ`, `CLAYDE_SCHEDULER_TIMEOUT` with their defaults. + +- [ ] **Step 3: README + CLAUDE.md** + +Add a "Scheduler" section to `README.md`: the `~/clayde-tasks/` format (cron/at frontmatter, body = prompt), the `done/` behaviour, notification model (silent success, framework failure ntfy, prompt-driven notify via the `ntfy-ping` skill), the whole-library skill mount, the auto permission mode, and the **bootstrapping caveat** (the CLI login must be established once and not left to lapse before the first keep-warm tick). Add a short pointer in `CLAUDE.md` under Configuration and Project Structure (new `scheduler/` package, `service/` rename). + +- [ ] **Step 4: Sanity-check compose** + +Run: `docker compose -f docker-compose.yml config >/dev/null && echo OK` +Expected: `OK`. + +- [ ] **Step 5: Commit** + +```bash +git add -A +git commit -m "scheduler: deployment mounts, config template, and docs" +``` + +--- + +## Self-Review + +**Spec coverage:** +- service/ move + Job rename → Task 1; Job.origin + span → Task 2; auto permission mode → Task 3; notify-by-origin → Task 4; origin prompt framing → Task 5; SKILL.md discovery → Task 6; croniter + settings → Task 7; task parsing → Task 8; state → Task 9; due-ness/lateness → Task 10; loop (enqueue/dedup/move) → Task 11; orchestrator wiring → Task 12; compose mounts + docs + bootstrapping caveat → Task 13. All spec sections mapped. +- Deferred by spec (non-goals): crash-safe delivery, backfill, per-task silence-on-failure — none implemented, as intended. Per-origin timeout (`scheduler_timeout`) is defined in settings but the worker still uses one timeout in v1; noted in Task 12 as a follow-up rather than silently dropped. + +**Placeholder scan:** No TBD/TODO; every code step carries real code; test steps carry real assertions. + +**Type consistency:** `Job(id, text, timestamp, origin="pebble")` consistent across Tasks 1–12. `invoke_claude_job` consistent (Tasks 1, 3). `build_system_prompt(..., origin=)` / `build_user_prompt(..., origin=)` consistent (Tasks 5, and worker call). `ScheduledTask` fields consistent between Task 8 (producer) and Tasks 10–11 (consumers: `.cron`, `.at`, `.tz`, `.prompt`, `.enabled`, `.path`). State helpers `get_last_fired`/`set_last_fired` consistent between Task 9 and Task 11. Due-ness helper names consistent between Task 10 and Task 11. diff --git a/docs/superpowers/specs/2026-09-20-scheduled-tasks-design.md b/docs/superpowers/specs/2026-09-20-scheduled-tasks-design.md new file mode 100644 index 0000000..0741ad9 --- /dev/null +++ b/docs/superpowers/specs/2026-09-20-scheduled-tasks-design.md @@ -0,0 +1,314 @@ +# Scheduled Tasks — File-Driven Scheduler — Design + +**Date:** 2026-09-20 (revised 2026-09-21) +**Status:** Proposed — awaiting review + +**Changes in this revision:** dropped the per-task `notify` field in favour of +prompt-driven notification plus framework failure-notify; mount the whole KB +skill library and adjust discovery to the directory-based `SKILL.md` format; +switch the shared runner from `--dangerously-skip-permissions` to auto +permission mode. + +## Goal + +Give Clayde a scheduler that runs prompts on a schedule, independent of any +interactive session or the Pebble watch. Each task is one markdown file in a +host-mounted directory: frontmatter says *when* to run (a one-off timestamp or +a recurring cron expression), the body is the prompt. Due tasks are dispatched +through the existing job pipeline — a fresh Claude CLI session with `/skills/` +available, cwd = KB. + +Primary uses: reminders to self, recurring maintenance prompts, and the +credential keep-warm ping that motivated this work (a recurring task whose CLI +run refreshes the container's OAuth token so its login never lapses from +disuse). + +## Non-goals + +- **Crash-safe delivery.** The job queue is in-memory. A restart in the window + between enqueue and execution loses that one run (the Pebble path has the + same property). Recurring tasks self-heal on the next tick; a fired one-off + would sit in `done/` un-run. A worker→scheduler completion callback would fix + this and is deliberately out of scope for v1. +- **Backfilling missed occurrences.** After downtime an overdue task fires + once, never once-per-missed-tick. +- **Authoring tasks from other devices.** The task directory is a dedicated + host dir, not the synced knowledge base. Tasks are created on the VM. +- **Sub-minute schedules.** Cron granularity is one minute. +- **A separate reminders feature.** A reminder is just a task whose prompt asks + the agent to notify. +- **Per-task silence-on-failure.** Scheduler failures always notify (below). + There is no knob to silence a failing task. + +## Refactor: neutral `service/` package + +The job type and execution machinery are no longer Pebble-specific once the +scheduler feeds the same queue. Lift the shared core out of `webhook/` into a +new `src/clayde/service/` package (behaviour-preserving move): + +| New path | From | Contents | +|----------|------|----------| +| `service/queue.py` | `webhook/queue.py` | `Job` (renamed from `PebbleJob`), `JobQueue`, `QueueFullError` | +| `service/worker.py` | `webhook/worker.py` | `worker_loop`, `process_job` | +| `service/runner.py` | `webhook/runner.py` | `invoke_claude`, `extract_notification_payload` | +| `service/notify.py` | `webhook/notify.py` | `send_ntfy`, `NotificationPayload` | +| `service/skills.py` | `webhook/skills.py` | skill discovery + prompt builders | + +`webhook/` keeps only the HTTP producer: `app.py`, `auth.py`. A new +`scheduler/` package is the second producer. `orchestrator.py` and +`webhook/__init__.py` update their imports. Test files move to +`tests/service/` accordingly; `tests/webhook/` keeps the app/auth tests. + +Telemetry: the worker span `clayde.pebble.process` becomes `clayde.job.process` +with an `origin` attribute (`pebble` | `scheduler`). The enqueue span in +`webhook/app.py` stays `clayde.pebble.enqueue` — that endpoint really is Pebble. + +### `Job` model + +```python +@dataclass(frozen=True) +class Job: + id: str + text: str # the prompt + timestamp: int # epoch seconds at enqueue + origin: str = "pebble" # "pebble" | "scheduler" +``` + +`origin` selects prompt framing and the success-notification behaviour. It +defaults to Pebble, so `webhook/app.py` constructs `Job` unchanged. + +## Task file format + +A dedicated host dir `~/clayde-tasks/`, mounted **read-write** at `/tasks` +(read-write so fired one-offs can be moved to `done/`; the dir is owned by +`ubuntu`/1000, the uid the container runs as). One markdown file per task: + +```markdown +--- +cron: "0 8 * * *" # recurring, 5-field cron +# at: 2026-09-21T08:00 # one-off, ISO-8601 local datetime (mutually exclusive with cron) +tz: Europe/Berlin # optional; default CLAYDE_SCHEDULER_TZ +enabled: true # optional; default true +title: keep-warm # optional; label for logs only +--- +Run a trivial health check and confirm you are alive. +``` + +Rules: + +- Exactly one of `cron` / `at` is required. Two distinct keys (not one + overloaded `schedule:`) so a malformed cron can't be misread as a timestamp. +- `cron` is a standard 5-field expression, parsed by `croniter`. +- `at` is an ISO-8601 local datetime, interpreted in `tz`. +- `tz` is an IANA name resolved via stdlib `zoneinfo`; default from + `CLAYDE_SCHEDULER_TZ`. +- `enabled: false` parks a task without deleting it. +- Malformed files (missing/both schedule keys, bad cron, unterminated + frontmatter) are logged at WARNING and skipped — same policy as skill + discovery. No ntfy on a malformed file (would spam every tick). + +### Directory & state layout + +``` +~/clayde-tasks/ + keep-warm.md # recurring + call-dentist.md # one-off + done/ # fired one-offs moved here (timestamp-prefixed) +``` + +Run-state lives in the container's own volume at `/data/scheduler_state.json`, +never in the task dir. It tracks only recurring tasks: + +```json +{"recurring": {"keep-warm.md": {"last_fired_at": "2026-09-20T08:00:00+02:00"}}} +``` + +One-offs need no state entry — moving the file to `done/` is their dedup. +State is keyed by filename relative to the task dir. + +## Scheduler loop + +A new `scheduler_loop()` coroutine (in `scheduler/loop.py`) joins the existing +`asyncio.gather` in `orchestrator._run_with_pebble()`, gated by +`CLAYDE_SCHEDULER_ENABLED`. Every `CLAYDE_SCHEDULER_INTERVAL_S` (default 30) it: + +1. Discovers and parses `/tasks/*.md` (`scheduler/tasks.py`); skips `done/`. +2. Evaluates due-ness (below). +3. For each due task, builds a `Job(origin="scheduler", text=)` and + enqueues it into the shared `JobQueue`. If the queue is full, log and leave + dedup uncommitted so it retries next tick. +4. Commits dedup **at enqueue time**: recurring → write `last_fired_at`; + one-off → move file to `done/-`. + +### Due-ness, missed runs, lateness + +- **Recurring:** each tick, compute the most recent scheduled occurrence ≤ now + via `croniter` in the task's tz. Fire iff that occurrence is later than the + stored `last_fired_at`; then set `last_fired_at` to it. Fires each occurrence + exactly once, never backfills. + - **First encounter (no state):** baseline `last_fired_at` to the last past + occurrence *without* firing. A "daily 08:00" task created at 15:00 first + fires the next day, not immediately. +- **One-off:** fire when `now ≥ at` and the file is still in the active set. +- **Missed while down:** both conditions above stay true after downtime, so an + overdue task fires once on the first tick after startup. No backfill. +- **Lateness annotation:** when the fire time is more than 60 s past the + scheduled time, prepend to the prompt text: + `[This task was scheduled for and is running late.]` + So Claude can phrase a late reminder appropriately. Applies to both one-off + overdue firing and a late recurring tick. + +## Notification + +No per-task notify field. Behaviour is by origin: + +- **Scheduler job, success:** the framework emits **no** ntfy. Any intentional + notification is the task's own responsibility — its prompt asks the agent to + notify (e.g. "notify me: ..."), which the agent does via the mounted + `ntfy-ping` skill. keep-warm, whose prompt says nothing about notifying, is + therefore silent on success. +- **Scheduler job, failure** (timeout, usage limit, CLI error, auth error, + worker crash): the framework emits its ntfy, because a run that didn't finish + cannot self-report. keep-warm thus stays silent day-to-day but shouts when + the login lapses (an auth error) — the one signal it exists to surface. +- **Pebble job:** unchanged — framework ntfy on every outcome. + +Implementation: `process_job` skips the **success-branch** `_notify` when +`job.origin == "scheduler"`; every failure branch notifies as it does today. +No changes to `Job`, `NotificationPayload`, or `extract_notification_payload` +for notification purposes. + +Dependency: an intentional notification needs the `ntfy-ping` skill reachable +(satisfied by the whole-library mount below) and pointed at the right topic — +`ntfy-ping`'s topic must be reconciled with `CLAYDE_NTFY_TOPIC`, or the task +prompt must target the correct topic. Verify during implementation. + +## Skill exposure + +Mount the whole personal skill library so scheduled tasks and Pebble commands +can use it. Add to `docker-compose.yml`, `clayde` service: + +``` +- ~/knowledge_base/skills:/skills/kb:ro +``` + +**Discovery change** (`service/skills.py`): KB skills are directory-based +(`/SKILL.md` plus reference/example `.md` files); discovery currently +treats every `*.md` under `/skills` as a skill candidate. Change +`discover_skills` to consider only `SKILL.md` files and flat top-level `*.md` +(the builtin `ping.md` format), silently ignoring other `.md`. Without this, +discovery logs the ~47 reference files as malformed on every job tick. Name +de-duplication and builtin-override ordering are unchanged. + +**Safety:** each skill dir carries its own credentials (api-email, api-gcal, +api-azure, api-ionos, and others), so the whole tree becoming reachable is a +real capability grant. It is gated by the auto permission mode (below), not by +omission. This is an accepted, classifier-mediated risk, not a hard boundary. + +## Permission mode + +Replace `--dangerously-skip-permissions` in the shared runner +(`service/runner.py`, `invoke_claude`) with: + +``` +--permission-mode auto --permission-prompts none +``` + +Auto mode's classifier auto-approves safe actions and denies dangerous ones +(credential reads, destructive commands); `--permission-prompts none` means any +action the classifier would escalate to a human is auto-denied, since headless +runs have no one to ask. Verified against Claude CLI 2.1.278 +(`--permission-mode` choices include `auto`; `--permission-prompts` includes +`none`). + +Applies to **both** scheduler and Pebble jobs (shared runner) — a hardening of +the existing webhook, at the cost that some commands that ran under +skip-permissions may now be denied. It is a classifier, not a sandbox: it +lowers blast radius but does not hard-guarantee against a mutating action +disguised as benign. + +## Prompt framing + +`service/skills.py` prompt builders are parametrised by origin: + +- **System prompt** opening line: "executing a scheduled task" for + `origin="scheduler"`, unchanged "request from a Pebble watch" for `pebble`. + The skills catalogue, timeout budget, and JSON-tail requirement are shared. +- **User prompt:** scheduler builds `\n` rather than the + Pebble `(timestamp N)\n`. + +## Settings (new, `CLAYDE_` prefix) + +| Key | Default | Purpose | +|-----|---------|---------| +| `CLAYDE_SCHEDULER_ENABLED` | `false` | Activate the scheduler loop | +| `CLAYDE_SCHEDULER_DIR` | `/tasks` | In-container task directory | +| `CLAYDE_SCHEDULER_INTERVAL_S` | `30` | Tick interval | +| `CLAYDE_SCHEDULER_TZ` | `Europe/Berlin` | Default timezone for tasks | +| `CLAYDE_SCHEDULER_TIMEOUT` | `300` | Per-run wall-clock budget (mirrors `pebble_timeout`) | + +`docker-compose.yml`: add `- ~/clayde-tasks:/tasks` (read-write) and +`- ~/knowledge_base/skills:/skills/kb:ro` to the `clayde` service. + +## Dependency + +Add `croniter` to `[project.dependencies]` in `pyproject.toml`. Chosen over a +hand-rolled cron parser: standard, small, correct on edge cases (DST, day/dow +interplay). `zoneinfo` is stdlib. + +## Module & file layout + +``` +src/clayde/ + service/ # NEW — shared job execution (moved from webhook/) + __init__.py # re-exports Job, JobQueue, QueueFullError, worker_loop + queue.py # Job, JobQueue, QueueFullError + worker.py # worker_loop, process_job (scheduler success = no ntfy) + runner.py # invoke_claude (auto permission mode), extract_notification_payload + notify.py # send_ntfy, NotificationPayload + skills.py # discover_skills (SKILL.md-aware), build_system_prompt(origin,...), build_user_prompt + webhook/ + __init__.py + app.py # HTTP endpoint only (imports Job/JobQueue from service) + auth.py + scheduler/ # NEW + __init__.py + tasks.py # ScheduledTask model, parse_task_file, discover_tasks + state.py # load_state, save_state, dedup helpers + loop.py # scheduler_loop + +tests/ + service/ # moved webhook execution tests + scheduler/ # NEW — tasks, state, loop + webhook/ # app + auth tests +``` + +## Testing + +`uv run pytest`. New / changed coverage: + +- **tasks.py:** valid cron / valid at / both keys (reject) / neither (reject) / + bad cron / bad tz / `enabled: false` / body extraction. +- **state.py:** load/save round-trip, missing file, dedup key by relative path. +- **loop.py:** recurring first-encounter baseline (no fire); recurring fires + once per occurrence; recurring single-fire after simulated downtime; one-off + fires and moves to `done/`; one-off not re-fired; lateness annotation present + when late and absent when on time; queue-full leaves dedup uncommitted. +- **worker.py:** `origin="scheduler"` success emits **no** ntfy; each scheduler + failure branch (timeout / usage limit / CLI error / auth / worker crash) + **does** emit ntfy; `origin="pebble"` still notifies on every outcome. +- **skills.py discovery:** `SKILL.md` under a skill dir is matched; reference + `.md` files are ignored without a WARNING; flat builtin `ping.md` matched; + name collision de-dup and builtin-override ordering unchanged. +- **runner.py:** the CLI argv contains `--permission-mode auto` and + `--permission-prompts none`, and no longer contains + `--dangerously-skip-permissions`. +- **Regression:** moved Pebble tests still pass under `tests/service/`; + `origin="pebble"` framing and always-notify unchanged. + +## Bootstrapping caveat + +The scheduler presumes the CLI login is established. The keep-warm task keeps +the lineage alive once running, but the login must be created once and not left +to lapse before the first keep-warm tick. Document this in the README alongside +the scheduler setup. diff --git a/pyproject.toml b/pyproject.toml index b875649..a43ca77 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,6 +5,7 @@ description = "Clayde — autonomous GitHub issue agent" requires-python = ">=3.12" dependencies = [ "anthropic>=0.40", + "croniter>=2.0", "fastapi>=0.115", "jinja2>=3.1.6", "opentelemetry-api>=1.20", diff --git a/src/clayde/config.py b/src/clayde/config.py index 69dc51f..4f4e308 100644 --- a/src/clayde/config.py +++ b/src/clayde/config.py @@ -55,6 +55,13 @@ def effective_git_name(self) -> str: pebble_queue_max: int = 100 pebble_host: str = "" + # Scheduler + scheduler_enabled: bool = False + scheduler_dir: str = "/tasks" + scheduler_interval_s: int = 30 + scheduler_tz: str = "Europe/Berlin" + scheduler_timeout: int = 300 + # ntfy notifications (Pebble outcome feedback) ntfy_topic: str = "7yuau0vyes" ntfy_base_url: str = "https://ntfy.sh" diff --git a/src/clayde/disk.py b/src/clayde/disk.py index 8f0f6ce..88449a3 100644 --- a/src/clayde/disk.py +++ b/src/clayde/disk.py @@ -20,7 +20,7 @@ from pathlib import Path from clayde.config import DATA_DIR, Settings -from clayde.webhook.notify import send_ntfy +from clayde.service.notify import send_ntfy log = logging.getLogger("clayde.disk") diff --git a/src/clayde/orchestrator.py b/src/clayde/orchestrator.py index 460b163..c26f09e 100644 --- a/src/clayde/orchestrator.py +++ b/src/clayde/orchestrator.py @@ -6,8 +6,9 @@ import uvicorn -from clayde.config import get_settings, setup_logging +from clayde.config import DATA_DIR, get_settings, setup_logging from clayde.freeshard.loop import run_cycle +from clayde.scheduler.loop import scheduler_loop from clayde.webhook import JobQueue, create_app, worker_loop log = logging.getLogger("clayde.orchestrator") @@ -15,6 +16,10 @@ _shutdown = False +def _scheduler_state_path() -> str: + return str(DATA_DIR / "scheduler_state.json") + + def _handle_signal(signum, frame): global _shutdown _shutdown = True @@ -52,13 +57,24 @@ async def _run_with_pebble() -> None: server = uvicorn.Server(config) async def worker_task() -> None: - await worker_loop( + await worker_loop(queue, kb_path=settings.kb_path) + + async def scheduler_task() -> None: + await scheduler_loop( queue, - timeout_s=settings.pebble_timeout, - kb_path=settings.kb_path, + tasks_dir=settings.scheduler_dir, + state_path=_scheduler_state_path(), + default_tz=settings.scheduler_tz, + interval_s=settings.scheduler_interval_s, + default_timeout_s=settings.scheduler_timeout, ) tasks = [server.serve(), worker_task()] + if settings.scheduler_enabled: + log.info("Scheduler loop enabled") + tasks.append(scheduler_task()) + else: + log.info("Scheduler loop disabled (CLAYDE_SCHEDULER_ENABLED not set)") if settings.fs_enabled: log.info("Freeshard loop enabled") tasks.append(_freeshard_loop(settings)) diff --git a/src/clayde/scheduler/__init__.py b/src/clayde/scheduler/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/clayde/scheduler/loop.py b/src/clayde/scheduler/loop.py new file mode 100644 index 0000000..b137396 --- /dev/null +++ b/src/clayde/scheduler/loop.py @@ -0,0 +1,109 @@ +"""Scheduler tick loop and its pure scheduling helpers.""" +from __future__ import annotations + +import asyncio +import logging +import shutil +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from croniter import croniter + +from clayde.scheduler.state import ( + load_state, save_state, get_last_fired, set_last_fired, +) +from clayde.scheduler.tasks import discover_tasks +from clayde.service.queue import Job, JobQueue, QueueFullError + +log = logging.getLogger("clayde.scheduler") + +_LATE_THRESHOLD = timedelta(seconds=60) + + +def baseline_recurring(cron: str, tz, now: datetime) -> datetime: + start = now.astimezone(tz) + timedelta(seconds=1) + return croniter(cron, start).get_prev(datetime) + + +def recurring_due(cron: str, tz, now: datetime, last_fired: datetime) -> datetime | None: + start = now.astimezone(tz) + timedelta(seconds=1) + prev = croniter(cron, start).get_prev(datetime) + return prev if prev > last_fired else None + + +def oneoff_due(at: datetime, now: datetime) -> bool: + return now >= at + + +def lateness_note(scheduled: datetime, now: datetime) -> str | None: + delta = now - scheduled + if delta <= _LATE_THRESHOLD: + return None + mins = int(delta.total_seconds() // 60) + when = scheduled.strftime("%Y-%m-%d %H:%M %Z") + dur = f"{mins} min" if mins else f"{int(delta.total_seconds())} s" + return f"[This task was scheduled for {when} and is running {dur} late.]" + + +def _move_to_done(path: Path, now: datetime) -> None: + done = path.parent / "done" + done.mkdir(exist_ok=True) + shutil.move(str(path), str(done / f"{int(now.timestamp())}-{path.name}")) + + +def run_tick(queue: JobQueue, *, tasks_dir: Path, state_path: Path, + default_tz: str, now: datetime, default_timeout_s: int) -> None: + state = load_state(state_path) + for task in discover_tasks(tasks_dir, default_tz, default_timeout_s): + if not task.enabled: + continue + key = task.path.name + scheduled: datetime | None = None + + if task.cron is not None: + last = get_last_fired(state, key) + if last is None: + set_last_fired(state, key, baseline_recurring(task.cron, task.tz, now)) + continue + scheduled = recurring_due(task.cron, task.tz, now, last) + elif oneoff_due(task.at, now): + scheduled = task.at + + if scheduled is None: + continue + + try: + note = lateness_note(scheduled, now) + text = f"{note}\n{task.prompt}" if note else task.prompt + job = Job(id=str(uuid.uuid4()), text=text, + timestamp=int(now.timestamp()), origin="scheduler", + timeout_s=task.timeout_s) + try: + queue.enqueue(job) + except QueueFullError: + log.warning("Queue full — deferring task %s", key) + continue + if task.cron is not None: + set_last_fired(state, key, scheduled) + else: + _move_to_done(task.path, now) + except Exception: + log.exception("Scheduled task failed, skipping: %s", task.path) + continue + + save_state(state_path, state) + + +async def scheduler_loop(queue: JobQueue, *, tasks_dir: str, state_path: str, + default_tz: str, interval_s: int, + default_timeout_s: int) -> None: + log.info("Scheduler loop started (dir=%s, interval=%ds)", tasks_dir, interval_s) + while True: + try: + run_tick(queue, tasks_dir=Path(tasks_dir), state_path=Path(state_path), + default_tz=default_tz, now=datetime.now(timezone.utc), + default_timeout_s=default_timeout_s) + except Exception: + log.exception("Scheduler tick failed — continuing") + await asyncio.sleep(interval_s) diff --git a/src/clayde/scheduler/state.py b/src/clayde/scheduler/state.py new file mode 100644 index 0000000..06f43d3 --- /dev/null +++ b/src/clayde/scheduler/state.py @@ -0,0 +1,33 @@ +"""Container-owned scheduler run-state (recurring dedup).""" +from __future__ import annotations + +import json +from datetime import datetime +from pathlib import Path + + +def load_state(path: Path) -> dict: + try: + data = json.loads(path.read_text()) + except (FileNotFoundError, json.JSONDecodeError): + data = {} + data.setdefault("recurring", {}) + return data + + +def save_state(path: Path, state: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(state, indent=2)) + tmp.replace(path) + + +def get_last_fired(state: dict, key: str) -> datetime | None: + entry = state.get("recurring", {}).get(key) + if not entry or "last_fired_at" not in entry: + return None + return datetime.fromisoformat(entry["last_fired_at"]) + + +def set_last_fired(state: dict, key: str, dt: datetime) -> None: + state.setdefault("recurring", {})[key] = {"last_fired_at": dt.isoformat()} diff --git a/src/clayde/scheduler/tasks.py b/src/clayde/scheduler/tasks.py new file mode 100644 index 0000000..cfb805b --- /dev/null +++ b/src/clayde/scheduler/tasks.py @@ -0,0 +1,113 @@ +"""Scheduled-task markdown files: model, parsing, discovery.""" +from __future__ import annotations + +import logging +import re +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +import yaml +from croniter import croniter + +log = logging.getLogger("clayde.scheduler") + +MAX_TIMEOUT_S = 14400 # 4 hours + +_TIMEOUT_RE = re.compile(r"^\s*(\d+)\s*([hms]?)\s*$") +_TIMEOUT_UNIT_SECONDS = {"": 1, "s": 1, "m": 60, "h": 3600} + + +@dataclass(frozen=True) +class ScheduledTask: + path: Path + prompt: str + cron: str | None + at: datetime | None + title: str | None + timeout_s: int + tz: ZoneInfo + enabled: bool + + +def _split_frontmatter(text: str) -> tuple[dict, str]: + if not text.startswith("---\n"): + raise ValueError("missing frontmatter") + end = text.find("\n---", 4) + if end == -1: + raise ValueError("unterminated frontmatter") + data = yaml.safe_load(text[4:end]) or {} + body = text[end + 4:].lstrip("\n") + if not isinstance(data, dict): + raise ValueError("frontmatter is not a mapping") + return data, body + + +def _parse_timeout(value, default_s: int) -> int: + if value is None: + requested = default_s + elif isinstance(value, int): + requested = value + elif isinstance(value, str): + m = _TIMEOUT_RE.match(value) + if not m: + raise ValueError(f"bad timeout {value!r}") + amount, unit = m.groups() + requested = int(amount) * _TIMEOUT_UNIT_SECONDS[unit] + else: + raise ValueError(f"bad timeout {value!r}") + + if requested > MAX_TIMEOUT_S: + log.warning( + "Requested timeout %ds exceeds the %ds cap — clamping", + requested, MAX_TIMEOUT_S, + ) + requested = MAX_TIMEOUT_S + return max(1, requested) + + +def parse_task_file(path: Path, default_tz: str, default_timeout_s: int) -> ScheduledTask: + data, body = _split_frontmatter(path.read_text()) + + cron = data.get("cron") + at_raw = data.get("at") + if (cron is None) == (at_raw is None): + raise ValueError("exactly one of 'cron' or 'at' is required") + + tz_name = data.get("tz", default_tz) + try: + tz = ZoneInfo(str(tz_name)) + except (ZoneInfoNotFoundError, ValueError) as e: + raise ValueError(f"bad tz {tz_name!r}") from e + + if cron is not None: + cron = str(cron) + if not croniter.is_valid(cron): + raise ValueError(f"bad cron {cron!r}") + at = None + else: + cron = None + base = at_raw if isinstance(at_raw, datetime) else datetime.fromisoformat(str(at_raw)) + at = base.replace(tzinfo=tz) if base.tzinfo is None else base + + enabled = bool(data.get("enabled", True)) + title = data.get("title") + timeout_s = _parse_timeout(data.get("timeout"), default_timeout_s) + return ScheduledTask( + path=path, prompt=body, cron=cron, at=at, tz=tz, + enabled=enabled, title=str(title) if title is not None else None, + timeout_s=timeout_s, + ) + + +def discover_tasks(root: Path, default_tz: str, default_timeout_s: int) -> list[ScheduledTask]: + if not root.exists(): + return [] + tasks: list[ScheduledTask] = [] + for p in sorted(root.glob("*.md")): + try: + tasks.append(parse_task_file(p, default_tz, default_timeout_s)) + except Exception as e: + log.warning("Skipping malformed task file %s: %s", p, e) + return tasks diff --git a/src/clayde/service/__init__.py b/src/clayde/service/__init__.py new file mode 100644 index 0000000..abcb58d --- /dev/null +++ b/src/clayde/service/__init__.py @@ -0,0 +1,4 @@ +from clayde.service.queue import Job, JobQueue, QueueFullError +from clayde.service.worker import worker_loop + +__all__ = ["Job", "JobQueue", "QueueFullError", "worker_loop"] diff --git a/src/clayde/webhook/notify.py b/src/clayde/service/notify.py similarity index 98% rename from src/clayde/webhook/notify.py rename to src/clayde/service/notify.py index 2549bf5..485b6f4 100644 --- a/src/clayde/webhook/notify.py +++ b/src/clayde/service/notify.py @@ -14,7 +14,7 @@ from clayde.telemetry import get_tracer -log = logging.getLogger("clayde.webhook.notify") +log = logging.getLogger("clayde.service.notify") def _encode_header_value(text: str) -> str: diff --git a/src/clayde/webhook/queue.py b/src/clayde/service/queue.py similarity index 67% rename from src/clayde/webhook/queue.py rename to src/clayde/service/queue.py index a2e4834..43faf5b 100644 --- a/src/clayde/webhook/queue.py +++ b/src/clayde/service/queue.py @@ -11,24 +11,26 @@ class QueueFullError(Exception): @dataclass(frozen=True) -class PebbleJob: +class Job: id: str text: str timestamp: int + origin: str = "pebble" + timeout_s: int = 300 class JobQueue: - """Thin wrapper over ``asyncio.Queue[PebbleJob]`` with non-blocking enqueue.""" + """Thin wrapper over ``asyncio.Queue[Job]`` with non-blocking enqueue.""" def __init__(self, maxsize: int): - self._q: asyncio.Queue[PebbleJob] = asyncio.Queue(maxsize=maxsize) + self._q: asyncio.Queue[Job] = asyncio.Queue(maxsize=maxsize) - def enqueue(self, job: PebbleJob) -> None: + def enqueue(self, job: Job) -> None: """Non-blocking enqueue. Raises ``QueueFullError`` when full.""" try: self._q.put_nowait(job) except asyncio.QueueFull as e: raise QueueFullError() from e - async def get(self) -> PebbleJob: + async def get(self) -> Job: return await self._q.get() diff --git a/src/clayde/webhook/runner.py b/src/clayde/service/runner.py similarity index 94% rename from src/clayde/webhook/runner.py rename to src/clayde/service/runner.py index 53d6b1c..24d9529 100644 --- a/src/clayde/webhook/runner.py +++ b/src/clayde/service/runner.py @@ -16,14 +16,14 @@ _make_cli_env, _resolve_cli_bin, ) -from clayde.webhook.notify import NotificationPayload +from clayde.service.notify import NotificationPayload -log = logging.getLogger("clayde.webhook.worker") +log = logging.getLogger("clayde.service.worker") _JSON_BLOCK_RE = re.compile(r"```json\s*\n(.*?)(?:\n\s*)?```", re.DOTALL) -async def invoke_claude_pebble( +async def invoke_claude_job( *, system_prompt: str, user_text: str, cwd: str, timeout_s: int, ) -> str: """Run the Claude CLI for a single Pebble request and return its result text. @@ -39,7 +39,8 @@ async def invoke_claude_pebble( "-p", user_text, "--append-system-prompt", system_prompt, "--output-format", "json", - "--dangerously-skip-permissions", + "--permission-mode", "auto", + "--permission-prompts", "none", ] log.info("Invoking Claude CLI (cwd=%s, timeout=%ds)", cwd, timeout_s) proc = await asyncio.create_subprocess_exec( diff --git a/src/clayde/webhook/skills.py b/src/clayde/service/skills.py similarity index 80% rename from src/clayde/webhook/skills.py rename to src/clayde/service/skills.py index 4fe5ddb..8057623 100644 --- a/src/clayde/webhook/skills.py +++ b/src/clayde/service/skills.py @@ -41,8 +41,13 @@ def _parse_skill(path: Path) -> Skill: return Skill(name=name, description=desc, path=path) +_INTRO = { + "pebble": "You are Clayde, executing a request from the user via a Pebble watch.", + "scheduler": "You are Clayde, executing a scheduled task.", +} + _SYSTEM_PROMPT_TEMPLATE = """\ -You are Clayde, executing a request from the user via a Pebble watch. +{intro} You have a hard wall-clock budget of {timeout_s} seconds for this entire request. If your process exceeds it, it is killed and the user gets @@ -69,11 +74,14 @@ def _parse_skill(path: Path) -> Skill: """ -def build_system_prompt(skills: list[Skill], timeout_s: int = 300) -> str: +def build_system_prompt( + skills: list[Skill], timeout_s: int = 300, origin: str = "pebble" +) -> str: """Build the system prompt sent to the Claude CLI for a Pebble request. ``timeout_s`` is the hard wall-clock budget enforced by the runner; it is - surfaced in the prompt so Claude can scope work to fit. + surfaced in the prompt so Claude can scope work to fit. ``origin`` + selects the opening framing line ("pebble" or "scheduler"). """ if not skills: skill_section = "Available skills: (none currently registered)" @@ -87,12 +95,15 @@ def build_system_prompt(skills: list[Skill], timeout_s: int = 300) -> str: f"{files}" ) return _SYSTEM_PROMPT_TEMPLATE.format( + intro=_INTRO.get(origin, _INTRO["pebble"]), skill_section=skill_section, timeout_s=timeout_s, ) -def build_user_prompt(text: str, timestamp: int) -> str: +def build_user_prompt(text: str, timestamp: int, origin: str = "pebble") -> str: """Build the user prompt (passed to ``claude -p``) for a Pebble request.""" + if origin == "scheduler": + return text return f"(timestamp {timestamp})\n{text}" @@ -101,9 +112,18 @@ def _is_builtin(path: Path) -> bool: return "builtin" in {p.name for p in path.parents} +def _is_skill_candidate(p: Path) -> bool: + # Directory skills use SKILL.md; the flat builtin format lives under builtin/. + return p.name == "SKILL.md" or p.parent.name == "builtin" + + def discover_skills(root: Path = SKILLS_ROOT) -> list[Skill]: """Recursively discover all skills under ``root``. + Only ``SKILL.md`` files (directory-style skills) and flat ``.md`` files + directly under a ``builtin/`` subdirectory are considered; other markdown + files (e.g. a skill's reference/example docs) are ignored before parsing. + Returns a list ordered alphabetically by full path. Non-builtin skills (those NOT under a ``builtin/`` subdirectory) are processed before builtin skills so that user-mounted overrides take priority over @@ -113,7 +133,7 @@ def discover_skills(root: Path = SKILLS_ROOT) -> list[Skill]: """ if not root.exists(): return [] - all_files = sorted(root.rglob("*.md")) + all_files = sorted(p for p in root.rglob("*.md") if _is_skill_candidate(p)) # Non-builtin first so user skills override shipped builtins on name collision. files = [f for f in all_files if not _is_builtin(f)] files += [f for f in all_files if _is_builtin(f)] diff --git a/src/clayde/webhook/worker.py b/src/clayde/service/worker.py similarity index 77% rename from src/clayde/webhook/worker.py rename to src/clayde/service/worker.py index 0755c1b..df145c4 100644 --- a/src/clayde/webhook/worker.py +++ b/src/clayde/service/worker.py @@ -12,17 +12,17 @@ ) from clayde.config import get_settings from clayde.telemetry import get_tracer -from clayde.webhook.notify import send_ntfy -from clayde.webhook.queue import JobQueue, PebbleJob -from clayde.webhook.runner import extract_notification_payload, invoke_claude_pebble -from clayde.webhook.skills import ( +from clayde.service.notify import send_ntfy +from clayde.service.queue import JobQueue, Job +from clayde.service.runner import extract_notification_payload, invoke_claude_job +from clayde.service.skills import ( SKILLS_ROOT, build_system_prompt, build_user_prompt, discover_skills, ) -log = logging.getLogger("clayde.webhook.worker") +log = logging.getLogger("clayde.service.worker") _FALLBACK_TITLE = "Pebble: done (no summary)" @@ -45,10 +45,11 @@ async def _notify(*, title: str, body: str, success: bool) -> None: ) -async def process_job(job: PebbleJob, *, timeout_s: int, kb_path: str) -> None: +async def process_job(job: Job, *, kb_path: str) -> None: """Process a single Pebble job. Emits exactly one ntfy notification.""" tracer = get_tracer() - with tracer.start_as_current_span("clayde.pebble.process") as span: + with tracer.start_as_current_span("clayde.job.process") as span: + span.set_attribute("job.origin", job.origin) span.set_attribute("pebble.job_id", job.id) span.set_attribute("pebble.timestamp", job.timestamp) span.set_attribute("pebble.text", job.text) @@ -56,17 +57,17 @@ async def process_job(job: PebbleJob, *, timeout_s: int, kb_path: str) -> None: skills = discover_skills(SKILLS_ROOT) span.set_attribute("pebble.skills_available", len(skills)) - system_prompt = build_system_prompt(skills, timeout_s=timeout_s) - user_text = build_user_prompt(job.text, job.timestamp) + system_prompt = build_system_prompt(skills, timeout_s=job.timeout_s, origin=job.origin) + user_text = build_user_prompt(job.text, job.timestamp, origin=job.origin) t0 = time.monotonic() outcome = "worker_error" try: - output = await invoke_claude_pebble( + output = await invoke_claude_job( system_prompt=system_prompt, user_text=user_text, cwd=kb_path, - timeout_s=timeout_s, + timeout_s=job.timeout_s, ) payload = extract_notification_payload(output) if payload.title == _FALLBACK_TITLE: @@ -75,16 +76,17 @@ async def process_job(job: PebbleJob, *, timeout_s: int, kb_path: str) -> None: outcome = "success" else: outcome = "claude_fail" - await _notify( - title=payload.title, body=payload.body, success=payload.success, - ) + if job.origin != "scheduler": + await _notify( + title=payload.title, body=payload.body, success=payload.success, + ) log.info("[%s] processed outcome=%s", job.id, outcome) except InvocationTimeoutError: outcome = "timeout" log.warning("[%s] timeout", job.id) await _notify( title="Pebble: timeout", - body=f"ran {timeout_s}s+", + body=f"ran {job.timeout_s}s+", success=False, ) except UsageLimitError: @@ -127,15 +129,12 @@ async def process_job(job: PebbleJob, *, timeout_s: int, kb_path: str) -> None: span.set_attribute("pebble.success", outcome == "success") -async def worker_loop(queue: JobQueue, *, timeout_s: int, kb_path: str) -> None: +async def worker_loop(queue: JobQueue, *, kb_path: str) -> None: """Pop jobs from the queue and process them serially. Runs until cancelled.""" - log.info( - "Pebble worker loop started (timeout_s=%d, kb_path=%s)", - timeout_s, kb_path, - ) + log.info("Pebble worker loop started (kb_path=%s)", kb_path) while True: job = await queue.get() try: - await process_job(job, timeout_s=timeout_s, kb_path=kb_path) + await process_job(job, kb_path=kb_path) except Exception: log.exception("[%s] unhandled error in process_job", job.id) diff --git a/src/clayde/webhook/__init__.py b/src/clayde/webhook/__init__.py index f3ece45..55e20d4 100644 --- a/src/clayde/webhook/__init__.py +++ b/src/clayde/webhook/__init__.py @@ -1,18 +1,4 @@ -"""Pebble webhook + skill framework.""" +from clayde.service import Job, JobQueue, QueueFullError, worker_loop +from clayde.webhook.app import create_app -from clayde.webhook.app import PebblePayload, create_app -from clayde.webhook.notify import NotificationPayload, send_ntfy -from clayde.webhook.queue import JobQueue, PebbleJob, QueueFullError -from clayde.webhook.worker import process_job, worker_loop - -__all__ = [ - "JobQueue", - "NotificationPayload", - "PebbleJob", - "PebblePayload", - "QueueFullError", - "create_app", - "process_job", - "send_ntfy", - "worker_loop", -] +__all__ = ["Job", "JobQueue", "QueueFullError", "worker_loop", "create_app"] diff --git a/src/clayde/webhook/app.py b/src/clayde/webhook/app.py index e8165f4..2c8de2b 100644 --- a/src/clayde/webhook/app.py +++ b/src/clayde/webhook/app.py @@ -12,8 +12,8 @@ from clayde.config import get_settings from clayde.telemetry import get_tracer from clayde.webhook.auth import verify_bearer -from clayde.webhook.notify import send_ntfy -from clayde.webhook.queue import JobQueue, PebbleJob, QueueFullError +from clayde.service.notify import send_ntfy +from clayde.service.queue import JobQueue, Job, QueueFullError log = logging.getLogger("clayde.webhook") @@ -39,7 +39,10 @@ async def receive( verify_bearer(authorization, expected=expected_token) job_id = str(uuid.uuid4()) - job = PebbleJob(id=job_id, text=payload.text, timestamp=payload.timestamp) + job = Job( + id=job_id, text=payload.text, timestamp=payload.timestamp, + timeout_s=get_settings().pebble_timeout, + ) tracer = get_tracer() with tracer.start_as_current_span("clayde.pebble.enqueue") as span: diff --git a/tests/scheduler/__init__.py b/tests/scheduler/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/scheduler/test_loop.py b/tests/scheduler/test_loop.py new file mode 100644 index 0000000..e2a549f --- /dev/null +++ b/tests/scheduler/test_loop.py @@ -0,0 +1,111 @@ +from datetime import datetime, timedelta +from pathlib import Path +from zoneinfo import ZoneInfo +import pytest +from clayde.service.queue import JobQueue +from clayde.scheduler.loop import run_tick + +TZ = "Europe/Berlin" + +def _w(d: Path, name: str, fm: str, body="do it"): + (d / name).write_text(f"---\n{fm}\n---\n{body}\n") + +async def _drain(q: JobQueue): + out = [] + while not q._q.empty(): + out.append(await q.get()) + return out + +async def test_first_encounter_baselines_without_firing(tmp_path): + _w(tmp_path, "k.md", 'cron: "0 8 * * *"') + q = JobQueue(maxsize=10) + now = datetime(2026, 9, 21, 15, 0, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=tmp_path / "s.json", default_tz=TZ, now=now, default_timeout_s=300) + assert await _drain(q) == [] + assert (tmp_path / "s.json").exists() + +async def test_recurring_fires_and_dedups(tmp_path): + _w(tmp_path, "k.md", 'cron: "0 8 * * *"') + q = JobQueue(maxsize=10) + sp = tmp_path / "s.json" + # seed state so it's not first-encounter + from clayde.scheduler.state import load_state, save_state, set_last_fired + st = load_state(sp); set_last_fired(st, "k.md", datetime(2026, 9, 20, 8, 0, tzinfo=ZoneInfo(TZ))); save_state(sp, st) + now = datetime(2026, 9, 21, 8, 0, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=sp, default_tz=TZ, now=now, default_timeout_s=300) + jobs = await _drain(q) + assert len(jobs) == 1 and jobs[0].origin == "scheduler" + # second tick same minute: no duplicate + run_tick(q, tasks_dir=tmp_path, state_path=sp, default_tz=TZ, now=now, default_timeout_s=300) + assert await _drain(q) == [] + +async def test_oneoff_fires_and_moves_to_done(tmp_path): + _w(tmp_path, "call.md", "at: 2026-09-21T08:00") + q = JobQueue(maxsize=10) + now = datetime(2026, 9, 21, 8, 1, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=tmp_path / "s.json", default_tz=TZ, now=now, default_timeout_s=300) + jobs = await _drain(q) + assert len(jobs) == 1 + assert not (tmp_path / "call.md").exists() + assert list((tmp_path / "done").glob("*call.md")) + +async def test_late_oneoff_prepends_note(tmp_path): + _w(tmp_path, "call.md", "at: 2026-09-21T08:00", body="ring the bell") + q = JobQueue(maxsize=10) + now = datetime(2026, 9, 21, 9, 0, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=tmp_path / "s.json", default_tz=TZ, now=now, default_timeout_s=300) + jobs = await _drain(q) + assert "late" in jobs[0].text.lower() and "ring the bell" in jobs[0].text + +async def test_job_gets_task_timeout(tmp_path): + _w(tmp_path, "call.md", 'at: 2026-09-21T08:00\ntimeout: 2h') + q = JobQueue(maxsize=10) + now = datetime(2026, 9, 21, 8, 1, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=tmp_path / "s.json", default_tz=TZ, now=now, default_timeout_s=300) + jobs = await _drain(q) + assert jobs[0].timeout_s == 7200 + +async def test_job_gets_default_timeout_when_absent(tmp_path): + _w(tmp_path, "call.md", "at: 2026-09-21T08:00") + q = JobQueue(maxsize=10) + now = datetime(2026, 9, 21, 8, 1, tzinfo=ZoneInfo(TZ)) + run_tick(q, tasks_dir=tmp_path, state_path=tmp_path / "s.json", default_tz=TZ, now=now, default_timeout_s=300) + jobs = await _drain(q) + assert jobs[0].timeout_s == 300 + +async def test_failing_task_does_not_lose_earlier_dedup(tmp_path, monkeypatch): + # "a-cron.md" sorts before "b-oneoff.md" so the recurring task fires + # first in the same tick as the one-off task that then blows up. + _w(tmp_path, "a-cron.md", 'cron: "0 8 * * *"') + _w(tmp_path, "b-oneoff.md", "at: 2026-09-21T08:00") + q = JobQueue(maxsize=10) + sp = tmp_path / "s.json" + from clayde.scheduler.state import load_state, save_state, set_last_fired, get_last_fired + st = load_state(sp) + set_last_fired(st, "a-cron.md", datetime(2026, 9, 20, 8, 0, tzinfo=ZoneInfo(TZ))) + save_state(sp, st) + now = datetime(2026, 9, 21, 8, 0, tzinfo=ZoneInfo(TZ)) + + import clayde.scheduler.loop as loop_mod + def _boom(path, now): + raise OSError("permission denied") + monkeypatch.setattr(loop_mod, "_move_to_done", _boom) + + run_tick(q, tasks_dir=tmp_path, state_path=sp, default_tz=TZ, now=now, default_timeout_s=300) + + jobs = await _drain(q) + # both jobs got enqueued before the one-off's move-to-done blew up + assert len(jobs) == 2 + + reloaded = load_state(sp) + fired_at = get_last_fired(reloaded, "a-cron.md") + assert fired_at is not None and fired_at == datetime(2026, 9, 21, 8, 0, tzinfo=ZoneInfo(TZ)) + + # a second tick at the same `now`: the recurring task must not re-fire + # (its dedup was persisted); the still-broken one-off re-enqueues, which + # is the documented (not this fix's) consequence of the move continuing + # to fail. + run_tick(q, tasks_dir=tmp_path, state_path=sp, default_tz=TZ, now=now, default_timeout_s=300) + second_jobs = await _drain(q) + assert len(second_jobs) == 1 + assert "do it" in second_jobs[0].text diff --git a/tests/scheduler/test_schedule.py b/tests/scheduler/test_schedule.py new file mode 100644 index 0000000..218916a --- /dev/null +++ b/tests/scheduler/test_schedule.py @@ -0,0 +1,41 @@ +from datetime import datetime, timedelta +from zoneinfo import ZoneInfo +from clayde.scheduler.loop import ( + baseline_recurring, recurring_due, oneoff_due, lateness_note, +) + +TZ = ZoneInfo("Europe/Berlin") + +def _at(y, m, d, hh, mm, ss=0): + return datetime(y, m, d, hh, mm, ss, tzinfo=TZ) + +def test_baseline_is_last_past_occurrence(tmp=None): + now = _at(2026, 9, 21, 15, 0) + assert baseline_recurring("0 8 * * *", TZ, now) == _at(2026, 9, 21, 8, 0) + +def test_recurring_fires_once_after_occurrence(): + now = _at(2026, 9, 21, 8, 0) + last = _at(2026, 9, 20, 8, 0) + assert recurring_due("0 8 * * *", TZ, now, last) == _at(2026, 9, 21, 8, 0) + +def test_recurring_not_due_when_already_fired(): + now = _at(2026, 9, 21, 8, 30) + last = _at(2026, 9, 21, 8, 0) + assert recurring_due("0 8 * * *", TZ, now, last) is None + +def test_recurring_single_fire_after_downtime(): + # down for two days; only the most recent occurrence fires, once + now = _at(2026, 9, 23, 9, 0) + last = _at(2026, 9, 20, 8, 0) + assert recurring_due("0 8 * * *", TZ, now, last) == _at(2026, 9, 23, 8, 0) + +def test_oneoff_due(): + assert oneoff_due(_at(2026, 9, 21, 8, 0), _at(2026, 9, 21, 8, 1)) is True + assert oneoff_due(_at(2026, 9, 21, 8, 0), _at(2026, 9, 21, 7, 59)) is False + +def test_lateness_note_present_when_late(): + note = lateness_note(_at(2026, 9, 21, 8, 0), _at(2026, 9, 21, 9, 0)) + assert note is not None and "late" in note.lower() + +def test_lateness_note_absent_when_on_time(): + assert lateness_note(_at(2026, 9, 21, 8, 0), _at(2026, 9, 21, 8, 0, 30)) is None diff --git a/tests/scheduler/test_state.py b/tests/scheduler/test_state.py new file mode 100644 index 0000000..7f1e8b9 --- /dev/null +++ b/tests/scheduler/test_state.py @@ -0,0 +1,17 @@ +from datetime import datetime, timezone +from clayde.scheduler.state import load_state, save_state, get_last_fired, set_last_fired + +def test_missing_file_is_empty(tmp_path): + assert load_state(tmp_path / "none.json") == {"recurring": {}} + +def test_roundtrip(tmp_path): + p = tmp_path / "s.json" + state = load_state(p) + dt = datetime(2026, 9, 20, 8, 0, tzinfo=timezone.utc) + set_last_fired(state, "keep-warm.md", dt) + save_state(p, state) + again = load_state(p) + assert get_last_fired(again, "keep-warm.md") == dt + +def test_get_missing_key_is_none(tmp_path): + assert get_last_fired(load_state(tmp_path / "s.json"), "x") is None diff --git a/tests/scheduler/test_tasks.py b/tests/scheduler/test_tasks.py new file mode 100644 index 0000000..1402ae4 --- /dev/null +++ b/tests/scheduler/test_tasks.py @@ -0,0 +1,97 @@ +import logging +from datetime import datetime +from pathlib import Path +import pytest +from clayde.scheduler.tasks import ( + parse_task_file, discover_tasks, ScheduledTask, MAX_TIMEOUT_S, +) + +DEFAULT_TIMEOUT_S = 300 + +def _w(p: Path, fm: str, body: str = "do the thing"): + p.write_text(f"---\n{fm}\n---\n{body}\n") + +def test_parse_cron(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"') + t = parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert t.cron == "0 8 * * *" and t.at is None and t.enabled is True + assert t.prompt.strip() == "do the thing" + +def test_parse_at(tmp_path): + f = tmp_path / "k.md"; _w(f, "at: 2026-09-21T08:00") + t = parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert t.at == datetime(2026, 9, 21, 8, 0, tzinfo=t.tz) and t.cron is None + +def test_both_keys_rejected(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\nat: 2026-09-21T08:00') + with pytest.raises(ValueError): + parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + +def test_neither_key_rejected(tmp_path): + f = tmp_path / "k.md"; _w(f, "title: x") + with pytest.raises(ValueError): + parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + +def test_bad_cron_rejected(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "not a cron"') + with pytest.raises(ValueError): + parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + +def test_bad_tz_rejected(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\ntz: Mars/Phobos') + with pytest.raises(ValueError): + parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + +def test_enabled_false(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\nenabled: false') + assert parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S).enabled is False + +def test_discover_skips_done_and_malformed(tmp_path, caplog): + _w(tmp_path / "good.md", 'cron: "0 8 * * *"') + (tmp_path / "bad.md").write_text("no frontmatter") + (tmp_path / "done").mkdir() + _w(tmp_path / "done" / "old.md", 'cron: "0 8 * * *"') + tasks = discover_tasks(tmp_path, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert [t.path.name for t in tasks] == ["good.md"] + +def test_timeout_absent_uses_default(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"') + t = parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert t.timeout_s == DEFAULT_TIMEOUT_S + +def test_timeout_hours(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\ntimeout: 4h') + t = parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert t.timeout_s == 14400 + +def test_timeout_minutes(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\ntimeout: 90m') + t = parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert t.timeout_s == 5400 + +def test_timeout_seconds_suffix(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\ntimeout: 45s') + t = parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert t.timeout_s == 45 + +def test_timeout_bare_seconds(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\ntimeout: 600') + t = parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert t.timeout_s == 600 + +def test_timeout_over_cap_clamped_and_warns(tmp_path, caplog): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\ntimeout: 9h') + with caplog.at_level(logging.WARNING): + t = parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert t.timeout_s == MAX_TIMEOUT_S == 14400 + assert any("timeout" in r.message.lower() for r in caplog.records) + +def test_timeout_malformed_rejected(tmp_path): + f = tmp_path / "k.md"; _w(f, 'cron: "0 8 * * *"\ntimeout: soon') + with pytest.raises(ValueError): + parse_task_file(f, "Europe/Berlin", DEFAULT_TIMEOUT_S) + +def test_timeout_malformed_skipped_by_discover(tmp_path): + _w(tmp_path / "bad.md", 'cron: "0 8 * * *"\ntimeout: soon') + tasks = discover_tasks(tmp_path, "Europe/Berlin", DEFAULT_TIMEOUT_S) + assert tasks == [] diff --git a/tests/service/__init__.py b/tests/service/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_webhook_notify.py b/tests/service/test_notify.py similarity index 98% rename from tests/test_webhook_notify.py rename to tests/service/test_notify.py index 1228759..8dc3848 100644 --- a/tests/test_webhook_notify.py +++ b/tests/service/test_notify.py @@ -9,7 +9,7 @@ import pytest import respx -from clayde.webhook.notify import NotificationPayload, _encode_header_value, send_ntfy +from clayde.service.notify import NotificationPayload, _encode_header_value, send_ntfy def test_notification_payload_clamps_length(): diff --git a/tests/service/test_queue.py b/tests/service/test_queue.py new file mode 100644 index 0000000..4ee692d --- /dev/null +++ b/tests/service/test_queue.py @@ -0,0 +1,56 @@ +import asyncio + +import pytest + +from clayde.service.queue import JobQueue, Job, QueueFullError + + +@pytest.mark.asyncio +async def test_enqueue_and_dequeue(): + q = JobQueue(maxsize=2) + job = Job(id="abc", text="hi", timestamp=1) + q.enqueue(job) + got = await q.get() + assert got == job + + +@pytest.mark.asyncio +async def test_enqueue_raises_when_full(): + q = JobQueue(maxsize=1) + q.enqueue(Job(id="a", text="", timestamp=0)) + with pytest.raises(QueueFullError): + q.enqueue(Job(id="b", text="", timestamp=0)) + + +@pytest.mark.asyncio +async def test_get_blocks_until_enqueued(): + q = JobQueue(maxsize=2) + job = Job(id="abc", text="hi", timestamp=1) + + async def producer(): + await asyncio.sleep(0.01) + q.enqueue(job) + + asyncio.create_task(producer()) + got = await asyncio.wait_for(q.get(), timeout=1.0) + assert got == job + + +def test_job_origin_defaults_to_pebble(): + job = Job(id="1", text="hi", timestamp=0) + assert job.origin == "pebble" + + +def test_job_origin_can_be_scheduler(): + job = Job(id="1", text="hi", timestamp=0, origin="scheduler") + assert job.origin == "scheduler" + + +def test_job_timeout_s_defaults_to_300(): + job = Job(id="1", text="hi", timestamp=0) + assert job.timeout_s == 300 + + +def test_job_timeout_s_can_be_set(): + job = Job(id="1", text="hi", timestamp=0, timeout_s=7200) + assert job.timeout_s == 7200 diff --git a/tests/test_webhook_runner.py b/tests/service/test_runner.py similarity index 79% rename from tests/test_webhook_runner.py rename to tests/service/test_runner.py index 35ecb43..15f77e1 100644 --- a/tests/test_webhook_runner.py +++ b/tests/service/test_runner.py @@ -5,7 +5,7 @@ import pytest from clayde.claude import CliInvocationError, InvocationTimeoutError, UsageLimitError -from clayde.webhook import runner +from clayde.service import runner class _FakeProc: @@ -35,7 +35,7 @@ async def fake_create(*args, **kwargs): async def test_runner_returns_result_text(fake_subproc, tmp_path): fake_subproc["proc"] = _FakeProc(json.dumps({"result": "all good"}).encode()) - out = await runner.invoke_claude_pebble( + out = await runner.invoke_claude_job( system_prompt="sys", user_text="hi", cwd=str(tmp_path), timeout_s=10, ) assert out == "all good" @@ -60,7 +60,7 @@ async def slow_communicate(): fake_subproc["proc"] = proc with pytest.raises(InvocationTimeoutError): - await runner.invoke_claude_pebble( + await runner.invoke_claude_job( system_prompt="s", user_text="t", cwd=str(tmp_path), timeout_s=0, ) proc.kill.assert_called_once() @@ -71,14 +71,14 @@ async def test_runner_raises_usage_limit_on_stderr(fake_subproc, tmp_path): stdout=b"{}", stderr=b"hit your usage limit", returncode=1, ) with pytest.raises(UsageLimitError): - await runner.invoke_claude_pebble( + await runner.invoke_claude_job( system_prompt="s", user_text="t", cwd=str(tmp_path), timeout_s=10, ) async def test_runner_returns_no_match_unchanged(fake_subproc, tmp_path): fake_subproc["proc"] = _FakeProc(json.dumps({"result": "No matching skill"}).encode()) - out = await runner.invoke_claude_pebble( + out = await runner.invoke_claude_job( system_prompt="s", user_text="t", cwd=str(tmp_path), timeout_s=10, ) assert out == "No matching skill" @@ -93,7 +93,7 @@ async def test_runner_raises_usage_limit_on_is_error_output(fake_subproc, tmp_pa returncode=1, ) with pytest.raises(UsageLimitError): - await runner.invoke_claude_pebble( + await runner.invoke_claude_job( system_prompt="s", user_text="t", cwd=str(tmp_path), timeout_s=10, ) @@ -105,7 +105,7 @@ async def test_runner_raises_runtime_error_on_auth_failure(fake_subproc, tmp_pat returncode=1, ) with pytest.raises(RuntimeError, match="authentication failed"): - await runner.invoke_claude_pebble( + await runner.invoke_claude_job( system_prompt="s", user_text="t", cwd=str(tmp_path), timeout_s=10, ) @@ -125,7 +125,7 @@ async def hanging_communicate(): fake_subproc["proc"] = proc task = asyncio.create_task( - runner.invoke_claude_pebble( + runner.invoke_claude_job( system_prompt="s", user_text="t", cwd=str(tmp_path), timeout_s=60, ) ) @@ -141,7 +141,7 @@ async def test_runner_raises_cli_invocation_error_on_nonzero(fake_subproc, tmp_p stdout=b'{"result": "boom"}', stderr=b"boom on stderr", returncode=2, ) with pytest.raises(CliInvocationError) as exc: - await runner.invoke_claude_pebble( + await runner.invoke_claude_job( system_prompt="sys", user_text="hi", cwd=str(tmp_path), timeout_s=5, ) assert "boom" in exc.value.stderr @@ -151,7 +151,33 @@ async def test_runner_returns_text_on_zero_exit(fake_subproc, tmp_path): fake_subproc["proc"] = _FakeProc( stdout=json.dumps({"result": "ok"}).encode(), stderr=b"", returncode=0, ) - out = await runner.invoke_claude_pebble( + out = await runner.invoke_claude_job( system_prompt="sys", user_text="hi", cwd=str(tmp_path), timeout_s=5, ) assert out == "ok" + + +async def test_invoke_uses_auto_permission_mode(monkeypatch, tmp_path): + captured = {} + + class FakeProc: + returncode = 0 + async def communicate(self): + return (b'{"result": "ok", "is_error": false}', b"") + def kill(self): pass + async def wait(self): return 0 + + async def fake_exec(*args, **kwargs): + captured["args"] = args + return FakeProc() + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec) + await runner.invoke_claude_job( + system_prompt="s", user_text="u", cwd=str(tmp_path), timeout_s=5, + ) + args = captured["args"] + assert "--permission-mode" in args + assert "auto" in args + assert "--permission-prompts" in args + assert "none" in args + assert "--dangerously-skip-permissions" not in args diff --git a/tests/test_webhook_runner_parse.py b/tests/service/test_runner_parse.py similarity index 96% rename from tests/test_webhook_runner_parse.py rename to tests/service/test_runner_parse.py index 004c49c..b79e88b 100644 --- a/tests/test_webhook_runner_parse.py +++ b/tests/service/test_runner_parse.py @@ -2,7 +2,7 @@ from __future__ import annotations -from clayde.webhook.runner import extract_notification_payload +from clayde.service.runner import extract_notification_payload def test_extracts_last_json_block(): diff --git a/tests/test_webhook_skills.py b/tests/service/test_skills.py similarity index 70% rename from tests/test_webhook_skills.py rename to tests/service/test_skills.py index fcc1e5d..c8aa5b9 100644 --- a/tests/test_webhook_skills.py +++ b/tests/service/test_skills.py @@ -2,7 +2,7 @@ import pytest -from clayde.webhook.skills import Skill, _parse_skill, discover_skills +from clayde.service.skills import Skill, _parse_skill, discover_skills def _write(path: Path, content: str) -> Path: @@ -66,16 +66,16 @@ def _write_skill(path: Path, name: str, description: str) -> Path: def test_discover_recursive_alpha_order(tmp_path): - _write_skill(tmp_path / "personal" / "b.md", "b-skill", "B") - _write_skill(tmp_path / "personal" / "a.md", "a-skill", "A") - _write_skill(tmp_path / "shared" / "z.md", "z-skill", "Z") + _write_skill(tmp_path / "personal" / "b-skill" / "SKILL.md", "b-skill", "B") + _write_skill(tmp_path / "personal" / "a-skill" / "SKILL.md", "a-skill", "A") + _write_skill(tmp_path / "shared" / "z-skill" / "SKILL.md", "z-skill", "Z") skills = discover_skills(tmp_path) assert [s.name for s in skills] == ["a-skill", "b-skill", "z-skill"] def test_discover_dedup_first_wins(tmp_path, caplog): - a = _write_skill(tmp_path / "a" / "first.md", "dup", "first one") - _write_skill(tmp_path / "b" / "second.md", "dup", "second one") + a = _write_skill(tmp_path / "a" / "dup" / "SKILL.md", "dup", "first one") + _write_skill(tmp_path / "b" / "dup" / "SKILL.md", "dup", "second one") with caplog.at_level("WARNING", logger="clayde.webhook"): skills = discover_skills(tmp_path) assert len(skills) == 1 @@ -84,8 +84,9 @@ def test_discover_dedup_first_wins(tmp_path, caplog): def test_discover_skips_malformed(tmp_path, caplog): - _write_skill(tmp_path / "ok.md", "ok-skill", "fine") - (tmp_path / "broken.md").write_text("not a skill file\n") + _write_skill(tmp_path / "ok-skill" / "SKILL.md", "ok-skill", "fine") + (tmp_path / "broken" / "SKILL.md").parent.mkdir(parents=True) + (tmp_path / "broken" / "SKILL.md").write_text("not a skill file\n") with caplog.at_level("WARNING", logger="clayde.webhook"): skills = discover_skills(tmp_path) assert [s.name for s in skills] == ["ok-skill"] @@ -97,7 +98,28 @@ def test_discover_missing_root(tmp_path): assert discover_skills(missing) == [] -from clayde.webhook.skills import build_system_prompt, build_user_prompt +def test_directory_skill_matched(tmp_path): + _write_skill(tmp_path / "kb" / "ntfy-ping" / "SKILL.md", "ntfy-ping", "send a push") + names = {s.name for s in discover_skills(tmp_path)} + assert "ntfy-ping" in names + + +def test_reference_md_ignored_without_warning(tmp_path, caplog): + _write_skill(tmp_path / "kb" / "foo" / "SKILL.md", "foo", "the foo skill") + (tmp_path / "kb" / "foo" / "references").mkdir(parents=True) + (tmp_path / "kb" / "foo" / "references" / "notes.md").write_text("# just notes\n") + names = {s.name for s in discover_skills(tmp_path)} + assert names == {"foo"} + assert "Failed to parse skill" not in caplog.text + + +def test_flat_builtin_md_matched(tmp_path): + _write_skill(tmp_path / "builtin" / "ping.md", "ping", "health check") + names = {s.name for s in discover_skills(tmp_path)} + assert "ping" in names + + +from clayde.service.skills import build_system_prompt, build_user_prompt def test_build_system_prompt_with_skills(): @@ -133,7 +155,7 @@ def test_build_user_prompt(): def test_prompt_no_longer_caps_to_one_skill(): - from clayde.webhook.skills import Skill, build_system_prompt + from clayde.service.skills import Skill, build_system_prompt from pathlib import Path p = build_system_prompt([ Skill(name="add-note", description="Save a note", path=Path("/skills/personal/add-note.md")), @@ -145,7 +167,7 @@ def test_prompt_no_longer_caps_to_one_skill(): def test_prompt_contains_json_contract(): - from clayde.webhook.skills import build_system_prompt + from clayde.service.skills import build_system_prompt p = build_system_prompt([]) assert '```json' in p assert '"title"' in p @@ -154,20 +176,21 @@ def test_prompt_contains_json_contract(): def test_prompt_when_no_skills_still_invites_judgement(): - from clayde.webhook.skills import build_system_prompt + from clayde.service.skills import build_system_prompt p = build_system_prompt([]) assert "judgement" in p.lower() or "judgment" in p.lower() def test_discovers_builtin_alongside_host(tmp_path): - from clayde.webhook.skills import discover_skills + from clayde.service.skills import discover_skills # Simulate the in-container layout: /skills/builtin + /skills/personal. (tmp_path / "builtin").mkdir() (tmp_path / "personal").mkdir() (tmp_path / "builtin" / "ping.md").write_text( "---\nname: ping\ndescription: Health check.\n---\n\npong\n" ) - (tmp_path / "personal" / "add-note.md").write_text( + (tmp_path / "personal" / "add-note").mkdir() + (tmp_path / "personal" / "add-note" / "SKILL.md").write_text( "---\nname: add-note\ndescription: Save a note.\n---\n\n...\n" ) skills = discover_skills(tmp_path) @@ -177,13 +200,14 @@ def test_discovers_builtin_alongside_host(tmp_path): def test_discover_personal_overrides_builtin(tmp_path, caplog): """Non-builtin skills (personal/shared) win over builtin on name collision.""" - from clayde.webhook.skills import discover_skills + from clayde.service.skills import discover_skills (tmp_path / "builtin").mkdir() (tmp_path / "personal").mkdir() (tmp_path / "builtin" / "voice-command.md").write_text( "---\nname: voice-command\ndescription: Builtin version.\n---\n\nBuiltin body.\n" ) - (tmp_path / "personal" / "voice-command.md").write_text( + (tmp_path / "personal" / "voice-command").mkdir() + (tmp_path / "personal" / "voice-command" / "SKILL.md").write_text( "---\nname: voice-command\ndescription: Personal override.\n---\n\nCustom body.\n" ) with caplog.at_level("WARNING", logger="clayde.webhook"): @@ -195,7 +219,7 @@ def test_discover_personal_overrides_builtin(tmp_path, caplog): def test_voice_command_builtin_skill_exists(): """The shipped voice-command builtin skill has the expected frontmatter.""" - from clayde.webhook import skills as skills_mod + from clayde.service import skills as skills_mod import importlib.resources builtin_dir = Path(skills_mod.__file__).parent.parent / "skills_builtin" vc_path = builtin_dir / "voice-command.md" @@ -206,3 +230,22 @@ def test_voice_command_builtin_skill_exists(): body = vc_path.read_text() assert "speech-to-text" in body or "voice" in body.lower() assert "/home/clayde/knowledge_base" in body + + +def test_system_prompt_scheduler_framing(): + p = build_system_prompt([], timeout_s=300, origin="scheduler") + assert "scheduled task" in p.lower() + assert "pebble watch" not in p.lower() + + +def test_system_prompt_pebble_framing_unchanged(): + p = build_system_prompt([], timeout_s=300, origin="pebble") + assert "pebble watch" in p.lower() + + +def test_user_prompt_scheduler_has_no_timestamp_prefix(): + assert build_user_prompt("do it", 123, origin="scheduler") == "do it" + + +def test_user_prompt_pebble_unchanged(): + assert build_user_prompt("do it", 123, origin="pebble") == "(timestamp 123)\ndo it" diff --git a/tests/test_webhook_worker.py b/tests/service/test_worker.py similarity index 52% rename from tests/test_webhook_worker.py rename to tests/service/test_worker.py index 813be91..d0636ff 100644 --- a/tests/test_webhook_worker.py +++ b/tests/service/test_worker.py @@ -11,8 +11,8 @@ InvocationTimeoutError, UsageLimitError, ) -from clayde.webhook import worker -from clayde.webhook.queue import PebbleJob +from clayde.service import worker +from clayde.service.queue import Job @dataclass @@ -36,12 +36,16 @@ async def fake_send(*, title, body, success, **_): @pytest.fixture def fake_skills(monkeypatch): monkeypatch.setattr(worker, "discover_skills", lambda root=None: []) - monkeypatch.setattr(worker, "build_system_prompt", lambda skills, timeout_s=300: "SYS") - monkeypatch.setattr(worker, "build_user_prompt", lambda text, ts: f"USER:{text}") + monkeypatch.setattr( + worker, "build_system_prompt", lambda skills, timeout_s=300, origin="pebble": "SYS" + ) + monkeypatch.setattr( + worker, "build_user_prompt", lambda text, ts, origin="pebble": f"USER:{text}" + ) -def _job(): - return PebbleJob(id="job-1", text="hello", timestamp=1000) +def _job(timeout_s=10): + return Job(id="job-1", text="hello", timestamp=1000, timeout_s=timeout_s) @pytest.mark.asyncio @@ -49,8 +53,8 @@ async def test_success_path_emits_one_success_ntfy(monkeypatch, captured_ntfy, f async def fake_invoke(**kwargs): return '```json\n{"title": "saved", "body": "wrote inbox/x.md", "success": true}\n```' - monkeypatch.setattr(worker, "invoke_claude_pebble", fake_invoke) - await worker.process_job(_job(), timeout_s=10, kb_path="/tmp") + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + await worker.process_job(_job(), kb_path="/tmp") assert len(captured_ntfy) == 1 assert captured_ntfy[0].title == "saved" assert captured_ntfy[0].success is True @@ -61,8 +65,8 @@ async def test_claude_reports_failure_via_json(monkeypatch, captured_ntfy, fake_ async def fake_invoke(**kwargs): return '```json\n{"title": "could not", "body": "no calendar set up", "success": false}\n```' - monkeypatch.setattr(worker, "invoke_claude_pebble", fake_invoke) - await worker.process_job(_job(), timeout_s=10, kb_path="/tmp") + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + await worker.process_job(_job(), kb_path="/tmp") assert len(captured_ntfy) == 1 assert captured_ntfy[0].success is False assert captured_ntfy[0].title == "could not" @@ -73,8 +77,8 @@ async def test_parse_fallback_on_missing_json(monkeypatch, captured_ntfy, fake_s async def fake_invoke(**kwargs): return "I did things but forgot the JSON." - monkeypatch.setattr(worker, "invoke_claude_pebble", fake_invoke) - await worker.process_job(_job(), timeout_s=10, kb_path="/tmp") + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + await worker.process_job(_job(), kb_path="/tmp") assert len(captured_ntfy) == 1 assert captured_ntfy[0].title == "Pebble: done (no summary)" assert captured_ntfy[0].success is True @@ -85,20 +89,34 @@ async def test_timeout_emits_fail_ntfy(monkeypatch, captured_ntfy, fake_skills): async def fake_invoke(**kwargs): raise InvocationTimeoutError("ran 10s+") - monkeypatch.setattr(worker, "invoke_claude_pebble", fake_invoke) - await worker.process_job(_job(), timeout_s=10, kb_path="/tmp") + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + await worker.process_job(_job(), kb_path="/tmp") assert len(captured_ntfy) == 1 assert captured_ntfy[0].title == "Pebble: timeout" assert captured_ntfy[0].success is False +@pytest.mark.asyncio +async def test_timeout_ntfy_body_reflects_the_jobs_own_timeout( + monkeypatch, captured_ntfy, fake_skills +): + async def fake_invoke(**kwargs): + assert kwargs["timeout_s"] == 7200 + raise InvocationTimeoutError("ran 7200s+") + + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + await worker.process_job(_job(timeout_s=7200), kb_path="/tmp") + assert len(captured_ntfy) == 1 + assert captured_ntfy[0].body == "ran 7200s+" + + @pytest.mark.asyncio async def test_usage_limit_emits_rate_limited_ntfy(monkeypatch, captured_ntfy, fake_skills): async def fake_invoke(**kwargs): raise UsageLimitError("limit hit") - monkeypatch.setattr(worker, "invoke_claude_pebble", fake_invoke) - await worker.process_job(_job(), timeout_s=10, kb_path="/tmp") + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + await worker.process_job(_job(), kb_path="/tmp") assert len(captured_ntfy) == 1 assert captured_ntfy[0].title == "Pebble: rate-limited" assert captured_ntfy[0].success is False @@ -109,8 +127,8 @@ async def test_cli_invocation_error_emits_fail_ntfy(monkeypatch, captured_ntfy, async def fake_invoke(**kwargs): raise CliInvocationError("stderr tail here") - monkeypatch.setattr(worker, "invoke_claude_pebble", fake_invoke) - await worker.process_job(_job(), timeout_s=10, kb_path="/tmp") + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + await worker.process_job(_job(), kb_path="/tmp") assert len(captured_ntfy) == 1 assert captured_ntfy[0].title == "Pebble: failed" assert "stderr tail" in captured_ntfy[0].body @@ -122,8 +140,8 @@ async def test_auth_error_emits_auth_ntfy(monkeypatch, captured_ntfy, fake_skill async def fake_invoke(**kwargs): raise RuntimeError("Claude CLI authentication failed") - monkeypatch.setattr(worker, "invoke_claude_pebble", fake_invoke) - await worker.process_job(_job(), timeout_s=10, kb_path="/tmp") + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + await worker.process_job(_job(), kb_path="/tmp") assert len(captured_ntfy) == 1 assert captured_ntfy[0].title == "Pebble: auth error" assert captured_ntfy[0].success is False @@ -134,9 +152,44 @@ async def test_unexpected_exception_emits_fail_ntfy(monkeypatch, captured_ntfy, async def fake_invoke(**kwargs): raise ValueError("something weird") - monkeypatch.setattr(worker, "invoke_claude_pebble", fake_invoke) - await worker.process_job(_job(), timeout_s=10, kb_path="/tmp") + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + await worker.process_job(_job(), kb_path="/tmp") assert len(captured_ntfy) == 1 assert captured_ntfy[0].title == "Pebble: failed" assert "ValueError" in captured_ntfy[0].body assert captured_ntfy[0].success is False + + +@pytest.mark.asyncio +async def test_scheduler_success_does_not_notify(monkeypatch, captured_ntfy, fake_skills): + async def fake_invoke(**kwargs): + return '```json\n{"title": "saved", "body": "wrote inbox/x.md", "success": true}\n```' + + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + job = Job(id="job-1", text="hello", timestamp=1000, origin="scheduler", timeout_s=10) + await worker.process_job(job, kb_path="/tmp") + assert len(captured_ntfy) == 0 + + +@pytest.mark.asyncio +async def test_scheduler_failure_notifies(monkeypatch, captured_ntfy, fake_skills): + async def fake_invoke(**kwargs): + raise InvocationTimeoutError("ran 10s+") + + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + job = Job(id="job-1", text="hello", timestamp=1000, origin="scheduler", timeout_s=10) + await worker.process_job(job, kb_path="/tmp") + assert len(captured_ntfy) == 1 + assert captured_ntfy[0].title == "Pebble: timeout" + + +@pytest.mark.asyncio +async def test_pebble_success_still_notifies(monkeypatch, captured_ntfy, fake_skills): + async def fake_invoke(**kwargs): + return '```json\n{"title": "saved", "body": "wrote inbox/x.md", "success": true}\n```' + + monkeypatch.setattr(worker, "invoke_claude_job", fake_invoke) + job = Job(id="job-1", text="hello", timestamp=1000, origin="pebble", timeout_s=10) + await worker.process_job(job, kb_path="/tmp") + assert len(captured_ntfy) == 1 + assert captured_ntfy[0].title == "saved" diff --git a/tests/test_config.py b/tests/test_config.py index e8f5146..35d1d4e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -164,3 +164,14 @@ def test_fs_enabled_defaults_off(): from clayde.config import Settings s = Settings(_env_file=None) assert s.fs_enabled is False + + +def test_scheduler_settings_defaults(monkeypatch): + from clayde.config import _reset_settings, get_settings + _reset_settings() + s = get_settings() + assert s.scheduler_enabled is False + assert s.scheduler_dir == "/tasks" + assert s.scheduler_interval_s == 30 + assert s.scheduler_tz == "Europe/Berlin" + assert s.scheduler_timeout == 300 diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py index d6d4f3a..df81a76 100644 --- a/tests/test_orchestrator.py +++ b/tests/test_orchestrator.py @@ -6,6 +6,11 @@ import pytest +def test_scheduler_state_path_under_data(): + from clayde.orchestrator import _scheduler_state_path + assert _scheduler_state_path().endswith("/scheduler_state.json") + + def test_run_loop_with_pebble_invokes_async_entry(monkeypatch): """run_loop() must hand off to the async Pebble entry point.""" from clayde import orchestrator @@ -37,7 +42,7 @@ async def fake_to_thread(func, *args, **kwargs): async def fake_serve(): pass - async def fake_worker_loop(queue, *, timeout_s, kb_path): + async def fake_worker_loop(queue, *, kb_path): pass monkeypatch.setattr(orchestrator, "_shutdown", False) @@ -49,6 +54,7 @@ async def fake_worker_loop(queue, *, timeout_s, kb_path): mock_settings.pebble_timeout = 60 mock_settings.kb_path = "/kb" mock_settings.fs_loop_interval_s = 0 + mock_settings.scheduler_enabled = False with ( patch("clayde.orchestrator.setup_logging"), diff --git a/tests/test_pebble_e2e.py b/tests/test_pebble_e2e.py index e94af0b..c47c0e0 100644 --- a/tests/test_pebble_e2e.py +++ b/tests/test_pebble_e2e.py @@ -10,9 +10,9 @@ from httpx import ASGITransport, AsyncClient from clayde.config import _reset_settings -from clayde.webhook import worker as worker_mod +from clayde.service import worker as worker_mod from clayde.webhook.app import create_app -from clayde.webhook.queue import JobQueue +from clayde.service.queue import JobQueue @pytest.mark.asyncio @@ -38,16 +38,18 @@ async def fake_invoke(**kwargs): "```\n" ) - monkeypatch.setattr(worker_mod, "invoke_claude_pebble", fake_invoke) + monkeypatch.setattr(worker_mod, "invoke_claude_job", fake_invoke) monkeypatch.setattr(worker_mod, "discover_skills", lambda root=None: []) - monkeypatch.setattr(worker_mod, "build_system_prompt", lambda skills, timeout_s=300: "SYS") - monkeypatch.setattr(worker_mod, "build_user_prompt", lambda text, ts: text) + monkeypatch.setattr( + worker_mod, "build_system_prompt", lambda skills, timeout_s=300, origin="pebble": "SYS" + ) + monkeypatch.setattr(worker_mod, "build_user_prompt", lambda text, ts, origin="pebble": text) # Real queue + real worker_loop. q = JobQueue(maxsize=4) app = create_app(queue=q, expected_token="tok") worker_task = asyncio.create_task( - worker_mod.worker_loop(q, timeout_s=10, kb_path=str(tmp_path)) + worker_mod.worker_loop(q, kb_path=str(tmp_path)) ) try: diff --git a/tests/test_webhook_queue.py b/tests/test_webhook_queue.py deleted file mode 100644 index ab90506..0000000 --- a/tests/test_webhook_queue.py +++ /dev/null @@ -1,36 +0,0 @@ -import asyncio - -import pytest - -from clayde.webhook.queue import JobQueue, PebbleJob, QueueFullError - - -@pytest.mark.asyncio -async def test_enqueue_and_dequeue(): - q = JobQueue(maxsize=2) - job = PebbleJob(id="abc", text="hi", timestamp=1) - q.enqueue(job) - got = await q.get() - assert got == job - - -@pytest.mark.asyncio -async def test_enqueue_raises_when_full(): - q = JobQueue(maxsize=1) - q.enqueue(PebbleJob(id="a", text="", timestamp=0)) - with pytest.raises(QueueFullError): - q.enqueue(PebbleJob(id="b", text="", timestamp=0)) - - -@pytest.mark.asyncio -async def test_get_blocks_until_enqueued(): - q = JobQueue(maxsize=2) - job = PebbleJob(id="abc", text="hi", timestamp=1) - - async def producer(): - await asyncio.sleep(0.01) - q.enqueue(job) - - asyncio.create_task(producer()) - got = await asyncio.wait_for(q.get(), timeout=1.0) - assert got == job diff --git a/tests/webhook/__init__.py b/tests/webhook/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_webhook_app.py b/tests/webhook/test_app.py similarity index 87% rename from tests/test_webhook_app.py rename to tests/webhook/test_app.py index c6cc39f..2853ddd 100644 --- a/tests/test_webhook_app.py +++ b/tests/webhook/test_app.py @@ -1,8 +1,9 @@ import pytest from fastapi.testclient import TestClient +from clayde.config import get_settings from clayde.webhook.app import PebblePayload, create_app -from clayde.webhook.queue import JobQueue +from clayde.service.queue import JobQueue @pytest.fixture @@ -39,6 +40,17 @@ def test_pebble_accepts_valid_request(client, queue): assert "id" in body and isinstance(body["id"], str) and len(body["id"]) > 0 +def test_pebble_job_carries_pebble_timeout(client, queue): + r = client.post( + "/webhook/pebble", + json={"text": "hello", "timestamp": 1778068506}, + headers={"Authorization": "Bearer test-token"}, + ) + assert r.status_code == 200 + job = queue._q.get_nowait() + assert job.timeout_s == get_settings().pebble_timeout + + def test_pebble_rejects_missing_token(client): r = client.post( "/webhook/pebble", @@ -89,7 +101,7 @@ async def _noop(**_): @pytest.mark.asyncio async def test_queue_full_emits_ntfy(monkeypatch): from clayde.webhook import app as app_mod - from clayde.webhook.queue import JobQueue, QueueFullError + from clayde.service.queue import JobQueue, QueueFullError calls = [] diff --git a/tests/test_webhook_auth.py b/tests/webhook/test_auth.py similarity index 100% rename from tests/test_webhook_auth.py rename to tests/webhook/test_auth.py diff --git a/uv.lock b/uv.lock index 8132c01..750084c 100644 --- a/uv.lock +++ b/uv.lock @@ -186,6 +186,7 @@ version = "0.1.0" source = { editable = "." } dependencies = [ { name = "anthropic" }, + { name = "croniter" }, { name = "fastapi" }, { name = "jinja2" }, { name = "opentelemetry-api" }, @@ -209,6 +210,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "anthropic", specifier = ">=0.40" }, + { name = "croniter", specifier = ">=2.0" }, { name = "fastapi", specifier = ">=0.115" }, { name = "httpx", marker = "extra == 'dev'", specifier = ">=0.27" }, { name = "jinja2", specifier = ">=3.1.6" }, @@ -247,6 +249,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, ] +[[package]] +name = "croniter" +version = "6.2.4" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "python-dateutil" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/37/57/2e2a65aee2a70483cb28e2b7e15a072d00a523207593b44400d4717bb100/croniter-6.2.4.tar.gz", hash = "sha256:fc124f751b1b04805c2a04b061898b436b45ab2320b045e1e052ea895de65189", size = 166267, upload-time = "2026-07-10T09:52:59.955Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cd/ba/d678e5bd329646ca51d3c92addbc77804e86d21f4b6b6a027218e6abb010/croniter-6.2.4-py3-none-any.whl", hash = "sha256:8ef3d544107a5c05a150a2d78f8bf5a8eb9c5c4d93405a736b824109574e3f4d", size = 46677, upload-time = "2026-07-10T09:52:58.425Z" }, +] + [[package]] name = "cryptography" version = "46.0.5" @@ -953,6 +967,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" }, ] +[[package]] +name = "python-dateutil" +version = "2.9.0.post0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "six" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432, upload-time = "2024-03-01T18:36:20.211Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" }, +] + [[package]] name = "python-dotenv" version = "1.2.2" @@ -1035,6 +1061,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1d/4a/221da6ca167db45693d8d26c7dc79ccfc978a440251bf6721c9aaf251ac0/respx-0.23.1-py2.py3-none-any.whl", hash = "sha256:b18004b029935384bccfa6d7d9d74b4ec9af73a081cc28600fffc0447f4b8c1a", size = 25557, upload-time = "2026-04-08T14:37:14.613Z" }, ] +[[package]] +name = "six" +version = "1.17.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" }, +] + [[package]] name = "sniffio" version = "1.3.1"