openvibers/openvibe.watch Commits
Watch: project-scoped watches; the run-job action starts a Run job for the owner in the watch's project (#53)
Changes
@@@ -228,7 +228,16 @@ A fired condition queues the watch's actions in the same transaction as the chec| `webhook` | a POST of the firing (`type`, `watch {id, name, url}`, `check_run_id`, `condition`, `observation {observed_at, value, previous_value}`) to the action's https URL, through the same SSRF guard as every fetch, never following a redirect ([server/net/post.js](server/net/post.js)). It is signed with the key `secret_env` names exactly as OpenVibe.Events signs a delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`; check it with openvibe-sdk/events `verifyDeliveryV2`). `X-OpenVibe-Delivery-Id` (`<check run>.<action index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 or a URL the guard refuses is `failed` at once. Anything else, or a lost connection, is retried on the backoff, then `failed`. The key must be granted to the owner (`WATCH_SECRET_GRANTS`, below) and set, 32 characters or more: otherwise `unavailable` || `actor-task` | a task for the owner on OpenVibe.Actor ([server/actor.js](server/actor.js)): the action's `brief`, followed by the firing as context (the watch, the value and what it was, its page), at most `budget_usd`. Watch names the owner in `X-OV-Subject`, so the task runs under **the owner's** allowance and appears in their task list, and Actor records Watch as `via`. The idempotency key `watch.<check run>.<action index>` makes a retried delivery the same task. Created is `sent`, with the task id recorded. A request Actor refuses as invalid (over the tier's budget, an unknown agent) is `failed` at once. Anything else, including a day's allowance spent, is retried on the backoff. An agent- or service-owned watch is `skipped`. Needs `actor.task.create` for `openvibe.actor` (Network grants it) and `WATCH_ACTOR_URL` || `actor-agent` | runs one of the person's OpenVibe.Actor agents by `agent_id`. Its optional `note` precedes the firing context (watch name and link, value and previous value, observation time); input is capped at 8000 characters without cutting the note. Watch uses `X-OV-Subject` for the owner. A created run is `sent` with its task id; a missing agent, limit or free-tier refusal is `failed` for that firing; service errors retry on the usual backoff. Only person-owned watches may use it. Needs the `actor.agent.run` grant for `openvibe.actor` and `WATCH_ACTOR_URL` |--| `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |++| `run-job` | one OpenVibe.Run job for the owner in the watch's project ([server/run-jobs.js](server/run-jobs.js)): the action's `run.job-create-request@1` fields (`class`, `artifact`, `args`, `limits`, `egress`, `ttl_ms`), with the firing (watch, check run, value, previous value, time) under `args[firing_arg]` when the action names one. Watch submits as itself with the owner in `X-OV-Subject` and the project in `X-OV-Project` (project delegation, ADR-003 amendment 2026-10-12): Run bills the project, checks the owner is still an admin or owner of it, makes the owner the requester and records Watch as `via`. Key `watch.<check run>.<action index>`, so a retry is the same job. Created is `sent` with the job id; Run's final refusals (400, 403, 404, 409, 413, 422) are `failed` at once; anything else retries on the backoff. `WATCH_RUN_URL` is Run's internal address (default `http://127.0.0.1:4920`). |++| `codes-run` | recorded `unavailable` (`watch.action_unavailable`): its dispatcher is a later step, and the watch page says so |++++A watch with a `run-job` action names a **project** (`project_id` on `watch.watch-request@1`, required for it:++`422 watch.project_required`). Only a person names one, and only a project OpenVibe.Network says they are an admin or++owner of and that is not archived (`GET /internal/projects/:project/members/:subject`, `network.project.member.read`),++checked when the project is named or changed: `403 project.not_delegated` otherwise, `503++watch.project_check_unavailable` when Network cannot answer. Run checks again before every job. The run-job action++needs `class` and `limits` (and `artifact` for `function` and `code`); the contract draft's `runtime`, `command` and++`workload` are refused.Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.@@@ -12,7 +12,7 @@"dotenv": "^16.4.7","express": "^4.21.2","fast-xml-parser": "^4.5.3",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.153.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3","pg": "^8.23.0"@@@ -769,9 +769,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.143.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",-- "integrity": "sha512-Cdbf+9RoPAUBibpM2zXQvtsiKMugb55dKubTt8pFZvAmxsLtGvX+zaC++yL1vOSub8FX3+GgqcGtoJS6tklrUg==",++ "version": "0.153.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.153.0",++ "integrity": "sha512-WQ7p7ATKtTky9nI9yE9Agkm3rNKuyWljmrMCdZaqxg+FjVaziTqrFz0a3YMBTXJmNf5JTGXn/ioOmWjO95/lCA==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -24,7 +24,7 @@"dotenv": "^16.4.7","express": "^4.21.2","fast-xml-parser": "^4.5.3",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.153.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3","pg": "^8.23.0"@@@ -37,9 +37,18 @@* owner: an agent- or service-owned watch is `skipped`.* actor-agent runs one of the owner's personal agents on Actor with a note and bounded firing context.* A missing agent or allowance refusal fails this firing; a service error retries.-- * codes-run, run-job-- * `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and-- * the watch page says so rather than pretending they ran.++ * run-job one OpenVibe.Run job for the owner in the watch's project (server/run-jobs.js): the action's++ * run.job-create-request@1 fields, with the firing under args[firing_arg] when it names one. Watch++ * submits as itself, the owner in X-OV-Subject and the project in X-OV-Project (project delegation,++ * ADR-003 amendment 2026-10-12), so Run bills the project, checks the owner is still an admin or++ * owner of it, makes the owner the requester and records Watch as via. The idempotency key++ * `watch.<check run>.<action index>` makes a retried delivery the same job. Created (or the same++ * job again) is `sent` with the job id. A refusal that cannot change on a retry (400, 403, 404, 409,++ * 413, 422: an invalid job, a project the owner may no longer use, over the project's caps) is++ * `failed` at once; anything else, Run or Network unreachable included, is retried on the backoff.++ * It needs a person as owner and a project: otherwise `skipped` / `unavailable`.++ * codes-run `unavailable` with `watch.action_unavailable`: its dispatcher is a later step of the plan, and the++ * watch page says so rather than pretending it ran.** The worker claims due rows with a short lease (FOR UPDATE SKIP LOCKED, next_attempt_at moved past the lease), so* two processes never deliver one row at the same time, and a process that dies mid-delivery leaves the row to the@@@ -50,11 +59,16 @@const { signDeliveryHeaders } = require('openvibe-sdk/events');const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const PROJECT_ID = /^prj_[0-9A-HJKMNP-TV-Z]{26}$/;const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000];const LEASE_MS = 2 * 60_000;const TITLE_MAX = 120;const MESSAGE_MAX = 300;--const LATER = new Set(['codes-run', 'run-job']);++const LATER = new Set(['codes-run']);++// Run's answers that a retry cannot change: the job is invalid, over the project's caps, or the owner may no longer++// act in the project (403 project.not_delegated / capability.denied).++const RUN_FINAL = new Set([400, 403, 404, 409, 413, 422]);++const JOB_FIELDS = ['class', 'artifact', 'args', 'limits', 'egress', 'ttl_ms'];const TASK_MAX = 20000;const TRIGGERED = 'watch.watch.triggered';const MIN_HOOK_SECRET = 32;@@@ -121,13 +135,36 @@ function actorContextFor({ config, watch, value, previousValue, observedAt }, mareturn [first, valueLine.length > room ? `${valueLine.slice(0, Math.max(0, room - 1))}…` : valueLine, last].join('\n').slice(0, maxLength);}++/**++ * The run.job-create-request@1 body a run-job action submits (without its idempotency key): the action's job fields,++ * and the firing under args[firing_arg] when the action names one.++ */++function runJobFor({ config, watch, run, action, value, previousValue, observedAt }) {++ const job = {};++ for (const k of JOB_FIELDS) if (action[k] !== undefined && action[k] !== null) job[k] = action[k];++ if (action.firing_arg) {++ job.args = {++ ...(job.args || {}),++ [action.firing_arg]: {++ watch_id: watch.id,++ watch_url: `${config.baseUrl}/watches/${encodeURIComponent(watch.id)}`,++ check_run_id: run.id,++ observed_at: observedAt ? new Date(observedAt).toISOString() : null,++ value: value === undefined ? null : value,++ previous_value: previousValue === undefined ? null : previousValue,++ },++ };++ }++ return job;++}++function actorAgentInputFor(args) {const note = typeof args.action.note === 'string' ? args.action.note : '';const prefix = note ? `${note}\n\n` : '';return prefix + actorContextFor(args, 8000 - prefix.length);}--function createActions({ db, config, notifications = null, post = null, actor = null, now = () => Date.now(), log = console }) {++function createActions({ db, config, notifications = null, post = null, actor = null, run: runJobs = null, now = () => Date.now(), log = console }) {// Read at each use, not once: what a process was started with is what it enforces, for rows queued before it too.const grants = () => config.secretGrants;const enabled = Boolean(notifications) && config.actions.enabled;@@@ -175,6 +212,10 @@ function createActions({ db, config, notifications = null, post = null, actor =if (USER_SUBJECT.test(String(owner || ''))) {request = { agent_id: action.agent_id, input: actorAgentInputFor({ config, watch, action, value, previousValue, observedAt }) };} else { state = 'skipped'; error = 'watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service'; }++ } else if (kind === 'run-job') {++ if (!USER_SUBJECT.test(String(owner || ''))) { state = 'skipped'; error = 'watch.action_no_person: a Run job is started for a person in their project, and this watch belongs to an agent or a service'; }++ else if (!PROJECT_ID.test(String(watch.project_id || ''))) { state = 'unavailable'; error = 'watch.project_required: a run-job action is billed to the watch\'s project, and this watch names none'; }++ else request = { project_id: watch.project_id, job: runJobFor({ config, watch, run, action, value, previousValue, observedAt }) };} else if (LATER.has(kind)) {state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;} else {@@@ -240,11 +281,26 @@ function createActions({ db, config, notifications = null, post = null, actor =}}++ async function deliverRunJob(row, request) {++ if (!runJobs) throw Object.assign(new Error('Run delivery is not configured in this process'), { code: 'run_unconfigured' });++ try {++ const job = await runJobs.submit({++ subjectId: row.owner_sub, projectId: request.project_id, job: request.job,++ idempotencyKey: `watch.${row.check_run_id}.${row.action_index}`,++ });++ return { state: 'sent', error: `Run job ${job.id || '(no id)'}${job.created ? '' : ' (already started)'}` };++ } catch (err) {++ if (err && RUN_FINAL.has(err.status)) return { state: 'failed', error: `watch.action_refused: ${err.code || err.status}: ${String(err.detail || err.message || '').slice(0, 200)}` };++ throw err;++ }++ }++async function deliver(row, attempt) {const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;if (row.kind === 'webhook') return await deliverWebhook(row, request, attempt);if (row.kind === 'actor-task') return await deliverActorTask(row, request);if (row.kind === 'actor-agent') return await deliverActorAgent(row, request);++ if (row.kind === 'run-job') return await deliverRunJob(row, request);if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };const out = await notifications.push({ subjectId: row.owner_sub, ...request });if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };@@@ -316,4 +372,4 @@ function createActions({ db, config, notifications = null, post = null, actor =return { queue, tick, start, stop, status, recent, enabled };}--module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, actorAgentInputFor, BACKOFF_MS };++module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, actorAgentInputFor, runJobFor, BACKOFF_MS, RUN_FINAL };@@@ -29,7 +29,7 @@ function watchesRouter({ registry, auth, limits, webhookSources = null }) {const fail = (res, req, err) => {if (err instanceof RegistryError) {-- const status = err.code === 'watch.internal' ? 500 : (err.code === 'watch.exists' ? 409 : 422);++ const status = err.status || (err.code === 'watch.internal' ? 500 : (err.code === 'watch.exists' ? 409 : 422));return http.sendProblem(res, status, err.code, { detail: err.message, ctx: req.ov });}throw err;@@@ -101,6 +101,8 @@ function load(env = process.env) {webhookTimeoutMs: int(env.WATCH_WEBHOOK_TIMEOUT_MS, 10000),// OpenVibe.Actor's internal address, for actor-task and actor-agent actions.actorUrl: (env.WATCH_ACTOR_URL || 'http://127.0.0.1:4950').replace(/\/+$/, ''),++ // OpenVibe.Run's internal address, for run-job actions (a job for the owner in the watch's project).++ runUrl: (env.WATCH_RUN_URL || 'http://127.0.0.1:4920').replace(/\/+$/, ''),},// Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins@@@ -16,6 +16,7 @@ const { createObservations } = require('./observations');const { createGuard } = require('./net/guard');const { createPoster } = require('./net/post');const { createActorTasks } = require('./actor');++const { createProjects, createRunJobs } = require('./run-jobs');const { createFetcher } = require('./net/fetcher');const { createSpacer } = require('./spacer');const { createCarriers } = require('./carriers');@@@ -40,7 +41,7 @@ const { createApp } = require('./app');const accountDataLib = require('./account-data');const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null, actor: givenActor = null } = {}) {++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null, actor: givenActor = null, run: givenRun = null, projects: givenProjects = null } = {}) {config = config || load();// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.const db = givenDb || await openDb(config, { log });@@@ -52,7 +53,19 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed'],validate: (env) => validate('events.event-envelope@1', env),});-- const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner, secretGrants: config.secretGrants });++ // Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own++ // client-credentials token; without the client secret they wait (actions.status() says delivery is off).++ // Actor tasks (step 6) and Run jobs (project delegation) use the same client and token provider (a token per audience).++ const client = config.oauth.clientSecret++ ? createClient({++ tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),++ baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl, run: config.actions.runUrl }, fetch: fetchImpl,++ })++ : null;++ // A watch that names a project: Network says whether its owner is an admin or owner of it, when it is saved (Run checks++ // again before every job). Without the client secret no project can be checked, so none can be named.++ const projects = givenProjects || (client ? createProjects(client, { networkUrl: config.networkInternalUrl }) : null);++ const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner, secretGrants: config.secretGrants, projects });const observations = createObservations({ db, now });// The only way out to the internet: SSRF guard (ports, private hosts, every redirect hop), byte// cap, one deadline, conditional GET; per-host spacing in front of every request.@@@ -61,20 +74,12 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchconst spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });// Every carrier goes out through the spacer, so two watches over one host never hammer it.const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });-- // Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own-- // client-credentials token; without the client secret they wait (actions.status() says delivery is off).-- // Actor tasks (step 6) use the same client and token provider (a token per audience).-- const client = config.oauth.clientSecret-- ? createClient({-- tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),-- baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl }, fetch: fetchImpl,-- })-- : null;const notifications = client ? createNotificationsClient(client) : null;const actorTasks = client ? createActorTasks(client, { baseUrl: config.actions.actorUrl }) : null;// Webhook actions leave through the same guard as every fetch (server/net/post.js).const post = givenPost || createPoster({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.actions.webhookTimeoutMs });-- const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, now, log });++ const runJobs = client ? createRunJobs(client, { baseUrl: config.actions.runUrl }) : null;++ const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, run: givenRun || runJobs, now, log });const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });const scheduler = createScheduler({ db, check, config, now, log });// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.@@@ -16,6 +16,13 @@* Ownership is the authority: every read and write is scoped to the acting subject (principal.sub,* or the person a first-party service names with X-OV-Subject), and another owner's watch is not* found — 404, never 403 (an id must not confirm that someone else's watch exists).++ *++ * A watch may name a project (project_id, ADR-003 amendment 2026-10-12): the project its run-job actions are billed++ * to. Only a person names one, and only a project Network says they are an admin or owner of (not archived), checked++ * when the project is first named or changed (403 project.not_delegated; 503 watch.project_check_unavailable when++ * Network cannot answer). OpenVibe.Run checks again before every job, so a later loss of the role stops the jobs. A++ * run-job action needs the watch to name a project (422 watch.project_required) and run.job-create-request@1's++ * class and limits (and artifact for function and code); the draft's runtime/command/workload are refused.*/const { validate, ids: contractIds } = require('openvibe-contracts');const safe = require('./safe-regex');@@@ -32,9 +39,13 @@ const MIN_EVERY_SEC = 900;const FORBIDDEN_HEADERS = new Set(['host', 'cookie', 'connection', 'content-length', 'transfer-encoding', 'user-agent', 'accept-encoding', 'if-none-match', 'if-modified-since']);const PERSON = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;const AGENT_ID = /^agt_[0-9A-HJKMNP-TV-Z]{26}$/;++/** The run-job fields of the contract's early draft, which never ran: refused, so nobody writes a job that cannot start. */++const RUN_JOB_DRAFT = ['runtime', 'command', 'workload'];++const ARTIFACT_CLASSES = new Set(['function', 'code']);class RegistryError extends Error {-- constructor(detail, code = 'watch.invalid') { super(detail); this.code = code; }++ /** status: the HTTP answer when it is not the default (422; 500 for watch.internal, 409 for watch.exists). */++ constructor(detail, code = 'watch.invalid', status = null) { super(detail); this.code = code; if (status) this.status = status; }}const parse = (v) => (v == null ? null : (typeof v === 'string' ? JSON.parse(v) : v));@@@ -71,6 +82,17 @@ function validateRequest(body, { create = false } = {}) {}const v = validate('watch.watch-request@1', body);if (!v.valid) throw new RegistryError(`body does not match watch.watch-request@1: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`);++ for (const [i, action] of (Array.isArray(body.action) ? body.action : []).entries()) {++ if (!action || action.kind !== 'run-job') continue;++ const draft = RUN_JOB_DRAFT.find((f) => action[f] !== undefined);++ if (draft) throw new RegistryError(`/action/${i}/${draft} is not accepted: a run-job action takes run.job-create-request@1 fields (class, artifact, args, limits, egress, ttl_ms)`);++ if (!action.class) throw new RegistryError(`/action/${i}/class is required (a platform.runtime-class@1 name)`);++ if (!action.limits) throw new RegistryError(`/action/${i}/limits is required (wall_ms, cpu_ms, mem_bytes)`);++ if (ARTIFACT_CLASSES.has(action.class) && !action.artifact) throw new RegistryError(`/action/${i}/artifact is required for a ${action.class} job (name and one exact version)`);++ if (action.firing_arg && action.args && Object.prototype.hasOwnProperty.call(action.args, action.firing_arg)) {++ throw new RegistryError(`/action/${i}/firing_arg names a key args already has (${action.firing_arg})`);++ }++ }if (create) {for (const field of ['source', 'extraction', 'condition', 'action']) {if (body[field] === undefined || body[field] === null) throw new RegistryError(`a new watch requires ${field}`);@@@ -132,7 +154,7 @@ function validateView(view) {return view;}--function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50, secretGrants = null }) {++function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50, secretGrants = null, projects = null }) {/** A named secret the owner may not use is refused when the watch is saved (server/secret-grants.js). */function checkSecrets(owner, rec) {if (!secretGrants) return;@@@ -144,6 +166,24 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxthrow new RegistryError('watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service');}}++ /**++ * The project a watch names (project delegation): a run-job action needs one, and a project newly named (`named`)++ * must be one Network says the owner is an admin or owner of. The refusal never says which check failed.++ */++ async function checkProject(owner, rec, named) {++ const project = rec.project_id || null;++ if (!project && Array.isArray(rec.action) && rec.action.some((a) => a && a.kind === 'run-job')) {++ throw new RegistryError('a run-job action is billed to a project: set project_id to a project you are an admin or owner of', 'watch.project_required');++ }++ if (!project || !named) return;++ const refused = new RegistryError('project_id must be a project you are an admin or owner of, and not archived', 'project.not_delegated', 403);++ if (!PERSON.test(String(owner || ''))) throw refused;++ if (!projects) throw new RegistryError('this Watch cannot check project memberships (no Network client)', 'watch.project_check_unavailable', 503);++ let r;++ try { r = await projects.check(project, owner); }++ catch { throw new RegistryError('Network could not confirm the project membership just now; try again', 'watch.project_check_unavailable', 503); }++ if (!r || !r.ok) throw refused;++ }const st = {get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),@@@ -265,6 +305,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxif (!owner) throw new RegistryError('no acting subject', 'watch.invalid');checkSecrets(owner, req);checkActorAgents(owner, req);++ await checkProject(owner, req, true);// A per-owner cap, so one caller cannot fill the registry (the API and the form both come// through here). The number is a config value (WATCH_MAX_PER_OWNER).const owned = await st.countOwned.get(owner);@@@ -288,7 +329,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox};const row = {id,-- project_id: null,++ project_id: req.project_id || null,owner_sub: owner,name: rec.name,status: rec.status,@@@ -328,12 +369,13 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxconst merged = { ...current, ...changes };delete merged.id; // an id never changesdelete merged.owner; // ownership never changes (a watch is the subject's)-- // project_id is not part of watch.watch-request@1: the row's own null is dropped, and a-- // caller that sends one fails the schema check below rather than being silently ignored.-- if (changes.project_id === undefined) delete merged.project_id;++ // The row's own null project is dropped (null is "none" on the wire too); a project the patch names, or the++ // one the watch already has, travels with the merged watch so a run-job action is judged against it.++ if (merged.project_id === null) delete merged.project_id;validateRequest(merged);checkSecrets(row.owner_sub, merged);checkActorAgents(row.owner_sub, merged);++ await checkProject(row.owner_sub, merged, changes.project_id !== undefined && changes.project_id !== row.project_id);const sets = [];const args = { id: row.id, updated_at: now(), updated_by: ownerOf(principal) };for (const [k, v] of Object.entries(changes)) {nnew file mode 100644@@@ -0,0 +1,54 @@++'use strict';++/**++ * Project delegation (ADR-003 amendment 2026-10-12), for a watch that names a project: Watch acts for the watch's owner++ * inside that project.++ *++ * const projects = createProjects(client, { networkUrl }); // client: openvibe-sdk/core createClient with Watch's token++ * await projects.check(projectId, subjectId) → { ok } (an admin or owner of the unarchived project), throws when Network cannot answer++ *++ * const run = createRunJobs(client, { baseUrl });++ * await run.submit({ subjectId, projectId, job, idempotencyKey }) → { id, state, created }++ *++ * check() reads the person's membership from Network (GET /internal/projects/:project/members/:subject, capability++ * network.project.member.read) when a watch is saved with a project. It is a courtesy to the person, who learns at++ * once that they cannot use that project; OpenVibe.Run asks Network again before every job, which is the check that++ * counts. submit() POSTs a run.job-create-request@1 body to Run with Watch's own token (run.job.submit, audience++ * openvibe.run), the owner in X-OV-Subject and the project in X-OV-Project: the job is billed to the project, the++ * person is its requester and Run records Watch as via. The call is made once; the action worker retries on its own++ * backoff. Refusals arrive as openvibe-sdk OpenVibeError (status, code).++ */++++const ADMIN_ROLES = new Set(['admin', 'owner']);++++function createProjects(client, { networkUrl }) {++ async function check(projectId, subjectId) {++ let out;++ try {++ out = await client.json({++ service: 'network', baseUrl: networkUrl, audience: 'openvibe.network', method: 'GET',++ path: `/internal/projects/${encodeURIComponent(projectId)}/members/${encodeURIComponent(subjectId)}`, retries: 0,++ });++ } catch (err) {++ if (err && err.status === 404) return { ok: false };++ throw err;++ }++ const ok = !!out && out.project_id === projectId && out.subject === subjectId && ADMIN_ROLES.has(out.role) && out.archived === false;++ return { ok };++ }++ return { check };++}++++function createRunJobs(client, { baseUrl }) {++ async function submit({ subjectId, projectId, job, idempotencyKey }) {++ const out = await client.json({++ service: 'run', baseUrl, audience: 'openvibe.run', method: 'POST', path: '/api/v1/jobs',++ headers: { 'X-OV-Subject': subjectId, 'X-OV-Project': projectId }, json: { ...job, idempotency_key: idempotencyKey },++ retries: 0, idempotencyKey: false,++ });++ const j = (out && out.job) || {};++ return { id: j.id || null, state: j.state || null, created: !!(out && out.created) };++ }++ return { submit };++}++++module.exports = { createProjects, createRunJobs, ADMIN_ROLES };@@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).*/--async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null } = {}) {++async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null, run = null, projects = null } = {}) {// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real// to fetch and verify against (no test touches the internet). Its URL is overridable via env.const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });@@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l});// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.const testdb = await require('./db').testDb();-- const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(now ? { now } : {}) });++ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(run ? { run } : {}), ...(projects ? { projects } : {}), ...(now ? { now } : {}) });const base = `http://127.0.0.1:${h.server.address().port}`;/** A session cookie for a person, as /auth/callback would have set it. */nnew file mode 100644@@@ -0,0 +1,230 @@++'use strict';++/**++ * Project-scoped watches and the run-job action (project delegation, ADR-003 amendment 2026-10-12):++ * - a watch names a project only when Network says its owner is an admin or owner of it (403 project.not_delegated,++ * 503 when Network cannot answer), checked when the project is named or changed;++ * - a run-job action needs a project and run.job-create-request@1's class and limits (artifact for function and++ * code); the contract draft's runtime/command/workload are refused;++ * - a fired run-job action submits one Run job for the owner in the project, the firing under args[firing_arg], with++ * the idempotency key watch.<check run>.<action index>; Run's final refusals fail at once, anything else retries.++ */++const assert = require('assert');++const { boot, request, site, suite, userToken, newUser, newAgent } = require('./helpers');++++const t = suite('project-run-jobs');++let svc;++let web;++let price = '19.99';++++const person = { sub: newUser() };++const viewer = { sub: newUser() };++const agent = { sub: newAgent() };++const P = 'prj_01JAB2C3D4E5F6G7H8J9K0MNPS';++const Q = 'prj_01JAB2C3D4E5F6G7H8J9K0MNPT';++++// Network, as server/run-jobs.js createProjects would ask it: who is an admin or owner of which project.++const checks = [];++let networkDown = false;++const projects = {++ async check(project, subject) {++ checks.push([project, subject]);++ if (networkDown) throw Object.assign(new Error('Network answered 503'), { status: 503 });++ return { ok: (project === P || project === Q) && subject === person.sub };++ },++};++// OpenVibe.Run, as server/run-jobs.js createRunJobs would reach it: every submit recorded; runMode decides the answer.++const submits = [];++let runMode = 'ok';++const run = {++ async submit(req) {++ submits.push(req);++ if (runMode === 'busy') throw Object.assign(new Error('Network could not confirm'), { status: 503, code: 'project.check_unavailable' });++ if (runMode === 'refuse') throw Object.assign(new Error('not delegated'), { status: 403, code: 'project.not_delegated', detail: 'a project is named only by a first-party service acting for an admin or owner' });++ if (runMode === 'caps') throw Object.assign(new Error('over the cap'), { status: 422, code: 'run.limits.exceeded', detail: 'limits.wall_ms is at most 900000 ms for a project' });++ return { id: `job_01JAB2C3D4E5F6G7H8J9K0MN${String(submits.length).padStart(2, '0')}`, state: 'queued', created: submits.length !== 3 };++ },++};++++const job = (over = {}) => ({++ kind: 'run-job', class: 'function', artifact: { name: 'restock', version: '1.0.0' },++ args: { sku: 'w-1' }, limits: { wall_ms: 60000, cpu_ms: 30000, mem_bytes: 134217728 }, egress: 'openvibe-only', ...over,++});++const def = (overrides = {}) => ({++ name: 'Widget price',++ source: { kind: 'http', url: `${web.origin}/price`, format: null },++ cadence: { every_sec: 900 },++ extraction: { kind: 'text' },++ condition: { op: 'lt', value: 10 },++ action: [job({ firing_arg: 'firing' })],++ project_id: P,++ ...overrides,++});++const refusedWith = (code, status) => (err) => { assert.strictEqual(err.code, code, err.message); if (status) assert.strictEqual(err.status, status); return true; };++const acts = async (id) => await svc.db.prepare('SELECT * FROM action_runs WHERE watch_id = ? ORDER BY rid').all(id);++const setPrice = async (v) => { await fetch(`${web.origin}/set?v=${v}`); };++async function fire(w) {++ await setPrice('19.99');++ await svc.check.run(w.id, { trigger: 'schedule' });++ await setPrice('7.50');++ const out = await svc.check.run(w.id, { trigger: 'schedule' });++ assert.strictEqual(out.run.state, 'condition_met');++ return out;++}++++t('boot', async () => {++ web = await site({++ '/price': () => ({ headers: { 'content-type': 'text/plain' }, body: price }),++ '/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },++ });++ const notifications = { async push() { return { sent: true }; } };++ svc = await boot({ notifications, run, projects });++});++++t('a watch names a project its owner is an admin or owner of; anyone else, and an agent, is refused', async () => {++ const w = await svc.registry.create(def(), person);++ assert.strictEqual(w.project_id, P);++ assert.deepStrictEqual(checks.pop(), [P, person.sub], 'checked with Network when named');++ assert.strictEqual(svc.registry.view(w).project_id, P, 'the view carries it (watch.watch@1)');++ await assert.rejects(() => svc.registry.create(def(), viewer), refusedWith('project.not_delegated', 403));++ await assert.rejects(() => svc.registry.create(def({ project_id: 'prj_01JAB2C3D4E5F6G7H8J9K0MNPV' }), person), refusedWith('project.not_delegated', 403), 'a project they do not run');++ await assert.rejects(() => svc.registry.create(def(), agent), refusedWith('project.not_delegated', 403), 'an agent never names a project');++ networkDown = true;++ try { await assert.rejects(() => svc.registry.create(def(), person), refusedWith('watch.project_check_unavailable', 503)); }++ finally { networkDown = false; }++ // A watch without run-job actions may name a project or not.++ const plain = await svc.registry.create(def({ action: [{ kind: 'notification' }], project_id: undefined }), person);++ assert.strictEqual(plain.project_id, null);++});++++t('a run-job action needs a project and Run\'s job fields; the draft\'s fields are refused', async () => {++ await assert.rejects(() => svc.registry.create(def({ project_id: undefined }), person), refusedWith('watch.project_required'));++ await assert.rejects(() => svc.registry.create(def({ project_id: null }), person), refusedWith('watch.project_required'));++ for (const [action, re] of [++ [{ kind: 'run-job', runtime: 'function', command: 'node x.js' }, /runtime is not accepted/],++ [job({ workload: {} }), /workload is not accepted/],++ [job({ class: undefined }), /class is required/],++ [job({ limits: undefined }), /limits is required/],++ [job({ artifact: undefined }), /artifact is required for a function job/],++ [job({ firing_arg: 'sku' }), /firing_arg names a key args already has/],++ ]) {++ await assert.rejects(() => svc.registry.create(def({ action: [action] }), person), (err) => {++ assert.strictEqual(err.code, 'watch.invalid', err.message);++ assert.match(err.message, re);++ return true;++ });++ }++ // A class that runs no artifact needs none (browser, linux…): Run places it when a node offers it.++ const browser = await svc.registry.create(def({ action: [job({ class: 'browser', artifact: undefined })] }), person);++ assert.ok(browser.id);++});++++t('patch: clearing the project of a run-job watch is refused; a new project is checked, an untouched one is not', async () => {++ const w = await svc.registry.create(def(), person);++ await assert.rejects(() => svc.registry.patch(w.id, { project_id: null }, person), refusedWith('watch.project_required'));++ const before = checks.length;++ await svc.registry.patch(w.id, { name: 'renamed' }, person);++ assert.strictEqual(checks.length, before, 'the project the watch already has is not asked about again (Run asks at every job)');++ const moved = await svc.registry.patch(w.id, { project_id: Q }, person);++ assert.strictEqual(moved.project_id, Q);++ assert.deepStrictEqual(checks.pop(), [Q, person.sub]);++ await assert.rejects(() => svc.registry.patch(w.id, { project_id: 'prj_01JAB2C3D4E5F6G7H8J9K0MNPV' }, person), refusedWith('project.not_delegated', 403));++ const plain = await svc.registry.patch(w.id, { action: [{ kind: 'notification' }], project_id: null }, person);++ assert.strictEqual(plain.project_id, null, 'without run-job actions the project may go');++});++++t('the API answers the refusals with their own status', async () => {++ const token = userToken(viewer.sub);++ const r = await request(svc.base, 'POST', '/api/v1/watches', { token, body: def() });++ assert.strictEqual(r.status, 403, r.text);++ assert.strictEqual(r.body.code, 'project.not_delegated');++ const missing = await request(svc.base, 'POST', '/api/v1/watches', { token: userToken(person.sub), body: def({ project_id: undefined }) });++ assert.strictEqual(missing.status, 422, missing.text);++ assert.strictEqual(missing.body.code, 'watch.project_required');++});++++t('a fired run-job action submits one job for the owner in the project, with the firing, under one key', async () => {++ submits.length = 0; runMode = 'busy';++ const w = await svc.registry.create(def({ name: 'restock' }), person);++ const out = await fire(w);++ let r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 1, failed: 0 }, 'Run (or Network behind it) unavailable: retried');++ runMode = 'ok';++ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);++ r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });++ const [first, second] = submits;++ assert.strictEqual(first.subjectId, person.sub, 'for the owner');++ assert.strictEqual(first.projectId, P, 'in the watch\'s project');++ assert.strictEqual(first.idempotencyKey, `watch.${out.run.id}.0`);++ assert.strictEqual(second.idempotencyKey, first.idempotencyKey, 'the retry is the same job');++ assert.deepStrictEqual({ ...first.job, args: undefined }, {++ class: 'function', artifact: { name: 'restock', version: '1.0.0' }, limits: { wall_ms: 60000, cpu_ms: 30000, mem_bytes: 134217728 }, egress: 'openvibe-only', args: undefined,++ }, 'run.job-create-request@1 fields only (no kind, no firing_arg)');++ assert.strictEqual(first.job.args.sku, 'w-1');++ const firing = first.job.args.firing;++ assert.deepStrictEqual([firing.watch_id, firing.check_run_id, firing.value, firing.previous_value], [w.id, out.run.id, '7.50', '19.99']);++ assert.ok(firing.watch_url.endsWith(`/watches/${w.id}`) && firing.observed_at);++ const [row] = await acts(w.id);++ assert.deepStrictEqual([row.state, row.last_error], ['sent', 'Run job job_01JAB2C3D4E5F6G7H8J9K0MN02']);++ const req = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;++ assert.strictEqual(req.project_id, P, 'the project is fixed when the action is queued');++});++++t('Run\'s final refusals fail at once; an agent-owned or project-less row never reaches Run', async () => {++ for (const mode of ['refuse', 'caps']) {++ runMode = mode;++ const w = await svc.registry.create(def({ name: `refused ${mode}` }), person);++ await fire(w);++ const r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 0, failed: 1 }, mode);++ assert.match((await acts(w.id))[0].last_error, mode === 'refuse' ? /watch\.action_refused: project\.not_delegated/ : /watch\.action_refused: run\.limits\.exceeded/);++ }++ runMode = 'ok';++ // A row queued for a watch whose project was cleared behind the registry's back (or never set): not sent.++ const w = await svc.registry.create(def({ name: 'no project' }), person);++ await svc.db.prepare('UPDATE watches SET project_id = NULL WHERE id = ?').run(w.id);++ const before = submits.length;++ await fire(w);++ const [row] = await acts(w.id);++ assert.deepStrictEqual([row.state, /project_required/.test(row.last_error)], ['unavailable', true]);++ await svc.actions.tick();++ assert.strictEqual(submits.length, before, 'nothing submitted');++});++++t('server/run-jobs.js asks Network and Run as Watch, for the owner, once', async () => {++ const { createProjects, createRunJobs } = require('../server/run-jobs');++ const calls = [];++ let answer = { project_id: P, subject: person.sub, role: 'admin', archived: false, added_at: null };++ let fail = null;++ const client = { async json(o) { calls.push(o); if (fail) throw fail; return answer; } };++ const projectsLive = createProjects(client, { networkUrl: 'http://127.0.0.1:4000' });++ assert.deepStrictEqual(await projectsLive.check(P, person.sub), { ok: true });++ const c = calls[0];++ assert.deepStrictEqual([c.service, c.baseUrl, c.audience, c.method, c.path, c.retries],++ ['network', 'http://127.0.0.1:4000', 'openvibe.network', 'GET', `/internal/projects/${P}/members/${person.sub}`, 0]);++ for (const [a, why] of [++ [{ ...answer, role: 'developer' }, 'a developer'], [{ ...answer, role: 'viewer' }, 'a viewer'],++ [{ ...answer, archived: true }, 'an archived project'], [{ ...answer, subject: viewer.sub }, 'an answer about someone else'],++ ]) { answer = a; assert.deepStrictEqual(await projectsLive.check(P, person.sub), { ok: false }, why); }++ answer = { ...answer, role: 'owner', archived: false, subject: person.sub };++ assert.deepStrictEqual(await projectsLive.check(P, person.sub), { ok: true }, 'an owner');++ fail = Object.assign(new Error('not found'), { status: 404 });++ assert.deepStrictEqual(await projectsLive.check(P, person.sub), { ok: false }, 'a non-member');++ fail = Object.assign(new Error('down'), { status: 503 });++ await assert.rejects(() => projectsLive.check(P, person.sub), /down/, 'Network down: the caller answers 503');++ fail = null;++++ answer = { created: true, job: { id: 'job_X', state: 'queued' } };++ const runLive = createRunJobs(client, { baseUrl: 'http://127.0.0.1:4920' });++ const outJob = await runLive.submit({ subjectId: person.sub, projectId: P, job: { class: 'function', limits: { wall_ms: 1, cpu_ms: 1, mem_bytes: 1 } }, idempotencyKey: 'watch.r.0' });++ assert.deepStrictEqual(outJob, { id: 'job_X', state: 'queued', created: true });++ const s = calls[calls.length - 1];++ assert.deepStrictEqual([s.service, s.baseUrl, s.audience, s.method, s.path, s.retries, s.idempotencyKey],++ ['run', 'http://127.0.0.1:4920', 'openvibe.run', 'POST', '/api/v1/jobs', 0, false]);++ assert.deepStrictEqual(s.headers, { 'X-OV-Subject': person.sub, 'X-OV-Project': P });++ assert.deepStrictEqual(s.json, { class: 'function', limits: { wall_ms: 1, cpu_ms: 1, mem_bytes: 1 }, idempotency_key: 'watch.r.0' });++});++++t('done', async () => { await web.close(); await svc.stop(); });++++t.run();
@@ -228,7 +228,16 @@ A fired condition queues the watch's actions in the same transaction as the chec
@@ -228,7 +228,16 @@ A fired condition queues the watch's actions in the same transaction as the chec
| `webhook` | a POST of the firing (`type`, `watch {id, name, url}`, `check_run_id`, `condition`, `observation {observed_at, value, previous_value}`) to the action's https URL, through the same SSRF guard as every fetch, never following a redirect ([server/net/post.js](server/net/post.js)). It is signed with the key `secret_env` names exactly as OpenVibe.Events signs a delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`; check it with openvibe-sdk/events `verifyDeliveryV2`). `X-OpenVibe-Delivery-Id` (`<check run>.<action index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 or a URL the guard refuses is `failed` at once. Anything else, or a lost connection, is retried on the backoff, then `failed`. The key must be granted to the owner (`WATCH_SECRET_GRANTS`, below) and set, 32 characters or more: otherwise `unavailable` |
| `webhook` | a POST of the firing (`type`, `watch {id, name, url}`, `check_run_id`, `condition`, `observation {observed_at, value, previous_value}`) to the action's https URL, through the same SSRF guard as every fetch, never following a redirect ([server/net/post.js](server/net/post.js)). It is signed with the key `secret_env` names exactly as OpenVibe.Events signs a delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`; check it with openvibe-sdk/events `verifyDeliveryV2`). `X-OpenVibe-Delivery-Id` (`<check run>.<action index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 or a URL the guard refuses is `failed` at once. Anything else, or a lost connection, is retried on the backoff, then `failed`. The key must be granted to the owner (`WATCH_SECRET_GRANTS`, below) and set, 32 characters or more: otherwise `unavailable` |
| `actor-task` | a task for the owner on OpenVibe.Actor ([server/actor.js](server/actor.js)): the action's `brief`, followed by the firing as context (the watch, the value and what it was, its page), at most `budget_usd`. Watch names the owner in `X-OV-Subject`, so the task runs under **the owner's** allowance and appears in their task list, and Actor records Watch as `via`. The idempotency key `watch.<check run>.<action index>` makes a retried delivery the same task. Created is `sent`, with the task id recorded. A request Actor refuses as invalid (over the tier's budget, an unknown agent) is `failed` at once. Anything else, including a day's allowance spent, is retried on the backoff. An agent- or service-owned watch is `skipped`. Needs `actor.task.create` for `openvibe.actor` (Network grants it) and `WATCH_ACTOR_URL` |
| `actor-task` | a task for the owner on OpenVibe.Actor ([server/actor.js](server/actor.js)): the action's `brief`, followed by the firing as context (the watch, the value and what it was, its page), at most `budget_usd`. Watch names the owner in `X-OV-Subject`, so the task runs under **the owner's** allowance and appears in their task list, and Actor records Watch as `via`. The idempotency key `watch.<check run>.<action index>` makes a retried delivery the same task. Created is `sent`, with the task id recorded. A request Actor refuses as invalid (over the tier's budget, an unknown agent) is `failed` at once. Anything else, including a day's allowance spent, is retried on the backoff. An agent- or service-owned watch is `skipped`. Needs `actor.task.create` for `openvibe.actor` (Network grants it) and `WATCH_ACTOR_URL` |
| `actor-agent` | runs one of the person's OpenVibe.Actor agents by `agent_id`. Its optional `note` precedes the firing context (watch name and link, value and previous value, observation time); input is capped at 8000 characters without cutting the note. Watch uses `X-OV-Subject` for the owner. A created run is `sent` with its task id; a missing agent, limit or free-tier refusal is `failed` for that firing; service errors retry on the usual backoff. Only person-owned watches may use it. Needs the `actor.agent.run` grant for `openvibe.actor` and `WATCH_ACTOR_URL` |
| `actor-agent` | runs one of the person's OpenVibe.Actor agents by `agent_id`. Its optional `note` precedes the firing context (watch name and link, value and previous value, observation time); input is capped at 8000 characters without cutting the note. Watch uses `X-OV-Subject` for the owner. A created run is `sent` with its task id; a missing agent, limit or free-tier refusal is `failed` for that firing; service errors retry on the usual backoff. Only person-owned watches may use it. Needs the `actor.agent.run` grant for `openvibe.actor` and `WATCH_ACTOR_URL` |
-| `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |
+| `run-job` | one OpenVibe.Run job for the owner in the watch's project ([server/run-jobs.js](server/run-jobs.js)): the action's `run.job-create-request@1` fields (`class`, `artifact`, `args`, `limits`, `egress`, `ttl_ms`), with the firing (watch, check run, value, previous value, time) under `args[firing_arg]` when the action names one. Watch submits as itself with the owner in `X-OV-Subject` and the project in `X-OV-Project` (project delegation, ADR-003 amendment 2026-10-12): Run bills the project, checks the owner is still an admin or owner of it, makes the owner the requester and records Watch as `via`. Key `watch.<check run>.<action index>`, so a retry is the same job. Created is `sent` with the job id; Run's final refusals (400, 403, 404, 409, 413, 422) are `failed` at once; anything else retries on the backoff. `WATCH_RUN_URL` is Run's internal address (default `http://127.0.0.1:4920`). |
+| `codes-run` | recorded `unavailable` (`watch.action_unavailable`): its dispatcher is a later step, and the watch page says so |
+
+A watch with a `run-job` action names a **project** (`project_id` on `watch.watch-request@1`, required for it:
+`422 watch.project_required`). Only a person names one, and only a project OpenVibe.Network says they are an admin or
+owner of and that is not archived (`GET /internal/projects/:project/members/:subject`, `network.project.member.read`),
+checked when the project is named or changed: `403 project.not_delegated` otherwise, `503
+watch.project_check_unavailable` when Network cannot answer. Run checks again before every job. The run-job action
+needs `class` and `limits` (and `artifact` for `function` and `code`); the contract draft's `runtime`, `command` and
+`workload` are refused.
Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.
Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.
Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.
Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.
@@ -12,7 +12,7 @@
@@ -12,7 +12,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.153.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -769,9 +769,9 @@
@@ -769,9 +769,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.143.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
- "integrity": "sha512-Cdbf+9RoPAUBibpM2zXQvtsiKMugb55dKubTt8pFZvAmxsLtGvX+zaC++yL1vOSub8FX3+GgqcGtoJS6tklrUg==",
+ "version": "0.153.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.153.0",
+ "integrity": "sha512-WQ7p7ATKtTky9nI9yE9Agkm3rNKuyWljmrMCdZaqxg+FjVaziTqrFz0a3YMBTXJmNf5JTGXn/ioOmWjO95/lCA==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -24,7 +24,7 @@
@@ -24,7 +24,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.153.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -37,9 +37,18 @@
@@ -37,9 +37,18 @@
* owner: an agent- or service-owned watch is `skipped`.
* owner: an agent- or service-owned watch is `skipped`.
* actor-agent runs one of the owner's personal agents on Actor with a note and bounded firing context.
* actor-agent runs one of the owner's personal agents on Actor with a note and bounded firing context.
* A missing agent or allowance refusal fails this firing; a service error retries.
* A missing agent or allowance refusal fails this firing; a service error retries.
- * codes-run, run-job
- * `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and
- * the watch page says so rather than pretending they ran.
+ * run-job one OpenVibe.Run job for the owner in the watch's project (server/run-jobs.js): the action's
+ * run.job-create-request@1 fields, with the firing under args[firing_arg] when it names one. Watch
+ * submits as itself, the owner in X-OV-Subject and the project in X-OV-Project (project delegation,
+ * ADR-003 amendment 2026-10-12), so Run bills the project, checks the owner is still an admin or
+ * owner of it, makes the owner the requester and records Watch as via. The idempotency key
+ * `watch.<check run>.<action index>` makes a retried delivery the same job. Created (or the same
+ * job again) is `sent` with the job id. A refusal that cannot change on a retry (400, 403, 404, 409,
+ * 413, 422: an invalid job, a project the owner may no longer use, over the project's caps) is
+ * `failed` at once; anything else, Run or Network unreachable included, is retried on the backoff.
+ * It needs a person as owner and a project: otherwise `skipped` / `unavailable`.
+ * codes-run `unavailable` with `watch.action_unavailable`: its dispatcher is a later step of the plan, and the
+ * watch page says so rather than pretending it ran.
*
*
* The worker claims due rows with a short lease (FOR UPDATE SKIP LOCKED, next_attempt_at moved past the lease), so
* The worker claims due rows with a short lease (FOR UPDATE SKIP LOCKED, next_attempt_at moved past the lease), so
* two processes never deliver one row at the same time, and a process that dies mid-delivery leaves the row to the
* two processes never deliver one row at the same time, and a process that dies mid-delivery leaves the row to the
@@ -50,11 +59,16 @@
@@ -50,11 +59,16 @@
const { signDeliveryHeaders } = require('openvibe-sdk/events');
const { signDeliveryHeaders } = require('openvibe-sdk/events');
const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const PROJECT_ID = /^prj_[0-9A-HJKMNP-TV-Z]{26}$/;
const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000];
const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000];
const LEASE_MS = 2 * 60_000;
const LEASE_MS = 2 * 60_000;
const TITLE_MAX = 120;
const TITLE_MAX = 120;
const MESSAGE_MAX = 300;
const MESSAGE_MAX = 300;
-const LATER = new Set(['codes-run', 'run-job']);
+const LATER = new Set(['codes-run']);
+// Run's answers that a retry cannot change: the job is invalid, over the project's caps, or the owner may no longer
+// act in the project (403 project.not_delegated / capability.denied).
+const RUN_FINAL = new Set([400, 403, 404, 409, 413, 422]);
+const JOB_FIELDS = ['class', 'artifact', 'args', 'limits', 'egress', 'ttl_ms'];
const TASK_MAX = 20000;
const TASK_MAX = 20000;
const TRIGGERED = 'watch.watch.triggered';
const TRIGGERED = 'watch.watch.triggered';
const MIN_HOOK_SECRET = 32;
const MIN_HOOK_SECRET = 32;
@@ -121,13 +135,36 @@ function actorContextFor({ config, watch, value, previousValue, observedAt }, ma
@@ -121,13 +135,36 @@ function actorContextFor({ config, watch, value, previousValue, observedAt }, ma
return [first, valueLine.length > room ? `${valueLine.slice(0, Math.max(0, room - 1))}…` : valueLine, last].join('\n').slice(0, maxLength);
return [first, valueLine.length > room ? `${valueLine.slice(0, Math.max(0, room - 1))}…` : valueLine, last].join('\n').slice(0, maxLength);
}
}
+/**
+ * The run.job-create-request@1 body a run-job action submits (without its idempotency key): the action's job fields,
+ * and the firing under args[firing_arg] when the action names one.
+ */
+function runJobFor({ config, watch, run, action, value, previousValue, observedAt }) {
+ const job = {};
+ for (const k of JOB_FIELDS) if (action[k] !== undefined && action[k] !== null) job[k] = action[k];
+ if (action.firing_arg) {
+ job.args = {
+ ...(job.args || {}),
+ [action.firing_arg]: {
+ watch_id: watch.id,
+ watch_url: `${config.baseUrl}/watches/${encodeURIComponent(watch.id)}`,
+ check_run_id: run.id,
+ observed_at: observedAt ? new Date(observedAt).toISOString() : null,
+ value: value === undefined ? null : value,
+ previous_value: previousValue === undefined ? null : previousValue,
+ },
+ };
+ }
+ return job;
+}
+
function actorAgentInputFor(args) {
function actorAgentInputFor(args) {
const note = typeof args.action.note === 'string' ? args.action.note : '';
const note = typeof args.action.note === 'string' ? args.action.note : '';
const prefix = note ? `${note}\n\n` : '';
const prefix = note ? `${note}\n\n` : '';
return prefix + actorContextFor(args, 8000 - prefix.length);
return prefix + actorContextFor(args, 8000 - prefix.length);
}
}
-function createActions({ db, config, notifications = null, post = null, actor = null, now = () => Date.now(), log = console }) {
+function createActions({ db, config, notifications = null, post = null, actor = null, run: runJobs = null, now = () => Date.now(), log = console }) {
// Read at each use, not once: what a process was started with is what it enforces, for rows queued before it too.
// Read at each use, not once: what a process was started with is what it enforces, for rows queued before it too.
const grants = () => config.secretGrants;
const grants = () => config.secretGrants;
const enabled = Boolean(notifications) && config.actions.enabled;
const enabled = Boolean(notifications) && config.actions.enabled;
@@ -175,6 +212,10 @@ function createActions({ db, config, notifications = null, post = null, actor =
@@ -175,6 +212,10 @@ function createActions({ db, config, notifications = null, post = null, actor =
if (USER_SUBJECT.test(String(owner || ''))) {
if (USER_SUBJECT.test(String(owner || ''))) {
request = { agent_id: action.agent_id, input: actorAgentInputFor({ config, watch, action, value, previousValue, observedAt }) };
request = { agent_id: action.agent_id, input: actorAgentInputFor({ config, watch, action, value, previousValue, observedAt }) };
} else { state = 'skipped'; error = 'watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service'; }
} else { state = 'skipped'; error = 'watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service'; }
+ } else if (kind === 'run-job') {
+ if (!USER_SUBJECT.test(String(owner || ''))) { state = 'skipped'; error = 'watch.action_no_person: a Run job is started for a person in their project, and this watch belongs to an agent or a service'; }
+ else if (!PROJECT_ID.test(String(watch.project_id || ''))) { state = 'unavailable'; error = 'watch.project_required: a run-job action is billed to the watch\'s project, and this watch names none'; }
+ else request = { project_id: watch.project_id, job: runJobFor({ config, watch, run, action, value, previousValue, observedAt }) };
} else if (LATER.has(kind)) {
} else if (LATER.has(kind)) {
state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;
state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;
} else {
} else {
@@ -240,11 +281,26 @@ function createActions({ db, config, notifications = null, post = null, actor =
@@ -240,11 +281,26 @@ function createActions({ db, config, notifications = null, post = null, actor =
}
}
}
}
+ async function deliverRunJob(row, request) {
+ if (!runJobs) throw Object.assign(new Error('Run delivery is not configured in this process'), { code: 'run_unconfigured' });
+ try {
+ const job = await runJobs.submit({
+ subjectId: row.owner_sub, projectId: request.project_id, job: request.job,
+ idempotencyKey: `watch.${row.check_run_id}.${row.action_index}`,
+ });
+ return { state: 'sent', error: `Run job ${job.id || '(no id)'}${job.created ? '' : ' (already started)'}` };
+ } catch (err) {
+ if (err && RUN_FINAL.has(err.status)) return { state: 'failed', error: `watch.action_refused: ${err.code || err.status}: ${String(err.detail || err.message || '').slice(0, 200)}` };
+ throw err;
+ }
+ }
+
async function deliver(row, attempt) {
async function deliver(row, attempt) {
const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;
const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;
if (row.kind === 'webhook') return await deliverWebhook(row, request, attempt);
if (row.kind === 'webhook') return await deliverWebhook(row, request, attempt);
if (row.kind === 'actor-task') return await deliverActorTask(row, request);
if (row.kind === 'actor-task') return await deliverActorTask(row, request);
if (row.kind === 'actor-agent') return await deliverActorAgent(row, request);
if (row.kind === 'actor-agent') return await deliverActorAgent(row, request);
+ if (row.kind === 'run-job') return await deliverRunJob(row, request);
if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };
if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };
const out = await notifications.push({ subjectId: row.owner_sub, ...request });
const out = await notifications.push({ subjectId: row.owner_sub, ...request });
if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };
if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };
@@ -316,4 +372,4 @@ function createActions({ db, config, notifications = null, post = null, actor =
@@ -316,4 +372,4 @@ function createActions({ db, config, notifications = null, post = null, actor =
return { queue, tick, start, stop, status, recent, enabled };
return { queue, tick, start, stop, status, recent, enabled };
}
}
-module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, actorAgentInputFor, BACKOFF_MS };
+module.exports = { createActions, notificationFor, webhookFor, actorTaskFor, actorAgentInputFor, runJobFor, BACKOFF_MS, RUN_FINAL };
@@ -29,7 +29,7 @@ function watchesRouter({ registry, auth, limits, webhookSources = null }) {
@@ -29,7 +29,7 @@ function watchesRouter({ registry, auth, limits, webhookSources = null }) {
const fail = (res, req, err) => {
const fail = (res, req, err) => {
if (err instanceof RegistryError) {
if (err instanceof RegistryError) {
- const status = err.code === 'watch.internal' ? 500 : (err.code === 'watch.exists' ? 409 : 422);
+ const status = err.status || (err.code === 'watch.internal' ? 500 : (err.code === 'watch.exists' ? 409 : 422));
return http.sendProblem(res, status, err.code, { detail: err.message, ctx: req.ov });
return http.sendProblem(res, status, err.code, { detail: err.message, ctx: req.ov });
}
}
throw err;
throw err;
@@ -101,6 +101,8 @@ function load(env = process.env) {
@@ -101,6 +101,8 @@ function load(env = process.env) {
webhookTimeoutMs: int(env.WATCH_WEBHOOK_TIMEOUT_MS, 10000),
webhookTimeoutMs: int(env.WATCH_WEBHOOK_TIMEOUT_MS, 10000),
// OpenVibe.Actor's internal address, for actor-task and actor-agent actions.
// OpenVibe.Actor's internal address, for actor-task and actor-agent actions.
actorUrl: (env.WATCH_ACTOR_URL || 'http://127.0.0.1:4950').replace(/\/+$/, ''),
actorUrl: (env.WATCH_ACTOR_URL || 'http://127.0.0.1:4950').replace(/\/+$/, ''),
+ // OpenVibe.Run's internal address, for run-job actions (a job for the owner in the watch's project).
+ runUrl: (env.WATCH_RUN_URL || 'http://127.0.0.1:4920').replace(/\/+$/, ''),
},
},
// Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins
// Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins
@@ -16,6 +16,7 @@ const { createObservations } = require('./observations');
@@ -16,6 +16,7 @@ const { createObservations } = require('./observations');
const { createGuard } = require('./net/guard');
const { createGuard } = require('./net/guard');
const { createPoster } = require('./net/post');
const { createPoster } = require('./net/post');
const { createActorTasks } = require('./actor');
const { createActorTasks } = require('./actor');
+const { createProjects, createRunJobs } = require('./run-jobs');
const { createFetcher } = require('./net/fetcher');
const { createFetcher } = require('./net/fetcher');
const { createSpacer } = require('./spacer');
const { createSpacer } = require('./spacer');
const { createCarriers } = require('./carriers');
const { createCarriers } = require('./carriers');
@@ -40,7 +41,7 @@ const { createApp } = require('./app');
@@ -40,7 +41,7 @@ const { createApp } = require('./app');
const accountDataLib = require('./account-data');
const accountDataLib = require('./account-data');
const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
-async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null, actor: givenActor = null } = {}) {
+async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null, actor: givenActor = null, run: givenRun = null, projects: givenProjects = null } = {}) {
config = config || load();
config = config || load();
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
const db = givenDb || await openDb(config, { log });
const db = givenDb || await openDb(config, { log });
@@ -52,7 +53,19 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -52,7 +53,19 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed'],
'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed'],
validate: (env) => validate('events.event-envelope@1', env),
validate: (env) => validate('events.event-envelope@1', env),
});
});
- const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner, secretGrants: config.secretGrants });
+ // Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own
+ // client-credentials token; without the client secret they wait (actions.status() says delivery is off).
+ // Actor tasks (step 6) and Run jobs (project delegation) use the same client and token provider (a token per audience).
+ const client = config.oauth.clientSecret
+ ? createClient({
+ tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),
+ baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl, run: config.actions.runUrl }, fetch: fetchImpl,
+ })
+ : null;
+ // A watch that names a project: Network says whether its owner is an admin or owner of it, when it is saved (Run checks
+ // again before every job). Without the client secret no project can be checked, so none can be named.
+ const projects = givenProjects || (client ? createProjects(client, { networkUrl: config.networkInternalUrl }) : null);
+ const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner, secretGrants: config.secretGrants, projects });
const observations = createObservations({ db, now });
const observations = createObservations({ db, now });
// The only way out to the internet: SSRF guard (ports, private hosts, every redirect hop), byte
// The only way out to the internet: SSRF guard (ports, private hosts, every redirect hop), byte
// cap, one deadline, conditional GET; per-host spacing in front of every request.
// cap, one deadline, conditional GET; per-host spacing in front of every request.
@@ -61,20 +74,12 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -61,20 +74,12 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });
const spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });
// Every carrier goes out through the spacer, so two watches over one host never hammer it.
// Every carrier goes out through the spacer, so two watches over one host never hammer it.
const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });
const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });
- // Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own
- // client-credentials token; without the client secret they wait (actions.status() says delivery is off).
- // Actor tasks (step 6) use the same client and token provider (a token per audience).
- const client = config.oauth.clientSecret
- ? createClient({
- tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),
- baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl }, fetch: fetchImpl,
- })
- : null;
const notifications = client ? createNotificationsClient(client) : null;
const notifications = client ? createNotificationsClient(client) : null;
const actorTasks = client ? createActorTasks(client, { baseUrl: config.actions.actorUrl }) : null;
const actorTasks = client ? createActorTasks(client, { baseUrl: config.actions.actorUrl }) : null;
// Webhook actions leave through the same guard as every fetch (server/net/post.js).
// Webhook actions leave through the same guard as every fetch (server/net/post.js).
const post = givenPost || createPoster({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.actions.webhookTimeoutMs });
const post = givenPost || createPoster({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.actions.webhookTimeoutMs });
- const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, now, log });
+ const runJobs = client ? createRunJobs(client, { baseUrl: config.actions.runUrl }) : null;
+ const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, run: givenRun || runJobs, now, log });
const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });
const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.
// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.
@@ -16,6 +16,13 @@
@@ -16,6 +16,13 @@
* Ownership is the authority: every read and write is scoped to the acting subject (principal.sub,
* Ownership is the authority: every read and write is scoped to the acting subject (principal.sub,
* or the person a first-party service names with X-OV-Subject), and another owner's watch is not
* or the person a first-party service names with X-OV-Subject), and another owner's watch is not
* found — 404, never 403 (an id must not confirm that someone else's watch exists).
* found — 404, never 403 (an id must not confirm that someone else's watch exists).
+ *
+ * A watch may name a project (project_id, ADR-003 amendment 2026-10-12): the project its run-job actions are billed
+ * to. Only a person names one, and only a project Network says they are an admin or owner of (not archived), checked
+ * when the project is first named or changed (403 project.not_delegated; 503 watch.project_check_unavailable when
+ * Network cannot answer). OpenVibe.Run checks again before every job, so a later loss of the role stops the jobs. A
+ * run-job action needs the watch to name a project (422 watch.project_required) and run.job-create-request@1's
+ * class and limits (and artifact for function and code); the draft's runtime/command/workload are refused.
*/
*/
const { validate, ids: contractIds } = require('openvibe-contracts');
const { validate, ids: contractIds } = require('openvibe-contracts');
const safe = require('./safe-regex');
const safe = require('./safe-regex');
@@ -32,9 +39,13 @@ const MIN_EVERY_SEC = 900;
@@ -32,9 +39,13 @@ const MIN_EVERY_SEC = 900;
const FORBIDDEN_HEADERS = new Set(['host', 'cookie', 'connection', 'content-length', 'transfer-encoding', 'user-agent', 'accept-encoding', 'if-none-match', 'if-modified-since']);
const FORBIDDEN_HEADERS = new Set(['host', 'cookie', 'connection', 'content-length', 'transfer-encoding', 'user-agent', 'accept-encoding', 'if-none-match', 'if-modified-since']);
const PERSON = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const PERSON = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const AGENT_ID = /^agt_[0-9A-HJKMNP-TV-Z]{26}$/;
const AGENT_ID = /^agt_[0-9A-HJKMNP-TV-Z]{26}$/;
+/** The run-job fields of the contract's early draft, which never ran: refused, so nobody writes a job that cannot start. */
+const RUN_JOB_DRAFT = ['runtime', 'command', 'workload'];
+const ARTIFACT_CLASSES = new Set(['function', 'code']);
class RegistryError extends Error {
class RegistryError extends Error {
- constructor(detail, code = 'watch.invalid') { super(detail); this.code = code; }
+ /** status: the HTTP answer when it is not the default (422; 500 for watch.internal, 409 for watch.exists). */
+ constructor(detail, code = 'watch.invalid', status = null) { super(detail); this.code = code; if (status) this.status = status; }
}
}
const parse = (v) => (v == null ? null : (typeof v === 'string' ? JSON.parse(v) : v));
const parse = (v) => (v == null ? null : (typeof v === 'string' ? JSON.parse(v) : v));
@@ -71,6 +82,17 @@ function validateRequest(body, { create = false } = {}) {
@@ -71,6 +82,17 @@ function validateRequest(body, { create = false } = {}) {
}
}
const v = validate('watch.watch-request@1', body);
const v = validate('watch.watch-request@1', body);
if (!v.valid) throw new RegistryError(`body does not match watch.watch-request@1: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`);
if (!v.valid) throw new RegistryError(`body does not match watch.watch-request@1: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`);
+ for (const [i, action] of (Array.isArray(body.action) ? body.action : []).entries()) {
+ if (!action || action.kind !== 'run-job') continue;
+ const draft = RUN_JOB_DRAFT.find((f) => action[f] !== undefined);
+ if (draft) throw new RegistryError(`/action/${i}/${draft} is not accepted: a run-job action takes run.job-create-request@1 fields (class, artifact, args, limits, egress, ttl_ms)`);
+ if (!action.class) throw new RegistryError(`/action/${i}/class is required (a platform.runtime-class@1 name)`);
+ if (!action.limits) throw new RegistryError(`/action/${i}/limits is required (wall_ms, cpu_ms, mem_bytes)`);
+ if (ARTIFACT_CLASSES.has(action.class) && !action.artifact) throw new RegistryError(`/action/${i}/artifact is required for a ${action.class} job (name and one exact version)`);
+ if (action.firing_arg && action.args && Object.prototype.hasOwnProperty.call(action.args, action.firing_arg)) {
+ throw new RegistryError(`/action/${i}/firing_arg names a key args already has (${action.firing_arg})`);
+ }
+ }
if (create) {
if (create) {
for (const field of ['source', 'extraction', 'condition', 'action']) {
for (const field of ['source', 'extraction', 'condition', 'action']) {
if (body[field] === undefined || body[field] === null) throw new RegistryError(`a new watch requires ${field}`);
if (body[field] === undefined || body[field] === null) throw new RegistryError(`a new watch requires ${field}`);
@@ -132,7 +154,7 @@ function validateView(view) {
@@ -132,7 +154,7 @@ function validateView(view) {
return view;
return view;
}
}
-function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50, secretGrants = null }) {
+function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50, secretGrants = null, projects = null }) {
/** A named secret the owner may not use is refused when the watch is saved (server/secret-grants.js). */
/** A named secret the owner may not use is refused when the watch is saved (server/secret-grants.js). */
function checkSecrets(owner, rec) {
function checkSecrets(owner, rec) {
if (!secretGrants) return;
if (!secretGrants) return;
@@ -144,6 +166,24 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -144,6 +166,24 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
throw new RegistryError('watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service');
throw new RegistryError('watch.action_no_person: an Actor agent runs for a person, and this watch belongs to an agent or a service');
}
}
}
}
+ /**
+ * The project a watch names (project delegation): a run-job action needs one, and a project newly named (`named`)
+ * must be one Network says the owner is an admin or owner of. The refusal never says which check failed.
+ */
+ async function checkProject(owner, rec, named) {
+ const project = rec.project_id || null;
+ if (!project && Array.isArray(rec.action) && rec.action.some((a) => a && a.kind === 'run-job')) {
+ throw new RegistryError('a run-job action is billed to a project: set project_id to a project you are an admin or owner of', 'watch.project_required');
+ }
+ if (!project || !named) return;
+ const refused = new RegistryError('project_id must be a project you are an admin or owner of, and not archived', 'project.not_delegated', 403);
+ if (!PERSON.test(String(owner || ''))) throw refused;
+ if (!projects) throw new RegistryError('this Watch cannot check project memberships (no Network client)', 'watch.project_check_unavailable', 503);
+ let r;
+ try { r = await projects.check(project, owner); }
+ catch { throw new RegistryError('Network could not confirm the project membership just now; try again', 'watch.project_check_unavailable', 503); }
+ if (!r || !r.ok) throw refused;
+ }
const st = {
const st = {
get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
@@ -265,6 +305,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -265,6 +305,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
if (!owner) throw new RegistryError('no acting subject', 'watch.invalid');
if (!owner) throw new RegistryError('no acting subject', 'watch.invalid');
checkSecrets(owner, req);
checkSecrets(owner, req);
checkActorAgents(owner, req);
checkActorAgents(owner, req);
+ await checkProject(owner, req, true);
// A per-owner cap, so one caller cannot fill the registry (the API and the form both come
// A per-owner cap, so one caller cannot fill the registry (the API and the form both come
// through here). The number is a config value (WATCH_MAX_PER_OWNER).
// through here). The number is a config value (WATCH_MAX_PER_OWNER).
const owned = await st.countOwned.get(owner);
const owned = await st.countOwned.get(owner);
@@ -288,7 +329,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -288,7 +329,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
};
};
const row = {
const row = {
id,
id,
- project_id: null,
+ project_id: req.project_id || null,
owner_sub: owner,
owner_sub: owner,
name: rec.name,
name: rec.name,
status: rec.status,
status: rec.status,
@@ -328,12 +369,13 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -328,12 +369,13 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
const merged = { ...current, ...changes };
const merged = { ...current, ...changes };
delete merged.id; // an id never changes
delete merged.id; // an id never changes
delete merged.owner; // ownership never changes (a watch is the subject's)
delete merged.owner; // ownership never changes (a watch is the subject's)
- // project_id is not part of watch.watch-request@1: the row's own null is dropped, and a
- // caller that sends one fails the schema check below rather than being silently ignored.
- if (changes.project_id === undefined) delete merged.project_id;
+ // The row's own null project is dropped (null is "none" on the wire too); a project the patch names, or the
+ // one the watch already has, travels with the merged watch so a run-job action is judged against it.
+ if (merged.project_id === null) delete merged.project_id;
validateRequest(merged);
validateRequest(merged);
checkSecrets(row.owner_sub, merged);
checkSecrets(row.owner_sub, merged);
checkActorAgents(row.owner_sub, merged);
checkActorAgents(row.owner_sub, merged);
+ await checkProject(row.owner_sub, merged, changes.project_id !== undefined && changes.project_id !== row.project_id);
const sets = [];
const sets = [];
const args = { id: row.id, updated_at: now(), updated_by: ownerOf(principal) };
const args = { id: row.id, updated_at: now(), updated_by: ownerOf(principal) };
for (const [k, v] of Object.entries(changes)) {
for (const [k, v] of Object.entries(changes)) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,54 @@
@@ -0,0 +1,54 @@
+'use strict';
+/**
+ * Project delegation (ADR-003 amendment 2026-10-12), for a watch that names a project: Watch acts for the watch's owner
+ * inside that project.
+ *
+ * const projects = createProjects(client, { networkUrl }); // client: openvibe-sdk/core createClient with Watch's token
+ * await projects.check(projectId, subjectId) → { ok } (an admin or owner of the unarchived project), throws when Network cannot answer
+ *
+ * const run = createRunJobs(client, { baseUrl });
+ * await run.submit({ subjectId, projectId, job, idempotencyKey }) → { id, state, created }
+ *
+ * check() reads the person's membership from Network (GET /internal/projects/:project/members/:subject, capability
+ * network.project.member.read) when a watch is saved with a project. It is a courtesy to the person, who learns at
+ * once that they cannot use that project; OpenVibe.Run asks Network again before every job, which is the check that
+ * counts. submit() POSTs a run.job-create-request@1 body to Run with Watch's own token (run.job.submit, audience
+ * openvibe.run), the owner in X-OV-Subject and the project in X-OV-Project: the job is billed to the project, the
+ * person is its requester and Run records Watch as via. The call is made once; the action worker retries on its own
+ * backoff. Refusals arrive as openvibe-sdk OpenVibeError (status, code).
+ */
+
+const ADMIN_ROLES = new Set(['admin', 'owner']);
+
+function createProjects(client, { networkUrl }) {
+ async function check(projectId, subjectId) {
+ let out;
+ try {
+ out = await client.json({
+ service: 'network', baseUrl: networkUrl, audience: 'openvibe.network', method: 'GET',
+ path: `/internal/projects/${encodeURIComponent(projectId)}/members/${encodeURIComponent(subjectId)}`, retries: 0,
+ });
+ } catch (err) {
+ if (err && err.status === 404) return { ok: false };
+ throw err;
+ }
+ const ok = !!out && out.project_id === projectId && out.subject === subjectId && ADMIN_ROLES.has(out.role) && out.archived === false;
+ return { ok };
+ }
+ return { check };
+}
+
+function createRunJobs(client, { baseUrl }) {
+ async function submit({ subjectId, projectId, job, idempotencyKey }) {
+ const out = await client.json({
+ service: 'run', baseUrl, audience: 'openvibe.run', method: 'POST', path: '/api/v1/jobs',
+ headers: { 'X-OV-Subject': subjectId, 'X-OV-Project': projectId }, json: { ...job, idempotency_key: idempotencyKey },
+ retries: 0, idempotencyKey: false,
+ });
+ const j = (out && out.job) || {};
+ return { id: j.id || null, state: j.state || null, created: !!(out && out.created) };
+ }
+ return { submit };
+}
+
+module.exports = { createProjects, createRunJobs, ADMIN_ROLES };
@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
*/
*/
-async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null } = {}) {
+async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null, run = null, projects = null } = {}) {
// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real
// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real
// to fetch and verify against (no test touches the internet). Its URL is overridable via env.
// to fetch and verify against (no test touches the internet). Its URL is overridable via env.
const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });
const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });
@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l
@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l
});
});
// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
const testdb = await require('./db').testDb();
const testdb = await require('./db').testDb();
- const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(now ? { now } : {}) });
+ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(run ? { run } : {}), ...(projects ? { projects } : {}), ...(now ? { now } : {}) });
const base = `http://127.0.0.1:${h.server.address().port}`;
const base = `http://127.0.0.1:${h.server.address().port}`;
/** A session cookie for a person, as /auth/callback would have set it. */
/** A session cookie for a person, as /auth/callback would have set it. */
new file mode 100644
new file mode 100644
@@ -0,0 +1,230 @@
@@ -0,0 +1,230 @@
+'use strict';
+/**
+ * Project-scoped watches and the run-job action (project delegation, ADR-003 amendment 2026-10-12):
+ * - a watch names a project only when Network says its owner is an admin or owner of it (403 project.not_delegated,
+ * 503 when Network cannot answer), checked when the project is named or changed;
+ * - a run-job action needs a project and run.job-create-request@1's class and limits (artifact for function and
+ * code); the contract draft's runtime/command/workload are refused;
+ * - a fired run-job action submits one Run job for the owner in the project, the firing under args[firing_arg], with
+ * the idempotency key watch.<check run>.<action index>; Run's final refusals fail at once, anything else retries.
+ */
+const assert = require('assert');
+const { boot, request, site, suite, userToken, newUser, newAgent } = require('./helpers');
+
+const t = suite('project-run-jobs');
+let svc;
+let web;
+let price = '19.99';
+
+const person = { sub: newUser() };
+const viewer = { sub: newUser() };
+const agent = { sub: newAgent() };
+const P = 'prj_01JAB2C3D4E5F6G7H8J9K0MNPS';
+const Q = 'prj_01JAB2C3D4E5F6G7H8J9K0MNPT';
+
+// Network, as server/run-jobs.js createProjects would ask it: who is an admin or owner of which project.
+const checks = [];
+let networkDown = false;
+const projects = {
+ async check(project, subject) {
+ checks.push([project, subject]);
+ if (networkDown) throw Object.assign(new Error('Network answered 503'), { status: 503 });
+ return { ok: (project === P || project === Q) && subject === person.sub };
+ },
+};
+// OpenVibe.Run, as server/run-jobs.js createRunJobs would reach it: every submit recorded; runMode decides the answer.
+const submits = [];
+let runMode = 'ok';
+const run = {
+ async submit(req) {
+ submits.push(req);
+ if (runMode === 'busy') throw Object.assign(new Error('Network could not confirm'), { status: 503, code: 'project.check_unavailable' });
+ if (runMode === 'refuse') throw Object.assign(new Error('not delegated'), { status: 403, code: 'project.not_delegated', detail: 'a project is named only by a first-party service acting for an admin or owner' });
+ if (runMode === 'caps') throw Object.assign(new Error('over the cap'), { status: 422, code: 'run.limits.exceeded', detail: 'limits.wall_ms is at most 900000 ms for a project' });
+ return { id: `job_01JAB2C3D4E5F6G7H8J9K0MN${String(submits.length).padStart(2, '0')}`, state: 'queued', created: submits.length !== 3 };
+ },
+};
+
+const job = (over = {}) => ({
+ kind: 'run-job', class: 'function', artifact: { name: 'restock', version: '1.0.0' },
+ args: { sku: 'w-1' }, limits: { wall_ms: 60000, cpu_ms: 30000, mem_bytes: 134217728 }, egress: 'openvibe-only', ...over,
+});
+const def = (overrides = {}) => ({
+ name: 'Widget price',
+ source: { kind: 'http', url: `${web.origin}/price`, format: null },
+ cadence: { every_sec: 900 },
+ extraction: { kind: 'text' },
+ condition: { op: 'lt', value: 10 },
+ action: [job({ firing_arg: 'firing' })],
+ project_id: P,
+ ...overrides,
+});
+const refusedWith = (code, status) => (err) => { assert.strictEqual(err.code, code, err.message); if (status) assert.strictEqual(err.status, status); return true; };
+const acts = async (id) => await svc.db.prepare('SELECT * FROM action_runs WHERE watch_id = ? ORDER BY rid').all(id);
+const setPrice = async (v) => { await fetch(`${web.origin}/set?v=${v}`); };
+async function fire(w) {
+ await setPrice('19.99');
+ await svc.check.run(w.id, { trigger: 'schedule' });
+ await setPrice('7.50');
+ const out = await svc.check.run(w.id, { trigger: 'schedule' });
+ assert.strictEqual(out.run.state, 'condition_met');
+ return out;
+}
+
+t('boot', async () => {
+ web = await site({
+ '/price': () => ({ headers: { 'content-type': 'text/plain' }, body: price }),
+ '/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },
+ });
+ const notifications = { async push() { return { sent: true }; } };
+ svc = await boot({ notifications, run, projects });
+});
+
+t('a watch names a project its owner is an admin or owner of; anyone else, and an agent, is refused', async () => {
+ const w = await svc.registry.create(def(), person);
+ assert.strictEqual(w.project_id, P);
+ assert.deepStrictEqual(checks.pop(), [P, person.sub], 'checked with Network when named');
+ assert.strictEqual(svc.registry.view(w).project_id, P, 'the view carries it (watch.watch@1)');
+ await assert.rejects(() => svc.registry.create(def(), viewer), refusedWith('project.not_delegated', 403));
+ await assert.rejects(() => svc.registry.create(def({ project_id: 'prj_01JAB2C3D4E5F6G7H8J9K0MNPV' }), person), refusedWith('project.not_delegated', 403), 'a project they do not run');
+ await assert.rejects(() => svc.registry.create(def(), agent), refusedWith('project.not_delegated', 403), 'an agent never names a project');
+ networkDown = true;
+ try { await assert.rejects(() => svc.registry.create(def(), person), refusedWith('watch.project_check_unavailable', 503)); }
+ finally { networkDown = false; }
+ // A watch without run-job actions may name a project or not.
+ const plain = await svc.registry.create(def({ action: [{ kind: 'notification' }], project_id: undefined }), person);
+ assert.strictEqual(plain.project_id, null);
+});
+
+t('a run-job action needs a project and Run\'s job fields; the draft\'s fields are refused', async () => {
+ await assert.rejects(() => svc.registry.create(def({ project_id: undefined }), person), refusedWith('watch.project_required'));
+ await assert.rejects(() => svc.registry.create(def({ project_id: null }), person), refusedWith('watch.project_required'));
+ for (const [action, re] of [
+ [{ kind: 'run-job', runtime: 'function', command: 'node x.js' }, /runtime is not accepted/],
+ [job({ workload: {} }), /workload is not accepted/],
+ [job({ class: undefined }), /class is required/],
+ [job({ limits: undefined }), /limits is required/],
+ [job({ artifact: undefined }), /artifact is required for a function job/],
+ [job({ firing_arg: 'sku' }), /firing_arg names a key args already has/],
+ ]) {
+ await assert.rejects(() => svc.registry.create(def({ action: [action] }), person), (err) => {
+ assert.strictEqual(err.code, 'watch.invalid', err.message);
+ assert.match(err.message, re);
+ return true;
+ });
+ }
+ // A class that runs no artifact needs none (browser, linux…): Run places it when a node offers it.
+ const browser = await svc.registry.create(def({ action: [job({ class: 'browser', artifact: undefined })] }), person);
+ assert.ok(browser.id);
+});
+
+t('patch: clearing the project of a run-job watch is refused; a new project is checked, an untouched one is not', async () => {
+ const w = await svc.registry.create(def(), person);
+ await assert.rejects(() => svc.registry.patch(w.id, { project_id: null }, person), refusedWith('watch.project_required'));
+ const before = checks.length;
+ await svc.registry.patch(w.id, { name: 'renamed' }, person);
+ assert.strictEqual(checks.length, before, 'the project the watch already has is not asked about again (Run asks at every job)');
+ const moved = await svc.registry.patch(w.id, { project_id: Q }, person);
+ assert.strictEqual(moved.project_id, Q);
+ assert.deepStrictEqual(checks.pop(), [Q, person.sub]);
+ await assert.rejects(() => svc.registry.patch(w.id, { project_id: 'prj_01JAB2C3D4E5F6G7H8J9K0MNPV' }, person), refusedWith('project.not_delegated', 403));
+ const plain = await svc.registry.patch(w.id, { action: [{ kind: 'notification' }], project_id: null }, person);
+ assert.strictEqual(plain.project_id, null, 'without run-job actions the project may go');
+});
+
+t('the API answers the refusals with their own status', async () => {
+ const token = userToken(viewer.sub);
+ const r = await request(svc.base, 'POST', '/api/v1/watches', { token, body: def() });
+ assert.strictEqual(r.status, 403, r.text);
+ assert.strictEqual(r.body.code, 'project.not_delegated');
+ const missing = await request(svc.base, 'POST', '/api/v1/watches', { token: userToken(person.sub), body: def({ project_id: undefined }) });
+ assert.strictEqual(missing.status, 422, missing.text);
+ assert.strictEqual(missing.body.code, 'watch.project_required');
+});
+
+t('a fired run-job action submits one job for the owner in the project, with the firing, under one key', async () => {
+ submits.length = 0; runMode = 'busy';
+ const w = await svc.registry.create(def({ name: 'restock' }), person);
+ const out = await fire(w);
+ let r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 1, failed: 0 }, 'Run (or Network behind it) unavailable: retried');
+ runMode = 'ok';
+ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);
+ r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });
+ const [first, second] = submits;
+ assert.strictEqual(first.subjectId, person.sub, 'for the owner');
+ assert.strictEqual(first.projectId, P, 'in the watch\'s project');
+ assert.strictEqual(first.idempotencyKey, `watch.${out.run.id}.0`);
+ assert.strictEqual(second.idempotencyKey, first.idempotencyKey, 'the retry is the same job');
+ assert.deepStrictEqual({ ...first.job, args: undefined }, {
+ class: 'function', artifact: { name: 'restock', version: '1.0.0' }, limits: { wall_ms: 60000, cpu_ms: 30000, mem_bytes: 134217728 }, egress: 'openvibe-only', args: undefined,
+ }, 'run.job-create-request@1 fields only (no kind, no firing_arg)');
+ assert.strictEqual(first.job.args.sku, 'w-1');
+ const firing = first.job.args.firing;
+ assert.deepStrictEqual([firing.watch_id, firing.check_run_id, firing.value, firing.previous_value], [w.id, out.run.id, '7.50', '19.99']);
+ assert.ok(firing.watch_url.endsWith(`/watches/${w.id}`) && firing.observed_at);
+ const [row] = await acts(w.id);
+ assert.deepStrictEqual([row.state, row.last_error], ['sent', 'Run job job_01JAB2C3D4E5F6G7H8J9K0MN02']);
+ const req = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;
+ assert.strictEqual(req.project_id, P, 'the project is fixed when the action is queued');
+});
+
+t('Run\'s final refusals fail at once; an agent-owned or project-less row never reaches Run', async () => {
+ for (const mode of ['refuse', 'caps']) {
+ runMode = mode;
+ const w = await svc.registry.create(def({ name: `refused ${mode}` }), person);
+ await fire(w);
+ const r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 0, failed: 1 }, mode);
+ assert.match((await acts(w.id))[0].last_error, mode === 'refuse' ? /watch\.action_refused: project\.not_delegated/ : /watch\.action_refused: run\.limits\.exceeded/);
+ }
+ runMode = 'ok';
+ // A row queued for a watch whose project was cleared behind the registry's back (or never set): not sent.
+ const w = await svc.registry.create(def({ name: 'no project' }), person);
+ await svc.db.prepare('UPDATE watches SET project_id = NULL WHERE id = ?').run(w.id);
+ const before = submits.length;
+ await fire(w);
+ const [row] = await acts(w.id);
+ assert.deepStrictEqual([row.state, /project_required/.test(row.last_error)], ['unavailable', true]);
+ await svc.actions.tick();
+ assert.strictEqual(submits.length, before, 'nothing submitted');
+});
+
+t('server/run-jobs.js asks Network and Run as Watch, for the owner, once', async () => {
+ const { createProjects, createRunJobs } = require('../server/run-jobs');
+ const calls = [];
+ let answer = { project_id: P, subject: person.sub, role: 'admin', archived: false, added_at: null };
+ let fail = null;
+ const client = { async json(o) { calls.push(o); if (fail) throw fail; return answer; } };
+ const projectsLive = createProjects(client, { networkUrl: 'http://127.0.0.1:4000' });
+ assert.deepStrictEqual(await projectsLive.check(P, person.sub), { ok: true });
+ const c = calls[0];
+ assert.deepStrictEqual([c.service, c.baseUrl, c.audience, c.method, c.path, c.retries],
+ ['network', 'http://127.0.0.1:4000', 'openvibe.network', 'GET', `/internal/projects/${P}/members/${person.sub}`, 0]);
+ for (const [a, why] of [
+ [{ ...answer, role: 'developer' }, 'a developer'], [{ ...answer, role: 'viewer' }, 'a viewer'],
+ [{ ...answer, archived: true }, 'an archived project'], [{ ...answer, subject: viewer.sub }, 'an answer about someone else'],
+ ]) { answer = a; assert.deepStrictEqual(await projectsLive.check(P, person.sub), { ok: false }, why); }
+ answer = { ...answer, role: 'owner', archived: false, subject: person.sub };
+ assert.deepStrictEqual(await projectsLive.check(P, person.sub), { ok: true }, 'an owner');
+ fail = Object.assign(new Error('not found'), { status: 404 });
+ assert.deepStrictEqual(await projectsLive.check(P, person.sub), { ok: false }, 'a non-member');
+ fail = Object.assign(new Error('down'), { status: 503 });
+ await assert.rejects(() => projectsLive.check(P, person.sub), /down/, 'Network down: the caller answers 503');
+ fail = null;
+
+ answer = { created: true, job: { id: 'job_X', state: 'queued' } };
+ const runLive = createRunJobs(client, { baseUrl: 'http://127.0.0.1:4920' });
+ const outJob = await runLive.submit({ subjectId: person.sub, projectId: P, job: { class: 'function', limits: { wall_ms: 1, cpu_ms: 1, mem_bytes: 1 } }, idempotencyKey: 'watch.r.0' });
+ assert.deepStrictEqual(outJob, { id: 'job_X', state: 'queued', created: true });
+ const s = calls[calls.length - 1];
+ assert.deepStrictEqual([s.service, s.baseUrl, s.audience, s.method, s.path, s.retries, s.idempotencyKey],
+ ['run', 'http://127.0.0.1:4920', 'openvibe.run', 'POST', '/api/v1/jobs', 0, false]);
+ assert.deepStrictEqual(s.headers, { 'X-OV-Subject': person.sub, 'X-OV-Project': P });
+ assert.deepStrictEqual(s.json, { class: 'function', limits: { wall_ms: 1, cpu_ms: 1, mem_bytes: 1 }, idempotency_key: 'watch.r.0' });
+});
+
+t('done', async () => { await web.close(); await svc.stop(); });
+
+t.run();