openvibers/openvibe.watch Commits

Routines: a watch with a cron, a scheduled check, and a five-failure pause (#232)

OP OpenVibers committed 6 days ago · d78e5102a7 · 13 files, +433 −14

* Routines: a watch with a cron, a scheduled check, and a five-failure pause

* Watch API list: the routine schedule route

Changes

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex fad958f..021d91f 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -1,5 +1,16 @@
# Changelog
++## 2026-11-09 — a routine is a watch with a cron (Actor's triggers)
++
++- **A schedule watch**: `POST /api/v1/watches/:id/schedule` saves a five-field cron and timezone (migration
++ `0021_actor_schedule`). The scheduler fires it like a cadence; `check.run` records the occurrence as a
++ `condition_met` run with carrier `schedule` and queues the watch's actions without fetching anything. The floor is 15
++ minutes, as it is for a cadence. The source stays a valid `watch.watch@1` kind (the subscription is created paused, a
++ webhook source is refused deliveries, and the routine never polls a page), so the released watch schema is unchanged.
++- **A watch that fails five checks in a row pauses itself** and says so (one `agent.step` ended step); resuming clears
++ the streak and arms it again. A routine's action then reaches Actor with `watch_id` and the Actor `trigger_id` from
++ the watch's `actor_trigger` label, so the run links back to its card.
++
What shipped on OpenVibe.Watch, newest first. The plan is T18 ("Watch complete"); each entry names
the step it landed.
ddiff --git a/migrations/0021_actor_schedule.sql b/migrations/0021_actor_schedule.sql
nnew file mode 100644
iindex 0000000..da8e445
---- /dev/null
++++ b/migrations/0021_actor_schedule.sql
@@@ -0,0 +1,7 @@
++-- phase: expand
++-- A watch that runs a routine instead of watching a page: the five-field cron (in the watch's timezone)
++-- Actor saves after creating the subscription. It is separate from cadence because a routine has no source
++-- to fetch; the scheduler fires an action and records the run, and nothing polls.
++ALTER TABLE watches ADD COLUMN IF NOT EXISTS schedule jsonb;
++CREATE INDEX IF NOT EXISTS idx_watches_schedule_due ON watches (next_due_at)
++ WHERE status = 'active' AND deleted_at IS NULL AND schedule IS NOT NULL;
ddiff --git a/server/actions.js b/server/actions.js
iindex 8487374..f50e021 100644
---- a/server/actions.js
++++ b/server/actions.js
@@@ -261,7 +261,14 @@ function createActions({ db, config, notifications = null, post = null, actor =
} else { state = 'skipped'; error = 'watch.action_no_person: an Actor task runs under a person\'s allowance, and this watch belongs to an agent or a service'; }
} else if (kind === 'actor-agent') {
if (USER_SUBJECT.test(String(owner || ''))) {
-- request = { agent_id: action.agent_id, input: actorAgentInputFor({ config, watch, action, value, previousValue, observedAt }) };
++ request = {
++ agent_id: action.agent_id,
++ input: actorAgentInputFor({ config, watch, action, value, previousValue, observedAt }),
++ // The subscription and the Actor trigger it was made for: Actor links the run to the card
++ // (actor.trigger@1) without parsing the watch URL out of the text.
++ watch_id: watch.id,
++ trigger_id: (watch.labels && watch.labels.actor_trigger) || null,
++ };
} else { state = 'skipped'; error = 'watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service'; }
} else if (kind === 'run-job') {
if (!USER_SUBJECT.test(String(owner || ''))) { state = 'skipped'; error = 'watch.action_no_person: a Run job is started for a person in their project, and this watch belongs to an agent or a service'; }
@@@ -344,7 +351,10 @@ function createActions({ db, config, notifications = null, post = null, actor =
async function deliverActorAgent(row, request) {
if (!actor) throw Object.assign(new Error('Actor delivery is not configured in this process'), { code: 'actor_unconfigured' });
try {
-- const task = await actor.runAgent({ subjectId: row.owner_sub, agentId: request.agent_id, input: request.input });
++ const task = await actor.runAgent({
++ subjectId: row.owner_sub, agentId: request.agent_id, input: request.input,
++ watchId: request.watch_id || null, triggerId: request.trigger_id || null,
++ });
return { state: 'sent', error: `Actor task ${task.id || '(no id)'}` };
} catch (err) {
if (err && err.status === 404 && err.code === 'actor.agent.not_found') return { state: 'failed', error: `watch.action_refused: ${err.code}: this agent was not found for the watch owner` };
ddiff --git a/server/actor.js b/server/actor.js
iindex 7119bf1..0627cbf 100644
---- a/server/actor.js
++++ b/server/actor.js
@@@ -4,7 +4,7 @@
*
* const actor = createActorTasks(client, { baseUrl }); // client: openvibe-sdk/core createClient with Watch's token
* await actor.create({ subjectId, task, budgetUsd, idempotencyKey }) → { id, state }
-- * await actor.runAgent({ subjectId, agentId, input }) → { id, state }
++ * await actor.runAgent({ subjectId, agentId, input, watchId, triggerId }) → { id, state }
*
* POST <actor>/api/v1/tasks with Watch's own token (actor.task.create, audience openvibe.actor), or
* POST <actor>/api/v1/me/agents/:agentId/run (actor.agent.run), with the owner in
@@@ -24,8 +24,10 @@ function createActorTasks(client, { baseUrl }) {
const out = await post('/api/v1/tasks', subjectId, json);
return { id: out && out.id, state: out && out.state };
}
-- async function runAgent({ subjectId, agentId, input }) {
-- const out = await post(`/api/v1/me/agents/${encodeURIComponent(agentId)}/run`, subjectId, { input });
++ async function runAgent({ subjectId, agentId, input, watchId = null, triggerId = null }) {
++ const out = await post(`/api/v1/me/agents/${encodeURIComponent(agentId)}/run`, subjectId, {
++ input, ...(watchId ? { watch_id: watchId } : {}), ...(triggerId ? { trigger_id: triggerId } : {}),
++ });
return { id: out && (out.task_id || (out.task && out.task.id) || out.id), state: out && out.state };
}
return { create, runAgent };
ddiff --git a/server/api/watches.js b/server/api/watches.js
iindex 24e3e25..70dddb7 100644
---- a/server/api/watches.js
++++ b/server/api/watches.js
@@@ -10,6 +10,7 @@
* PATCH /api/v1/watches/:id change fields; pause/resume through status
* POST /api/v1/watches/:id/pause status = paused
* POST /api/v1/watches/:id/resume status = active (and re-armed)
++ * POST /api/v1/watches/:id/schedule a routine's five-field cron and timezone (Actor's subscription)
* DELETE /api/v1/watches/:id soft delete (observations stay resolvable)
*
* Another owner's watch is 404, never 403: an id must not confirm that someone else's watch exists.
@@@ -100,6 +101,17 @@ function watchesRouter({ registry, auth, limits, webhookSources = null, sharing
router.post('/api/v1/watches/:id/pause', manage, setStatus('paused'));
router.post('/api/v1/watches/:id/resume', manage, setStatus('active'));
++ // The routine's cron: Actor saves it after creating the subscription as the person. It is separate from the
++ // watch request (watch.watch-request@1 knows source and cadence, not a schedule); the watch stays paused until it
++ // is resumed, so a routine never fires between the create and this call.
++ router.post('/api/v1/watches/:id/schedule', manage, async (req, res) => {
++ try {
++ const out = await registry.setSchedule(req.params.id, req.body || {}, req.principal);
++ if (!out) return notFound(res, req);
++ res.json(out);
++ } catch (err) { fail(res, req, err); }
++ });
++
// A new secret for a webhook watch (the old one stops working at once); also how a watch that became a webhook
// watch gets its first.
router.post('/api/v1/watches/:id/webhook-secret', manage, async (req, res) => {
ddiff --git a/server/app.js b/server/app.js
iindex 8a86bd4..d695cd7 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -47,6 +47,7 @@ const TEXT_INDEX = [
'',
'GET /api/v1/watches, /api/v1/watches/:id (watch.watch.read)',
'POST/PATCH/DELETE /api/v1/watches[/:id], POST /api/v1/watches/:id/pause, /resume (watch.watch.manage)',
++ 'POST /api/v1/watches/:id/schedule (watch.watch.manage)',
'GET /api/v1/watches/:id/observations, /api/v1/watches/:id/checks (watch.observation.read)',
'POST /api/v1/watches/:id/check (watch.check.run)',
'GET /api/health, /api/ready, /release.json',
ddiff --git a/server/check.js b/server/check.js
iindex a839caa..8454f7c 100644
---- a/server/check.js
++++ b/server/check.js
@@@ -193,11 +193,40 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
}
// A run that fired queues the watch's actions in this same commit (server/actions.js): once per trigger.
if (fired && actions) await actions.queue({ watch, run, ...fired });
++ // A watch that has failed five checks in a row pauses itself: one broken page must not keep spending
++ // checks (or waking its owner) until someone notices. Resuming clears the streak and arms it again.
++ if (failures >= 5 && watch.status === 'active') {
++ await db.prepare("UPDATE watches SET status = 'paused', next_due_at = 0, updated_at = ? WHERE id = ? AND status = 'active' AND deleted_at IS NULL")
++ .run(run.finished_at, watch.id);
++ await emitAgentStep(outbox, watch, {
++ topic: 'ended', status: 'stopped', startedAt: watch.created_at, endedAt: run.finished_at,
++ reason: `paused after ${failures} failed checks in a row: ${run.detail || run.error_code || run.state}`.slice(0, 500),
++ baseUrl: config.baseUrl, now,
++ });
++ }
});
outbox.kick().catch(() => { });
return { run, observation, met: run.triggers > 0, failures };
}
++ /**
++ * A routine: a watch with a schedule and no source to fetch. The run is the action itself, recorded as a
++ * condition_met check (so the watch's trail and next occurrence stay truthful) and queued once per
++ * occurrence. A quiet test records the run but fires nothing.
++ */
++ async function runScheduled({ row, watch, trigger, startedAt, quiet = false }) {
++ const finishedAt = now();
++ const value = { scheduled_at: new Date(startedAt).toISOString(), cron: watch.schedule.cron, timezone: watch.schedule.timezone || 'UTC' };
++ const run = {
++ ...baseRun(watch, trigger, 'schedule', startedAt),
++ state: quiet ? 'ok' : 'condition_met', triggers: quiet ? 0 : 1, finished_at: finishedAt,
++ detail: quiet ? 'quiet test: nothing fired' : null,
++ };
++ const fired = quiet ? null : { owner: row.owner_sub, value, previousValue: null, observedAt: finishedAt };
++ const out = await settle({ watch, run, failures: 0, successAt: finishedAt, events: [], fired });
++ return { ...out, met: !quiet };
++ }
++
// ── One check ────────────────────────────────────────────────────────────────────────────
async function runOnce(row, trigger, input = null, { quiet = false } = {}) {
@@@ -221,6 +250,11 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
return { disabled: true, ...(await skip('disabled', `watch_${row.status}`, `the watch is ${row.status}`)) };
}
++ // A routine has no source to fetch: the occurrence itself is the run, and its actions are queued below.
++ if (watch.schedule && watch.schedule.cron) {
++ return await runScheduled({ row, watch, trigger, startedAt, quiet });
++ }
++
// An extraction this release cannot run (an AI reading is an action, step 10): never fetch.
if (watch.extraction && watch.extraction.kind === 'ai') {
const out = await skip('skipped', 'watch.extraction_unavailable', 'ai extraction is an AI action, not a check (plan T18 step 10). Nothing was fetched.');
ddiff --git a/server/cron.js b/server/cron.js
nnew file mode 100644
iindex 0000000..b83d906
---- /dev/null
++++ b/server/cron.js
@@@ -0,0 +1,155 @@
++'use strict';
++/**
++ * A five-field cron, for a watch that runs a routine rather than polling a page: `minute hour
++ * day-of-month month day-of-week`, with `*`, lists, ranges, steps and the usual three-letter month
++ * and day names (`mon-fri`, `jan`). Day-of-month and day-of-week follow the classic rule: when both
++ * are restricted either may match; when one is `*` only the other decides.
++ *
++ * Time zones are Node's own Intl data (no table of ours), so a schedule in Europe/Berlin follows
++ * the clock changes; the two-pass wall-clock conversion settles the offset for the day it is on.
++ * The floor is fifteen minutes between two occurrences (the same floor Watch's cadence has): a
++ * `*​/5` cron is refused at save, never silently throttled.
++ *
++ * parse(expr) → { minutes, hours, days, months, weekdays, dayAny, weekdayAny }
++ * next(expr, timezone, afterMs) → ms of the first occurrence strictly after afterMs | null
++ * validate(expr, timezone) → { valid, error, next_at }
++ */
++
++const MONTHS = { jan: 1, feb: 2, mar: 3, apr: 4, may: 5, jun: 6, jul: 7, aug: 8, sep: 9, oct: 10, nov: 11, dec: 12 };
++const DAYS = { sun: 0, mon: 1, tue: 2, wed: 3, thu: 4, fri: 5, sat: 6 };
++const MIN_EVERY_MS = 15 * 60_000;
++const MAX_DAYS = 1500; // enough for any calendar rule, including the 29 February gap
++
++const ZONE_PARTS = { timeZone: 'UTC', hourCycle: 'h23', year: 'numeric', month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit', second: '2-digit' };
++
++function isZone(zone) {
++ if (typeof zone !== 'string' || !zone.trim()) return false;
++ try { new Intl.DateTimeFormat('en-US', { timeZone: zone.trim() }); return true; } catch { return false; }
++}
++
++function partsInZone(date, zone) {
++ const out = {};
++ for (const part of new Intl.DateTimeFormat('en-US', { ...ZONE_PARTS, timeZone: zone }).formatToParts(date)) {
++ if (part.type !== 'literal') out[part.type] = part.value;
++ }
++ return {
++ year: Number(out.year), month: Number(out.month), day: Number(out.day),
++ hour: Number(out.hour) % 24, minute: Number(out.minute), second: Number(out.second),
++ };
++}
++
++/** The zone's offset from UTC at an instant, in milliseconds. */
++function offsetMs(date, zone) {
++ const p = partsInZone(date, zone);
++ const asUtc = Date.UTC(p.year, p.month - 1, p.day, p.hour, p.minute, p.second);
++ return asUtc - Math.floor(date.getTime() / 1000) * 1000;
++}
++
++/** The UTC instant of a wall-clock time in a zone (two passes settle the offset, DST included). */
++function zonedToUtc({ year, month, day, hour = 0, minute = 0 }, zone) {
++ const wall = Date.UTC(year, month - 1, day, hour, minute, 0);
++ let ts = wall - offsetMs(new Date(wall), zone);
++ ts = wall - offsetMs(new Date(ts), zone);
++ return new Date(ts);
++}
++
++function parseField(text, { min, max, names = null, weekday = false }) {
++ const out = new Set();
++ const source = String(text || '').trim().toLowerCase();
++ if (!source) throw new Error('an empty cron field');
++ for (const part of source.split(',')) {
++ if (!part) throw new Error(`bad cron field "${source}"`);
++ const [rangeText, stepText] = part.split('/');
++ const step = stepText === undefined ? 1 : Number(stepText);
++ if (!Number.isInteger(step) || step < 1 || step > max - min + 1) throw new Error(`bad cron step "${part}"`);
++ let lo, hi;
++ if (rangeText === '*' || rangeText === '?') { lo = min; hi = max; }
++ else {
++ const [a, b] = rangeText.split('-');
++ const value = (v) => {
++ const n = names && Object.prototype.hasOwnProperty.call(names, v) ? names[v] : (/^\d+$/.test(v) ? Number(v) : NaN);
++ if (!Number.isInteger(n)) throw new Error(`bad cron value "${v}"`);
++ return n;
++ };
++ lo = value(a);
++ hi = b === undefined ? (stepText === undefined ? lo : max) : value(b);
++ if (weekday) { if (lo === 7) lo = 0; if (hi === 7) hi = 0; }
++ if (lo < min || lo > max || hi < min || hi > max || lo > hi) {
++ if (!(weekday && lo === 0 && hi === 6 && rangeText.includes('7'))) throw new Error(`cron value out of range in "${part}"`);
++ }
++ }
++ for (let v = lo; v <= hi; v += step) out.add(v);
++ }
++ return out;
++}
++
++function parse(expr) {
++ const fields = String(expr || '').trim().split(/\s+/);
++ if (fields.length !== 5) throw new Error('a cron has five fields: minute hour day-of-month month day-of-week');
++ return {
++ expr: fields.join(' '),
++ minutes: parseField(fields[0], { min: 0, max: 59 }),
++ hours: parseField(fields[1], { min: 0, max: 23 }),
++ days: parseField(fields[2], { min: 1, max: 31 }),
++ months: parseField(fields[3], { min: 1, max: 12, names: MONTHS }),
++ weekdays: parseField(fields[4], { min: 0, max: 6, names: DAYS, weekday: true }),
++ dayAny: fields[2] === '*' || fields[2] === '?',
++ weekdayAny: fields[4] === '*' || fields[4] === '?',
++ };
++}
++
++const sorted = (set) => [...set].sort((a, b) => a - b);
++
++/** The first occurrence strictly after `afterMs` in `timezone`; null when the rule never runs. */
++function next(expr, timezone, afterMs = Date.now()) {
++ const rule = typeof expr === 'string' ? parse(expr) : expr;
++ const zone = isZone(timezone) ? timezone : 'UTC';
++ const after = Number(afterMs) || Date.now();
++ const start = partsInZone(new Date(after), zone);
++ const hours = sorted(rule.hours);
++ const minutes = sorted(rule.minutes);
++ for (let offset = 0; offset <= MAX_DAYS; offset += 1) {
++ const date = new Date(Date.UTC(start.year, start.month - 1, start.day + offset));
++ const year = date.getUTCFullYear();
++ const month = date.getUTCMonth() + 1;
++ const day = date.getUTCDate();
++ if (!rule.months.has(month)) continue;
++ const weekday = date.getUTCDay();
++ const byDay = rule.days.has(day);
++ const byWeekday = rule.weekdays.has(weekday);
++ const dayMatches = rule.dayAny && rule.weekdayAny ? true
++ : rule.dayAny ? byWeekday
++ : rule.weekdayAny ? byDay
++ : byDay || byWeekday;
++ if (!dayMatches) continue;
++ for (const hour of hours) {
++ for (const minute of minutes) {
++ const at = zonedToUtc({ year, month, day, hour, minute }, zone).getTime();
++ if (at > after) return at;
++ }
++ }
++ }
++ return null;
++}
++
++/** Whether the rule is usable: a zone Node knows, an occurrence within reach, and 15 minutes or more between runs. */
++function validate(expr, timezone = 'UTC') {
++ if (!isZone(timezone)) return { valid: false, error: `unknown time zone "${timezone}"`, next_at: null };
++ let rule;
++ try { rule = parse(expr); } catch (err) { return { valid: false, error: err.message, next_at: null }; }
++ const now = Date.now();
++ const first = next(rule, timezone, now);
++ if (first == null) return { valid: false, error: 'this cron never runs', next_at: null };
++ let previous = first;
++ for (let i = 0; i < 4; i += 1) {
++ const following = next(rule, timezone, previous);
++ if (following == null) break;
++ if (following - previous < MIN_EVERY_MS) {
++ return { valid: false, error: `the smallest interval is 15 minutes; this cron runs again after ${Math.round((following - previous) / 1000)} seconds`, next_at: new Date(first).toISOString() };
++ }
++ previous = following;
++ }
++ return { valid: true, error: null, next_at: new Date(first).toISOString() };
++}
++
++module.exports = { parse, next, validate, isZone, MIN_EVERY_MS, partsInZone, zonedToUtc };
ddiff --git a/server/registry.js b/server/registry.js
iindex ac73f09..7997c93 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -26,6 +26,7 @@
*/
const { validate, ids: contractIds } = require('openvibe-contracts');
const safe = require('./safe-regex');
++const cron = require('./cron');
const { envelope } = require('./events/envelope');
const { emitAgentStep } = require('./events/agent-step');
@@@ -150,6 +151,11 @@ function validateRequest(body, { create = false } = {}) {
* the same shape OpenVibe.Sources uses for a failing source). 0 for a source that never polls.
*/
function nextDue(rec, at, { failures = 0, maxBackoffMs = Infinity } = {}) {
++ // A routine: the next occurrence of its cron, in its own timezone. A failed check still waits for the
++ // next occurrence (there is no page to back off from); an unknown or empty rule never runs.
++ if (rec && rec.schedule && rec.schedule.cron) {
++ try { return cron.next(rec.schedule.cron, rec.schedule.timezone || 'UTC', at) || 0; } catch { return 0; }
++ }
if (!rec || !rec.cadence || !rec.cadence.every_sec) return 0;
const every = rec.cadence.every_sec * 1000;
if (failures > 0) return at + Math.min(every * 2 ** Math.min(failures - 1, 10), Math.max(maxBackoffMs, every));
@@@ -228,10 +234,11 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
project_env: (v) => v,
};
-- /** The stored record, request-shaped (what a PATCH merges into). */
++ /** The stored record, request-shaped (what a PATCH merges into). `schedule` is kept out of the
++ * enumerable shape: watch.watch-request@1 does not know it, but the check engine and nextDue read it. */
function fromRow(row) {
if (!row) return null;
-- return {
++ const out = {
id: row.id,
project_id: row.project_id,
...(row.project_env ? { project_env: row.project_env } : {}),
@@@ -248,12 +255,14 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
budget: parse(row.budget) || {},
retention: parse(row.retention) || {},
};
++ Object.defineProperty(out, 'schedule', { value: parse(row.schedule), enumerable: false });
++ return out;
}
/** What the operator sees about a watch's runtime, never invented: counts and times only. */
function health(row) {
const iso = (v) => (v == null ? null : new Date(v).toISOString());
-- const polls = row.cadence != null;
++ const polls = row.cadence != null || row.schedule != null;
let status;
if (row.status === 'paused') status = 'paused';
else if (row.status === 'disabled') status = 'disabled';
@@@ -419,7 +428,9 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
}
if (!sets.length) return await st.get.get(row.id);
// A changed watch gets a fresh cadence: re-arm from now, so editing never fires late work.
-- const rec = { ...current, ...changes };
++ const rec = { ...current, ...changes, schedule: parse(row.schedule) };
++ // A resume is a fresh start: the failures that paused a routine (or that a person paused around) begin again.
++ if (rec.status === 'active' && row.status !== 'active') sets.push('consecutive_failures = 0', 'not_before = 0');
args.next_due_at = rec.status === 'active' ? arm(rec, args.updated_at) : 0;
sets.push('next_due_at = @next_due_at');
// The first resume is the arm: a watch the site tested is paused with no armed_at until now.
@@@ -447,6 +458,23 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
return await st.get.get(row.id);
}
++ /** Save the routine's cron (Actor's route after it creates the subscription). The row stays paused until resumed. */
++ async function setSchedule(id, schedule, principal) {
++ const row = await get(id, principal);
++ if (!row) return null;
++ const cronExpr = String((schedule && schedule.cron) || '').trim();
++ const timezone = String((schedule && schedule.timezone) || 'UTC').trim() || 'UTC';
++ const verdict = cron.validate(cronExpr, timezone);
++ if (!verdict.valid) throw new RegistryError(verdict.error, 'watch.invalid');
++ const rec = { ...fromRow(row), schedule: { cron: cronExpr, timezone } };
++ const at = now();
++ const next = row.status === 'active' ? nextDue(rec, at, {}) : 0;
++ await db.query(`UPDATE watches SET schedule = $2, next_due_at = $3, armed_at = COALESCE(armed_at, $4),
++ updated_at = $4, updated_by = $5 WHERE id = $1 AND deleted_at IS NULL`,
++ [row.id, JSON.stringify(rec.schedule), next, at, ownerOf(principal)]);
++ return { id: row.id, schedule: rec.schedule, next_due_at: next ? new Date(next).toISOString() : null };
++ }
++
/** Soft delete: the rows an observation refers to stay resolvable. */
async function remove(id, principal) {
const row = await get(id, principal);
@@@ -482,7 +510,8 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
/** Active watches whose turn has come, oldest due first (the scheduler's queue). */
async function due(limit) {
-- return await db.prepare(`SELECT id FROM watches WHERE status = 'active' AND deleted_at IS NULL AND cadence IS NOT NULL
++ return await db.prepare(`SELECT id FROM watches WHERE status = 'active' AND deleted_at IS NULL
++ AND (cadence IS NOT NULL OR schedule IS NOT NULL)
AND next_due_at <= @now AND not_before <= @now ORDER BY next_due_at, id LIMIT @limit`).all({ now: now(), limit });
}
@@@ -497,7 +526,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
};
}
-- return { validate: validateRequest, validateView, view, health, fromRow, create, get, all, patch, remove, due, onEventSourceChange: (fn) => { onEventSource = fn; } };
++ return { validate: validateRequest, validateView, view, health, fromRow, create, get, all, patch, setSchedule, remove, due, onEventSourceChange: (fn) => { onEventSource = fn; } };
}
module.exports = { createRegistry, validateRequest, validateView, RegistryError, defaultName, nextDue, POLLING, PUSHED, RUN_STATES, STATUSES };
ddiff --git a/server/scheduler.js b/server/scheduler.js
iindex aaa33bb..e3353d4 100644
---- a/server/scheduler.js
++++ b/server/scheduler.js
@@@ -1,15 +1,17 @@
'use strict';
/**
* In-process scheduler: every WATCH_TICK_MS, start checks for active watches whose next_due_at and
-- * not_before have passed, oldest due first, up to WATCH_MAX_CONCURRENT checks at a time. The due
-- * query is the one the migration's partial index serves, and check.run() itself refuses a second
++ * not_before have passed, oldest due first, up to WATCH_MAX_CONCURRENT checks at a time. A routine
++ * (schedule) watch is due the same way; its check runs the action without fetching anything. The due
++ * query is the one the migrations' partial indexes serve, and check.run() itself refuses a second
* check of a watch already in flight (so a manual check and a scheduled one never overlap).
*
* Copied from OpenVibe.Sources/server/scheduler.js with Watch's due query (watches, not sources;
* cadence IS NOT NULL, since a source pushed to by an event or webhook never polls).
*/
function createScheduler({ db, check, config, now = () => Date.now(), log = console }) {
-- const due = db.prepare(`SELECT id FROM watches WHERE status = 'active' AND deleted_at IS NULL AND cadence IS NOT NULL
++ const due = db.prepare(`SELECT id FROM watches WHERE status = 'active' AND deleted_at IS NULL
++ AND (cadence IS NOT NULL OR schedule IS NOT NULL)
AND next_due_at <= @now AND not_before <= @now ORDER BY next_due_at, id LIMIT @limit`);
let timer = null;
const running = new Set();
ddiff --git a/server/webhook-sources.js b/server/webhook-sources.js
iindex 5c2c054..f6a4695 100644
---- a/server/webhook-sources.js
++++ b/server/webhook-sources.js
@@@ -319,6 +319,9 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
const row = /^wch_[0-9A-HJKMNP-TV-Z]{26}$/.test(id) ? await st.watch.get(id) : null;
const source = row ? (typeof row.source === 'string' ? JSON.parse(row.source) : row.source) : null;
if (!row || !source || source.kind !== 'webhook') return problem(res, 404, 'watch.not_found', 'no webhook watch at this address');
++ // A routine uses the same stored address so the create stays inside watch.watch@1's source kinds; it is not
++ // a page anyone can push to, and a delivery here must not run the person's agent out of turn.
++ if (row.schedule) return problem(res, 404, 'watch.not_found', 'this address belongs to a routine, not a webhook watch');
const kept = await st.get.get(id) || {};
const provider = kept.provider || null;
if (!verified(kept.secret || null, raw, req.headers, { provider, now: now() })) {
ddiff --git a/test/check.test.js b/test/check.test.js
iindex 600f66d..95106d9 100644
---- a/test/check.test.js
++++ b/test/check.test.js
@@@ -326,6 +326,20 @@ t('the trail carries previous_hash, and an identical body records no second obse
assert.strictEqual((await observations(w.id)).length, 1);
});
++t('a watch that fails five checks in a row pauses itself, and resuming gives it a fresh streak', async () => {
++ const w = await create(person, { source: { kind: 'http', url: `${web.origin}/broken`, format: null }, extraction: { kind: 'json' } });
++ for (let i = 0; i < 5; i += 1) await check(w.id);
++ let r = await row(w.id);
++ assert.strictEqual(r.status, 'paused');
++ assert.strictEqual(Number(r.consecutive_failures), 5);
++ const disabled = await check(w.id);
++ assert.strictEqual(disabled.disabled, true, 'a paused watch is not fetched again');
++ const resumed = await svc.registry.patch(w.id, { status: 'active' }, person);
++ assert.strictEqual(resumed.status, 'active');
++ assert.strictEqual(Number(resumed.consecutive_failures), 0, 'the resume is a fresh streak');
++ assert.ok(resumed.next_due_at > Date.now());
++});
++
t('done', async () => { await web.close(); await svc.stop(); });
t.run();
ddiff --git a/test/schedule.test.js b/test/schedule.test.js
nnew file mode 100644
iindex 0000000..fc3ce26
---- /dev/null
++++ b/test/schedule.test.js
@@@ -0,0 +1,139 @@
++'use strict';
++/**
++ * Routines: a watch with a five-field cron instead of a source to fetch. The cron parser, the API route
++ * that saves it (owner-only, 15-minute floor), the scheduled check that queues the watch's action once
++ * per occurrence, the scheduler's due query, and the actor-agent delivery carrying the Actor trigger id.
++ */
++const assert = require('assert');
++const { ids } = require('openvibe-contracts');
++const { boot, request, userToken, suite, newUser, newAgent } = require('./helpers');
++const cron = require('../server/cron');
++
++const t = suite('schedule');
++let svc;
++const person = newUser();
++const other = newUser();
++const ME = userToken(person);
++const OTHER = userToken(other);
++const agent = newAgent();
++const actorCalls = [];
++const actor = {
++ async create(req) { actorCalls.push(req); return { id: ids.newId('task'), state: 'queued' }; },
++ async runAgent(req) { actorCalls.push(req); return { id: ids.newId('task'), state: 'queued' }; },
++};
++const notifications = { async push() { return { sent: true }; } };
++const api = (method, p, opts = {}) => request(svc.base, method, p, opts);
++
++const routine = (overrides = {}) => ({
++ name: 'Actor routine: summarise my inbox',
++ status: 'paused',
++ source: { kind: 'webhook' },
++ cadence: null,
++ extraction: { kind: 'text' },
++ condition: { op: 'exists' },
++ action: [{ kind: 'actor-agent', agent_id: agent, note: 'summarise my inbox' }],
++ labels: { actor_trigger: 'trg_01M6VNTQD7NB3QH8K4R2S9T6VW' },
++ ...overrides,
++});
++
++t('cron: weekdays at 9 in Berlin, Sunday evening, and the 15-minute floor', async () => {
++ const monday = Date.UTC(2026, 10, 9, 9); // Mon 9 Nov 2026 10:00 Berlin, after 09:00 local
++ const next = new Date(cron.next('0 9 * * 1-5', 'Europe/Berlin', monday));
++ assert.strictEqual(next.toISOString(), '2026-11-10T08:00:00.000Z', 'Tuesday 09:00 CET');
++ // Friday 09:00 to Monday 09:00 is the weekend: still the next weekday occurrence.
++ const friday = Date.UTC(2026, 10, 13, 8);
++ assert.strictEqual(new Date(cron.next('0 9 * * 1-5', 'Europe/Berlin', friday)).toISOString(), '2026-11-16T08:00:00.000Z');
++ assert.strictEqual(new Date(cron.next('0 19 * * 0', 'Europe/Berlin', Date.UTC(2026, 10, 8, 12))).toISOString(), '2026-11-08T18:00:00.000Z', 'Sunday 19:00 CET');
++ assert.strictEqual(cron.validate('0 9 * * 1-5', 'Europe/Berlin').valid, true);
++ const fast = cron.validate('*/5 * * * *', 'Europe/Berlin');
++ assert.strictEqual(fast.valid, false);
++ assert.match(fast.error, /15 minutes/);
++ assert.strictEqual(cron.validate('0 9 * * 1-5', 'Mars/Olympus').valid, false, 'an unknown zone is refused');
++ assert.strictEqual(cron.validate('0 9 * *', 'UTC').valid, false, 'five fields are required');
++ assert.strictEqual(new Date(cron.next('30 8 * * *', 'UTC', Date.UTC(2026, 10, 9, 8, 0))).toISOString(), '2026-11-09T08:30:00.000Z');
++});
++
++t('boot', async () => {
++ svc = await boot({ actor, notifications, worker: 'off' });
++});
++
++t('the schedule route is the person\'s, validates the cron and 404s another person', async () => {
++ const made = await api('POST', '/api/v1/watches', { token: ME, body: routine() });
++ assert.strictEqual(made.status, 201, made.text);
++ const id = made.body.watch.id;
++ assert.strictEqual(made.body.watch.status, 'paused');
++
++ const bad = await api('POST', `/api/v1/watches/${id}/schedule`, { token: ME, body: { cron: '*/5 * * * *', timezone: 'Europe/Berlin' } });
++ assert.strictEqual(bad.status, 422, bad.text);
++ assert.match(bad.text, /15 minutes/);
++
++ const ok = await api('POST', `/api/v1/watches/${id}/schedule`, { token: ME, body: { cron: '0 9 * * 1-5', timezone: 'Europe/Berlin' } });
++ assert.strictEqual(ok.status, 200, ok.text);
++ assert.deepStrictEqual(ok.body.schedule, { cron: '0 9 * * 1-5', timezone: 'Europe/Berlin' });
++ assert.strictEqual(ok.body.next_due_at, null, 'a paused routine is not armed');
++
++ assert.strictEqual((await api('POST', `/api/v1/watches/${id}/schedule`, { token: OTHER, body: { cron: '0 9 * * 1-5', timezone: 'Europe/Berlin' } })).status, 404, 'another person gets 404');
++ assert.strictEqual((await api('POST', '/api/v1/watches/wch_01M6VNTQD796C68VF8B182VAGS/schedule', { token: ME, body: { cron: '0 9 * * 1-5', timezone: 'UTC' } })).status, 404, 'a ghost id is 404');
++
++ const resumed = await api('POST', `/api/v1/watches/${id}/resume`, { token: ME });
++ assert.strictEqual(resumed.status, 200, resumed.text);
++ const view = await api('GET', `/api/v1/watches/${id}`, { token: ME });
++ assert.ok(view.body.watch.health.next_due_at, 'a resumed routine shows its next run');
++ assert.ok(view.body.watch.health.next_due_at > new Date().toISOString());
++});
++
++t('a scheduled check runs the action once and rearms for the next occurrence', async () => {
++ const made = await api('POST', '/api/v1/watches', { token: ME, body: routine({ name: 'nightly brief', labels: { actor_trigger: 'trg_01M6VNTQD7NB3QH8K4R2S9T6VW' } }) });
++ const id = made.body.watch.id;
++ await api('POST', `/api/v1/watches/${id}/schedule`, { token: ME, body: { cron: '0 7 * * *', timezone: 'UTC' } });
++ await api('POST', `/api/v1/watches/${id}/resume`, { token: ME });
++
++ const out = await svc.check.run(id, { trigger: 'schedule' });
++ assert.strictEqual(out.run.state, 'condition_met');
++ assert.strictEqual(out.run.triggers, 1);
++ assert.strictEqual(out.run.carrier, 'schedule');
++ const runs = await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);
++ assert.strictEqual(runs.length, 1);
++ const actions = await svc.db.prepare('SELECT * FROM action_runs WHERE watch_id = ?').all(id);
++ assert.strictEqual(actions.length, 1);
++ assert.strictEqual(actions[0].kind, 'actor-agent');
++ const request = typeof actions[0].request === 'string' ? JSON.parse(actions[0].request) : actions[0].request;
++ assert.strictEqual(request.trigger_id, 'trg_01M6VNTQD7NB3QH8K4R2S9T6VW');
++ assert.strictEqual(request.watch_id, id);
++
++ await svc.actions.tick();
++ const call = actorCalls.find((c) => c.agentId === agent && c.triggerId === 'trg_01M6VNTQD7NB3QH8K4R2S9T6VW');
++ assert.ok(call, 'the actor-agent delivery names the subscription and the Actor trigger');
++ assert.strictEqual(call.watchId, id);
++
++ const row = await svc.db.prepare('SELECT * FROM watches WHERE id = ?').get(id);
++ assert.ok(row.next_due_at > Date.now(), 'the next occurrence is armed');
++ assert.strictEqual(Number(row.consecutive_failures), 0);
++});
++
++t('the scheduler due query includes a routine and never a paused one', async () => {
++ const made = await api('POST', '/api/v1/watches', { token: ME, body: routine({ name: 'due routine' }) });
++ const id = made.body.watch.id;
++ await api('POST', `/api/v1/watches/${id}/schedule`, { token: ME, body: { cron: '0 7 * * *', timezone: 'UTC' } });
++ await api('POST', `/api/v1/watches/${id}/resume`, { token: ME });
++ await svc.db.prepare('UPDATE watches SET next_due_at = 0 WHERE id = ?').run(id);
++ assert.ok((await svc.registry.due(10)).some((w) => w.id === id), 'a due routine is in the scheduler queue');
++ const idsRun = await svc.scheduler.tick();
++ assert.ok(idsRun.includes(id));
++ await new Promise((r) => setTimeout(r, 100));
++ const row = await svc.db.prepare('SELECT status, next_due_at FROM watches WHERE id = ?').get(id);
++ assert.strictEqual(row.status, 'active');
++ assert.ok(row.next_due_at > Date.now());
++});
++
++t('a routine is not a webhook address anyone can push to', async () => {
++ const made = await api('POST', '/api/v1/watches', { token: ME, body: routine({ name: 'not a hook' }) });
++ assert.strictEqual(made.status, 201, made.text);
++ const id = made.body.watch.id;
++ await api('POST', `/api/v1/watches/${id}/schedule`, { token: ME, body: { cron: '0 7 * * *', timezone: 'UTC' } });
++ const pushed = await fetch(`${svc.base}/hooks/${id}`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' });
++ assert.strictEqual(pushed.status, 404);
++});
++
++t.done = true;
++(async () => { await t.run(); await svc.stop(); })().catch((err) => { console.error(err); process.exit(1); });