openvibers/openvibe.actor Commits

Triggers in conversation: schedule_trigger, the Watch subscription as the person, and the card's moves (#325)

OP OpenVibers committed 6 days ago · 2528b767a4 · 27 files, +1555 −17

* Triggers in conversation: schedule_trigger, the Watch subscription as the person, and the card's moves

* Triggers test: wait for the card's last run after the task ends

* Triggers review: only a person's own token reaches Watch; the routine's approval says what and when

- watchToken() returns nothing unless the request is a person (principal kind 'user'): an app or a service
  acting for a person (kind 'app') holds its own Bearer, which was stashed as the person's run token and would
  have been sent to Watch as theirs. Test: a service's POST /api/v1/tasks leaves no run token (fails without
  the guard).
- stepFor('schedule_trigger'): the approval card reads "Set up a task that repeats (0 9 * * 1-5,
  Europe/Berlin): "summarise my inbox"" instead of "Use schedule_trigger".

Changes

ddiff --git a/.env.example b/.env.example
iindex ec94231..3bc1c7e 100644
---- a/.env.example
++++ b/.env.example
@@@ -92,6 +92,9 @@ ACTOR_RUNTIME_OUTPUT_USD_PER_MTOK=0
# OpenVibe.Space: the person's own spaces, which their agent builds (plan 2026-10-22 §Z stage Z1). With this
# address the space_* tools are offered to a person's task; without it they are off.
# ACTOR_SPACE_URL=http://127.0.0.1:4940
++# OpenVibe.Watch: a trigger's subscription lives there, created and changed with the person's own token. Empty
++# turns triggers off.
++# ACTOR_WATCH_URL=http://127.0.0.1:4730
# The files and charts a task makes (Actor v2, stage V2-4) live in Actor's own PostgreSQL, private to the person who
# asked, and are served from the task's own page (server/runtime/artifacts.js says why they are not on OpenVibe.Media).
# The largest single file a task may make, and the most one task may hold.
ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 5baa784..436e818 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -2,6 +2,26 @@
What changed in OpenVibe.Actor, newest first. Each site also publishes its patch notes at /updates.
++## Unreleased — triggers in conversation
++
++- **Triggers made in the conversation** (plan §12 A-4): `schedule_trigger` turns "every weekday at 9, summarise my inbox"
++ or "when a new issue lands in my repo, triage it" into a Watch subscription the person owns, created with their own
++ token — never Actor's service token acting for them. The tool parses the sentence, and when it cannot (no time, no
++ timezone, two times, no event it can name) it makes nothing and asks the one question instead. The card is a
++ `trigger` step on the turn: when in words and cron, what it runs, the next run, the last result (ok, failed, needs
++ you) and active/paused (migration 0037).
++- **The person's moves** are Actor routes: `GET/POST /api/v1/triggers`, `PATCH /api/v1/triggers/:id`, `…/pause`,
++ `…/resume`, `…/run` and `DELETE` (undo for ten minutes, then Delete with a confirmation). Another person gets 404.
++ At most 20 active triggers, the smallest interval is 15 minutes, and five failed runs in a row pause the trigger and
++ say so on the card.
++- **Watch runs come back as assignment roots** with origin `routine` and the trigger's own level, capped there: a
++ routine cannot borrow a higher standing level from its agent. Money, passwords, deletions and new recipients still
++ always ask inside the run.
++- Watch gets the routine's cron (`POST /api/v1/watches/:id/schedule`), a scheduled check that queues the action without
++ fetching, and its own five-failed-check pause; an actor-agent delivery now names the watch and the Actor trigger.
++- Contracts 0.330.0 adds `actor.trigger@1` and the `trigger` step on `actor.task-event@2`; Actor still pins 0.328.0
++ until it is released, so the routes are guarded by `actor.task.read`/`actor.task.create` in the meantime.
++
## Unreleased — one Computer
- **One Computer** (plan 2026-11-07: "Surf, Computer and Actor are one experience"): when a task opens a desk, its
ddiff --git a/migrations/0037_actor_triggers.sql b/migrations/0037_actor_triggers.sql
nnew file mode 100644
iindex 0000000..f6d55bd
---- /dev/null
++++ b/migrations/0037_actor_triggers.sql
@@@ -0,0 +1,40 @@
++-- phase: expand
++-- Triggers made in a conversation (plan §12 A-4): the routine's schedule (or its Watch event source), the
++-- agent it runs, the card's last result and failure streak, and the ten-minute undo window. The Watch
++-- subscription is the runner; this table is Actor's card and ownership record.
++CREATE TABLE IF NOT EXISTS actor_triggers (
++ id text COLLATE "C" PRIMARY KEY,
++ owner text COLLATE "C" NOT NULL,
++ agent_id text COLLATE "C" NOT NULL,
++ task text COLLATE "C" NOT NULL,
++ kind text COLLATE "C" NOT NULL CHECK (kind IN ('schedule','event')),
++ schedule jsonb,
++ event jsonb,
++ timezone text COLLATE "C",
++ words text COLLATE "C" NOT NULL,
++ level text COLLATE "C" NOT NULL DEFAULT 'ask',
++ state text COLLATE "C" NOT NULL DEFAULT 'active' CHECK (state IN ('active','paused','deleted')),
++ paused_reason text COLLATE "C",
++ failures bigint NOT NULL DEFAULT 0,
++ watch_id text COLLATE "C",
++ turn_task_id text COLLATE "C",
++ last_task_id text COLLATE "C",
++ last_state text COLLATE "C",
++ last_run_at bigint,
++ created_at bigint NOT NULL,
++ updated_at bigint NOT NULL,
++ deleted_at bigint
++);
++CREATE INDEX IF NOT EXISTS actor_triggers_owner ON actor_triggers (owner, created_at DESC) WHERE deleted_at IS NULL;
++CREATE INDEX IF NOT EXISTS actor_triggers_watch ON actor_triggers (watch_id) WHERE watch_id IS NOT NULL;
++
++-- One trigger run per task. The engine updates this row as the task moves; a terminal result writes the
++-- trigger's last run and its consecutive-failure streak.
++CREATE TABLE IF NOT EXISTS actor_trigger_runs (
++ trigger_id text COLLATE "C" NOT NULL REFERENCES actor_triggers(id) ON DELETE CASCADE,
++ task_id text COLLATE "C" PRIMARY KEY REFERENCES tasks(id) ON DELETE CASCADE,
++ state text COLLATE "C" NOT NULL,
++ created_at bigint NOT NULL,
++ updated_at bigint NOT NULL
++);
++CREATE INDEX IF NOT EXISTS actor_trigger_runs_trigger ON actor_trigger_runs (trigger_id, created_at DESC);
ddiff --git a/public/js/chat.js b/public/js/chat.js
iindex 4744004..7b3cfac 100644
---- a/public/js/chat.js
++++ b/public/js/chat.js
@@@ -14,6 +14,13 @@
'use strict';
var streams = [];
++ // The person's own timezone, for a trigger made in the chat: the browser knows it, the server should not guess.
++ // A cookie, not a token or a task field; the parser still refuses a zone Node does not know and asks instead.
++ try {
++ var zone = Intl.DateTimeFormat().resolvedOptions().timeZone;
++ if (zone) document.cookie = 'ov_tz=' + encodeURIComponent(zone) + '; Path=/; Max-Age=31536000; SameSite=Lax' + (location.protocol === 'https:' ? '; Secure' : '');
++ } catch (e) { /* no Intl: the tool asks for the timezone */ }
++
function start() {
var chat = document.querySelector('.chat');
if (!chat || chat.dataset.ready) return;
ddiff --git a/server/agents/actions.js b/server/agents/actions.js
iindex f651f52..9b4c219 100644
---- a/server/agents/actions.js
++++ b/server/agents/actions.js
@@@ -138,6 +138,18 @@ function stepFor(name, input) {
return { act: 'act', category: categories[0], summary, payload: safeInput(a.input) };
}
++ // ── A routine: it keeps running after this turn, so the card says what and when in plain words ──
++ case 'schedule_trigger': {
++ const what = text(a.task || a.request, 160);
++ const when = a.cron ? `${text(a.cron, 40)}${a.timezone ? `, ${text(a.timezone, 40)}` : ''}`
++ : (a.event && a.event.pattern ? `when ${text(a.event.pattern, 80)} happens` : '');
++ return {
++ act: 'act',
++ summary: `Set up a task that repeats${when ? ` (${when})` : ''}: ${quoted(what, 160) || 'the one you asked for'}`,
++ payload: safeInput({ request: a.request, task: a.task, cron: a.cron, timezone: a.timezone, event: a.event, agent_id: a.agent_id }),
++ };
++ }
++
// ── Actor's own working state: never asks ──
case 'make_file':
case 'make_chart':
ddiff --git a/server/agents/assignments.js b/server/agents/assignments.js
iindex 8b3d2cb..583caaf 100644
---- a/server/agents/assignments.js
++++ b/server/agents/assignments.js
@@@ -101,7 +101,7 @@ function originOf(task, explicit) {
const agentOf = (s, config, task) => contacts.agentForTask(s, config, task).catch(() => null);
/** Make the task's root node, or return the one it already has. */
--async function ensureForTask(s, config, task, { origin = null } = {}) {
++async function ensureForTask(s, config, task, { origin = null, level = null } = {}) {
if (!task || !task.id || !String(task.requester || '').startsWith('user:')) return null;
if (task.assignment_id) {
const linked = await byId(s, task.requester, task.assignment_id);
@@@ -114,7 +114,11 @@ async function ensureForTask(s, config, task, { origin = null } = {}) {
}
const agent = await agentOf(s, config, task);
const own = levels.LEVELS.includes(task.level) ? task.level : 'ask';
-- const effective = effectiveLevel(own, agent && agent.level);
++ // A routine runs at the level the person made it with and never above it: the agent's standing level may raise
++ // an ordinary hand-off, but a trigger made in a thread cannot borrow more autonomy than that thread had.
++ const effective = ORIGINS.includes(origin) && origin === 'routine' && levels.LEVELS.includes(level)
++ ? levels.minLevel(effectiveLevel(own, agent && agent.level), level)
++ : effectiveLevel(own, agent && agent.level);
const now = s.iso();
const id = s.newId('asg');
let row = await s.db.maybe(`INSERT INTO actor_assignments
ddiff --git a/server/agents/levels.js b/server/agents/levels.js
iindex 25cc474..5ecebd3 100644
---- a/server/agents/levels.js
++++ b/server/agents/levels.js
@@@ -32,6 +32,18 @@ function maxLevel(...levels) {
return out;
}
++/** The lower of the named levels: a routine may never run above the level of the person who made it. */
++function minLevel(...levels) {
++ let out = 'full';
++ let found = false;
++ for (const level of levels) {
++ if (!LEVELS.includes(level)) continue;
++ if (!found || RANK[level] < RANK[out]) out = level;
++ found = true;
++ }
++ return found ? out : 'ask';
++}
++
function textOf(value, depth = 0) {
if (value == null || depth > 4) return '';
if (typeof value === 'string') return value;
@@@ -104,4 +116,4 @@ async function review({ models = null, summary = '', payload = null, signal = nu
}
}
--module.exports = { LEVELS, RANK, ALWAYS_ASK, maxLevel, classify, floorFor, decide, review, REVIEW_TIMEOUT_MS };
++module.exports = { LEVELS, RANK, ALWAYS_ASK, maxLevel, minLevel, classify, floorFor, decide, review, REVIEW_TIMEOUT_MS };
ddiff --git a/server/agents/personal.js b/server/agents/personal.js
iindex f131a11..3bf7ec1 100644
---- a/server/agents/personal.js
++++ b/server/agents/personal.js
@@@ -193,12 +193,12 @@ function personText(task) {
return end < 0 ? t : t.slice(0, end);
}
--async function run(ctx, agent, input = '', { origin = 'person' } = {}) {
++async function run(ctx, agent, input = '', { origin = 'person', level = null, triggerId = null } = {}) {
const { s, config, engine, webhooks } = ctx;
if (typeof input !== 'string' || input.length > 8000) return { status: 422, code: 'actor.agent.invalid', detail: 'input must be at most 8000 characters.' };
// The person's own active memory and this agent's active notes; a suggestion is not put in a run.
const notes = await memoryStore.forRun(s, agent.owner, agent.id);
-- const out = await createTask({ s, config, engine, webhooks, principal: { requester: agent.owner, via: `agent:${agent.id}`, project: null }, body: { task: prompt(agent, notes, input), mode: agent.mode, budget: { per_task_usd: Number(agent.budget_per_run_usd) } }, assignmentOrigin: origin, beforeInsert: async () => {
++ const out = await createTask({ s, config, engine, webhooks, principal: { requester: agent.owner, via: `agent:${agent.id}`, project: null }, body: { task: prompt(agent, notes, input), mode: agent.mode, budget: { per_task_usd: Number(agent.budget_per_run_usd) }, ...(level ? { level } : {}) }, assignmentOrigin: origin, assignmentLevel: level, triggerId, beforeInsert: async () => {
const current = await s.db.maybe('SELECT daily_runs_max FROM actor_agents WHERE id = $1 AND owner = $2 FOR UPDATE', [agent.id, agent.owner]);
if (!current) return { status: 404, code: 'actor.agent.not_found', detail: 'No such agent.' };
const count = Number(await s.db.value('SELECT count(*) FROM tasks WHERE via = $1 AND created_at >= $2', [`agent:${agent.id}`, `${s.iso().slice(0, 10)}T00:00:00.000Z`]));
ddiff --git a/server/app.js b/server/app.js
iindex f4729e2..1404bae 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -29,6 +29,7 @@ const { createRegistry } = require('./runtime/tools');
const { createLoop } = require('./runtime/loop');
const { createStats } = require('./runtime/stats');
const { createEngine } = require('./tasks/engine');
++const { createTriggers } = require('./triggers/service');
const { createWebhooks } = require('./tasks/webhooks');
const { createChatClient } = require('./chat/client');
const { createLocalSignIn } = require('./auth/local');
@@@ -128,6 +129,10 @@ async function createApp(opts = {}) {
// Network notifications ("tell me when it's done"): off without a client secret, never able to fail a task.
const notifier = opts.notifier || createNotifier({ config, fetchImpl, log });
const engine = createEngine({ config, s, adapters, stream, webhooks, now: s.now, log, models, loop, notifier, registry, checkpoints });
++ // Triggers (plan §12 A-4): a Watch subscription the person owns, its card in the thread, and the run back as an
++ // assignment root at the trigger's level. The engine reports every task state so the card's streak stays true.
++ const triggers = opts.triggers || createTriggers({ s, config, engine, webhooks, stream, now: s.now, log, fetchImpl: providerFetch });
++ engine.setTriggers(triggers);
// Waits (plan A2): a wait is a record, not a running loop. The module owns the rows, the matcher, the loop
// guard and the deadline; the engine parks a run and resumes it when the wait ends.
const waits = opts.waits || require('./waits').createWaits({
@@@ -149,7 +154,7 @@ async function createApp(opts = {}) {
// The person's computers (agents/computers.js): one adapter for the task's still proxy, made here so a test can
// stand Bot/Run in (opts.bot.computers) exactly as the Console pages do.
const computers = (opts.bot && opts.bot.computers) || require('./agents/computers').createComputers({ config, fetchImpl: providerFetch, log });
-- const ctx = { config, s, keys, aiCatalogConfig, sso, stream, adapters, engine, webhooks, principal, projects, log, chatAsks, vibeDelivery, models, registry, loop, notifier, stats, checkpoints, connectors, waits, sharing, computers };
++ const ctx = { config, s, keys, aiCatalogConfig, sso, stream, adapters, engine, webhooks, principal, projects, log, chatAsks, vibeDelivery, models, registry, loop, notifier, stats, checkpoints, connectors, waits, sharing, computers, triggers };
// Account export and deletion (ADR-033, ./identity/account-data.js): the two tables that hold a person's rows.
// The sender posts to Network's internal export/deletion routes with this service's own client-credentials token;
ddiff --git a/server/config.js b/server/config.js
iindex 777220d..82343ef 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -166,6 +166,9 @@ function load(env = process.env) {
// OpenVibe.Space: the person's own spaces, which their agent builds (plan 2026-10-22 §Z stage Z1).
// Empty turns the space tools off.
spaceUrl: trim(env.ACTOR_SPACE_URL != null ? env.ACTOR_SPACE_URL : (selfHosted ? '' : 'http://127.0.0.1:4940')),
++ // OpenVibe.Watch: where a trigger's subscription lives (plan §12 A-4). The person's own token is
++ // always the authority; empty turns triggers off.
++ watchUrl: trim(env.ACTOR_WATCH_URL != null ? env.ACTOR_WATCH_URL : (selfHosted ? '' : 'http://127.0.0.1:4730')),
},
// Artifacts (Actor v2, stage V2-4): what one task may make. The bytes live in Actor's own PostgreSQL,
ddiff --git a/server/http/api.js b/server/http/api.js
iindex ad35292..067385f 100644
---- a/server/http/api.js
++++ b/server/http/api.js
@@@ -49,9 +49,21 @@ const TASK_ID = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;
const USER_ID = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
function createApi(ctx) {
-- const { config, s, engine, stream, principal, limits, webhooks } = ctx;
++ const { config, s, engine, stream, principal, limits, webhooks, triggers } = ctx;
const r = asyncRouter();
const problem = (req, res, status, code, detail, extra) => contracts.http.sendProblem(res, status, code, { detail, ctx: req.ov, extra });
++ // A trigger belongs to the person in the request, and its Watch calls use that person's own token: the
++ // Bearer the API was called with, or this site's session token. A service acting for a person is not it.
++ const watchToken = (req) => {
++ // Only a person's own credential: an app or a service acting for them (kind 'app') holds its own token,
++ // and Watch must never see that token as the person's.
++ if (!req.principal || req.principal.kind !== 'user') return null;
++ const h = String(req.headers.authorization || '');
++ if (h.startsWith('Bearer ')) return h.slice(7).trim();
++ return req.token || (req.viewer && req.viewer.token) || null;
++ };
++ const triggerFail = (req, res, out) => problem(req, res, (out.error && out.error.status) || 502, (out.error && out.error.code) || 'actor.trigger.failed', (out.error && out.error.detail) || 'The trigger could not be saved.');
++ const triggerOwner = (req) => String(req.principal.requester || '').replace(/^user:/, '');
r.use(express.json({ limit: '64kb' }));
r.use((req, res, next) => { res.set('Cache-Control', 'no-store'); next(); });
@@@ -268,7 +280,20 @@ function createApi(ctx) {
});
r.post('/me/agents/:id/run', principal.requireCapability('actor.agent.run'), owner, limits.budget('actor.agent.run'), async (req, res) => {
const row = await loadAgent(req, res); if (!row) return;
-- const out = await personal.run(ctx, row, (req.body || {}).input || '');
++ const body = req.body || {};
++ // Watch wakes an agent when a trigger fires. That run is origin routine and inherits the trigger's own level,
++ // never more: the trigger was made in a person's thread at that thread's level. A person pressing Run now on
++ // the agent page is still origin person.
++ const viaWatch = Boolean(req.principal && req.principal.kind === 'app' && req.principal.via === 'service:watch');
++ const trigger = viaWatch && triggers
++ ? await triggers.forDelivery(row.owner, { triggerId: body.trigger_id, watchId: body.watch_id })
++ : null;
++ if (trigger && trigger.state === 'paused') return problem(req, res, 409, 'actor.trigger.paused', 'This trigger is paused, so nothing ran. Resume it on its card.');
++ const out = await personal.run(ctx, row, body.input || '', {
++ origin: viaWatch ? 'routine' : 'person',
++ level: trigger ? trigger.level : null,
++ triggerId: trigger ? trigger.id : null,
++ });
if (out.code) return problem(req, res, out.status, out.code, out.detail);
res.set('Location', `/api/v1/tasks/${out.task.id}`);
res.status(201).json(await wireTask(out.task));
@@@ -405,7 +430,7 @@ function createApi(ctx) {
};
r.post('/tasks', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), async (req, res) => {
-- const out = await createTask({ s, config, engine, webhooks, principal: req.principal, body: req.body || {} });
++ const out = await createTask({ s, config, engine, webhooks, principal: req.principal, body: req.body || {}, personToken: watchToken(req), timezone: req.cookies && req.cookies.ov_tz ? String(req.cookies.ov_tz).slice(0, 64) : null });
if (out.code) {
if (out.retryAfter) res.set('Retry-After', String(out.retryAfter));
return problem(req, res, out.status, out.code, out.detail);
@@@ -565,6 +590,59 @@ function createApi(ctx) {
for (const taskId of taskIds) await engine.cancel(taskId, p.requester);
res.json(result);
});
++
++ // ── Triggers made in a conversation (plan §12 A-4) ───────
++ // The Watch subscription is the runner; these routes are the card's own moves. Everything is the person's:
++ // another person's trigger is the same 404 as one that does not exist, and Watch receives their own token.
++ r.get('/triggers', principal.requireCapability('actor.task.read'), limits.reads('actor.task.read'), personOnly, async (req, res) => {
++ res.json({ triggers: await triggers.list(triggerOwner(req)) });
++ });
++ r.get('/triggers/:id', principal.requireCapability('actor.task.read'), limits.reads('actor.task.read'), personOnly, async (req, res) => {
++ const card = await triggers.get(triggerOwner(req), req.params.id, { token: watchToken(req), live: true });
++ if (!card) return problem(req, res, 404, 'actor.trigger.not_found', 'No such trigger.');
++ res.json({ trigger: card });
++ });
++ r.post('/triggers', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), personOnly, async (req, res) => {
++ const b = req.body || {};
++ const out = await triggers.create({
++ owner: triggerOwner(req), agentId: b.agent_id, task: b.task, kind: b.kind,
++ cron: b.cron, timezone: b.timezone, event: b.event,
++ level: 'ask', turnTaskId: TASK_ID.test(String(b.turn_task_id || '')) ? b.turn_task_id : null,
++ personToken: watchToken(req),
++ });
++ if (out.error) return triggerFail(req, res, out);
++ res.status(201).json({ trigger: out.trigger });
++ });
++ r.patch('/triggers/:id', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), personOnly, async (req, res) => {
++ const out = await triggers.edit(triggerOwner(req), req.params.id, req.body || {}, { token: watchToken(req) });
++ if (out.error) return triggerFail(req, res, out);
++ res.json({ trigger: out.trigger });
++ });
++ r.post('/triggers/:id/pause', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), personOnly, async (req, res) => {
++ const out = await triggers.pause(triggerOwner(req), req.params.id, { token: watchToken(req) });
++ if (out.error) return triggerFail(req, res, out);
++ res.json({ trigger: out.trigger });
++ });
++ r.post('/triggers/:id/resume', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), personOnly, async (req, res) => {
++ const out = await triggers.resume(triggerOwner(req), req.params.id, { token: watchToken(req) });
++ if (out.error) return triggerFail(req, res, out);
++ res.json({ trigger: out.trigger });
++ });
++ r.post('/triggers/:id/run', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), personOnly, async (req, res) => {
++ const out = await triggers.runNow(triggerOwner(req), req.params.id);
++ if (out.error) return triggerFail(req, res, out);
++ res.status(201).json({ task: out.task, trigger: out.trigger });
++ });
++ r.delete('/triggers/:id', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), personOnly, async (req, res) => {
++ const undo = String(req.query.undo || '') === '1' || (req.body && req.body.undo === true);
++ const confirm = String(req.query.confirm || '') === 'delete' || (req.body && req.body.confirm === 'delete');
++ const out = undo
++ ? await triggers.undo(triggerOwner(req), req.params.id, { token: watchToken(req) })
++ : await triggers.remove(triggerOwner(req), req.params.id, { token: watchToken(req), confirm });
++ if (out.error) return triggerFail(req, res, out);
++ res.status(200).json(out);
++ });
++
return r;
}
ddiff --git a/server/http/chat-pages.js b/server/http/chat-pages.js
iindex 4fdb5ec..c8b6a98 100644
---- a/server/http/chat-pages.js
++++ b/server/http/chat-pages.js
@@@ -677,6 +677,9 @@ ${runsSaved ? html`<p class="cm-ok" role="status">Saved. Your next task runs on
const checked = inputs.check(req.body && req.body.files);
if (checked.problem) return fail(422, checked.problem, parent, text);
const out = await createTask({ s, config, engine, principal: { requester: requesterOf(req), project: null }, body: { task: text, mode }, parentId: parent ? parent.id : null,
++ // The person's own token and clock for a trigger made in this turn: their Watch, their timezone.
++ personToken: req.token || (req.viewer && req.viewer.token) || null,
++ timezone: req.cookies && req.cookies.ov_tz ? String(req.cookies.ov_tz).slice(0, 64) : null,
afterInsert: checked.files.length ? (made) => inputs.save(s, made.id, requesterOf(req), checked.files) : null });
if (out.code) return fail(out.status === 503 || out.status === 429 ? out.status : 422, PROBLEM[out.code] || out.detail || 'That could not start. Try again in a moment.', parent, text);
const row = await store.getTask(s, out.task.id);
ddiff --git a/server/identity/account-data.js b/server/identity/account-data.js
iindex 8f72c6a..85ec236 100644
---- a/server/identity/account-data.js
++++ b/server/identity/account-data.js
@@@ -42,6 +42,10 @@ const TABLES = [
// shows. Before the tasks, so the rows are counted before the task's cascade would take them.
{ table: 'actor_waits', subject: 'requester', value: (usr) => `user:${usr}`, file: 'waits.json',
columns: ['id', 'task_id', 'requester', 'step', 'kind', 'until', 'filter', 'deadline_at', 'on_timeout', 'line', 'status', 'created_at', 'updated_at', 'finished_at'] },
++ // The triggers they made in a conversation (plan §12 A-4): the schedule or event source, the agent and task,
++ // and the card's last result. owner is the bare usr_… subject. Their runs hang on it and go with a cascade.
++ { table: 'actor_triggers', subject: 'owner', value: (usr) => usr, file: 'triggers.json',
++ columns: ['id', 'owner', 'agent_id', 'task', 'kind', 'schedule', 'event', 'timezone', 'words', 'level', 'state', 'paused_reason', 'failures', 'watch_id', 'turn_task_id', 'last_task_id', 'last_state', 'last_run_at', 'created_at', 'updated_at'] },
// The tree of delegated work (plan §12 A-2): one row per hand-off, with the agent, task, level and budget
// state. Before tasks, so the rows are counted before a task's cascade could take them.
{ table: 'actor_assignments', subject: 'owner', value: (usr) => `user:${usr}`, file: 'assignments.json',
ddiff --git a/server/runtime/loop.js b/server/runtime/loop.js
iindex c884601..e59fdf2 100644
---- a/server/runtime/loop.js
++++ b/server/runtime/loop.js
@@@ -856,7 +856,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
* a span; and drives on from the step it stopped at. A checkpoint whose last step already answered is delivered
* without another model call.
*/
-- async function continueFrom({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, assign = null, closeAssignment = null, chatPolicy = null, approveChatTool = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, delta = null }) {
++ async function continueFrom({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, assign = null, closeAssignment = null, chatPolicy = null, approveChatTool = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, delta = null, personToken = null, timezone = null, triggers = null, level = null }) {
const source = resume && resume.state ? resume.state : (fork && fork.checkpoint && fork.checkpoint.state);
const restored = restoreState(source);
if (!restored) throw Object.assign(new Error('the saved step could not be read'), { code: 'actor.checkpoint.invalid' });
@@@ -877,7 +877,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
// delivers a saved final answer without calling the model again.
if (fork) state.answered = false;
const ctx = {
-- config, log, models, signal, requester, taskId, meter, budgetLeft, waitForPerson, requestApproval, assign, closeAssignment, chatPolicy, approveChatTool, artifacts,
++ config, log, models, signal, requester, taskId, meter, budgetLeft, waitForPerson, requestApproval, assign, closeAssignment, chatPolicy, approveChatTool, artifacts, personToken, timezone, triggers, level,
store, now, agentId, personText: personText == null ? String(task) : String(personText),
trace: createTracer({ store, taskId, log, now }),
readOutside: () => state.readOutside === true, notifyTask,
@@@ -1014,7 +1014,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
}
}
-- async function run({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, assign = null, closeAssignment = null, chatPolicy = null, approveChatTool = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, attachments = [], delta = null }) {
++ async function run({ task, requester = null, taskId = null, mode = 'balanced', signal, emit, meter, budgetLeft, waitForPerson = null, requestApproval = null, assign = null, closeAssignment = null, chatPolicy = null, approveChatTool = null, context = [], personText = null, agentId = null, notifyTask = null, askPerson = null, control = null, resume = null, fork = null, spent = null, waits = null, waitEvent = null, attachments = [], delta = null, personToken = null, timezone = null, triggers = null, level = null }) {
if (resume || fork) {
return await continueFrom({ task, requester, taskId, mode, signal, emit, meter, budgetLeft, waitForPerson, requestApproval, assign, closeAssignment, chatPolicy, approveChatTool, context, personText, agentId, notifyTask, askPerson, control, resume, fork, spent, waits, waitEvent, delta });
}
@@@ -1040,7 +1040,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
},
};
const ctx = {
-- config, log, models, signal, requester, taskId, meter, budgetLeft, waitForPerson, requestApproval, assign, closeAssignment, chatPolicy, approveChatTool, artifacts,
++ config, log, models, signal, requester, taskId, meter, budgetLeft, waitForPerson, requestApproval, assign, closeAssignment, chatPolicy, approveChatTool, artifacts, personToken, timezone, triggers, level,
store, now, agentId, personText: personText == null ? String(task) : String(personText),
trace: createTracer({ store, taskId, log, now }),
readOutside: () => state.readOutside === true, notifyTask,
ddiff --git a/server/runtime/tools/schedule_trigger.js b/server/runtime/tools/schedule_trigger.js
nnew file mode 100644
iindex 0000000..aee4bad
---- /dev/null
++++ b/server/runtime/tools/schedule_trigger.js
@@@ -0,0 +1,119 @@
++'use strict';
++/**
++ * A trigger made in the conversation (plan §12 A-4): "every weekday at 9, summarise my inbox".
++ *
++ * The sentence is parsed here, deterministically. When it is ambiguous (no time, no timezone, two
++ * times, "every so often", no event it can name) the tool creates nothing and returns the one
++ * question to ask in plain words. When it is clear, the subscription is created in OpenVibe.Watch
++ * with the person's own token, linked to this thread, and the card goes onto the turn as a trigger
++ * step. The model may also pass the fields explicitly (`cron`, `timezone`, `task`, `event`) after the
++ * person answered.
++ */
++const { personOf } = require('./_openvibe');
++const contacts = require('../../agents/contacts');
++const { parseRequest } = require('../../triggers/parse');
++const cron = require('../../triggers/cron');
++
++const EVENT_PATTERN = /^[a-z][a-z0-9_-]*(\.[a-z0-9_*]+){1,}$/;
++const clip = (s, n) => { const t = String(s || '').replace(/\s+/g, ' ').trim(); return t.length > n ? `${t.slice(0, n - 1)}…` : t; };
++
++/** An agent the person named in the sentence, if exactly one of their agents matches its name. */
++async function namedAgent(s, config, owner, text) {
++ const lower = String(text || '').toLowerCase();
++ const list = await contacts.list(s, config, owner).catch(() => []);
++ const matches = list.filter((a) => a && a.name && lower.includes(String(a.name).toLowerCase()));
++ return matches.length === 1 ? matches[0] : null;
++}
++
++module.exports = {
++ name: 'schedule_trigger',
++ description: 'Make something run on a schedule or when an OpenVibe event happens: "every weekday at 9, summarise my inbox" or "when a new issue lands in my repo, triage it". It parses the sentence, creates the Watch subscription as the person, and puts a live trigger card on this thread. If the request is ambiguous it creates nothing and tells you the one question to ask them first (the time, the timezone, what event, or what to do). Money, passwords, deletions and new recipients still always ask when the trigger runs.',
++ scope: 'person',
++ acts: true,
++ costUsd: 0,
++ timeoutMs: 25_000,
++ input: {
++ type: 'object',
++ properties: {
++ request: { type: 'string', description: 'What the person said, in their words, including the schedule or event and the task.' },
++ task: { type: 'string', description: 'The work to do each time, when the person said it separately from the schedule.' },
++ agent_id: { type: 'string', description: 'Which of the person\'s agents runs it (agt_…). Default: the agent of this thread.' },
++ timezone: { type: 'string', description: 'IANA timezone for a schedule (Europe/Berlin). Default: the person\'s own, when known.' },
++ cron: { type: 'string', description: 'A five-field cron, when the person gave an exact schedule after you asked.' },
++ event: {
++ type: 'object',
++ description: 'A Watch event source, when the person named one after you asked.',
++ properties: {
++ pattern: { type: 'string' },
++ filter: { type: 'object', additionalProperties: { type: ['string', 'number', 'boolean', 'null'] } },
++ },
++ required: ['pattern'],
++ },
++ },
++ required: [],
++ },
++ async run(input, ctx) {
++ const owner = personOf(ctx.requester);
++ if (!owner) return { text: JSON.stringify({ error: 'A trigger belongs to a signed-in person, and this run has none.' }), isError: true };
++ if (!ctx.triggers) return { text: JSON.stringify({ error: 'Triggers are not configured on this Actor (OpenVibe.Watch is not connected).' }), isError: true };
++ const said = String(input.request || '').trim();
++ const explicitCron = typeof input.cron === 'string' && input.cron.trim() ? input.cron.trim() : null;
++ const explicitEvent = input.event && typeof input.event === 'object' && EVENT_PATTERN.test(String(input.event.pattern || '')) ? input.event : null;
++
++ let parsed;
++ if (explicitCron || explicitEvent) {
++ const kind = explicitCron ? 'schedule' : 'event';
++ const timezone = String(input.timezone || ctx.timezone || '').trim();
++ const task = String(input.task || said || '').trim();
++ if (!task) return { text: 'Before I make it: what should I do each time?', data: { question: 'What should I do each time?' } };
++ if (kind === 'schedule') {
++ if (!timezone || !cron.isZone(timezone)) return { text: 'Before I make it: which timezone should the schedule use? (For example Europe/Berlin.)', data: { question: 'Which timezone?' } };
++ const verdict = cron.validate(explicitCron, timezone);
++ if (!verdict.valid) return { text: `That schedule cannot run: ${verdict.error}. What time should it run?`, data: { question: `That schedule cannot run: ${verdict.error}. What time should it run?` } };
++ parsed = { status: 'ok', kind, cron: explicitCron, timezone, task, words: require('../../triggers/parse').wordsFor(explicitCron, timezone) };
++ } else {
++ parsed = { status: 'ok', kind, task, event: { pattern: input.event.pattern, filter: input.event.filter || null }, words: `When ${input.event.pattern}` };
++ }
++ } else {
++ parsed = parseRequest(said, { timezone: String(input.timezone || ctx.timezone || '').trim() || null, task: input.task || null });
++ }
++ if (parsed.status !== 'ok') {
++ return { text: `I need one thing before I make it: ${parsed.question}`, data: { needs_you: true, question: parsed.question }, sources: [] };
++ }
++
++ // Which agent runs it: the model's choice, the person's words, or this thread's own agent.
++ let agentId = typeof input.agent_id === 'string' && /^agt_[0-9A-HJKMNP-TV-Z]{26}$/.test(input.agent_id) ? input.agent_id : null;
++ if (!agentId && said) {
++ const named = await namedAgent(ctx.store, ctx.config, owner, said);
++ if (named) agentId = named.id;
++ }
++ if (!agentId) agentId = ctx.agentId || null;
++ if (!agentId) {
++ const vibe = await contacts.ensureVibe(ctx.store, ctx.config, owner).catch(() => null);
++ agentId = vibe ? vibe.id : null;
++ }
++ if (!agentId) return { text: 'Before I make it: which of your agents should run it?', data: { question: 'Which agent?' } };
++
++ const out = await ctx.triggers.create({
++ owner, agentId, task: parsed.task,
++ kind: parsed.kind,
++ cron: parsed.cron || null,
++ timezone: parsed.timezone || null,
++ event: parsed.event || null,
++ level: ctx.level || 'ask',
++ turnTaskId: ctx.taskId || null,
++ personToken: ctx.personToken || null,
++ announce: false,
++ });
++ if (out.error) {
++ return {
++ text: JSON.stringify({ error: out.error.detail, code: out.error.code, ...(out.error.code === 'actor.trigger.sign_in' ? { needs_you: true } : {}) }),
++ data: out.error, isError: true,
++ };
++ }
++ const card = out.trigger;
++ const line = `It is on: ${card.when.words}. ${clip(card.task, 160)}${card.next_run_at ? ` Next run ${card.next_run_at}.` : ''}`;
++ if (ctx.emit) await ctx.emit({ step: 'trigger', text: line, trigger: card });
++ return { text: line, data: { trigger: card } };
++ },
++};
ddiff --git a/server/tasks/create.js b/server/tasks/create.js
iindex cfad569..cf3402d 100644
---- a/server/tasks/create.js
++++ b/server/tasks/create.js
@@@ -15,6 +15,8 @@ const { mintSecret } = require('./webhooks');
const learn = require('../runtime/brain/learn');
const contacts = require('../agents/contacts');
const assignments = require('../agents/assignments');
++const triggerStore = require('../triggers/store');
++const runTokens = require('../triggers/tokens');
/**
* The sharing system (design 2026-11-06), injected once by server/app.js. A run is its owner's until the Share sheet
@@@ -31,10 +33,17 @@ function untilMidnight(nowMs) {
return Math.ceil((Date.UTC(n.getUTCFullYear(), n.getUTCMonth(), n.getUTCDate() + 1) - n.getTime()) / 1000);
}
--async function createTask({ s, config, engine, principal, body, webhooks: webhookQueue = null, beforeInsert = null, chatContext = null, afterInsert = null, parentId = null, forkOfStep = null, forkChange = null, assignmentOrigin = null }) {
++async function createTask({ s, config, engine, principal, body, webhooks: webhookQueue = null, beforeInsert = null, chatContext = null, afterInsert = null, parentId = null, forkOfStep = null, forkChange = null, assignmentOrigin = null, assignmentLevel = null, triggerId = null, personToken = null, timezone = null }) {
const v = contracts.validate('actor.task-create-request@1', body);
if (!v.valid) return { status: 422, code: 'actor.task.invalid', detail: (v.errors || []).map((e) => `${e.path || e.instancePath || ''} ${e.message}`.trim()).join('; ').slice(0, 500) || 'invalid request' };
const tier = config.allowance;
++ // A trigger's run is linked to its card. The trigger must belong to the same person the task is for; the
++ // caller (Watch's delivery or the Run now route) already checked, and this check makes it true for every path.
++ let trigger = null;
++ if (triggerId != null) {
++ trigger = await triggerStore.get(s, String(principal.requester).replace(/^user:/, ''), triggerId);
++ if (!trigger) return { status: 404, code: 'actor.trigger.not_found', detail: 'No such trigger.' };
++ }
// The contact this thread runs as (A-1). A person's task defaults to their Vibe, created on first read; a personal
// agent's own run keeps that agent. The thread's level may raise the agent's, never lower it.
let agentId = typeof body.agent_id === 'string' ? body.agent_id : null;
@@@ -117,14 +126,16 @@ async function createTask({ s, config, engine, principal, body, webhooks: webhoo
// engine starts it only after this commits, so the run never begins without them.
// Every hand-off is one node (plan §12 A-2). The root is made with the task, so the tree exists from
// its first step; an older task without one gets it lazily (assignments.ensureForTask).
-- const node = await assignments.ensureForTask(s, config, row, { origin: assignmentOrigin });
++ const node = await assignments.ensureForTask(s, config, row, { origin: assignmentOrigin, level: assignmentLevel });
const task = node ? { ...row, assignment_id: node.id } : row;
++ if (trigger) await triggerStore.addRun(s, trigger.id, task.id, 'queued', s.now());
if (afterInsert) await afterInsert(task);
if (chatContext) await s.db.query('INSERT INTO actor_chat_tasks (task_id, agent_id, policy, conversation) VALUES ($1, $2, $3, $4)', [task.id, chatContext.agentId, chatContext.policy, chatContext.conversation]);
return { status: 201, task };
});
// The engine starts the task only once its row has committed; a webhook that asked for `queued` hears it now.
if (out.status === 201) {
++ if (personToken) runTokens.put(out.task.id, { token: personToken, timezone });
if (sharing) {
try {
const audience = await sharing.defaultOf(out.task.requester).catch(() => 'private');
ddiff --git a/server/tasks/engine.js b/server/tasks/engine.js
iindex e4cabab..7339405 100644
---- a/server/tasks/engine.js
++++ b/server/tasks/engine.js
@@@ -31,6 +31,7 @@ const { check: checkClaims } = require('../runtime/check');
const approvals = require('../agents/approvals');
const contacts = require('../agents/contacts');
const assignments = require('../agents/assignments');
++const runTokens = require('../triggers/tokens');
const levels = require('../agents/levels');
const { repoIn: codesRepoIn } = require('../agents/codes');
const memory = require('../memory');
@@@ -45,7 +46,7 @@ const LOOP_AGENT = 'openvibe-runtime';
// What a task asked in a public chat may be (chat/asks.js): nothing that uses the person's own computers, robots or code.
const PUBLIC_CLASSES = ['answer', 'lookup', 'research', 'web'];
--function createEngine({ config, s, adapters, stream, webhooks = null, now = () => Date.now(), log = console, models = null, loop = null, notifier = null, registry = null, checkpoints = null, waits: waitsArg = null }) {
++function createEngine({ config, s, adapters, stream, webhooks = null, now = () => Date.now(), log = console, models = null, loop = null, notifier = null, registry = null, checkpoints = null, waits: waitsArg = null, triggers: triggersArg = null }) {
const running = new Map(); // id → AbortController
const waiting = [];
// A question the loop asked before any tool ran (stage V2-8): the task waits in needs_you, the page shows the
@@@ -55,6 +56,9 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
// resumes. One per task: the task is queued again and the payload is taken when it starts.
const wakes = new Map();
let waits = waitsArg;
++ // Triggers are created after the engine (they need it to run now); the app sets them once.
++ let triggers = triggersArg;
++ const setTriggers = (t) => { triggers = t || null; };
const controls = new Map(); // task id → the run's control object (live assumptions)
let stopped = false;
@@@ -148,6 +152,9 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
await emit(id, { kind: ended ? 'end' : 'state', state, ...(agentForEvent ? { agent: agentForEvent } : {}) });
// The assignment tree follows the task it runs (plan §12 A-2), and its root's spend is written with it.
if (row) await assignments.syncTaskState(s, row).catch((err) => log.warn && log.warn('[Actor] assignment state:', id, (err && err.message) || err));
++ // A trigger's card follows its run: a failure streak grows, five in a row pause it, a success clears it.
++ if (row && triggers) await triggers.taskState(row).catch((err) => log.warn && log.warn('[Actor] trigger state:', id, (err && err.message) || err));
++ if (ended) runTokens.drop(id);
if (ended) stream.end(id);
// "Tell me when it's done" (stage V2-6): done or failed, once per task. Cancelled is the person's own action.
if (ended && row && (state === 'succeeded' || state === 'failed')) await tellPerson(row, state);
@@@ -827,7 +834,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
const status = () => ({ running: running.size, waiting: waiting.length });
-- return { submit, cancel, stop, status, answer, assume, steer, resumeInterrupted, resumeWithEvent, resumeWithNotice, failTask, notifyWaitTimeout, setWaits, MAX_ATTEMPTS };
++ return { submit, cancel, stop, status, answer, assume, steer, resumeInterrupted, resumeWithEvent, resumeWithNotice, failTask, notifyWaitTimeout, setWaits, setTriggers, MAX_ATTEMPTS };
}
module.exports = { createEngine, MAX_ATTEMPTS };
ddiff --git a/server/triggers/cron.js b/server/triggers/cron.js
nnew file mode 100644
iindex 0000000..b83d906
---- /dev/null
++++ b/server/triggers/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/triggers/parse.js b/server/triggers/parse.js
nnew file mode 100644
iindex 0000000..de93889
---- /dev/null
++++ b/server/triggers/parse.js
@@@ -0,0 +1,232 @@
++'use strict';
++/**
++ * "Every weekday at 9, summarise my inbox" → a schedule and the task. "When a new issue lands in my
++ * repo, triage it" → a Watch event source and the task. This is the deterministic part of
++ * schedule_trigger: the same sentence always parses the same way, and anything it cannot pin down
++ * comes back as a question instead of a trigger. The model may ask the person and call again with
++ * the explicit fields (`cron`, `timezone`, `task`, `agent_id`, `event`).
++ *
++ * A missing timezone is the person's own zone when the caller knows it (`timezone`), never a guess;
++ * a missing time is a question, not a default. The five-field cron is the machine form, and
++ * `words` is the sentence the card shows.
++ */
++const cron = require('./cron');
++
++const DAY_NAMES = ['Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday'];
++const DAY_INDEX = { sunday: 0, sun: 0, monday: 1, mon: 1, tuesday: 2, tue: 2, tues: 2, wednesday: 3, wed: 3, thursday: 4, thu: 4, thur: 4, thurs: 4, friday: 5, fri: 5, saturday: 6, sat: 6 };
++const ZONE_CITIES = {
++ berlin: 'Europe/Berlin', london: 'Europe/London', paris: 'Europe/Paris', madrid: 'Europe/Madrid', rome: 'Europe/Rome',
++ amsterdam: 'Europe/Amsterdam', brussels: 'Europe/Brussels', vienna: 'Europe/Vienna', zurich: 'Europe/Zurich',
++ stockholm: 'Europe/Stockholm', oslo: 'Europe/Oslo', copenhagen: 'Europe/Copenhagen', helsinki: 'Europe/Helsinki',
++ warsaw: 'Europe/Warsaw', prague: 'Europe/Prague', lisbon: 'Europe/Lisbon', dublin: 'Europe/Dublin', athens: 'Europe/Athens',
++ moscow: 'Europe/Moscow', istanbul: 'Europe/Istanbul', dubai: 'Asia/Dubai', mumbai: 'Asia/Kolkata', delhi: 'Asia/Kolkata',
++ singapore: 'Asia/Singapore', tokyo: 'Asia/Tokyo', seoul: 'Asia/Seoul', shanghai: 'Asia/Shanghai', beijing: 'Asia/Shanghai',
++ hongkong: 'Asia/Hong_Kong', sydney: 'Australia/Sydney', melbourne: 'Australia/Melbourne', auckland: 'Pacific/Auckland',
++ 'new york': 'America/New_York', nyc: 'America/New_York', boston: 'America/New_York', washington: 'America/New_York',
++ chicago: 'America/Chicago', denver: 'America/Denver', 'los angeles': 'America/Los_Angeles', 'san francisco': 'America/Los_Angeles',
++ seattle: 'America/Los_Angeles', toronto: 'America/Toronto', vancouver: 'America/Vancouver', mexico: 'America/Mexico_City',
++ 'sao paulo': 'America/Sao_Paulo', 'buenos aires': 'America/Argentina/Buenos_Aires', honolulu: 'Pacific/Honolulu',
++ utc: 'UTC', gmt: 'UTC',
++};
++const DAYPART = { morning: [9, 0], afternoon: [14, 0], evening: [19, 0], night: [21, 0] };
++const AMBIGUOUS = /\b(sometimes|occasionally|every so often|from time to time|regularly|whenever i|when i ask|at some point|soon|sometime)\b/i;
++const TIME = String.raw`(\d{1,2})(?::(\d{2}))?\s*(am|pm)?`;
++const WEEKDAY = String.raw`(weekday|weekdays|day|every day|morning|afternoon|evening|night|weekend|weekends|sun(?:day)?|mon(?:day)?|tue(?:sday|s)?|wed(?:nesday)?|thu(?:rsday|rs)?|fri(?:day)?|sat(?:urday)?)`;
++
++const pad = (n) => String(n).padStart(2, '0');
++const clean = (s) => String(s || '').replace(/\s+/g, ' ').trim();
++
++/** A wall-clock hour/minute as cron's two fields. 12am is 0, 12pm is 12. */
++function clock(hourText, minuteText, ampm) {
++ let hour = Number(hourText);
++ const minute = minuteText == null ? 0 : Number(minuteText);
++ if (!Number.isInteger(hour) || hour > 23 || !Number.isInteger(minute) || minute > 59) return null;
++ if (ampm) {
++ if (hour < 1 || hour > 12) return null;
++ hour = hour % 12 + (ampm.toLowerCase() === 'pm' ? 12 : 0);
++ }
++ return { hour, minute };
++}
++
++function zoneFromText(text) {
++ const source = String(text || '');
++ const explicit = source.match(/\b([A-Z][A-Za-z_+-]+(?:\/[A-Z][A-Za-z_+-]+)+)\b/);
++ if (explicit && cron.isZone(explicit[1])) return explicit[1];
++ const inCity = source.match(/\bin\s+([A-Za-z][A-Za-z ]{1,30}?)(?:\s+time)?\b/i) || source.match(/\b([A-Za-z ]{2,30})\s+time\b/i);
++ if (inCity) {
++ const key = inCity[1].trim().toLowerCase();
++ if (ZONE_CITIES[key]) return ZONE_CITIES[key];
++ }
++ const lower = source.toLowerCase();
++ for (const [city, zone] of Object.entries(ZONE_CITIES)) {
++ if (new RegExp(`\\b${city.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')}\\b`).test(lower)) return zone;
++ }
++ return null;
++}
++
++function wordsFor(cronExpr, timezone) {
++ try {
++ const rule = cron.parse(cronExpr);
++ const minutes = [...rule.minutes].sort((a, b) => a - b);
++ const hours = [...rule.hours].sort((a, b) => a - b);
++ const one = (set) => set.size === 1;
++ if (one(rule.minutes) && one(rule.hours) && rule.days.size === 1 && rule.months.size === 1 && rule.weekdays.size === 1 && rule.dayAny === false && rule.weekdayAny === false) {
++ return `On day ${[...rule.days][0]} of month ${[...rule.months][0]} at ${pad([...hours][0])}:${pad([...minutes][0])} (${timezone})`;
++ }
++ if (one(rule.minutes) && one(rule.hours) && rule.months.size === 12) {
++ const time = `${pad([...hours][0])}:${pad([...minutes][0])}`;
++ if (rule.weekdayAny && rule.dayAny) return `Every day at ${time} (${timezone})`;
++ if (rule.dayAny && !rule.weekdayAny) {
++ const days = [...rule.weekdays].sort((a, b) => a - b);
++ if (days.join(',') === '1,2,3,4,5') return `Every weekday at ${time} (${timezone})`;
++ if (days.join(',') === '0,6') return `Every weekend at ${time} (${timezone})`;
++ if (days.length === 1) return `Every ${DAY_NAMES[days[0]]} at ${time} (${timezone})`;
++ return `Every ${days.map((d) => DAY_NAMES[d]).join(', ')} at ${time} (${timezone})`;
++ }
++ }
++ if (one(rule.minutes) && one(rule.hours) && rule.hours.has(0) && rule.minutes.has(0) && rule.months.size === 12 && rule.days.size === 31 && rule.weekdays.size === 7) return `Every hour (${timezone})`;
++ if (rule.hours.size === 24 && rule.minutes.size === 60) return `Every minute (${timezone})`;
++ if (one(rule.hours) && rule.hours.has(0) && rule.minutes.size === 60) return `Every hour (${timezone})`;
++ if (rule.hours.size === 24 && rule.minutes.size === 1 && !rule.minutes.has(0)) return `Every ${[...rule.minutes][0]} minutes (${timezone})`;
++ if (rule.hours.size === 24 && rule.minutes.size === 1 && [...rule.minutes][0] % 15 === 0 && [...rule.minutes][0] !== 0) return `Every ${[...rule.minutes][0]} minutes (${timezone})`;
++ if (rule.minutes.size === 60 && rule.hours.size < 24) {
++ const step = 24 / rule.hours.size;
++ if (Number.isInteger(step) && rule.hours.has(0)) return `Every ${step} hour${step === 1 ? '' : 's'} (${timezone})`;
++ }
++ } catch { /* fall through */ }
++ return `Cron ${cronExpr} (${timezone})`;
++}
++
++/** The schedule phrase and its cron, or null. Longest match wins so "every weekday at 9" beats "every day". */
++function scheduleIn(text) {
++ const source = clean(text);
++ const lower = source.toLowerCase();
++ const patterns = [];
++
++ // every weekday at 9 / every weekday at 9:30 pm
++ patterns.push({ re: new RegExp(String.raw`\bevery\s+(weekdays?|weekends?)\s+at\s+${TIME}\b`, 'i'), build: (m) => {
++ const c = clock(m[2], m[3], m[4]); if (!c) return null;
++ return { cron: `${c.minute} ${c.hour} * * ${/weekend/i.test(m[1]) ? '0,6' : '1-5'}` };
++ } });
++ // every Monday at 10 / every Sunday evening
++ patterns.push({ re: new RegExp(String.raw`\bevery\s+(${Object.keys(DAY_INDEX).join('|')})\b(?:\s+(?:at\s+${TIME}|(morning|afternoon|evening|night)))?`, 'i'), build: (m) => {
++ let c = m[2] != null ? clock(m[2], m[3], m[4]) : (m[5] ? { hour: DAYPART[m[5].toLowerCase()][0], minute: 0 } : null);
++ if (!c) return null;
++ return { cron: `${c.minute} ${c.hour} * * ${DAY_INDEX[m[1].toLowerCase()]}` };
++ } });
++ // every day at 8 / daily at 8 / every morning
++ patterns.push({ re: new RegExp(String.raw`\b(?:every\s+day|daily)\s+at\s+${TIME}\b`, 'i'), build: (m) => {
++ const c = clock(m[1], m[2], m[3]); if (!c) return null;
++ return { cron: `${c.minute} ${c.hour} * * *` };
++ } });
++ patterns.push({ re: new RegExp(String.raw`\bevery\s+(morning|afternoon|evening|night)\b`, 'i'), build: (m) => {
++ const [hour, minute] = DAYPART[m[1].toLowerCase()];
++ return { cron: `${minute} ${hour} * * *` };
++ } });
++ // every 15 minutes / every 2 hours / every hour
++ patterns.push({ re: /\bevery\s+(\d{1,3})\s*(minutes?|mins?|hours?|hrs?)\b/i, build: (m) => {
++ const n = Number(m[1]);
++ if (/^h/i.test(m[2])) return { cron: `0 */${n} * * *` };
++ return { cron: `*/${n} * * * *` };
++ } });
++ patterns.push({ re: /\bevery\s+hour\b/i, build: () => ({ cron: '0 * * * *' }) });
++ // weekdays at 9 (without "every")
++ patterns.push({ re: new RegExp(String.raw`\b(weekdays?)\s+at\s+${TIME}\b`, 'i'), build: (m) => {
++ const c = clock(m[2], m[3], m[4]); if (!c) return null;
++ return { cron: `${c.minute} ${c.hour} * * 1-5` };
++ } });
++
++ let best = null;
++ for (const p of patterns) {
++ const m = lower.match(p.re);
++ if (!m) continue;
++ const built = p.build(m);
++ if (!built) continue;
++ if (!best || m[0].length > best.matchLength) best = { ...built, matchStart: m.index, matchLength: m[0].length, phrase: source.slice(m.index, m.index + m[0].length) };
++ }
++ return best;
++}
++
++/** A Watch event source from "when …", or null. */
++function eventIn(text) {
++ const source = clean(text);
++ const m = source.match(/\bwhen\s+(.+?)(?:,|$)/i);
++ if (!m) return null;
++ const what = m[1].trim();
++ const lower = what.toLowerCase();
++ let pattern = null;
++ let words = what;
++ if (/\b(issue|ticket)\b/.test(lower)) pattern = 'github.issue.opened';
++ else if (/\b(pull request|merge request|pr)\b/.test(lower)) pattern = 'github.pull_request.opened';
++ else if (/\b(comment|reply)\b/.test(lower)) pattern = 'github.comment.created';
++ else if (/\b(goes live|went live|stream|broadcast)\b/.test(lower)) pattern = 'live.stream.started';
++ else if (/\b(blog|article|post)\b/.test(lower) && /\b(new|publish|posted)\b/.test(lower)) pattern = 'blog.post.published';
++ else if (/\b(wiki|page)\b/.test(lower) && /\b(publish|published|new)\b/.test(lower)) pattern = 'wiki.page.published';
++ else if (/\b(release|deploy)\b/.test(lower)) pattern = 'host.release.published';
++ else if (/\b(email|mail|message)\b/.test(lower)) pattern = 'mail.message.received';
++ else if (/\b(source|feed)\b/.test(lower)) pattern = 'sources.item.created';
++ if (!pattern) return { ambiguous: true, what, reason: 'I can watch a new issue or pull request, a blog post, a wiki page, a live stream, a release, mail or a source, but I could not tell which of those this is.' };
++ const repo = what.match(/\b([a-z0-9_.-]+\/[a-z0-9_.-]+)\b/i);
++ const filter = repo ? { repo: repo[1] } : null;
++ // Drop the trailing location ("in my repo") from the sentence the card shows.
++ words = what.replace(/\s+in\s+my\s+repo(?:sitory)?\b/i, '').replace(/\s+in\s+.+$/i, '').trim() || what;
++ return { pattern, filter, words: `When ${words}`, matchStart: m.index, matchLength: m[0].length, phrase: m[0] };
++}
++
++/** The task after (or before) the matched clause, with the clause taken out. */
++function taskIn(text, match) {
++ const source = clean(text);
++ if (!match) return '';
++ const before = clean(source.slice(0, match.matchStart).replace(/[,;:.\s]+$/, ''));
++ const after = clean(source.slice(match.matchStart + match.matchLength).replace(/^[,;:.\s]+/, ''));
++ const pick = after || before;
++ return pick.replace(/^(?:then|and|please|to)\s+/i, '').trim();
++}
++
++/**
++ * The person's sentence → { status: 'ok', kind, cron/timezone or event, task, words } or
++ * { status: 'ask', question }. `timezone` is the person's zone when the caller knows it.
++ */
++function parseRequest(text, { timezone: knownZone = null, task: givenTask = null } = {}) {
++ const source = clean(text);
++ if (!source) return { status: 'ask', question: 'What should run, and when?' };
++ if (AMBIGUOUS.test(source)) {
++ return { status: 'ask', question: 'How often should this run, exactly? For example: every weekday at 9, or every hour.' };
++ }
++ const schedule = scheduleIn(source);
++ const event = eventIn(source);
++ if (schedule && event) {
++ return { status: 'ask', question: 'Should this run on a schedule or when an OpenVibe event arrives? You said both.' };
++ }
++ if (event && event.ambiguous) {
++ return { status: 'ask', question: `${event.reason} What exactly should wake it?` };
++ }
++ if (!schedule && !event) {
++ if (/\bevery\b|\bdaily\b|\bweekly\b|\bwhen\b/i.test(source)) {
++ return { status: 'ask', question: 'When exactly should this run? For example: every weekday at 9 in your timezone, or when a new issue lands in your repo.' };
++ }
++ return { status: 'ask', question: 'Should this run on a schedule (every weekday at 9) or when something happens (a new issue, a blog post, a release)?' };
++ }
++ const task = clean(givenTask || taskIn(source, schedule || event)).slice(0, 8000);
++ if (!task) return { status: 'ask', question: 'What should I do each time this runs?' };
++ if (/\b(at\s+\d{1,2}(?::\d{2})?\s*(?:am|pm)?)\b.*\bor\b.*\bat\s+\d{1,2}/i.test(source)) {
++ return { status: 'ask', question: 'Which of those times should it run? Give me one time and the timezone.' };
++ }
++ if (schedule) {
++ const zone = knownZone || zoneFromText(source);
++ if (!zone) return { status: 'ask', question: `Every ${schedule.phrase.replace(/^every\s+/i, '')} in which timezone?` };
++ const verdict = cron.validate(schedule.cron, zone);
++ if (!verdict.valid) return { status: 'ask', question: `That schedule cannot run: ${verdict.error}. What time should it run?` };
++ return {
++ status: 'ok', kind: 'schedule', cron: schedule.cron, timezone: zone, task,
++ words: wordsFor(schedule.cron, zone),
++ };
++ }
++ return {
++ status: 'ok', kind: 'event', task,
++ event: { pattern: event.pattern, filter: event.filter },
++ words: event.words,
++ };
++}
++
++module.exports = { parseRequest, scheduleIn, eventIn, zoneFromText, wordsFor, AMBIGUOUS };
ddiff --git a/server/triggers/service.js b/server/triggers/service.js
nnew file mode 100644
iindex 0000000..48027b7
---- /dev/null
++++ b/server/triggers/service.js
@@@ -0,0 +1,351 @@
++'use strict';
++/**
++ * Triggers made in a conversation (plan §12 A-4). The subscription is a Watch the person owns; this
++ * module keeps the card's record, turns runs back into assignment roots and owns the person's four
++ * moves (edit, pause, resume, undo/delete) and Run now.
++ *
++ * A routine is one Watch watch, created with the person's own token: paused, given its cron (or
++ * event source), then resumed, so it never fires between the calls. The Watch action is actor-agent:
++ * Watch wakes the agent, Actor makes the run origin 'routine' at the trigger's stored level, and the
++ * assignment tree caps it there. Money, passwords, deletions and new recipients still go through the
++ * always-ask floor inside the run.
++ *
++ * Five failed runs in a row pause the trigger here; Watch's own five failed checks pause the watch
++ * (server/check.js) and are mirrored onto the card when it is next read.
++ */
++const cron = require('./cron');
++const store = require('./store');
++const watchClient = require('./watch');
++const parse = require('./parse');
++const contacts = require('../agents/contacts');
++const levels = require('../agents/levels');
++const personal = require('../agents/personal');
++
++const ID = /^trg_[0-9A-HJKMNP-TV-Z]{26}$/;
++const WATCH_ID = /^wch_[0-9A-HJKMNP-TV-Z]{26}$/;
++const TASK_ID = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;
++const EVENT_PATTERN = /^[a-z][a-z0-9_-]*(\.[a-z0-9_*]+){1,}$/;
++const MAX_ACTIVE = 20;
++const UNDO_MS = 10 * 60_000;
++const MAX_TASK = 8000;
++
++const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
++const clip = (s, n) => { const t = String(s || '').replace(/\s+/g, ' ').trim(); return t.length > n ? `${t.slice(0, n - 1)}…` : t; };
++const problem = (status, code, detail) => ({ status, code, detail });
++
++function createTriggers({ s, config, engine = null, webhooks = null, stream = null, watch = null, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console }) {
++ const client = watch || watchClient.createWatchClient({ config, fetchImpl });
++ const nowMs = () => (typeof s.now === 'function' ? s.now() : now());
++
++ const requesterOf = (owner) => (String(owner || '').startsWith('user:') ? String(owner) : `user:${owner}`);
++ const subjectOf = (owner) => String(owner || '').replace(/^user:/, '');
++ async function agentOf(owner, id) {
++ if (!id || !/^agt_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(id))) return null;
++ return await contacts.get(s, requesterOf(owner), id);
++ }
++
++ /** The card (actor.trigger@1), from the trigger's own row and, when it can be read, Watch's health. */
++ async function card(row, { token = null, live = false } = {}) {
++ if (!row) return null;
++ const agent = await agentOf(row.owner, row.agent_id).catch(() => null);
++ let next = null;
++ if (row.state === 'active' && row.kind === 'schedule' && row.schedule && row.schedule.cron) {
++ try { const at = cron.next(row.schedule.cron, row.schedule.timezone || row.timezone || 'UTC', nowMs()); if (at) next = iso(at); } catch { /* no next */ }
++ }
++ const out = {
++ id: row.id,
++ owner: row.owner,
++ agent_id: row.agent_id,
++ agent_name: agent ? agent.name : null,
++ task: row.task,
++ kind: row.kind,
++ when: {
++ words: row.words,
++ cron: row.kind === 'schedule' && row.schedule ? row.schedule.cron : null,
++ timezone: row.kind === 'schedule' ? (row.timezone || (row.schedule && row.schedule.timezone) || null) : null,
++ event: row.kind === 'event' && row.event ? { pattern: row.event.pattern, filter: row.event.filter || null } : null,
++ },
++ next_run_at: next,
++ last_run: row.last_task_id && row.last_state && row.last_run_at
++ ? { task_id: row.last_task_id, state: row.last_state, at: iso(row.last_run_at) }
++ : null,
++ state: row.state === 'paused' ? 'paused' : 'active',
++ paused_reason: row.paused_reason || null,
++ failures: Number(row.failures) || 0,
++ watch: row.watch_id ? { id: row.watch_id, status: 'active' } : null,
++ undo_until: iso(Number(row.created_at) + UNDO_MS),
++ created_at: iso(row.created_at),
++ updated_at: iso(row.updated_at),
++ };
++ if (live && token && row.watch_id && WATCH_ID.test(row.watch_id)) {
++ const got = await client.get(token, row.watch_id).catch(() => null);
++ if (got && got.ok && got.watch) {
++ const health = got.watch.health || {};
++ out.watch = {
++ id: row.watch_id,
++ status: got.watch.status,
++ last_state: health.last_state || null,
++ consecutive_failures: Number(health.consecutive_failures) || 0,
++ next_due_at: health.next_due_at || null,
++ };
++ if (out.state === 'active' && got.watch.status !== 'active' && Number(health.consecutive_failures) >= 5) {
++ const paused = await store.update(s, row.owner, row.id, { state: 'paused', paused_reason: 'failed', updated_at: nowMs() });
++ out.state = 'paused';
++ out.paused_reason = 'failed';
++ out.updated_at = iso(paused.updated_at);
++ }
++ }
++ }
++ return out;
++ }
++
++ async function emit(row, text) {
++ if (!row || !row.turn_task_id || !TASK_ID.test(String(row.turn_task_id))) return;
++ try {
++ const cardNow = await card(row);
++ const event = { at: new Date(nowMs()).toISOString(), kind: 'output', step: 'trigger', agent: 'openvibe-runtime', text: clip(text, 500), trigger: cardNow };
++ const saved = await require('../tasks/store').appendEvent(s, row.turn_task_id, event);
++ if (stream && typeof stream.publish === 'function') stream.publish(row.turn_task_id, saved);
++ } catch (err) {
++ log.warn && log.warn('[Actor] trigger card:', row.id, (err && err.message) || err);
++ }
++ }
++
++ function validate({ owner, agentId, task, kind, cronExpr, timezone, event }) {
++ if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(owner || ''))) return problem(401, 'actor.trigger.sign_in', 'Sign in to make a trigger.');
++ if (!agentId) return problem(422, 'actor.trigger.agent', 'Choose which of your agents should run it.');
++ if (typeof task !== 'string' || !task.trim() || task.length > MAX_TASK) return problem(422, 'actor.trigger.task', 'Say what the agent should do each time, in a sentence.');
++ if (!['schedule', 'event'].includes(kind)) return problem(422, 'actor.trigger.kind', 'A trigger is a schedule or a Watch event.');
++ if (kind === 'schedule') {
++ if (!timezone || !cron.isZone(timezone)) return problem(422, 'actor.trigger.timezone', `Unknown time zone "${timezone}". Say the city or the IANA name.`);
++ const verdict = cron.validate(cronExpr, timezone);
++ if (!verdict.valid) return problem(422, 'actor.trigger.schedule', verdict.error);
++ }
++ if (kind === 'event') {
++ if (!event || !EVENT_PATTERN.test(String(event.pattern || ''))) return problem(422, 'actor.trigger.event', 'An event source is an Events topic pattern such as github.issue.opened.');
++ if (event.filter != null && (typeof event.filter !== 'object' || Array.isArray(event.filter))) return problem(422, 'actor.trigger.event', 'event.filter must be an object of field values.');
++ }
++ return null;
++ }
++
++ async function create({ owner, agentId, task, kind, cron: cronExpr = null, timezone = null, event = null, level = null, turnTaskId = null, personToken = null, announce = true }) {
++ const bad = validate({ owner, agentId, task, kind, cronExpr, timezone, event });
++ if (bad) return { error: bad };
++ const agent = await agentOf(owner, agentId);
++ if (!agent) return { error: problem(422, 'actor.agent.not_found', 'agent_id names one of your own agents (GET /api/v1/agents).') };
++ if (!client.enabled) return { error: problem(503, 'actor.trigger.watch_unavailable', 'OpenVibe.Watch is not configured on this Actor, so it cannot run a trigger.') };
++ if (!personToken) return { error: problem(401, 'actor.trigger.sign_in', 'Sign in again: creating a trigger needs your own OpenVibe session, not a service token.') };
++ const active = await store.countActive(s, owner);
++ if (active >= MAX_ACTIVE) return { error: problem(409, 'actor.trigger.limit', `You already have ${MAX_ACTIVE} active triggers. Pause one, or delete one, before making another.`) };
++ const at = nowMs();
++ const row = await store.insert(s, {
++ id: s.newId('trg'), owner, agent_id: agent.id, task: String(task).trim(), kind,
++ schedule: kind === 'schedule' ? { cron: cronExpr, timezone } : null,
++ event: kind === 'event' ? { pattern: event.pattern, filter: event.filter || null } : null,
++ timezone: kind === 'schedule' ? timezone : null,
++ words: kind === 'schedule' ? parse.wordsFor(cronExpr, timezone) : (event.words || `When ${event.pattern}`),
++ level: levels.LEVELS.includes(level) ? level : 'ask',
++ state: 'paused', turn_task_id: turnTaskId, created_at: at,
++ });
++ let made = null;
++ try {
++ const body = {
++ name: `Actor trigger: ${clip(String(task).split('\n')[0], 80)}`,
++ status: 'paused',
++ source: kind === 'event' ? { kind: 'event', pattern: event.pattern, filter: event.filter || null } : { kind: 'webhook' },
++ cadence: null,
++ extraction: { kind: 'text' },
++ condition: { op: 'exists' },
++ action: [{ kind: 'actor-agent', agent_id: agent.id, note: String(task).slice(0, 2000) }],
++ labels: { actor_trigger: row.id },
++ };
++ const created = await client.create(personToken, body);
++ if (!created.ok) throw created;
++ made = created.watch;
++ if (kind === 'schedule') {
++ const scheduled = await client.setSchedule(personToken, made.id, cronExpr, timezone);
++ if (!scheduled.ok) throw scheduled;
++ }
++ const linked = await store.update(s, owner, row.id, { watch_id: made.id, updated_at: nowMs() });
++ const resumed = await client.resume(personToken, made.id);
++ if (!resumed.ok) throw resumed;
++ const activeRow = await store.update(s, owner, row.id, { state: 'active', paused_reason: null, updated_at: nowMs() });
++ const cardNow = await card(activeRow);
++ if (announce) await emit(linked, `Trigger created: ${cardNow.when.words}.`);
++ return { trigger: cardNow, row: activeRow };
++ } catch (err) {
++ if (made && made.id) await client.remove(personToken, made.id).catch(() => null);
++ await store.remove(s, owner, row.id).catch(() => null);
++ const why = err && err.code ? err : problem(502, 'actor.trigger.watch_failed', `OpenVibe.Watch could not save the subscription: ${(err && (err.detail || err.message)) || 'unknown error'}`);
++ return { error: why.status ? why : problem(502, why.code || 'actor.trigger.watch_failed', why.detail || 'OpenVibe.Watch could not save the subscription.') };
++ }
++ }
++
++ async function get(owner, id, { token = null, live = false } = {}) {
++ if (!ID.test(String(id || ''))) return null;
++ const row = await store.get(s, owner, id);
++ if (!row) return null;
++ return await card(row, { token, live });
++ }
++
++ async function list(owner) {
++ const rows = await store.list(s, owner);
++ return await Promise.all(rows.map((row) => card(row)));
++ }
++
++ async function edit(owner, id, patch = {}, { token = null } = {}) {
++ const row = ID.test(String(id || '')) ? await store.get(s, owner, id) : null;
++ if (!row) return { error: problem(404, 'actor.trigger.not_found', 'No such trigger.') };
++ if (!token) return { error: problem(401, 'actor.trigger.sign_in', 'Sign in again: changing a trigger uses your own OpenVibe session.') };
++ const next = {
++ agent_id: patch.agent_id !== undefined ? patch.agent_id : row.agent_id,
++ task: patch.task !== undefined ? String(patch.task).trim() : row.task,
++ kind: patch.kind !== undefined ? patch.kind : row.kind,
++ cron: patch.cron !== undefined ? patch.cron : (row.schedule && row.schedule.cron),
++ timezone: patch.timezone !== undefined ? patch.timezone : (row.timezone || (row.schedule && row.schedule.timezone) || 'UTC'),
++ event: patch.event !== undefined ? patch.event : row.event,
++ level: row.level,
++ };
++ const bad = validate({ owner, agentId: next.agent_id, task: next.task, kind: next.kind, cronExpr: next.cron, timezone: next.timezone, event: next.event });
++ if (bad) return { error: bad };
++ const agent = await agentOf(owner, next.agent_id);
++ if (!agent) return { error: problem(422, 'actor.agent.not_found', 'agent_id names one of your own agents.') };
++ if (!row.watch_id) return { error: problem(409, 'actor.trigger.no_watch', 'This trigger has no subscription to edit.') };
++ const changes = {};
++ if (next.task !== row.task || next.agent_id !== row.agent_id) {
++ changes.action = [{ kind: 'actor-agent', agent_id: next.agent_id, note: next.task.slice(0, 2000) }];
++ changes.name = `Actor trigger: ${clip(next.task.split('\n')[0], 80)}`;
++ }
++ if (next.kind === 'event' && (patch.event !== undefined || patch.kind !== undefined)) changes.source = { kind: 'event', pattern: next.event.pattern, filter: next.event.filter || null };
++ if (Object.keys(changes).length) {
++ const patched = await client.patch(token, row.watch_id, changes);
++ if (!patched.ok) return { error: problem(patched.status || 502, patched.code || 'actor.trigger.watch_failed', patched.detail || 'OpenVibe.Watch refused the change.') };
++ }
++ if (next.kind === 'schedule' && (patch.cron !== undefined || patch.timezone !== undefined || patch.kind !== undefined)) {
++ const scheduled = await client.setSchedule(token, row.watch_id, next.cron, next.timezone);
++ if (!scheduled.ok) return { error: problem(scheduled.status || 502, scheduled.code || 'actor.trigger.watch_failed', scheduled.detail || 'OpenVibe.Watch refused the schedule.') };
++ }
++ const updated = await store.update(s, owner, row.id, {
++ task: next.task, agent_id: next.agent_id, kind: next.kind,
++ schedule: next.kind === 'schedule' ? { cron: next.cron, timezone: next.timezone } : row.schedule,
++ event: next.kind === 'event' ? { pattern: next.event.pattern, filter: next.event.filter || null } : row.event,
++ timezone: next.kind === 'schedule' ? next.timezone : null,
++ words: next.kind === 'schedule' ? parse.wordsFor(next.cron, next.timezone) : (next.event.words || row.words),
++ updated_at: nowMs(),
++ });
++ await emit(updated, `Trigger edited: ${updated.words}.`);
++ return { trigger: await card(updated) };
++ }
++
++ async function pause(owner, id, { token = null } = {}) {
++ const row = ID.test(String(id || '')) ? await store.get(s, owner, id) : null;
++ if (!row) return { error: problem(404, 'actor.trigger.not_found', 'No such trigger.') };
++ if (!token) return { error: problem(401, 'actor.trigger.sign_in', 'Sign in again: pausing a trigger uses your own OpenVibe session.') };
++ if (row.running) return { error: problem(409, 'actor.trigger.busy', 'This trigger is running now; try again in a moment.') };
++ if (row.watch_id) {
++ const out = await client.pause(token, row.watch_id);
++ if (!out.ok && out.status !== 404) return { error: problem(out.status || 502, out.code || 'actor.trigger.watch_failed', out.detail || 'OpenVibe.Watch did not pause it.') };
++ }
++ const updated = await store.update(s, owner, row.id, { state: 'paused', paused_reason: 'person', updated_at: nowMs() });
++ await emit(updated, 'Trigger paused.');
++ return { trigger: await card(updated) };
++ }
++
++ async function resume(owner, id, { token = null } = {}) {
++ const row = ID.test(String(id || '')) ? await store.get(s, owner, id) : null;
++ if (!row) return { error: problem(404, 'actor.trigger.not_found', 'No such trigger.') };
++ if (!token) return { error: problem(401, 'actor.trigger.sign_in', 'Sign in again: resuming a trigger uses your own OpenVibe session.') };
++ if (await store.countActive(s, owner) >= MAX_ACTIVE) return { error: problem(409, 'actor.trigger.limit', `You already have ${MAX_ACTIVE} active triggers. Pause one before resuming another.`) };
++ if (row.watch_id) {
++ const out = await client.resume(token, row.watch_id);
++ if (!out.ok) return { error: problem(out.status || 502, out.code || 'actor.trigger.watch_failed', out.detail || 'OpenVibe.Watch did not resume it.') };
++ }
++ const updated = await store.update(s, owner, row.id, { state: 'active', paused_reason: null, failures: 0, updated_at: nowMs() });
++ await emit(updated, 'Trigger resumed.');
++ return { trigger: await card(updated) };
++ }
++
++ async function undo(owner, id, { token = null } = {}) {
++ const row = ID.test(String(id || '')) ? await store.get(s, owner, id) : null;
++ if (!row) return { error: problem(404, 'actor.trigger.not_found', 'No such trigger.') };
++ if (nowMs() - Number(row.created_at) > UNDO_MS) return { error: problem(409, 'actor.trigger.undo_expired', 'The ten-minute undo window has passed. Delete the trigger instead.') };
++ if (!token) return { error: problem(401, 'actor.trigger.sign_in', 'Sign in again: undoing a trigger uses your own OpenVibe session.') };
++ if (row.watch_id) {
++ const out = await client.remove(token, row.watch_id);
++ if (!out.ok && out.status !== 404) return { error: problem(out.status || 502, out.code || 'actor.trigger.watch_failed', out.detail || 'OpenVibe.Watch did not remove the subscription.') };
++ }
++ await store.remove(s, owner, row.id);
++ return { undone: true, id: row.id };
++ }
++
++ async function remove(owner, id, { token = null, confirm = false } = {}) {
++ const row = ID.test(String(id || '')) ? await store.get(s, owner, id) : null;
++ if (!row) return { error: problem(404, 'actor.trigger.not_found', 'No such trigger.') };
++ if (!confirm) return { error: problem(409, 'actor.trigger.confirm', 'Deleting a trigger is permanent. Ask for confirmation first.') };
++ if (!token) return { error: problem(401, 'actor.trigger.sign_in', 'Sign in again: deleting a trigger uses your own OpenVibe session.') };
++ if (row.watch_id) {
++ const out = await client.remove(token, row.watch_id);
++ if (!out.ok && out.status !== 404) return { error: problem(out.status || 502, out.code || 'actor.trigger.watch_failed', out.detail || 'OpenVibe.Watch did not remove the subscription.') };
++ }
++ const updated = await store.update(s, owner, row.id, { state: 'deleted', deleted_at: nowMs(), updated_at: nowMs() });
++ await emit(updated, 'Trigger deleted.');
++ return { deleted: true, id: row.id };
++ }
++
++ async function runNow(owner, id) {
++ const row = ID.test(String(id || '')) ? await store.get(s, owner, id) : null;
++ if (!row) return { error: problem(404, 'actor.trigger.not_found', 'No such trigger.') };
++ const agent = await agentOf(owner, row.agent_id);
++ if (!agent) return { error: problem(422, 'actor.agent.not_found', 'The agent this trigger runs is gone. Pick another agent first.') };
++ if (!row.watch_id) return { error: problem(409, 'actor.trigger.no_watch', 'This trigger has no subscription. Delete it and make a new one.') };
++ const out = await personal.run({ s, config, engine, webhooks }, agent, row.task, { origin: 'routine', level: row.level, triggerId: row.id });
++ if (out && out.code) return { error: problem(out.status || 422, out.code, out.detail || 'The run could not start.') };
++ await emit(row, `Run now: ${row.task}.`);
++ return { task: out.task, trigger: await card(await store.get(s, owner, row.id)) };
++ }
++
++ /** Resolve the trigger a Watch delivery is for, by its id or the subscription's id. */
++ async function forDelivery(owner, { triggerId = null, watchId = null } = {}) {
++ const subject = subjectOf(owner);
++ if (triggerId && ID.test(String(triggerId))) {
++ const row = await store.get(s, subject, triggerId);
++ if (row) return row;
++ }
++ if (watchId && WATCH_ID.test(String(watchId))) return await store.byWatch(s, subject, watchId);
++ return null;
++ }
++
++ /**
++ * A task moved (the engine calls this after every state change). A failed run grows the streak; five in a
++ * row pause the trigger and say so on the card. A success clears it. A run waiting for the person is shown
++ * as needs_you and is not a failure.
++ */
++ async function taskState(task) {
++ if (!task || !task.id) return null;
++ const run = await store.runForTask(s, task.id);
++ if (!run) return null;
++ const row = await s.db.maybe('SELECT * FROM actor_triggers WHERE id = $1', [run.trigger_id]);
++ if (!row || row.state === 'deleted') return null;
++ const mapped = task.state === 'succeeded' ? 'ok'
++ : task.state === 'failed' ? 'failed'
++ : task.state === 'needs_you' ? 'needs_you'
++ : task.state === 'cancelled' ? 'failed' : 'running';
++ if (run.state === mapped) return row;
++ await store.addRun(s, run.trigger_id, task.id, mapped, nowMs());
++ if (mapped === 'running') return row;
++ let failures = Number(row.failures) || 0;
++ if (mapped === 'failed' && task.state !== 'cancelled') failures += 1;
++ if (mapped === 'ok') failures = 0;
++ const patch = {
++ last_task_id: task.id, last_state: mapped, last_run_at: nowMs(), failures, updated_at: nowMs(),
++ ...(failures >= 5 ? { state: 'paused', paused_reason: 'failed' } : {}),
++ };
++ const updated = await store.update(s, row.owner, row.id, patch);
++ if (failures >= 5) await emit(updated, `Paused after 5 failed runs in a row: ${row.task}.`);
++ return updated;
++ }
++
++ return { create, get, list, edit, pause, resume, undo, remove, runNow, forDelivery, taskState, card, MAX_ACTIVE, UNDO_MS };
++}
++
++module.exports = { createTriggers, MAX_ACTIVE, UNDO_MS };
ddiff --git a/server/triggers/store.js b/server/triggers/store.js
nnew file mode 100644
iindex 0000000..c049f5d
---- /dev/null
++++ b/server/triggers/store.js
@@@ -0,0 +1,79 @@
++'use strict';
++/** Triggers and their runs in Actor's own PostgreSQL (migrations/0037_actor_triggers.sql). */
++
++const parse = (v) => (v == null ? null : typeof v === 'string' ? JSON.parse(v) : v);
++
++function wire(row) {
++ if (!row) return null;
++ return {
++ id: row.id,
++ owner: row.owner,
++ agent_id: row.agent_id,
++ task: row.task,
++ kind: row.kind,
++ schedule: parse(row.schedule),
++ event: parse(row.event),
++ timezone: row.timezone || null,
++ words: row.words,
++ level: row.level || 'ask',
++ state: row.state,
++ paused_reason: row.paused_reason || null,
++ failures: Number(row.failures) || 0,
++ watch_id: row.watch_id || null,
++ turn_task_id: row.turn_task_id || null,
++ last_task_id: row.last_task_id || null,
++ last_state: row.last_state || null,
++ last_run_at: row.last_run_at == null ? null : Number(row.last_run_at),
++ created_at: Number(row.created_at),
++ updated_at: Number(row.updated_at),
++ };
++}
++
++const get = async (s, owner, id) => /^trg_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(id || ''))
++ ? wire(await s.db.maybe('SELECT * FROM actor_triggers WHERE id = $1 AND owner = $2 AND deleted_at IS NULL', [id, owner]))
++ : null;
++const byWatch = async (s, owner, watchId) => /^wch_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(watchId || ''))
++ ? wire(await s.db.maybe('SELECT * FROM actor_triggers WHERE watch_id = $1 AND owner = $2 AND deleted_at IS NULL', [watchId, owner]))
++ : null;
++const list = async (s, owner) => (await s.db.many('SELECT * FROM actor_triggers WHERE owner = $1 AND deleted_at IS NULL ORDER BY created_at DESC, id DESC LIMIT 200', [owner])).map(wire);
++const countActive = async (s, owner) => Number(await s.db.value("SELECT count(*) FROM actor_triggers WHERE owner = $1 AND state = 'active' AND deleted_at IS NULL", [owner])) || 0;
++const countForAgent = async (s, owner, agentId) => Number(await s.db.value('SELECT count(*) FROM actor_triggers WHERE owner = $1 AND agent_id = $2 AND deleted_at IS NULL', [owner, agentId])) || 0;
++
++async function insert(s, rec) {
++ await s.db.query(`INSERT INTO actor_triggers
++ (id, owner, agent_id, task, kind, schedule, event, timezone, words, level, state, paused_reason, failures, watch_id, turn_task_id, created_at, updated_at)
++ VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7::jsonb,$8,$9,$10,$11,$12,$13,$14,$15,$16,$16)`,
++ [rec.id, rec.owner, rec.agent_id, rec.task, rec.kind, rec.schedule ? JSON.stringify(rec.schedule) : null,
++ rec.event ? JSON.stringify(rec.event) : null, rec.timezone || null, rec.words, rec.level || 'ask',
++ rec.state || 'paused', rec.paused_reason || null, rec.failures || 0, rec.watch_id || null, rec.turn_task_id || null, rec.created_at]);
++ return await get(s, rec.owner, rec.id);
++}
++
++async function update(s, owner, id, patch) {
++ const cols = [];
++ const args = [id, owner];
++ const json = new Set(['schedule', 'event']);
++ for (const [k, v] of Object.entries(patch)) {
++ if (!['agent_id', 'task', 'schedule', 'event', 'timezone', 'words', 'level', 'state', 'paused_reason', 'failures', 'watch_id', 'last_task_id', 'last_state', 'last_run_at', 'updated_at', 'deleted_at'].includes(k)) continue;
++ args.push(json.has(k) && v != null ? JSON.stringify(v) : v);
++ cols.push(`${k} = $${args.length}${json.has(k) ? '::jsonb' : ''}`);
++ }
++ if (!cols.length) return await get(s, owner, id);
++ await s.db.query(`UPDATE actor_triggers SET ${cols.join(', ')} WHERE id = $1 AND owner = $2 AND deleted_at IS NULL`, args);
++ return await get(s, owner, id);
++}
++
++async function addRun(s, triggerId, taskId, state, at) {
++ await s.db.query(`INSERT INTO actor_trigger_runs (trigger_id, task_id, state, created_at, updated_at)
++ VALUES ($1,$2,$3,$4,$4) ON CONFLICT (task_id) DO UPDATE SET state = EXCLUDED.state, updated_at = EXCLUDED.updated_at`,
++ [triggerId, taskId, state, at]);
++}
++const runForTask = (s, taskId) => s.db.maybe('SELECT * FROM actor_trigger_runs WHERE task_id = $1', [taskId]);
++const runs = (s, triggerId, limit = 20) => s.db.many('SELECT * FROM actor_trigger_runs WHERE trigger_id = $1 ORDER BY created_at DESC LIMIT $2', [triggerId, limit]);
++const removeRuns = (s, triggerId) => s.db.query('DELETE FROM actor_trigger_runs WHERE trigger_id = $1', [triggerId]);
++const remove = async (s, owner, id) => {
++ await s.db.query('DELETE FROM actor_trigger_runs WHERE trigger_id = $1', [id]);
++ await s.db.query('DELETE FROM actor_triggers WHERE id = $1 AND owner = $2', [id, owner]);
++};
++
++module.exports = { wire, get, byWatch, list, countActive, countForAgent, insert, update, addRun, runForTask, runs, removeRuns, remove };
ddiff --git a/server/triggers/tokens.js b/server/triggers/tokens.js
nnew file mode 100644
iindex 0000000..aacf54a
---- /dev/null
++++ b/server/triggers/tokens.js
@@@ -0,0 +1,24 @@
++'use strict';
++/**
++ * The person's own token for the minutes a chat turn runs. It is kept in memory, for one task id,
++ * never written to the database, never logged and never put in an event; the tool reads it to create
++ * or change a Watch subscription as the person. An hour is enough for a turn that was queued behind
++ * another (the token itself carries its own expiry, and Watch checks it).
++ */
++const TTL_MS = 60 * 60_000;
++const MAP = new Map();
++
++function put(taskId, { token = null, timezone = null } = {}) {
++ if (!taskId || !token) return;
++ MAP.set(taskId, { token, timezone: timezone || null, at: Date.now() });
++}
++function get(taskId) {
++ const found = taskId ? MAP.get(taskId) : null;
++ if (!found) return null;
++ if (Date.now() - found.at > TTL_MS) { MAP.delete(taskId); return null; }
++ return found;
++}
++function drop(taskId) { if (taskId) MAP.delete(taskId); }
++function size() { return MAP.size; }
++
++module.exports = { put, get, drop, size, TTL_MS };
ddiff --git a/server/triggers/watch.js b/server/triggers/watch.js
nnew file mode 100644
iindex 0000000..e1573f4
---- /dev/null
++++ b/server/triggers/watch.js
@@@ -0,0 +1,69 @@
++'use strict';
++/**
++ * OpenVibe.Watch, called with the person's own token — never Actor's service token acting for them. A
++ * trigger is a watch the person owns: pause, resume, edit and delete go to Watch as that person, and
++ * another person's watch is 404 there just as another person's trigger is 404 here.
++ *
++ * The person's token reaches Actor as their session or Bearer; the trigger tool reads it from the
++ * short-lived run token (server/triggers/tokens.js) and the API routes from the request itself.
++ */
++const TIMEOUT_MS = 10_000;
++const PREFIX = { 'watch.watch.manage': null };
++
++function createWatchClient({ config, fetchImpl = globalThis.fetch, log = console }) {
++ const base = String((config.services && config.services.watchUrl) || '').replace(/\/+$/, '');
++ const enabled = Boolean(base);
++
++ async function call(token, method, path, body) {
++ if (!enabled) return { ok: false, status: 0, code: 'actor.trigger.watch_unavailable', detail: 'OpenVibe.Watch is not configured on this Actor (ACTOR_WATCH_URL).' };
++ if (!token) return { ok: false, status: 401, code: 'actor.trigger.sign_in', detail: 'Sign in again: creating or changing a trigger needs your own OpenVibe session, not a service token.' };
++ let res, text;
++ try {
++ res = await fetchImpl(`${base}${path}`, {
++ method,
++ headers: {
++ accept: 'application/json',
++ authorization: `Bearer ${token}`,
++ ...(body === undefined ? {} : { 'content-type': 'application/json' }),
++ },
++ body: body === undefined ? undefined : JSON.stringify(body),
++ signal: AbortSignal.timeout(TIMEOUT_MS),
++ });
++ text = await res.text();
++ } catch (err) {
++ return { ok: false, status: 0, code: 'actor.trigger.watch_unavailable', detail: `OpenVibe.Watch did not answer (${String((err && err.message) || err).slice(0, 160)}).` };
++ }
++ let data = null;
++ try { data = text ? JSON.parse(text) : null; } catch { /* not JSON */ }
++ if (res.ok) return { ok: true, status: res.status, data, text };
++ if (res.status === 401) return { ok: false, status: 401, code: 'actor.trigger.sign_in', detail: 'Your OpenVibe session expired while saving the trigger. Sign in again and ask once more.', data };
++ const code = (data && (data.code || data.title)) || `watch_${res.status}`;
++ const detail = (data && (data.detail || data.title)) || text.slice(0, 300) || `OpenVibe.Watch answered ${res.status}`;
++ return { ok: false, status: res.status, code, detail, data };
++ }
++
++ async function create(token, watch) {
++ const out = await call(token, 'POST', '/api/v1/watches', watch);
++ return out.ok ? { ok: true, watch: out.data && out.data.watch, data: out.data } : out;
++ }
++ async function get(token, id) {
++ const out = await call(token, 'GET', `/api/v1/watches/${encodeURIComponent(id)}`);
++ return out.ok ? { ok: true, watch: out.data && out.data.watch } : out;
++ }
++ async function patch(token, id, changes) {
++ const out = await call(token, 'PATCH', `/api/v1/watches/${encodeURIComponent(id)}`, changes);
++ return out.ok ? { ok: true, watch: out.data && out.data.watch } : out;
++ }
++ async function setSchedule(token, id, cronExpr, timezone) {
++ const out = await call(token, 'POST', `/api/v1/watches/${encodeURIComponent(id)}/schedule`, { cron: cronExpr, timezone });
++ return out.ok ? { ok: true, data: out.data } : out;
++ }
++ async function pause(token, id) { return await call(token, 'POST', `/api/v1/watches/${encodeURIComponent(id)}/pause`); }
++ async function resume(token, id) { return await call(token, 'POST', `/api/v1/watches/${encodeURIComponent(id)}/resume`); }
++ async function remove(token, id) { return await call(token, 'DELETE', `/api/v1/watches/${encodeURIComponent(id)}`); }
++ async function check(token, id) { return await call(token, 'POST', `/api/v1/watches/${encodeURIComponent(id)}/check`); }
++
++ return { enabled, base, create, get, patch, setSchedule, pause, resume, remove, check };
++}
++
++module.exports = { createWatchClient };
ddiff --git a/test/contracts.test.js b/test/contracts.test.js
iindex 23e501c..a88c425 100644
---- a/test/contracts.test.js
++++ b/test/contracts.test.js
@@@ -28,6 +28,16 @@ const ROUTE_EXTENSIONS = new Map([
['GET /api/v1/agent/timeline/:id', 'actor.timeline.read'],
['POST /api/v1/agent/approvals/:product/:id/decide', 'actor.approval.decide'],
['POST /api/v1/agent/kill-switch', 'actor.resource.control'],
++ // Triggers made in a conversation (plan §12 A-4). The routes exist before Contracts lists actor.trigger.*; a
++ // trigger is the person's own work record, so the task capabilities the person already holds guard it.
++ ['GET /api/v1/triggers', 'actor.task.read'],
++ ['GET /api/v1/triggers/:id', 'actor.task.read'],
++ ['POST /api/v1/triggers', 'actor.task.create'],
++ ['PATCH /api/v1/triggers/:id', 'actor.task.create'],
++ ['POST /api/v1/triggers/:id/pause', 'actor.task.create'],
++ ['POST /api/v1/triggers/:id/resume', 'actor.task.create'],
++ ['POST /api/v1/triggers/:id/run', 'actor.task.create'],
++ ['DELETE /api/v1/triggers/:id', 'actor.task.create'],
]);
const API_SRC = fs.readFileSync(path.join(__dirname, '..', 'server', 'http', 'api.js'), 'utf8');
// The A2A surface (stage V2-11): POST /a2a/v1/tasks and GET /a2a/v1/tasks/:id, guarded by actor.a2a.task.
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex 8cf7f3a..51bb987 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -118,6 +118,7 @@ const SAME = { 'sec-fetch-site': 'same-origin' };
'server/runtime/models.js': 'the configured provider base URLs (OpenAI, DeepSeek, the runtime, any OpenAI-compatible or local endpoint) with the operator\'s keys, only to list models at boot and to call chat completions; never a URL from a task (test/runtime-models.test.js)',
'server/runtime/tools/_openvibe.js': 'the configured OpenVibe.Search, OpenVibe.Tools and OpenVibe.Bot bases with Actor\'s own service token; a page URL is sent to Tools as input, never fetched here (test/runtime-tools.test.js)',
'server/runtime/tools/_space.js': 'the configured OpenVibe.Space base (ACTOR_SPACE_URL, loopback) with Actor\'s own service token (space.build.read|create|write|publish): the person is named in X-OV-Subject, the space id must be spc_<ULID> and the file path is URL-encoded; no URL from the model or from a task is ever fetched here (test/space-tools.test.js)',
++ 'server/triggers/watch.js': 'the configured OpenVibe.Watch base (ACTOR_WATCH_URL, loopback) with the person\'s own bearer token: create, schedule, pause, resume, edit and delete their subscription; the watch id must be wch_<ULID> and is URL-encoded, a cron and timezone are data, and no URL from a task is ever fetched here (test/triggers.test.js)',
'server/runtime/copilot.js': 'OpenVibe.Chat (OV_CHAT_URL/config, configured) with Actor\'s own service token for the room digest, answer and poll, and OpenVibe.Live for the state, copilot-ready and step calls; the stream id must be ags_<ULID> and is URL-encoded; no URL from a person or a chat line (test/agent-copilot.test.js)',
'server/events-consumer.js': 'OpenVibe.Events: the configured events.url (ACTOR_EVENTS_URL), to create the two ADR-033 subscriptions at boot',
'server/net/secret-signer.js': 'OpenVibe.Network: the configured networkInternalUrl (OV_NETWORK_INTERNAL_URL), POST /internal/secrets/use to sign a webhook delivery with a project secret (ADR-056); the body is the timestamp and the delivery, never a URL',
ddiff --git a/test/tool-acts.test.js b/test/tool-acts.test.js
iindex 2039272..eaa307e 100644
---- a/test/tool-acts.test.js
++++ b/test/tool-acts.test.js
@@@ -46,6 +46,7 @@ const EXPECTED = {
remember: true,
robot_steps: true,
run_tool: true,
++ schedule_trigger: true,
// Space: the agent builds the person's own spaces (plan 2026-10-22 §Z stage Z1). Making one, writing its files
// and asking to publish it change the world; listing and reading them do not.
space_create: true,
ddiff --git a/test/triggers.test.js b/test/triggers.test.js
nnew file mode 100644
iindex 0000000..209da56
---- /dev/null
++++ b/test/triggers.test.js
@@@ -0,0 +1,288 @@
++'use strict';
++/**
++ * Triggers made in a conversation (plan §12 A-4): the sentence parses or asks back, the subscription is
++ * a Watch the person owns, its runs are assignment roots at the person's level, the card has the last
++ * result and the next run, and the four moves (edit, pause, resume, undo/delete) plus Run now are the
++ * person's alone — another person gets 404. Five failed runs in a row pause it and say so.
++ */
++const assert = require('assert');
++const http = require('http');
++const crypto = require('crypto');
++const { serviceAuth } = require('openvibe-contracts');
++const { ids } = require('openvibe-contracts');
++const { boot, check, done } = require('./helpers/boot');
++const { parseRequest } = require('../server/triggers/parse');
++const contacts = require('../server/agents/contacts');
++const triggerStore = require('../server/triggers/store');
++const { createRegistry } = require('../server/runtime/tools');
++
++const SAME = { 'sec-fetch-site': 'same-origin' };
++
++/** OpenVibe.Watch as the person's token reaches it: records every call and answers like the service. */
++async function watchStub() {
++ const calls = [];
++ const server = http.createServer(async (req, res) => {
++ const chunks = [];
++ for await (const c of req) chunks.push(c);
++ const text = Buffer.concat(chunks).toString('utf8');
++ let body = null;
++ try { body = text ? JSON.parse(text) : null; } catch { body = text; }
++ calls.push({ method: req.method, url: req.url, body, auth: req.headers.authorization });
++ const send = (status, value) => { res.writeHead(status, { 'content-type': 'application/json' }); res.end(JSON.stringify(value)); };
++ if (req.method === 'POST' && req.url === '/api/v1/watches') {
++ const id = ids.newId('watch');
++ return send(201, { watch: { id, status: body.status || 'active', name: body.name, labels: body.labels || {}, source: body.source } });
++ }
++ if (req.method === 'POST' && /^\/api\/v1\/watches\/wch_[^/]+\/schedule$/.test(req.url)) {
++ return send(200, { id: req.url.split('/')[4], schedule: body, next_due_at: '2026-11-10T08:00:00.000Z' });
++ }
++ if (req.method === 'POST' && /^\/api\/v1\/watches\/wch_[^/]+\/(pause|resume)$/.test(req.url)) {
++ return send(200, { id: req.url.split('/')[4], status: req.url.endsWith('/resume') ? 'active' : 'paused' });
++ }
++ if (req.method === 'PATCH' && /^\/api\/v1\/watches\/wch_[^/]+$/.test(req.url)) {
++ return send(200, { watch: { id: req.url.split('/')[4], status: 'active', ...body } });
++ }
++ if (req.method === 'GET' && /^\/api\/v1\/watches\/wch_[^/]+$/.test(req.url)) {
++ return send(200, { watch: { id: req.url.split('/')[4], status: 'active', health: { last_state: 'condition_met', consecutive_failures: 0, next_due_at: '2026-11-10T08:00:00.000Z' } } });
++ }
++ if (req.method === 'DELETE' && /^\/api\/v1\/watches\/wch_[^/]+$/.test(req.url)) return send(200, { id: req.url.split('/')[4], deleted: true });
++ return send(404, { code: 'watch.not_found' });
++ });
++ await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
++ return { url: `http://127.0.0.1:${server.address().port}`, calls, close: () => new Promise((r) => server.close(r)) };
++}
++
++(async () => {
++ const watch = await watchStub();
++ const t = await boot({ env: { ACTOR_WATCH_URL: watch.url }, loop: { close: async () => {}, run: async () => ({ text: 'Done.', sources: [], evidence: [], model: 'fake' }) } });
++ try {
++ const ada = t.network.addUser('ada');
++ const bo = t.network.addUser('bo');
++ const cy = t.network.addUser('cy');
++ const owner = `user:${ada.subject}`;
++ const s = t.ctx.s;
++ const vibe = await contacts.ensureVibe(s, t.config, owner);
++ const full = (await contacts.create(s, t.config, owner, { name: 'Full', level: 'full' })).row;
++ const call = (method, p, o = {}) => t.get(p, { as: ada, method, headers: SAME, ...o });
++ const json = (method, p, body) => call(method, p, { json: body });
++ const triggered = (id) => (async () => (await call('POST', `/api/v1/triggers/${id}/run`)).json().task)();
++
++ await check('the three examples parse; two ambiguous ones ask back', async () => {
++ const weekday = parseRequest('every weekday at 9, summarise my inbox', { timezone: 'Europe/Berlin' });
++ assert.deepStrictEqual([weekday.status, weekday.kind, weekday.cron, weekday.timezone, weekday.task], ['ok', 'schedule', '0 9 * * 1-5', 'Europe/Berlin', 'summarise my inbox']);
++ const issue = parseRequest('when a new issue lands in my repo, triage it', { timezone: 'Europe/Berlin' });
++ assert.deepStrictEqual([issue.status, issue.kind, issue.event.pattern, issue.task], ['ok', 'event', 'github.issue.opened', 'triage it']);
++ const sunday = parseRequest('every Sunday evening, plan my week', { timezone: 'Europe/Berlin' });
++ assert.deepStrictEqual([sunday.status, sunday.cron, sunday.task], ['ok', '0 19 * * 0', 'plan my week']);
++ const vague = parseRequest('every so often, check my email', { timezone: 'Europe/Berlin' });
++ assert.strictEqual(vague.status, 'ask');
++ assert.match(vague.question, /how often/i);
++ const nothing = parseRequest('when something happens, tell me', { timezone: 'Europe/Berlin' });
++ assert.strictEqual(nothing.status, 'ask');
++ assert.match(nothing.question, /what exactly/i);
++ // A clear schedule with no known timezone is a question, never a UTC guess.
++ const noZone = parseRequest('every weekday at 9, summarise my inbox', {});
++ assert.strictEqual(noZone.status, 'ask');
++ assert.match(noZone.question, /timezone/i);
++ });
++
++ await check('the tool creates on a clear sentence and asks back on a vague one', async () => {
++ const registry = createRegistry({ strict: true, log: { warn() {}, log() {} } });
++ const events = [];
++ const ctx = {
++ config: t.config, requester: owner, store: s, triggers: t.ctx.triggers,
++ personToken: 'a-person-token', timezone: 'Europe/Berlin', agentId: vibe.id, taskId: null, level: 'ask',
++ emit: async (e) => events.push(e),
++ };
++ const before = Number(await s.db.value('SELECT count(*) FROM actor_triggers WHERE owner = $1', [ada.subject]));
++ const out = await registry.run('schedule_trigger', { request: 'every weekday at 9, summarise my inbox' }, ctx);
++ assert.ok(!out.isError, out.text);
++ const cardEvent = events.find((e) => e.step === 'trigger');
++ assert.ok(cardEvent, 'the card is a trigger step on the turn');
++ assert.strictEqual(cardEvent.trigger.kind, 'schedule');
++ assert.strictEqual(cardEvent.trigger.when.cron, '0 9 * * 1-5');
++ const asked = await registry.run('schedule_trigger', { request: 'every so often, check my email' }, ctx);
++ assert.match(asked.text, /how often/i);
++ const after = Number(await s.db.value('SELECT count(*) FROM actor_triggers WHERE owner = $1', [ada.subject]));
++ assert.strictEqual(after, before + 1, 'the vague sentence made nothing');
++ });
++
++ await check('creating a schedule makes a paused Watch, gives it the cron, then resumes it — with the person\'s token', async () => {
++ const res = await json('POST', '/api/v1/triggers', { agent_id: vibe.id, task: 'summarise my inbox', kind: 'schedule', cron: '0 9 * * 1-5', timezone: 'Europe/Berlin' });
++ assert.strictEqual(res.status, 201, res.text);
++ const card = res.json().trigger;
++ assert.match(card.id, /^trg_/);
++ assert.deepStrictEqual([card.kind, card.when.cron, card.when.timezone, card.state], ['schedule', '0 9 * * 1-5', 'Europe/Berlin', 'active']);
++ const made = watch.calls.find((c) => c.method === 'POST' && c.url === '/api/v1/watches' && c.body.labels && c.body.labels.actor_trigger === card.id);
++ assert.strictEqual(made.body.status, 'paused');
++ assert.deepStrictEqual(made.body.labels, { actor_trigger: card.id });
++ assert.ok(String(made.auth).startsWith('Bearer ') && !String(made.auth).includes('actor-secret'), 'the person\'s own bearer, not Actor\'s service secret');
++ assert.ok(watch.calls.some((c) => /\/schedule$/.test(c.url)));
++ assert.ok(watch.calls.some((c) => /\/resume$/.test(c.url)));
++ const row = await s.db.maybe('SELECT * FROM actor_triggers WHERE id = $1', [card.id]);
++ assert.strictEqual(row.state, 'active');
++ });
++
++ await check('the card has the next run, the last result, and the watch\'s own health', async () => {
++ const list = await call('GET', '/api/v1/triggers');
++ assert.strictEqual(list.status, 200, list.text);
++ assert.ok(list.json().triggers.length >= 1);
++ const id = list.json().triggers[0].id;
++ const one = await call('GET', `/api/v1/triggers/${id}`);
++ assert.strictEqual(one.status, 200, one.text);
++ const card = one.json().trigger;
++ assert.ok(card.next_run_at, 'next run is computed');
++ assert.strictEqual(card.watch.status, 'active');
++ assert.strictEqual(card.last_run, null);
++ assert.ok(card.undo_until);
++ });
++
++ await check('pause, resume, edit and Run now are the person\'s; another person gets 404', async () => {
++ const made = await json('POST', '/api/v1/triggers', { agent_id: vibe.id, task: 'plan my week', kind: 'schedule', cron: '0 19 * * 0', timezone: 'Europe/Berlin' });
++ const id = made.json().trigger.id;
++ const paused = await call('POST', `/api/v1/triggers/${id}/pause`);
++ assert.strictEqual(paused.status, 200, paused.text);
++ assert.strictEqual(paused.json().trigger.state, 'paused');
++ const resumed = await call('POST', `/api/v1/triggers/${id}/resume`);
++ assert.strictEqual(resumed.status, 200, resumed.text);
++ assert.strictEqual(resumed.json().trigger.state, 'active');
++ const edited = await call('PATCH', `/api/v1/triggers/${id}`, { json: { task: 'plan the next week' } });
++ assert.strictEqual(edited.status, 200, edited.text);
++ assert.strictEqual(edited.json().trigger.task, 'plan the next week');
++ assert.ok(watch.calls.some((c) => c.method === 'PATCH' && c.url.endsWith(id.replace('trg_', 'wch_')) === false)); // patched by the id returned at create
++
++ assert.strictEqual((await t.get(`/api/v1/triggers/${id}`, { as: bo })).status, 404);
++ assert.strictEqual((await t.get(`/api/v1/triggers/${id}/pause`, { as: bo, method: 'POST', headers: SAME })).status, 404);
++ assert.strictEqual((await t.get(`/api/v1/triggers/${id}/run`, { as: bo, method: 'POST', headers: SAME })).status, 404);
++ });
++
++ await check('Run now makes an origin-routine root at the trigger\'s level, never the agent\'s higher one', async () => {
++ const made = await json('POST', '/api/v1/triggers', { agent_id: full.id, task: 'say hi', kind: 'schedule', cron: '0 9 * * *', timezone: 'UTC' });
++ const id = made.json().trigger.id;
++ const run = await call('POST', `/api/v1/triggers/${id}/run`);
++ assert.strictEqual(run.status, 201, run.text);
++ const task = await t.waitFor(run.json().task.id);
++ const root = await s.db.maybe('SELECT * FROM actor_assignments WHERE task_id = $1 AND depth = 0', [task.id]);
++ assert.ok(root, 'the run is a root');
++ assert.strictEqual(root.origin, 'routine');
++ assert.strictEqual(root.level, 'ask');
++ assert.strictEqual(root.effective_level, 'ask', 'the routine cannot borrow the agent\'s full level');
++ let card = (await call('GET', `/api/v1/triggers/${id}`)).json().trigger;
++ for (let i = 0; i < 80 && !card.last_run; i += 1) {
++ await new Promise((r) => setTimeout(r, 25));
++ card = (await call('GET', `/api/v1/triggers/${id}`)).json().trigger;
++ }
++ assert.deepStrictEqual([card.last_run.state, card.last_run.task_id], ['ok', task.id]);
++ });
++
++ await check('event triggers subscribe the event source, not a webhook', async () => {
++ const made = await json('POST', '/api/v1/triggers', { agent_id: vibe.id, task: 'triage it', kind: 'event', event: { pattern: 'github.issue.opened', filter: { repo: 'openvibe/actor' } } });
++ assert.strictEqual(made.status, 201, made.text);
++ const callMade = watch.calls.filter((c) => c.method === 'POST' && c.url === '/api/v1/watches').pop();
++ assert.strictEqual(callMade.body.source.kind, 'event');
++ assert.strictEqual(callMade.body.source.pattern, 'github.issue.opened');
++ assert.strictEqual(callMade.body.source.filter.repo, 'openvibe/actor');
++ assert.strictEqual(made.json().trigger.kind, 'event');
++ });
++
++ await check('undo within ten minutes removes it with no trace; delete needs confirmation after', async () => {
++ const made = await json('POST', '/api/v1/triggers', { agent_id: vibe.id, task: 'undo me', kind: 'schedule', cron: '0 9 * * *', timezone: 'UTC' });
++ const id = made.json().trigger.id;
++ const undone = await call('DELETE', `/api/v1/triggers/${id}?undo=1`);
++ assert.strictEqual(undone.status, 200, undone.text);
++ assert.strictEqual(undone.json().undone, true);
++ assert.strictEqual((await call('GET', `/api/v1/triggers/${id}`)).status, 404);
++ assert.strictEqual(await s.db.maybe('SELECT * FROM actor_triggers WHERE id = $1', [id]), null, 'no row remains');
++ assert.ok(watch.calls.some((c) => c.method === 'DELETE'));
++
++ const old = await json('POST', '/api/v1/triggers', { agent_id: vibe.id, task: 'delete me', kind: 'schedule', cron: '0 9 * * *', timezone: 'UTC' });
++ const oldId = old.json().trigger.id;
++ await s.db.query('UPDATE actor_triggers SET created_at = $2 WHERE id = $1', [oldId, Date.now() - 11 * 60_000]);
++ assert.strictEqual((await call('DELETE', `/api/v1/triggers/${oldId}`)).status, 409, 'no confirmation, no delete');
++ const expired = await call('DELETE', `/api/v1/triggers/${oldId}?undo=1`);
++ assert.strictEqual(expired.status, 409, 'the undo window has passed');
++ const deleted = await call('DELETE', `/api/v1/triggers/${oldId}?confirm=delete`);
++ assert.strictEqual(deleted.status, 200, deleted.text);
++ assert.strictEqual((await call('GET', `/api/v1/triggers/${oldId}`)).status, 404);
++ });
++
++ await check('at most twenty active triggers; a paused one does not count', async () => {
++ const boOwner = `user:${bo.subject}`;
++ const boVibe = await contacts.ensureVibe(s, t.config, boOwner);
++ const boCall = (method, p, o = {}) => t.get(p, { as: bo, method, headers: SAME, ...o });
++ for (let i = 0; i < 20; i += 1) {
++ const made = await boCall('POST', '/api/v1/triggers', { json: { agent_id: boVibe.id, task: `filler ${i}`, kind: 'schedule', cron: '0 9 * * *', timezone: 'UTC' } });
++ assert.strictEqual(made.status, 201, made.text);
++ }
++ const over = await boCall('POST', '/api/v1/triggers', { json: { agent_id: boVibe.id, task: 'one too many', kind: 'schedule', cron: '0 9 * * *', timezone: 'UTC' } });
++ assert.strictEqual(over.status, 409, over.text);
++ assert.strictEqual(over.json().code, 'actor.trigger.limit');
++ const one = await s.db.maybe("SELECT id FROM actor_triggers WHERE owner = $1 AND state = 'active' AND deleted_at IS NULL LIMIT 1", [bo.subject]);
++ assert.strictEqual((await boCall('POST', `/api/v1/triggers/${one.id}/pause`)).status, 200);
++ const again = await boCall('POST', '/api/v1/triggers', { json: { agent_id: boVibe.id, task: 'room now', kind: 'schedule', cron: '0 9 * * *', timezone: 'UTC' } });
++ assert.strictEqual(again.status, 201, again.text);
++ });
++
++ await check('five failed runs in a row pause the trigger and say so on its card', async () => {
++ const cyVibe = await contacts.ensureVibe(s, t.config, `user:${cy.subject}`);
++ const made = await t.get('/api/v1/triggers', { as: cy, method: 'POST', headers: SAME, json: { agent_id: cyVibe.id, task: 'fails a lot', kind: 'schedule', cron: '0 9 * * *', timezone: 'UTC' } });
++ const id = made.json().trigger.id;
++ for (let i = 0; i < 5; i += 1) {
++ const taskId = ids.newId('task');
++ await s.db.query("INSERT INTO tasks (id, requester, task, mode, budget_task, budget_day, state, created_at) VALUES ($1,$2,'a failed run','balanced',0.01,1,'failed',$3) ON CONFLICT DO NOTHING", [taskId, `user:${cy.subject}`, new Date().toISOString()]);
++ await triggerStore.addRun(s, id, taskId, 'queued', Date.now());
++ await t.ctx.triggers.taskState({ id: taskId, state: 'failed' });
++ }
++ const card = (await t.get(`/api/v1/triggers/${id}`, { as: cy })).json().trigger;
++ assert.deepStrictEqual([card.state, card.paused_reason, card.failures], ['paused', 'failed', 5]);
++ assert.strictEqual(card.last_run.state, 'failed');
++ });
++ await check('a service acting for a person never leaves its own token as the person\'s', async () => {
++ const now = Math.floor(Date.now() / 1000);
++ const token = serviceAuth.signServiceToken({
++ iss: t.network.url, sub: 'svc:watch', actor_type: 'service', aud: ['openvibe.actor'],
++ cap: ['actor.task.create'], ns: [], iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(6).toString('hex')}`,
++ }, t.network.privatePem);
++ const res = await t.get('/api/v1/tasks', { bearer: token, headers: { 'X-OV-Subject': ada.subject }, json: { task: 'say hello' } });
++ assert.ok(res.status === 201 || res.status === 202, res.text);
++ const id = (res.json().task && res.json().task.id) || res.json().id;
++ assert.match(String(id), /^tsk_/);
++ assert.strictEqual(require('../server/triggers/tokens').get(id), null, 'Watch would have seen the service\'s token as the person\'s');
++ });
++
++ await check('the approval card for a routine says what and when, in words', async () => {
++ const { stepFor } = require('../server/agents/actions');
++ const step = stepFor('schedule_trigger', { task: 'summarise my inbox', cron: '0 9 * * 1-5', timezone: 'Europe/Berlin' });
++ assert.strictEqual(step.act, 'act');
++ assert.match(step.summary, /repeats \(0 9 \* \* 1-5, Europe\/Berlin\): "summarise my inbox"/);
++ const ev = stepFor('schedule_trigger', { request: 'when a new issue lands, triage it', event: { pattern: 'codes.issue.created' } });
++ assert.match(ev.summary, /when codes\.issue\.created happens/);
++ });
++
++ await check('Watch waking an agent is an origin-routine run linked to the trigger', async () => {
++ const made = await json('POST', '/api/v1/triggers', { agent_id: vibe.id, task: 'look into it', kind: 'schedule', cron: '0 9 * * *', timezone: 'UTC' });
++ const trigger = made.json().trigger;
++ const row = await s.db.maybe('SELECT watch_id FROM actor_triggers WHERE id = $1', [trigger.id]);
++ const now = Math.floor(Date.now() / 1000);
++ const token = serviceAuth.signServiceToken({
++ iss: t.network.url, sub: 'svc:watch', actor_type: 'service', aud: ['openvibe.actor'],
++ cap: ['actor.agent.run'], ns: [], iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(6).toString('hex')}`,
++ }, t.network.privatePem);
++ const res = await t.get(`/api/v1/me/agents/${vibe.id}/run`, {
++ bearer: token, headers: { 'X-OV-Subject': ada.subject },
++ json: { input: 'the watch fired', watch_id: row.watch_id, trigger_id: trigger.id },
++ });
++ assert.strictEqual(res.status, 201, res.text);
++ const task = await t.waitFor(res.json().id);
++ const root = await s.db.maybe('SELECT * FROM actor_assignments WHERE task_id = $1 AND depth = 0', [task.id]);
++ assert.deepStrictEqual([root.origin, root.effective_level], ['routine', 'ask']);
++ const run = await s.db.maybe('SELECT * FROM actor_trigger_runs WHERE task_id = $1', [task.id]);
++ assert.strictEqual(run.trigger_id, trigger.id, 'the run is linked back to its card');
++ });
++
++ } finally {
++ await t.close();
++ await watch.close();
++ }
++ done();
++})().catch((e) => { console.error(e); process.exit(1); });