openvibers/openvibe.actor Commits
Actor: task webhooks (signed per task, retried, public https endpoints only) (#29)
Changes
@@@ -51,6 +51,9 @@ ACTOR_LOCAL_LLM_MODEL=#ACTOR_MAX_COST_USD_PER_DAY=2#ACTOR_MAX_RUNNING=4#ACTOR_TASK_TIMEOUT_MS=180000++# Task webhooks: how often due deliveries go out, and how long one delivery may take++#ACTOR_WEBHOOK_INTERVAL_MS=5000++#ACTOR_WEBHOOK_TIMEOUT_MS=10000# optional: defaults to true in production# COOKIE_SECURE=true@@@ -41,7 +41,7 @@ Rate cards are typed in by a person from each provider's price page, with the pa| Route | Capability | ||---|---|---|--| `POST /api/v1/tasks` | `actor.task.create` | `actor.task-create-request@1` → `platform.task@1` (201, queued) |++| `POST /api/v1/tasks` | `actor.task.create` | `actor.task-create-request@1` → `platform.task@1` (201, queued; with `webhooks`, the answer carries `webhook_secret` once) || `GET /api/v1/tasks/:id` | `actor.task.read` | the task || `GET /api/v1/tasks/:id/events` | `actor.task.read` | server-sent events, `actor.task-event@1`, resumable with `Last-Event-ID` || `POST /api/v1/tasks/:id/cancel` | `actor.task.create` | stop it |@@@ -59,6 +59,12 @@ The resource index is for OpenVibe.Services on loopback and is blocked at the pu**What a caller can see:** a task is visible only to its requester, or to tokens of its project. Anyone else gets 404, the same answer as a task that doesn't exist.++**Webhooks:** a task may register up to 8 https endpoints, each with the state changes to deliver (`webhooks: [{ url, events }]`, the names of `state`).++- **Signing:** the answer that creates the task carries `webhook_secret`, minted for that task. It is in no read or list, never exported, and erased once the task has ended and its deliveries are done. Each delivery is `actor.task-webhook@1` (`{ type: "actor.task.state", delivery_id, state, task }`, the task as it was then), signed like an Events delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`), so a receiver checks it with openvibe-sdk/events `verifyDeliveryV2(rawBody, headers, secret)`. The delivery id stays the same across retries; dedupe on it.++- **Outcomes:** 2xx is delivered; 410, or a URL refused at send time, ends the delivery; anything else is retried after 10 s, 1 min, 5 min, 30 min, 2 h, 6 h and 12 h, then fails.++- **Where Actor may connect:** a webhook URL is the one address a caller chooses (`server/net/webhook-post.js`): https on port 443 or 8443, no credentials, public addresses only. It is checked at creation (`actor.webhook.refused`) and again at send time through openvibe-shared/egress, so a name that later resolves to a private address is refused. Redirects are never followed.++- **Not yet:** a project secret named by reference (`secret_ref`) is refused until project secrets exist.++**Errors:** RFC 9457 problems. The codes are on [openvibe.actor/docs](https://openvibe.actor/docs#errors).## Money@@@ -193,7 +199,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).<!-- versions:start -->--- openvibe-contracts: v0.129.0++- openvibe-contracts: v0.133.0- openvibe-sdk: v0.38.0- openvibe-shared: v3.0.1<!-- versions:end -->@@@ -7,11 +7,11 @@"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9","port": 4950,"domain": "openvibe.actor",-- "contracts": "openvibe-contracts v0.129.0",++ "contracts": "openvibe-contracts v0.133.0","sdk": "openvibe-sdk v0.38.0","shared": "openvibe-shared v3.0.1","packages": [-- "openvibe-contracts v0.129.0",++ "openvibe-contracts v0.133.0","openvibe-sdk v0.38.0","openvibe-shared v3.0.1"],@@@ -46,7 +46,8 @@"other platforms' agents","private mode on OpenVibe hardware","cross-family answer checks",-- "live task progress"++ "live task progress",++ "task webhooks (signed per task, retried with backoff, public https endpoints only)"],"updated": "2026-10-10"}nnew file mode 100644@@@ -0,0 +1,24 @@++-- phase: expand++-- OpenVibe.Actor: task webhooks (plan T17, actor.task-webhook@1). `webhooks` is what the task registered (url + the++-- state changes to deliver); `webhook_secret` signs its deliveries, minted per task, shown once to its creator and++-- erased once the task has ended and its deliveries are done. Each delivery is a row, retried with backoff; a delivery++-- never delays or changes the task. The secret is never exported (server/identity/account-data.js lists the columns).++ALTER TABLE tasks ADD COLUMN IF NOT EXISTS webhooks jsonb;++ALTER TABLE tasks ADD COLUMN IF NOT EXISTS webhook_secret text COLLATE "C";++++CREATE TABLE IF NOT EXISTS webhook_deliveries (++ id text COLLATE "C" PRIMARY KEY, -- whd_<ULID>, stable across retries++ task_id text COLLATE "C" NOT NULL REFERENCES tasks (id) ON DELETE CASCADE,++ url text NOT NULL,++ state text COLLATE "C" NOT NULL, -- the task state it announces++ body jsonb NOT NULL, -- actor.task-webhook@1 as it was then++ status text COLLATE "C" NOT NULL DEFAULT 'pending', -- pending | sending | delivered | failed++ attempts integer NOT NULL DEFAULT 0,++ next_at bigint NOT NULL, -- epoch ms of the next attempt++ last_status integer,++ last_error text,++ created_at text COLLATE "C" NOT NULL,++ finished_at text COLLATE "C"++);++CREATE INDEX IF NOT EXISTS webhook_deliveries_due ON webhook_deliveries (next_at) WHERE status = 'pending';++CREATE INDEX IF NOT EXISTS webhook_deliveries_task ON webhook_deliveries (task_id);@@@ -15,7 +15,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "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.133.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.1","pg": "^8.23.0"@@@ -871,9 +871,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.133.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.133.0",++ "integrity": "sha512-kER/ktuvD74FU2xKKByWZ/8fvFbVTwhWyGhdCuMJL9XU5F9pmWV27+PL5MZ4VCMN9EolNyHh5FgprhhOW+M4kQ==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -27,7 +27,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "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.133.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.1","pg": "^8.23.0"@@@ -24,6 +24,8 @@ const { createSso } = require('./auth/sso');const catalog = require('./agents/catalog');const { createAdapters } = require('./agents/adapters');const { createEngine } = require('./tasks/engine');++const { createWebhooks } = require('./tasks/webhooks');++const { createWebhookPoster } = require('./net/webhook-post');const { createTaskStream } = require('./tasks/stream');const { createPrincipal } = require('./http/principal');const { createApi } = require('./http/api');@@@ -42,7 +44,8 @@ const VERSION = require('../package.json').version;/*** opts: config, store, now (clock), fetchImpl (Network), providerFetch (model providers and OpenVibe services), log,* limitsNow, callerLimits (false: count nobody, tests only), valkey, adapters (tests: stand-in agents),-- * accountSend (a stand-in for Network's internal routes; tests only)++ * accountSend (a stand-in for Network's internal routes; tests only), webhookPost (a stand-in for the guarded webhook++ * poster; tests only)*/async function createApp(opts = {}) {const config = opts.config || configLib.load();@@@ -54,9 +57,14 @@ async function createApp(opts = {}) {const sso = createSso({ config, keys, fetchImpl, now: s.now, log });const stream = createTaskStream({ log });const adapters = opts.adapters || createAdapters({ config, rates: catalog.RATES, fetchImpl: opts.providerFetch || globalThis.fetch, log });-- const engine = createEngine({ config, s, adapters, stream, now: s.now, log });++ // Task webhooks (plan T17): the only connections to an address a caller chose, through server/net/webhook-post.js.++ const webhooks = createWebhooks({++ s, config, now: s.now, log,++ post: opts.webhookPost || createWebhookPoster({ userAgent: `OpenVibe.Actor/${VERSION} (+https://openvibe.actor/docs#webhooks)`, timeoutMs: config.webhooks.timeoutMs }),++ });++ const engine = createEngine({ config, s, adapters, stream, webhooks, now: s.now, log });const principal = createPrincipal({ config, keys });-- const ctx = { config, s, keys, sso, stream, adapters, engine, principal, log };++ const ctx = { config, s, keys, sso, stream, adapters, engine, webhooks, principal, log };// Account export and deletion (ADR-033, ./identity/account-data.js): the two tables that hold a person's rows.// The sender posts to Network's internal export/deletion routes with this service's own client-credentials token;@@@ -79,6 +79,11 @@ function load(env = process.env) {maxRunning: Math.max(1, int(env.ACTOR_MAX_RUNNING, 4)),// A task that runs longer than this is stopped and fails (actor.task.timeout).taskTimeoutMs: Math.max(10_000, int(env.ACTOR_TASK_TIMEOUT_MS, 180_000)),++ // Task webhooks (server/tasks/webhooks.js): how often due deliveries are sent, and how long one may take.++ webhooks: {++ intervalMs: Math.max(1000, int(env.ACTOR_WEBHOOK_INTERVAL_MS, 5000)),++ timeoutMs: Math.max(1000, int(env.ACTOR_WEBHOOK_TIMEOUT_MS, 10_000)),++ },// OpenVibe.Events → this service (server/events-consumer.js). The secret signs a delivery (comma-separated// for rotation, 32+ characters each); unset turns POST /internal/events off (503). The url is where the@@@ -5,7 +5,8 @@** GET /agents actor.agent.read (public) the agent systems, their rate cards and availability* POST /route actor.agent.read (public) which agent a task would go to, and why (nothing runs)-- * POST /tasks actor.task.create give Actor a task (actor.task-create-request@1 → platform.task@1)++ * POST /tasks actor.task.create give Actor a task (actor.task-create-request@1 → platform.task@1;++ * with webhooks, the answer carries webhook_secret once)* GET /tasks actor.task.list your tasks, newest first* GET /tasks/:id actor.task.read one task* GET /tasks/:id/events actor.task.read its live stream (actor.task-event@1, Last-Event-ID)@@@ -26,7 +27,7 @@ const { createTask } = require('../tasks/create');const TASK_ID = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;function createApi(ctx) {-- const { config, s, engine, stream, principal, limits } = ctx;++ const { config, s, engine, stream, principal, limits, webhooks } = ctx;const r = asyncRouter();const problem = (req, res, status, code, detail, extra) => contracts.http.sendProblem(res, status, code, { detail, ctx: req.ov, extra });@@@ -60,13 +61,14 @@ function createApi(ctx) {const wire = (row) => store.toWire(row, { baseUrl: config.baseUrl });r.post('/tasks', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), async (req, res) => {-- const out = await createTask({ s, config, engine, principal: req.principal, body: req.body || {} });++ const out = await createTask({ s, config, engine, webhooks, principal: req.principal, body: req.body || {} });if (out.code) {if (out.retryAfter) res.set('Retry-After', String(out.retryAfter));return problem(req, res, out.status, out.code, out.detail);}if (out.status === 201) res.set('Location', `/api/v1/tasks/${out.task.id}`);-- return res.status(out.status).json(wire(out.task));++ // The signing secret, to its creator only, in this answer (and an idempotent repeat of it) and never again.++ return res.status(out.status).json({ ...wire(out.task), ...(out.task.webhook_secret ? { webhook_secret: out.task.webhook_secret } : {}) });});r.get('/tasks', principal.requireCapability('actor.task.list'), limits.reads('actor.task.list'), async (req, res) => {@@@ -214,6 +214,7 @@ ${pageRows.next ? html`<p><a href="/tasks?before=${pageRows.next}">Older tasks</}const t = store.toWire(row, { baseUrl: config.baseUrl });const events = await store.eventsAfter(s, row.id, 0);++ const deliveries = row.webhooks ? await ctx.webhooks.forTask(row.id) : [];const open = store.OPEN.includes(row.state);const cls = (events.find((e) => e.step === 'plan' && !e.agent && /"(\w+)" task/.test(e.text || '')) || { text: '' }).text.match(/"(\w+)"/);const kind = cls ? CLASS_TEXT[cls[1]] || cls[1] : null;@@@ -235,11 +236,22 @@ ${outcome(t, row)}${open ? html`<p class="muted small live-note" id="live-note"><span class="dot-pulse" aria-hidden="true"></span> Updating as the agent works.</p>` : ''}</section>${t.explanation ? explanationBlock(t.explanation) : ''}++${deliveries.length ? deliveriesBlock(deliveries) : ''}<p class="muted small api-hint">In code: <code>GET /api/v1/tasks/${row.id}</code> · <a href="/docs">API</a></p></article>`,});});++ /** The task's webhook deliveries: where, which state, and how it went (the receiver's answer code only). */++ const DELIVERY_TEXT = { pending: 'Waiting to retry', sending: 'Sending', delivered: 'Delivered', failed: 'Not delivered' };++ function deliveriesBlock(rows) {++ return html`<details class="deliveries"><summary>Webhooks (${rows.length})</summary>++${table(['State', 'Endpoint', 'Delivery', 'Attempts'], rows.map((d) => [++ html`<code>${d.state}</code>`, html`<code class="small">${d.url}</code>`,++ html`${DELIVERY_TEXT[d.status] || d.status}${d.last_status ? html` <span class="muted small">(${d.last_status})</span>` : ''}`, String(d.attempts),++ ]))}</details>`;++ }++/** The outcome card: the first thing under the task. */function outcome(t, row) {if (t.state === 'succeeded') return resultBlock(t.result, t);@@@ -395,7 +407,7 @@ data: {"task_id":"tsk_…","seq":3,"kind":"output","agent":"openvibe-runtime","s<li><strong>An app or agent</strong>: a Network token for audience <code>openvibe.actor</code> with the <code>actor.task.*</code> capability the route needs, granted on <a href="https://openvibe.services">OpenVibe.Services</a>. Tasks belong to the app's project.</li><li><strong>Anyone</strong>: <code>GET /api/v1/agents</code> and <code>POST /api/v1/route</code> need no token.</li></ul>${table(['Route', 'Capability', 'What it does'], [-- [html`<code>POST /api/v1/tasks</code>`, html`<code>actor.task.create</code>`, html`Create a task: <code>{ task, mode?, budget?: { per_task_usd, per_day_usd }, agent?, idempotency_key? }</code>. Answers 201 with the task (queued).`],++ [html`<code>POST /api/v1/tasks</code>`, html`<code>actor.task.create</code>`, html`Create a task: <code>{ task, mode?, budget?: { per_task_usd, per_day_usd }, agent?, idempotency_key?, webhooks?: [{ url, events }] }</code>. Answers 201 with the task (queued); with webhooks, also its <code>webhook_secret</code>, this once (<a href="#webhooks">webhooks</a>).`],[html`<code>GET /api/v1/tasks/:id</code>`, html`<code>actor.task.read</code>`, 'The task: state, result, cost, explanation, error.'],[html`<code>GET /api/v1/tasks/:id/events</code>`, html`<code>actor.task.read</code>`, html`Server-sent events (<code>actor.task-event@1</code>): <code>state</code>, <code>output</code> (each step), <code>end</code>. Resumes with <code>Last-Event-ID</code>.`],[html`<code>POST /api/v1/tasks/:id/cancel</code>`, html`<code>actor.task.create</code>`, 'Stop it. What was spent stays spent.'],@@@ -404,12 +416,18 @@ ${table(['Route', 'Capability', 'What it does'], [[html`<code>POST /api/v1/route</code>`, 'public', html`<code>{ task, mode? }</code> → which agent would take it and why, for every candidate. Nothing runs and nothing is charged.`],])}${raw(showcase.code({ title: 'Create a task, then watch it', samples: [{ label: 'Create', lang: 'bash', code: create }, { label: 'Watch', lang: 'bash', code: watch }] }))}++<h2 id="webhooks">Webhooks</h2>++<p>Give a task up to 8 HTTPS endpoints and the state changes each wants (<code>queued</code>, <code>running</code>, <code>verifying</code>, <code>succeeded</code>, <code>failed</code>, <code>cancelled</code>). Actor POSTs <a href="https://openvibe.services/docs/contracts/actor.task-webhook@1"><code>actor.task-webhook@1</code></a> to each: <code>{ type: "actor.task.state", delivery_id, state, task }</code>, with the task as it was at that moment.</p>++<ul><li><strong>Signed per task.</strong> The answer that creates the task carries <code>webhook_secret</code>, and nothing else ever shows it. Each delivery has <code>X-OpenVibe-Timestamp</code> and <code>X-OpenVibe-Signature-V2</code>, exactly like an OpenVibe.Events delivery, so <code>verifyDeliveryV2(rawBody, headers, secret)</code> from <code>openvibe-sdk/events</code> checks it (and the five-minute window).</li>++<li><strong>Retried.</strong> A 2xx answer is delivered. 410 Gone ends it. Anything else is retried after 10 s, 1 min, 5 min, 30 min, 2 h, 6 h and 12 h. <code>delivery_id</code> (also <code>X-OpenVibe-Delivery-Id</code>) is the same on every attempt: dedupe on it.</li>++<li><strong>Public endpoints only.</strong> HTTPS on port 443 or 8443, no credentials in the URL, and only public addresses, checked when the task is created and again when each delivery is sent. Redirects are not followed.</li></ul><h2 id="limits">Limits and money</h2><p>The free tier: ${tier.tasksPerDay} tasks a day, up to ${usd(tier.perTaskUsd)} a task and ${usd(tier.perDayUsd)} a day (<a href="/limits.json">limits.json</a>). A budget above the tier is refused (<code>actor.budget.over_tier</code>), never lowered. A task whose cheapest capable agent would cost more than its budget fails before running (<code>actor.budget.exceeded</code>). <code>cost.usd</code> is what each model call and web search really cost at the <a href="/agents">published prices</a>.</p><h2 id="errors">Errors</h2><p>Every refusal is RFC 9457 <code>application/problem+json</code> with a stable <code>code</code>. A task that fails says why in <code>error</code>:</p>${table(['Code', 'Meaning'], [[html`<code>actor.task.invalid</code>`, 'The request does not match actor.task-create-request@1.'],++ [html`<code>actor.webhook.refused</code>`, 'A webhook URL Actor will not deliver to: not HTTPS on 443/8443, credentials in it, or not a public address.'],[html`<code>actor.budget.over_tier</code>`, 'The budget is above what the free tier allows.'],[html`<code>actor.allowance.exhausted</code> / <code>actor.budget.day_spent</code>`, 'Today\'s free tasks or budget are used; Retry-After says when it resets.'],[html`<code>actor.capacity.spent</code>`, 'Actor\'s free capacity for everyone is used up for today.'],@@@ -25,7 +25,12 @@ const { createAccountData, TOPICS } = require('openvibe-sdk/account-data');* store `user:usr_…` (server/tasks/store.js takes the requester straight from the principal).*/const TABLES = [-- { table: 'tasks', subject: 'requester', value: (usr) => `user:${usr}`, file: 'tasks.json' },++ // Every column but webhook_secret (a task's signing secret is a credential, not the person's data).++ {++ table: 'tasks', subject: 'requester', value: (usr) => `user:${usr}`, file: 'tasks.json',++ columns: ['id', 'requester', 'project_id', 'task', 'mode', 'agent', 'budget_task', 'budget_day', 'state', 'result', 'error_code', 'error_detail',++ 'cost_usd', 'free_usd', 'explanation', 'cancel_at', 'cancel_by', 'last_seq', 'idem_key', 'idem_hash', 'created_at', 'finished_at', 'via', 'webhooks'],++ },// The person's own per-UTC-day spend accounting. Deleted: nothing in Actor keeps a person's spend record for a// reason that outlives them. The `*` row (everyone's daily ceiling) does not match `user:usr_…` and stays — it is// the operator's total, a separate row.@@@ -19,7 +19,7 @@ const PRUNE_EVERY_MS = 6 * 60 * 60_000;function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {return gracefulStop({name: 'Actor', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,-- close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.engine.stop(), () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],++ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.engine.stop(), () => ctx.webhooks.stop(), () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],});}@@@ -29,6 +29,10 @@ async function start() {// A task a stopped process left open did not finish: it is recorded so, never re-run.const interrupted = await store.failInterrupted(ctx.s);if (interrupted.length) console.warn(`[Actor] ${interrupted.length} task(s) interrupted by the last stop were marked failed`);++ // Their webhooks hear `failed` like any other end; deliveries a stop left half-sent go out again (same delivery id).++ for (const id of interrupted) await ctx.webhooks.enqueue(id, 'failed').catch((err) => console.warn('[Actor] webhook enqueue failed:', id, err && err.message));++ await ctx.webhooks.recover();++ ctx.webhooks.start();const server = app.listen(config.port, config.host, () => {const available = require('./agents/catalog').list(config).filter((a) => a.available).map((a) => a.id);nnew file mode 100644@@@ -0,0 +1,62 @@++'use strict';++/**++ * The one place Actor connects to an address a caller chose: a task's webhook (plan T17). Everything else Actor fetches++ * goes to a configured base (test/security-ssrf.test.js keeps that list).++ *++ * checkWebhookUrl(url) → URL, or throws WebhookRefused: https only, port 443 or 8443, no credentials, not a name++ * that can only mean this machine or a private network, not a non-public IP literal++ * createWebhookPoster({ userAgent, timeoutMs, lookup }) → post(url, { headers, body }) → { status }++ *++ * The poster checks the URL again at send time and connects through openvibe-shared/egress safeLookup, which fails++ * unless every address the name resolves to is public, so a DNS answer cannot change between the check and the++ * connect. A redirect is never followed (a 3xx is the receiver's answer like any other), and at most a few KiB of the++ * answer are read and discarded: nothing a receiver says is stored.++ */++const https = require('https');++const net = require('net');++const egress = require('openvibe-shared/egress');++++const PORTS = new Set(['', '443', '8443']);++const MAX_RESPONSE_BYTES = 4096;++++class WebhookRefused extends Error {++ constructor(detail) { super(detail); this.code = 'actor.webhook.refused'; }++}++++function checkWebhookUrl(input) {++ let u;++ try { u = new URL(String(input)); } catch { throw new WebhookRefused('not a URL'); }++ if (u.protocol !== 'https:') throw new WebhookRefused('a webhook URL is https');++ if (u.username || u.password) throw new WebhookRefused('credentials in the URL are not allowed');++ if (!PORTS.has(u.port)) throw new WebhookRefused('a webhook URL uses port 443 or 8443');++ const host = egress.normalizeHost(u.hostname);++ if (net.isIP(host) ? !egress.isPublicAddress(host) : egress.isInternalName(host)) throw new WebhookRefused(`${host} is not a public address`);++ return u;++}++++function createWebhookPoster({ userAgent = 'OpenVibe.Actor', timeoutMs = 10_000, lookup = egress.safeLookup } = {}) {++ return async function post(url, { headers = {}, body = '' } = {}) {++ const u = checkWebhookUrl(url);++ const payload = Buffer.from(String(body));++ return await new Promise((resolve, reject) => {++ let settled = false;++ const finish = (fn, v) => { if (!settled) { settled = true; fn(v); } };++ const req = https.request(u, {++ method: 'POST', lookup, timeout: timeoutMs,++ headers: { 'Content-Type': 'application/json', 'Content-Length': payload.length, 'User-Agent': userAgent, ...headers },++ }, (res) => {++ let seen = 0;++ res.on('data', (chunk) => { seen += chunk.length; if (seen > MAX_RESPONSE_BYTES) res.destroy(); });++ const done = () => finish(resolve, { status: res.statusCode });++ res.on('end', done);++ res.on('close', done);++ res.on('error', done);++ });++ req.on('timeout', () => req.destroy(Object.assign(new Error(`no answer within ${timeoutMs} ms`), { code: 'ETIMEDOUT' })));++ req.on('error', (err) => finish(reject, err));++ req.end(payload);++ });++ };++}++++module.exports = { checkWebhookUrl, createWebhookPoster, WebhookRefused };@@@ -10,13 +10,15 @@ const crypto = require('crypto');const contracts = require('openvibe-contracts');const store = require('./store');const catalog = require('../agents/catalog');++const { checkWebhookUrl } = require('../net/webhook-post');++const { mintSecret } = require('./webhooks');function untilMidnight(nowMs) {const n = new Date(nowMs);return Math.ceil((Date.UTC(n.getUTCFullYear(), n.getUTCMonth(), n.getUTCDate() + 1) - n.getTime()) / 1000);}--async function createTask({ s, config, engine, principal, body }) {++async function createTask({ s, config, engine, principal, body, webhooks: webhookQueue = null }) {const v = contracts.validate('actor.task-create-request@1', body);if (!v.valid) return { status: 422, code: 'actor.task.invalid', detail: (v.errors || []).map((e) => `${e.path || e.instancePath || ''} ${e.message}`.trim()).join('; ').slice(0, 500) || 'invalid request' };const tier = config.allowance;@@@ -26,8 +28,14 @@ async function createTask({ s, config, engine, principal, body }) {return { status: 422, code: 'actor.budget.over_tier', detail: `The free tier allows up to $${tier.perTaskUsd} per task and $${tier.perDayUsd} a day; paid tiers come with OpenVibe.Billing.` };}if (body.agent && !catalog.byId(body.agent)) return { status: 422, code: 'actor.agent.unknown', detail: `no agent "${body.agent}"; GET /api/v1/agents lists them` };++ // Webhooks (plan T17): each URL must be one Actor may deliver to (https, port 443/8443, a public address); the++ // send-time check repeats it against what the name resolves to then.++ const webhooks = (body.webhooks || []).map((h) => ({ url: h.url, events: h.events }));++ for (const h of webhooks) {++ try { checkWebhookUrl(h.url); } catch (err) { return { status: 422, code: 'actor.webhook.refused', detail: `${h.url.slice(0, 200)}: ${err.message}` }; }++ }-- const hash = crypto.createHash('sha256').update(JSON.stringify({ task: body.task, mode: body.mode || 'balanced', budget: [perTask, perDay], agent: body.agent || null })).digest('hex');++ const hash = crypto.createHash('sha256').update(JSON.stringify({ task: body.task, mode: body.mode || 'balanced', budget: [perTask, perDay], agent: body.agent || null, ...(webhooks.length ? { webhooks } : {}) })).digest('hex');if (body.idempotency_key) {const prior = await store.byIdempotency(s, principal.requester, body.idempotency_key);if (prior) {@@@ -50,11 +58,15 @@ async function createTask({ s, config, engine, principal, body }) {const row = await store.insertTask(s, {id: s.newId('tsk'), requester: principal.requester, via: principal.via || null, project_id: principal.project || null, task: body.task, mode: body.mode || 'balanced', agent: body.agent || null,budget_task: perTask, budget_day: perDay, idem_key: body.idempotency_key || null, idem_hash: body.idempotency_key ? hash : null, created_at: s.iso(),++ webhooks, webhook_secret: webhooks.length ? mintSecret() : null,});return { status: 201, task: row };});-- // The engine starts the task only once its row has committed.-- if (out.status === 201) engine.submit(out.task.id);++ // The engine starts the task only once its row has committed; a webhook that asked for `queued` hears it now.++ if (out.status === 201) {++ if (webhooks.length && webhookQueue) await webhookQueue.enqueue(out.task.id, 'queued');++ engine.submit(out.task.id);++ }return out;}@@@ -22,7 +22,7 @@ const { check } = require('../agents/check');const MAX_ATTEMPTS = 2;--function createEngine({ config, s, adapters, stream, now = () => Date.now(), log = console }) {++function createEngine({ config, s, adapters, stream, webhooks = null, now = () => Date.now(), log = console }) {const running = new Map(); // id → AbortControllerconst waiting = [];let stopped = false;@@@ -39,6 +39,13 @@ function createEngine({ config, s, adapters, stream, now = () => Date.now(), logconst ended = ['succeeded', 'failed', 'cancelled'].includes(state);await emit(id, { kind: ended ? 'end' : 'state', state, ...(agentForEvent ? { agent: agentForEvent } : {}) });if (ended) stream.end(id);++ // Webhooks that asked for this state hear it; a delivery never holds the task up or fails it.++ if (webhooks && row && row.webhooks) {++ try {++ await webhooks.enqueue(id, state);++ if (ended) await webhooks.forgetSecretIfDone(id);++ } catch (err) { log.warn('[Actor] webhook enqueue failed:', id, err && err.message); }++ }return row;}@@@ -22,9 +22,10 @@ function subjectRef(requester) {async function insertTask(s, t) {await s.db.query(-- `INSERT INTO tasks (id, requester, project_id, task, mode, agent, budget_task, budget_day, state, idem_key, idem_hash, created_at, via)-- VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'queued', $9, $10, $11, $12)`,-- [t.id, t.requester, t.project_id || null, t.task, t.mode, t.agent || null, t.budget_task, t.budget_day, t.idem_key || null, t.idem_hash || null, t.created_at, t.via || null]);++ `INSERT INTO tasks (id, requester, project_id, task, mode, agent, budget_task, budget_day, state, idem_key, idem_hash, created_at, via, webhooks, webhook_secret)++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'queued', $9, $10, $11, $12, $13::jsonb, $14)`,++ [t.id, t.requester, t.project_id || null, t.task, t.mode, t.agent || null, t.budget_task, t.budget_day, t.idem_key || null, t.idem_hash || null, t.created_at, t.via || null,++ t.webhooks && t.webhooks.length ? JSON.stringify(t.webhooks) : null, t.webhook_secret || null]);await addSpend(s, t.requester, 0, { tasks: 1 });return getTask(s, t.id);}@@@ -144,7 +145,10 @@ function toWire(row, { baseUrl }) {};if (row.project_id) t.project_id = row.project_id;if (t.explanation && !t.explanation.length) t.explanation = null;++ // What the task registered; the signing secret is never part of the task as read (only the creating answer has it).++ const hooks = parse(row.webhooks);++ if (hooks && hooks.length) t.webhooks = hooks.map((h) => ({ url: h.url, events: h.events }));return t;}--module.exports = { insertTask, getTask, byIdempotency, listTasks, updateTask, appendEvent, eventsAfter, addSpend, spendToday, lockSpendToday, failInterrupted, prune, toWire, subjectRef, dayOf, OPEN, RETENTION_DAYS };++module.exports = { insertTask, getTask, byIdempotency, listTasks, updateTask, appendEvent, eventsAfter, addSpend, spendToday, lockSpendToday, failInterrupted, prune, toWire, subjectRef, dayOf, parseJson: parse, OPEN, RETENTION_DAYS };nnew file mode 100644@@@ -0,0 +1,129 @@++'use strict';++/**++ * Task webhooks (plan T17, actor.task-webhook@1): a task registers up to 8 https endpoints and the state changes each++ * wants. When the task reaches one, a delivery row is queued with the body as it is then; this worker POSTs it,++ * signed with the task's own secret exactly like an Events delivery (X-OpenVibe-Timestamp, X-OpenVibe-Signature-V2:++ * HMAC-SHA256 of "<t>.<raw body>"), so a receiver checks it with openvibe-sdk/events verifyDeliveryV2.++ *++ * 2xx delivered++ * 410, or a refused URL failed at once (the receiver is gone, or the URL is not a public address any more)++ * anything else retried: 10 s, 1 min, 5 min, 30 min, 2 h, 6 h, 12 h, then failed (about a day)++ *++ * A delivery never delays or changes the task. Once a task has ended and none of its deliveries is pending, its secret++ * is erased. The delivery id is stable across attempts (dedupe on it). One Actor process sends: rows are claimed with++ * FOR UPDATE SKIP LOCKED all the same, and a row a crash left 'sending' is pending again at the next start.++ */++const crypto = require('crypto');++const { signDeliveryHeaders } = require('openvibe-sdk/events');++const store = require('./store');++const { WebhookRefused } = require('../net/webhook-post');++++const BACKOFF_MS = [10_000, 60_000, 300_000, 1_800_000, 7_200_000, 21_600_000, 43_200_000];++const BATCH = 20;++++/** A fresh per-task secret: whsec_ + 32 random bytes, base64url (43 characters). */++const mintSecret = () => `whsec_${crypto.randomBytes(32).toString('base64url')}`;++++function createWebhooks({ s, config, post, now = () => Date.now(), log = console }) {++ let timer = null;++ let running = null;++++ /** Queue the deliveries a task's state change asks for (none when it registered no webhook for that state). */++ async function enqueue(taskId, state) {++ const row = await store.getTask(s, taskId);++ const hooks = store.parseJson(row && row.webhooks) || [];++ const wanted = hooks.filter((h) => Array.isArray(h.events) && h.events.includes(state));++ if (!wanted.length) return 0;++ const task = store.toWire(row, { baseUrl: config.baseUrl });++ for (const h of wanted) {++ const id = s.newId('whd');++ const body = { type: 'actor.task.state', delivery_id: id, state, task };++ await s.db.query(`INSERT INTO webhook_deliveries (id, task_id, url, state, body, status, attempts, next_at, created_at)++ VALUES ($1, $2, $3, $4, $5::jsonb, 'pending', 0, $6, $7)`, [id, taskId, h.url, state, JSON.stringify(body), now(), s.iso()]);++ }++ return wanted.length;++ }++++ async function claim() {++ return await s.tx(async () => await s.db.many(`UPDATE webhook_deliveries SET status = 'sending'++ WHERE id IN (SELECT id FROM webhook_deliveries WHERE status = 'pending' AND next_at <= $1 ORDER BY next_at, id LIMIT $2 FOR UPDATE SKIP LOCKED)++ RETURNING *`, [now(), BATCH]));++ }++++ async function finish(d, status, { lastStatus = null, error = null, attempts }) {++ await s.db.query(`UPDATE webhook_deliveries SET status = $2, attempts = $3, last_status = $4, last_error = $5, finished_at = $6 WHERE id = $1`,++ [d.id, status, attempts, lastStatus, error, s.iso()]);++ await forgetSecretIfDone(d.task_id);++ }++++ /** The task has ended and nothing is left to send: its secret goes. */++ async function forgetSecretIfDone(taskId) {++ await s.db.query(`UPDATE tasks SET webhook_secret = NULL WHERE id = $1 AND webhook_secret IS NOT NULL AND state NOT IN ('queued', 'running', 'verifying')++ AND NOT EXISTS (SELECT 1 FROM webhook_deliveries WHERE task_id = $1 AND status IN ('pending', 'sending'))`, [taskId]);++ }++++ async function sendOne(d) {++ const attempts = Number(d.attempts) + 1;++ const secret = await s.db.value('SELECT webhook_secret FROM tasks WHERE id = $1', [d.task_id]);++ if (!secret) return finish(d, 'failed', { error: 'the task has no signing secret any more', attempts });++ const body = JSON.stringify(store.parseJson(d.body));++ const headers = {++ ...signDeliveryHeaders(body, secret, { now: now() }),++ 'X-OpenVibe-Delivery-Id': d.id,++ 'X-OpenVibe-Task-Id': d.task_id,++ 'X-OpenVibe-Task-State': d.state,++ 'X-OpenVibe-Delivery-Attempt': String(attempts),++ };++ let res = null;++ let error = null;++ try { res = await post(d.url, { headers, body }); } catch (err) { error = err; }++ if (res && res.status >= 200 && res.status < 300) return finish(d, 'delivered', { lastStatus: res.status, attempts });++ if ((res && res.status === 410) || error instanceof WebhookRefused) {++ return finish(d, 'failed', { lastStatus: res ? res.status : null, error: error ? error.message : 'the receiver answered 410 Gone', attempts });++ }++ const why = error ? String(error.message || error).slice(0, 300) : `HTTP ${res.status}`;++ if (attempts > BACKOFF_MS.length) return finish(d, 'failed', { lastStatus: res ? res.status : null, error: why, attempts });++ await s.db.query(`UPDATE webhook_deliveries SET status = 'pending', attempts = $2, next_at = $3, last_status = $4, last_error = $5 WHERE id = $1`,++ [d.id, attempts, now() + BACKOFF_MS[attempts - 1], res ? res.status : null, why]);++ return undefined;++ }++++ /** One pass over what is due. One pass at a time. */++ function tick() {++ if (running) return running;++ running = (async () => {++ let sent = 0;++ for (const d of await claim()) {++ try { await sendOne(d); sent++; } catch (err) {++ log.warn('[Actor] webhook delivery crashed:', d.id, err && err.message);++ await s.db.query(`UPDATE webhook_deliveries SET status = 'pending', next_at = $2 WHERE id = $1 AND status = 'sending'`, [d.id, now() + BACKOFF_MS[0]]).catch(() => {});++ }++ }++ return sent;++ })().finally(() => { running = null; });++ return running;++ }++++ /** At start: a row a stopped process left 'sending' is pending again (the receiver dedupes on the delivery id). */++ async function recover() {++ return await s.db.exec(`UPDATE webhook_deliveries SET status = 'pending' WHERE status = 'sending'`);++ }++++ function start(intervalMs = config.webhooks.intervalMs) {++ if (timer) return;++ timer = setInterval(() => { tick().catch((err) => log.warn('[Actor] webhook pass failed:', err && err.message)); }, intervalMs);++ timer.unref();++ }++ async function stop() {++ if (timer) clearInterval(timer);++ timer = null;++ if (running) await running.catch(() => {});++ }++++ /** A task's deliveries, newest first (the task page shows them; the receiver URL is the requester's own). */++ const forTask = (taskId) => s.db.many('SELECT id, url, state, status, attempts, last_status, last_error, created_at, finished_at FROM webhook_deliveries WHERE task_id = $1 ORDER BY created_at DESC, id DESC', [taskId]);++++ return { enqueue, tick, recover, start, stop, forTask, forgetSecretIfDone, BACKOFF_MS };++}++++module.exports = { createWebhooks, mintSecret, BACKOFF_MS };@@@ -42,7 +42,7 @@ async function boot(opts = {}) {const { createStore } = require('../../server/db');const testdb = await require('./db').testDb();const store = createStore(testdb.db, { now: opts.now });-- const built = await createApp({ config, store, log, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, adapters: opts.adapters, accountSend: opts.accountSend });++ const built = await createApp({ config, store, log, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, adapters: opts.adapters, accountSend: opts.accountSend, webhookPost: opts.webhookPost });await built.ctx.keys.ensure();const server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });const base = `http://127.0.0.1:${server.address().port}`;@@@ -85,6 +85,7 @@ async function boot(opts = {}) {logs: () => captured.join('\n'),async close() {await built.ctx.engine.stop();++ await built.ctx.webhooks.stop();await new Promise((r) => server.close(r));built.ctx.keys.client.stop();await testdb.close();@@@ -108,6 +108,7 @@ const SAME = { 'sec-fetch-site': 'same-origin' };'server/agents/providers.js': 'the configured provider base URLs (DeepSeek, OpenAI, the open model) with the operator\'s keys','server/agents/tools.js': 'the configured OpenVibe.Tools and OpenVibe.Search bases; ids come from Tools\' catalog, a page URL goes to Tools as input','server/events-consumer.js': 'OpenVibe.Events: the configured events.url (ACTOR_EVENTS_URL), to create the two ADR-033 subscriptions at boot',++ 'server/net/webhook-post.js': 'a task\'s webhook URL (plan T17), the one caller-chosen address: https on 443/8443 to public addresses only, checked at creation and at send time through openvibe-shared/egress safeLookup, never a redirect (test/webhooks.test.js)',};const root = path.join(__dirname, '..');const found = [];@@@ -178,13 +178,14 @@ const SAME = { 'sec-fetch-site': 'same-origin' };assert.strictEqual(t.providers.requests.filter((q) => q.path === '/openai/responses').length, 0);});-- await check('a budget above the free tier is refused, never lowered; an unknown field or a webhook is refused', async () => {++ await check('a budget above the free tier is refused, never lowered; an unknown field or a webhook secret by reference is refused', async () => {const over = await create({ task: 'x', budget: { per_task_usd: 5 } });assert.strictEqual(over.status, 422);assert.strictEqual(over.json().code, 'actor.budget.over_tier');-- const hooks = await create({ task: 'x', webhooks: [{ url: 'https://example.com', events: ['succeeded'] }] });++ const hooks = await create({ task: 'x', webhooks: [{ url: 'https://example.com', events: ['succeeded'], secret_ref: 'HOOK_KEY' }] });assert.strictEqual(hooks.status, 422);assert.strictEqual(hooks.json().code, 'actor.task.invalid');++ assert.strictEqual((await create({ task: 'x', surprise: true })).json().code, 'actor.task.invalid');assert.strictEqual((await create({ task: ' ' })).status, 422);});nnew file mode 100644@@@ -0,0 +1,200 @@++'use strict';++/**++ * Task webhooks (plan T17, actor.task-webhook@1): a task registers https endpoints and the states each wants; the++ * creating answer carries the task's signing secret once (reads and lists never do); each delivery is signed like an++ * Events delivery (openvibe-sdk/events verifyDeliveryV2 accepts it), carries the task as it was and a delivery id that++ * stays the same across retries; 2xx is delivered, 410 or a refused URL ends it, anything else is retried on the++ * backoff and then fails; the secret is erased once the task has ended and its deliveries are done, and it is never++ * exported. A URL that is not a public https address on 443/8443 is refused at creation, and the real poster refuses++ * a name that resolves to a private address at send time.++ */++const assert = require('assert');++const contracts = require('openvibe-contracts');++const { verifyDeliveryV2 } = require('openvibe-sdk/events');++const { createSafeLookup } = require('openvibe-shared/egress');++const { boot, check, done } = require('./helpers/boot');++const { checkWebhookUrl, createWebhookPoster, WebhookRefused } = require('../server/net/webhook-post');++++const SAME = { 'sec-fetch-site': 'same-origin' };++const HOOK = 'https://hooks.example.com/openvibe/actor';++++(async () => {++ // Actor's clock (the store's, which sign-in checks token expiry against too): real time, moved on by the retry checks.++ let clock = Date.now();++ const sent = [];++ let answers = []; // what the stand-in receiver answers next: a status, or an Error to throw++ const webhookPost = async (url, { headers, body }) => {++ checkWebhookUrl(url); // the same check the real poster makes first++ sent.push({ url, headers, body });++ const a = answers.length ? answers.shift() : 200;++ if (a instanceof Error) throw a;++ return { status: a };++ };++ const t = await boot({ webhookPost, now: () => clock });++ const kim = t.network.addUser('kim');++ const create = (json) => t.get('/api/v1/tasks', { as: kim, json, headers: SAME });++ const tick = () => t.ctx.webhooks.tick();++ const deliveries = (taskId) => t.ctx.s.db.many('SELECT * FROM webhook_deliveries WHERE task_id = $1 ORDER BY created_at, id', [taskId]);++ const secretOf = (taskId) => t.ctx.s.db.value('SELECT webhook_secret FROM tasks WHERE id = $1', [taskId]);++ // The engine queues a delivery just after it records the state: wait for the rows before sending.++ async function queued(taskId, n) {++ for (let i = 0; i < 200; i++) { if ((await deliveries(taskId)).length >= n) return; await new Promise((res) => setTimeout(res, 20)); }++ throw new Error(`${taskId}: ${n} deliveries never queued`);++ }++++ let first;++ await check('the creating answer carries the signing secret once; reads and lists show the webhooks but never the secret', async () => {++ t.providers.reset();++ const r = await create({ task: 'Who handles mail for openvibe.network?', webhooks: [{ url: HOOK, events: ['queued', 'succeeded', 'failed'] }] });++ assert.strictEqual(r.status, 201, r.text);++ first = r.json();++ assert.match(first.webhook_secret, /^whsec_[A-Za-z0-9_-]{43}$/);++ assert.ok(contracts.validate('platform.task@1', first).valid, JSON.stringify(contracts.validate('platform.task@1', first).errors));++ assert.deepStrictEqual(first.webhooks, [{ url: HOOK, events: ['queued', 'succeeded', 'failed'] }]);++ const read = (await t.get(`/api/v1/tasks/${first.id}`, { as: kim })).json();++ assert.strictEqual('webhook_secret' in read, false, 'a read never carries the secret');++ assert.deepStrictEqual(read.webhooks, first.webhooks);++ const list = (await t.get('/api/v1/tasks', { as: kim })).json();++ assert.ok(!JSON.stringify(list).includes(first.webhook_secret), 'a list never carries the secret');++ });++++ await check('each delivery is signed like an Events delivery, carries the task as it was, and validates', async () => {++ await t.waitFor(first.id);++ await queued(first.id, 2);++ await tick();++ assert.deepStrictEqual(sent.map((d) => d.url), [HOOK, HOOK]);++ for (const d of sent) {++ assert.ok(verifyDeliveryV2(d.body, d.headers, first.webhook_secret, { now: clock }), 'verifyDeliveryV2 accepts it with the task\'s secret');++ assert.ok(!verifyDeliveryV2(d.body, d.headers, 'whsec_someone-elses-secret', { now: clock }), 'and nothing else');++ const body = JSON.parse(d.body);++ const v = contracts.validate('actor.task-webhook@1', body);++ assert.ok(v.valid, JSON.stringify(v.errors));++ assert.strictEqual(body.task.id, first.id);++ assert.strictEqual(d.headers['X-OpenVibe-Delivery-Id'], body.delivery_id);++ assert.strictEqual(d.headers['X-OpenVibe-Task-Id'], first.id);++ assert.strictEqual(d.headers['X-OpenVibe-Task-State'], body.state);++ assert.strictEqual(d.headers['X-OpenVibe-Delivery-Attempt'], '1');++ assert.ok(!d.body.includes(first.webhook_secret), 'the body never carries the secret');++ }++ assert.deepStrictEqual(sent.map((d) => JSON.parse(d.body).state), ['queued', 'succeeded']);++ assert.strictEqual(JSON.parse(sent[0].body).task.state, 'queued', 'the queued delivery has the task as it was then');++ assert.strictEqual(JSON.parse(sent[1].body).task.result.agent, 'openvibe-runtime');++ assert.deepStrictEqual((await deliveries(first.id)).map((d) => d.status), ['delivered', 'delivered']);++ assert.strictEqual(await secretOf(first.id), null, 'the task ended and its deliveries are done: the secret is gone');++ const pageHtml = (await t.get(`/tasks/${first.id}`, { as: kim })).text;++ assert.match(pageHtml, /<summary>Webhooks \(2\)<\/summary>/, 'the task page lists its deliveries');++ assert.ok(pageHtml.includes('Delivered') && pageHtml.includes(HOOK) && !pageHtml.includes('whsec_'), 'with how each went, and never the secret');++ });++++ await check('an idempotent repeat answers the same secret while the task still has one', async () => {++ t.providers.reset();++ const body = { task: 'Say hi', idempotency_key: 'hooks-idem-1', webhooks: [{ url: HOOK, events: ['failed'] }] };++ const a = await create(body);++ const b = await create(body);++ assert.deepStrictEqual([a.status, b.status], [201, 200]);++ assert.strictEqual(b.json().webhook_secret, a.json().webhook_secret);++ assert.strictEqual((await create({ ...body, webhooks: [{ url: `${HOOK}/other`, events: ['failed'] }] })).status, 409, 'other webhooks are another body');++ await t.waitFor(a.json().id);++ for (let i = 0; i < 200 && await secretOf(a.json().id); i++) await new Promise((res) => setTimeout(res, 20));++ assert.strictEqual(await secretOf(a.json().id), null, 'no delivery for its end state: the secret goes when it ends');++ assert.deepStrictEqual(await deliveries(a.json().id), []);++ });++++ await check('a failing receiver is retried on the backoff with the same delivery id; 2xx then delivers it', async () => {++ t.providers.reset();++ sent.length = 0;++ const r = (await create({ task: 'Who handles mail for openvibe.network?', webhooks: [{ url: HOOK, events: ['succeeded'] }] })).json();++ await t.waitFor(r.id);++ await queued(r.id, 1);++ answers = [500, Object.assign(new Error('connect ECONNREFUSED'), { code: 'ECONNREFUSED' })];++ await tick();++ let [d] = await deliveries(r.id);++ assert.deepStrictEqual([d.status, d.attempts, d.last_status, Number(d.next_at)], ['pending', 1, 500, clock + 10_000]);++ await tick();++ assert.strictEqual(sent.length, 1, 'not due yet: nothing sent');++ clock += 10_000;++ await tick();++ [d] = await deliveries(r.id);++ assert.deepStrictEqual([d.status, d.attempts, Number(d.next_at)], ['pending', 2, clock + 60_000]);++ assert.match(d.last_error, /ECONNREFUSED/);++ clock += 60_000;++ await tick();++ [d] = await deliveries(r.id);++ assert.strictEqual(d.status, 'delivered');++ assert.deepStrictEqual(sent.map((x) => x.headers['X-OpenVibe-Delivery-Id']), [d.id, d.id, d.id], 'one delivery id across the attempts');++ assert.deepStrictEqual(sent.map((x) => x.headers['X-OpenVibe-Delivery-Attempt']), ['1', '2', '3']);++ assert.ok(verifyDeliveryV2(sent[2].body, sent[2].headers, r.webhook_secret, { now: clock }), 'each attempt is signed afresh (its own timestamp)');++ assert.strictEqual(await secretOf(r.id), null);++ });++++ await check('410 Gone or a URL refused at send time ends the delivery at once; endless failures end after the backoff', async () => {++ t.providers.reset();++ const r = (await create({ task: 'Who handles mail for openvibe.network?', webhooks: [{ url: `${HOOK}/a`, events: ['succeeded'] }, { url: `${HOOK}/b`, events: ['succeeded'] }, { url: `${HOOK}/c`, events: ['succeeded'] }] })).json();++ await t.waitFor(r.id);++ await queued(r.id, 3);++ answers = [410, new WebhookRefused('hooks.example.com is not a public address')];++ for (let i = 0; i < 20; i++) answers.push(503);++ await tick();++ const byUrl = async () => Object.fromEntries((await deliveries(r.id)).map((d) => [d.url.slice(-1), d]));++ let d = await byUrl();++ assert.deepStrictEqual([d.a.status, d.a.last_status], ['failed', 410]);++ assert.strictEqual(d.b.status, 'failed');++ assert.match(d.b.last_error, /not a public address/);++ assert.strictEqual(d.c.status, 'pending');++ for (const wait of t.ctx.webhooks.BACKOFF_MS) { clock += wait; await tick(); }++ d = await byUrl();++ assert.deepStrictEqual([d.c.status, d.c.attempts, d.c.last_status], ['failed', t.ctx.webhooks.BACKOFF_MS.length + 1, 503]);++ assert.strictEqual(await secretOf(r.id), null, 'nothing left to send: the secret is gone');++ clock = Date.now(); // back to real time: the next checks sign in with tokens that expire in real time++ answers = [];++ });++++ await check('cancel is delivered to a webhook that asked for it', async () => {++ t.providers.reset();++ sent.length = 0;++ t.providers.state.cls = 'lookup';++ t.providers.state.runtime = 'hang';++ const r = (await create({ task: 'MX of example.com please', webhooks: [{ url: HOOK, events: ['cancelled'] }] })).json();++ await t.waitFor(r.id, ['running']);++ await new Promise((res) => setTimeout(res, 150));++ await t.get(`/api/v1/tasks/${r.id}/cancel`, { as: kim, method: 'POST', headers: SAME });++ await t.waitFor(r.id, ['cancelled']);++ await queued(r.id, 1);++ await tick();++ assert.deepStrictEqual(sent.map((x) => JSON.parse(x.body).state), ['cancelled']);++ assert.strictEqual(JSON.parse(sent[0].body).task.cancel.requested_by.id, kim.subject);++ assert.deepStrictEqual((await deliveries(r.id)).map((d) => d.status), ['delivered']);++ assert.strictEqual(await secretOf(r.id), null);++ });++++ await check('a URL Actor may not deliver to is refused at creation; a project secret by reference is not accepted yet', async () => {++ for (const url of ['https://127.0.0.1/x', 'https://localhost/x', 'https://10.0.0.7/x', 'https://[::1]/x', 'https://169.254.169.254/latest', 'https://hooks.example.com:8080/x', 'https://user:[email protected]/x', 'https://intranet/x']) {++ const r = await create({ task: 'x', webhooks: [{ url, events: ['succeeded'] }] });++ assert.strictEqual(r.status, 422, `${url}: ${r.status} ${r.text.slice(0, 200)}`);++ assert.strictEqual(r.json().code, 'actor.webhook.refused', url);++ }++ const http = await create({ task: 'x', webhooks: [{ url: 'http://hooks.example.com/x', events: ['succeeded'] }] });++ assert.deepStrictEqual([http.status, http.json().code], [422, 'actor.task.invalid'], 'plain http is not a webhook URL');++ const ref = await create({ task: 'x', webhooks: [{ url: HOOK, events: ['succeeded'], secret_ref: 'HOOK_KEY' }] });++ assert.deepStrictEqual([ref.status, ref.json().code], [422, 'actor.task.invalid']);++ assert.ok(checkWebhookUrl('https://hooks.example.com:8443/x'), 'port 8443 is allowed');++ });++++ await check('the real poster refuses a name that resolves to a private address, at send time', async () => {++ const lookup = createSafeLookup({ lookup: (host, opts, cb) => cb(null, [{ address: '10.1.2.3', family: 4 }]) });++ const post = createWebhookPoster({ lookup, timeoutMs: 2000 });++ await assert.rejects(post('https://hooks.example.com/x', { body: '{}' }), /10\.1\.2\.3|not public|no public|EGRESS|denied/i);++ await assert.rejects(post('https://127.0.0.1/x', { body: '{}' }), WebhookRefused);++ });++++ await check('the signing secret is never exported with the person\'s tasks', async () => {++ const { TABLES } = require('../server/identity/account-data');++ const tasks = TABLES.find((x) => x.table === 'tasks');++ assert.ok(Array.isArray(tasks.columns) && tasks.columns.includes('webhooks') && !tasks.columns.includes('webhook_secret'));++ const cols = (await t.ctx.s.db.many("SELECT column_name FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'tasks'")).map((r) => r.column_name);++ assert.deepStrictEqual(cols.filter((c) => !tasks.columns.includes(c)), ['webhook_secret'], 'every other column of tasks is exported');++ });++++ await t.close();++ done();++})().catch((e) => { console.error(e); process.exit(1); });
@@ -51,6 +51,9 @@ ACTOR_LOCAL_LLM_MODEL=
@@ -51,6 +51,9 @@ ACTOR_LOCAL_LLM_MODEL=
#ACTOR_MAX_COST_USD_PER_DAY=2
#ACTOR_MAX_COST_USD_PER_DAY=2
#ACTOR_MAX_RUNNING=4
#ACTOR_MAX_RUNNING=4
#ACTOR_TASK_TIMEOUT_MS=180000
#ACTOR_TASK_TIMEOUT_MS=180000
+# Task webhooks: how often due deliveries go out, and how long one delivery may take
+#ACTOR_WEBHOOK_INTERVAL_MS=5000
+#ACTOR_WEBHOOK_TIMEOUT_MS=10000
# optional: defaults to true in production
# optional: defaults to true in production
# COOKIE_SECURE=true
# COOKIE_SECURE=true
@@ -41,7 +41,7 @@ Rate cards are typed in by a person from each provider's price page, with the pa
@@ -41,7 +41,7 @@ Rate cards are typed in by a person from each provider's price page, with the pa
| Route | Capability | |
| Route | Capability | |
|---|---|---|
|---|---|---|
-| `POST /api/v1/tasks` | `actor.task.create` | `actor.task-create-request@1` → `platform.task@1` (201, queued) |
+| `POST /api/v1/tasks` | `actor.task.create` | `actor.task-create-request@1` → `platform.task@1` (201, queued; with `webhooks`, the answer carries `webhook_secret` once) |
| `GET /api/v1/tasks/:id` | `actor.task.read` | the task |
| `GET /api/v1/tasks/:id` | `actor.task.read` | the task |
| `GET /api/v1/tasks/:id/events` | `actor.task.read` | server-sent events, `actor.task-event@1`, resumable with `Last-Event-ID` |
| `GET /api/v1/tasks/:id/events` | `actor.task.read` | server-sent events, `actor.task-event@1`, resumable with `Last-Event-ID` |
| `POST /api/v1/tasks/:id/cancel` | `actor.task.create` | stop it |
| `POST /api/v1/tasks/:id/cancel` | `actor.task.create` | stop it |
@@ -59,6 +59,12 @@ The resource index is for OpenVibe.Services on loopback and is blocked at the pu
@@ -59,6 +59,12 @@ The resource index is for OpenVibe.Services on loopback and is blocked at the pu
**What a caller can see:** a task is visible only to its requester, or to tokens of its project. Anyone else gets 404, the same answer as a task that doesn't exist.
**What a caller can see:** a task is visible only to its requester, or to tokens of its project. Anyone else gets 404, the same answer as a task that doesn't exist.
+**Webhooks:** a task may register up to 8 https endpoints, each with the state changes to deliver (`webhooks: [{ url, events }]`, the names of `state`).
+- **Signing:** the answer that creates the task carries `webhook_secret`, minted for that task. It is in no read or list, never exported, and erased once the task has ended and its deliveries are done. Each delivery is `actor.task-webhook@1` (`{ type: "actor.task.state", delivery_id, state, task }`, the task as it was then), signed like an Events delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`), so a receiver checks it with openvibe-sdk/events `verifyDeliveryV2(rawBody, headers, secret)`. The delivery id stays the same across retries; dedupe on it.
+- **Outcomes:** 2xx is delivered; 410, or a URL refused at send time, ends the delivery; anything else is retried after 10 s, 1 min, 5 min, 30 min, 2 h, 6 h and 12 h, then fails.
+- **Where Actor may connect:** a webhook URL is the one address a caller chooses (`server/net/webhook-post.js`): https on port 443 or 8443, no credentials, public addresses only. It is checked at creation (`actor.webhook.refused`) and again at send time through openvibe-shared/egress, so a name that later resolves to a private address is refused. Redirects are never followed.
+- **Not yet:** a project secret named by reference (`secret_ref`) is refused until project secrets exist.
+
**Errors:** RFC 9457 problems. The codes are on [openvibe.actor/docs](https://openvibe.actor/docs#errors).
**Errors:** RFC 9457 problems. The codes are on [openvibe.actor/docs](https://openvibe.actor/docs#errors).
## Money
## Money
@@ -193,7 +199,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).
@@ -193,7 +199,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).
Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
<!-- versions:start -->
<!-- versions:start -->
-- openvibe-contracts: v0.129.0
+- openvibe-contracts: v0.133.0
- openvibe-sdk: v0.38.0
- openvibe-sdk: v0.38.0
- openvibe-shared: v3.0.1
- openvibe-shared: v3.0.1
<!-- versions:end -->
<!-- versions:end -->
@@ -7,11 +7,11 @@
@@ -7,11 +7,11 @@
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
"port": 4950,
"port": 4950,
"domain": "openvibe.actor",
"domain": "openvibe.actor",
- "contracts": "openvibe-contracts v0.129.0",
+ "contracts": "openvibe-contracts v0.133.0",
"sdk": "openvibe-sdk v0.38.0",
"sdk": "openvibe-sdk v0.38.0",
"shared": "openvibe-shared v3.0.1",
"shared": "openvibe-shared v3.0.1",
"packages": [
"packages": [
- "openvibe-contracts v0.129.0",
+ "openvibe-contracts v0.133.0",
"openvibe-sdk v0.38.0",
"openvibe-sdk v0.38.0",
"openvibe-shared v3.0.1"
"openvibe-shared v3.0.1"
],
],
@@ -46,7 +46,8 @@
@@ -46,7 +46,8 @@
"other platforms' agents",
"other platforms' agents",
"private mode on OpenVibe hardware",
"private mode on OpenVibe hardware",
"cross-family answer checks",
"cross-family answer checks",
- "live task progress"
+ "live task progress",
+ "task webhooks (signed per task, retried with backoff, public https endpoints only)"
],
],
"updated": "2026-10-10"
"updated": "2026-10-10"
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,24 @@
@@ -0,0 +1,24 @@
+-- phase: expand
+-- OpenVibe.Actor: task webhooks (plan T17, actor.task-webhook@1). `webhooks` is what the task registered (url + the
+-- state changes to deliver); `webhook_secret` signs its deliveries, minted per task, shown once to its creator and
+-- erased once the task has ended and its deliveries are done. Each delivery is a row, retried with backoff; a delivery
+-- never delays or changes the task. The secret is never exported (server/identity/account-data.js lists the columns).
+ALTER TABLE tasks ADD COLUMN IF NOT EXISTS webhooks jsonb;
+ALTER TABLE tasks ADD COLUMN IF NOT EXISTS webhook_secret text COLLATE "C";
+
+CREATE TABLE IF NOT EXISTS webhook_deliveries (
+ id text COLLATE "C" PRIMARY KEY, -- whd_<ULID>, stable across retries
+ task_id text COLLATE "C" NOT NULL REFERENCES tasks (id) ON DELETE CASCADE,
+ url text NOT NULL,
+ state text COLLATE "C" NOT NULL, -- the task state it announces
+ body jsonb NOT NULL, -- actor.task-webhook@1 as it was then
+ status text COLLATE "C" NOT NULL DEFAULT 'pending', -- pending | sending | delivered | failed
+ attempts integer NOT NULL DEFAULT 0,
+ next_at bigint NOT NULL, -- epoch ms of the next attempt
+ last_status integer,
+ last_error text,
+ created_at text COLLATE "C" NOT NULL,
+ finished_at text COLLATE "C"
+);
+CREATE INDEX IF NOT EXISTS webhook_deliveries_due ON webhook_deliveries (next_at) WHERE status = 'pending';
+CREATE INDEX IF NOT EXISTS webhook_deliveries_task ON webhook_deliveries (task_id);
@@ -15,7 +15,7 @@
@@ -15,7 +15,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "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.133.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.1",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -871,9 +871,9 @@
@@ -871,9 +871,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.133.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.133.0",
+ "integrity": "sha512-kER/ktuvD74FU2xKKByWZ/8fvFbVTwhWyGhdCuMJL9XU5F9pmWV27+PL5MZ4VCMN9EolNyHh5FgprhhOW+M4kQ==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -27,7 +27,7 @@
@@ -27,7 +27,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "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.133.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.1",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -24,6 +24,8 @@ const { createSso } = require('./auth/sso');
@@ -24,6 +24,8 @@ const { createSso } = require('./auth/sso');
const catalog = require('./agents/catalog');
const catalog = require('./agents/catalog');
const { createAdapters } = require('./agents/adapters');
const { createAdapters } = require('./agents/adapters');
const { createEngine } = require('./tasks/engine');
const { createEngine } = require('./tasks/engine');
+const { createWebhooks } = require('./tasks/webhooks');
+const { createWebhookPoster } = require('./net/webhook-post');
const { createTaskStream } = require('./tasks/stream');
const { createTaskStream } = require('./tasks/stream');
const { createPrincipal } = require('./http/principal');
const { createPrincipal } = require('./http/principal');
const { createApi } = require('./http/api');
const { createApi } = require('./http/api');
@@ -42,7 +44,8 @@ const VERSION = require('../package.json').version;
@@ -42,7 +44,8 @@ const VERSION = require('../package.json').version;
/**
/**
* opts: config, store, now (clock), fetchImpl (Network), providerFetch (model providers and OpenVibe services), log,
* opts: config, store, now (clock), fetchImpl (Network), providerFetch (model providers and OpenVibe services), log,
* limitsNow, callerLimits (false: count nobody, tests only), valkey, adapters (tests: stand-in agents),
* limitsNow, callerLimits (false: count nobody, tests only), valkey, adapters (tests: stand-in agents),
- * accountSend (a stand-in for Network's internal routes; tests only)
+ * accountSend (a stand-in for Network's internal routes; tests only), webhookPost (a stand-in for the guarded webhook
+ * poster; tests only)
*/
*/
async function createApp(opts = {}) {
async function createApp(opts = {}) {
const config = opts.config || configLib.load();
const config = opts.config || configLib.load();
@@ -54,9 +57,14 @@ async function createApp(opts = {}) {
@@ -54,9 +57,14 @@ async function createApp(opts = {}) {
const sso = createSso({ config, keys, fetchImpl, now: s.now, log });
const sso = createSso({ config, keys, fetchImpl, now: s.now, log });
const stream = createTaskStream({ log });
const stream = createTaskStream({ log });
const adapters = opts.adapters || createAdapters({ config, rates: catalog.RATES, fetchImpl: opts.providerFetch || globalThis.fetch, log });
const adapters = opts.adapters || createAdapters({ config, rates: catalog.RATES, fetchImpl: opts.providerFetch || globalThis.fetch, log });
- const engine = createEngine({ config, s, adapters, stream, now: s.now, log });
+ // Task webhooks (plan T17): the only connections to an address a caller chose, through server/net/webhook-post.js.
+ const webhooks = createWebhooks({
+ s, config, now: s.now, log,
+ post: opts.webhookPost || createWebhookPoster({ userAgent: `OpenVibe.Actor/${VERSION} (+https://openvibe.actor/docs#webhooks)`, timeoutMs: config.webhooks.timeoutMs }),
+ });
+ const engine = createEngine({ config, s, adapters, stream, webhooks, now: s.now, log });
const principal = createPrincipal({ config, keys });
const principal = createPrincipal({ config, keys });
- const ctx = { config, s, keys, sso, stream, adapters, engine, principal, log };
+ const ctx = { config, s, keys, sso, stream, adapters, engine, webhooks, principal, log };
// Account export and deletion (ADR-033, ./identity/account-data.js): the two tables that hold a person's rows.
// Account export and deletion (ADR-033, ./identity/account-data.js): the two tables that hold a person's rows.
// The sender posts to Network's internal export/deletion routes with this service's own client-credentials token;
// The sender posts to Network's internal export/deletion routes with this service's own client-credentials token;
@@ -79,6 +79,11 @@ function load(env = process.env) {
@@ -79,6 +79,11 @@ function load(env = process.env) {
maxRunning: Math.max(1, int(env.ACTOR_MAX_RUNNING, 4)),
maxRunning: Math.max(1, int(env.ACTOR_MAX_RUNNING, 4)),
// A task that runs longer than this is stopped and fails (actor.task.timeout).
// A task that runs longer than this is stopped and fails (actor.task.timeout).
taskTimeoutMs: Math.max(10_000, int(env.ACTOR_TASK_TIMEOUT_MS, 180_000)),
taskTimeoutMs: Math.max(10_000, int(env.ACTOR_TASK_TIMEOUT_MS, 180_000)),
+ // Task webhooks (server/tasks/webhooks.js): how often due deliveries are sent, and how long one may take.
+ webhooks: {
+ intervalMs: Math.max(1000, int(env.ACTOR_WEBHOOK_INTERVAL_MS, 5000)),
+ timeoutMs: Math.max(1000, int(env.ACTOR_WEBHOOK_TIMEOUT_MS, 10_000)),
+ },
// OpenVibe.Events → this service (server/events-consumer.js). The secret signs a delivery (comma-separated
// OpenVibe.Events → this service (server/events-consumer.js). The secret signs a delivery (comma-separated
// for rotation, 32+ characters each); unset turns POST /internal/events off (503). The url is where the
// for rotation, 32+ characters each); unset turns POST /internal/events off (503). The url is where the
@@ -5,7 +5,8 @@
@@ -5,7 +5,8 @@
*
*
* GET /agents actor.agent.read (public) the agent systems, their rate cards and availability
* GET /agents actor.agent.read (public) the agent systems, their rate cards and availability
* POST /route actor.agent.read (public) which agent a task would go to, and why (nothing runs)
* POST /route actor.agent.read (public) which agent a task would go to, and why (nothing runs)
- * POST /tasks actor.task.create give Actor a task (actor.task-create-request@1 → platform.task@1)
+ * POST /tasks actor.task.create give Actor a task (actor.task-create-request@1 → platform.task@1;
+ * with webhooks, the answer carries webhook_secret once)
* GET /tasks actor.task.list your tasks, newest first
* GET /tasks actor.task.list your tasks, newest first
* GET /tasks/:id actor.task.read one task
* GET /tasks/:id actor.task.read one task
* GET /tasks/:id/events actor.task.read its live stream (actor.task-event@1, Last-Event-ID)
* GET /tasks/:id/events actor.task.read its live stream (actor.task-event@1, Last-Event-ID)
@@ -26,7 +27,7 @@ const { createTask } = require('../tasks/create');
@@ -26,7 +27,7 @@ const { createTask } = require('../tasks/create');
const TASK_ID = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;
const TASK_ID = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;
function createApi(ctx) {
function createApi(ctx) {
- const { config, s, engine, stream, principal, limits } = ctx;
+ const { config, s, engine, stream, principal, limits, webhooks } = ctx;
const r = asyncRouter();
const r = asyncRouter();
const problem = (req, res, status, code, detail, extra) => contracts.http.sendProblem(res, status, code, { detail, ctx: req.ov, extra });
const problem = (req, res, status, code, detail, extra) => contracts.http.sendProblem(res, status, code, { detail, ctx: req.ov, extra });
@@ -60,13 +61,14 @@ function createApi(ctx) {
@@ -60,13 +61,14 @@ function createApi(ctx) {
const wire = (row) => store.toWire(row, { baseUrl: config.baseUrl });
const wire = (row) => store.toWire(row, { baseUrl: config.baseUrl });
r.post('/tasks', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), async (req, res) => {
r.post('/tasks', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), async (req, res) => {
- const out = await createTask({ s, config, engine, principal: req.principal, body: req.body || {} });
+ const out = await createTask({ s, config, engine, webhooks, principal: req.principal, body: req.body || {} });
if (out.code) {
if (out.code) {
if (out.retryAfter) res.set('Retry-After', String(out.retryAfter));
if (out.retryAfter) res.set('Retry-After', String(out.retryAfter));
return problem(req, res, out.status, out.code, out.detail);
return problem(req, res, out.status, out.code, out.detail);
}
}
if (out.status === 201) res.set('Location', `/api/v1/tasks/${out.task.id}`);
if (out.status === 201) res.set('Location', `/api/v1/tasks/${out.task.id}`);
- return res.status(out.status).json(wire(out.task));
+ // The signing secret, to its creator only, in this answer (and an idempotent repeat of it) and never again.
+ return res.status(out.status).json({ ...wire(out.task), ...(out.task.webhook_secret ? { webhook_secret: out.task.webhook_secret } : {}) });
});
});
r.get('/tasks', principal.requireCapability('actor.task.list'), limits.reads('actor.task.list'), async (req, res) => {
r.get('/tasks', principal.requireCapability('actor.task.list'), limits.reads('actor.task.list'), async (req, res) => {
@@ -214,6 +214,7 @@ ${pageRows.next ? html`<p><a href="/tasks?before=${pageRows.next}">Older tasks</
@@ -214,6 +214,7 @@ ${pageRows.next ? html`<p><a href="/tasks?before=${pageRows.next}">Older tasks</
}
}
const t = store.toWire(row, { baseUrl: config.baseUrl });
const t = store.toWire(row, { baseUrl: config.baseUrl });
const events = await store.eventsAfter(s, row.id, 0);
const events = await store.eventsAfter(s, row.id, 0);
+ const deliveries = row.webhooks ? await ctx.webhooks.forTask(row.id) : [];
const open = store.OPEN.includes(row.state);
const open = store.OPEN.includes(row.state);
const cls = (events.find((e) => e.step === 'plan' && !e.agent && /"(\w+)" task/.test(e.text || '')) || { text: '' }).text.match(/"(\w+)"/);
const cls = (events.find((e) => e.step === 'plan' && !e.agent && /"(\w+)" task/.test(e.text || '')) || { text: '' }).text.match(/"(\w+)"/);
const kind = cls ? CLASS_TEXT[cls[1]] || cls[1] : null;
const kind = cls ? CLASS_TEXT[cls[1]] || cls[1] : null;
@@ -235,11 +236,22 @@ ${outcome(t, row)}
@@ -235,11 +236,22 @@ ${outcome(t, row)}
${open ? html`<p class="muted small live-note" id="live-note"><span class="dot-pulse" aria-hidden="true"></span> Updating as the agent works.</p>` : ''}
${open ? html`<p class="muted small live-note" id="live-note"><span class="dot-pulse" aria-hidden="true"></span> Updating as the agent works.</p>` : ''}
</section>
</section>
${t.explanation ? explanationBlock(t.explanation) : ''}
${t.explanation ? explanationBlock(t.explanation) : ''}
+${deliveries.length ? deliveriesBlock(deliveries) : ''}
<p class="muted small api-hint">In code: <code>GET /api/v1/tasks/${row.id}</code> · <a href="/docs">API</a></p>
<p class="muted small api-hint">In code: <code>GET /api/v1/tasks/${row.id}</code> · <a href="/docs">API</a></p>
</article>`,
</article>`,
});
});
});
});
+ /** The task's webhook deliveries: where, which state, and how it went (the receiver's answer code only). */
+ const DELIVERY_TEXT = { pending: 'Waiting to retry', sending: 'Sending', delivered: 'Delivered', failed: 'Not delivered' };
+ function deliveriesBlock(rows) {
+ return html`<details class="deliveries"><summary>Webhooks (${rows.length})</summary>
+${table(['State', 'Endpoint', 'Delivery', 'Attempts'], rows.map((d) => [
+ html`<code>${d.state}</code>`, html`<code class="small">${d.url}</code>`,
+ html`${DELIVERY_TEXT[d.status] || d.status}${d.last_status ? html` <span class="muted small">(${d.last_status})</span>` : ''}`, String(d.attempts),
+ ]))}</details>`;
+ }
+
/** The outcome card: the first thing under the task. */
/** The outcome card: the first thing under the task. */
function outcome(t, row) {
function outcome(t, row) {
if (t.state === 'succeeded') return resultBlock(t.result, t);
if (t.state === 'succeeded') return resultBlock(t.result, t);
@@ -395,7 +407,7 @@ data: {"task_id":"tsk_…","seq":3,"kind":"output","agent":"openvibe-runtime","s
@@ -395,7 +407,7 @@ data: {"task_id":"tsk_…","seq":3,"kind":"output","agent":"openvibe-runtime","s
<li><strong>An app or agent</strong>: a Network token for audience <code>openvibe.actor</code> with the <code>actor.task.*</code> capability the route needs, granted on <a href="https://openvibe.services">OpenVibe.Services</a>. Tasks belong to the app's project.</li>
<li><strong>An app or agent</strong>: a Network token for audience <code>openvibe.actor</code> with the <code>actor.task.*</code> capability the route needs, granted on <a href="https://openvibe.services">OpenVibe.Services</a>. Tasks belong to the app's project.</li>
<li><strong>Anyone</strong>: <code>GET /api/v1/agents</code> and <code>POST /api/v1/route</code> need no token.</li></ul>
<li><strong>Anyone</strong>: <code>GET /api/v1/agents</code> and <code>POST /api/v1/route</code> need no token.</li></ul>
${table(['Route', 'Capability', 'What it does'], [
${table(['Route', 'Capability', 'What it does'], [
- [html`<code>POST /api/v1/tasks</code>`, html`<code>actor.task.create</code>`, html`Create a task: <code>{ task, mode?, budget?: { per_task_usd, per_day_usd }, agent?, idempotency_key? }</code>. Answers 201 with the task (queued).`],
+ [html`<code>POST /api/v1/tasks</code>`, html`<code>actor.task.create</code>`, html`Create a task: <code>{ task, mode?, budget?: { per_task_usd, per_day_usd }, agent?, idempotency_key?, webhooks?: [{ url, events }] }</code>. Answers 201 with the task (queued); with webhooks, also its <code>webhook_secret</code>, this once (<a href="#webhooks">webhooks</a>).`],
[html`<code>GET /api/v1/tasks/:id</code>`, html`<code>actor.task.read</code>`, 'The task: state, result, cost, explanation, error.'],
[html`<code>GET /api/v1/tasks/:id</code>`, html`<code>actor.task.read</code>`, 'The task: state, result, cost, explanation, error.'],
[html`<code>GET /api/v1/tasks/:id/events</code>`, html`<code>actor.task.read</code>`, html`Server-sent events (<code>actor.task-event@1</code>): <code>state</code>, <code>output</code> (each step), <code>end</code>. Resumes with <code>Last-Event-ID</code>.`],
[html`<code>GET /api/v1/tasks/:id/events</code>`, html`<code>actor.task.read</code>`, html`Server-sent events (<code>actor.task-event@1</code>): <code>state</code>, <code>output</code> (each step), <code>end</code>. Resumes with <code>Last-Event-ID</code>.`],
[html`<code>POST /api/v1/tasks/:id/cancel</code>`, html`<code>actor.task.create</code>`, 'Stop it. What was spent stays spent.'],
[html`<code>POST /api/v1/tasks/:id/cancel</code>`, html`<code>actor.task.create</code>`, 'Stop it. What was spent stays spent.'],
@@ -404,12 +416,18 @@ ${table(['Route', 'Capability', 'What it does'], [
@@ -404,12 +416,18 @@ ${table(['Route', 'Capability', 'What it does'], [
[html`<code>POST /api/v1/route</code>`, 'public', html`<code>{ task, mode? }</code> → which agent would take it and why, for every candidate. Nothing runs and nothing is charged.`],
[html`<code>POST /api/v1/route</code>`, 'public', html`<code>{ task, mode? }</code> → which agent would take it and why, for every candidate. Nothing runs and nothing is charged.`],
])}
])}
${raw(showcase.code({ title: 'Create a task, then watch it', samples: [{ label: 'Create', lang: 'bash', code: create }, { label: 'Watch', lang: 'bash', code: watch }] }))}
${raw(showcase.code({ title: 'Create a task, then watch it', samples: [{ label: 'Create', lang: 'bash', code: create }, { label: 'Watch', lang: 'bash', code: watch }] }))}
+<h2 id="webhooks">Webhooks</h2>
+<p>Give a task up to 8 HTTPS endpoints and the state changes each wants (<code>queued</code>, <code>running</code>, <code>verifying</code>, <code>succeeded</code>, <code>failed</code>, <code>cancelled</code>). Actor POSTs <a href="https://openvibe.services/docs/contracts/actor.task-webhook@1"><code>actor.task-webhook@1</code></a> to each: <code>{ type: "actor.task.state", delivery_id, state, task }</code>, with the task as it was at that moment.</p>
+<ul><li><strong>Signed per task.</strong> The answer that creates the task carries <code>webhook_secret</code>, and nothing else ever shows it. Each delivery has <code>X-OpenVibe-Timestamp</code> and <code>X-OpenVibe-Signature-V2</code>, exactly like an OpenVibe.Events delivery, so <code>verifyDeliveryV2(rawBody, headers, secret)</code> from <code>openvibe-sdk/events</code> checks it (and the five-minute window).</li>
+<li><strong>Retried.</strong> A 2xx answer is delivered. 410 Gone ends it. Anything else is retried after 10 s, 1 min, 5 min, 30 min, 2 h, 6 h and 12 h. <code>delivery_id</code> (also <code>X-OpenVibe-Delivery-Id</code>) is the same on every attempt: dedupe on it.</li>
+<li><strong>Public endpoints only.</strong> HTTPS on port 443 or 8443, no credentials in the URL, and only public addresses, checked when the task is created and again when each delivery is sent. Redirects are not followed.</li></ul>
<h2 id="limits">Limits and money</h2>
<h2 id="limits">Limits and money</h2>
<p>The free tier: ${tier.tasksPerDay} tasks a day, up to ${usd(tier.perTaskUsd)} a task and ${usd(tier.perDayUsd)} a day (<a href="/limits.json">limits.json</a>). A budget above the tier is refused (<code>actor.budget.over_tier</code>), never lowered. A task whose cheapest capable agent would cost more than its budget fails before running (<code>actor.budget.exceeded</code>). <code>cost.usd</code> is what each model call and web search really cost at the <a href="/agents">published prices</a>.</p>
<p>The free tier: ${tier.tasksPerDay} tasks a day, up to ${usd(tier.perTaskUsd)} a task and ${usd(tier.perDayUsd)} a day (<a href="/limits.json">limits.json</a>). A budget above the tier is refused (<code>actor.budget.over_tier</code>), never lowered. A task whose cheapest capable agent would cost more than its budget fails before running (<code>actor.budget.exceeded</code>). <code>cost.usd</code> is what each model call and web search really cost at the <a href="/agents">published prices</a>.</p>
<h2 id="errors">Errors</h2>
<h2 id="errors">Errors</h2>
<p>Every refusal is RFC 9457 <code>application/problem+json</code> with a stable <code>code</code>. A task that fails says why in <code>error</code>:</p>
<p>Every refusal is RFC 9457 <code>application/problem+json</code> with a stable <code>code</code>. A task that fails says why in <code>error</code>:</p>
${table(['Code', 'Meaning'], [
${table(['Code', 'Meaning'], [
[html`<code>actor.task.invalid</code>`, 'The request does not match actor.task-create-request@1.'],
[html`<code>actor.task.invalid</code>`, 'The request does not match actor.task-create-request@1.'],
+ [html`<code>actor.webhook.refused</code>`, 'A webhook URL Actor will not deliver to: not HTTPS on 443/8443, credentials in it, or not a public address.'],
[html`<code>actor.budget.over_tier</code>`, 'The budget is above what the free tier allows.'],
[html`<code>actor.budget.over_tier</code>`, 'The budget is above what the free tier allows.'],
[html`<code>actor.allowance.exhausted</code> / <code>actor.budget.day_spent</code>`, 'Today\'s free tasks or budget are used; Retry-After says when it resets.'],
[html`<code>actor.allowance.exhausted</code> / <code>actor.budget.day_spent</code>`, 'Today\'s free tasks or budget are used; Retry-After says when it resets.'],
[html`<code>actor.capacity.spent</code>`, 'Actor\'s free capacity for everyone is used up for today.'],
[html`<code>actor.capacity.spent</code>`, 'Actor\'s free capacity for everyone is used up for today.'],
@@ -25,7 +25,12 @@ const { createAccountData, TOPICS } = require('openvibe-sdk/account-data');
@@ -25,7 +25,12 @@ const { createAccountData, TOPICS } = require('openvibe-sdk/account-data');
* store `user:usr_…` (server/tasks/store.js takes the requester straight from the principal).
* store `user:usr_…` (server/tasks/store.js takes the requester straight from the principal).
*/
*/
const TABLES = [
const TABLES = [
- { table: 'tasks', subject: 'requester', value: (usr) => `user:${usr}`, file: 'tasks.json' },
+ // Every column but webhook_secret (a task's signing secret is a credential, not the person's data).
+ {
+ table: 'tasks', subject: 'requester', value: (usr) => `user:${usr}`, file: 'tasks.json',
+ columns: ['id', 'requester', 'project_id', 'task', 'mode', 'agent', 'budget_task', 'budget_day', 'state', 'result', 'error_code', 'error_detail',
+ 'cost_usd', 'free_usd', 'explanation', 'cancel_at', 'cancel_by', 'last_seq', 'idem_key', 'idem_hash', 'created_at', 'finished_at', 'via', 'webhooks'],
+ },
// The person's own per-UTC-day spend accounting. Deleted: nothing in Actor keeps a person's spend record for a
// The person's own per-UTC-day spend accounting. Deleted: nothing in Actor keeps a person's spend record for a
// reason that outlives them. The `*` row (everyone's daily ceiling) does not match `user:usr_…` and stays — it is
// reason that outlives them. The `*` row (everyone's daily ceiling) does not match `user:usr_…` and stays — it is
// the operator's total, a separate row.
// the operator's total, a separate row.
@@ -19,7 +19,7 @@ const PRUNE_EVERY_MS = 6 * 60 * 60_000;
@@ -19,7 +19,7 @@ const PRUNE_EVERY_MS = 6 * 60 * 60_000;
function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
return gracefulStop({
return gracefulStop({
name: 'Actor', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
name: 'Actor', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
- close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.engine.stop(), () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],
+ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.engine.stop(), () => ctx.webhooks.stop(), () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],
});
});
}
}
@@ -29,6 +29,10 @@ async function start() {
@@ -29,6 +29,10 @@ async function start() {
// A task a stopped process left open did not finish: it is recorded so, never re-run.
// A task a stopped process left open did not finish: it is recorded so, never re-run.
const interrupted = await store.failInterrupted(ctx.s);
const interrupted = await store.failInterrupted(ctx.s);
if (interrupted.length) console.warn(`[Actor] ${interrupted.length} task(s) interrupted by the last stop were marked failed`);
if (interrupted.length) console.warn(`[Actor] ${interrupted.length} task(s) interrupted by the last stop were marked failed`);
+ // Their webhooks hear `failed` like any other end; deliveries a stop left half-sent go out again (same delivery id).
+ for (const id of interrupted) await ctx.webhooks.enqueue(id, 'failed').catch((err) => console.warn('[Actor] webhook enqueue failed:', id, err && err.message));
+ await ctx.webhooks.recover();
+ ctx.webhooks.start();
const server = app.listen(config.port, config.host, () => {
const server = app.listen(config.port, config.host, () => {
const available = require('./agents/catalog').list(config).filter((a) => a.available).map((a) => a.id);
const available = require('./agents/catalog').list(config).filter((a) => a.available).map((a) => a.id);
new file mode 100644
new file mode 100644
@@ -0,0 +1,62 @@
@@ -0,0 +1,62 @@
+'use strict';
+/**
+ * The one place Actor connects to an address a caller chose: a task's webhook (plan T17). Everything else Actor fetches
+ * goes to a configured base (test/security-ssrf.test.js keeps that list).
+ *
+ * checkWebhookUrl(url) → URL, or throws WebhookRefused: https only, port 443 or 8443, no credentials, not a name
+ * that can only mean this machine or a private network, not a non-public IP literal
+ * createWebhookPoster({ userAgent, timeoutMs, lookup }) → post(url, { headers, body }) → { status }
+ *
+ * The poster checks the URL again at send time and connects through openvibe-shared/egress safeLookup, which fails
+ * unless every address the name resolves to is public, so a DNS answer cannot change between the check and the
+ * connect. A redirect is never followed (a 3xx is the receiver's answer like any other), and at most a few KiB of the
+ * answer are read and discarded: nothing a receiver says is stored.
+ */
+const https = require('https');
+const net = require('net');
+const egress = require('openvibe-shared/egress');
+
+const PORTS = new Set(['', '443', '8443']);
+const MAX_RESPONSE_BYTES = 4096;
+
+class WebhookRefused extends Error {
+ constructor(detail) { super(detail); this.code = 'actor.webhook.refused'; }
+}
+
+function checkWebhookUrl(input) {
+ let u;
+ try { u = new URL(String(input)); } catch { throw new WebhookRefused('not a URL'); }
+ if (u.protocol !== 'https:') throw new WebhookRefused('a webhook URL is https');
+ if (u.username || u.password) throw new WebhookRefused('credentials in the URL are not allowed');
+ if (!PORTS.has(u.port)) throw new WebhookRefused('a webhook URL uses port 443 or 8443');
+ const host = egress.normalizeHost(u.hostname);
+ if (net.isIP(host) ? !egress.isPublicAddress(host) : egress.isInternalName(host)) throw new WebhookRefused(`${host} is not a public address`);
+ return u;
+}
+
+function createWebhookPoster({ userAgent = 'OpenVibe.Actor', timeoutMs = 10_000, lookup = egress.safeLookup } = {}) {
+ return async function post(url, { headers = {}, body = '' } = {}) {
+ const u = checkWebhookUrl(url);
+ const payload = Buffer.from(String(body));
+ return await new Promise((resolve, reject) => {
+ let settled = false;
+ const finish = (fn, v) => { if (!settled) { settled = true; fn(v); } };
+ const req = https.request(u, {
+ method: 'POST', lookup, timeout: timeoutMs,
+ headers: { 'Content-Type': 'application/json', 'Content-Length': payload.length, 'User-Agent': userAgent, ...headers },
+ }, (res) => {
+ let seen = 0;
+ res.on('data', (chunk) => { seen += chunk.length; if (seen > MAX_RESPONSE_BYTES) res.destroy(); });
+ const done = () => finish(resolve, { status: res.statusCode });
+ res.on('end', done);
+ res.on('close', done);
+ res.on('error', done);
+ });
+ req.on('timeout', () => req.destroy(Object.assign(new Error(`no answer within ${timeoutMs} ms`), { code: 'ETIMEDOUT' })));
+ req.on('error', (err) => finish(reject, err));
+ req.end(payload);
+ });
+ };
+}
+
+module.exports = { checkWebhookUrl, createWebhookPoster, WebhookRefused };
@@ -10,13 +10,15 @@ const crypto = require('crypto');
@@ -10,13 +10,15 @@ const crypto = require('crypto');
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
const store = require('./store');
const store = require('./store');
const catalog = require('../agents/catalog');
const catalog = require('../agents/catalog');
+const { checkWebhookUrl } = require('../net/webhook-post');
+const { mintSecret } = require('./webhooks');
function untilMidnight(nowMs) {
function untilMidnight(nowMs) {
const n = new Date(nowMs);
const n = new Date(nowMs);
return Math.ceil((Date.UTC(n.getUTCFullYear(), n.getUTCMonth(), n.getUTCDate() + 1) - n.getTime()) / 1000);
return Math.ceil((Date.UTC(n.getUTCFullYear(), n.getUTCMonth(), n.getUTCDate() + 1) - n.getTime()) / 1000);
}
}
-async function createTask({ s, config, engine, principal, body }) {
+async function createTask({ s, config, engine, principal, body, webhooks: webhookQueue = null }) {
const v = contracts.validate('actor.task-create-request@1', body);
const v = contracts.validate('actor.task-create-request@1', body);
if (!v.valid) return { status: 422, code: 'actor.task.invalid', detail: (v.errors || []).map((e) => `${e.path || e.instancePath || ''} ${e.message}`.trim()).join('; ').slice(0, 500) || 'invalid request' };
if (!v.valid) return { status: 422, code: 'actor.task.invalid', detail: (v.errors || []).map((e) => `${e.path || e.instancePath || ''} ${e.message}`.trim()).join('; ').slice(0, 500) || 'invalid request' };
const tier = config.allowance;
const tier = config.allowance;
@@ -26,8 +28,14 @@ async function createTask({ s, config, engine, principal, body }) {
@@ -26,8 +28,14 @@ async function createTask({ s, config, engine, principal, body }) {
return { status: 422, code: 'actor.budget.over_tier', detail: `The free tier allows up to $${tier.perTaskUsd} per task and $${tier.perDayUsd} a day; paid tiers come with OpenVibe.Billing.` };
return { status: 422, code: 'actor.budget.over_tier', detail: `The free tier allows up to $${tier.perTaskUsd} per task and $${tier.perDayUsd} a day; paid tiers come with OpenVibe.Billing.` };
}
}
if (body.agent && !catalog.byId(body.agent)) return { status: 422, code: 'actor.agent.unknown', detail: `no agent "${body.agent}"; GET /api/v1/agents lists them` };
if (body.agent && !catalog.byId(body.agent)) return { status: 422, code: 'actor.agent.unknown', detail: `no agent "${body.agent}"; GET /api/v1/agents lists them` };
+ // Webhooks (plan T17): each URL must be one Actor may deliver to (https, port 443/8443, a public address); the
+ // send-time check repeats it against what the name resolves to then.
+ const webhooks = (body.webhooks || []).map((h) => ({ url: h.url, events: h.events }));
+ for (const h of webhooks) {
+ try { checkWebhookUrl(h.url); } catch (err) { return { status: 422, code: 'actor.webhook.refused', detail: `${h.url.slice(0, 200)}: ${err.message}` }; }
+ }
- const hash = crypto.createHash('sha256').update(JSON.stringify({ task: body.task, mode: body.mode || 'balanced', budget: [perTask, perDay], agent: body.agent || null })).digest('hex');
+ const hash = crypto.createHash('sha256').update(JSON.stringify({ task: body.task, mode: body.mode || 'balanced', budget: [perTask, perDay], agent: body.agent || null, ...(webhooks.length ? { webhooks } : {}) })).digest('hex');
if (body.idempotency_key) {
if (body.idempotency_key) {
const prior = await store.byIdempotency(s, principal.requester, body.idempotency_key);
const prior = await store.byIdempotency(s, principal.requester, body.idempotency_key);
if (prior) {
if (prior) {
@@ -50,11 +58,15 @@ async function createTask({ s, config, engine, principal, body }) {
@@ -50,11 +58,15 @@ async function createTask({ s, config, engine, principal, body }) {
const row = await store.insertTask(s, {
const row = await store.insertTask(s, {
id: s.newId('tsk'), requester: principal.requester, via: principal.via || null, project_id: principal.project || null, task: body.task, mode: body.mode || 'balanced', agent: body.agent || null,
id: s.newId('tsk'), requester: principal.requester, via: principal.via || null, project_id: principal.project || null, task: body.task, mode: body.mode || 'balanced', agent: body.agent || null,
budget_task: perTask, budget_day: perDay, idem_key: body.idempotency_key || null, idem_hash: body.idempotency_key ? hash : null, created_at: s.iso(),
budget_task: perTask, budget_day: perDay, idem_key: body.idempotency_key || null, idem_hash: body.idempotency_key ? hash : null, created_at: s.iso(),
+ webhooks, webhook_secret: webhooks.length ? mintSecret() : null,
});
});
return { status: 201, task: row };
return { status: 201, task: row };
});
});
- // The engine starts the task only once its row has committed.
- if (out.status === 201) engine.submit(out.task.id);
+ // The engine starts the task only once its row has committed; a webhook that asked for `queued` hears it now.
+ if (out.status === 201) {
+ if (webhooks.length && webhookQueue) await webhookQueue.enqueue(out.task.id, 'queued');
+ engine.submit(out.task.id);
+ }
return out;
return out;
}
}
@@ -22,7 +22,7 @@ const { check } = require('../agents/check');
@@ -22,7 +22,7 @@ const { check } = require('../agents/check');
const MAX_ATTEMPTS = 2;
const MAX_ATTEMPTS = 2;
-function createEngine({ config, s, adapters, stream, now = () => Date.now(), log = console }) {
+function createEngine({ config, s, adapters, stream, webhooks = null, now = () => Date.now(), log = console }) {
const running = new Map(); // id → AbortController
const running = new Map(); // id → AbortController
const waiting = [];
const waiting = [];
let stopped = false;
let stopped = false;
@@ -39,6 +39,13 @@ function createEngine({ config, s, adapters, stream, now = () => Date.now(), log
@@ -39,6 +39,13 @@ function createEngine({ config, s, adapters, stream, now = () => Date.now(), log
const ended = ['succeeded', 'failed', 'cancelled'].includes(state);
const ended = ['succeeded', 'failed', 'cancelled'].includes(state);
await emit(id, { kind: ended ? 'end' : 'state', state, ...(agentForEvent ? { agent: agentForEvent } : {}) });
await emit(id, { kind: ended ? 'end' : 'state', state, ...(agentForEvent ? { agent: agentForEvent } : {}) });
if (ended) stream.end(id);
if (ended) stream.end(id);
+ // Webhooks that asked for this state hear it; a delivery never holds the task up or fails it.
+ if (webhooks && row && row.webhooks) {
+ try {
+ await webhooks.enqueue(id, state);
+ if (ended) await webhooks.forgetSecretIfDone(id);
+ } catch (err) { log.warn('[Actor] webhook enqueue failed:', id, err && err.message); }
+ }
return row;
return row;
}
}
@@ -22,9 +22,10 @@ function subjectRef(requester) {
@@ -22,9 +22,10 @@ function subjectRef(requester) {
async function insertTask(s, t) {
async function insertTask(s, t) {
await s.db.query(
await s.db.query(
- `INSERT INTO tasks (id, requester, project_id, task, mode, agent, budget_task, budget_day, state, idem_key, idem_hash, created_at, via)
- VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'queued', $9, $10, $11, $12)`,
- [t.id, t.requester, t.project_id || null, t.task, t.mode, t.agent || null, t.budget_task, t.budget_day, t.idem_key || null, t.idem_hash || null, t.created_at, t.via || null]);
+ `INSERT INTO tasks (id, requester, project_id, task, mode, agent, budget_task, budget_day, state, idem_key, idem_hash, created_at, via, webhooks, webhook_secret)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'queued', $9, $10, $11, $12, $13::jsonb, $14)`,
+ [t.id, t.requester, t.project_id || null, t.task, t.mode, t.agent || null, t.budget_task, t.budget_day, t.idem_key || null, t.idem_hash || null, t.created_at, t.via || null,
+ t.webhooks && t.webhooks.length ? JSON.stringify(t.webhooks) : null, t.webhook_secret || null]);
await addSpend(s, t.requester, 0, { tasks: 1 });
await addSpend(s, t.requester, 0, { tasks: 1 });
return getTask(s, t.id);
return getTask(s, t.id);
}
}
@@ -144,7 +145,10 @@ function toWire(row, { baseUrl }) {
@@ -144,7 +145,10 @@ function toWire(row, { baseUrl }) {
};
};
if (row.project_id) t.project_id = row.project_id;
if (row.project_id) t.project_id = row.project_id;
if (t.explanation && !t.explanation.length) t.explanation = null;
if (t.explanation && !t.explanation.length) t.explanation = null;
+ // What the task registered; the signing secret is never part of the task as read (only the creating answer has it).
+ const hooks = parse(row.webhooks);
+ if (hooks && hooks.length) t.webhooks = hooks.map((h) => ({ url: h.url, events: h.events }));
return t;
return t;
}
}
-module.exports = { insertTask, getTask, byIdempotency, listTasks, updateTask, appendEvent, eventsAfter, addSpend, spendToday, lockSpendToday, failInterrupted, prune, toWire, subjectRef, dayOf, OPEN, RETENTION_DAYS };
+module.exports = { insertTask, getTask, byIdempotency, listTasks, updateTask, appendEvent, eventsAfter, addSpend, spendToday, lockSpendToday, failInterrupted, prune, toWire, subjectRef, dayOf, parseJson: parse, OPEN, RETENTION_DAYS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,129 @@
@@ -0,0 +1,129 @@
+'use strict';
+/**
+ * Task webhooks (plan T17, actor.task-webhook@1): a task registers up to 8 https endpoints and the state changes each
+ * wants. When the task reaches one, a delivery row is queued with the body as it is then; this worker POSTs it,
+ * signed with the task's own secret exactly like an Events delivery (X-OpenVibe-Timestamp, X-OpenVibe-Signature-V2:
+ * HMAC-SHA256 of "<t>.<raw body>"), so a receiver checks it with openvibe-sdk/events verifyDeliveryV2.
+ *
+ * 2xx delivered
+ * 410, or a refused URL failed at once (the receiver is gone, or the URL is not a public address any more)
+ * anything else retried: 10 s, 1 min, 5 min, 30 min, 2 h, 6 h, 12 h, then failed (about a day)
+ *
+ * A delivery never delays or changes the task. Once a task has ended and none of its deliveries is pending, its secret
+ * is erased. The delivery id is stable across attempts (dedupe on it). One Actor process sends: rows are claimed with
+ * FOR UPDATE SKIP LOCKED all the same, and a row a crash left 'sending' is pending again at the next start.
+ */
+const crypto = require('crypto');
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+const store = require('./store');
+const { WebhookRefused } = require('../net/webhook-post');
+
+const BACKOFF_MS = [10_000, 60_000, 300_000, 1_800_000, 7_200_000, 21_600_000, 43_200_000];
+const BATCH = 20;
+
+/** A fresh per-task secret: whsec_ + 32 random bytes, base64url (43 characters). */
+const mintSecret = () => `whsec_${crypto.randomBytes(32).toString('base64url')}`;
+
+function createWebhooks({ s, config, post, now = () => Date.now(), log = console }) {
+ let timer = null;
+ let running = null;
+
+ /** Queue the deliveries a task's state change asks for (none when it registered no webhook for that state). */
+ async function enqueue(taskId, state) {
+ const row = await store.getTask(s, taskId);
+ const hooks = store.parseJson(row && row.webhooks) || [];
+ const wanted = hooks.filter((h) => Array.isArray(h.events) && h.events.includes(state));
+ if (!wanted.length) return 0;
+ const task = store.toWire(row, { baseUrl: config.baseUrl });
+ for (const h of wanted) {
+ const id = s.newId('whd');
+ const body = { type: 'actor.task.state', delivery_id: id, state, task };
+ await s.db.query(`INSERT INTO webhook_deliveries (id, task_id, url, state, body, status, attempts, next_at, created_at)
+ VALUES ($1, $2, $3, $4, $5::jsonb, 'pending', 0, $6, $7)`, [id, taskId, h.url, state, JSON.stringify(body), now(), s.iso()]);
+ }
+ return wanted.length;
+ }
+
+ async function claim() {
+ return await s.tx(async () => await s.db.many(`UPDATE webhook_deliveries SET status = 'sending'
+ WHERE id IN (SELECT id FROM webhook_deliveries WHERE status = 'pending' AND next_at <= $1 ORDER BY next_at, id LIMIT $2 FOR UPDATE SKIP LOCKED)
+ RETURNING *`, [now(), BATCH]));
+ }
+
+ async function finish(d, status, { lastStatus = null, error = null, attempts }) {
+ await s.db.query(`UPDATE webhook_deliveries SET status = $2, attempts = $3, last_status = $4, last_error = $5, finished_at = $6 WHERE id = $1`,
+ [d.id, status, attempts, lastStatus, error, s.iso()]);
+ await forgetSecretIfDone(d.task_id);
+ }
+
+ /** The task has ended and nothing is left to send: its secret goes. */
+ async function forgetSecretIfDone(taskId) {
+ await s.db.query(`UPDATE tasks SET webhook_secret = NULL WHERE id = $1 AND webhook_secret IS NOT NULL AND state NOT IN ('queued', 'running', 'verifying')
+ AND NOT EXISTS (SELECT 1 FROM webhook_deliveries WHERE task_id = $1 AND status IN ('pending', 'sending'))`, [taskId]);
+ }
+
+ async function sendOne(d) {
+ const attempts = Number(d.attempts) + 1;
+ const secret = await s.db.value('SELECT webhook_secret FROM tasks WHERE id = $1', [d.task_id]);
+ if (!secret) return finish(d, 'failed', { error: 'the task has no signing secret any more', attempts });
+ const body = JSON.stringify(store.parseJson(d.body));
+ const headers = {
+ ...signDeliveryHeaders(body, secret, { now: now() }),
+ 'X-OpenVibe-Delivery-Id': d.id,
+ 'X-OpenVibe-Task-Id': d.task_id,
+ 'X-OpenVibe-Task-State': d.state,
+ 'X-OpenVibe-Delivery-Attempt': String(attempts),
+ };
+ let res = null;
+ let error = null;
+ try { res = await post(d.url, { headers, body }); } catch (err) { error = err; }
+ if (res && res.status >= 200 && res.status < 300) return finish(d, 'delivered', { lastStatus: res.status, attempts });
+ if ((res && res.status === 410) || error instanceof WebhookRefused) {
+ return finish(d, 'failed', { lastStatus: res ? res.status : null, error: error ? error.message : 'the receiver answered 410 Gone', attempts });
+ }
+ const why = error ? String(error.message || error).slice(0, 300) : `HTTP ${res.status}`;
+ if (attempts > BACKOFF_MS.length) return finish(d, 'failed', { lastStatus: res ? res.status : null, error: why, attempts });
+ await s.db.query(`UPDATE webhook_deliveries SET status = 'pending', attempts = $2, next_at = $3, last_status = $4, last_error = $5 WHERE id = $1`,
+ [d.id, attempts, now() + BACKOFF_MS[attempts - 1], res ? res.status : null, why]);
+ return undefined;
+ }
+
+ /** One pass over what is due. One pass at a time. */
+ function tick() {
+ if (running) return running;
+ running = (async () => {
+ let sent = 0;
+ for (const d of await claim()) {
+ try { await sendOne(d); sent++; } catch (err) {
+ log.warn('[Actor] webhook delivery crashed:', d.id, err && err.message);
+ await s.db.query(`UPDATE webhook_deliveries SET status = 'pending', next_at = $2 WHERE id = $1 AND status = 'sending'`, [d.id, now() + BACKOFF_MS[0]]).catch(() => {});
+ }
+ }
+ return sent;
+ })().finally(() => { running = null; });
+ return running;
+ }
+
+ /** At start: a row a stopped process left 'sending' is pending again (the receiver dedupes on the delivery id). */
+ async function recover() {
+ return await s.db.exec(`UPDATE webhook_deliveries SET status = 'pending' WHERE status = 'sending'`);
+ }
+
+ function start(intervalMs = config.webhooks.intervalMs) {
+ if (timer) return;
+ timer = setInterval(() => { tick().catch((err) => log.warn('[Actor] webhook pass failed:', err && err.message)); }, intervalMs);
+ timer.unref();
+ }
+ async function stop() {
+ if (timer) clearInterval(timer);
+ timer = null;
+ if (running) await running.catch(() => {});
+ }
+
+ /** A task's deliveries, newest first (the task page shows them; the receiver URL is the requester's own). */
+ const forTask = (taskId) => s.db.many('SELECT id, url, state, status, attempts, last_status, last_error, created_at, finished_at FROM webhook_deliveries WHERE task_id = $1 ORDER BY created_at DESC, id DESC', [taskId]);
+
+ return { enqueue, tick, recover, start, stop, forTask, forgetSecretIfDone, BACKOFF_MS };
+}
+
+module.exports = { createWebhooks, mintSecret, BACKOFF_MS };
@@ -42,7 +42,7 @@ async function boot(opts = {}) {
@@ -42,7 +42,7 @@ async function boot(opts = {}) {
const { createStore } = require('../../server/db');
const { createStore } = require('../../server/db');
const testdb = await require('./db').testDb();
const testdb = await require('./db').testDb();
const store = createStore(testdb.db, { now: opts.now });
const store = createStore(testdb.db, { now: opts.now });
- const built = await createApp({ config, store, log, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, adapters: opts.adapters, accountSend: opts.accountSend });
+ const built = await createApp({ config, store, log, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, adapters: opts.adapters, accountSend: opts.accountSend, webhookPost: opts.webhookPost });
await built.ctx.keys.ensure();
await built.ctx.keys.ensure();
const server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });
const server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });
const base = `http://127.0.0.1:${server.address().port}`;
const base = `http://127.0.0.1:${server.address().port}`;
@@ -85,6 +85,7 @@ async function boot(opts = {}) {
@@ -85,6 +85,7 @@ async function boot(opts = {}) {
logs: () => captured.join('\n'),
logs: () => captured.join('\n'),
async close() {
async close() {
await built.ctx.engine.stop();
await built.ctx.engine.stop();
+ await built.ctx.webhooks.stop();
await new Promise((r) => server.close(r));
await new Promise((r) => server.close(r));
built.ctx.keys.client.stop();
built.ctx.keys.client.stop();
await testdb.close();
await testdb.close();
@@ -108,6 +108,7 @@ const SAME = { 'sec-fetch-site': 'same-origin' };
@@ -108,6 +108,7 @@ const SAME = { 'sec-fetch-site': 'same-origin' };
'server/agents/providers.js': 'the configured provider base URLs (DeepSeek, OpenAI, the open model) with the operator\'s keys',
'server/agents/providers.js': 'the configured provider base URLs (DeepSeek, OpenAI, the open model) with the operator\'s keys',
'server/agents/tools.js': 'the configured OpenVibe.Tools and OpenVibe.Search bases; ids come from Tools\' catalog, a page URL goes to Tools as input',
'server/agents/tools.js': 'the configured OpenVibe.Tools and OpenVibe.Search bases; ids come from Tools\' catalog, a page URL goes to Tools as input',
'server/events-consumer.js': 'OpenVibe.Events: the configured events.url (ACTOR_EVENTS_URL), to create the two ADR-033 subscriptions at boot',
'server/events-consumer.js': 'OpenVibe.Events: the configured events.url (ACTOR_EVENTS_URL), to create the two ADR-033 subscriptions at boot',
+ 'server/net/webhook-post.js': 'a task\'s webhook URL (plan T17), the one caller-chosen address: https on 443/8443 to public addresses only, checked at creation and at send time through openvibe-shared/egress safeLookup, never a redirect (test/webhooks.test.js)',
};
};
const root = path.join(__dirname, '..');
const root = path.join(__dirname, '..');
const found = [];
const found = [];
@@ -178,13 +178,14 @@ const SAME = { 'sec-fetch-site': 'same-origin' };
@@ -178,13 +178,14 @@ const SAME = { 'sec-fetch-site': 'same-origin' };
assert.strictEqual(t.providers.requests.filter((q) => q.path === '/openai/responses').length, 0);
assert.strictEqual(t.providers.requests.filter((q) => q.path === '/openai/responses').length, 0);
});
});
- await check('a budget above the free tier is refused, never lowered; an unknown field or a webhook is refused', async () => {
+ await check('a budget above the free tier is refused, never lowered; an unknown field or a webhook secret by reference is refused', async () => {
const over = await create({ task: 'x', budget: { per_task_usd: 5 } });
const over = await create({ task: 'x', budget: { per_task_usd: 5 } });
assert.strictEqual(over.status, 422);
assert.strictEqual(over.status, 422);
assert.strictEqual(over.json().code, 'actor.budget.over_tier');
assert.strictEqual(over.json().code, 'actor.budget.over_tier');
- const hooks = await create({ task: 'x', webhooks: [{ url: 'https://example.com', events: ['succeeded'] }] });
+ const hooks = await create({ task: 'x', webhooks: [{ url: 'https://example.com', events: ['succeeded'], secret_ref: 'HOOK_KEY' }] });
assert.strictEqual(hooks.status, 422);
assert.strictEqual(hooks.status, 422);
assert.strictEqual(hooks.json().code, 'actor.task.invalid');
assert.strictEqual(hooks.json().code, 'actor.task.invalid');
+ assert.strictEqual((await create({ task: 'x', surprise: true })).json().code, 'actor.task.invalid');
assert.strictEqual((await create({ task: ' ' })).status, 422);
assert.strictEqual((await create({ task: ' ' })).status, 422);
});
});
new file mode 100644
new file mode 100644
@@ -0,0 +1,200 @@
@@ -0,0 +1,200 @@
+'use strict';
+/**
+ * Task webhooks (plan T17, actor.task-webhook@1): a task registers https endpoints and the states each wants; the
+ * creating answer carries the task's signing secret once (reads and lists never do); each delivery is signed like an
+ * Events delivery (openvibe-sdk/events verifyDeliveryV2 accepts it), carries the task as it was and a delivery id that
+ * stays the same across retries; 2xx is delivered, 410 or a refused URL ends it, anything else is retried on the
+ * backoff and then fails; the secret is erased once the task has ended and its deliveries are done, and it is never
+ * exported. A URL that is not a public https address on 443/8443 is refused at creation, and the real poster refuses
+ * a name that resolves to a private address at send time.
+ */
+const assert = require('assert');
+const contracts = require('openvibe-contracts');
+const { verifyDeliveryV2 } = require('openvibe-sdk/events');
+const { createSafeLookup } = require('openvibe-shared/egress');
+const { boot, check, done } = require('./helpers/boot');
+const { checkWebhookUrl, createWebhookPoster, WebhookRefused } = require('../server/net/webhook-post');
+
+const SAME = { 'sec-fetch-site': 'same-origin' };
+const HOOK = 'https://hooks.example.com/openvibe/actor';
+
+(async () => {
+ // Actor's clock (the store's, which sign-in checks token expiry against too): real time, moved on by the retry checks.
+ let clock = Date.now();
+ const sent = [];
+ let answers = []; // what the stand-in receiver answers next: a status, or an Error to throw
+ const webhookPost = async (url, { headers, body }) => {
+ checkWebhookUrl(url); // the same check the real poster makes first
+ sent.push({ url, headers, body });
+ const a = answers.length ? answers.shift() : 200;
+ if (a instanceof Error) throw a;
+ return { status: a };
+ };
+ const t = await boot({ webhookPost, now: () => clock });
+ const kim = t.network.addUser('kim');
+ const create = (json) => t.get('/api/v1/tasks', { as: kim, json, headers: SAME });
+ const tick = () => t.ctx.webhooks.tick();
+ const deliveries = (taskId) => t.ctx.s.db.many('SELECT * FROM webhook_deliveries WHERE task_id = $1 ORDER BY created_at, id', [taskId]);
+ const secretOf = (taskId) => t.ctx.s.db.value('SELECT webhook_secret FROM tasks WHERE id = $1', [taskId]);
+ // The engine queues a delivery just after it records the state: wait for the rows before sending.
+ async function queued(taskId, n) {
+ for (let i = 0; i < 200; i++) { if ((await deliveries(taskId)).length >= n) return; await new Promise((res) => setTimeout(res, 20)); }
+ throw new Error(`${taskId}: ${n} deliveries never queued`);
+ }
+
+ let first;
+ await check('the creating answer carries the signing secret once; reads and lists show the webhooks but never the secret', async () => {
+ t.providers.reset();
+ const r = await create({ task: 'Who handles mail for openvibe.network?', webhooks: [{ url: HOOK, events: ['queued', 'succeeded', 'failed'] }] });
+ assert.strictEqual(r.status, 201, r.text);
+ first = r.json();
+ assert.match(first.webhook_secret, /^whsec_[A-Za-z0-9_-]{43}$/);
+ assert.ok(contracts.validate('platform.task@1', first).valid, JSON.stringify(contracts.validate('platform.task@1', first).errors));
+ assert.deepStrictEqual(first.webhooks, [{ url: HOOK, events: ['queued', 'succeeded', 'failed'] }]);
+ const read = (await t.get(`/api/v1/tasks/${first.id}`, { as: kim })).json();
+ assert.strictEqual('webhook_secret' in read, false, 'a read never carries the secret');
+ assert.deepStrictEqual(read.webhooks, first.webhooks);
+ const list = (await t.get('/api/v1/tasks', { as: kim })).json();
+ assert.ok(!JSON.stringify(list).includes(first.webhook_secret), 'a list never carries the secret');
+ });
+
+ await check('each delivery is signed like an Events delivery, carries the task as it was, and validates', async () => {
+ await t.waitFor(first.id);
+ await queued(first.id, 2);
+ await tick();
+ assert.deepStrictEqual(sent.map((d) => d.url), [HOOK, HOOK]);
+ for (const d of sent) {
+ assert.ok(verifyDeliveryV2(d.body, d.headers, first.webhook_secret, { now: clock }), 'verifyDeliveryV2 accepts it with the task\'s secret');
+ assert.ok(!verifyDeliveryV2(d.body, d.headers, 'whsec_someone-elses-secret', { now: clock }), 'and nothing else');
+ const body = JSON.parse(d.body);
+ const v = contracts.validate('actor.task-webhook@1', body);
+ assert.ok(v.valid, JSON.stringify(v.errors));
+ assert.strictEqual(body.task.id, first.id);
+ assert.strictEqual(d.headers['X-OpenVibe-Delivery-Id'], body.delivery_id);
+ assert.strictEqual(d.headers['X-OpenVibe-Task-Id'], first.id);
+ assert.strictEqual(d.headers['X-OpenVibe-Task-State'], body.state);
+ assert.strictEqual(d.headers['X-OpenVibe-Delivery-Attempt'], '1');
+ assert.ok(!d.body.includes(first.webhook_secret), 'the body never carries the secret');
+ }
+ assert.deepStrictEqual(sent.map((d) => JSON.parse(d.body).state), ['queued', 'succeeded']);
+ assert.strictEqual(JSON.parse(sent[0].body).task.state, 'queued', 'the queued delivery has the task as it was then');
+ assert.strictEqual(JSON.parse(sent[1].body).task.result.agent, 'openvibe-runtime');
+ assert.deepStrictEqual((await deliveries(first.id)).map((d) => d.status), ['delivered', 'delivered']);
+ assert.strictEqual(await secretOf(first.id), null, 'the task ended and its deliveries are done: the secret is gone');
+ const pageHtml = (await t.get(`/tasks/${first.id}`, { as: kim })).text;
+ assert.match(pageHtml, /<summary>Webhooks \(2\)<\/summary>/, 'the task page lists its deliveries');
+ assert.ok(pageHtml.includes('Delivered') && pageHtml.includes(HOOK) && !pageHtml.includes('whsec_'), 'with how each went, and never the secret');
+ });
+
+ await check('an idempotent repeat answers the same secret while the task still has one', async () => {
+ t.providers.reset();
+ const body = { task: 'Say hi', idempotency_key: 'hooks-idem-1', webhooks: [{ url: HOOK, events: ['failed'] }] };
+ const a = await create(body);
+ const b = await create(body);
+ assert.deepStrictEqual([a.status, b.status], [201, 200]);
+ assert.strictEqual(b.json().webhook_secret, a.json().webhook_secret);
+ assert.strictEqual((await create({ ...body, webhooks: [{ url: `${HOOK}/other`, events: ['failed'] }] })).status, 409, 'other webhooks are another body');
+ await t.waitFor(a.json().id);
+ for (let i = 0; i < 200 && await secretOf(a.json().id); i++) await new Promise((res) => setTimeout(res, 20));
+ assert.strictEqual(await secretOf(a.json().id), null, 'no delivery for its end state: the secret goes when it ends');
+ assert.deepStrictEqual(await deliveries(a.json().id), []);
+ });
+
+ await check('a failing receiver is retried on the backoff with the same delivery id; 2xx then delivers it', async () => {
+ t.providers.reset();
+ sent.length = 0;
+ const r = (await create({ task: 'Who handles mail for openvibe.network?', webhooks: [{ url: HOOK, events: ['succeeded'] }] })).json();
+ await t.waitFor(r.id);
+ await queued(r.id, 1);
+ answers = [500, Object.assign(new Error('connect ECONNREFUSED'), { code: 'ECONNREFUSED' })];
+ await tick();
+ let [d] = await deliveries(r.id);
+ assert.deepStrictEqual([d.status, d.attempts, d.last_status, Number(d.next_at)], ['pending', 1, 500, clock + 10_000]);
+ await tick();
+ assert.strictEqual(sent.length, 1, 'not due yet: nothing sent');
+ clock += 10_000;
+ await tick();
+ [d] = await deliveries(r.id);
+ assert.deepStrictEqual([d.status, d.attempts, Number(d.next_at)], ['pending', 2, clock + 60_000]);
+ assert.match(d.last_error, /ECONNREFUSED/);
+ clock += 60_000;
+ await tick();
+ [d] = await deliveries(r.id);
+ assert.strictEqual(d.status, 'delivered');
+ assert.deepStrictEqual(sent.map((x) => x.headers['X-OpenVibe-Delivery-Id']), [d.id, d.id, d.id], 'one delivery id across the attempts');
+ assert.deepStrictEqual(sent.map((x) => x.headers['X-OpenVibe-Delivery-Attempt']), ['1', '2', '3']);
+ assert.ok(verifyDeliveryV2(sent[2].body, sent[2].headers, r.webhook_secret, { now: clock }), 'each attempt is signed afresh (its own timestamp)');
+ assert.strictEqual(await secretOf(r.id), null);
+ });
+
+ await check('410 Gone or a URL refused at send time ends the delivery at once; endless failures end after the backoff', async () => {
+ t.providers.reset();
+ const r = (await create({ task: 'Who handles mail for openvibe.network?', webhooks: [{ url: `${HOOK}/a`, events: ['succeeded'] }, { url: `${HOOK}/b`, events: ['succeeded'] }, { url: `${HOOK}/c`, events: ['succeeded'] }] })).json();
+ await t.waitFor(r.id);
+ await queued(r.id, 3);
+ answers = [410, new WebhookRefused('hooks.example.com is not a public address')];
+ for (let i = 0; i < 20; i++) answers.push(503);
+ await tick();
+ const byUrl = async () => Object.fromEntries((await deliveries(r.id)).map((d) => [d.url.slice(-1), d]));
+ let d = await byUrl();
+ assert.deepStrictEqual([d.a.status, d.a.last_status], ['failed', 410]);
+ assert.strictEqual(d.b.status, 'failed');
+ assert.match(d.b.last_error, /not a public address/);
+ assert.strictEqual(d.c.status, 'pending');
+ for (const wait of t.ctx.webhooks.BACKOFF_MS) { clock += wait; await tick(); }
+ d = await byUrl();
+ assert.deepStrictEqual([d.c.status, d.c.attempts, d.c.last_status], ['failed', t.ctx.webhooks.BACKOFF_MS.length + 1, 503]);
+ assert.strictEqual(await secretOf(r.id), null, 'nothing left to send: the secret is gone');
+ clock = Date.now(); // back to real time: the next checks sign in with tokens that expire in real time
+ answers = [];
+ });
+
+ await check('cancel is delivered to a webhook that asked for it', async () => {
+ t.providers.reset();
+ sent.length = 0;
+ t.providers.state.cls = 'lookup';
+ t.providers.state.runtime = 'hang';
+ const r = (await create({ task: 'MX of example.com please', webhooks: [{ url: HOOK, events: ['cancelled'] }] })).json();
+ await t.waitFor(r.id, ['running']);
+ await new Promise((res) => setTimeout(res, 150));
+ await t.get(`/api/v1/tasks/${r.id}/cancel`, { as: kim, method: 'POST', headers: SAME });
+ await t.waitFor(r.id, ['cancelled']);
+ await queued(r.id, 1);
+ await tick();
+ assert.deepStrictEqual(sent.map((x) => JSON.parse(x.body).state), ['cancelled']);
+ assert.strictEqual(JSON.parse(sent[0].body).task.cancel.requested_by.id, kim.subject);
+ assert.deepStrictEqual((await deliveries(r.id)).map((d) => d.status), ['delivered']);
+ assert.strictEqual(await secretOf(r.id), null);
+ });
+
+ await check('a URL Actor may not deliver to is refused at creation; a project secret by reference is not accepted yet', async () => {
+ for (const url of ['https://127.0.0.1/x', 'https://localhost/x', 'https://10.0.0.7/x', 'https://[::1]/x', 'https://169.254.169.254/latest', 'https://hooks.example.com:8080/x', 'https://user:[email protected]/x', 'https://intranet/x']) {
+ const r = await create({ task: 'x', webhooks: [{ url, events: ['succeeded'] }] });
+ assert.strictEqual(r.status, 422, `${url}: ${r.status} ${r.text.slice(0, 200)}`);
+ assert.strictEqual(r.json().code, 'actor.webhook.refused', url);
+ }
+ const http = await create({ task: 'x', webhooks: [{ url: 'http://hooks.example.com/x', events: ['succeeded'] }] });
+ assert.deepStrictEqual([http.status, http.json().code], [422, 'actor.task.invalid'], 'plain http is not a webhook URL');
+ const ref = await create({ task: 'x', webhooks: [{ url: HOOK, events: ['succeeded'], secret_ref: 'HOOK_KEY' }] });
+ assert.deepStrictEqual([ref.status, ref.json().code], [422, 'actor.task.invalid']);
+ assert.ok(checkWebhookUrl('https://hooks.example.com:8443/x'), 'port 8443 is allowed');
+ });
+
+ await check('the real poster refuses a name that resolves to a private address, at send time', async () => {
+ const lookup = createSafeLookup({ lookup: (host, opts, cb) => cb(null, [{ address: '10.1.2.3', family: 4 }]) });
+ const post = createWebhookPoster({ lookup, timeoutMs: 2000 });
+ await assert.rejects(post('https://hooks.example.com/x', { body: '{}' }), /10\.1\.2\.3|not public|no public|EGRESS|denied/i);
+ await assert.rejects(post('https://127.0.0.1/x', { body: '{}' }), WebhookRefused);
+ });
+
+ await check('the signing secret is never exported with the person\'s tasks', async () => {
+ const { TABLES } = require('../server/identity/account-data');
+ const tasks = TABLES.find((x) => x.table === 'tasks');
+ assert.ok(Array.isArray(tasks.columns) && tasks.columns.includes('webhooks') && !tasks.columns.includes('webhook_secret'));
+ const cols = (await t.ctx.s.db.many("SELECT column_name FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'tasks'")).map((r) => r.column_name);
+ assert.deepStrictEqual(cols.filter((c) => !tasks.columns.includes(c)), ['webhook_secret'], 'every other column of tasks is exported');
+ });
+
+ await t.close();
+ done();
+})().catch((e) => { console.error(e); process.exit(1); });