openvibers/openvibe.watch Commits
Watch: an actor-agent action runs the owner's personal agent when the watch fires (#47)
Changes
@@@ -98,7 +98,7 @@ VALKEY_PREFIX=#WATCH_HOOK_EXAMPLE_SECRET=# A receiver answers within this, or the attempt is retried on the backoff.#WATCH_WEBHOOK_TIMEOUT_MS=10000--# OpenVibe.Actor's internal address, for actor-task actions (Watch's token needs actor.task.create).++# OpenVibe.Actor's internal address, for actor-task and actor-agent actions (Watch needs actor.task.create and actor.agent.run grants).#WATCH_ACTOR_URL=http://127.0.0.1:4950# Who may use which named secret (server/secret-grants.js). A name is not a capability: without a grant@@@ -227,6 +227,7 @@ A fired condition queues the watch's actions in the same transaction as the chec| `event` | `watch.watch.triggered` is emitted with every firing in the run's own commit, so that action is recorded `sent` as it is queued (the event names a person: an agent- or service-owned watch is `skipped`). Any other event type is not declared by Watch: `unavailable` || `webhook` | a POST of the firing (`type`, `watch {id, name, url}`, `check_run_id`, `condition`, `observation {observed_at, value, previous_value}`) to the action's https URL, through the same SSRF guard as every fetch, never following a redirect ([server/net/post.js](server/net/post.js)). It is signed with the key `secret_env` names exactly as OpenVibe.Events signs a delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`; check it with openvibe-sdk/events `verifyDeliveryV2`). `X-OpenVibe-Delivery-Id` (`<check run>.<action index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 or a URL the guard refuses is `failed` at once. Anything else, or a lost connection, is retried on the backoff, then `failed`. The key must be granted to the owner (`WATCH_SECRET_GRANTS`, below) and set, 32 characters or more: otherwise `unavailable` || `actor-task` | a task for the owner on OpenVibe.Actor ([server/actor.js](server/actor.js)): the action's `brief`, followed by the firing as context (the watch, the value and what it was, its page), at most `budget_usd`. Watch names the owner in `X-OV-Subject`, so the task runs under **the owner's** allowance and appears in their task list, and Actor records Watch as `via`. The idempotency key `watch.<check run>.<action index>` makes a retried delivery the same task. Created is `sent`, with the task id recorded. A request Actor refuses as invalid (over the tier's budget, an unknown agent) is `failed` at once. Anything else, including a day's allowance spent, is retried on the backoff. An agent- or service-owned watch is `skipped`. Needs `actor.task.create` for `openvibe.actor` (Network grants it) and `WATCH_ACTOR_URL` |++| `actor-agent` | runs one of the person's OpenVibe.Actor agents by `agent_id`. Its optional `note` precedes the firing context (watch name and link, value and previous value, observation time); input is capped at 8000 characters without cutting the note. Watch uses `X-OV-Subject` for the owner. A created run is `sent` with its task id; a missing agent, limit or free-tier refusal is `failed` for that firing; service errors retry on the usual backoff. Only person-owned watches may use it. Needs the `actor.agent.run` grant for `openvibe.actor` and `WATCH_ACTOR_URL` || `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.@@@ -487,7 +488,7 @@ POSTs have their own per-address limit zones (nginx refuses `limit_req` inside ``sudo ovhost deploy watch`.<!-- versions:start -->--- openvibe-contracts: v0.129.0++- openvibe-contracts: v0.143.0- openvibe-sdk: v0.44.0- openvibe-shared: v4.0.2<!-- versions:end -->@@@ -7,7 +7,7 @@"runtime": "node 22, express, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9, fast-xml-parser","port": 4730,"domain": "openvibe.watch (a public server-rendered site + a loopback-only API; vhost in deploy/nginx; README \"Public site\" and \"Public host\")",-- "contracts": "openvibe-contracts v0.129.0",++ "contracts": "openvibe-contracts v0.143.0 (actor-agent requires v0.143.0)","sdk": "openvibe-sdk v0.44.0","shared": "openvibe-shared v4.0.2","capabilities": [@@@ -35,6 +35,15 @@"run (step 6)","node (step 7)"],++ "actionKinds": [++ "notification",++ "event",++ "webhook",++ "actor-task",++ "actor-agent",++ "codes-run",++ "run-job"++ ],"seededWatches": 0,"deployed": true,"deployment": "openvibe-ovh: unit openvibe-watch.service on 127.0.0.1:4730 (release layout /opt/openvibe.watch, env /etc/openvibe/watch.env, database ov_watch), deployed with `sudo ovhost deploy watch`; the public site openvibe.watch (sign-in, watch management, plan T18 step 8) has been live since 2026-10-08. On 2026-10-10 it held no watches yet, so no check has run against a real site in production.",@@@ -12,7 +12,7 @@"dotenv": "^16.4.7","express": "^4.21.2","fast-xml-parser": "^4.5.3",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.0.2","pg": "^8.23.0"@@@ -769,9 +769,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.129.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",-- "integrity": "sha512-Ly0F/ap6wmAf8hYpNrOhMLsU3cZYBjsFDW2P1F/xVmHVsRDzAPAZDXtFcohP0UNeZgyxgrR0BI0pPG5guD/Efw==",++ "version": "0.143.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",++ "integrity": "sha512-Cdbf+9RoPAUBibpM2zXQvtsiKMugb55dKubTt8pFZvAmxsLtGvX+zaC++yL1vOSub8FX3+GgqcGtoJS6tklrUg==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -24,7 +24,7 @@"dotenv": "^16.4.7","express": "^4.21.2","fast-xml-parser": "^4.5.3",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.0.2","pg": "^8.23.0"@@@ -33,8 +33,8 @@"node": ">=22.0.0"},"devDependencies": {++ "@electric-sql/pglite": "^0.5.8","ajv": "^8.20.0",-- "ajv-formats": "^3.0.1",-- "@electric-sql/pglite": "^0.5.8"++ "ajv-formats": "^3.0.1"}}@@@ -35,6 +35,8 @@* Actor refuses as invalid (over the tier's budget, an unknown agent) is `failed` at once; anything* else, including an allowance spent for the day, is retried on the backoff. It needs a person as* owner: an agent- or service-owned watch is `skipped`.++ * actor-agent runs one of the owner's personal agents on Actor with a note and bounded firing context.++ * A missing agent or allowance refusal fails this firing; a service error retries.* codes-run, run-job* `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and* the watch page says so rather than pretending they ran.@@@ -103,16 +105,26 @@ function webhookFor({ config, watch, run, value, previousValue, observedAt }) {/** The task an actor-task action gives Actor: its brief, then what fired, so the agent knows why it was woken. */function actorTaskFor({ config, watch, action, value, previousValue, observedAt }) {++ const context = actorContextFor({ config, watch, value, previousValue, observedAt });++ const brief = String(action.brief || '').trim();++ return `${brief.slice(0, TASK_MAX - context.length - 2)}\n\n${context}`;++}++++function actorContextFor({ config, watch, value, previousValue, observedAt }, maxLength = Infinity) {const now = valueText(value);const before = previousValue === undefined || previousValue === null ? null : valueText(previousValue);const when = observedAt ? new Date(observedAt).toISOString() : 'just now';-- const context = [-- `Context: your watch “${watch.name}” fired at ${when}.`,-- `Value: ${now}${before !== null && before !== now ? ` (was ${before})` : ''}.`,-- `Watch: ${config.baseUrl}/watches/${encodeURIComponent(watch.id)}`,-- ].join('\n');-- const brief = String(action.brief || '').trim();-- return `${brief.slice(0, TASK_MAX - context.length - 2)}\n\n${context}`;++ const first = `Context: your watch “${watch.name}” fired at ${when}.`;++ const last = `Watch: ${config.baseUrl}/watches/${encodeURIComponent(watch.id)}`;++ const valueLine = `Value: ${now}${before !== null && before !== now ? ` (was ${before})` : ''}.`;++ const room = Math.max(0, maxLength - first.length - last.length - 2);++ return [first, valueLine.length > room ? `${valueLine.slice(0, Math.max(0, room - 1))}…` : valueLine, last].join('\n').slice(0, maxLength);++}++++function actorAgentInputFor(args) {++ const note = typeof args.action.note === 'string' ? args.action.note : '';++ const prefix = note ? `${note}\n\n` : '';++ return prefix + actorContextFor(args, 8000 - prefix.length);}function createActions({ db, config, notifications = null, post = null, actor = null, now = () => Date.now(), log = console }) {@@@ -159,6 +171,10 @@ function createActions({ db, config, notifications = null, post = null, actor =if (USER_SUBJECT.test(String(owner || ''))) {request = { task: actorTaskFor({ config, watch, action, value, previousValue, observedAt }), budget_usd: action.budget_usd == null ? null : action.budget_usd };} else { state = 'skipped'; error = 'watch.action_no_person: an Actor task runs under a person\'s allowance, and this watch belongs to an agent or a service'; }++ } else if (kind === 'actor-agent') {++ if (USER_SUBJECT.test(String(owner || ''))) {++ request = { agent_id: action.agent_id, input: actorAgentInputFor({ config, watch, action, value, previousValue, observedAt }) };++ } else { state = 'skipped'; error = 'watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service'; }} else if (LATER.has(kind)) {state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;} else {@@@ -210,10 +226,25 @@ function createActions({ db, config, notifications = null, post = null, actor =}}++ async function deliverActorAgent(row, request) {++ if (!actor) throw Object.assign(new Error('Actor delivery is not configured in this process'), { code: 'actor_unconfigured' });++ try {++ const task = await actor.runAgent({ subjectId: row.owner_sub, agentId: request.agent_id, input: request.input });++ return { state: 'sent', error: `Actor task ${task.id || '(no id)'}` };++ } catch (err) {++ if (err && err.status === 404 && err.code === 'actor.agent.not_found') return { state: 'failed', error: `watch.action_refused: ${err.code}: this agent was not found for the watch owner` };++ if (err && (err.status === 429 || err.status === 400 || err.status === 422 || (err.status === 403 && /(?:limit|free.tier|over_tier)/i.test(String(err.code || ''))))) {++ return { state: 'failed', error: `watch.action_refused: ${err.code || err.status}: ${String(err.detail || err.message || '').slice(0, 200)}` };++ }++ throw err;++ }++ }++async function deliver(row, attempt) {const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;if (row.kind === 'webhook') return await deliverWebhook(row, request, attempt);if (row.kind === 'actor-task') return await deliverActorTask(row, request);++ if (row.kind === 'actor-agent') return await deliverActorAgent(row, request);if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };const out = await notifications.push({ subjectId: row.owner_sub, ...request });if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };@@@ -285,4 +316,4 @@ function createActions({ db, config, notifications = null, post = null, actor =return { queue, tick, start, stop, status, recent, enabled };}--module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, BACKOFF_MS };++module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, actorAgentInputFor, BACKOFF_MS };@@@ -1,11 +1,13 @@'use strict';/**-- * OpenVibe.Actor, for a watch's actor-task action (plan T18 step 6): give Actor a task FOR the watch's owner.++ * OpenVibe.Actor, for a watch's actor-task and actor-agent actions: run work FOR the watch's owner.** const actor = createActorTasks(client, { baseUrl }); // client: openvibe-sdk/core createClient with Watch's token* await actor.create({ subjectId, task, budgetUsd, idempotencyKey }) → { id, state }++ * await actor.runAgent({ subjectId, agentId, input }) → { id, state }*-- * POST <actor>/api/v1/tasks with Watch's own token (actor.task.create, audience openvibe.actor) and the owner in++ * POST <actor>/api/v1/tasks with Watch's own token (actor.task.create, audience openvibe.actor), or++ * POST <actor>/api/v1/me/agents/:agentId/run (actor.agent.run), with the owner in* X-OV-Subject: Actor makes the task the person's (their allowance, their daily budget, their task list) and records* Watch as `via` (Actor#25). The idempotency key makes a retried delivery the same task, never a second one. The call* is made once; the action worker retries on its own backoff. Actor's refusals arrive as openvibe-sdk OpenVibeError@@@ -13,15 +15,20 @@*/function createActorTasks(client, { baseUrl }) {++ const post = (path, subjectId, json) => client.json({++ service: 'actor', baseUrl, audience: 'openvibe.actor', method: 'POST', path,++ headers: { 'X-OV-Subject': subjectId }, json, retries: 0, idempotencyKey: false,++ });async function create({ subjectId, task, budgetUsd = null, idempotencyKey }) {const json = { task, idempotency_key: idempotencyKey, ...(budgetUsd == null ? {} : { budget: { per_task_usd: budgetUsd } }) };-- const out = await client.json({-- service: 'actor', baseUrl, audience: 'openvibe.actor', method: 'POST', path: '/api/v1/tasks',-- headers: { 'X-OV-Subject': subjectId }, json, retries: 0, idempotencyKey: false,-- });++ const out = await post('/api/v1/tasks', subjectId, json);return { id: out && out.id, state: out && out.state };}-- return { create };++ async function runAgent({ subjectId, agentId, input }) {++ const out = await post(`/api/v1/me/agents/${encodeURIComponent(agentId)}/run`, subjectId, { input });++ return { id: out && (out.task_id || (out.task && out.task.id) || out.id), state: out && out.state };++ }++ return { create, runAgent };}module.exports = { createActorTasks };@@@ -99,7 +99,7 @@ function load(env = process.env) {batch: int(env.WATCH_ACTIONS_BATCH, 20),// A webhook receiver answers within this, or the attempt is retried on the backoff.webhookTimeoutMs: int(env.WATCH_WEBHOOK_TIMEOUT_MS, 10000),-- // OpenVibe.Actor's internal address, for actor-task actions (Watch's token needs actor.task.create).++ // OpenVibe.Actor's internal address, for actor-task and actor-agent actions.actorUrl: (env.WATCH_ACTOR_URL || 'http://127.0.0.1:4950').replace(/\/+$/, ''),},@@@ -121,7 +121,7 @@ function createPageRoutes(ctx) {template, name: '', url: '', kind: t.kind,extraction: t.extraction, selector: '', value_path: t.valuePath || '',op: t.defaultOp || t.ops[0], value: '', for_sec: '',-- cadence: '3600', notify: 'on',++ cadence: '3600', notify: 'on', agent_on: '', agent_id: '', agent_note: '',};}@@@ -132,6 +132,7 @@ function createPageRoutes(ctx) {function postedFields(template, body) {const fields = { ...blankFields(template), ...(body || {}) };fields.notify = body && body.notify === 'on' ? 'on' : '';++ fields.agent_on = body && body.agent_on === 'on' ? 'on' : '';return fields;}@@@ -163,9 +164,25 @@ function createPageRoutes(ctx) {for_sec: (watch.condition && Number.isInteger(watch.condition.for_sec)) ? String(watch.condition.for_sec) : '',cadence: String((watch.cadence && watch.cadence.every_sec) || 3600),notify: notifyOn(watch.action) ? 'on' : '',++ agent_on: agentAction(watch.action) ? 'on' : '',++ agent_id: (agentAction(watch.action) || {}).agent_id || '',++ agent_note: (agentAction(watch.action) || {}).note || '',};}const notifyOn = (action) => Array.isArray(action) && action.some((a) => a && a.kind === 'notification');++ const agentAction = (action) => Array.isArray(action) ? action.find((a) => a && a.kind === 'actor-agent') : null;++++ function actionsFrom(fields, errors) {++ const action = fields.notify === 'on' ? [...NOTIFY_ACTION] : [];++ if (fields.agent_on === 'on') {++ const agentId = String(fields.agent_id || '').trim();++ const note = String(fields.agent_note || '');++ if (!/^agt_[0-9A-HJKMNP-TV-Z]{26}$/.test(agentId)) errors.agent_id = 'Give a valid Actor agent id (agt_ followed by 26 characters).';++ if (note.length > 2000) errors.agent_note = 'The note can be at most 2000 characters.';++ action.push({ kind: 'actor-agent', agent_id: agentId, ...(note ? { note } : {}) });++ }++ return action.length ? action : QUIET_ACTION;++ }/** The condition a form says, checked against the template's ops. Returns { condition } or { error }. */function conditionFrom(fields, template) {@@@ -198,6 +215,7 @@ function createPageRoutes(ctx) {function buildBody(fields, template) {const t = TEMPLATES[template];const errors = {};++ const action = actionsFrom(fields, errors);// A webhook watch has no address to fetch and no cadence: it is pushed to (server/webhook-sources.js).if (template === 'webhook') {const name = String(fields.name == null ? '' : fields.name).trim();@@@ -211,7 +229,7 @@ function createPageRoutes(ctx) {cadence: null,extraction: valuePath ? { kind: 'json', value_path: valuePath } : { kind: 'text' },condition: c.condition,-- action: fields.notify === 'on' ? NOTIFY_ACTION : QUIET_ACTION,++ action,};if (name) body.name = name;return { body };@@@ -252,7 +270,7 @@ function createPageRoutes(ctx) {cadence: { every_sec: cadenceSec, jitter_sec: 0 },extraction,condition: c.condition,-- action: fields.notify === 'on' ? NOTIFY_ACTION : QUIET_ACTION,++ action,};if (name) body.name = name; // omitted, not undefined: the registry names it from the sourcereturn { body };@@@ -265,8 +283,8 @@ function createPageRoutes(ctx) {*/function registryField(message) {const m = String(message);-- const named = m.match(/\/(?:source|extraction|condition|cadence|name|action)\/(url|selector|value_path|value|for_sec|every_sec)\b/);-- if (named) return named[1] === 'every_sec' ? 'cadence' : named[1];++ const named = m.match(/\/(?:source|extraction|condition|cadence|name|action)(?:\/\d+)?\/(url|selector|value_path|value|for_sec|every_sec|agent_id|note)\b/);++ if (named) return named[1] === 'every_sec' ? 'cadence' : (named[1] === 'note' ? 'agent_note' : named[1]);if (/\/name\b/.test(m)) return 'name';if (/\/condition\b/.test(m)) return 'op';if (/\/cadence\b/.test(m)) return 'cadence';@@@ -363,9 +381,12 @@ ${conditionBlock}<fieldset><legend>${pushed ? 'How to tell you' : 'How often, and how to tell you'}</legend>${pushed ? html`<p class="muted small">Checked each time a signed delivery arrives (at most 60 an hour), never on a cadence.</p>` : field('Cadence', cadence, { name: 'cadence', error: errors.cadence, hint: 'Watch never checks more often than this, and can wait longer when the site asks it to.' })}${checkbox('Notify me when it fires', { name: 'notify', checked: fields.notify === 'on', hint: 'When the condition fires you get one notification on OpenVibe (your notifications on any OpenVibe site), with the value and a link here.' })}++${checkbox('Run one of my agents on OpenVibe.Actor', { name: 'agent_on', checked: fields.agent_on === 'on', hint: 'When the condition fires, Watch gives your agent the value and a link to this watch.' })}++${field('Agent id', html`<input type="text" id="${idOf('agent_id')}" name="agent_id" value="${fields.agent_id}" maxlength="30" placeholder="agt_…">`, { name: 'agent_id', error: errors.agent_id, hint: 'Find the id on your agent’s OpenVibe.Actor page. Required when running an agent.' })}++${field('Note for the agent (optional)', html`<textarea id="${idOf('agent_note')}" name="agent_note" maxlength="2000" rows="3">${fields.agent_note}</textarea>`, { name: 'agent_note', error: errors.agent_note })}</fieldset><p class="form-actions"><button class="btn primary" type="submit">${submit}</button> <a href="/watches">Cancel</a></p>--<p class="muted small">Notifications are delivered today. Events, webhooks and tasks for OpenVibe.Actor, Codes and Run are later steps: a watch that asks for one records it as not available yet.</p>++<p class="muted small">Notifications and Actor agents are delivered when the watch fires. Codes and Run actions are later steps.</p></form>`;}@@@ -558,6 +579,9 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: 'Createconst obs = await observations.list(row.id, { before: ridBefore(req.query.obs_before), limit: obsLimit(req.query.obs_limit) });const runs = await check.list(row.id, { before: ridBefore(req.query.checks_before), limit: obsLimit(req.query.checks_limit) });const acts = actions ? await actions.recent(row.id, 10) : [];++ const configuredActions = (v.action || []).map((a) => a.kind === 'actor-agent'++ ? html`<div>Run agent <a href="https://openvibe.actor/me/agents/${a.agent_id}"><code>${a.agent_id}</code></a>${a.note ? html` — ${a.note}` : ''}</div>`++ : html`<div><code>${a.kind}</code></div>`);const obsRows = obs.map((o) => [time(new Date(Number(o.observed_at)).toISOString()),@@@ -609,6 +633,7 @@ ${v.source.kind === 'webhook'? html`<dt>Webhook address</dt><dd><code>${config.baseUrl}/hooks/${v.id}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a new secret</button></form><br><span class="muted small">The secret was shown once, when the watch was made. A new one replaces it at once.</span></dd>`: html`<dt>Source</dt><dd><code>${v.source.kind}</code> · <code>${v.source.url || (v.source.pattern ? v.source.pattern : '(none)')}</code></dd>`}<dt>Cadence</dt><dd>${v.cadence ? W.everyWords(v.cadence.every_sec) : (v.source.kind === 'webhook' ? 'when a delivery arrives' : (v.source.kind === 'event' ? 'when its event arrives' : 'not polled'))}</dd>++<dt>Actions</dt><dd>${configuredActions}</dd><dt>Last check</dt><dd>${v.health.last_check_at ? html`${time(v.health.last_check_at)} — ${W.stateWords(v.health.last_state)}` : html`<span class="muted">not checked yet</span>`}</dd><dt>Next due</dt><dd>${v.health.next_due_at ? time(v.health.next_due_at) : html`<span class="muted">—</span>`}</dd>${v.health.consecutive_failures ? html`<dt>Failing</dt><dd>${v.health.consecutive_failures} check(s) in a row</dd>` : ''}@@@ -760,6 +785,7 @@ ${watchForm({ template: fields.template, fields, errors: {}, action: `/watches/$const template = getTemplate((req.body || {}).template);const fields = { ...fieldsOfWatch(registry.view(row)), ...(req.body || {}) };fields.notify = req.body && req.body.notify === 'on' ? 'on' : '';++ fields.agent_on = req.body && req.body.agent_on === 'on' ? 'on' : '';const built = buildBody(fields, template);let errors = built.errors;if (!errors) {@@@ -30,6 +30,8 @@ const STATUSES = ['active', 'paused', 'disabled', 'failed'];const MIN_EVERY_SEC = 900;/** Headers the fetcher decides; a watch may never override them (SSRF and conditional-GET safety). */const FORBIDDEN_HEADERS = new Set(['host', 'cookie', 'connection', 'content-length', 'transfer-encoding', 'user-agent', 'accept-encoding', 'if-none-match', 'if-modified-since']);++const PERSON = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const AGENT_ID = /^agt_[0-9A-HJKMNP-TV-Z]{26}$/;class RegistryError extends Error {constructor(detail, code = 'watch.invalid') { super(detail); this.code = code; }@@@ -59,6 +61,14 @@ function defaultName(source) {*/function validateRequest(body, { create = false } = {}) {if (!body || typeof body !== 'object' || Array.isArray(body)) throw new RegistryError('body must be a watch object');++ for (const [i, action] of (Array.isArray(body.action) ? body.action : []).entries()) {++ if (action && action.kind === 'actor-agent') {++ if (!AGENT_ID.test(action.agent_id)) throw new RegistryError(`/action/${i}/agent_id must be an Actor agent id`);++ if (action.note !== undefined && action.note !== null && (typeof action.note !== 'string' || action.note.length > 2000)) {++ throw new RegistryError(`/action/${i}/note must be a string of at most 2000 characters or null`);++ }++ }++ }const v = validate('watch.watch-request@1', body);if (!v.valid) throw new RegistryError(`body does not match watch.watch-request@1: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`);if (create) {@@@ -129,6 +139,11 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxconst bad = secretGrants.check(owner, rec);if (bad.length) throw new RegistryError(`${bad[0].path}: ${bad[0].reason}`, 'watch.secret_not_granted');}++ function checkActorAgents(owner, rec) {++ if (!PERSON.test(String(owner || '')) && Array.isArray(rec.action) && rec.action.some((a) => a && a.kind === 'actor-agent')) {++ throw new RegistryError('watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service');++ }++ }const st = {get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),@@@ -249,6 +264,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxconst owner = ownerOf(principal);if (!owner) throw new RegistryError('no acting subject', 'watch.invalid');checkSecrets(owner, req);++ checkActorAgents(owner, req);// A per-owner cap, so one caller cannot fill the registry (the API and the form both come// through here). The number is a config value (WATCH_MAX_PER_OWNER).const owned = await st.countOwned.get(owner);@@@ -317,6 +333,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxif (changes.project_id === undefined) delete merged.project_id;validateRequest(merged);checkSecrets(row.owner_sub, merged);++ checkActorAgents(row.owner_sub, merged);const sets = [];const args = { id: row.id, updated_at: now(), updated_by: ownerOf(principal) };for (const [k, v] of Object.entries(changes)) {@@@ -53,7 +53,16 @@ const actor = {if (actorMode === 'busy') throw Object.assign(new Error('capacity'), { status: 503, code: 'actor.capacity.spent' });return { id: `tsk_01JAB2C3D4E5F6G7H8J9K0MN${String(tasks.length).padStart(2, '0')}`, state: 'queued' };},++ async runAgent(req) {++ tasks.push(req);++ if (actorMode === 'missing') throw Object.assign(new Error('not found'), { status: 404, code: 'actor.agent.not_found' });++ if (actorMode === 'limit') throw Object.assign(new Error('free tier limit'), { status: 429, code: 'actor.agent.limit' });++ if (actorMode === 'free') throw Object.assign(new Error('free tier refusal'), { status: 403, code: 'actor.agent.free_tier' });++ if (actorMode === 'busy') throw Object.assign(new Error('capacity'), { status: 503, code: 'actor.unavailable' });++ return { id: 'tsk_AGENT', state: 'queued' };++ },};++const agentId = 'agt_01JAB2C3D4E5F6G7H8J9K0MNPQ';const def = (overrides = {}) => ({name: 'Widget price',source: { kind: 'http', url: `${web.origin}/price`, format: null },@@@ -318,6 +327,62 @@ t('server/actor.js asks Actor as Watch, for the owner, once', async () => {assert.deepStrictEqual(c.json, { task: 'do it', idempotency_key: 'watch.r.0', budget: { per_task_usd: 0.2 } });await createActorTasks(client, { baseUrl: 'x' }).create({ subjectId: person.sub, task: 'no budget', idempotencyKey: 'watch.r.1' });assert.ok(!('budget' in calls[1].json), 'no budget named: Actor applies the tier\'s');++ const agentRun = await createActorTasks(client, { baseUrl: 'http://127.0.0.1:4950' }).runAgent({ subjectId: person.sub, agentId, input: 'hello' });++ assert.deepStrictEqual(agentRun, { id: 'tsk_X', state: 'queued' });++ assert.deepStrictEqual([calls[2].baseUrl, calls[2].audience, calls[2].path, calls[2].retries],++ ['http://127.0.0.1:4950', 'openvibe.actor', `/api/v1/me/agents/${agentId}/run`, 0]);++ assert.deepStrictEqual(calls[2].headers, { 'X-OV-Subject': person.sub });++ assert.deepStrictEqual(calls[2].json, { input: 'hello' });++});++++t('actor-agent validation checks id, note and person ownership', async () => {++ const action = [{ kind: 'actor-agent', agent_id: agentId, note: 'Review this change.' }];++ const w = await create(person, { action });++ assert.deepStrictEqual(svc.registry.view(w).action, action);++ await assert.rejects(create(person, { action: [{ ...action[0], agent_id: 'agt_bad' }] }), /agent_id/);++ await assert.rejects(create(person, { action: [{ ...action[0], note: 'x'.repeat(2001) }] }), /note/);++ await assert.rejects(create(agent, { action }), (e) => e.code === 'watch.invalid' && /action_no_person/.test(e.message));++ await assert.rejects(create({ sub: 'svc:watch-test' }, { action }), (e) => e.code === 'watch.invalid' && /action_no_person/.test(e.message));++ await assert.rejects(svc.registry.patch(w.id, { action: [{ ...action[0], agent_id: 'agt_bad' }] }, person), /agent_id/);++});++++t('actor-agent sends note and bounded firing context, records its task, and classifies refusals', async () => {++ const { actorAgentInputFor } = require('../server/actions');++ const longInput = actorAgentInputFor({ config: svc.config, watch: { name: 'long value', id: 'wch_x' }, action: { note: 'N'.repeat(2000) }, value: 'V'.repeat(20000), previousValue: 'before', observedAt: Date.now() });++ assert.strictEqual(longInput.length, 8000);++ assert.ok(longInput.startsWith(`${'N'.repeat(2000)}\n\nContext:`), 'the note is never cut');++ tasks.length = 0; actorMode = 'ok';++ const note = 'N'.repeat(2000);++ const w = await create(person, { name: 'cheap widget', action: [{ kind: 'actor-agent', agent_id: agentId, note }] });++ await fire(w);++ let r = await svc.actions.tick();++ assert.strictEqual(r.sent, 1);++ assert.deepStrictEqual([tasks[0].subjectId, tasks[0].agentId], [person.sub, agentId]);++ assert.ok(tasks[0].input.startsWith(`${note}\n\nContext: your watch “cheap widget” fired at `));++ assert.match(tasks[0].input, /Value: 7\.50 \(was 19\.99\)\./);++ assert.ok(tasks[0].input.includes(`https://openvibe.watch/watches/${w.id}`));++ assert.ok(tasks[0].input.length <= 8000);++ assert.deepStrictEqual([(await acts(w.id))[0].state, (await acts(w.id))[0].last_error], ['sent', 'Actor task tsk_AGENT']);++++ for (const [kind, expected] of [['missing', /actor\.agent\.not_found/], ['limit', /actor\.agent\.limit/], ['free', /actor\.agent\.free_tier/], ['busy', /actor\.unavailable/]]) {++ actorMode = kind;++ const next = await create(person, { action: [{ kind: 'actor-agent', agent_id: agentId }] });++ await fire(next);++ r = await svc.actions.tick();++ const row = (await acts(next.id))[0];++ assert.strictEqual(row.state, kind === 'busy' ? 'queued' : 'failed');++ assert.match(row.last_error, expected);++ assert.strictEqual(kind === 'busy' ? r.retried : r.failed, 1);++ if (kind === 'busy') {++ // Actor is back: the queued run goes out on the next due tick, so the queue ends empty.++ actorMode = 'ok';++ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(next.id);++ r = await svc.actions.tick();++ assert.strictEqual(r.sent, 1);++ assert.strictEqual((await acts(next.id))[0].state, 'sent');++ }++ }++ actorMode = 'ok';});t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {@@@ -19,6 +19,7 @@ let session; // the cookie from a real sign-inconst create = (fields) => svc.get('/watches/new', { cookie: session, form: fields });const indexOf = (res) => (String(res.location || '').match(/\/watches\/(wch_[0-9A-HJKMNP-TV-Z]{26})/) || [])[1] || null;++const agentId = 'agt_01JAB2C3D4E5F6G7H8J9K0MNPQ';t('boot', async () => {web = await site({ '/price': () => ({ body: '<html><body><span class="price">19.99</span></body></html>' }) });@@@ -140,6 +141,26 @@ t('notify me off stores an event action instead of a notification, and says so',assert.deepStrictEqual(read.body.watch.action, [{ kind: 'event', event_type: 'watch.watch.triggered' }]);});++t('the plain create and edit forms save an Actor agent action', async () => {++ const form = await svc.get('/watches/new?template=page', { cookie: session });++ assert.match(form.text, /Run one of my agents on OpenVibe\.Actor/);++ assert.match(form.text, /name="agent_id"/);++ const made = await create({ template: 'page', url: `${web.origin}/price`, cadence: '3600', agent_on: 'on', agent_id: agentId, agent_note: 'Check this value.' });++ assert.strictEqual(made.status, 303);++ const id = indexOf(made);++ const view = await request(svc.base, 'GET', `/api/v1/watches/${id}`, { token: ME });++ assert.deepStrictEqual(view.body.watch.action, [{ kind: 'actor-agent', agent_id: agentId, note: 'Check this value.' }]);++ const page = await svc.get(`/watches/${id}`, { cookie: session });++ assert.ok(page.text.includes(`https://openvibe.actor/me/agents/${agentId}`));++ const edit = await svc.get(`/watches/${id}/edit`, { cookie: session });++ assert.match(edit.text, /name="agent_on"[^>]*checked/);++ assert.ok(edit.text.includes('Check this value.'));++ const saved = await svc.get(`/watches/${id}/edit`, { cookie: session, form: { template: 'page', url: `${web.origin}/price`, cadence: '3600', agent_on: 'on', agent_id: agentId, agent_note: 'New note.' } });++ assert.strictEqual(saved.status, 303);++ const updated = await request(svc.base, 'GET', `/api/v1/watches/${id}`, { token: ME });++ assert.strictEqual(updated.body.watch.action[0].note, 'New note.');++});++t('validation errors re-render the form with the registry\'s message and the person\'s input', async () => {const badUrl = await create({ template: 'page', name: 'Keep me', url: 'not-a-url', selector: '.x', cadence: '3600' });assert.strictEqual(badUrl.status, 422);
@@ -98,7 +98,7 @@ VALKEY_PREFIX=
@@ -98,7 +98,7 @@ VALKEY_PREFIX=
#WATCH_HOOK_EXAMPLE_SECRET=
#WATCH_HOOK_EXAMPLE_SECRET=
# A receiver answers within this, or the attempt is retried on the backoff.
# A receiver answers within this, or the attempt is retried on the backoff.
#WATCH_WEBHOOK_TIMEOUT_MS=10000
#WATCH_WEBHOOK_TIMEOUT_MS=10000
-# OpenVibe.Actor's internal address, for actor-task actions (Watch's token needs actor.task.create).
+# OpenVibe.Actor's internal address, for actor-task and actor-agent actions (Watch needs actor.task.create and actor.agent.run grants).
#WATCH_ACTOR_URL=http://127.0.0.1:4950
#WATCH_ACTOR_URL=http://127.0.0.1:4950
# Who may use which named secret (server/secret-grants.js). A name is not a capability: without a grant
# Who may use which named secret (server/secret-grants.js). A name is not a capability: without a grant
@@ -227,6 +227,7 @@ A fired condition queues the watch's actions in the same transaction as the chec
@@ -227,6 +227,7 @@ A fired condition queues the watch's actions in the same transaction as the chec
| `event` | `watch.watch.triggered` is emitted with every firing in the run's own commit, so that action is recorded `sent` as it is queued (the event names a person: an agent- or service-owned watch is `skipped`). Any other event type is not declared by Watch: `unavailable` |
| `event` | `watch.watch.triggered` is emitted with every firing in the run's own commit, so that action is recorded `sent` as it is queued (the event names a person: an agent- or service-owned watch is `skipped`). Any other event type is not declared by Watch: `unavailable` |
| `webhook` | a POST of the firing (`type`, `watch {id, name, url}`, `check_run_id`, `condition`, `observation {observed_at, value, previous_value}`) to the action's https URL, through the same SSRF guard as every fetch, never following a redirect ([server/net/post.js](server/net/post.js)). It is signed with the key `secret_env` names exactly as OpenVibe.Events signs a delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`; check it with openvibe-sdk/events `verifyDeliveryV2`). `X-OpenVibe-Delivery-Id` (`<check run>.<action index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 or a URL the guard refuses is `failed` at once. Anything else, or a lost connection, is retried on the backoff, then `failed`. The key must be granted to the owner (`WATCH_SECRET_GRANTS`, below) and set, 32 characters or more: otherwise `unavailable` |
| `webhook` | a POST of the firing (`type`, `watch {id, name, url}`, `check_run_id`, `condition`, `observation {observed_at, value, previous_value}`) to the action's https URL, through the same SSRF guard as every fetch, never following a redirect ([server/net/post.js](server/net/post.js)). It is signed with the key `secret_env` names exactly as OpenVibe.Events signs a delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`; check it with openvibe-sdk/events `verifyDeliveryV2`). `X-OpenVibe-Delivery-Id` (`<check run>.<action index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 or a URL the guard refuses is `failed` at once. Anything else, or a lost connection, is retried on the backoff, then `failed`. The key must be granted to the owner (`WATCH_SECRET_GRANTS`, below) and set, 32 characters or more: otherwise `unavailable` |
| `actor-task` | a task for the owner on OpenVibe.Actor ([server/actor.js](server/actor.js)): the action's `brief`, followed by the firing as context (the watch, the value and what it was, its page), at most `budget_usd`. Watch names the owner in `X-OV-Subject`, so the task runs under **the owner's** allowance and appears in their task list, and Actor records Watch as `via`. The idempotency key `watch.<check run>.<action index>` makes a retried delivery the same task. Created is `sent`, with the task id recorded. A request Actor refuses as invalid (over the tier's budget, an unknown agent) is `failed` at once. Anything else, including a day's allowance spent, is retried on the backoff. An agent- or service-owned watch is `skipped`. Needs `actor.task.create` for `openvibe.actor` (Network grants it) and `WATCH_ACTOR_URL` |
| `actor-task` | a task for the owner on OpenVibe.Actor ([server/actor.js](server/actor.js)): the action's `brief`, followed by the firing as context (the watch, the value and what it was, its page), at most `budget_usd`. Watch names the owner in `X-OV-Subject`, so the task runs under **the owner's** allowance and appears in their task list, and Actor records Watch as `via`. The idempotency key `watch.<check run>.<action index>` makes a retried delivery the same task. Created is `sent`, with the task id recorded. A request Actor refuses as invalid (over the tier's budget, an unknown agent) is `failed` at once. Anything else, including a day's allowance spent, is retried on the backoff. An agent- or service-owned watch is `skipped`. Needs `actor.task.create` for `openvibe.actor` (Network grants it) and `WATCH_ACTOR_URL` |
+| `actor-agent` | runs one of the person's OpenVibe.Actor agents by `agent_id`. Its optional `note` precedes the firing context (watch name and link, value and previous value, observation time); input is capped at 8000 characters without cutting the note. Watch uses `X-OV-Subject` for the owner. A created run is `sent` with its task id; a missing agent, limit or free-tier refusal is `failed` for that firing; service errors retry on the usual backoff. Only person-owned watches may use it. Needs the `actor.agent.run` grant for `openvibe.actor` and `WATCH_ACTOR_URL` |
| `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |
| `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |
Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.
Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.
@@ -487,7 +488,7 @@ POSTs have their own per-address limit zones (nginx refuses `limit_req` inside `
@@ -487,7 +488,7 @@ POSTs have their own per-address limit zones (nginx refuses `limit_req` inside `
`sudo ovhost deploy watch`.
`sudo ovhost deploy watch`.
<!-- versions:start -->
<!-- versions:start -->
-- openvibe-contracts: v0.129.0
+- openvibe-contracts: v0.143.0
- openvibe-sdk: v0.44.0
- openvibe-sdk: v0.44.0
- openvibe-shared: v4.0.2
- openvibe-shared: v4.0.2
<!-- versions:end -->
<!-- versions:end -->
@@ -7,7 +7,7 @@
@@ -7,7 +7,7 @@
"runtime": "node 22, express, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9, fast-xml-parser",
"runtime": "node 22, express, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9, fast-xml-parser",
"port": 4730,
"port": 4730,
"domain": "openvibe.watch (a public server-rendered site + a loopback-only API; vhost in deploy/nginx; README \"Public site\" and \"Public host\")",
"domain": "openvibe.watch (a public server-rendered site + a loopback-only API; vhost in deploy/nginx; README \"Public site\" and \"Public host\")",
- "contracts": "openvibe-contracts v0.129.0",
+ "contracts": "openvibe-contracts v0.143.0 (actor-agent requires v0.143.0)",
"sdk": "openvibe-sdk v0.44.0",
"sdk": "openvibe-sdk v0.44.0",
"shared": "openvibe-shared v4.0.2",
"shared": "openvibe-shared v4.0.2",
"capabilities": [
"capabilities": [
@@ -35,6 +35,15 @@
@@ -35,6 +35,15 @@
"run (step 6)",
"run (step 6)",
"node (step 7)"
"node (step 7)"
],
],
+ "actionKinds": [
+ "notification",
+ "event",
+ "webhook",
+ "actor-task",
+ "actor-agent",
+ "codes-run",
+ "run-job"
+ ],
"seededWatches": 0,
"seededWatches": 0,
"deployed": true,
"deployed": true,
"deployment": "openvibe-ovh: unit openvibe-watch.service on 127.0.0.1:4730 (release layout /opt/openvibe.watch, env /etc/openvibe/watch.env, database ov_watch), deployed with `sudo ovhost deploy watch`; the public site openvibe.watch (sign-in, watch management, plan T18 step 8) has been live since 2026-10-08. On 2026-10-10 it held no watches yet, so no check has run against a real site in production.",
"deployment": "openvibe-ovh: unit openvibe-watch.service on 127.0.0.1:4730 (release layout /opt/openvibe.watch, env /etc/openvibe/watch.env, database ov_watch), deployed with `sudo ovhost deploy watch`; the public site openvibe.watch (sign-in, watch management, plan T18 step 8) has been live since 2026-10-08. On 2026-10-10 it held no watches yet, so no check has run against a real site in production.",
@@ -12,7 +12,7 @@
@@ -12,7 +12,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.0.2",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.0.2",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -769,9 +769,9 @@
@@ -769,9 +769,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.129.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
- "integrity": "sha512-Ly0F/ap6wmAf8hYpNrOhMLsU3cZYBjsFDW2P1F/xVmHVsRDzAPAZDXtFcohP0UNeZgyxgrR0BI0pPG5guD/Efw==",
+ "version": "0.143.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
+ "integrity": "sha512-Cdbf+9RoPAUBibpM2zXQvtsiKMugb55dKubTt8pFZvAmxsLtGvX+zaC++yL1vOSub8FX3+GgqcGtoJS6tklrUg==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -24,7 +24,7 @@
@@ -24,7 +24,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.0.2",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.0.2",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -33,8 +33,8 @@
@@ -33,8 +33,8 @@
"node": ">=22.0.0"
"node": ">=22.0.0"
},
},
"devDependencies": {
"devDependencies": {
+ "@electric-sql/pglite": "^0.5.8",
"ajv": "^8.20.0",
"ajv": "^8.20.0",
- "ajv-formats": "^3.0.1",
- "@electric-sql/pglite": "^0.5.8"
+ "ajv-formats": "^3.0.1"
}
}
}
}
@@ -35,6 +35,8 @@
@@ -35,6 +35,8 @@
* Actor refuses as invalid (over the tier's budget, an unknown agent) is `failed` at once; anything
* Actor refuses as invalid (over the tier's budget, an unknown agent) is `failed` at once; anything
* else, including an allowance spent for the day, is retried on the backoff. It needs a person as
* else, including an allowance spent for the day, is retried on the backoff. It needs a person as
* owner: an agent- or service-owned watch is `skipped`.
* owner: an agent- or service-owned watch is `skipped`.
+ * actor-agent runs one of the owner's personal agents on Actor with a note and bounded firing context.
+ * A missing agent or allowance refusal fails this firing; a service error retries.
* codes-run, run-job
* codes-run, run-job
* `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and
* `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and
* the watch page says so rather than pretending they ran.
* the watch page says so rather than pretending they ran.
@@ -103,16 +105,26 @@ function webhookFor({ config, watch, run, value, previousValue, observedAt }) {
@@ -103,16 +105,26 @@ function webhookFor({ config, watch, run, value, previousValue, observedAt }) {
/** The task an actor-task action gives Actor: its brief, then what fired, so the agent knows why it was woken. */
/** The task an actor-task action gives Actor: its brief, then what fired, so the agent knows why it was woken. */
function actorTaskFor({ config, watch, action, value, previousValue, observedAt }) {
function actorTaskFor({ config, watch, action, value, previousValue, observedAt }) {
+ const context = actorContextFor({ config, watch, value, previousValue, observedAt });
+ const brief = String(action.brief || '').trim();
+ return `${brief.slice(0, TASK_MAX - context.length - 2)}\n\n${context}`;
+}
+
+function actorContextFor({ config, watch, value, previousValue, observedAt }, maxLength = Infinity) {
const now = valueText(value);
const now = valueText(value);
const before = previousValue === undefined || previousValue === null ? null : valueText(previousValue);
const before = previousValue === undefined || previousValue === null ? null : valueText(previousValue);
const when = observedAt ? new Date(observedAt).toISOString() : 'just now';
const when = observedAt ? new Date(observedAt).toISOString() : 'just now';
- const context = [
- `Context: your watch “${watch.name}” fired at ${when}.`,
- `Value: ${now}${before !== null && before !== now ? ` (was ${before})` : ''}.`,
- `Watch: ${config.baseUrl}/watches/${encodeURIComponent(watch.id)}`,
- ].join('\n');
- const brief = String(action.brief || '').trim();
- return `${brief.slice(0, TASK_MAX - context.length - 2)}\n\n${context}`;
+ const first = `Context: your watch “${watch.name}” fired at ${when}.`;
+ const last = `Watch: ${config.baseUrl}/watches/${encodeURIComponent(watch.id)}`;
+ const valueLine = `Value: ${now}${before !== null && before !== now ? ` (was ${before})` : ''}.`;
+ const room = Math.max(0, maxLength - first.length - last.length - 2);
+ return [first, valueLine.length > room ? `${valueLine.slice(0, Math.max(0, room - 1))}…` : valueLine, last].join('\n').slice(0, maxLength);
+}
+
+function actorAgentInputFor(args) {
+ const note = typeof args.action.note === 'string' ? args.action.note : '';
+ const prefix = note ? `${note}\n\n` : '';
+ return prefix + actorContextFor(args, 8000 - prefix.length);
}
}
function createActions({ db, config, notifications = null, post = null, actor = null, now = () => Date.now(), log = console }) {
function createActions({ db, config, notifications = null, post = null, actor = null, now = () => Date.now(), log = console }) {
@@ -159,6 +171,10 @@ function createActions({ db, config, notifications = null, post = null, actor =
@@ -159,6 +171,10 @@ function createActions({ db, config, notifications = null, post = null, actor =
if (USER_SUBJECT.test(String(owner || ''))) {
if (USER_SUBJECT.test(String(owner || ''))) {
request = { task: actorTaskFor({ config, watch, action, value, previousValue, observedAt }), budget_usd: action.budget_usd == null ? null : action.budget_usd };
request = { task: actorTaskFor({ config, watch, action, value, previousValue, observedAt }), budget_usd: action.budget_usd == null ? null : action.budget_usd };
} else { state = 'skipped'; error = 'watch.action_no_person: an Actor task runs under a person\'s allowance, and this watch belongs to an agent or a service'; }
} else { state = 'skipped'; error = 'watch.action_no_person: an Actor task runs under a person\'s allowance, and this watch belongs to an agent or a service'; }
+ } else if (kind === 'actor-agent') {
+ if (USER_SUBJECT.test(String(owner || ''))) {
+ request = { agent_id: action.agent_id, input: actorAgentInputFor({ config, watch, action, value, previousValue, observedAt }) };
+ } else { state = 'skipped'; error = 'watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service'; }
} else if (LATER.has(kind)) {
} else if (LATER.has(kind)) {
state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;
state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;
} else {
} else {
@@ -210,10 +226,25 @@ function createActions({ db, config, notifications = null, post = null, actor =
@@ -210,10 +226,25 @@ function createActions({ db, config, notifications = null, post = null, actor =
}
}
}
}
+ async function deliverActorAgent(row, request) {
+ if (!actor) throw Object.assign(new Error('Actor delivery is not configured in this process'), { code: 'actor_unconfigured' });
+ try {
+ const task = await actor.runAgent({ subjectId: row.owner_sub, agentId: request.agent_id, input: request.input });
+ return { state: 'sent', error: `Actor task ${task.id || '(no id)'}` };
+ } catch (err) {
+ if (err && err.status === 404 && err.code === 'actor.agent.not_found') return { state: 'failed', error: `watch.action_refused: ${err.code}: this agent was not found for the watch owner` };
+ if (err && (err.status === 429 || err.status === 400 || err.status === 422 || (err.status === 403 && /(?:limit|free.tier|over_tier)/i.test(String(err.code || ''))))) {
+ return { state: 'failed', error: `watch.action_refused: ${err.code || err.status}: ${String(err.detail || err.message || '').slice(0, 200)}` };
+ }
+ throw err;
+ }
+ }
+
async function deliver(row, attempt) {
async function deliver(row, attempt) {
const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;
const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;
if (row.kind === 'webhook') return await deliverWebhook(row, request, attempt);
if (row.kind === 'webhook') return await deliverWebhook(row, request, attempt);
if (row.kind === 'actor-task') return await deliverActorTask(row, request);
if (row.kind === 'actor-task') return await deliverActorTask(row, request);
+ if (row.kind === 'actor-agent') return await deliverActorAgent(row, request);
if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };
if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };
const out = await notifications.push({ subjectId: row.owner_sub, ...request });
const out = await notifications.push({ subjectId: row.owner_sub, ...request });
if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };
if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };
@@ -285,4 +316,4 @@ function createActions({ db, config, notifications = null, post = null, actor =
@@ -285,4 +316,4 @@ function createActions({ db, config, notifications = null, post = null, actor =
return { queue, tick, start, stop, status, recent, enabled };
return { queue, tick, start, stop, status, recent, enabled };
}
}
-module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, BACKOFF_MS };
+module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, actorAgentInputFor, BACKOFF_MS };
@@ -1,11 +1,13 @@
@@ -1,11 +1,13 @@
'use strict';
'use strict';
/**
/**
- * OpenVibe.Actor, for a watch's actor-task action (plan T18 step 6): give Actor a task FOR the watch's owner.
+ * OpenVibe.Actor, for a watch's actor-task and actor-agent actions: run work FOR the watch's owner.
*
*
* const actor = createActorTasks(client, { baseUrl }); // client: openvibe-sdk/core createClient with Watch's token
* const actor = createActorTasks(client, { baseUrl }); // client: openvibe-sdk/core createClient with Watch's token
* await actor.create({ subjectId, task, budgetUsd, idempotencyKey }) → { id, state }
* await actor.create({ subjectId, task, budgetUsd, idempotencyKey }) → { id, state }
+ * await actor.runAgent({ subjectId, agentId, input }) → { id, state }
*
*
- * POST <actor>/api/v1/tasks with Watch's own token (actor.task.create, audience openvibe.actor) and the owner in
+ * POST <actor>/api/v1/tasks with Watch's own token (actor.task.create, audience openvibe.actor), or
+ * POST <actor>/api/v1/me/agents/:agentId/run (actor.agent.run), with the owner in
* X-OV-Subject: Actor makes the task the person's (their allowance, their daily budget, their task list) and records
* X-OV-Subject: Actor makes the task the person's (their allowance, their daily budget, their task list) and records
* Watch as `via` (Actor#25). The idempotency key makes a retried delivery the same task, never a second one. The call
* Watch as `via` (Actor#25). The idempotency key makes a retried delivery the same task, never a second one. The call
* is made once; the action worker retries on its own backoff. Actor's refusals arrive as openvibe-sdk OpenVibeError
* is made once; the action worker retries on its own backoff. Actor's refusals arrive as openvibe-sdk OpenVibeError
@@ -13,15 +15,20 @@
@@ -13,15 +15,20 @@
*/
*/
function createActorTasks(client, { baseUrl }) {
function createActorTasks(client, { baseUrl }) {
+ const post = (path, subjectId, json) => client.json({
+ service: 'actor', baseUrl, audience: 'openvibe.actor', method: 'POST', path,
+ headers: { 'X-OV-Subject': subjectId }, json, retries: 0, idempotencyKey: false,
+ });
async function create({ subjectId, task, budgetUsd = null, idempotencyKey }) {
async function create({ subjectId, task, budgetUsd = null, idempotencyKey }) {
const json = { task, idempotency_key: idempotencyKey, ...(budgetUsd == null ? {} : { budget: { per_task_usd: budgetUsd } }) };
const json = { task, idempotency_key: idempotencyKey, ...(budgetUsd == null ? {} : { budget: { per_task_usd: budgetUsd } }) };
- const out = await client.json({
- service: 'actor', baseUrl, audience: 'openvibe.actor', method: 'POST', path: '/api/v1/tasks',
- headers: { 'X-OV-Subject': subjectId }, json, retries: 0, idempotencyKey: false,
- });
+ const out = await post('/api/v1/tasks', subjectId, json);
return { id: out && out.id, state: out && out.state };
return { id: out && out.id, state: out && out.state };
}
}
- return { create };
+ async function runAgent({ subjectId, agentId, input }) {
+ const out = await post(`/api/v1/me/agents/${encodeURIComponent(agentId)}/run`, subjectId, { input });
+ return { id: out && (out.task_id || (out.task && out.task.id) || out.id), state: out && out.state };
+ }
+ return { create, runAgent };
}
}
module.exports = { createActorTasks };
module.exports = { createActorTasks };
@@ -99,7 +99,7 @@ function load(env = process.env) {
@@ -99,7 +99,7 @@ function load(env = process.env) {
batch: int(env.WATCH_ACTIONS_BATCH, 20),
batch: int(env.WATCH_ACTIONS_BATCH, 20),
// A webhook receiver answers within this, or the attempt is retried on the backoff.
// A webhook receiver answers within this, or the attempt is retried on the backoff.
webhookTimeoutMs: int(env.WATCH_WEBHOOK_TIMEOUT_MS, 10000),
webhookTimeoutMs: int(env.WATCH_WEBHOOK_TIMEOUT_MS, 10000),
- // OpenVibe.Actor's internal address, for actor-task actions (Watch's token needs actor.task.create).
+ // OpenVibe.Actor's internal address, for actor-task and actor-agent actions.
actorUrl: (env.WATCH_ACTOR_URL || 'http://127.0.0.1:4950').replace(/\/+$/, ''),
actorUrl: (env.WATCH_ACTOR_URL || 'http://127.0.0.1:4950').replace(/\/+$/, ''),
},
},
@@ -121,7 +121,7 @@ function createPageRoutes(ctx) {
@@ -121,7 +121,7 @@ function createPageRoutes(ctx) {
template, name: '', url: '', kind: t.kind,
template, name: '', url: '', kind: t.kind,
extraction: t.extraction, selector: '', value_path: t.valuePath || '',
extraction: t.extraction, selector: '', value_path: t.valuePath || '',
op: t.defaultOp || t.ops[0], value: '', for_sec: '',
op: t.defaultOp || t.ops[0], value: '', for_sec: '',
- cadence: '3600', notify: 'on',
+ cadence: '3600', notify: 'on', agent_on: '', agent_id: '', agent_note: '',
};
};
}
}
@@ -132,6 +132,7 @@ function createPageRoutes(ctx) {
@@ -132,6 +132,7 @@ function createPageRoutes(ctx) {
function postedFields(template, body) {
function postedFields(template, body) {
const fields = { ...blankFields(template), ...(body || {}) };
const fields = { ...blankFields(template), ...(body || {}) };
fields.notify = body && body.notify === 'on' ? 'on' : '';
fields.notify = body && body.notify === 'on' ? 'on' : '';
+ fields.agent_on = body && body.agent_on === 'on' ? 'on' : '';
return fields;
return fields;
}
}
@@ -163,9 +164,25 @@ function createPageRoutes(ctx) {
@@ -163,9 +164,25 @@ function createPageRoutes(ctx) {
for_sec: (watch.condition && Number.isInteger(watch.condition.for_sec)) ? String(watch.condition.for_sec) : '',
for_sec: (watch.condition && Number.isInteger(watch.condition.for_sec)) ? String(watch.condition.for_sec) : '',
cadence: String((watch.cadence && watch.cadence.every_sec) || 3600),
cadence: String((watch.cadence && watch.cadence.every_sec) || 3600),
notify: notifyOn(watch.action) ? 'on' : '',
notify: notifyOn(watch.action) ? 'on' : '',
+ agent_on: agentAction(watch.action) ? 'on' : '',
+ agent_id: (agentAction(watch.action) || {}).agent_id || '',
+ agent_note: (agentAction(watch.action) || {}).note || '',
};
};
}
}
const notifyOn = (action) => Array.isArray(action) && action.some((a) => a && a.kind === 'notification');
const notifyOn = (action) => Array.isArray(action) && action.some((a) => a && a.kind === 'notification');
+ const agentAction = (action) => Array.isArray(action) ? action.find((a) => a && a.kind === 'actor-agent') : null;
+
+ function actionsFrom(fields, errors) {
+ const action = fields.notify === 'on' ? [...NOTIFY_ACTION] : [];
+ if (fields.agent_on === 'on') {
+ const agentId = String(fields.agent_id || '').trim();
+ const note = String(fields.agent_note || '');
+ if (!/^agt_[0-9A-HJKMNP-TV-Z]{26}$/.test(agentId)) errors.agent_id = 'Give a valid Actor agent id (agt_ followed by 26 characters).';
+ if (note.length > 2000) errors.agent_note = 'The note can be at most 2000 characters.';
+ action.push({ kind: 'actor-agent', agent_id: agentId, ...(note ? { note } : {}) });
+ }
+ return action.length ? action : QUIET_ACTION;
+ }
/** The condition a form says, checked against the template's ops. Returns { condition } or { error }. */
/** The condition a form says, checked against the template's ops. Returns { condition } or { error }. */
function conditionFrom(fields, template) {
function conditionFrom(fields, template) {
@@ -198,6 +215,7 @@ function createPageRoutes(ctx) {
@@ -198,6 +215,7 @@ function createPageRoutes(ctx) {
function buildBody(fields, template) {
function buildBody(fields, template) {
const t = TEMPLATES[template];
const t = TEMPLATES[template];
const errors = {};
const errors = {};
+ const action = actionsFrom(fields, errors);
// A webhook watch has no address to fetch and no cadence: it is pushed to (server/webhook-sources.js).
// A webhook watch has no address to fetch and no cadence: it is pushed to (server/webhook-sources.js).
if (template === 'webhook') {
if (template === 'webhook') {
const name = String(fields.name == null ? '' : fields.name).trim();
const name = String(fields.name == null ? '' : fields.name).trim();
@@ -211,7 +229,7 @@ function createPageRoutes(ctx) {
@@ -211,7 +229,7 @@ function createPageRoutes(ctx) {
cadence: null,
cadence: null,
extraction: valuePath ? { kind: 'json', value_path: valuePath } : { kind: 'text' },
extraction: valuePath ? { kind: 'json', value_path: valuePath } : { kind: 'text' },
condition: c.condition,
condition: c.condition,
- action: fields.notify === 'on' ? NOTIFY_ACTION : QUIET_ACTION,
+ action,
};
};
if (name) body.name = name;
if (name) body.name = name;
return { body };
return { body };
@@ -252,7 +270,7 @@ function createPageRoutes(ctx) {
@@ -252,7 +270,7 @@ function createPageRoutes(ctx) {
cadence: { every_sec: cadenceSec, jitter_sec: 0 },
cadence: { every_sec: cadenceSec, jitter_sec: 0 },
extraction,
extraction,
condition: c.condition,
condition: c.condition,
- action: fields.notify === 'on' ? NOTIFY_ACTION : QUIET_ACTION,
+ action,
};
};
if (name) body.name = name; // omitted, not undefined: the registry names it from the source
if (name) body.name = name; // omitted, not undefined: the registry names it from the source
return { body };
return { body };
@@ -265,8 +283,8 @@ function createPageRoutes(ctx) {
@@ -265,8 +283,8 @@ function createPageRoutes(ctx) {
*/
*/
function registryField(message) {
function registryField(message) {
const m = String(message);
const m = String(message);
- const named = m.match(/\/(?:source|extraction|condition|cadence|name|action)\/(url|selector|value_path|value|for_sec|every_sec)\b/);
- if (named) return named[1] === 'every_sec' ? 'cadence' : named[1];
+ const named = m.match(/\/(?:source|extraction|condition|cadence|name|action)(?:\/\d+)?\/(url|selector|value_path|value|for_sec|every_sec|agent_id|note)\b/);
+ if (named) return named[1] === 'every_sec' ? 'cadence' : (named[1] === 'note' ? 'agent_note' : named[1]);
if (/\/name\b/.test(m)) return 'name';
if (/\/name\b/.test(m)) return 'name';
if (/\/condition\b/.test(m)) return 'op';
if (/\/condition\b/.test(m)) return 'op';
if (/\/cadence\b/.test(m)) return 'cadence';
if (/\/cadence\b/.test(m)) return 'cadence';
@@ -363,9 +381,12 @@ ${conditionBlock}
@@ -363,9 +381,12 @@ ${conditionBlock}
<fieldset><legend>${pushed ? 'How to tell you' : 'How often, and how to tell you'}</legend>
<fieldset><legend>${pushed ? 'How to tell you' : 'How often, and how to tell you'}</legend>
${pushed ? html`<p class="muted small">Checked each time a signed delivery arrives (at most 60 an hour), never on a cadence.</p>` : field('Cadence', cadence, { name: 'cadence', error: errors.cadence, hint: 'Watch never checks more often than this, and can wait longer when the site asks it to.' })}
${pushed ? html`<p class="muted small">Checked each time a signed delivery arrives (at most 60 an hour), never on a cadence.</p>` : field('Cadence', cadence, { name: 'cadence', error: errors.cadence, hint: 'Watch never checks more often than this, and can wait longer when the site asks it to.' })}
${checkbox('Notify me when it fires', { name: 'notify', checked: fields.notify === 'on', hint: 'When the condition fires you get one notification on OpenVibe (your notifications on any OpenVibe site), with the value and a link here.' })}
${checkbox('Notify me when it fires', { name: 'notify', checked: fields.notify === 'on', hint: 'When the condition fires you get one notification on OpenVibe (your notifications on any OpenVibe site), with the value and a link here.' })}
+${checkbox('Run one of my agents on OpenVibe.Actor', { name: 'agent_on', checked: fields.agent_on === 'on', hint: 'When the condition fires, Watch gives your agent the value and a link to this watch.' })}
+${field('Agent id', html`<input type="text" id="${idOf('agent_id')}" name="agent_id" value="${fields.agent_id}" maxlength="30" placeholder="agt_…">`, { name: 'agent_id', error: errors.agent_id, hint: 'Find the id on your agent’s OpenVibe.Actor page. Required when running an agent.' })}
+${field('Note for the agent (optional)', html`<textarea id="${idOf('agent_note')}" name="agent_note" maxlength="2000" rows="3">${fields.agent_note}</textarea>`, { name: 'agent_note', error: errors.agent_note })}
</fieldset>
</fieldset>
<p class="form-actions"><button class="btn primary" type="submit">${submit}</button> <a href="/watches">Cancel</a></p>
<p class="form-actions"><button class="btn primary" type="submit">${submit}</button> <a href="/watches">Cancel</a></p>
-<p class="muted small">Notifications are delivered today. Events, webhooks and tasks for OpenVibe.Actor, Codes and Run are later steps: a watch that asks for one records it as not available yet.</p>
+<p class="muted small">Notifications and Actor agents are delivered when the watch fires. Codes and Run actions are later steps.</p>
</form>`;
</form>`;
}
}
@@ -558,6 +579,9 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: 'Create
@@ -558,6 +579,9 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: 'Create
const obs = await observations.list(row.id, { before: ridBefore(req.query.obs_before), limit: obsLimit(req.query.obs_limit) });
const obs = await observations.list(row.id, { before: ridBefore(req.query.obs_before), limit: obsLimit(req.query.obs_limit) });
const runs = await check.list(row.id, { before: ridBefore(req.query.checks_before), limit: obsLimit(req.query.checks_limit) });
const runs = await check.list(row.id, { before: ridBefore(req.query.checks_before), limit: obsLimit(req.query.checks_limit) });
const acts = actions ? await actions.recent(row.id, 10) : [];
const acts = actions ? await actions.recent(row.id, 10) : [];
+ const configuredActions = (v.action || []).map((a) => a.kind === 'actor-agent'
+ ? html`<div>Run agent <a href="https://openvibe.actor/me/agents/${a.agent_id}"><code>${a.agent_id}</code></a>${a.note ? html` — ${a.note}` : ''}</div>`
+ : html`<div><code>${a.kind}</code></div>`);
const obsRows = obs.map((o) => [
const obsRows = obs.map((o) => [
time(new Date(Number(o.observed_at)).toISOString()),
time(new Date(Number(o.observed_at)).toISOString()),
@@ -609,6 +633,7 @@ ${v.source.kind === 'webhook'
@@ -609,6 +633,7 @@ ${v.source.kind === 'webhook'
? html`<dt>Webhook address</dt><dd><code>${config.baseUrl}/hooks/${v.id}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a new secret</button></form><br><span class="muted small">The secret was shown once, when the watch was made. A new one replaces it at once.</span></dd>`
? html`<dt>Webhook address</dt><dd><code>${config.baseUrl}/hooks/${v.id}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a new secret</button></form><br><span class="muted small">The secret was shown once, when the watch was made. A new one replaces it at once.</span></dd>`
: html`<dt>Source</dt><dd><code>${v.source.kind}</code> · <code>${v.source.url || (v.source.pattern ? v.source.pattern : '(none)')}</code></dd>`}
: html`<dt>Source</dt><dd><code>${v.source.kind}</code> · <code>${v.source.url || (v.source.pattern ? v.source.pattern : '(none)')}</code></dd>`}
<dt>Cadence</dt><dd>${v.cadence ? W.everyWords(v.cadence.every_sec) : (v.source.kind === 'webhook' ? 'when a delivery arrives' : (v.source.kind === 'event' ? 'when its event arrives' : 'not polled'))}</dd>
<dt>Cadence</dt><dd>${v.cadence ? W.everyWords(v.cadence.every_sec) : (v.source.kind === 'webhook' ? 'when a delivery arrives' : (v.source.kind === 'event' ? 'when its event arrives' : 'not polled'))}</dd>
+<dt>Actions</dt><dd>${configuredActions}</dd>
<dt>Last check</dt><dd>${v.health.last_check_at ? html`${time(v.health.last_check_at)} — ${W.stateWords(v.health.last_state)}` : html`<span class="muted">not checked yet</span>`}</dd>
<dt>Last check</dt><dd>${v.health.last_check_at ? html`${time(v.health.last_check_at)} — ${W.stateWords(v.health.last_state)}` : html`<span class="muted">not checked yet</span>`}</dd>
<dt>Next due</dt><dd>${v.health.next_due_at ? time(v.health.next_due_at) : html`<span class="muted">—</span>`}</dd>
<dt>Next due</dt><dd>${v.health.next_due_at ? time(v.health.next_due_at) : html`<span class="muted">—</span>`}</dd>
${v.health.consecutive_failures ? html`<dt>Failing</dt><dd>${v.health.consecutive_failures} check(s) in a row</dd>` : ''}
${v.health.consecutive_failures ? html`<dt>Failing</dt><dd>${v.health.consecutive_failures} check(s) in a row</dd>` : ''}
@@ -760,6 +785,7 @@ ${watchForm({ template: fields.template, fields, errors: {}, action: `/watches/$
@@ -760,6 +785,7 @@ ${watchForm({ template: fields.template, fields, errors: {}, action: `/watches/$
const template = getTemplate((req.body || {}).template);
const template = getTemplate((req.body || {}).template);
const fields = { ...fieldsOfWatch(registry.view(row)), ...(req.body || {}) };
const fields = { ...fieldsOfWatch(registry.view(row)), ...(req.body || {}) };
fields.notify = req.body && req.body.notify === 'on' ? 'on' : '';
fields.notify = req.body && req.body.notify === 'on' ? 'on' : '';
+ fields.agent_on = req.body && req.body.agent_on === 'on' ? 'on' : '';
const built = buildBody(fields, template);
const built = buildBody(fields, template);
let errors = built.errors;
let errors = built.errors;
if (!errors) {
if (!errors) {
@@ -30,6 +30,8 @@ const STATUSES = ['active', 'paused', 'disabled', 'failed'];
@@ -30,6 +30,8 @@ const STATUSES = ['active', 'paused', 'disabled', 'failed'];
const MIN_EVERY_SEC = 900;
const MIN_EVERY_SEC = 900;
/** Headers the fetcher decides; a watch may never override them (SSRF and conditional-GET safety). */
/** Headers the fetcher decides; a watch may never override them (SSRF and conditional-GET safety). */
const FORBIDDEN_HEADERS = new Set(['host', 'cookie', 'connection', 'content-length', 'transfer-encoding', 'user-agent', 'accept-encoding', 'if-none-match', 'if-modified-since']);
const FORBIDDEN_HEADERS = new Set(['host', 'cookie', 'connection', 'content-length', 'transfer-encoding', 'user-agent', 'accept-encoding', 'if-none-match', 'if-modified-since']);
+const PERSON = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const AGENT_ID = /^agt_[0-9A-HJKMNP-TV-Z]{26}$/;
class RegistryError extends Error {
class RegistryError extends Error {
constructor(detail, code = 'watch.invalid') { super(detail); this.code = code; }
constructor(detail, code = 'watch.invalid') { super(detail); this.code = code; }
@@ -59,6 +61,14 @@ function defaultName(source) {
@@ -59,6 +61,14 @@ function defaultName(source) {
*/
*/
function validateRequest(body, { create = false } = {}) {
function validateRequest(body, { create = false } = {}) {
if (!body || typeof body !== 'object' || Array.isArray(body)) throw new RegistryError('body must be a watch object');
if (!body || typeof body !== 'object' || Array.isArray(body)) throw new RegistryError('body must be a watch object');
+ for (const [i, action] of (Array.isArray(body.action) ? body.action : []).entries()) {
+ if (action && action.kind === 'actor-agent') {
+ if (!AGENT_ID.test(action.agent_id)) throw new RegistryError(`/action/${i}/agent_id must be an Actor agent id`);
+ if (action.note !== undefined && action.note !== null && (typeof action.note !== 'string' || action.note.length > 2000)) {
+ throw new RegistryError(`/action/${i}/note must be a string of at most 2000 characters or null`);
+ }
+ }
+ }
const v = validate('watch.watch-request@1', body);
const v = validate('watch.watch-request@1', body);
if (!v.valid) throw new RegistryError(`body does not match watch.watch-request@1: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`);
if (!v.valid) throw new RegistryError(`body does not match watch.watch-request@1: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`);
if (create) {
if (create) {
@@ -129,6 +139,11 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -129,6 +139,11 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
const bad = secretGrants.check(owner, rec);
const bad = secretGrants.check(owner, rec);
if (bad.length) throw new RegistryError(`${bad[0].path}: ${bad[0].reason}`, 'watch.secret_not_granted');
if (bad.length) throw new RegistryError(`${bad[0].path}: ${bad[0].reason}`, 'watch.secret_not_granted');
}
}
+ function checkActorAgents(owner, rec) {
+ if (!PERSON.test(String(owner || '')) && Array.isArray(rec.action) && rec.action.some((a) => a && a.kind === 'actor-agent')) {
+ throw new RegistryError('watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service');
+ }
+ }
const st = {
const st = {
get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
@@ -249,6 +264,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -249,6 +264,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
const owner = ownerOf(principal);
const owner = ownerOf(principal);
if (!owner) throw new RegistryError('no acting subject', 'watch.invalid');
if (!owner) throw new RegistryError('no acting subject', 'watch.invalid');
checkSecrets(owner, req);
checkSecrets(owner, req);
+ checkActorAgents(owner, req);
// A per-owner cap, so one caller cannot fill the registry (the API and the form both come
// A per-owner cap, so one caller cannot fill the registry (the API and the form both come
// through here). The number is a config value (WATCH_MAX_PER_OWNER).
// through here). The number is a config value (WATCH_MAX_PER_OWNER).
const owned = await st.countOwned.get(owner);
const owned = await st.countOwned.get(owner);
@@ -317,6 +333,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -317,6 +333,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
if (changes.project_id === undefined) delete merged.project_id;
if (changes.project_id === undefined) delete merged.project_id;
validateRequest(merged);
validateRequest(merged);
checkSecrets(row.owner_sub, merged);
checkSecrets(row.owner_sub, merged);
+ checkActorAgents(row.owner_sub, merged);
const sets = [];
const sets = [];
const args = { id: row.id, updated_at: now(), updated_by: ownerOf(principal) };
const args = { id: row.id, updated_at: now(), updated_by: ownerOf(principal) };
for (const [k, v] of Object.entries(changes)) {
for (const [k, v] of Object.entries(changes)) {
@@ -53,7 +53,16 @@ const actor = {
@@ -53,7 +53,16 @@ const actor = {
if (actorMode === 'busy') throw Object.assign(new Error('capacity'), { status: 503, code: 'actor.capacity.spent' });
if (actorMode === 'busy') throw Object.assign(new Error('capacity'), { status: 503, code: 'actor.capacity.spent' });
return { id: `tsk_01JAB2C3D4E5F6G7H8J9K0MN${String(tasks.length).padStart(2, '0')}`, state: 'queued' };
return { id: `tsk_01JAB2C3D4E5F6G7H8J9K0MN${String(tasks.length).padStart(2, '0')}`, state: 'queued' };
},
},
+ async runAgent(req) {
+ tasks.push(req);
+ if (actorMode === 'missing') throw Object.assign(new Error('not found'), { status: 404, code: 'actor.agent.not_found' });
+ if (actorMode === 'limit') throw Object.assign(new Error('free tier limit'), { status: 429, code: 'actor.agent.limit' });
+ if (actorMode === 'free') throw Object.assign(new Error('free tier refusal'), { status: 403, code: 'actor.agent.free_tier' });
+ if (actorMode === 'busy') throw Object.assign(new Error('capacity'), { status: 503, code: 'actor.unavailable' });
+ return { id: 'tsk_AGENT', state: 'queued' };
+ },
};
};
+const agentId = 'agt_01JAB2C3D4E5F6G7H8J9K0MNPQ';
const def = (overrides = {}) => ({
const def = (overrides = {}) => ({
name: 'Widget price',
name: 'Widget price',
source: { kind: 'http', url: `${web.origin}/price`, format: null },
source: { kind: 'http', url: `${web.origin}/price`, format: null },
@@ -318,6 +327,62 @@ t('server/actor.js asks Actor as Watch, for the owner, once', async () => {
@@ -318,6 +327,62 @@ t('server/actor.js asks Actor as Watch, for the owner, once', async () => {
assert.deepStrictEqual(c.json, { task: 'do it', idempotency_key: 'watch.r.0', budget: { per_task_usd: 0.2 } });
assert.deepStrictEqual(c.json, { task: 'do it', idempotency_key: 'watch.r.0', budget: { per_task_usd: 0.2 } });
await createActorTasks(client, { baseUrl: 'x' }).create({ subjectId: person.sub, task: 'no budget', idempotencyKey: 'watch.r.1' });
await createActorTasks(client, { baseUrl: 'x' }).create({ subjectId: person.sub, task: 'no budget', idempotencyKey: 'watch.r.1' });
assert.ok(!('budget' in calls[1].json), 'no budget named: Actor applies the tier\'s');
assert.ok(!('budget' in calls[1].json), 'no budget named: Actor applies the tier\'s');
+ const agentRun = await createActorTasks(client, { baseUrl: 'http://127.0.0.1:4950' }).runAgent({ subjectId: person.sub, agentId, input: 'hello' });
+ assert.deepStrictEqual(agentRun, { id: 'tsk_X', state: 'queued' });
+ assert.deepStrictEqual([calls[2].baseUrl, calls[2].audience, calls[2].path, calls[2].retries],
+ ['http://127.0.0.1:4950', 'openvibe.actor', `/api/v1/me/agents/${agentId}/run`, 0]);
+ assert.deepStrictEqual(calls[2].headers, { 'X-OV-Subject': person.sub });
+ assert.deepStrictEqual(calls[2].json, { input: 'hello' });
+});
+
+t('actor-agent validation checks id, note and person ownership', async () => {
+ const action = [{ kind: 'actor-agent', agent_id: agentId, note: 'Review this change.' }];
+ const w = await create(person, { action });
+ assert.deepStrictEqual(svc.registry.view(w).action, action);
+ await assert.rejects(create(person, { action: [{ ...action[0], agent_id: 'agt_bad' }] }), /agent_id/);
+ await assert.rejects(create(person, { action: [{ ...action[0], note: 'x'.repeat(2001) }] }), /note/);
+ await assert.rejects(create(agent, { action }), (e) => e.code === 'watch.invalid' && /action_no_person/.test(e.message));
+ await assert.rejects(create({ sub: 'svc:watch-test' }, { action }), (e) => e.code === 'watch.invalid' && /action_no_person/.test(e.message));
+ await assert.rejects(svc.registry.patch(w.id, { action: [{ ...action[0], agent_id: 'agt_bad' }] }, person), /agent_id/);
+});
+
+t('actor-agent sends note and bounded firing context, records its task, and classifies refusals', async () => {
+ const { actorAgentInputFor } = require('../server/actions');
+ const longInput = actorAgentInputFor({ config: svc.config, watch: { name: 'long value', id: 'wch_x' }, action: { note: 'N'.repeat(2000) }, value: 'V'.repeat(20000), previousValue: 'before', observedAt: Date.now() });
+ assert.strictEqual(longInput.length, 8000);
+ assert.ok(longInput.startsWith(`${'N'.repeat(2000)}\n\nContext:`), 'the note is never cut');
+ tasks.length = 0; actorMode = 'ok';
+ const note = 'N'.repeat(2000);
+ const w = await create(person, { name: 'cheap widget', action: [{ kind: 'actor-agent', agent_id: agentId, note }] });
+ await fire(w);
+ let r = await svc.actions.tick();
+ assert.strictEqual(r.sent, 1);
+ assert.deepStrictEqual([tasks[0].subjectId, tasks[0].agentId], [person.sub, agentId]);
+ assert.ok(tasks[0].input.startsWith(`${note}\n\nContext: your watch “cheap widget” fired at `));
+ assert.match(tasks[0].input, /Value: 7\.50 \(was 19\.99\)\./);
+ assert.ok(tasks[0].input.includes(`https://openvibe.watch/watches/${w.id}`));
+ assert.ok(tasks[0].input.length <= 8000);
+ assert.deepStrictEqual([(await acts(w.id))[0].state, (await acts(w.id))[0].last_error], ['sent', 'Actor task tsk_AGENT']);
+
+ for (const [kind, expected] of [['missing', /actor\.agent\.not_found/], ['limit', /actor\.agent\.limit/], ['free', /actor\.agent\.free_tier/], ['busy', /actor\.unavailable/]]) {
+ actorMode = kind;
+ const next = await create(person, { action: [{ kind: 'actor-agent', agent_id: agentId }] });
+ await fire(next);
+ r = await svc.actions.tick();
+ const row = (await acts(next.id))[0];
+ assert.strictEqual(row.state, kind === 'busy' ? 'queued' : 'failed');
+ assert.match(row.last_error, expected);
+ assert.strictEqual(kind === 'busy' ? r.retried : r.failed, 1);
+ if (kind === 'busy') {
+ // Actor is back: the queued run goes out on the next due tick, so the queue ends empty.
+ actorMode = 'ok';
+ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(next.id);
+ r = await svc.actions.tick();
+ assert.strictEqual(r.sent, 1);
+ assert.strictEqual((await acts(next.id))[0].state, 'sent');
+ }
+ }
+ actorMode = 'ok';
});
});
t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {
t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {
@@ -19,6 +19,7 @@ let session; // the cookie from a real sign-in
@@ -19,6 +19,7 @@ let session; // the cookie from a real sign-in
const create = (fields) => svc.get('/watches/new', { cookie: session, form: fields });
const create = (fields) => svc.get('/watches/new', { cookie: session, form: fields });
const indexOf = (res) => (String(res.location || '').match(/\/watches\/(wch_[0-9A-HJKMNP-TV-Z]{26})/) || [])[1] || null;
const indexOf = (res) => (String(res.location || '').match(/\/watches\/(wch_[0-9A-HJKMNP-TV-Z]{26})/) || [])[1] || null;
+const agentId = 'agt_01JAB2C3D4E5F6G7H8J9K0MNPQ';
t('boot', async () => {
t('boot', async () => {
web = await site({ '/price': () => ({ body: '<html><body><span class="price">19.99</span></body></html>' }) });
web = await site({ '/price': () => ({ body: '<html><body><span class="price">19.99</span></body></html>' }) });
@@ -140,6 +141,26 @@ t('notify me off stores an event action instead of a notification, and says so',
@@ -140,6 +141,26 @@ t('notify me off stores an event action instead of a notification, and says so',
assert.deepStrictEqual(read.body.watch.action, [{ kind: 'event', event_type: 'watch.watch.triggered' }]);
assert.deepStrictEqual(read.body.watch.action, [{ kind: 'event', event_type: 'watch.watch.triggered' }]);
});
});
+t('the plain create and edit forms save an Actor agent action', async () => {
+ const form = await svc.get('/watches/new?template=page', { cookie: session });
+ assert.match(form.text, /Run one of my agents on OpenVibe\.Actor/);
+ assert.match(form.text, /name="agent_id"/);
+ const made = await create({ template: 'page', url: `${web.origin}/price`, cadence: '3600', agent_on: 'on', agent_id: agentId, agent_note: 'Check this value.' });
+ assert.strictEqual(made.status, 303);
+ const id = indexOf(made);
+ const view = await request(svc.base, 'GET', `/api/v1/watches/${id}`, { token: ME });
+ assert.deepStrictEqual(view.body.watch.action, [{ kind: 'actor-agent', agent_id: agentId, note: 'Check this value.' }]);
+ const page = await svc.get(`/watches/${id}`, { cookie: session });
+ assert.ok(page.text.includes(`https://openvibe.actor/me/agents/${agentId}`));
+ const edit = await svc.get(`/watches/${id}/edit`, { cookie: session });
+ assert.match(edit.text, /name="agent_on"[^>]*checked/);
+ assert.ok(edit.text.includes('Check this value.'));
+ const saved = await svc.get(`/watches/${id}/edit`, { cookie: session, form: { template: 'page', url: `${web.origin}/price`, cadence: '3600', agent_on: 'on', agent_id: agentId, agent_note: 'New note.' } });
+ assert.strictEqual(saved.status, 303);
+ const updated = await request(svc.base, 'GET', `/api/v1/watches/${id}`, { token: ME });
+ assert.strictEqual(updated.body.watch.action[0].note, 'New note.');
+});
+
t('validation errors re-render the form with the registry\'s message and the person\'s input', async () => {
t('validation errors re-render the form with the registry\'s message and the person\'s input', async () => {
const badUrl = await create({ template: 'page', name: 'Keep me', url: 'not-a-url', selector: '.x', cadence: '3600' });
const badUrl = await create({ template: 'page', name: 'Keep me', url: 'not-a-url', selector: '.x', cadence: '3600' });
assert.strictEqual(badUrl.status, 422);
assert.strictEqual(badUrl.status, 422);