openvibers/openvibe.run Commits
Run: a person's own machine runs only that person's jobs (owner-matched placement); a delegating service cancels and streams its jobs (#42)
Changes
@@@ -52,9 +52,10 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs- Project delegation (ADR-003 amendment 2026-10-12): a first-party service acting for a person (`X-OV-Subject`) mayname a project with `X-OV-Project`. Run asks Network whether the person is an admin or owner of the unarchivedproject (`network.project.member.read`, a yes cached 60 s), then the person is the requester and the service is-- stored as `via` (`migrations/0002_job_via.sql`). The service submits and reads only the jobs it started for that-- person; list, cancel and stream are `403 project.not_delegated`, and so is any other caller naming a project. Network-- unreachable is `503 project.check_unavailable`. Watch's `run-job` action is the first user.++ stored as `via` (`migrations/0002_job_via.sql`). The service submits, reads, cancels and gets a stream ticket for only++ the jobs it started for that person (another person's or another service's job is `404`); list and a delegated Bearer++ stream are `403 project.not_delegated`, and so is any other caller naming a project. Network unreachable is++ `503 project.check_unavailable`. Watch's `run-job` action and OpenVibe.Codes' coding jobs use it.- The `job_<ULID>` id (minted here, passed unchanged to Bot as `platform.job@1` `id`, so every metering reading ofthe job is `run:<id>:<n>`).- Idempotent submit: one row per `(project_id, idempotency_key)`, a repeat answering the first job, a different body@@@ -67,9 +68,11 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobsobjective: balanced }`, the contract's own default) and the project caps on a job's limits.- Placement (plan T14 R3, `server/dispatch/placement.js`): the job's requirements are stated by Run and the choiceamong offers is `openvibe-sdk/placement` `plan()` — Run writes no placer of its own (ADR-036). The job's-- `worker:<class>` capability is what a candidate must advertise, and a user-owned offer is eligible only when the-- job's requirements name `user-owned` (the contract's own rule; the SDK's fallback trust list would otherwise-- admit it).++ `worker:<class>` capability is what a candidate must advertise. A user-owned offer is a person's own machine++ (contracts 0.161.0): it is eligible only for that person's own job (requester `user:usr_…` equal to the offer's++ `owner`) whose requirements name `user-owned`. Network keeps such offers off the public list, so Run reads them,++ owners included, from `GET /internal/resources?kind=node&trust=user-owned` with its own client++ (`network.registry.read`); a user-owned offer on the public list is never used.- The poller (`server/jobs/poller.js`): the ttl sweep, the send, the mirror, the cancel-on-Bot and the job's stdoutinto the stream.@@@ -7,7 +7,7 @@"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db) + Valkey 9","port": 4920,"domain": "openvibe.run",-- "contracts": "openvibe-contracts v0.155.0",++ "contracts": "openvibe-contracts v0.161.0","sdk": "openvibe-sdk v0.44.0","shared": "openvibe-shared v4.1.3","packages": [@@@ -50,7 +50,8 @@"Bot owns the node link and the per-second metering of the jobs it carries (plan T14 L1); Run holds a Billing token of its own (billing.usage.record, audience openvibe.billing) used only for the external managed-browser provider's jobs","plan T14 R4a built, not deployed: the managed-browser provider adapter (server/dispatch/browser-provider.js; RUN_BROWSER_PROVIDER_URL / _API_KEY / _TIMEOUT_MS, off while unset) places class browser jobs on kind=provider offers named browser-provider, validates every answer against browser-provider.job-create-result@1 / job-state@1 and never adopts a malformed one; Run meters those jobs itself, the provider's claimed wall time clamped to Run's observed duration + 5 s, and relays the readings from run_usage_outbox (migrations/0003_usage_outbox.sql) to Billing (RUN_BILLING_URL; unset, they wait and the jobs still run)","truthful readiness (GET /api/ready: db required; valkey, Network JWKS, Events, Bot reachability and the poller's own status reported, never claimed) and the loopback-only /metrics gauges (jobs queued, active, ended, outbox backlog, streams)",-- "plan T11 step 4 built, not deployed: /llms.txt and /llms-full.txt (server/web/discovery.js, public pages and docs only, never the API, admin, a job or a stream) and IndexNow (INDEXNOW_KEY: serves GET /<key>.txt and pings the public pages once at boot; unset, the key route is 404 and nothing is sent)"++ "plan T11 step 4 built, not deployed: /llms.txt and /llms-full.txt (server/web/discovery.js, public pages and docs only, never the API, admin, a job or a stream) and IndexNow (INDEXNOW_KEY: serves GET /<key>.txt and pings the public pages once at boot; unset, the key route is 404 and nothing is sent)",++ "a person's own machine (user-owned offer read from Network's internal list) runs only that person's jobs naming user-owned; a delegating service cancels and streams (by ticket) the jobs it started for a person"],-- "updated": "2026-10-13"++ "updated": "2026-10-14"}@@@ -15,7 +15,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.155.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.161.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"@@@ -872,8 +872,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.155.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.155.0",++ "version": "0.161.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.161.0",++ "integrity": "sha512-2F1KBisvjjMbLI72G4TLsL1Z5aehb+69/nYQGFtF2tcmCeF5ecNvVJaJLRmHnB1cK/vJt+rxIZssE3xdyoFGrQ==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -26,7 +26,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.155.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.161.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"@@@ -178,7 +178,8 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now// ── POST /jobs/:id/cancel — run.job.cancel ─────────────────────────────────r.post('/jobs/:id/cancel', guard('run.job.cancel'), wrap(async (req, res) => {-- const row = await getJob(req, req.params.id);++ // A delegating service cancels only the jobs it started for that person (getOwn: via and requester match).++ const row = await getJob(req, req.params.id, { delegated: true });if (!row) fail(404, 'run.job_not_found', 'no such job');const at = now();const out = await db.tx(async (t) => {@@@ -196,7 +197,8 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now// ── POST /jobs/:id/stream/ticket — run.job.stream ──────────────────────────r.post('/jobs/:id/stream/ticket', guard('run.job.stream'), wrap(async (req, res) => {-- const row = await getJob(req, req.params.id);++ // A delegating service gets a ticket only for a job it started for that person; the ticket opens that one stream.++ const row = await getJob(req, req.params.id, { delegated: true });if (!row) fail(404, 'run.job_not_found', 'no such job');if (!tickets.get()) fail(503, 'run.stream.disabled', 'stream tickets are disabled: no stream key is configured (RUN_STREAM_PRIVATE_KEY or RUN_STREAM_KEY_FILE)');const { ticket, claims } = tickets.mint({ jobId: row.id, projectId: row.project_id });@@@ -40,6 +40,7 @@*/const contracts = require('openvibe-contracts');const { plan } = require('openvibe-sdk/placement');++const { createServiceTokenClient } = require('openvibe-sdk/auth');const { requirementsDefault } = require('../util');const { PROVIDER: BROWSER_PROVIDER } = require('./browser-provider');@@@ -60,6 +61,8 @@ class PlacementError extends Error {}const obj = (v) => (v && typeof v === 'object' && !Array.isArray(v) ? v : {});++/** The person a job was requested by (`user:usr_…`, as Run stores the requester), or null for a service or an app. */++const personOf = (requester) => (/^user:(usr_[0-9A-HJKMNP-TV-Z]{26})$/.exec(String(requester || '')) || [])[1] || null;/*** createPlacement({ config, fetchImpl, now, log, providers }) → { choose, offers, status }. `providers` maps@@@ -72,6 +75,39 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Datconst { offersTtlMs, timeoutMs } = config.dispatch;const caches = new Map(); // kind → { at, offers, fetched }const external = Object.entries(providers).filter(([name, classes]) => EXTERNAL_PROVIDERS.has(name) && Array.isArray(classes) && classes.length);++ // A person's own machines (contracts 0.161.0): never on the public list, read from Network's internal one with++ // network.registry.read, owner included. Without Run's client secret there are none to place on.++ const { clientId, clientSecret } = obj(config.oauth);++ const registryTokens = clientSecret++ ? createServiceTokenClient({ tokenUrl: `${config.network.internalUrl}/oauth/token`, clientId, clientSecret, audience: 'openvibe.network', scope: 'network.registry.read', fetch: fetchImpl, now })++ : null;++ let owned = { at: 0, offers: [] };++ /** Every live user-owned node offer that names its owner, from the internal list, cached like the public ones. */++ async function ownedOffers() {++ if (!registryTokens) return [];++ if (owned.at && now() - owned.at < offersTtlMs) return owned.offers;++ const href = `${config.network.internalUrl}/internal/resources?kind=node&trust=user-owned`;++ let res;++ try {++ res = await fetchImpl(href, { headers: { Accept: 'application/json', ...(await registryTokens.authHeaders()) }, signal: AbortSignal.timeout(timeoutMs) });++ } catch (e) {++ const why = e && (e.name === 'TimeoutError' || e.name === 'AbortError') ? `no answer within ${timeoutMs} ms` : 'unreachable';++ throw new PlacementError('run.dispatch.no_offers', `Network did not answer for the people's own machines (${why})`);++ }++ if (!res.ok) throw new PlacementError('run.dispatch.no_offers', `Network answered HTTP ${res.status} for the people's own machines`);++ const doc = await res.json().catch(() => null);++ const list = doc && Array.isArray(doc.offers) ? doc.offers : null;++ if (!list) throw new PlacementError('run.dispatch.no_offers', 'the internal offers answer carries no offers array');++ const good = list.filter((o) => {++ if (!contracts.validate('platform.resource-offer@1', o).valid) {++ log.warn(`[Run] offer ${obj(o).offer_id || '?'} does not match platform.resource-offer@1: dropped`);++ return false;++ }++ return o.trust === 'user-owned' && typeof o.owner === 'string' && o.kind === 'node' && typeof o.node_id === 'string' && o.node_id.length > 0;++ });++ owned = { at: now(), offers: good };++ return good;++ }/** The published offers of one kind, cached briefly. A Network that does not answer is a retryable error. */async function offers(kind = 'node') {@@@ -111,11 +147,19 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Datconst want = workerCapability(row.class);const requirements = { ...obj(row.job).requirements || requirementsDefault(row.class) };const userOwnedNamed = Array.isArray(requirements.trust) && requirements.trust.includes('user-owned');-- const trusted = (o) => (o.capabilities || []).includes(want) && (o.trust !== 'user-owned' || userOwnedNamed);++ // A user-owned offer is a person's own machine: eligible only for that person's own job, and only when the job++ // names user-owned (contracts 0.161.0). The public list carries none; one that slipped in is never used, since++ // only the internal read states the owner this rule needs.++ const trusted = (o) => (o.capabilities || []).includes(want) && o.trust !== 'user-owned';const candidates = (await offers('node')).filter((o) => o.kind === 'node'&& typeof o.node_id === 'string' && o.node_id.length > 0&& !EXTERNAL_PROVIDERS.has(o.provider)&& trusted(o));++ const person = personOf(row.requester);++ if (userOwnedNamed && person) {++ candidates.push(...(await ownedOffers()).filter((o) => o.owner === person && (o.capabilities || []).includes(want)++ && !EXTERNAL_PROVIDERS.has(o.provider)));++ }// An external provider that runs this class: its kind=provider offers are candidates too.const names = new Set(external.filter(([, classes]) => classes.includes(row.class)).map(([name]) => name));if (names.size) {@@@ -154,4 +198,4 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat};}--module.exports = { createPlacement, PlacementError, workerCapability, EXTERNAL_PROVIDERS };++module.exports = { createPlacement, PlacementError, workerCapability, EXTERNAL_PROVIDERS, personOf };@@@ -73,16 +73,28 @@ async function checks(t) {assert.strictEqual(project.status, 200, project.text);});-- await check('list, cancel and stream stay with project tokens', async () => {++ await check('the delegating service cancels and streams only the jobs it started for that person; list stays with project tokens', async () => {refused(await t.call('GET', '/api/v1/jobs', as(ada, P)), 'list');-- refused(await t.call('POST', `/api/v1/jobs/${jobId}/cancel`, as(ada, P)), 'cancel');-- refused(await t.call('POST', `/api/v1/jobs/${jobId}/stream/ticket`, as(ada, P)), 'stream ticket');const d = as(ada, P);-- const stream = await t.sse(`/api/v1/jobs/${jobId}/stream`, { headers: { Authorization: `Bearer ${d.token}`, ...d.headers } });-- assert.strictEqual(stream.status, 403, stream.text);-- assert.strictEqual(stream.json.code, 'project.not_delegated');-- const [row] = await t.db.many('SELECT state FROM run_jobs WHERE id = $1', [jobId]);-- assert.strictEqual(row.state, 'queued', 'the refused cancel changed nothing');++ const bearer = await t.sse(`/api/v1/jobs/${jobId}/stream`, { headers: { Authorization: `Bearer ${d.token}`, ...d.headers } });++ assert.strictEqual(bearer.status, 403, bearer.text);++ assert.strictEqual(bearer.json.code, 'project.not_delegated', 'a delegated bearer does not stream: it asks for a ticket');++ // Another person, another service: the same 404 as a job that does not exist, for the ticket and the cancel.++ for (const other of [as(bo, P), as(ada, P, { sub: 'svc:actor' })]) {++ for (const path of [`/api/v1/jobs/${jobId}/stream/ticket`, `/api/v1/jobs/${jobId}/cancel`]) {++ const r = await t.call('POST', path, other);++ assert.strictEqual(r.status, 404, `${path}: ${r.text}`);++ assert.strictEqual(r.json.code, 'run.job_not_found');++ }++ }++ const [still] = await t.db.many('SELECT state FROM run_jobs WHERE id = $1', [jobId]);++ assert.strictEqual(still.state, 'queued', 'the refused cancels changed nothing');++ const ticket = await t.call('POST', `/api/v1/jobs/${jobId}/stream/ticket`, as(ada, P));++ assert.strictEqual(ticket.status, 200, ticket.text);++ assert.ok(typeof ticket.json.ticket === 'string' && ticket.json.job_id === jobId, JSON.stringify(ticket.json));++ const cancelled = await t.call('POST', `/api/v1/jobs/${jobId}/cancel`, as(ada, P));++ assert.strictEqual(cancelled.status, 200, cancelled.text);++ assert.strictEqual(cancelled.json.state, 'cancelled');});await check('a developer, a viewer, a non-member and an archived project are refused alike', async () => {@@@ -397,23 +397,38 @@ async function main() {bot.devices.set(DEVICE, { classes: ['function'] });});-- await check('a user-owned node is eligible only when the job\'s requirements name it', async () => {++ await check('a person\'s own machine runs only that person\'s jobs, and only when the job names user-owned', async () => {bot.devices.set(OTHER_DEVICE, { classes: ['function'] });-- t.setOffers([offer({ offer_id: 'off_user', trust: 'user-owned', node_id: OTHER_DEVICE })]);-- const projectA = PROJECT();-- const plain = await submit({}, { project: projectA });-- const denied = (await t.dispatch.poll()).find((a) => a.job_id === plain.id);-- assert.strictEqual(denied.state, 'queued'); // user-owned is never a default: no eligible node-- const readA = await read(plain.id, projectA);-- assert.strictEqual(readA.json.state, 'queued');-- assert.strictEqual(readA.json.placement, null);-- await t.db.exec('UPDATE run_jobs SET created_at = created_at - 100000000 WHERE id = $1', [plain.id]);++ const ada = ids.newId('user');++ const bo = ids.newId('user');++ const P = PROJECT();++ t.network.members.set(`${P} ${ada}`, { role: 'owner' });++ t.network.members.set(`${P} ${bo}`, { role: 'owner' });++ t.setOffers([offer({ offer_id: 'off_user', trust: 'user-owned', owner: ada, provider: 'self', node_id: OTHER_DEVICE })]);++ const named = { requirements: { kind: 'run.function', mobility: 'job', latency_class: 'background', objective: 'balanced', trust: ['user-owned'] } };++ const asPerson = async (subject, over) => {++ const r = await t.call('POST', '/api/v1/jobs', { body: t.request(over), token: t.network.signService({ sub: 'svc:codes', cap: ['run.job.submit', 'run.job.read'] }), headers: { 'X-OV-Subject': subject, 'X-OV-Project': P } });++ assert.strictEqual(r.status, 201, r.text);++ return r.json.job;++ };++ const settle = async (job) => (await t.dispatch.poll()).find((a) => a.job_id === job.id);++ // A job that does not name user-owned never lands on anyone's machine, its owner's included.++ assert.strictEqual((await settle(await asPerson(ada, {}))).state, 'queued');++ // A job naming user-owned that is another person's, or no person's (a project's own token), waits too.++ assert.strictEqual((await settle(await asPerson(bo, named))).state, 'queued');++ assert.strictEqual((await settle(await submit(named, { project: PROJECT() }))).state, 'queued');++ // The owner's own job is placed on their machine, its owner read from Network's internal list.++ const reads = t.network.internalReads.n;++ const mine = await settle(await asPerson(ada, named));++ assert.strictEqual(mine.state, 'placed');++ assert.strictEqual(mine.node_id, OTHER_DEVICE);++ assert.ok(t.network.internalReads.n > reads, 'the owner came from the internal read');++ assert.ok(t.network.tokenRequests.scopes.includes('network.registry.read'));++ // The public list never carried it.++ const pub = await (await fetch(`${t.network.url}/api/v1/offers?kind=node`)).json();++ assert.strictEqual(pub.offers.length, 0);++ await t.db.exec(`UPDATE run_jobs SET created_at = created_at - 100000000 WHERE state = 'queued'`);await t.dispatch.sweep();-- const projectB = PROJECT();-- const named = await submit({ requirements: { kind: 'run.function', mobility: 'job', latency_class: 'background', objective: 'balanced', trust: ['user-owned', 'first-party'] } }, { project: projectB });-- const placed = (await t.dispatch.poll()).find((a) => a.job_id === named.id);-- assert.strictEqual(placed.state, 'placed');-- assert.strictEqual(placed.node_id, OTHER_DEVICE);bot.devices.delete(OTHER_DEVICE);});@@@ -530,9 +545,10 @@ async function main() {});await check('the Bot token is minted once and reused, never per call', async () => {-- assert.strictEqual(t.network.tokenRequests.n, 1, JSON.stringify(t.network.tokenRequests));-- assert.deepStrictEqual(t.network.tokenRequests.audiences, ['openvibe.bot']);-- assert.deepStrictEqual(t.network.tokenRequests.scopes, ['bot.job.dispatch']);++ // Other tokens are Run's Network reads (the membership check and the people's own machines).++ const req = t.network.tokenRequests;++ const bot = req.audiences.map((aud, i) => [aud, req.scopes[i]]).filter(([aud]) => aud === 'openvibe.bot');++ assert.deepStrictEqual(bot, [['openvibe.bot', 'bot.job.dispatch']], JSON.stringify(req));});await check('the stream accepts a Bearer service token holding run.job.stream, and only that', async () => {@@@ -40,6 +40,7 @@ async function startNetwork() {const issuer = 'http://network.test';const tokenRequests = { n: 0, audiences: [], scopes: [] };const offers = []; // what the registry publishes; a test replaces its contents (t.setOffers)++ const internalReads = { n: 0 };// Project memberships Network states (GET /internal/projects/:project/members/:subject, project delegation):// `${project} ${subject}` → { role, archived }; memberReads counts the reads, memberFail makes them answer 500.const members = new Map();@@@ -66,10 +67,22 @@ async function startNetwork() {if (req.method === 'GET' && req.url.startsWith('/api/v1/offers')) {const query = new URL(req.url, 'http://network.test').searchParams;const kind = query.get('kind');-- const docs = kind ? offers.filter((o) => o.kind === kind) : offers;++ // A person's own machine is never public (contracts 0.161.0): the public list leaves user-owned offers out.++ const docs = (kind ? offers.filter((o) => o.kind === kind) : offers).filter((o) => o.trust !== 'user-owned');res.writeHead(200, { 'Content-Type': 'application/json' });return res.end(JSON.stringify({ offers: docs, count: docs.length, generated_at: new Date().toISOString(), filters: kind ? { kind } : {} }));}++ // The internal list (network.registry.read), user-owned offers and their owners included.++ if (req.method === 'GET' && req.url.startsWith('/internal/resources')) {++ internalReads.n++;++ const token = String(req.headers.authorization || '').replace(/^Bearer /, '');++ const v = serviceAuth.verifyServiceToken(token, { publicKey: publicPem, issuer, audience: 'openvibe.network' });++ if (!v.ok || !(v.claims.cap || []).includes('network.registry.read')) { res.writeHead(403, { 'Content-Type': 'application/json' }); return res.end('{}'); }++ const query = new URL(req.url, 'http://network.test').searchParams;++ const docs = offers.filter((o) => (!query.get('kind') || o.kind === query.get('kind')) && (!query.get('trust') || o.trust === query.get('trust')));++ res.writeHead(200, { 'Content-Type': 'application/json' });++ return res.end(JSON.stringify({ offers: docs, count: docs.length, generated_at: new Date().toISOString(), filters: {} }));++ }if (req.method === 'POST' && req.url === '/oauth/token') {let body = '';req.on('data', (chunk) => { body += chunk; });@@@ -100,7 +113,7 @@ async function startNetwork() {});const url = await listen(server);return {-- url, issuer, publicPem, privatePem, tokenRequests, offers, members, memberState,++ url, issuer, publicPem, privatePem, tokenRequests, offers, members, memberState, internalReads,/** Publish exactly these platform.resource-offer@1 documents (the array keeps its identity). */setOffers(list) { offers.splice(0, offers.length, ...list); },close: () => new Promise((r) => server.close(r)),
@@ -52,9 +52,10 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs
@@ -52,9 +52,10 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs
- Project delegation (ADR-003 amendment 2026-10-12): a first-party service acting for a person (`X-OV-Subject`) may
- Project delegation (ADR-003 amendment 2026-10-12): a first-party service acting for a person (`X-OV-Subject`) may
name a project with `X-OV-Project`. Run asks Network whether the person is an admin or owner of the unarchived
name a project with `X-OV-Project`. Run asks Network whether the person is an admin or owner of the unarchived
project (`network.project.member.read`, a yes cached 60 s), then the person is the requester and the service is
project (`network.project.member.read`, a yes cached 60 s), then the person is the requester and the service is
- stored as `via` (`migrations/0002_job_via.sql`). The service submits and reads only the jobs it started for that
- person; list, cancel and stream are `403 project.not_delegated`, and so is any other caller naming a project. Network
- unreachable is `503 project.check_unavailable`. Watch's `run-job` action is the first user.
+ stored as `via` (`migrations/0002_job_via.sql`). The service submits, reads, cancels and gets a stream ticket for only
+ the jobs it started for that person (another person's or another service's job is `404`); list and a delegated Bearer
+ stream are `403 project.not_delegated`, and so is any other caller naming a project. Network unreachable is
+ `503 project.check_unavailable`. Watch's `run-job` action and OpenVibe.Codes' coding jobs use it.
- The `job_<ULID>` id (minted here, passed unchanged to Bot as `platform.job@1` `id`, so every metering reading of
- The `job_<ULID>` id (minted here, passed unchanged to Bot as `platform.job@1` `id`, so every metering reading of
the job is `run:<id>:<n>`).
the job is `run:<id>:<n>`).
- Idempotent submit: one row per `(project_id, idempotency_key)`, a repeat answering the first job, a different body
- Idempotent submit: one row per `(project_id, idempotency_key)`, a repeat answering the first job, a different body
@@ -67,9 +68,11 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs
@@ -67,9 +68,11 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs
objective: balanced }`, the contract's own default) and the project caps on a job's limits.
objective: balanced }`, the contract's own default) and the project caps on a job's limits.
- Placement (plan T14 R3, `server/dispatch/placement.js`): the job's requirements are stated by Run and the choice
- Placement (plan T14 R3, `server/dispatch/placement.js`): the job's requirements are stated by Run and the choice
among offers is `openvibe-sdk/placement` `plan()` — Run writes no placer of its own (ADR-036). The job's
among offers is `openvibe-sdk/placement` `plan()` — Run writes no placer of its own (ADR-036). The job's
- `worker:<class>` capability is what a candidate must advertise, and a user-owned offer is eligible only when the
- job's requirements name `user-owned` (the contract's own rule; the SDK's fallback trust list would otherwise
- admit it).
+ `worker:<class>` capability is what a candidate must advertise. A user-owned offer is a person's own machine
+ (contracts 0.161.0): it is eligible only for that person's own job (requester `user:usr_…` equal to the offer's
+ `owner`) whose requirements name `user-owned`. Network keeps such offers off the public list, so Run reads them,
+ owners included, from `GET /internal/resources?kind=node&trust=user-owned` with its own client
+ (`network.registry.read`); a user-owned offer on the public list is never used.
- The poller (`server/jobs/poller.js`): the ttl sweep, the send, the mirror, the cancel-on-Bot and the job's stdout
- The poller (`server/jobs/poller.js`): the ttl sweep, the send, the mirror, the cancel-on-Bot and the job's stdout
into the stream.
into the stream.
@@ -7,7 +7,7 @@
@@ -7,7 +7,7 @@
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db) + Valkey 9",
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db) + Valkey 9",
"port": 4920,
"port": 4920,
"domain": "openvibe.run",
"domain": "openvibe.run",
- "contracts": "openvibe-contracts v0.155.0",
+ "contracts": "openvibe-contracts v0.161.0",
"sdk": "openvibe-sdk v0.44.0",
"sdk": "openvibe-sdk v0.44.0",
"shared": "openvibe-shared v4.1.3",
"shared": "openvibe-shared v4.1.3",
"packages": [
"packages": [
@@ -50,7 +50,8 @@
@@ -50,7 +50,8 @@
"Bot owns the node link and the per-second metering of the jobs it carries (plan T14 L1); Run holds a Billing token of its own (billing.usage.record, audience openvibe.billing) used only for the external managed-browser provider's jobs",
"Bot owns the node link and the per-second metering of the jobs it carries (plan T14 L1); Run holds a Billing token of its own (billing.usage.record, audience openvibe.billing) used only for the external managed-browser provider's jobs",
"plan T14 R4a built, not deployed: the managed-browser provider adapter (server/dispatch/browser-provider.js; RUN_BROWSER_PROVIDER_URL / _API_KEY / _TIMEOUT_MS, off while unset) places class browser jobs on kind=provider offers named browser-provider, validates every answer against browser-provider.job-create-result@1 / job-state@1 and never adopts a malformed one; Run meters those jobs itself, the provider's claimed wall time clamped to Run's observed duration + 5 s, and relays the readings from run_usage_outbox (migrations/0003_usage_outbox.sql) to Billing (RUN_BILLING_URL; unset, they wait and the jobs still run)",
"plan T14 R4a built, not deployed: the managed-browser provider adapter (server/dispatch/browser-provider.js; RUN_BROWSER_PROVIDER_URL / _API_KEY / _TIMEOUT_MS, off while unset) places class browser jobs on kind=provider offers named browser-provider, validates every answer against browser-provider.job-create-result@1 / job-state@1 and never adopts a malformed one; Run meters those jobs itself, the provider's claimed wall time clamped to Run's observed duration + 5 s, and relays the readings from run_usage_outbox (migrations/0003_usage_outbox.sql) to Billing (RUN_BILLING_URL; unset, they wait and the jobs still run)",
"truthful readiness (GET /api/ready: db required; valkey, Network JWKS, Events, Bot reachability and the poller's own status reported, never claimed) and the loopback-only /metrics gauges (jobs queued, active, ended, outbox backlog, streams)",
"truthful readiness (GET /api/ready: db required; valkey, Network JWKS, Events, Bot reachability and the poller's own status reported, never claimed) and the loopback-only /metrics gauges (jobs queued, active, ended, outbox backlog, streams)",
- "plan T11 step 4 built, not deployed: /llms.txt and /llms-full.txt (server/web/discovery.js, public pages and docs only, never the API, admin, a job or a stream) and IndexNow (INDEXNOW_KEY: serves GET /<key>.txt and pings the public pages once at boot; unset, the key route is 404 and nothing is sent)"
+ "plan T11 step 4 built, not deployed: /llms.txt and /llms-full.txt (server/web/discovery.js, public pages and docs only, never the API, admin, a job or a stream) and IndexNow (INDEXNOW_KEY: serves GET /<key>.txt and pings the public pages once at boot; unset, the key route is 404 and nothing is sent)",
+ "a person's own machine (user-owned offer read from Network's internal list) runs only that person's jobs naming user-owned; a delegating service cancels and streams (by ticket) the jobs it started for a person"
],
],
- "updated": "2026-10-13"
+ "updated": "2026-10-14"
}
}
@@ -15,7 +15,7 @@
@@ -15,7 +15,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.155.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.161.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"
@@ -872,8 +872,9 @@
@@ -872,8 +872,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.155.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.155.0",
+ "version": "0.161.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.161.0",
+ "integrity": "sha512-2F1KBisvjjMbLI72G4TLsL1Z5aehb+69/nYQGFtF2tcmCeF5ecNvVJaJLRmHnB1cK/vJt+rxIZssE3xdyoFGrQ==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -26,7 +26,7 @@
@@ -26,7 +26,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.155.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.161.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"
@@ -178,7 +178,8 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
@@ -178,7 +178,8 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
// ── POST /jobs/:id/cancel — run.job.cancel ─────────────────────────────────
// ── POST /jobs/:id/cancel — run.job.cancel ─────────────────────────────────
r.post('/jobs/:id/cancel', guard('run.job.cancel'), wrap(async (req, res) => {
r.post('/jobs/:id/cancel', guard('run.job.cancel'), wrap(async (req, res) => {
- const row = await getJob(req, req.params.id);
+ // A delegating service cancels only the jobs it started for that person (getOwn: via and requester match).
+ const row = await getJob(req, req.params.id, { delegated: true });
if (!row) fail(404, 'run.job_not_found', 'no such job');
if (!row) fail(404, 'run.job_not_found', 'no such job');
const at = now();
const at = now();
const out = await db.tx(async (t) => {
const out = await db.tx(async (t) => {
@@ -196,7 +197,8 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
@@ -196,7 +197,8 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
// ── POST /jobs/:id/stream/ticket — run.job.stream ──────────────────────────
// ── POST /jobs/:id/stream/ticket — run.job.stream ──────────────────────────
r.post('/jobs/:id/stream/ticket', guard('run.job.stream'), wrap(async (req, res) => {
r.post('/jobs/:id/stream/ticket', guard('run.job.stream'), wrap(async (req, res) => {
- const row = await getJob(req, req.params.id);
+ // A delegating service gets a ticket only for a job it started for that person; the ticket opens that one stream.
+ const row = await getJob(req, req.params.id, { delegated: true });
if (!row) fail(404, 'run.job_not_found', 'no such job');
if (!row) fail(404, 'run.job_not_found', 'no such job');
if (!tickets.get()) fail(503, 'run.stream.disabled', 'stream tickets are disabled: no stream key is configured (RUN_STREAM_PRIVATE_KEY or RUN_STREAM_KEY_FILE)');
if (!tickets.get()) fail(503, 'run.stream.disabled', 'stream tickets are disabled: no stream key is configured (RUN_STREAM_PRIVATE_KEY or RUN_STREAM_KEY_FILE)');
const { ticket, claims } = tickets.mint({ jobId: row.id, projectId: row.project_id });
const { ticket, claims } = tickets.mint({ jobId: row.id, projectId: row.project_id });
@@ -40,6 +40,7 @@
@@ -40,6 +40,7 @@
*/
*/
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
const { plan } = require('openvibe-sdk/placement');
const { plan } = require('openvibe-sdk/placement');
+const { createServiceTokenClient } = require('openvibe-sdk/auth');
const { requirementsDefault } = require('../util');
const { requirementsDefault } = require('../util');
const { PROVIDER: BROWSER_PROVIDER } = require('./browser-provider');
const { PROVIDER: BROWSER_PROVIDER } = require('./browser-provider');
@@ -60,6 +61,8 @@ class PlacementError extends Error {
@@ -60,6 +61,8 @@ class PlacementError extends Error {
}
}
const obj = (v) => (v && typeof v === 'object' && !Array.isArray(v) ? v : {});
const obj = (v) => (v && typeof v === 'object' && !Array.isArray(v) ? v : {});
+/** The person a job was requested by (`user:usr_…`, as Run stores the requester), or null for a service or an app. */
+const personOf = (requester) => (/^user:(usr_[0-9A-HJKMNP-TV-Z]{26})$/.exec(String(requester || '')) || [])[1] || null;
/**
/**
* createPlacement({ config, fetchImpl, now, log, providers }) → { choose, offers, status }. `providers` maps
* createPlacement({ config, fetchImpl, now, log, providers }) → { choose, offers, status }. `providers` maps
@@ -72,6 +75,39 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -72,6 +75,39 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
const { offersTtlMs, timeoutMs } = config.dispatch;
const { offersTtlMs, timeoutMs } = config.dispatch;
const caches = new Map(); // kind → { at, offers, fetched }
const caches = new Map(); // kind → { at, offers, fetched }
const external = Object.entries(providers).filter(([name, classes]) => EXTERNAL_PROVIDERS.has(name) && Array.isArray(classes) && classes.length);
const external = Object.entries(providers).filter(([name, classes]) => EXTERNAL_PROVIDERS.has(name) && Array.isArray(classes) && classes.length);
+ // A person's own machines (contracts 0.161.0): never on the public list, read from Network's internal one with
+ // network.registry.read, owner included. Without Run's client secret there are none to place on.
+ const { clientId, clientSecret } = obj(config.oauth);
+ const registryTokens = clientSecret
+ ? createServiceTokenClient({ tokenUrl: `${config.network.internalUrl}/oauth/token`, clientId, clientSecret, audience: 'openvibe.network', scope: 'network.registry.read', fetch: fetchImpl, now })
+ : null;
+ let owned = { at: 0, offers: [] };
+ /** Every live user-owned node offer that names its owner, from the internal list, cached like the public ones. */
+ async function ownedOffers() {
+ if (!registryTokens) return [];
+ if (owned.at && now() - owned.at < offersTtlMs) return owned.offers;
+ const href = `${config.network.internalUrl}/internal/resources?kind=node&trust=user-owned`;
+ let res;
+ try {
+ res = await fetchImpl(href, { headers: { Accept: 'application/json', ...(await registryTokens.authHeaders()) }, signal: AbortSignal.timeout(timeoutMs) });
+ } catch (e) {
+ const why = e && (e.name === 'TimeoutError' || e.name === 'AbortError') ? `no answer within ${timeoutMs} ms` : 'unreachable';
+ throw new PlacementError('run.dispatch.no_offers', `Network did not answer for the people's own machines (${why})`);
+ }
+ if (!res.ok) throw new PlacementError('run.dispatch.no_offers', `Network answered HTTP ${res.status} for the people's own machines`);
+ const doc = await res.json().catch(() => null);
+ const list = doc && Array.isArray(doc.offers) ? doc.offers : null;
+ if (!list) throw new PlacementError('run.dispatch.no_offers', 'the internal offers answer carries no offers array');
+ const good = list.filter((o) => {
+ if (!contracts.validate('platform.resource-offer@1', o).valid) {
+ log.warn(`[Run] offer ${obj(o).offer_id || '?'} does not match platform.resource-offer@1: dropped`);
+ return false;
+ }
+ return o.trust === 'user-owned' && typeof o.owner === 'string' && o.kind === 'node' && typeof o.node_id === 'string' && o.node_id.length > 0;
+ });
+ owned = { at: now(), offers: good };
+ return good;
+ }
/** The published offers of one kind, cached briefly. A Network that does not answer is a retryable error. */
/** The published offers of one kind, cached briefly. A Network that does not answer is a retryable error. */
async function offers(kind = 'node') {
async function offers(kind = 'node') {
@@ -111,11 +147,19 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -111,11 +147,19 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
const want = workerCapability(row.class);
const want = workerCapability(row.class);
const requirements = { ...obj(row.job).requirements || requirementsDefault(row.class) };
const requirements = { ...obj(row.job).requirements || requirementsDefault(row.class) };
const userOwnedNamed = Array.isArray(requirements.trust) && requirements.trust.includes('user-owned');
const userOwnedNamed = Array.isArray(requirements.trust) && requirements.trust.includes('user-owned');
- const trusted = (o) => (o.capabilities || []).includes(want) && (o.trust !== 'user-owned' || userOwnedNamed);
+ // A user-owned offer is a person's own machine: eligible only for that person's own job, and only when the job
+ // names user-owned (contracts 0.161.0). The public list carries none; one that slipped in is never used, since
+ // only the internal read states the owner this rule needs.
+ const trusted = (o) => (o.capabilities || []).includes(want) && o.trust !== 'user-owned';
const candidates = (await offers('node')).filter((o) => o.kind === 'node'
const candidates = (await offers('node')).filter((o) => o.kind === 'node'
&& typeof o.node_id === 'string' && o.node_id.length > 0
&& typeof o.node_id === 'string' && o.node_id.length > 0
&& !EXTERNAL_PROVIDERS.has(o.provider)
&& !EXTERNAL_PROVIDERS.has(o.provider)
&& trusted(o));
&& trusted(o));
+ const person = personOf(row.requester);
+ if (userOwnedNamed && person) {
+ candidates.push(...(await ownedOffers()).filter((o) => o.owner === person && (o.capabilities || []).includes(want)
+ && !EXTERNAL_PROVIDERS.has(o.provider)));
+ }
// An external provider that runs this class: its kind=provider offers are candidates too.
// An external provider that runs this class: its kind=provider offers are candidates too.
const names = new Set(external.filter(([, classes]) => classes.includes(row.class)).map(([name]) => name));
const names = new Set(external.filter(([, classes]) => classes.includes(row.class)).map(([name]) => name));
if (names.size) {
if (names.size) {
@@ -154,4 +198,4 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -154,4 +198,4 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
};
};
}
}
-module.exports = { createPlacement, PlacementError, workerCapability, EXTERNAL_PROVIDERS };
+module.exports = { createPlacement, PlacementError, workerCapability, EXTERNAL_PROVIDERS, personOf };
@@ -73,16 +73,28 @@ async function checks(t) {
@@ -73,16 +73,28 @@ async function checks(t) {
assert.strictEqual(project.status, 200, project.text);
assert.strictEqual(project.status, 200, project.text);
});
});
- await check('list, cancel and stream stay with project tokens', async () => {
+ await check('the delegating service cancels and streams only the jobs it started for that person; list stays with project tokens', async () => {
refused(await t.call('GET', '/api/v1/jobs', as(ada, P)), 'list');
refused(await t.call('GET', '/api/v1/jobs', as(ada, P)), 'list');
- refused(await t.call('POST', `/api/v1/jobs/${jobId}/cancel`, as(ada, P)), 'cancel');
- refused(await t.call('POST', `/api/v1/jobs/${jobId}/stream/ticket`, as(ada, P)), 'stream ticket');
const d = as(ada, P);
const d = as(ada, P);
- const stream = await t.sse(`/api/v1/jobs/${jobId}/stream`, { headers: { Authorization: `Bearer ${d.token}`, ...d.headers } });
- assert.strictEqual(stream.status, 403, stream.text);
- assert.strictEqual(stream.json.code, 'project.not_delegated');
- const [row] = await t.db.many('SELECT state FROM run_jobs WHERE id = $1', [jobId]);
- assert.strictEqual(row.state, 'queued', 'the refused cancel changed nothing');
+ const bearer = await t.sse(`/api/v1/jobs/${jobId}/stream`, { headers: { Authorization: `Bearer ${d.token}`, ...d.headers } });
+ assert.strictEqual(bearer.status, 403, bearer.text);
+ assert.strictEqual(bearer.json.code, 'project.not_delegated', 'a delegated bearer does not stream: it asks for a ticket');
+ // Another person, another service: the same 404 as a job that does not exist, for the ticket and the cancel.
+ for (const other of [as(bo, P), as(ada, P, { sub: 'svc:actor' })]) {
+ for (const path of [`/api/v1/jobs/${jobId}/stream/ticket`, `/api/v1/jobs/${jobId}/cancel`]) {
+ const r = await t.call('POST', path, other);
+ assert.strictEqual(r.status, 404, `${path}: ${r.text}`);
+ assert.strictEqual(r.json.code, 'run.job_not_found');
+ }
+ }
+ const [still] = await t.db.many('SELECT state FROM run_jobs WHERE id = $1', [jobId]);
+ assert.strictEqual(still.state, 'queued', 'the refused cancels changed nothing');
+ const ticket = await t.call('POST', `/api/v1/jobs/${jobId}/stream/ticket`, as(ada, P));
+ assert.strictEqual(ticket.status, 200, ticket.text);
+ assert.ok(typeof ticket.json.ticket === 'string' && ticket.json.job_id === jobId, JSON.stringify(ticket.json));
+ const cancelled = await t.call('POST', `/api/v1/jobs/${jobId}/cancel`, as(ada, P));
+ assert.strictEqual(cancelled.status, 200, cancelled.text);
+ assert.strictEqual(cancelled.json.state, 'cancelled');
});
});
await check('a developer, a viewer, a non-member and an archived project are refused alike', async () => {
await check('a developer, a viewer, a non-member and an archived project are refused alike', async () => {
@@ -397,23 +397,38 @@ async function main() {
@@ -397,23 +397,38 @@ async function main() {
bot.devices.set(DEVICE, { classes: ['function'] });
bot.devices.set(DEVICE, { classes: ['function'] });
});
});
- await check('a user-owned node is eligible only when the job\'s requirements name it', async () => {
+ await check('a person\'s own machine runs only that person\'s jobs, and only when the job names user-owned', async () => {
bot.devices.set(OTHER_DEVICE, { classes: ['function'] });
bot.devices.set(OTHER_DEVICE, { classes: ['function'] });
- t.setOffers([offer({ offer_id: 'off_user', trust: 'user-owned', node_id: OTHER_DEVICE })]);
- const projectA = PROJECT();
- const plain = await submit({}, { project: projectA });
- const denied = (await t.dispatch.poll()).find((a) => a.job_id === plain.id);
- assert.strictEqual(denied.state, 'queued'); // user-owned is never a default: no eligible node
- const readA = await read(plain.id, projectA);
- assert.strictEqual(readA.json.state, 'queued');
- assert.strictEqual(readA.json.placement, null);
- await t.db.exec('UPDATE run_jobs SET created_at = created_at - 100000000 WHERE id = $1', [plain.id]);
+ const ada = ids.newId('user');
+ const bo = ids.newId('user');
+ const P = PROJECT();
+ t.network.members.set(`${P} ${ada}`, { role: 'owner' });
+ t.network.members.set(`${P} ${bo}`, { role: 'owner' });
+ t.setOffers([offer({ offer_id: 'off_user', trust: 'user-owned', owner: ada, provider: 'self', node_id: OTHER_DEVICE })]);
+ const named = { requirements: { kind: 'run.function', mobility: 'job', latency_class: 'background', objective: 'balanced', trust: ['user-owned'] } };
+ const asPerson = async (subject, over) => {
+ const r = await t.call('POST', '/api/v1/jobs', { body: t.request(over), token: t.network.signService({ sub: 'svc:codes', cap: ['run.job.submit', 'run.job.read'] }), headers: { 'X-OV-Subject': subject, 'X-OV-Project': P } });
+ assert.strictEqual(r.status, 201, r.text);
+ return r.json.job;
+ };
+ const settle = async (job) => (await t.dispatch.poll()).find((a) => a.job_id === job.id);
+ // A job that does not name user-owned never lands on anyone's machine, its owner's included.
+ assert.strictEqual((await settle(await asPerson(ada, {}))).state, 'queued');
+ // A job naming user-owned that is another person's, or no person's (a project's own token), waits too.
+ assert.strictEqual((await settle(await asPerson(bo, named))).state, 'queued');
+ assert.strictEqual((await settle(await submit(named, { project: PROJECT() }))).state, 'queued');
+ // The owner's own job is placed on their machine, its owner read from Network's internal list.
+ const reads = t.network.internalReads.n;
+ const mine = await settle(await asPerson(ada, named));
+ assert.strictEqual(mine.state, 'placed');
+ assert.strictEqual(mine.node_id, OTHER_DEVICE);
+ assert.ok(t.network.internalReads.n > reads, 'the owner came from the internal read');
+ assert.ok(t.network.tokenRequests.scopes.includes('network.registry.read'));
+ // The public list never carried it.
+ const pub = await (await fetch(`${t.network.url}/api/v1/offers?kind=node`)).json();
+ assert.strictEqual(pub.offers.length, 0);
+ await t.db.exec(`UPDATE run_jobs SET created_at = created_at - 100000000 WHERE state = 'queued'`);
await t.dispatch.sweep();
await t.dispatch.sweep();
- const projectB = PROJECT();
- const named = await submit({ requirements: { kind: 'run.function', mobility: 'job', latency_class: 'background', objective: 'balanced', trust: ['user-owned', 'first-party'] } }, { project: projectB });
- const placed = (await t.dispatch.poll()).find((a) => a.job_id === named.id);
- assert.strictEqual(placed.state, 'placed');
- assert.strictEqual(placed.node_id, OTHER_DEVICE);
bot.devices.delete(OTHER_DEVICE);
bot.devices.delete(OTHER_DEVICE);
});
});
@@ -530,9 +545,10 @@ async function main() {
@@ -530,9 +545,10 @@ async function main() {
});
});
await check('the Bot token is minted once and reused, never per call', async () => {
await check('the Bot token is minted once and reused, never per call', async () => {
- assert.strictEqual(t.network.tokenRequests.n, 1, JSON.stringify(t.network.tokenRequests));
- assert.deepStrictEqual(t.network.tokenRequests.audiences, ['openvibe.bot']);
- assert.deepStrictEqual(t.network.tokenRequests.scopes, ['bot.job.dispatch']);
+ // Other tokens are Run's Network reads (the membership check and the people's own machines).
+ const req = t.network.tokenRequests;
+ const bot = req.audiences.map((aud, i) => [aud, req.scopes[i]]).filter(([aud]) => aud === 'openvibe.bot');
+ assert.deepStrictEqual(bot, [['openvibe.bot', 'bot.job.dispatch']], JSON.stringify(req));
});
});
await check('the stream accepts a Bearer service token holding run.job.stream, and only that', async () => {
await check('the stream accepts a Bearer service token holding run.job.stream, and only that', async () => {
@@ -40,6 +40,7 @@ async function startNetwork() {
@@ -40,6 +40,7 @@ async function startNetwork() {
const issuer = 'http://network.test';
const issuer = 'http://network.test';
const tokenRequests = { n: 0, audiences: [], scopes: [] };
const tokenRequests = { n: 0, audiences: [], scopes: [] };
const offers = []; // what the registry publishes; a test replaces its contents (t.setOffers)
const offers = []; // what the registry publishes; a test replaces its contents (t.setOffers)
+ const internalReads = { n: 0 };
// Project memberships Network states (GET /internal/projects/:project/members/:subject, project delegation):
// Project memberships Network states (GET /internal/projects/:project/members/:subject, project delegation):
// `${project} ${subject}` → { role, archived }; memberReads counts the reads, memberFail makes them answer 500.
// `${project} ${subject}` → { role, archived }; memberReads counts the reads, memberFail makes them answer 500.
const members = new Map();
const members = new Map();
@@ -66,10 +67,22 @@ async function startNetwork() {
@@ -66,10 +67,22 @@ async function startNetwork() {
if (req.method === 'GET' && req.url.startsWith('/api/v1/offers')) {
if (req.method === 'GET' && req.url.startsWith('/api/v1/offers')) {
const query = new URL(req.url, 'http://network.test').searchParams;
const query = new URL(req.url, 'http://network.test').searchParams;
const kind = query.get('kind');
const kind = query.get('kind');
- const docs = kind ? offers.filter((o) => o.kind === kind) : offers;
+ // A person's own machine is never public (contracts 0.161.0): the public list leaves user-owned offers out.
+ const docs = (kind ? offers.filter((o) => o.kind === kind) : offers).filter((o) => o.trust !== 'user-owned');
res.writeHead(200, { 'Content-Type': 'application/json' });
res.writeHead(200, { 'Content-Type': 'application/json' });
return res.end(JSON.stringify({ offers: docs, count: docs.length, generated_at: new Date().toISOString(), filters: kind ? { kind } : {} }));
return res.end(JSON.stringify({ offers: docs, count: docs.length, generated_at: new Date().toISOString(), filters: kind ? { kind } : {} }));
}
}
+ // The internal list (network.registry.read), user-owned offers and their owners included.
+ if (req.method === 'GET' && req.url.startsWith('/internal/resources')) {
+ internalReads.n++;
+ const token = String(req.headers.authorization || '').replace(/^Bearer /, '');
+ const v = serviceAuth.verifyServiceToken(token, { publicKey: publicPem, issuer, audience: 'openvibe.network' });
+ if (!v.ok || !(v.claims.cap || []).includes('network.registry.read')) { res.writeHead(403, { 'Content-Type': 'application/json' }); return res.end('{}'); }
+ const query = new URL(req.url, 'http://network.test').searchParams;
+ const docs = offers.filter((o) => (!query.get('kind') || o.kind === query.get('kind')) && (!query.get('trust') || o.trust === query.get('trust')));
+ res.writeHead(200, { 'Content-Type': 'application/json' });
+ return res.end(JSON.stringify({ offers: docs, count: docs.length, generated_at: new Date().toISOString(), filters: {} }));
+ }
if (req.method === 'POST' && req.url === '/oauth/token') {
if (req.method === 'POST' && req.url === '/oauth/token') {
let body = '';
let body = '';
req.on('data', (chunk) => { body += chunk; });
req.on('data', (chunk) => { body += chunk; });
@@ -100,7 +113,7 @@ async function startNetwork() {
@@ -100,7 +113,7 @@ async function startNetwork() {
});
});
const url = await listen(server);
const url = await listen(server);
return {
return {
- url, issuer, publicPem, privatePem, tokenRequests, offers, members, memberState,
+ url, issuer, publicPem, privatePem, tokenRequests, offers, members, memberState, internalReads,
/** Publish exactly these platform.resource-offer@1 documents (the array keeps its identity). */
/** Publish exactly these platform.resource-offer@1 documents (the array keeps its identity). */
setOffers(list) { offers.splice(0, offers.length, ...list); },
setOffers(list) { offers.splice(0, offers.length, ...list); },
close: () => new Promise((r) => server.close(r)),
close: () => new Promise((r) => server.close(r)),