openvibers/openvibe.examples Commits
tools-job-proof (WS-L task 4): an SDK-only service client submits a converter job, drops and reattaches its progress stream with Last-Event-ID, downloads the Media-stored result and finds its tools.job.created/started/succeeded events in the Events store; --result for the scheduled run; tested against the mock platform
Changes
@@@ -164,6 +164,34 @@ nine steps passed, `1` when one failed, `2` when it refused to start.It does not yet cover webhook delivery to an external endpoint (that needs a public https host)or publishing an app release in OpenVibe.Codes.++### Tools job proof++++`scripts/tools-job-proof.js` (`npm run tools-job-proof`) is the scheduled end-to-end proof of++OpenVibe.Tools jobs (roadmap WS-L task 4), again with the SDK only. A **service principal** runs it,++because sandbox jobs are never announced to Events or copied to Media:++++1. **token**: client credentials for `tools.job.create` and `tools.job.read` (`openvibe.tools`) and++ `events.event.read` (`openvibe.events`).++2. **cursor**: the head of the Events store for `tools.job.*`, before anything is submitted.++3. **submit**: `img.process` converts a small generated PNG to WebP.++4. **reattach**: the progress stream is dropped after its first event and reattached with++ `Last-Event-ID`, as the Tools UI does after a reload. Every later event must arrive once, in++ order, up to `job.succeeded`.++5. **result**: the job reads back `succeeded`, and its file is stored in Media (`storage: "media"`++ with a media id). The download must be a WebP that matches the listed sha256. Tools keeps no++ local copy of a Media-stored result.++6. **events**: `tools.job.created`, `tools.job.started` and `tools.job.succeeded` for the job are++ in the Events store.++++```bash++OV_CLIENT_ID=… OV_CLIENT_SECRET=… npm run tools-job-proof [-- --result last.json]++```++++`OV_OAUTH_CLIENT_ID` and `OV_OAUTH_CLIENT_SECRET` also work; Network's service-principal setup++writes those names. OpenVibe.Host runs it every six hours as the `probe` principal++(`openvibe-toolsjob.timer`). CI runs it against the mock platform (`test/tools-job-proof.test.js`),++and it refuses to run in CI against production.++## What still does not work- **No consent screen.** Network's account chooser names the app but does not list the@@@ -13,7 +13,8 @@"e2e": "node --env-file-if-exists=.env scripts/e2e.js","new-app": "node --env-file-if-exists=.env scripts/new-sandbox-app.js","developer-path": "node --env-file-if-exists=.env scripts/developer-path.js",-- "developer-path:mock": "node scripts/developer-path-mock.js"++ "developer-path:mock": "node scripts/developer-path-mock.js",++ "tools-job-proof": "node --env-file-if-exists=.env scripts/tools-job-proof.js"},"engines": {"node": ">=22"nnew file mode 100644@@@ -0,0 +1,219 @@++#!/usr/bin/env node++'use strict';++/**++ * npm run tools-job-proof: the scheduled end-to-end proof of OpenVibe.Tools jobs (roadmap WS-L task 4).++ * A first-party client that uses the public API and openvibe-sdk only, with a service principal's++ * client credentials (production environment: sandbox jobs are never announced or copied to Media):++ *++ * 1. token client credentials for tools.job.create + tools.job.read (openvibe.tools) and++ * events.event.read (openvibe.events)++ * 2. cursor the head of the Events store for tools.job.*, taken before anything is submitted++ * 3. submit a converter job: img.process, convert a small generated PNG to WebP++ * 4. reattach follow the job's progress stream, drop it after the first event and reattach with++ * Last-Event-ID the way the Tools UI does after a reload: every later event once, in++ * order, up to job.succeeded++ * 5. result the durable job reads back succeeded; its result file is stored in OpenVibe.Media++ * (storage "media" with a media id) and downloads as a WebP with the listed sha256++ * (Tools serves a Media-stored result from Media only: no local copy is kept)++ * 6. events tools.job.created, tools.job.started and tools.job.succeeded for this job are in the++ * Events store, after the cursor++ *++ * OV_CLIENT_ID=… OV_CLIENT_SECRET=… npm run tools-job-proof [-- --result <file>]++ *++ * OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET (what Network's service-principal setup writes) work too.++ * Optional: OV_NETWORK_URL, OV_TOOLS_JOBS_URL. Refuses to run in CI against the production Network; CI++ * runs it against openvibe-sdk/testing's mock platform (test/tools-job-proof.test.js). Never prints the++ * secret or a token. --result writes step names and outcomes only (OpenVibe.Host openvibe-toolsjob.timer).++ * Exit 0 when every step passed, 1 when one failed, 2 when it refused to start.++ */++const crypto = require('node:crypto');++const fs = require('node:fs');++const zlib = require('node:zlib');++const { createClient, isOpenVibeError } = require('openvibe-sdk/core');++const { createServiceTokenClient } = require('openvibe-sdk/auth');++const { createJobsClient } = require('openvibe-sdk/jobs');++const { createEventsClient } = require('openvibe-sdk/events');++++const PRODUCTION_NETWORK = 'https://openvibe.network';++const STEPS = ['token', 'cursor', 'submit', 'reattach', 'result', 'events'];++const EVENT_TYPES = ['tools.job.created', 'tools.job.started', 'tools.job.succeeded'];++const STEP_TIMEOUT_MS = 180000;++++const describeError = (err) => (isOpenVibeError(err) || (err && err.code && err.status !== undefined)++ ? `${err.code}${err.status ? ` (${err.status})` : ''}${err.detail ? `: ${err.detail}` : ''}${err.requestId ? ` request=${err.requestId}` : ''}`++ : String((err && err.message) || err));++const sleep = (ms) => new Promise((r) => setTimeout(r, ms));++++function timeout(promise, what) {++ let timer;++ return Promise.race([promise, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`${what} took longer than ${STEP_TIMEOUT_MS / 1000} s`)), STEP_TIMEOUT_MS); })])++ .finally(() => clearTimeout(timer));++}++++/** A small valid RGB PNG (a gradient), made here so the proof needs no file. */++function tinyPng(w = 16, h = 16) {++ const table = Array.from({ length: 256 }, (_, n) => { let c = n; for (let k = 0; k < 8; k++) c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1; return c >>> 0; });++ const crc = (buf) => { let c = 0xffffffff; for (const b of buf) c = table[(c ^ b) & 0xff] ^ (c >>> 8); return (c ^ 0xffffffff) >>> 0; };++ const chunk = (type, data) => {++ const body = Buffer.concat([Buffer.from(type, 'ascii'), data]);++ const len = Buffer.alloc(4); len.writeUInt32BE(data.length);++ const sum = Buffer.alloc(4); sum.writeUInt32BE(crc(body));++ return Buffer.concat([len, body, sum]);++ };++ const ihdr = Buffer.alloc(13);++ ihdr.writeUInt32BE(w, 0); ihdr.writeUInt32BE(h, 4); ihdr[8] = 8; ihdr[9] = 2; // 8-bit RGB++ const raw = Buffer.alloc((w * 3 + 1) * h);++ for (let y = 0; y < h; y++) for (let x = 0; x < w; x++) raw.set([x * 16, y * 16, 160], y * (w * 3 + 1) + 1 + x * 3);++ return Buffer.concat([Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]), chunk('IHDR', ihdr), chunk('IDAT', zlib.deflateSync(raw)), chunk('IEND', Buffer.alloc(0))]);++}++++const isWebp = (b) => b.length > 12 && b.toString('ascii', 0, 4) === 'RIFF' && b.toString('ascii', 8, 12) === 'WEBP';++++async function runToolsJobProof({ network = PRODUCTION_NETWORK, toolsUrl, clientId, clientSecret, fetch: fetchImpl = globalThis.fetch, log = (m) => console.log(m), eventsWaitMs = 90000, pollMs = 3000, reconnectDelayMs } = {}) {++ network = network.replace(/\/+$/, '');++ const say = (m) => log(` ${m}`);++ const st = { jobs: null, events: null, cursor: 0, jobId: null, job: null };++ const steps = [];++ let failed = false;++++ async function step(name, fn) {++ log(`[${steps.length + 1}/${STEPS.length}] ${name}`);++ if (failed) { steps.push({ name, ok: false, skipped: true }); log(' skipped (an earlier step failed)\n'); return; }++ try {++ await timeout(fn(), name);++ steps.push({ name, ok: true });++ log(' ok\n');++ } catch (err) {++ failed = true;++ steps.push({ name, ok: false, error: describeError(err) });++ log(` FAILED: ${describeError(err)}\n`);++ }++ }++++ await step('token', async () => {++ const tokens = createServiceTokenClient({++ network, clientId, clientSecret, fetch: fetchImpl,++ scope: { 'openvibe.tools': 'tools.job.create tools.job.read', 'openvibe.events': 'events.event.read' },++ });++ // Ask for both now, so a missing grant fails here and not halfway through.++ for (const audience of ['openvibe.tools', 'openvibe.events']) {++ const t = await tokens.getToken({ audience });++ if (!t) throw new Error(`no token for ${audience}`);++ }++ const client = createClient({ network, fetch: fetchImpl, tokenProvider: tokens });++ st.jobs = createJobsClient(client, toolsUrl ? { baseUrl: toolsUrl } : undefined);++ st.events = createEventsClient(client);++ say('tokens for openvibe.tools (tools.job.create, tools.job.read) and openvibe.events (events.event.read)');++ });++++ await step('cursor', async () => {++ const page = await st.events.pull({ topic: 'tools.job.*', afterSeq: Number.MAX_SAFE_INTEGER, limit: 1 });++ st.cursor = Number(page.latest_seq) || 0;++ say(`Events head: seq ${st.cursor}`);++ });++++ await step('submit', async () => {++ const png = tinyPng();++ const { job, replayed } = await st.jobs.submit({++ type: 'img.process', input: { tool: 'convert', format: 'webp' },++ files: [{ name: 'tools-job-proof.png', data: png, type: 'image/png' }],++ idempotencyKey: `proof-${crypto.randomUUID()}`,++ });++ if (replayed) throw new Error('a fresh Idempotency-Key was answered as a replay');++ st.jobId = job.id;++ say(`submitted ${job.id} (${job.state}), ${png.length} bytes of PNG`);++ });++++ await step('reattach', async () => {++ // First connection: one event, then the stream is dropped (a reload, a lost connection).++ let first = null;++ for await (const e of st.jobs.events(st.jobId, { reconnectDelayMs })) { first = e; break; }++ if (!first || first.id == null) throw new Error('the progress stream sent no event with an id');++ say(`first stream: ${first.event} (id ${first.id}), then dropped`);++ // Reattach from the last id seen: only later events, each once, in order, up to the end.++ const later = [];++ for await (const e of st.jobs.events(st.jobId, { lastEventId: first.id, reconnectDelayMs })) later.push(e);++ const ids = later.map((e) => Number(e.id));++ if (ids.some((id, i) => !(id > (i ? ids[i - 1] : Number(first.id))))) throw new Error(`reattached stream replayed or reordered events: ${[first.id, ...ids].join(', ')}`);++ const last = later.length ? later[later.length - 1] : first;++ if (last.event !== 'job.succeeded') throw new Error(`the job ended with ${last.event}${last.job && last.job.error ? ` (${last.job.error.code})` : ''}`);++ say(`reattached with Last-Event-ID ${first.id}: ${later.map((e) => e.event).join(' → ') || '(finished already, 204)'}`);++ });++++ await step('result', async () => {++ const job = await st.jobs.get(st.jobId);++ if (!job) throw new Error('the job is gone (404)');++ if (job.state !== 'succeeded') throw new Error(`the job reads back ${job.state}`);++ const f = job.result && job.result.files && job.result.files[0];++ if (!f) throw new Error('the job has no result file');++ if (f.storage !== 'media' || !f.media || !f.media.media_id) throw new Error(`the result is stored "${f.storage}", not in OpenVibe.Media`);++ const res = await st.jobs.file(st.jobId, 0);++ const bytes = Buffer.from(await res.arrayBuffer());++ const sha = crypto.createHash('sha256').update(bytes).digest('hex');++ if (f.sha256 && sha !== f.sha256) throw new Error('the downloaded result does not match its sha256');++ if (!isWebp(bytes)) throw new Error('the result is not a WebP image');++ st.job = job;++ say(`succeeded; ${f.name} (${bytes.length} bytes, WebP) is Media object ${f.media.media_id}, downloaded with sha256 ${sha.slice(0, 16)}…`);++ });++++ await step('events', async () => {++ const found = new Map();++ let after = st.cursor;++ const deadline = Date.now() + eventsWaitMs;++ for (;;) {++ let page;++ do {++ page = await st.events.pull({ topic: 'tools.job.*', afterSeq: after, limit: 200 });++ for (const { seq, event } of page.events || []) {++ if (event && event.subject && event.subject.id === st.jobId && EVENT_TYPES.includes(event.event_type)) found.set(event.event_type, seq);++ }++ after = page.next_after_seq != null ? page.next_after_seq : after;++ } while ((page.events || []).length && after < page.latest_seq);++ if (EVENT_TYPES.every((t) => found.has(t))) break;++ if (Date.now() >= deadline) throw new Error(`not in the Events store after ${eventsWaitMs / 1000} s: ${EVENT_TYPES.filter((t) => !found.has(t)).join(', ')}`);++ await sleep(pollMs);++ }++ say(EVENT_TYPES.map((t) => `${t} seq ${found.get(t)}`).join(', '));++ });++++ const ok = steps.every((s) => s.ok);++ log(`${steps.filter((s) => s.ok).length}/${steps.length} steps ok`);++ return { ok, steps, jobId: st.jobId };++}++++/** The outcome as JSON (written to file.tmp, then renamed): step names and ok only. → the record */++function writeResult(file, result, { network, started = Date.now(), now = Date.now() } = {}) {++ const record = {++ ok: !!result.ok, network, started_at: new Date(started).toISOString(), finished_at: new Date(now).toISOString(),++ steps: result.steps.map((st) => ({ name: st.name, ok: !!st.ok, ...(st.skipped ? { skipped: true } : {}) })),++ };++ fs.writeFileSync(`${file}.tmp`, `${JSON.stringify(record, null, 2)}\n`, { mode: 0o644 });++ fs.renameSync(`${file}.tmp`, file);++ return record;++}++++async function main(argv = process.argv.slice(2), env = process.env) {++ const network = (env.OV_NETWORK_URL || PRODUCTION_NETWORK).replace(/\/+$/, '');++ if (env.CI && new URL(network).host === new URL(PRODUCTION_NETWORK).host) {++ console.error('tools-job-proof: refusing to run against the production Network in CI (CI is set). Nothing was sent.');++ return 2;++ }++ const clientId = env.OV_CLIENT_ID || env.OV_OAUTH_CLIENT_ID;++ const clientSecret = env.OV_CLIENT_SECRET || env.OV_OAUTH_CLIENT_SECRET;++ if (!clientId || !clientSecret) {++ console.error('tools-job-proof: missing OV_CLIENT_ID and OV_CLIENT_SECRET (or OV_OAUTH_CLIENT_ID and OV_OAUTH_CLIENT_SECRET). Nothing was sent.');++ return 2;++ }++ console.log(`OpenVibe Tools job proof against ${network}\n`);++ const started = Date.now();++ const result = await runToolsJobProof({ network, toolsUrl: env.OV_TOOLS_JOBS_URL || undefined, clientId, clientSecret });++ const out = argv.indexOf('--result');++ if (out >= 0 && argv[out + 1]) writeResult(argv[out + 1], result, { network, started });++ return result.ok ? 0 : 1;++}++++module.exports = { runToolsJobProof, writeResult, tinyPng, STEPS, EVENT_TYPES, PRODUCTION_NETWORK };++++if (require.main === module) {++ main().then((code) => { process.exitCode = code; }, (err) => { console.error(`tools-job-proof: ${describeError(err)}`); process.exitCode = 1; });++}nnew file mode 100644@@@ -0,0 +1,137 @@++'use strict';++/**++ * scripts/tools-job-proof.js against openvibe-sdk/testing's mock platform: a service principal with++ * tools.job.* and events.event.read, a mock img.process converter. The mock keeps job results locally++ * and has no Tools relay, so a thin layer plays Tools' production side: the job view says where++ * Tools put the result (storage "media"), and tools.job.created/started/succeeded reach the Events++ * store shortly after the submit, as the outbox relay posts them. Also: each check fails when its++ * part is missing, --result holds step names only, and the script refuses to start without++ * credentials or in CI against production. No network.++ */++const assert = require('node:assert/strict');++const fs = require('node:fs');++const os = require('node:os');++const path = require('node:path');++const zlib = require('node:zlib');++const { createMockPlatform } = require('openvibe-sdk/testing');++const { runToolsJobProof, writeResult, tinyPng, STEPS } = require('../scripts/tools-job-proof');++++globalThis.fetch = () => { throw new Error('network access in a test'); };++++const TOOLS = 'https://openvibe.tools';++const GRANTS = [['tools.job.create', 'openvibe.tools'], ['tools.job.read', 'openvibe.tools'], ['events.event.read', 'openvibe.events']];++const webp = (from) => Buffer.concat([Buffer.from('RIFF'), Buffer.alloc(4), Buffer.from('WEBPVP8 '), from.subarray(0, 32)]);++++function setup({ grants = GRANTS, media = true, relay = true } = {}) {++ const platform = createMockPlatform({++ jobs: {++ stepMs: 5,++ handlers: {++ 'img.process': async ({ input, files, progress }) => {++ await progress(50, 'converting');++ return { data: { format: input.format }, files: [{ name: 'tools-job-proof.webp', mime: 'image/webp', bytes: webp(files[0].bytes) }] };++ },++ },++ },++ });++ platform.addClient('probe', { secret: 'probe-secret', grants });++ const announce = (type, id) => platform.publishEvent({ event_type: type, source: 'tools', priority: type === 'tools.job.succeeded' ? 'important' : 'low', actor: { type: 'service', id: 'tools' }, subject: { type: 'job', id }, visibility: 'internal' }, 'svc:tools');++ const fetch = async (input, init = {}) => {++ const url = new URL(typeof input === 'string' ? input : input.url);++ const method = (init.method || 'GET').toUpperCase();++ const res = await platform.fetch(input, init);++ if (url.origin !== TOOLS) return res;++ if (relay && method === 'POST' && url.pathname === '/api/v1/jobs' && res.status === 202) {++ const { id } = await res.clone().json();++ announce('tools.job.created', id);++ setTimeout(() => { announce('tools.job.started', id); announce('tools.job.succeeded', id); }, 30);++ }++ if (media && method === 'GET' && /^\/api\/v1\/jobs\/job_[^/]+$/.test(url.pathname) && res.ok) {++ const job = await res.json();++ if (job.result) job.result.files = job.result.files.map((f, i) => ({ ...f, storage: 'media', media: { media_id: `med_proof${i}`, role: 'output' } }));++ return new Response(JSON.stringify(job), { status: res.status, headers: res.headers });++ }++ return res;++ };++ return { platform, fetch };++}++++const run = (fetch, extra = {}) => runToolsJobProof({ fetch, clientId: 'probe', clientSecret: 'probe-secret', log: () => {}, eventsWaitMs: 2000, pollMs: 20, reconnectDelayMs: 10, ...extra });++++(async () => {++ // The generated input is a real PNG: signature, IHDR 16x16 RGB, and IDAT that inflates to its rows.++ const png = tinyPng();++ assert.deepEqual([...png.subarray(0, 8)], [0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]);++ assert.equal(png.toString('ascii', 12, 16), 'IHDR');++ assert.equal(png.readUInt32BE(16), 16);++ const idatLen = png.readUInt32BE(33);++ assert.equal(png.toString('ascii', 37, 41), 'IDAT');++ assert.equal(zlib.inflateSync(png.subarray(41, 41 + idatLen)).length, (16 * 3 + 1) * 16);++++ // 1. Every step passes; the stream was dropped and reattached with Last-Event-ID.++ {++ const { platform, fetch } = setup();++ const result = await run(fetch);++ assert.deepEqual(result.steps.map((s) => [s.name, s.ok]), STEPS.map((n) => [n, true]), JSON.stringify(result.steps));++ assert.equal(result.ok, true);++ const streams = platform.stats.requests.filter((r) => r.url.endsWith('/events'));++ assert.equal(streams.length, 2, 'one stream, then one reattach');++ assert.equal(streams[1].headers['last-event-id'], '1');++ const file = platform.stats.requests.filter((r) => /\/files\/0/.test(r.url));++ assert.equal(file.length, 1, 'the result was downloaded');++ }++++ // 2. A result that stayed on the Tools host fails the result step; the rest is skipped.++ {++ const { fetch } = setup({ media: false });++ const result = await run(fetch);++ const r = result.steps.find((s) => s.name === 'result');++ assert.equal(r.ok, false);++ assert.match(r.error, /not in OpenVibe\.Media/);++ assert.deepEqual(result.steps.find((s) => s.name === 'events'), { name: 'events', ok: false, skipped: true });++ }++++ // 3. Events that never reach the store fail the events step.++ {++ const { fetch } = setup({ relay: false });++ const result = await run(fetch, { eventsWaitMs: 100 });++ assert.equal(result.ok, false);++ assert.match(result.steps.find((s) => s.name === 'events').error, /tools\.job\.created, tools\.job\.started, tools\.job\.succeeded/);++ }++++ // 4. A principal without events.event.read fails at the token step, before submitting anything.++ {++ const { platform, fetch } = setup({ grants: GRANTS.slice(0, 2) });++ const result = await run(fetch);++ assert.equal(result.steps[0].name, 'token');++ assert.equal(result.steps[0].ok, false);++ assert.equal(platform.stats.requests.filter((r) => r.method === 'POST' && r.url.endsWith('/api/v1/jobs')).length, 0);++ }++++ // 5. --result: step names and outcomes, no error text, id or secret.++ {++ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-toolsproof-'));++ const file = path.join(dir, 'last.json');++ const rec = writeResult(file, { ok: false, jobId: 'job_secretish', steps: [{ name: 'token', ok: true }, { name: 'cursor', ok: false, error: 'events.forbidden (403) request=req_1' }, { name: 'submit', ok: false, skipped: true }] }, { network: 'https://openvibe.network', started: 0, now: 1000 });++ const text = fs.readFileSync(file, 'utf8');++ assert.deepEqual(JSON.parse(text), rec);++ assert.deepEqual(rec.steps, [{ name: 'token', ok: true }, { name: 'cursor', ok: false }, { name: 'submit', ok: false, skipped: true }]);++ assert.doesNotMatch(text, /forbidden|req_1|job_secretish/);++ fs.rmSync(dir, { recursive: true, force: true });++ }++++ // 6. It refuses to start without credentials, and in CI against production.++ {++ const script = path.join(__dirname, '..', 'scripts', 'tools-job-proof.js');++ const { spawnSync } = require('node:child_process');++ const clean = Object.fromEntries(Object.entries(process.env).filter(([k]) => !/^(OV_|CI$)/.test(k)));++ const none = spawnSync(process.execPath, [script], { env: clean, encoding: 'utf8' });++ assert.equal(none.status, 2);++ assert.match(none.stderr, /missing OV_CLIENT_ID/);++ const ci = spawnSync(process.execPath, [script], { env: { ...clean, CI: 'true', OV_CLIENT_ID: 'x', OV_CLIENT_SECRET: 'y' }, encoding: 'utf8' });++ assert.equal(ci.status, 2);++ assert.match(ci.stderr, /refusing to run against the production Network in CI/);++ }++++ console.log('tools-job-proof: all checks passed');++})().catch((err) => { console.error(err); process.exitCode = 1; });
@@ -164,6 +164,34 @@ nine steps passed, `1` when one failed, `2` when it refused to start.
@@ -164,6 +164,34 @@ nine steps passed, `1` when one failed, `2` when it refused to start.
It does not yet cover webhook delivery to an external endpoint (that needs a public https host)
It does not yet cover webhook delivery to an external endpoint (that needs a public https host)
or publishing an app release in OpenVibe.Codes.
or publishing an app release in OpenVibe.Codes.
+### Tools job proof
+
+`scripts/tools-job-proof.js` (`npm run tools-job-proof`) is the scheduled end-to-end proof of
+OpenVibe.Tools jobs (roadmap WS-L task 4), again with the SDK only. A **service principal** runs it,
+because sandbox jobs are never announced to Events or copied to Media:
+
+1. **token**: client credentials for `tools.job.create` and `tools.job.read` (`openvibe.tools`) and
+ `events.event.read` (`openvibe.events`).
+2. **cursor**: the head of the Events store for `tools.job.*`, before anything is submitted.
+3. **submit**: `img.process` converts a small generated PNG to WebP.
+4. **reattach**: the progress stream is dropped after its first event and reattached with
+ `Last-Event-ID`, as the Tools UI does after a reload. Every later event must arrive once, in
+ order, up to `job.succeeded`.
+5. **result**: the job reads back `succeeded`, and its file is stored in Media (`storage: "media"`
+ with a media id). The download must be a WebP that matches the listed sha256. Tools keeps no
+ local copy of a Media-stored result.
+6. **events**: `tools.job.created`, `tools.job.started` and `tools.job.succeeded` for the job are
+ in the Events store.
+
+```bash
+OV_CLIENT_ID=… OV_CLIENT_SECRET=… npm run tools-job-proof [-- --result last.json]
+```
+
+`OV_OAUTH_CLIENT_ID` and `OV_OAUTH_CLIENT_SECRET` also work; Network's service-principal setup
+writes those names. OpenVibe.Host runs it every six hours as the `probe` principal
+(`openvibe-toolsjob.timer`). CI runs it against the mock platform (`test/tools-job-proof.test.js`),
+and it refuses to run in CI against production.
+
## What still does not work
## What still does not work
- **No consent screen.** Network's account chooser names the app but does not list the
- **No consent screen.** Network's account chooser names the app but does not list the
@@ -13,7 +13,8 @@
@@ -13,7 +13,8 @@
"e2e": "node --env-file-if-exists=.env scripts/e2e.js",
"e2e": "node --env-file-if-exists=.env scripts/e2e.js",
"new-app": "node --env-file-if-exists=.env scripts/new-sandbox-app.js",
"new-app": "node --env-file-if-exists=.env scripts/new-sandbox-app.js",
"developer-path": "node --env-file-if-exists=.env scripts/developer-path.js",
"developer-path": "node --env-file-if-exists=.env scripts/developer-path.js",
- "developer-path:mock": "node scripts/developer-path-mock.js"
+ "developer-path:mock": "node scripts/developer-path-mock.js",
+ "tools-job-proof": "node --env-file-if-exists=.env scripts/tools-job-proof.js"
},
},
"engines": {
"engines": {
"node": ">=22"
"node": ">=22"
new file mode 100644
new file mode 100644
@@ -0,0 +1,219 @@
@@ -0,0 +1,219 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * npm run tools-job-proof: the scheduled end-to-end proof of OpenVibe.Tools jobs (roadmap WS-L task 4).
+ * A first-party client that uses the public API and openvibe-sdk only, with a service principal's
+ * client credentials (production environment: sandbox jobs are never announced or copied to Media):
+ *
+ * 1. token client credentials for tools.job.create + tools.job.read (openvibe.tools) and
+ * events.event.read (openvibe.events)
+ * 2. cursor the head of the Events store for tools.job.*, taken before anything is submitted
+ * 3. submit a converter job: img.process, convert a small generated PNG to WebP
+ * 4. reattach follow the job's progress stream, drop it after the first event and reattach with
+ * Last-Event-ID the way the Tools UI does after a reload: every later event once, in
+ * order, up to job.succeeded
+ * 5. result the durable job reads back succeeded; its result file is stored in OpenVibe.Media
+ * (storage "media" with a media id) and downloads as a WebP with the listed sha256
+ * (Tools serves a Media-stored result from Media only: no local copy is kept)
+ * 6. events tools.job.created, tools.job.started and tools.job.succeeded for this job are in the
+ * Events store, after the cursor
+ *
+ * OV_CLIENT_ID=… OV_CLIENT_SECRET=… npm run tools-job-proof [-- --result <file>]
+ *
+ * OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET (what Network's service-principal setup writes) work too.
+ * Optional: OV_NETWORK_URL, OV_TOOLS_JOBS_URL. Refuses to run in CI against the production Network; CI
+ * runs it against openvibe-sdk/testing's mock platform (test/tools-job-proof.test.js). Never prints the
+ * secret or a token. --result writes step names and outcomes only (OpenVibe.Host openvibe-toolsjob.timer).
+ * Exit 0 when every step passed, 1 when one failed, 2 when it refused to start.
+ */
+const crypto = require('node:crypto');
+const fs = require('node:fs');
+const zlib = require('node:zlib');
+const { createClient, isOpenVibeError } = require('openvibe-sdk/core');
+const { createServiceTokenClient } = require('openvibe-sdk/auth');
+const { createJobsClient } = require('openvibe-sdk/jobs');
+const { createEventsClient } = require('openvibe-sdk/events');
+
+const PRODUCTION_NETWORK = 'https://openvibe.network';
+const STEPS = ['token', 'cursor', 'submit', 'reattach', 'result', 'events'];
+const EVENT_TYPES = ['tools.job.created', 'tools.job.started', 'tools.job.succeeded'];
+const STEP_TIMEOUT_MS = 180000;
+
+const describeError = (err) => (isOpenVibeError(err) || (err && err.code && err.status !== undefined)
+ ? `${err.code}${err.status ? ` (${err.status})` : ''}${err.detail ? `: ${err.detail}` : ''}${err.requestId ? ` request=${err.requestId}` : ''}`
+ : String((err && err.message) || err));
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+
+function timeout(promise, what) {
+ let timer;
+ return Promise.race([promise, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`${what} took longer than ${STEP_TIMEOUT_MS / 1000} s`)), STEP_TIMEOUT_MS); })])
+ .finally(() => clearTimeout(timer));
+}
+
+/** A small valid RGB PNG (a gradient), made here so the proof needs no file. */
+function tinyPng(w = 16, h = 16) {
+ const table = Array.from({ length: 256 }, (_, n) => { let c = n; for (let k = 0; k < 8; k++) c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1; return c >>> 0; });
+ const crc = (buf) => { let c = 0xffffffff; for (const b of buf) c = table[(c ^ b) & 0xff] ^ (c >>> 8); return (c ^ 0xffffffff) >>> 0; };
+ const chunk = (type, data) => {
+ const body = Buffer.concat([Buffer.from(type, 'ascii'), data]);
+ const len = Buffer.alloc(4); len.writeUInt32BE(data.length);
+ const sum = Buffer.alloc(4); sum.writeUInt32BE(crc(body));
+ return Buffer.concat([len, body, sum]);
+ };
+ const ihdr = Buffer.alloc(13);
+ ihdr.writeUInt32BE(w, 0); ihdr.writeUInt32BE(h, 4); ihdr[8] = 8; ihdr[9] = 2; // 8-bit RGB
+ const raw = Buffer.alloc((w * 3 + 1) * h);
+ for (let y = 0; y < h; y++) for (let x = 0; x < w; x++) raw.set([x * 16, y * 16, 160], y * (w * 3 + 1) + 1 + x * 3);
+ return Buffer.concat([Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]), chunk('IHDR', ihdr), chunk('IDAT', zlib.deflateSync(raw)), chunk('IEND', Buffer.alloc(0))]);
+}
+
+const isWebp = (b) => b.length > 12 && b.toString('ascii', 0, 4) === 'RIFF' && b.toString('ascii', 8, 12) === 'WEBP';
+
+async function runToolsJobProof({ network = PRODUCTION_NETWORK, toolsUrl, clientId, clientSecret, fetch: fetchImpl = globalThis.fetch, log = (m) => console.log(m), eventsWaitMs = 90000, pollMs = 3000, reconnectDelayMs } = {}) {
+ network = network.replace(/\/+$/, '');
+ const say = (m) => log(` ${m}`);
+ const st = { jobs: null, events: null, cursor: 0, jobId: null, job: null };
+ const steps = [];
+ let failed = false;
+
+ async function step(name, fn) {
+ log(`[${steps.length + 1}/${STEPS.length}] ${name}`);
+ if (failed) { steps.push({ name, ok: false, skipped: true }); log(' skipped (an earlier step failed)\n'); return; }
+ try {
+ await timeout(fn(), name);
+ steps.push({ name, ok: true });
+ log(' ok\n');
+ } catch (err) {
+ failed = true;
+ steps.push({ name, ok: false, error: describeError(err) });
+ log(` FAILED: ${describeError(err)}\n`);
+ }
+ }
+
+ await step('token', async () => {
+ const tokens = createServiceTokenClient({
+ network, clientId, clientSecret, fetch: fetchImpl,
+ scope: { 'openvibe.tools': 'tools.job.create tools.job.read', 'openvibe.events': 'events.event.read' },
+ });
+ // Ask for both now, so a missing grant fails here and not halfway through.
+ for (const audience of ['openvibe.tools', 'openvibe.events']) {
+ const t = await tokens.getToken({ audience });
+ if (!t) throw new Error(`no token for ${audience}`);
+ }
+ const client = createClient({ network, fetch: fetchImpl, tokenProvider: tokens });
+ st.jobs = createJobsClient(client, toolsUrl ? { baseUrl: toolsUrl } : undefined);
+ st.events = createEventsClient(client);
+ say('tokens for openvibe.tools (tools.job.create, tools.job.read) and openvibe.events (events.event.read)');
+ });
+
+ await step('cursor', async () => {
+ const page = await st.events.pull({ topic: 'tools.job.*', afterSeq: Number.MAX_SAFE_INTEGER, limit: 1 });
+ st.cursor = Number(page.latest_seq) || 0;
+ say(`Events head: seq ${st.cursor}`);
+ });
+
+ await step('submit', async () => {
+ const png = tinyPng();
+ const { job, replayed } = await st.jobs.submit({
+ type: 'img.process', input: { tool: 'convert', format: 'webp' },
+ files: [{ name: 'tools-job-proof.png', data: png, type: 'image/png' }],
+ idempotencyKey: `proof-${crypto.randomUUID()}`,
+ });
+ if (replayed) throw new Error('a fresh Idempotency-Key was answered as a replay');
+ st.jobId = job.id;
+ say(`submitted ${job.id} (${job.state}), ${png.length} bytes of PNG`);
+ });
+
+ await step('reattach', async () => {
+ // First connection: one event, then the stream is dropped (a reload, a lost connection).
+ let first = null;
+ for await (const e of st.jobs.events(st.jobId, { reconnectDelayMs })) { first = e; break; }
+ if (!first || first.id == null) throw new Error('the progress stream sent no event with an id');
+ say(`first stream: ${first.event} (id ${first.id}), then dropped`);
+ // Reattach from the last id seen: only later events, each once, in order, up to the end.
+ const later = [];
+ for await (const e of st.jobs.events(st.jobId, { lastEventId: first.id, reconnectDelayMs })) later.push(e);
+ const ids = later.map((e) => Number(e.id));
+ if (ids.some((id, i) => !(id > (i ? ids[i - 1] : Number(first.id))))) throw new Error(`reattached stream replayed or reordered events: ${[first.id, ...ids].join(', ')}`);
+ const last = later.length ? later[later.length - 1] : first;
+ if (last.event !== 'job.succeeded') throw new Error(`the job ended with ${last.event}${last.job && last.job.error ? ` (${last.job.error.code})` : ''}`);
+ say(`reattached with Last-Event-ID ${first.id}: ${later.map((e) => e.event).join(' → ') || '(finished already, 204)'}`);
+ });
+
+ await step('result', async () => {
+ const job = await st.jobs.get(st.jobId);
+ if (!job) throw new Error('the job is gone (404)');
+ if (job.state !== 'succeeded') throw new Error(`the job reads back ${job.state}`);
+ const f = job.result && job.result.files && job.result.files[0];
+ if (!f) throw new Error('the job has no result file');
+ if (f.storage !== 'media' || !f.media || !f.media.media_id) throw new Error(`the result is stored "${f.storage}", not in OpenVibe.Media`);
+ const res = await st.jobs.file(st.jobId, 0);
+ const bytes = Buffer.from(await res.arrayBuffer());
+ const sha = crypto.createHash('sha256').update(bytes).digest('hex');
+ if (f.sha256 && sha !== f.sha256) throw new Error('the downloaded result does not match its sha256');
+ if (!isWebp(bytes)) throw new Error('the result is not a WebP image');
+ st.job = job;
+ say(`succeeded; ${f.name} (${bytes.length} bytes, WebP) is Media object ${f.media.media_id}, downloaded with sha256 ${sha.slice(0, 16)}…`);
+ });
+
+ await step('events', async () => {
+ const found = new Map();
+ let after = st.cursor;
+ const deadline = Date.now() + eventsWaitMs;
+ for (;;) {
+ let page;
+ do {
+ page = await st.events.pull({ topic: 'tools.job.*', afterSeq: after, limit: 200 });
+ for (const { seq, event } of page.events || []) {
+ if (event && event.subject && event.subject.id === st.jobId && EVENT_TYPES.includes(event.event_type)) found.set(event.event_type, seq);
+ }
+ after = page.next_after_seq != null ? page.next_after_seq : after;
+ } while ((page.events || []).length && after < page.latest_seq);
+ if (EVENT_TYPES.every((t) => found.has(t))) break;
+ if (Date.now() >= deadline) throw new Error(`not in the Events store after ${eventsWaitMs / 1000} s: ${EVENT_TYPES.filter((t) => !found.has(t)).join(', ')}`);
+ await sleep(pollMs);
+ }
+ say(EVENT_TYPES.map((t) => `${t} seq ${found.get(t)}`).join(', '));
+ });
+
+ const ok = steps.every((s) => s.ok);
+ log(`${steps.filter((s) => s.ok).length}/${steps.length} steps ok`);
+ return { ok, steps, jobId: st.jobId };
+}
+
+/** The outcome as JSON (written to file.tmp, then renamed): step names and ok only. → the record */
+function writeResult(file, result, { network, started = Date.now(), now = Date.now() } = {}) {
+ const record = {
+ ok: !!result.ok, network, started_at: new Date(started).toISOString(), finished_at: new Date(now).toISOString(),
+ steps: result.steps.map((st) => ({ name: st.name, ok: !!st.ok, ...(st.skipped ? { skipped: true } : {}) })),
+ };
+ fs.writeFileSync(`${file}.tmp`, `${JSON.stringify(record, null, 2)}\n`, { mode: 0o644 });
+ fs.renameSync(`${file}.tmp`, file);
+ return record;
+}
+
+async function main(argv = process.argv.slice(2), env = process.env) {
+ const network = (env.OV_NETWORK_URL || PRODUCTION_NETWORK).replace(/\/+$/, '');
+ if (env.CI && new URL(network).host === new URL(PRODUCTION_NETWORK).host) {
+ console.error('tools-job-proof: refusing to run against the production Network in CI (CI is set). Nothing was sent.');
+ return 2;
+ }
+ const clientId = env.OV_CLIENT_ID || env.OV_OAUTH_CLIENT_ID;
+ const clientSecret = env.OV_CLIENT_SECRET || env.OV_OAUTH_CLIENT_SECRET;
+ if (!clientId || !clientSecret) {
+ console.error('tools-job-proof: missing OV_CLIENT_ID and OV_CLIENT_SECRET (or OV_OAUTH_CLIENT_ID and OV_OAUTH_CLIENT_SECRET). Nothing was sent.');
+ return 2;
+ }
+ console.log(`OpenVibe Tools job proof against ${network}\n`);
+ const started = Date.now();
+ const result = await runToolsJobProof({ network, toolsUrl: env.OV_TOOLS_JOBS_URL || undefined, clientId, clientSecret });
+ const out = argv.indexOf('--result');
+ if (out >= 0 && argv[out + 1]) writeResult(argv[out + 1], result, { network, started });
+ return result.ok ? 0 : 1;
+}
+
+module.exports = { runToolsJobProof, writeResult, tinyPng, STEPS, EVENT_TYPES, PRODUCTION_NETWORK };
+
+if (require.main === module) {
+ main().then((code) => { process.exitCode = code; }, (err) => { console.error(`tools-job-proof: ${describeError(err)}`); process.exitCode = 1; });
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,137 @@
@@ -0,0 +1,137 @@
+'use strict';
+/**
+ * scripts/tools-job-proof.js against openvibe-sdk/testing's mock platform: a service principal with
+ * tools.job.* and events.event.read, a mock img.process converter. The mock keeps job results locally
+ * and has no Tools relay, so a thin layer plays Tools' production side: the job view says where
+ * Tools put the result (storage "media"), and tools.job.created/started/succeeded reach the Events
+ * store shortly after the submit, as the outbox relay posts them. Also: each check fails when its
+ * part is missing, --result holds step names only, and the script refuses to start without
+ * credentials or in CI against production. No network.
+ */
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const os = require('node:os');
+const path = require('node:path');
+const zlib = require('node:zlib');
+const { createMockPlatform } = require('openvibe-sdk/testing');
+const { runToolsJobProof, writeResult, tinyPng, STEPS } = require('../scripts/tools-job-proof');
+
+globalThis.fetch = () => { throw new Error('network access in a test'); };
+
+const TOOLS = 'https://openvibe.tools';
+const GRANTS = [['tools.job.create', 'openvibe.tools'], ['tools.job.read', 'openvibe.tools'], ['events.event.read', 'openvibe.events']];
+const webp = (from) => Buffer.concat([Buffer.from('RIFF'), Buffer.alloc(4), Buffer.from('WEBPVP8 '), from.subarray(0, 32)]);
+
+function setup({ grants = GRANTS, media = true, relay = true } = {}) {
+ const platform = createMockPlatform({
+ jobs: {
+ stepMs: 5,
+ handlers: {
+ 'img.process': async ({ input, files, progress }) => {
+ await progress(50, 'converting');
+ return { data: { format: input.format }, files: [{ name: 'tools-job-proof.webp', mime: 'image/webp', bytes: webp(files[0].bytes) }] };
+ },
+ },
+ },
+ });
+ platform.addClient('probe', { secret: 'probe-secret', grants });
+ const announce = (type, id) => platform.publishEvent({ event_type: type, source: 'tools', priority: type === 'tools.job.succeeded' ? 'important' : 'low', actor: { type: 'service', id: 'tools' }, subject: { type: 'job', id }, visibility: 'internal' }, 'svc:tools');
+ const fetch = async (input, init = {}) => {
+ const url = new URL(typeof input === 'string' ? input : input.url);
+ const method = (init.method || 'GET').toUpperCase();
+ const res = await platform.fetch(input, init);
+ if (url.origin !== TOOLS) return res;
+ if (relay && method === 'POST' && url.pathname === '/api/v1/jobs' && res.status === 202) {
+ const { id } = await res.clone().json();
+ announce('tools.job.created', id);
+ setTimeout(() => { announce('tools.job.started', id); announce('tools.job.succeeded', id); }, 30);
+ }
+ if (media && method === 'GET' && /^\/api\/v1\/jobs\/job_[^/]+$/.test(url.pathname) && res.ok) {
+ const job = await res.json();
+ if (job.result) job.result.files = job.result.files.map((f, i) => ({ ...f, storage: 'media', media: { media_id: `med_proof${i}`, role: 'output' } }));
+ return new Response(JSON.stringify(job), { status: res.status, headers: res.headers });
+ }
+ return res;
+ };
+ return { platform, fetch };
+}
+
+const run = (fetch, extra = {}) => runToolsJobProof({ fetch, clientId: 'probe', clientSecret: 'probe-secret', log: () => {}, eventsWaitMs: 2000, pollMs: 20, reconnectDelayMs: 10, ...extra });
+
+(async () => {
+ // The generated input is a real PNG: signature, IHDR 16x16 RGB, and IDAT that inflates to its rows.
+ const png = tinyPng();
+ assert.deepEqual([...png.subarray(0, 8)], [0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]);
+ assert.equal(png.toString('ascii', 12, 16), 'IHDR');
+ assert.equal(png.readUInt32BE(16), 16);
+ const idatLen = png.readUInt32BE(33);
+ assert.equal(png.toString('ascii', 37, 41), 'IDAT');
+ assert.equal(zlib.inflateSync(png.subarray(41, 41 + idatLen)).length, (16 * 3 + 1) * 16);
+
+ // 1. Every step passes; the stream was dropped and reattached with Last-Event-ID.
+ {
+ const { platform, fetch } = setup();
+ const result = await run(fetch);
+ assert.deepEqual(result.steps.map((s) => [s.name, s.ok]), STEPS.map((n) => [n, true]), JSON.stringify(result.steps));
+ assert.equal(result.ok, true);
+ const streams = platform.stats.requests.filter((r) => r.url.endsWith('/events'));
+ assert.equal(streams.length, 2, 'one stream, then one reattach');
+ assert.equal(streams[1].headers['last-event-id'], '1');
+ const file = platform.stats.requests.filter((r) => /\/files\/0/.test(r.url));
+ assert.equal(file.length, 1, 'the result was downloaded');
+ }
+
+ // 2. A result that stayed on the Tools host fails the result step; the rest is skipped.
+ {
+ const { fetch } = setup({ media: false });
+ const result = await run(fetch);
+ const r = result.steps.find((s) => s.name === 'result');
+ assert.equal(r.ok, false);
+ assert.match(r.error, /not in OpenVibe\.Media/);
+ assert.deepEqual(result.steps.find((s) => s.name === 'events'), { name: 'events', ok: false, skipped: true });
+ }
+
+ // 3. Events that never reach the store fail the events step.
+ {
+ const { fetch } = setup({ relay: false });
+ const result = await run(fetch, { eventsWaitMs: 100 });
+ assert.equal(result.ok, false);
+ assert.match(result.steps.find((s) => s.name === 'events').error, /tools\.job\.created, tools\.job\.started, tools\.job\.succeeded/);
+ }
+
+ // 4. A principal without events.event.read fails at the token step, before submitting anything.
+ {
+ const { platform, fetch } = setup({ grants: GRANTS.slice(0, 2) });
+ const result = await run(fetch);
+ assert.equal(result.steps[0].name, 'token');
+ assert.equal(result.steps[0].ok, false);
+ assert.equal(platform.stats.requests.filter((r) => r.method === 'POST' && r.url.endsWith('/api/v1/jobs')).length, 0);
+ }
+
+ // 5. --result: step names and outcomes, no error text, id or secret.
+ {
+ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-toolsproof-'));
+ const file = path.join(dir, 'last.json');
+ const rec = writeResult(file, { ok: false, jobId: 'job_secretish', steps: [{ name: 'token', ok: true }, { name: 'cursor', ok: false, error: 'events.forbidden (403) request=req_1' }, { name: 'submit', ok: false, skipped: true }] }, { network: 'https://openvibe.network', started: 0, now: 1000 });
+ const text = fs.readFileSync(file, 'utf8');
+ assert.deepEqual(JSON.parse(text), rec);
+ assert.deepEqual(rec.steps, [{ name: 'token', ok: true }, { name: 'cursor', ok: false }, { name: 'submit', ok: false, skipped: true }]);
+ assert.doesNotMatch(text, /forbidden|req_1|job_secretish/);
+ fs.rmSync(dir, { recursive: true, force: true });
+ }
+
+ // 6. It refuses to start without credentials, and in CI against production.
+ {
+ const script = path.join(__dirname, '..', 'scripts', 'tools-job-proof.js');
+ const { spawnSync } = require('node:child_process');
+ const clean = Object.fromEntries(Object.entries(process.env).filter(([k]) => !/^(OV_|CI$)/.test(k)));
+ const none = spawnSync(process.execPath, [script], { env: clean, encoding: 'utf8' });
+ assert.equal(none.status, 2);
+ assert.match(none.stderr, /missing OV_CLIENT_ID/);
+ const ci = spawnSync(process.execPath, [script], { env: { ...clean, CI: 'true', OV_CLIENT_ID: 'x', OV_CLIENT_SECRET: 'y' }, encoding: 'utf8' });
+ assert.equal(ci.status, 2);
+ assert.match(ci.stderr, /refusing to run against the production Network in CI/);
+ }
+
+ console.log('tools-job-proof: all checks passed');
+})().catch((err) => { console.error(err); process.exitCode = 1; });