openvibers/openvibe.watch Commits
Watch: actor-task actions give Actor a task for the owner (T18 step 6) (#34)
Changes
@@@ -98,6 +98,8 @@ 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).++#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# nobody can use it. A credential needs owners and the origins it may be sent to; a webhook key, owners.@@@ -178,7 +178,8 @@ A fired condition queues the watch's actions in the same transaction as the chec| `notification` | the owner gets one OpenVibe notification through OpenVibe.Network (`openvibe-sdk/notifications`, type `WATCH_TRIGGERED`): the watch's name, the value (and the value before), and a link to the watch page. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then `failed`; an owner Network does not know (a deleted account) is `skipped`; a person who switched the category off is Network's decision and counts as `sent`. A watch owned by an agent or a service has no one to notify (`skipped`). Needs the Network OAuth client secret and the grants `identity.subject.resolve` + `network.notifications.push` || `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`, `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |++| `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` |++| `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.Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.@@@ -28,7 +28,14 @@* The key must be granted to the watch's owner (WATCH_SECRET_GRANTS, server/secret-grants.js) and* set (32 characters or more), checked when it is queued and again at delivery: otherwise* `unavailable`.-- * actor-task, codes-run, run-job++ * actor-task a task for the owner on OpenVibe.Actor (server/actor.js): the action's brief, with the firing as++ * context, under the owner's own allowance (Watch names them in X-OV-Subject; Actor records Watch as++ * `via`), at most the action's budget_usd. The idempotency key `watch.<check run>.<action index>`++ * makes a retried delivery the same task. Created is `sent` (the task id is recorded). A request++ * 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`.++ * 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.*@@@ -45,7 +52,8 @@ const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000]const LEASE_MS = 2 * 60_000;const TITLE_MAX = 120;const MESSAGE_MAX = 300;--const LATER = new Set(['actor-task', 'codes-run', 'run-job']);++const LATER = new Set(['codes-run', 'run-job']);++const TASK_MAX = 20000;const TRIGGERED = 'watch.watch.triggered';const MIN_HOOK_SECRET = 32;// A URL the guard refuses (server/net/guard.js) will be refused on every attempt: the delivery fails at once.@@@ -93,7 +101,21 @@ function webhookFor({ config, watch, run, value, previousValue, observedAt }) {};}--function createActions({ db, config, notifications = null, post = null, now = () => Date.now(), log = console }) {++/** 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 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}`;++}++++function createActions({ db, config, notifications = null, post = null, actor = null, now = () => Date.now(), log = console }) {// Read at each use, not once: what a process was started with is what it enforces, for rows queued before it too.const grants = () => config.secretGrants;const enabled = Boolean(notifications) && config.actions.enabled;@@@ -133,6 +155,10 @@ function createActions({ db, config, notifications = null, post = null, now = ()const why = grants() ? grants().refusal(action.secret_env, owner) : `${action.secret_env} is not granted to any owner`;if (why) { state = 'unavailable'; error = `watch.action_secret_not_granted: ${why}`; }else request = { url: action.url, secret_env: action.secret_env, body: webhookFor({ config, watch, run, value, previousValue, observedAt }) };++ } else if (kind === 'actor-task') {++ 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 (LATER.has(kind)) {state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;} else {@@@ -169,9 +195,25 @@ function createActions({ db, config, notifications = null, post = null, now = ()throw Object.assign(new Error(`the receiver answered ${res.status}`), { code: `http_${res.status}` });}++ async function deliverActorTask(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.create({++ subjectId: row.owner_sub, task: request.task, budgetUsd: request.budget_usd,++ idempotencyKey: `watch.${row.check_run_id}.${row.action_index}`,++ });++ return { state: 'sent', error: `Actor task ${task.id || '(no id)'}` };++ } catch (err) {++ // Actor answered that this request can never succeed: over the tier's budget, an unknown agent, invalid.++ if (err && (err.status === 422 || err.status === 400)) 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 !== '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 };@@@ -243,4 +285,4 @@ function createActions({ db, config, notifications = null, post = null, now = ()return { queue, tick, start, stop, status, recent, enabled };}--module.exports = { createActions, notificationFor, webhookFor, BACKOFF_MS };++module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, BACKOFF_MS };nnew file mode 100644@@@ -0,0 +1,27 @@++'use strict';++/**++ * OpenVibe.Actor, for a watch's actor-task action (plan T18 step 6): give Actor a task 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 }++ *++ * POST <actor>/api/v1/tasks with Watch's own token (actor.task.create, audience openvibe.actor) and 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++ * (status, code).++ */++++function createActorTasks(client, { baseUrl }) {++ 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,++ });++ return { id: out && out.id, state: out && out.state };++ }++ return { create };++}++++module.exports = { createActorTasks };@@@ -88,6 +88,8 @@ 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).++ actorUrl: (env.WATCH_ACTOR_URL || 'http://127.0.0.1:4950').replace(/\/+$/, ''),},// Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins@@@ -15,6 +15,7 @@ const { createRegistry } = require('./registry');const { createObservations } = require('./observations');const { createGuard } = require('./net/guard');const { createPoster } = require('./net/post');++const { createActorTasks } = require('./actor');const { createFetcher } = require('./net/fetcher');const { createSpacer } = require('./spacer');const { createCarriers } = require('./carriers');@@@ -37,7 +38,7 @@ const { createApp } = require('./app');const accountDataLib = require('./account-data');const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null } = {}) {++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null, actor: givenActor = null } = {}) {config = config || load();// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.const db = givenDb || await openDb(config, { log });@@@ -60,15 +61,18 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchconst carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });// Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own// client-credentials token; without the client secret they wait (actions.status() says delivery is off).-- const notifications = config.oauth.clientSecret-- ? createNotificationsClient(createClient({++ // Actor tasks (step 6) use the same client and token provider (a token per audience).++ const client = config.oauth.clientSecret++ ? createClient({tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),-- baseUrls: { network: config.networkInternalUrl }, fetch: fetchImpl,-- }))++ baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl }, fetch: fetchImpl,++ }): null;++ const notifications = client ? createNotificationsClient(client) : null;++ const actorTasks = client ? createActorTasks(client, { baseUrl: config.actions.actorUrl }) : null;// Webhook actions leave through the same guard as every fetch (server/net/post.js).const post = givenPost || createPoster({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.actions.webhookTimeoutMs });-- const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, now, log });++ const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, now, log });const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });const scheduler = createScheduler({ db, check, config, now, log });const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });@@@ -5,7 +5,9 @@* refusal with backoff, skips an owner Network does not know and an owner that is not a person, and records the kinds* whose dispatchers are later steps as unavailable. A webhook action POSTs the firing signed like an Events delivery,* with a delivery id that is the same on every attempt; a busy receiver is retried, 410 and a refused URL fail at once;-- * its key must be granted to the owner (refused at save, unavailable at delivery once withdrawn) and long enough. The++ * its key must be granted to the owner (refused at save, unavailable at delivery once withdrawn) and long enough. An++ * actor-task action gives Actor a task for the owner (X-OV-Subject) with the firing as context and a stable++ * idempotency key; an invalid request fails at once, a busy Actor is retried, an agent-owned watch is skipped. The* watch page says what happened.*/const assert = require('assert');@@@ -41,6 +43,17 @@ async function post(url, { headers, body }) {return { status: { ok: 204, busy: 503, gone: 410 }[hookMode] };}const hook = (env = 'WATCH_HOOK_TEST') => ({ kind: 'webhook', url: 'https://hooks.example.org/w', secret_env: env });++// OpenVibe.Actor, as server/actor.js would reach it: every create recorded; actorMode decides the answer.++const tasks = [];++let actorMode = 'ok';++const actor = {++ async create(req) {++ tasks.push(req);++ if (actorMode === 'refuse') throw Object.assign(new Error('over the tier'), { status: 422, code: 'actor.budget.over_tier', detail: 'The free tier allows up to $0.25 per task' });++ 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' };++ },++};const def = (overrides = {}) => ({name: 'Widget price',source: { kind: 'http', url: `${web.origin}/price`, format: null },@@@ -70,7 +83,7 @@ t('boot', async () => {'/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },});svc = await boot({-- notifications, post,++ notifications, post, actor,env: {WATCH_HOOK_TEST: HOOK_KEY, WATCH_HOOK_SHORT: 'too-short', WATCH_HOOK_THEIRS: HOOK_KEY,WATCH_SECRET_GRANTS: JSON.stringify({@@@ -82,7 +95,7 @@ t('boot', async () => {t('a check that does not fire queues nothing; one that fires queues each action once, with its run', async () => {mode = 'ok'; pushes.length = 0;-- const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'actor-task', brief: 'look into it' }] });++ const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'codes-run', repo: 'OpenVibers/OpenVibe.Watch', brief: 'look into it' }] });await setPrice('19.99');await check(w.id);assert.deepStrictEqual(await acts(w.id), [], 'no trigger, no action');@@@ -90,7 +103,7 @@ t('a check that does not fire queues nothing; one that fires queues each actionconst out = await check(w.id);const rows = await acts(w.id);assert.deepStrictEqual(rows.map((r) => [r.kind, r.state, r.check_run_id, Number(r.action_index)]),-- [['notification', 'queued', out.run.id, 0], ['actor-task', 'unavailable', out.run.id, 1]]);++ [['notification', 'queued', out.run.id, 0], ['codes-run', 'unavailable', out.run.id, 1]]);assert.match(rows[1].last_error, /watch\.action_unavailable/);const r = await svc.actions.tick();@@@ -254,6 +267,59 @@ t('a webhook key must be granted to the owner (refused at save, unavailable onceassert.strictEqual(posts.length, 0, 'nothing was posted');});++t('an actor-task action gives Actor a task for the owner, with the firing as context and one idempotency key', async () => {++ tasks.length = 0; actorMode = 'busy';++ const w = await create(person, { name: 'cheap widget', action: [{ kind: 'actor-task', brief: 'Buy two if it is still in stock.', budget_usd: 0.1 }] });++ const out = await fire(w);++ let r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 1, failed: 0 }, 'a busy Actor is retried');++ actorMode = 'ok';++ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);++ r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });++ assert.strictEqual(tasks.length, 2);++ const [first, second] = tasks;++ assert.strictEqual(first.subjectId, person.sub, 'the owner\'s task');++ assert.strictEqual(first.idempotencyKey, `watch.${out.run.id}.0`);++ assert.strictEqual(second.idempotencyKey, first.idempotencyKey, 'the retry is the same task');++ assert.strictEqual(first.budgetUsd, 0.1);++ assert.match(first.task, /^Buy two if it is still in stock\.\n\nContext: your watch “cheap widget” fired at /);++ assert.match(first.task, /Value: 7\.50 \(was 19\.99\)\./);++ assert.ok(first.task.includes(`https://openvibe.watch/watches/${w.id}`));++ const [row] = await acts(w.id);++ assert.deepStrictEqual([row.state, row.last_error], ['sent', 'Actor task tsk_01JAB2C3D4E5F6G7H8J9K0MN02']);++++ actorMode = 'refuse';++ const over = await create(person, { name: 'too dear', action: [{ kind: 'actor-task', brief: 'x', budget_usd: 50 }] });++ await fire(over);++ r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 0, failed: 1 }, 'a request Actor refuses as invalid fails at once');++ assert.match((await acts(over.id))[0].last_error, /watch\.action_refused: actor\.budget\.over_tier/);++ actorMode = 'ok';++++ const before = tasks.length;++ const bot = await create(agent, { name: 'agent task', action: [{ kind: 'actor-task', brief: 'x' }] });++ await fire(bot);++ const [b] = await acts(bot.id);++ assert.deepStrictEqual([b.state, /action_no_person/.test(b.last_error)], ['skipped', true]);++ await svc.actions.tick();++ assert.strictEqual(tasks.length, before, 'no task for an agent-owned watch');++});++++t('server/actor.js asks Actor as Watch, for the owner, once', async () => {++ const { createActorTasks } = require('../server/actor');++ const calls = [];++ const client = { async json(o) { calls.push(o); return { id: 'tsk_X', state: 'queued' }; } };++ const out = await createActorTasks(client, { baseUrl: 'http://127.0.0.1:4950' }).create({ subjectId: person.sub, task: 'do it', budgetUsd: 0.2, idempotencyKey: 'watch.r.0' });++ assert.deepStrictEqual(out, { id: 'tsk_X', state: 'queued' });++ const c = calls[0];++ assert.deepStrictEqual([c.service, c.baseUrl, c.audience, c.method, c.path, c.retries], ['actor', 'http://127.0.0.1:4950', 'openvibe.actor', 'POST', '/api/v1/tasks', 0]);++ assert.deepStrictEqual(c.headers, { 'X-OV-Subject': person.sub });++ 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');++});++t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {const st = await svc.actions.status();assert.strictEqual(st.enabled, true);@@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).*/--async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null } = {}) {++async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null } = {}) {// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real// to fetch and verify against (no test touches the internet). Its URL is overridable via env.const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });@@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l});// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.const testdb = await require('./db').testDb();-- const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(now ? { now } : {}) });++ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(now ? { now } : {}) });const base = `http://127.0.0.1:${h.server.address().port}`;/** A session cookie for a person, as /auth/callback would have set it. */
@@ -98,6 +98,8 @@ VALKEY_PREFIX=
@@ -98,6 +98,8 @@ 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).
+#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
# nobody can use it. A credential needs owners and the origins it may be sent to; a webhook key, owners.
# nobody can use it. A credential needs owners and the origins it may be sent to; a webhook key, owners.
@@ -178,7 +178,8 @@ A fired condition queues the watch's actions in the same transaction as the chec
@@ -178,7 +178,8 @@ A fired condition queues the watch's actions in the same transaction as the chec
| `notification` | the owner gets one OpenVibe notification through OpenVibe.Network (`openvibe-sdk/notifications`, type `WATCH_TRIGGERED`): the watch's name, the value (and the value before), and a link to the watch page. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then `failed`; an owner Network does not know (a deleted account) is `skipped`; a person who switched the category off is Network's decision and counts as `sent`. A watch owned by an agent or a service has no one to notify (`skipped`). Needs the Network OAuth client secret and the grants `identity.subject.resolve` + `network.notifications.push` |
| `notification` | the owner gets one OpenVibe notification through OpenVibe.Network (`openvibe-sdk/notifications`, type `WATCH_TRIGGERED`): the watch's name, the value (and the value before), and a link to the watch page. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then `failed`; an owner Network does not know (a deleted account) is `skipped`; a person who switched the category off is Network's decision and counts as `sent`. A watch owned by an agent or a service has no one to notify (`skipped`). Needs the Network OAuth client secret and the grants `identity.subject.resolve` + `network.notifications.push` |
| `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`, `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |
+| `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` |
+| `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.
Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.
Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.
@@ -28,7 +28,14 @@
@@ -28,7 +28,14 @@
* The key must be granted to the watch's owner (WATCH_SECRET_GRANTS, server/secret-grants.js) and
* The key must be granted to the watch's owner (WATCH_SECRET_GRANTS, server/secret-grants.js) and
* set (32 characters or more), checked when it is queued and again at delivery: otherwise
* set (32 characters or more), checked when it is queued and again at delivery: otherwise
* `unavailable`.
* `unavailable`.
- * actor-task, codes-run, run-job
+ * actor-task a task for the owner on OpenVibe.Actor (server/actor.js): the action's brief, with the firing as
+ * context, under the owner's own allowance (Watch names them in X-OV-Subject; Actor records Watch as
+ * `via`), at most the action's budget_usd. The idempotency key `watch.<check run>.<action index>`
+ * makes a retried delivery the same task. Created is `sent` (the task id is recorded). A request
+ * 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`.
+ * 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.
*
*
@@ -45,7 +52,8 @@ const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000]
@@ -45,7 +52,8 @@ const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000]
const LEASE_MS = 2 * 60_000;
const LEASE_MS = 2 * 60_000;
const TITLE_MAX = 120;
const TITLE_MAX = 120;
const MESSAGE_MAX = 300;
const MESSAGE_MAX = 300;
-const LATER = new Set(['actor-task', 'codes-run', 'run-job']);
+const LATER = new Set(['codes-run', 'run-job']);
+const TASK_MAX = 20000;
const TRIGGERED = 'watch.watch.triggered';
const TRIGGERED = 'watch.watch.triggered';
const MIN_HOOK_SECRET = 32;
const MIN_HOOK_SECRET = 32;
// A URL the guard refuses (server/net/guard.js) will be refused on every attempt: the delivery fails at once.
// A URL the guard refuses (server/net/guard.js) will be refused on every attempt: the delivery fails at once.
@@ -93,7 +101,21 @@ function webhookFor({ config, watch, run, value, previousValue, observedAt }) {
@@ -93,7 +101,21 @@ function webhookFor({ config, watch, run, value, previousValue, observedAt }) {
};
};
}
}
-function createActions({ db, config, notifications = null, post = null, now = () => Date.now(), log = console }) {
+/** 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 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}`;
+}
+
+function createActions({ db, config, notifications = null, post = null, actor = null, now = () => Date.now(), log = console }) {
// Read at each use, not once: what a process was started with is what it enforces, for rows queued before it too.
// Read at each use, not once: what a process was started with is what it enforces, for rows queued before it too.
const grants = () => config.secretGrants;
const grants = () => config.secretGrants;
const enabled = Boolean(notifications) && config.actions.enabled;
const enabled = Boolean(notifications) && config.actions.enabled;
@@ -133,6 +155,10 @@ function createActions({ db, config, notifications = null, post = null, now = ()
@@ -133,6 +155,10 @@ function createActions({ db, config, notifications = null, post = null, now = ()
const why = grants() ? grants().refusal(action.secret_env, owner) : `${action.secret_env} is not granted to any owner`;
const why = grants() ? grants().refusal(action.secret_env, owner) : `${action.secret_env} is not granted to any owner`;
if (why) { state = 'unavailable'; error = `watch.action_secret_not_granted: ${why}`; }
if (why) { state = 'unavailable'; error = `watch.action_secret_not_granted: ${why}`; }
else request = { url: action.url, secret_env: action.secret_env, body: webhookFor({ config, watch, run, value, previousValue, observedAt }) };
else request = { url: action.url, secret_env: action.secret_env, body: webhookFor({ config, watch, run, value, previousValue, observedAt }) };
+ } else if (kind === 'actor-task') {
+ 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 (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 {
@@ -169,9 +195,25 @@ function createActions({ db, config, notifications = null, post = null, now = ()
@@ -169,9 +195,25 @@ function createActions({ db, config, notifications = null, post = null, now = ()
throw Object.assign(new Error(`the receiver answered ${res.status}`), { code: `http_${res.status}` });
throw Object.assign(new Error(`the receiver answered ${res.status}`), { code: `http_${res.status}` });
}
}
+ async function deliverActorTask(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.create({
+ subjectId: row.owner_sub, task: request.task, budgetUsd: request.budget_usd,
+ idempotencyKey: `watch.${row.check_run_id}.${row.action_index}`,
+ });
+ return { state: 'sent', error: `Actor task ${task.id || '(no id)'}` };
+ } catch (err) {
+ // Actor answered that this request can never succeed: over the tier's budget, an unknown agent, invalid.
+ if (err && (err.status === 422 || err.status === 400)) 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 !== '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 };
@@ -243,4 +285,4 @@ function createActions({ db, config, notifications = null, post = null, now = ()
@@ -243,4 +285,4 @@ function createActions({ db, config, notifications = null, post = null, now = ()
return { queue, tick, start, stop, status, recent, enabled };
return { queue, tick, start, stop, status, recent, enabled };
}
}
-module.exports = { createActions, notificationFor, webhookFor, BACKOFF_MS };
+module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, BACKOFF_MS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,27 @@
@@ -0,0 +1,27 @@
+'use strict';
+/**
+ * OpenVibe.Actor, for a watch's actor-task action (plan T18 step 6): give Actor a task 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 }
+ *
+ * POST <actor>/api/v1/tasks with Watch's own token (actor.task.create, audience openvibe.actor) and 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
+ * (status, code).
+ */
+
+function createActorTasks(client, { baseUrl }) {
+ 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,
+ });
+ return { id: out && out.id, state: out && out.state };
+ }
+ return { create };
+}
+
+module.exports = { createActorTasks };
@@ -88,6 +88,8 @@ function load(env = process.env) {
@@ -88,6 +88,8 @@ 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).
+ actorUrl: (env.WATCH_ACTOR_URL || 'http://127.0.0.1:4950').replace(/\/+$/, ''),
},
},
// Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins
// Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins
@@ -15,6 +15,7 @@ const { createRegistry } = require('./registry');
@@ -15,6 +15,7 @@ const { createRegistry } = require('./registry');
const { createObservations } = require('./observations');
const { createObservations } = require('./observations');
const { createGuard } = require('./net/guard');
const { createGuard } = require('./net/guard');
const { createPoster } = require('./net/post');
const { createPoster } = require('./net/post');
+const { createActorTasks } = require('./actor');
const { createFetcher } = require('./net/fetcher');
const { createFetcher } = require('./net/fetcher');
const { createSpacer } = require('./spacer');
const { createSpacer } = require('./spacer');
const { createCarriers } = require('./carriers');
const { createCarriers } = require('./carriers');
@@ -37,7 +38,7 @@ const { createApp } = require('./app');
@@ -37,7 +38,7 @@ const { createApp } = require('./app');
const accountDataLib = require('./account-data');
const accountDataLib = require('./account-data');
const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
-async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null } = {}) {
+async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null, actor: givenActor = null } = {}) {
config = config || load();
config = config || load();
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
const db = givenDb || await openDb(config, { log });
const db = givenDb || await openDb(config, { log });
@@ -60,15 +61,18 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -60,15 +61,18 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });
const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });
// Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own
// Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own
// client-credentials token; without the client secret they wait (actions.status() says delivery is off).
// client-credentials token; without the client secret they wait (actions.status() says delivery is off).
- const notifications = config.oauth.clientSecret
- ? createNotificationsClient(createClient({
+ // Actor tasks (step 6) use the same client and token provider (a token per audience).
+ const client = config.oauth.clientSecret
+ ? createClient({
tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),
tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),
- baseUrls: { network: config.networkInternalUrl }, fetch: fetchImpl,
- }))
+ baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl }, fetch: fetchImpl,
+ })
: null;
: null;
+ const notifications = client ? createNotificationsClient(client) : null;
+ const actorTasks = client ? createActorTasks(client, { baseUrl: config.actions.actorUrl }) : null;
// Webhook actions leave through the same guard as every fetch (server/net/post.js).
// Webhook actions leave through the same guard as every fetch (server/net/post.js).
const post = givenPost || createPoster({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.actions.webhookTimeoutMs });
const post = givenPost || createPoster({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.actions.webhookTimeoutMs });
- const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, now, log });
+ const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, now, log });
const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });
const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
@@ -5,7 +5,9 @@
@@ -5,7 +5,9 @@
* refusal with backoff, skips an owner Network does not know and an owner that is not a person, and records the kinds
* refusal with backoff, skips an owner Network does not know and an owner that is not a person, and records the kinds
* whose dispatchers are later steps as unavailable. A webhook action POSTs the firing signed like an Events delivery,
* whose dispatchers are later steps as unavailable. A webhook action POSTs the firing signed like an Events delivery,
* with a delivery id that is the same on every attempt; a busy receiver is retried, 410 and a refused URL fail at once;
* with a delivery id that is the same on every attempt; a busy receiver is retried, 410 and a refused URL fail at once;
- * its key must be granted to the owner (refused at save, unavailable at delivery once withdrawn) and long enough. The
+ * its key must be granted to the owner (refused at save, unavailable at delivery once withdrawn) and long enough. An
+ * actor-task action gives Actor a task for the owner (X-OV-Subject) with the firing as context and a stable
+ * idempotency key; an invalid request fails at once, a busy Actor is retried, an agent-owned watch is skipped. The
* watch page says what happened.
* watch page says what happened.
*/
*/
const assert = require('assert');
const assert = require('assert');
@@ -41,6 +43,17 @@ async function post(url, { headers, body }) {
@@ -41,6 +43,17 @@ async function post(url, { headers, body }) {
return { status: { ok: 204, busy: 503, gone: 410 }[hookMode] };
return { status: { ok: 204, busy: 503, gone: 410 }[hookMode] };
}
}
const hook = (env = 'WATCH_HOOK_TEST') => ({ kind: 'webhook', url: 'https://hooks.example.org/w', secret_env: env });
const hook = (env = 'WATCH_HOOK_TEST') => ({ kind: 'webhook', url: 'https://hooks.example.org/w', secret_env: env });
+// OpenVibe.Actor, as server/actor.js would reach it: every create recorded; actorMode decides the answer.
+const tasks = [];
+let actorMode = 'ok';
+const actor = {
+ async create(req) {
+ tasks.push(req);
+ if (actorMode === 'refuse') throw Object.assign(new Error('over the tier'), { status: 422, code: 'actor.budget.over_tier', detail: 'The free tier allows up to $0.25 per task' });
+ 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' };
+ },
+};
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 },
@@ -70,7 +83,7 @@ t('boot', async () => {
@@ -70,7 +83,7 @@ t('boot', async () => {
'/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },
'/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },
});
});
svc = await boot({
svc = await boot({
- notifications, post,
+ notifications, post, actor,
env: {
env: {
WATCH_HOOK_TEST: HOOK_KEY, WATCH_HOOK_SHORT: 'too-short', WATCH_HOOK_THEIRS: HOOK_KEY,
WATCH_HOOK_TEST: HOOK_KEY, WATCH_HOOK_SHORT: 'too-short', WATCH_HOOK_THEIRS: HOOK_KEY,
WATCH_SECRET_GRANTS: JSON.stringify({
WATCH_SECRET_GRANTS: JSON.stringify({
@@ -82,7 +95,7 @@ t('boot', async () => {
@@ -82,7 +95,7 @@ t('boot', async () => {
t('a check that does not fire queues nothing; one that fires queues each action once, with its run', async () => {
t('a check that does not fire queues nothing; one that fires queues each action once, with its run', async () => {
mode = 'ok'; pushes.length = 0;
mode = 'ok'; pushes.length = 0;
- const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'actor-task', brief: 'look into it' }] });
+ const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'codes-run', repo: 'OpenVibers/OpenVibe.Watch', brief: 'look into it' }] });
await setPrice('19.99');
await setPrice('19.99');
await check(w.id);
await check(w.id);
assert.deepStrictEqual(await acts(w.id), [], 'no trigger, no action');
assert.deepStrictEqual(await acts(w.id), [], 'no trigger, no action');
@@ -90,7 +103,7 @@ t('a check that does not fire queues nothing; one that fires queues each action
@@ -90,7 +103,7 @@ t('a check that does not fire queues nothing; one that fires queues each action
const out = await check(w.id);
const out = await check(w.id);
const rows = await acts(w.id);
const rows = await acts(w.id);
assert.deepStrictEqual(rows.map((r) => [r.kind, r.state, r.check_run_id, Number(r.action_index)]),
assert.deepStrictEqual(rows.map((r) => [r.kind, r.state, r.check_run_id, Number(r.action_index)]),
- [['notification', 'queued', out.run.id, 0], ['actor-task', 'unavailable', out.run.id, 1]]);
+ [['notification', 'queued', out.run.id, 0], ['codes-run', 'unavailable', out.run.id, 1]]);
assert.match(rows[1].last_error, /watch\.action_unavailable/);
assert.match(rows[1].last_error, /watch\.action_unavailable/);
const r = await svc.actions.tick();
const r = await svc.actions.tick();
@@ -254,6 +267,59 @@ t('a webhook key must be granted to the owner (refused at save, unavailable once
@@ -254,6 +267,59 @@ t('a webhook key must be granted to the owner (refused at save, unavailable once
assert.strictEqual(posts.length, 0, 'nothing was posted');
assert.strictEqual(posts.length, 0, 'nothing was posted');
});
});
+t('an actor-task action gives Actor a task for the owner, with the firing as context and one idempotency key', async () => {
+ tasks.length = 0; actorMode = 'busy';
+ const w = await create(person, { name: 'cheap widget', action: [{ kind: 'actor-task', brief: 'Buy two if it is still in stock.', budget_usd: 0.1 }] });
+ const out = await fire(w);
+ let r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 1, failed: 0 }, 'a busy Actor is retried');
+ actorMode = 'ok';
+ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);
+ r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });
+ assert.strictEqual(tasks.length, 2);
+ const [first, second] = tasks;
+ assert.strictEqual(first.subjectId, person.sub, 'the owner\'s task');
+ assert.strictEqual(first.idempotencyKey, `watch.${out.run.id}.0`);
+ assert.strictEqual(second.idempotencyKey, first.idempotencyKey, 'the retry is the same task');
+ assert.strictEqual(first.budgetUsd, 0.1);
+ assert.match(first.task, /^Buy two if it is still in stock\.\n\nContext: your watch “cheap widget” fired at /);
+ assert.match(first.task, /Value: 7\.50 \(was 19\.99\)\./);
+ assert.ok(first.task.includes(`https://openvibe.watch/watches/${w.id}`));
+ const [row] = await acts(w.id);
+ assert.deepStrictEqual([row.state, row.last_error], ['sent', 'Actor task tsk_01JAB2C3D4E5F6G7H8J9K0MN02']);
+
+ actorMode = 'refuse';
+ const over = await create(person, { name: 'too dear', action: [{ kind: 'actor-task', brief: 'x', budget_usd: 50 }] });
+ await fire(over);
+ r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 0, failed: 1 }, 'a request Actor refuses as invalid fails at once');
+ assert.match((await acts(over.id))[0].last_error, /watch\.action_refused: actor\.budget\.over_tier/);
+ actorMode = 'ok';
+
+ const before = tasks.length;
+ const bot = await create(agent, { name: 'agent task', action: [{ kind: 'actor-task', brief: 'x' }] });
+ await fire(bot);
+ const [b] = await acts(bot.id);
+ assert.deepStrictEqual([b.state, /action_no_person/.test(b.last_error)], ['skipped', true]);
+ await svc.actions.tick();
+ assert.strictEqual(tasks.length, before, 'no task for an agent-owned watch');
+});
+
+t('server/actor.js asks Actor as Watch, for the owner, once', async () => {
+ const { createActorTasks } = require('../server/actor');
+ const calls = [];
+ const client = { async json(o) { calls.push(o); return { id: 'tsk_X', state: 'queued' }; } };
+ const out = await createActorTasks(client, { baseUrl: 'http://127.0.0.1:4950' }).create({ subjectId: person.sub, task: 'do it', budgetUsd: 0.2, idempotencyKey: 'watch.r.0' });
+ assert.deepStrictEqual(out, { id: 'tsk_X', state: 'queued' });
+ const c = calls[0];
+ assert.deepStrictEqual([c.service, c.baseUrl, c.audience, c.method, c.path, c.retries], ['actor', 'http://127.0.0.1:4950', 'openvibe.actor', 'POST', '/api/v1/tasks', 0]);
+ assert.deepStrictEqual(c.headers, { 'X-OV-Subject': person.sub });
+ 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');
+});
+
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 () => {
const st = await svc.actions.status();
const st = await svc.actions.status();
assert.strictEqual(st.enabled, true);
assert.strictEqual(st.enabled, true);
@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
*/
*/
-async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null } = {}) {
+async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null } = {}) {
// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real
// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real
// to fetch and verify against (no test touches the internet). Its URL is overridable via env.
// to fetch and verify against (no test touches the internet). Its URL is overridable via env.
const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });
const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });
@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l
@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l
});
});
// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
const testdb = await require('./db').testDb();
const testdb = await require('./db').testDb();
- const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(now ? { now } : {}) });
+ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(now ? { now } : {}) });
const base = `http://127.0.0.1:${h.server.address().port}`;
const base = `http://127.0.0.1:${h.server.address().port}`;
/** A session cookie for a person, as /auth/callback would have set it. */
/** A session cookie for a person, as /auth/callback would have set it. */