openvibers/openvibe.run Commits
Manage compute pools with strategies, autoscaling and durable rental caps (#206)
Changes
@@@ -256,3 +256,7 @@ until Run has a product surface; the vhost below is installed then.- openvibe-sdk: v0.94.0- openvibe-shared: v4.153.0<!-- versions:end -->++++Compute placement, pool autoscaling, rental caps and the operator view are documented in++[docs/compute-broker.md](docs/compute-broker.md). Rentals default to quote-only; keys and caps++are never inferred from available credentials.@@@ -8,11 +8,11 @@"port": 4920,"domain": "openvibe.run","contracts": "openvibe-contracts v0.338.0",-- "sdk": "openvibe-sdk v0.94.0",++ "sdk": "openvibe-sdk v0.100.0","shared": "openvibe-shared v4.153.0","packages": ["openvibe-contracts v0.338.0",-- "openvibe-sdk v0.94.0",++ "openvibe-sdk v0.100.0","openvibe-shared v4.153.0"],"capabilities": [nnew file mode 100644@@@ -0,0 +1,98 @@++# Computer pools and compute broker++++Run owns scheduling, session leases, autoscaling and rental accounting. Computer is the front door;++Exchange shows GPU quotes; AI retains GPU inference runtimes and its nano-USD ledger.++++## Existing machines and strategies++++Set `RUN_COMPUTE_POOLS` to a JSON object keyed by pool name, and `RUN_COMPUTE_INVENTORY` to an++operator-controlled JSON object keyed by Bot machine id. These are non-secret configuration.++For example, a browser pool has `kind: "browser"`, `strategy: "automatic"`, `currency: "EUR"`,++`min_warm: 1`, `max_machines: 3`, `slots_per_machine: 8`, `target_wait_ms: 30000`,++`idle_grace_ms: 600000`, `max_hourly: 0.10`, `providers: ["hetzner", "ovh"]`.++A machine entry has `pool`, `tier` (`home`, `dedicated`, `community`, `partner`), `ram_mb`,++`measured_peak_ram_mb`, `region`, `hourly`, `currency`, optionally measured `latency_ms`,++`queue_ms`, `gpu_mb`, and `egress: "ovh-tunnel"`. Own-machine identity always comes from Bot.++Public offers cannot claim an OpenVibe trust tier. Missing RAM/price means ineligible, not free.++++`POST /api/v1/sessions?pool=browser&strategy=cheapest` keeps the released request contract.++Strategies are `automatic`, `cheapest`, `fastest`, `private`, `own`; the request overrides the++pool default. The pool and strategy are part of idempotency and persist on the session.++`GET /api/v1/sessions/:id?explain=1` adds the recorded decision, tier, price and reasons for its++owner. Normal reads remain contract-compatible. Fastest uses measured latency; an unmeasured++host has no advantage. Region and privacy constraints never relax to satisfy a queue.++++Automatic walks own → home → dedicated → community → partner, rejecting health, capacity,++queue-wait and price failures first. New rentals are ranked with their configured disk/network++allowance; unavailable keys or exhausted provider caps do not hide eligible alternatives.++Home public egress requires an OVH tunnel. RAM packing++uses at least 300 MB/browser and 2048 MB/desktop, subtracts 1 GB for the system, and respects++Bot's slot ceiling. Configure measured peaks from load tests; these defaults are not measurements.++++For jobs, `RUN_COMPUTE_JOB_POOL` applies the pool policy to Fabric offers after the existing++SDK planner's hard checks. Inventory `free_ram_mb` and `free_slots` cover offers whose public++projection omits capacity. Job `requirements.objective` selects cheapest, lowest-latency or++private; an explicit `node` remains pinned. Draining rentals reject both sessions and jobs.++The dispatch result retains placement reasons and estimated hourly cost.++++## Rental switches: all default off++++- `RUN_COMPUTE_ENABLED=1` enables pool reconciliation; `RUN_COMPUTE_DRY_RUN=0` permits provision.++ Dry run defaults on. Cleanup of already-created rentals continues when either switch is disabled.++- `RUN_COMPUTE_OWNER` is the Network subject whose Bot host list the operator manages.++- `RUN_COMPUTE_OPERATORS` is a comma-separated allowlist of Network user subject ids.++- For each provider prefix `RUN_COMPUTE_HETZNER_`, `RUN_COMPUTE_OVH_`, `RUN_COMPUTE_VAST_`,++ `RUN_COMPUTE_RUNPOD_`, set `KEY`, `MONTHLY_CAP`, `CURRENCY`, and `IMAGE`. Cap is a decimal++ amount in that currency, default `0`. Image must be a prepared, isolated Link worker image.++ Never put credentials in the inventory, pool JSON or image. Enrollment is an operator step.++- Also set `EXTRA_HOURLY` to a conservative bound for disk/egress and `COSTS_BOUNDED=1` only++ after enforcing that bound in the worker/network. Without both, live provisioning is refused.++ Public catalogue prices alone do not bound bandwidth or persistent-volume charges.++- OVH additionally needs `PROJECT`, `APPLICATION_SECRET`, and `CONSUMER_KEY`; `KEY` is the++ application key. This adapter uses EU API/IE catalogue, EUR including catalogue tax. Hetzner++ Cloud quotes EUR including tax and provisions IPv6 only. Vast and RunPod quote USD.++ Do not mix currencies or guess an exchange rate.++- GPU inference should use AI's existing `ai.gpus` settings and ledger. Leave Run's Vast and++ RunPod caps at zero unless intentionally operating a separate worker pool with a separate budget.++++No-key quotes: auction, OVH public catalogue, Vast, RunPod, and community offer reads, subject to++upstream availability. Hetzner Cloud needs a token; OVH needs credentials to resolve live project++flavors. Auction purchase is quote-only pending cancellation/liability implementation. Community++inference offers are not hourly machine rentals; use AI's existing routing or an existing Bot desk.++++The broker reserves a bounded runtime (at most one hour) plus a deletion allowance before calling++a provider. Integer micro-currency arithmetic and database locks serialize cap checks. Unknown++creation outcomes block new rentals and retain their reservation across restarts and months.++A pending release also blocks new purchases from that provider until deletion is confirmed.++Every provision, release, drain and strategy change is journalled. Deletion must be confirmed by++provider health before releasing the reservation; a refused deletion continues to accrue cost and++can exceed a reservation. The journal flags this. No software can guarantee an upstream provider++will honor deletion while unavailable; the operator must handle failed releases promptly.++++## Operator view and lifecycle++++`GET /api/v1/compute` shows pools, strategies, machines, sessions, current hourly cost, spend,++committed budget and cap; `?format=json` gives the same data. Only allowlisted signed-in operators++can read it. Strategy forms require same-origin writes. Quotes are at++`GET /api/v1/compute/pools/:pool/quotes`; they never spend.++++A prepared worker that enrolls a fresh Link identity under `RUN_COMPUTE_OWNER` and reports the++provider-assigned rental label as its machine name is bound automatically. Bot must identify it++as that owner's own machine; community/shared name claims are never accepted. Its quote supplies++the pool and RAM inventory. The image's enrollment mechanism is an operator prerequisite.++++For workers without this naming integration, put the machine id in trusted inventory and bind it++with `POST /api/v1/compute/rentals/:id/bind`, JSON `{ "machine": "dev_…" }`, as the operator with++an Origin matching Run. Automatic secure enrollment itself is not implemented here; do not bake reusable Node credentials into an image. Unbound rentals++count as starting capacity and are deleted at their deadline; never regarded as usable desktops.++++Ordinary scale-down marks a host draining, excludes it from placement, waits for zero sessions,++then requests deletion. At the runtime/budget limit, Run ends sessions before release. Unknown++host observations never count as zero occupancy. Each rental's stop deadline survives restart.++Desktop sizing is available, but the current released Bot session API creates browser desks only;++full Linux desktop execution still needs the Computer/Node runtime work and security review.++++Checks: `npm test -- compute sessions dispatch`; the full suite is `OV_ALLOW_SUITE=1 npm test`.++The SDK dependency is pinned to the reviewed job commit until its PR is merged and released;++replace that pin with the resulting SDK release before shipping. Do not deploy this PR before++Opus reviews money, worker enrollment, privacy and cancellation behavior.nnew file mode 100644@@@ -0,0 +1,22 @@++-- phase: expand++-- Expand only. Operator budgets are configuration; all financial commitments survive process restarts.++CREATE TABLE IF NOT EXISTS run_compute_budget_locks (provider text PRIMARY KEY);++CREATE TABLE IF NOT EXISTS run_compute_rentals (++ id text PRIMARY KEY, pool text NOT NULL, provider text NOT NULL, remote_id text, machine text,++ state text NOT NULL, month text NOT NULL, currency text NOT NULL, hourly_micros numeric(30,0) NOT NULL,++ reserved_micros numeric(30,0) NOT NULL, spent_micros numeric(30,0) NOT NULL DEFAULT 0,++ quote jsonb NOT NULL, created_at bigint NOT NULL, deadline bigint NOT NULL, released_at bigint,++ idle_since bigint, last_error text++);++CREATE INDEX IF NOT EXISTS run_compute_rentals_live ON run_compute_rentals(state, provider);++CREATE TABLE IF NOT EXISTS run_compute_journal (++ id bigserial PRIMARY KEY, at bigint NOT NULL, event text NOT NULL, rental text, pool text, detail jsonb NOT NULL++);++CREATE TABLE IF NOT EXISTS run_compute_pools (id text PRIMARY KEY, strategy text NOT NULL, last_machine text, last_tier text);++ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS compute_pool text;++ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS compute_strategy text;++ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS compute_decision jsonb;++CREATE TABLE IF NOT EXISTS run_compute_charges (++ rental text NOT NULL REFERENCES run_compute_rentals(id), month text NOT NULL,++ amount_micros numeric(30,0) NOT NULL, PRIMARY KEY (rental, month)++);@@@ -16,7 +16,7 @@"helmet": "^8.1.0","iovalkey": "^0.4.0","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.153.0","pg": "^8.23.0"},@@@ -1163,9 +1163,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.94.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",-- "integrity": "sha512-p3buZSOSFV81OdZgs5GoSIfMCQNJjW23eAIXIt0RVFIOvh3TByNNEMmuCanxRybFIbzh+9IDMbBwHJrBLl3JeA==",++ "version": "0.100.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0",++ "integrity": "sha512-P7Mi0jqC4PACvNZKgZQqGumUYSAin5X1B1ZTpyvcZ/6P9CIg6xDZew6Zmn/Hqp5VafqVKg09Y6VbEBjj2krtQg==","license": "MIT","bin": {"openvibe-download": "bin/openvibe-download.js"@@@ -27,7 +27,7 @@"helmet": "^8.1.0","iovalkey": "^0.4.0","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.153.0","pg": "^8.23.0"},@@@ -80,7 +80,7 @@ function sessionsRouter({ sessions }) {r.post('/sessions', wrap(async (req, res) => {const w = who(req, res, 'run.session.create');if (!w) return;-- const { row, created } = await sessions.create(w.owner, w.via, req.body && typeof req.body === 'object' ? req.body : {});++ const { row, created } = await sessions.create(w.owner, w.via, req.body && typeof req.body === 'object' ? req.body : {}, { pool: req.query.pool, strategy: req.query.strategy });res.status(created ? 201 : 200).json(sessions.present(row));}));r.get('/sessions', wrap(async (req, res) => {@@@ -93,7 +93,7 @@ function sessionsRouter({ sessions }) {}));r.get('/sessions/:id', wrap(async (req, res) => {const row = await mine(req, res, 'run.session.read');-- if (row) res.json(sessions.present(row));++ if (row) res.json(req.query.explain === '1' ? { ...sessions.present(row), compute: row.compute_decision || null } : sessions.present(row));}));r.post('/sessions/:id/commands', wrap(async (req, res) => {const row = await mine(req, res, 'run.session.control');@@@ -62,7 +62,16 @@ function createApp(opts = {}) {const apiAuth = createApiAuth({ config, keys, userAuth, membership, log });// Run sessions (contracts 0.223.0): browser desks Bot leases per session; the tick starts in server/index.js.const desks = opts.desks || createDeskClient({ config, fetchImpl, now });-- const sessions = opts.sessions || createSessions({ config, db, desks, outbox, now, log });++ const broker = opts.broker || require('./compute/broker').createBroker({ config, db, desks, now, fetchImpl, log });++ const sessions = opts.sessions || createSessions({ config, db, desks, outbox, now, log, broker });++ broker.attachSessions(async (id) => { const row = await sessions.get(id); if (row) await sessions.finish(row, 'ttl', 'compute rental budget exhausted'); });++ broker.attachJobs(async (machine) => {++ const jobs = require('./jobs/store');++ for (const row of await db.many("SELECT * FROM run_jobs WHERE node_id=$1 AND state IN ('placed','running')", [machine])) {++ const result = await jobs.cancel(db, row.id, row.project_id, now());++ if (result.row) await dispatch.cancel(result.row);++ }++ });const release = require('openvibe-shared/release').createRelease({ service: 'run', root: path.join(__dirname, '..') });const app = express();@@@ -120,6 +129,7 @@ function createApp(opts = {}) {express.json({ limit: JSON_LIMIT }),(req, res, next) => { res.setHeader('Cache-Control', 'no-store'); next(); },apiAuth.middleware,++ require('./compute/routes').computeRouter({ broker, config }),sessionsRouter({ sessions }),v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now, log }));@@@ -202,7 +212,7 @@ function createApp(opts = {}) {return res.status(500).type('text/plain').send('Something went wrong\n');});-- Object.assign(app.locals, { config, db, indexnow, valkey, keys, outbox, events, stream, tickets, dispatch, apiAuth, metrics, desks, sessions });++ Object.assign(app.locals, { config, db, indexnow, valkey, keys, outbox, events, stream, tickets, dispatch, apiAuth, metrics, desks, sessions, broker });return app;}nnew file mode 100644@@@ -0,0 +1,237 @@++'use strict';++const { randomUUID }=require('node:crypto');++const {choose,scale,micros,charge,createProviders,STRATEGIES}=require('openvibe-sdk/compute');++const obj=v=>typeof v==='string'?JSON.parse(v):v;++const bare=v=>String(v || '').replace(/^user:/,'');++function createBroker({db,config,desks,now=Date.now,fetchImpl=globalThis.fetch,adapters,log=console}) {++ const cfg=config.compute || {pools:{},inventory:{},providers:{},operators:[],dry_run:true};++ const providers=adapters || createProviders({fetch:fetchImpl,config:cfg.providers,now});++ let timer=null,running=null,finish=null,finishJobs=null;++ const month=()=>new Date(now()).toISOString().slice(0,7);++ const journal=(event,detail={},r={})=>db.query('INSERT INTO run_compute_journal (at,event,rental,pool,detail) VALUES ($1,$2,$3,$4,$5)',[now(),event,r.id || null,r.pool || null,JSON.stringify(detail)]);++ const live=()=>db.many("SELECT * FROM run_compute_rentals WHERE state <> 'released'");++ async function poolConfig(id) {++ const p=Object.hasOwn(cfg.pools,id)?cfg.pools[id]:null; if(!p) throw Object.assign(new Error('unknown compute pool'),{status:422,code:'run.compute.pool'});++ const saved=await db.maybe('SELECT * FROM run_compute_pools WHERE id=$1',[id]);++ return {...p,strategy:saved?.strategy || p.strategy || 'automatic',last_machine:saved?.last_machine};++ }++ async function inventory(hosts,owner) {++ const rentals=await live();++ const counts=await db.many("SELECT machine, count(*)::int AS n FROM run_sessions WHERE state IN ('running','starting') AND machine IS NOT NULL GROUP BY machine");++ const jobs=await db.many("SELECT node_id,count(*)::int AS n FROM run_jobs WHERE state IN ('placed','running') GROUP BY node_id");++ return hosts.map(h=>{++ const rental=rentals.find(r=>r.machine===h.machine_id);++ const rented=rental?obj(rental.quote):{};++ const v={...(rental?{pool:rental.pool,ram_mb:rented.ram_mb,region:rented.region,tier:rented.tier}:{}),...(cfg.inventory[h.machine_id] || {})};++ const ramPerSession=Math.max(cfg.pools[v.pool]?.kind==='desktop'?2048:300,Number(v.measured_peak_ram_mb) || 0,Number(cfg.pools[v.pool]?.measured_peak_ram_mb) || 0);++ const sessions=Math.max(counts.find(c=>c.machine===h.machine_id)?.n || 0,(h.max_browser_desks || 0)-(h.free || 0))+(jobs.find(j=>j.node_id===h.machine_id)?.n || 0);++ return {...v,id:h.machine_id,owner:h.trust==='user-owned'?bare(owner):null,++ tier:rental?(rented.tier || 'partner'):h.trust==='user-owned'?'own':h.trust==='community'?'community':h.trust==='first-party'?(v.tier || 'dedicated'):'partner',++ hourly:rental?Number(rental.hourly_micros)/1e6:(h.trust==='user-owned'?0:v.hourly ?? null),currency:rental?.currency || v.currency || 'USD',++ health:h.online?'up':'down',free_slots:Number(h.free) || 0,++ free_ram_mb:Number(h.free_ram_mb ?? (v.ram_mb!=null?Math.max(0,v.ram_mb-1024-sessions*ramPerSession):0)),++ sessions,draining:rental?['draining','releasing','uncertain'].includes(rental.state):Boolean(v.draining),managed:Boolean(rental),++ state:rental?.state || 'running',idle_since:rental?.idle_since==null?undefined:Number(rental.idle_since)};++ });++ }++ async function decide(row,hosts) {++ const pool=await poolConfig(row.compute_pool);++ const rentalMachines=await inventory(hosts,row.owner);++ const allowed=row.requirements?.trust;++ const filtered=hosts.filter(h=>!allowed || allowed.includes(h.trust)).map(h=>h.machine_id);++ const candidates=rentalMachines.filter(m=>filtered.includes(m.id) && (!m.pool || m.pool===row.compute_pool));++ const d=choose({strategy:row.compute_strategy || pool.strategy,owner:bare(row.owner),kind:row.kind,machine:row.machine_asked || undefined,++ region:row.requirements?.region==='nearest'?undefined:row.requirements?.region,max_hourly:row.requirements?.max_cost_usd == null ? undefined : row.requirements.max_cost_usd / Math.max(1,row.ttl_s || 3600) * 3600,public_egress:pool.public_egress!==false,ram_mb:pool.measured_peak_ram_mb},candidates,pool,{now:now(),current:null});++ if(d.selected) {++ const prev=await db.maybe('SELECT * FROM run_compute_pools WHERE id=$1',[row.compute_pool]);++ if(prev?.last_tier && prev.last_tier!==d.tier) await journal('strategy.switch',{from:prev.last_tier,to:d.tier,strategy:d.strategy,reasons:d.reasons},{pool:row.compute_pool});++ await db.query('INSERT INTO run_compute_pools (id,strategy,last_machine,last_tier) VALUES ($1,$2,$3,$4) ON CONFLICT (id) DO UPDATE SET last_machine=EXCLUDED.last_machine,last_tier=EXCLUDED.last_tier',[row.compute_pool,pool.strategy,d.selected,d.tier]);++ }++ return d;++ }++ async function quotes(poolId) {++ const p=await poolConfig(poolId);++ const results=await Promise.all((p.providers || []).map(async name=>{++ if(!providers[name]) return {provider:name,quotes:[],error:'unsupported provider'};++ try {return {provider:name,quotes:await providers[name].quote({gpu_mb:p.gpu_mb || 0})};}++ catch {return {provider:name,quotes:[],error:'catalogue unavailable or credentials required'};}++ }));++ return results;++ }++ async function reserve(pool,q) {++ const pc=cfg.providers[q.provider] || {};++ if(!pc.key || micros(pc.monthly_cap || '0')===0n) return {reason:'quote only: key and positive monthly cap required'};++ if(q.provisionable===false || !pc.image || !pc.costs_bounded || pc.extra_hourly==null) return {reason:'quote only: prepared image and bounded disk/egress costs required'};++ if(q.currency!==pc.currency || q.expires_at<now() || !Number.isFinite(q.hourly) || q.hourly<=0) return {reason:'invalid or expired quote'};++ const p=await poolConfig(pool);++ const hourly=BigInt(Math.ceil(q.hourly*1e6))+micros(pc.extra_hourly);++ if(p.max_hourly!=null && hourly>micros(p.max_hourly)) return {reason:'total price exceeds pool ceiling'};++ const runtime=Math.min(3600000,Math.max(60000,p.max_runtime_ms || 3600000));++ // Reserve one extra billing quantum for asynchronous deletion. Stop before the reservation's deadline.++ const reserved=charge(hourly,runtime+300000,q.billing_seconds)+micros(q.setup || 0);++ const cap=micros(pc.monthly_cap);++ return db.tx(async t=>{++ await t.query('INSERT INTO run_compute_budget_locks (provider) VALUES ($1) ON CONFLICT DO NOTHING',['pool:'+pool]);++ await t.maybe('SELECT provider FROM run_compute_budget_locks WHERE provider=$1 FOR UPDATE',['pool:'+pool]);++ await t.query('INSERT INTO run_compute_budget_locks (provider) VALUES ($1) ON CONFLICT DO NOTHING',[q.provider]);++ await t.maybe('SELECT provider FROM run_compute_budget_locks WHERE provider=$1 FOR UPDATE',[q.provider]);++ const poolRows=await t.many("SELECT id FROM run_compute_rentals WHERE pool=$1 AND state <> 'released'",[pool]);++ if(poolRows.length >= (p.max_machines ?? 10)) return {reason:'pool machine limit reached'};++ const rows=await t.many("SELECT * FROM run_compute_rentals WHERE provider=$1 AND (month=$2 OR state <> 'released')",[q.provider,month()]);++ if(rows.some(r=>['uncertain','releasing'].includes(r.state))) return {reason:'an unresolved provider call blocks new rentals'};++ const settled=await t.value('SELECT COALESCE(SUM(c.amount_micros),0) FROM run_compute_charges c JOIN run_compute_rentals r ON r.id=c.rental WHERE r.provider=$1 AND c.month=$2',[q.provider,month()]);++ const committed=BigInt(settled)+rows.filter(r=>r.state!=='released').reduce((sum,r)=>sum+BigInt(r.reserved_micros),0n);++ if(committed+reserved>cap) return {reason:'monthly cap reached'};++ const r={id:'ov-'+randomUUID(),pool,provider:q.provider,month:month(),currency:q.currency,created_at:now(),deadline:now()+runtime,quote:q};++ await t.query("INSERT INTO run_compute_rentals (id,pool,provider,state,month,currency,hourly_micros,reserved_micros,quote,created_at,deadline) VALUES ($1,$2,$3,'provisioning',$4,$5,$6,$7,$8,$9,$10)",[r.id,pool,r.provider,r.month,r.currency,hourly.toString(),reserved.toString(),JSON.stringify(q),r.created_at,r.deadline]);++ await t.query('INSERT INTO run_compute_journal (at,event,rental,pool,detail) VALUES ($1,$2,$3,$4,$5)',[now(),'provision.reserved',r.id,pool,JSON.stringify({reserved_micros:reserved.toString(),hourly_micros:hourly.toString(),quote:q.id})]);++ return {rental:r};++ });++ }++ async function provision(pool,q) {++ if(cfg.dry_run || !cfg.enabled) return {dry_run:true,provider:q.provider,quote:q.id,hourly:q.hourly,currency:q.currency};++ const result=await reserve(pool,q); if(!result.rental) return result;++ const r=result.rental;++ try {++ const remote=await providers[q.provider].provision(q,{label:r.id,image:cfg.providers[q.provider].image});++ if(!remote?.id) throw new Error('no remote id');++ await db.query("UPDATE run_compute_rentals SET remote_id=$2,state='starting' WHERE id=$1",[r.id,String(remote.id)]);++ await journal('provision.accepted',{remote_id:String(remote.id)},r);++ return {rental:r.id};++ } catch {++ await db.query("UPDATE run_compute_rentals SET state='uncertain',last_error='provision outcome unknown; reservation retained' WHERE id=$1",[r.id]);++ await journal('provision.uncertain',{},r); return {reason:'provider outcome unknown; reservation retained'};++ }++ }++ async function release(r) {++ if(!r.remote_id) return;++ await db.query("UPDATE run_compute_rentals SET state='releasing' WHERE id=$1",[r.id]);++ await journal('release.requested',{},r);++ try {++ await providers[r.provider].release(r.remote_id);++ const h=await providers[r.provider].health(r.remote_id);++ if(!h.deleted) return; // A stop or accepted DELETE is not proof that billing ended.++ const q=obj(r.quote);++ const spent=charge(BigInt(r.hourly_micros),now()-Number(r.created_at),q.billing_seconds)+micros(q.setup || 0);++ await db.tx(async t=>{++ const locked=await t.maybe('SELECT state FROM run_compute_rentals WHERE id=$1 FOR UPDATE',[r.id]);++ if(locked.state==='released')return;++ const begin=Number(r.created_at),end=now();++ const setup=micros(q.setup || 0),runtime=spent-setup;++ let allocated=0n;++ for(let start=begin;start<end;) {++ const d=new Date(start),boundary=Math.min(end,Date.UTC(d.getUTCFullYear(),d.getUTCMonth()+1,1));++ const share=boundary===end?runtime-allocated:runtime*BigInt(boundary-start)/BigInt(end-begin);++ const value=share+(start===begin?setup:0n);allocated+=share;++ await t.query('INSERT INTO run_compute_charges (rental,month,amount_micros) VALUES ($1,$2,$3) ON CONFLICT DO NOTHING',[r.id,new Date(start).toISOString().slice(0,7),value.toString()]);++ start=boundary;++ }++ await t.query("UPDATE run_compute_rentals SET state='released',spent_micros=$2,released_at=$3,last_error=NULL WHERE id=$1",[r.id,spent.toString(),now()]);++ });++ await journal('release.confirmed',{spent_micros:spent.toString(),over_reservation:spent>BigInt(r.reserved_micros)},r);++ } catch {await db.query("UPDATE run_compute_rentals SET last_error='release unconfirmed; billing reservation retained' WHERE id=$1",[r.id]);}++ }++ async function tickOnce() {++ const rentals=await live();++ for(const r of rentals) {++ const pc=cfg.providers[r.provider] || {};++ if(r.state==='provisioning') { // another process may still be completing the call; do not retry it++ if(now()-Number(r.created_at)>60000) await db.query("UPDATE run_compute_rentals SET state='uncertain' WHERE id=$1 AND state='provisioning'",[r.id]);++ continue;++ }++ if(r.state==='releasing') {await release(r); continue;}++ const settled=BigInt(await db.value('SELECT COALESCE(SUM(c.amount_micros),0) FROM run_compute_charges c JOIN run_compute_rentals p ON p.id=c.rental WHERE p.provider=$1 AND c.month=$2',[r.provider,month()]));++ const committed=rentals.filter(x=>x.provider===r.provider).reduce((sum,x)=>sum+BigInt(x.reserved_micros),settled);++ let unhealthy=false;++ if(r.remote_id) { try { const h=await providers[r.provider].health(r.remote_id); unhealthy=h.deleted || ['error','stopped','off','dead'].includes(String(h.state).toLowerCase()); } catch { /* unknown health keeps the reservation and deadline */ } }++ const stop=!cfg.enabled || now()>=Number(r.deadline) || committed>micros(pc.monthly_cap || '0') || unhealthy;++ if(stop && r.remote_id) {++ await db.query("UPDATE run_compute_rentals SET state='draining' WHERE id=$1",[r.id]);++ if(r.machine && finish) for(const s of await db.many("SELECT id FROM run_sessions WHERE machine=$1 AND state <> 'ended'",[r.machine])) await finish(s.id);++ if(r.machine && finishJobs) await finishJobs(r.machine);++ const jobs=r.machine ? await db.value("SELECT count(*)::int FROM run_jobs WHERE node_id=$1 AND state IN ('placed','running')",[r.machine]):0;++ const count= r.machine ? await db.value("SELECT count(*)::int FROM run_sessions WHERE machine=$1 AND state <> 'ended'",[r.machine]):0;++ if(!count && !jobs) await release(r);++ }++ }++ if(!cfg.enabled || !cfg.owner || !desks.enabled) return;++ let hosts; try {hosts=await desks.hosts(cfg.owner);} catch {return;}++ // Bot returns a name only for this authenticated owner's own machines. A community seller cannot++ // claim a paid rental by copying its label. The prepared image must enroll a fresh Link identity.++ for(const r of await live()) {++ if(r.state!=='starting' || r.machine || !r.remote_id)continue;++ const matches=hosts.filter(h=>h.trust==='user-owned' && h.name===r.id && h.online);++ if(matches.length!==1)continue;++ const updated=await db.maybe("UPDATE run_compute_rentals SET machine=$2,state='running' WHERE id=$1 AND machine IS NULL AND state='starting' RETURNING id",[r.id,matches[0].machine_id]);++ if(updated)await journal('machine.bound',{machine:matches[0].machine_id,source:'authenticated owner host label'},r);++ }++++ for(const id of Object.keys(cfg.pools)) {++ const p=await poolConfig(id);++ const all=await inventory(hosts,cfg.owner);++ const rows=(await live()).filter(r=>r.pool===id);++ const ms=all.filter(m=>m.pool===id || rows.some(r=>r.machine===m.id));++ for(const r of rows) {++ const m=ms.find(m=>m.id===r.machine);++ if(m) {++ if(m.sessions===0 && r.idle_since==null) await db.query('UPDATE run_compute_rentals SET idle_since=$2 WHERE id=$1',[r.id,now()]);++ if(m.sessions>0 && r.idle_since!=null) await db.query('UPDATE run_compute_rentals SET idle_since=NULL WHERE id=$1',[r.id]);++ } else ms.push({id:r.id,state:'provisioning',health:'starting',sessions:null});++ }++ const waiting=await db.many("SELECT created_at,compute_strategy FROM run_sessions WHERE compute_pool=$1 AND state='placing' AND machine_asked IS NULL AND (compute_strategy IS NULL OR compute_strategy <> 'own') ORDER BY created_at",[id]);++ const desired=scale(p,ms,waiting.map(s=>({created_at:Number(s.created_at)})),{now:now()});++ for(const a of desired.actions) {++ const r=rows.find(r=>r.machine===a.id || r.id===a.id);++ if(a.action==='drain' && r && !cfg.dry_run) {await db.query("UPDATE run_compute_rentals SET state='draining' WHERE id=$1",[r.id]);await journal('drain.started',{reason:a.reason},r);}++ if(a.action==='release' && r && !cfg.dry_run) await release(r);++ if(a.action==='provision') {++ const qs=(await quotes(id)).flatMap(x=>x.quotes).filter(q=>q.provisionable!==false && q.hourly!=null && q.currency===(p.currency || 'USD') && q.ram_mb>=((p.kind==='desktop'?2048:300)+1024));++ let candidates=qs.flatMap(q=>{++ const pc=cfg.providers[q.provider] || {};++ if(!cfg.dry_run && (!pc.key || micros(pc.monthly_cap || '0')===0n || !pc.image || !pc.costs_bounded || pc.extra_hourly==null))return [];++ const extra=pc.extra_hourly==null?0:Number(micros(pc.extra_hourly))/1e6;++ return [{...q,hourly:q.hourly+extra,id:q.provider+':'+q.id,health:'up',free_ram_mb:q.ram_mb-1024,free_slots:p.slots_per_machine || 1}];++ });++ while(true) {++ const d=choose({kind:p.kind,strategy:waiting[0]?.compute_strategy || p.strategy,region:p.region,gpu_mb:p.gpu_mb},candidates,p,{now:now()});++ const q=qs.find(q=>q.provider+':'+q.id===d.selected);++ if(!q){await journal('scale.decision',{reason:a.reason,decision:d,result:{reason:'no eligible rental quote'}},{pool:id});break;}++ const result=await provision(id,q);++ await journal('scale.decision',{reason:a.reason,decision:d,result},{pool:id});++ if(result.rental || result.dry_run || /unknown/.test(result.reason || ''))break;++ // A provider's exhausted cap or pending release must not hide eligible alternatives.++ candidates=candidates.filter(c=>c.provider!==q.provider);++ }++ break; // one fresh quote/reservation per pool/pass, observes previous launches on next pass++ }++ }++ }++ }++ function tick(){if(running)return running;running=tickOnce().finally(()=>{running=null;});return running;}++ async function status() {++ const rentals=await db.many('SELECT * FROM run_compute_rentals ORDER BY created_at DESC LIMIT 200');++ const budgets=[];++ for(const [provider,pc] of Object.entries(cfg.providers || {})) {++ const rs=await db.many("SELECT * FROM run_compute_rentals WHERE provider=$1 AND (month=$2 OR state <> 'released')",[provider,month()]);++ const settled=BigInt(await db.value('SELECT COALESCE(SUM(c.amount_micros),0) FROM run_compute_charges c JOIN run_compute_rentals r ON r.id=c.rental WHERE r.provider=$1 AND c.month=$2',[provider,month()]));++ let spent=settled,committed=settled,hourly=0n;++ for(const r of rs.filter(r=>r.state!=='released')){const h=BigInt(r.hourly_micros);spent+=charge(h,now()-Math.max(Number(r.created_at),Date.parse(month()+'-01T00:00:00Z')),obj(r.quote).billing_seconds);committed+=BigInt(r.reserved_micros);hourly+=h;}++ budgets.push({provider,currency:pc.currency,cap:pc.monthly_cap,configured:Boolean(pc.key),spent_micros:spent.toString(),committed_micros:committed.toString(),hourly_micros:hourly.toString()});++ }++ let machines=[];++ if(cfg.owner && desks.enabled) {try {machines=await inventory(await desks.hosts(cfg.owner),cfg.owner);} catch {machines=Object.entries(cfg.inventory).map(([id,v])=>({...v,id,health:'unknown'}));}}++ return {enabled:cfg.enabled,dry_run:cfg.dry_run,machines,pools:await Promise.all(Object.keys(cfg.pools).map(async id=>({id,...await poolConfig(id)}))),budgets,rentals,++ sessions:await db.many("SELECT machine,compute_pool,count(*)::int AS sessions FROM run_sessions WHERE state IN ('starting','running') GROUP BY machine,compute_pool"),++ journal:await db.many('SELECT * FROM run_compute_journal ORDER BY id DESC LIMIT 100')};++ }++ return {decide,inventory,quotes,reserve,provision,release,tick,status,poolConfig,++ async available(hosts){const blocked=new Set((await live()).filter(r=>['draining','releasing','uncertain'].includes(r.state)).map(r=>r.machine));return hosts.filter(h=>!blocked.has(h.machine_id));},++ async setStrategy(id,strategy){await poolConfig(id);if(!STRATEGIES.includes(strategy))throw Object.assign(new Error('invalid strategy'),{status:422});await db.query('INSERT INTO run_compute_pools (id,strategy) VALUES ($1,$2) ON CONFLICT(id) DO UPDATE SET strategy=EXCLUDED.strategy',[id,strategy]);await journal('strategy.changed',{strategy},{pool:id});},++ async bind(id,machine){const r=await db.maybe("SELECT * FROM run_compute_rentals WHERE id=$1 AND state='starting'",[id]);if(!r)throw new Error('rental not starting');if(!cfg.inventory[machine] || cfg.inventory[machine].pool!==r.pool)throw new Error('machine must be in configured pool inventory');await db.query("UPDATE run_compute_rentals SET machine=$2,state='running' WHERE id=$1",[id,machine]);await journal('machine.bound',{machine},r);},++ attachSessions(fn){finish=fn;},++ attachJobs(fn){finishJobs=fn;},++ start(){if(!timer){timer=setInterval(()=>tick().catch(()=>log.warn('[Run] compute reconcile failed')),cfg.tick_ms || 30000);timer.unref?.();}},++ async stop(){if(timer)clearInterval(timer);timer=null;if(running)await running;},++ };++}++module.exports={createBroker};nnew file mode 100644@@@ -0,0 +1,28 @@++'use strict';++const { STRATEGIES, micros } = require('openvibe-sdk/compute');++function loadCompute(env) {++ let pools = {}, inventory = {};++ for (const [name, put] of [['RUN_COMPUTE_POOLS', v => { pools=v; }],['RUN_COMPUTE_INVENTORY', v => { inventory=v; }]]) {++ if (!env[name]) continue;++ const v=JSON.parse(env[name]);++ if (!v || typeof v !== 'object' || Array.isArray(v)) throw new Error(`${name} must be an object`);++ put(v);++ }++ for (const [id,p] of Object.entries(pools)) {++ if (!/^[a-z0-9-]{1,40}$/.test(id) || !STRATEGIES.includes(p.strategy || 'automatic')) throw new Error('invalid compute pool');++ if (p.kind && !['browser','desktop'].includes(p.kind)) throw new Error('invalid pool kind');++ }++ const providers={};++ for (const name of ['hetzner','ovh','vast','runpod']) {++ const prefix='RUN_COMPUTE_'+name.toUpperCase()+'_';++ const cap=env[prefix+'MONTHLY_CAP'] || '0'; micros(cap);++ providers[name]={ key:env[prefix+'KEY'] || '', monthly_cap:cap, currency:env[prefix+'CURRENCY'] || (['hetzner','ovh'].includes(name)?'EUR':'USD'),++ image:env[prefix+'IMAGE'] || '', extra_hourly:env[prefix+'EXTRA_HOURLY'] || null,++ costs_bounded:env[prefix+'COSTS_BOUNDED']==='1',++ project:env[prefix+'PROJECT'] || '', application_secret:env[prefix+'APPLICATION_SECRET'] || '', consumer_key:env[prefix+'CONSUMER_KEY'] || '' };++ }++ return { enabled:env.RUN_COMPUTE_ENABLED==='1', dry_run:env.RUN_COMPUTE_DRY_RUN!=='0', pools, inventory, providers,++ operators:(env.RUN_COMPUTE_OPERATORS || '').split(',').map(s=>s.trim()).filter(Boolean),++ owner:env.RUN_COMPUTE_OWNER || '', job_pool:env.RUN_COMPUTE_JOB_POOL || '', tick_ms:30000 };++}++module.exports={loadCompute};nnew file mode 100644@@@ -0,0 +1,25 @@++'use strict';++const express=require('express');++const {STRATEGIES}=require('openvibe-sdk/compute');++const esc=s=>String(s ?? '').replace(/[&<>"']/g,c=>({'&':'&','<':'<','>':'>','"':'"',"'":'''}[c]));++function computeRouter({broker,config}) {++ const r=express.Router();++ const wrap=fn=>(req,res,next)=>Promise.resolve(fn(req,res)).catch(next);++ r.use('/compute',(req,res,next)=>{++ if(req.principal?.kind!=='user' || !(config.compute?.operators || []).includes(req.principal.subject))return res.status(403).json({error:'operator access required'});++ if(req.method!=='GET' && req.get('origin')!==config.baseUrl)return res.status(403).json({error:'same-origin write required'});++ next();++ });++ r.get('/compute',wrap(async(req,res)=>{++ const s=await broker.status();++ if(req.query.format==='json')return res.json(s);++ const money=v=>(Number(v)/1e6).toFixed(4);++ const machines=[...s.machines.map(m=>({...m,machine:m.id,provider:s.rentals.find(r=>r.machine===m.id)?.provider || 'existing',hourly_micros:m.hourly==null?null:m.hourly*1e6})),...s.rentals.filter(r=>!s.machines.some(m=>m.id===r.machine)).map(r=>({...r,tier:(typeof r.quote==='string'?JSON.parse(r.quote):r.quote).tier}))];++ res.type('html').send(`<!doctype html><html lang="en"><meta charset="utf-8"><meta name="viewport" content="width=device-width"><title>Compute pools · OpenVibe Run</title><style>:root{color-scheme:light dark;font:16px/1.6 system-ui}body{margin:0;background:light-dark(#f6f8fa,#101519);color:light-dark(#16232c,#e6edf2)}main{max-width:1100px;margin:auto;padding:24px}h1{font-size:2rem;line-height:1.2}h2{margin-top:2rem}section{padding:20px;margin:16px 0;border:1px solid #70809060;border-radius:12px}button,select{font:inherit;min-height:44px;padding:8px 12px;border-radius:6px;border:1px solid #70809080}form{display:flex;flex-wrap:wrap;gap:12px;margin:16px 0}a{color:light-dark(#075a85,#87d6ff)}table{display:block;overflow-x:auto;border-collapse:collapse;max-width:100%;text-align:left}th,td{padding:10px;border-bottom:1px solid #70809060;white-space:nowrap}ol{padding-left:24px}li{overflow-wrap:anywhere;margin:12px 0}@media(max-width:600px){main{padding:16px}h1{font-size:1.7rem}}</style><body><main><h1>Compute pools</h1><p>${s.dry_run?'Dry run — rentals disabled':'Live rentals within configured caps'} · ${s.enabled?'Reconciler enabled':'Reconciler disabled'}</p><h2>Pools</h2>${s.pools.map(p=>`<section><h3>${esc(p.id)}</h3><p>Strategy: ${esc(p.strategy)} · Warm minimum: ${esc(p.min_warm || 0)} · Queue target: ${esc(p.target_wait_ms ?? 30000)} ms</p><form method="post" action="/api/v1/compute/pools/${esc(p.id)}/strategy"><select name="strategy">${STRATEGIES.map(v=>`<option${v===p.strategy?' selected':''}>${v}</option>`).join('')}</select><button>Apply strategy</button></form><a href="/api/v1/compute/pools/${esc(p.id)}/quotes">Quotes and dry-run prices</a></section>`).join('')}<h2>Money</h2><table><tr><th>Provider</th><th>Currency</th><th>Per hour</th><th>This month</th><th>Committed</th><th>Cap</th></tr>${s.budgets.map(b=>`<tr><td>${esc(b.provider)}</td><td>${esc(b.currency)}</td><td>${money(b.hourly_micros)}</td><td>${money(b.spent_micros)}</td><td>${money(b.committed_micros)}</td><td>${esc(b.cap)}</td></tr>`).join('')}</table><h2>Machines</h2><table><tr><th>Pool</th><th>Machine / rental</th><th>Tier / provider</th><th>State</th><th>Sessions</th><th>Per hour</th></tr>${machines.map(m=>`<tr><td>${esc(m.pool)}</td><td>${esc(m.machine || m.id)}</td><td>${esc(m.tier)} / ${esc(m.provider)}</td><td>${esc(m.state || m.health)}</td><td>${s.sessions.find(x=>x.machine && x.machine===m.machine)?.sessions || 0}</td><td>${m.hourly_micros==null?'unknown':money(m.hourly_micros)} ${esc(m.currency)}</td></tr>`).join('')}</table><h2>Last decisions and switches</h2><ol>${s.journal.map(j=>`<li>${esc(new Date(Number(j.at)).toISOString())} ${esc(j.event)} ${esc(JSON.stringify(j.detail))}</li>`).join('')}</ol></main></body></html>`);++ }));++ r.get('/compute/pools/:pool/quotes',wrap(async(req,res)=>res.json({dry_run:true,providers:await broker.quotes(req.params.pool)})));++ r.post('/compute/pools/:pool/strategy',express.urlencoded({extended:false,limit:'2kb'}),wrap(async(req,res)=>{await broker.setStrategy(req.params.pool,req.body.strategy);res.redirect(303,'/api/v1/compute');}));++ r.post('/compute/rentals/:id/bind',wrap(async(req,res)=>{await broker.bind(req.params.id,String(req.body.machine || ''));res.json({ok:true});}));++ return r;++}++module.exports={computeRouter};@@@ -34,6 +34,7 @@ function loadConfig(env = process.env) {const networkUrl = trim(env.OV_NETWORK_URL || 'https://openvibe.network');const baseUrl = trim(env.BASE_URL || (isProduction ? 'https://openvibe.run' : `http://localhost:${port}`));return {++ compute: require('./compute/config').loadCompute(env),nodeEnv,isProduction,port,@@@ -53,7 +53,7 @@ function createDispatcher({ config = null, db = null, store = null, events = nulconst learned = outcomes || (db ? createOutcomes({ db, now, log }) : null);const benchPriors = priors || createPriors({ config, fetchImpl, now, log, ttlMs: config && config.bench ? config.bench.priorsTtlMs : undefined });// Provider offers are placement candidates only while the adapter can be called.-- const placer = placement || createPlacement({ config, fetchImpl, now, log, providers: adapter.enabled ? { [BROWSER_PROVIDER]: ['browser'] } : {}, outcomes: learned, priors: benchPriors });++ const placer = placement || createPlacement({ config, db, fetchImpl, now, log, providers: adapter.enabled ? { [BROWSER_PROVIDER]: ['browser'] } : {}, outcomes: learned, priors: benchPriors });const poller = createPoller({ config, db, store, events, stream, bot: botClient, provider: adapter, usageRelay: relay, placement: placer, outcomes: learned, now, log });const enabled = Boolean(botClient.enabled || adapter.enabled);const note = botClient.enabled@@@ -39,7 +39,7 @@* answer with no eligible candidate is null, which the poller settles run.job.unplaceable.*/const contracts = require('openvibe-contracts');--const { place, optionKey } = require('openvibe-sdk/placement');++const { place, plan, optionKey } = require('openvibe-sdk/placement');const { createServiceTokenClient } = require('openvibe-sdk/auth');const { requirementsDefault } = require('../util');const { PROVIDER: BROWSER_PROVIDER } = require('./browser-provider');@@@ -106,7 +106,7 @@ const personOf = (requester) => (/^user:(usr_[0-9A-HJKMNP-TV-Z]{26})$/.exec(Stri* createPlacement({ config, fetchImpl, now, log, providers }) → { choose, offers, status }. `providers` maps* the name of each external adapter that is on to the classes it runs ({ 'browser-provider': ['browser'] }).*/--function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console, providers = {}, outcomes = null, priors = null } = {}) {++function createPlacement({ config, db = null, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console, providers = {}, outcomes = null, priors = null } = {}) {const { url } = config.network;const offersUrlOf = (kind) => `${url}/api/v1/offers?kind=${kind}`;const offersUrl = offersUrlOf('node');@@@ -192,6 +192,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Datreturn { ok: false, code: 'run.job.node_offline', detail: 'node must be a machine id (dev_…)' };}const age = fresh ? 0 : offersTtlMs;++ if (db && await db.maybe("SELECT id FROM run_compute_rentals WHERE machine=$1 AND state IN ('draining','releasing','uncertain') LIMIT 1", [nodeId])) return { ok: false, code: 'run.job.node_offline', detail: 'the machine is draining' };const offer = [...(await offers('node', { maxAgeMs: age })), ...(await ownedOffers({ maxAgeMs: age }))].find((o) => o && o.kind === 'node' && o.node_id === nodeId);if (!offer) return { ok: false, code: 'run.job.node_offline', detail: `machine ${nodeId} is not connected to Run` };@@@ -230,7 +231,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat// 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'++ let 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));@@@ -249,6 +250,31 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Datif (!candidates.length) return null;const req = { ...requirements, capabilities: [...new Set([...(requirements.capabilities || []), want])] };contracts.assertValid('platform.workload-requirements@1', req);++ let computeDecision = null;++ if (db) {++ const draining = new Set((await db.many("SELECT machine FROM run_compute_rentals WHERE state IN ('draining','releasing','uncertain')")).map(r => r.machine));++ candidates = candidates.filter(o => !draining.has(o.node_id));++ }++ const poolId = config.compute && config.compute.job_pool;++ const pool = poolId && config.compute.pools[poolId];++ if (pool) {++ const saved = db && await db.maybe('SELECT strategy FROM run_compute_pools WHERE id=$1', [poolId]);++ const hard = plan(req, candidates, { now: now() });++ const eligible = new Set(hard.candidates.filter(c => c.eligible).map(c => c.id));++ const ms = candidates.filter(o => eligible.has(o.offer_id)).map(o => {++ const v = config.compute.inventory[o.node_id] || {};++ return { ...v, id: o.offer_id, owner: o.owner, tier: o.trust === 'user-owned' ? 'own' : o.trust === 'community' ? 'community' : o.trust === 'first-party' ? (v.tier || 'dedicated') : 'partner',++ health: o.health.status, region: o.region, currency: v.currency || 'USD', hourly: v.hourly ?? (o.trust === 'user-owned' ? 0 : null),++ free_ram_mb: o.capacity?.memory?.available_mb ?? v.free_ram_mb ?? 0,++ free_slots: o.capacity?.workers?.[row.class] ?? v.free_slots ?? 0,++ latency_ms: o.latency_ms?.p95, gpu_mb: o.capacity?.gpu?.memory_mb ?? v.gpu_mb };++ });++ const strategy = { cheapest: 'cheapest', 'lowest-latency': 'fastest', private: 'private', 'first-party-only': 'private' }[req.objective] || saved?.strategy || pool.strategy;++ computeDecision = require('openvibe-sdk/compute').choose({ strategy, owner: person, kind: row.class,++ ram_mb: req.resources?.memory_mb, region: req.region, public_egress: pool.public_egress !== false }, ms, pool, { now: now() });++ candidates = candidates.filter(o => o.offer_id === computeDecision.selected);++ }++// Placement A1: Bench's priors per option key (an hour's cache), the person's learned counts (one read) and the// job's class as the step kind. An unreadable table or a quiet Bench leaves the defaults, never a failed placement.const learned = outcomes && person ? await outcomes.forPerson(person) : [];@@@ -274,7 +300,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat// A provider offer has no device: its own id stands for where the job went (the read result's node).node_id: chosen.kind === 'provider' ? (chosen.node_id || chosen.offer_id) : chosen.node_id,provider: chosen.provider ?? null, region: chosen.region ?? null, kind: chosen.kind,-- offer_id: chosen.offer_id, trust: chosen.trust, reasons: result.reasons,++ offer_id: chosen.offer_id, trust: chosen.trust, reasons: [...result.reasons, ...(computeDecision ? [...computeDecision.reasons, `estimated ${computeDecision.hourly} ${computeDecision.currency}/hour`] : [])],option_key: optionKey(chosen), step_kind: row.class,explain: publicExplain({ line: placed.line, ranked: placed.ranked, rejected: placed.rejected }),};@@@ -38,7 +38,7 @@ async function main() {const accountData = accountDataLib.create({ db, log: console });const accountSend = config.oauth.clientSecret ? createNetworkSender({ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret }) : null;const app = createApp({ config, db, valkey, registry, accountData, accountSend });-- const { keys, outbox, stream, tickets, dispatch, indexnow, sessions } = app.locals;++ const { keys, outbox, stream, tickets, dispatch, indexnow, sessions, broker } = app.locals;keys.start();const ticketKey = tickets.start();@@@ -50,6 +50,7 @@ async function main() {dispatch.start();// Run sessions (server/sessions/): place, watch and meter the open ones. Off without the Bot token.sessions.start();++ broker.start();}let accountSubscription = null;@@@ -88,6 +89,7 @@ async function main() {() => keys.stop(),() => tickets.stop(),() => dispatch.stop(),++ () => broker.stop(),() => sessions.stop(),() => stream.close(),() => outbox.stop(),@@@ -43,7 +43,7 @@ class SessionError extends Error {constructor(status, code, detail) { super(detail || code); this.status = status; this.code = code; this.detail = detail; }}--function createSessions({ config, db, desks, outbox, now = () => Date.now(), log = console }) {++function createSessions({ config, db, desks, outbox, now = () => Date.now(), log = console, broker = null }) {const cfg = config.sessions || {};const tickMs = cfg.tickMs || 2000;const leaseCheckMs = cfg.leaseCheckMs || 30_000;@@@ -88,10 +88,16 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log}// ── create, act, finish ───────────────────────────────────────────────────────────────────────────────────-- async function create(owner, via, body) {++ async function create(owner, via, body, compute = {}) {++ if (compute.pool && (typeof compute.pool !== 'string' || !/^[a-z0-9-]{1,40}$/.test(compute.pool))) throw new SessionError(422, 'run.compute.pool', 'invalid compute pool');++ const pool = compute.pool || null;++ const strategy = compute.strategy || null;++ if (strategy && !require('openvibe-sdk/compute').STRATEGIES.includes(strategy)) throw new SessionError(422, 'run.compute.strategy', 'unknown compute strategy');++ if (strategy && !pool) throw new SessionError(422, 'run.compute.pool', 'a strategy requires a compute pool');++ if (pool) { if (!broker) throw new SessionError(503, 'run.compute.off'); await broker.poolConfig(pool); }const v = contracts.validate('run.session-create-request@1', body);if (!v.valid) throw new SessionError(422, 'run.session.invalid', `invalid session request: ${(v.errors && v.errors[0] && v.errors[0].message) || 'bad value'}`.slice(0, 300));-- const digest = store.digestOf(body);++ const digest = store.digestOf(pool ? { ...body, compute_pool: pool, compute_strategy: strategy } : body);if (body.idempotency_key) {const earlier = await db.maybe('SELECT * FROM run_sessions WHERE owner = $1 AND idempotency_key = $2 AND created_at > $3', [owner, body.idempotency_key, now() - 86_400_000]);if (earlier) {@@@ -104,7 +110,7 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), logconst row = {id: store.newSessionId(), owner, via: via || null, kind: 'browser', state: 'placing',width: body.width || 1280, height: body.height || 800, ttl_s: body.ttl_s || 3600, idle_s: body.idle_s || 900,-- machine_asked: body.machine || null, requirements: body.requirements || null, idempotency_key: body.idempotency_key || null,++ compute_pool: pool, compute_strategy: strategy, compute_decision: null, machine_asked: body.machine || null, requirements: body.requirements || null, idempotency_key: body.idempotency_key || null,request_digest: digest, reasons: [], task_id: body.task_id || null, curtain: false, created_at: at, updated_at: at,};const out = await store.insert(db, row);@@@ -219,14 +225,20 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), logif (now() - row.created_at >= row.ttl_s * 1000) return finish(row, 'unplaceable', 'no machine had room before the session\'s time ran out', { release: false });let hosts;try { hosts = await hostsFor(bare(row.owner)); } catch (e) { if (!e.retryable) log.warn(`[Run] session ${row.id}: desk hosts: ${e.code}`); return row; }-- const list = candidates(row, hosts);++ if (broker) hosts = await broker.available(hosts);++ let list = candidates(row, hosts);++ let decision = null;++ if (row.compute_pool && broker) {++ decision = await broker.decide(row, hosts);++ list = decision.selected ? hosts.filter(h => h.machine_id === decision.selected) : [];++ }if (!list.length) {if (row.machine_asked && !hosts.some((h) => h.machine_id === row.machine_asked)) return finish(row, 'unplaceable', 'that machine is not open to your agent (or it runs no browser desks)', { release: false });return row;}const desk = store.deskOf(row.id);for (const h of list) {-- await store.update(db, row.id, { state: 'starting', machine: h.machine_id, desk, trust: h.trust, offer_id: h.shared ? `desk:${h.machine_id}` : null, reasons: reasonsOf(row, h), updated_at: now() });++ await store.update(db, row.id, { state: 'starting', machine: h.machine_id, desk, trust: h.trust, offer_id: h.shared ? `desk:${h.machine_id}` : null, reasons: [...reasonsOf(row, h), ...(decision ? [...decision.reasons, `estimated ${decision.hourly} ${decision.currency}/hour`] : [])], compute_decision: decision, updated_at: now() });let lease;try {lease = await desks.lease(h.machine_id, bare(row.owner), { session: row.id, desk, kind: 'browser', width: row.width, height: row.height });@@@ -13,9 +13,9 @@ const crypto = require('crypto');const { ids } = require('openvibe-contracts');const COLUMNS = ['id', 'owner', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'idempotency_key',-- 'request_digest', 'machine', 'desk', 'trust', 'offer_id', 'reasons', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason',++ 'compute_pool', 'compute_strategy', 'compute_decision', 'request_digest', 'machine', 'desk', 'trust', 'offer_id', 'reasons', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason','end_detail', 'metered_minutes', 'lease_checked_at', 'task_id', 'curtain', 'updated_at'];--const JSON_COLUMNS = new Set(['requirements', 'reasons']);++const JSON_COLUMNS = new Set(['requirements', 'reasons', 'compute_decision']);const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());const json = (v, dflt) => { if (v == null) return dflt; if (typeof v === 'object') return v; try { return JSON.parse(v); } catch { return dflt; } };nnew file mode 100644@@@ -0,0 +1,23 @@++'use strict';++const assert=require('node:assert/strict');++const {boot}=require('./helpers/app');++const {ids}=require('openvibe-contracts');++(async()=>{++ const owner=ids.newId('user'), stranger=ids.newId('user');++ const machine=ids.newId('device');++ const desks={enabled:true,note:'stub',hosts:async()=>[{machine_id:machine,trust:'user-owned',online:true,free:2,max_browser_desks:2}],lease:async()=>({ok:true}),get:async()=>({ok:true})};++ const t=await boot({appOpts:{desks},env:{RUN_COMPUTE_OPERATORS:owner,RUN_COMPUTE_POOLS:JSON.stringify({browser:{strategy:'own',currency:'USD'}}),RUN_COMPUTE_INVENTORY:JSON.stringify({[machine]:{pool:'browser',ram_mb:4096,region:'us'}})}});++ try {++ const as=u=>({token:t.network.signUser({subject:u})});++ assert.equal((await t.call('GET','/api/v1/compute?format=json',as(stranger))).status,403);++ assert.equal((await t.call('GET','/api/v1/compute?format=json',as(owner))).status,200);++ const form=await t.call('GET','/api/v1/compute',as(owner));assert.equal(form.status,200);assert.match(form.text,/Compute pools/);++ const start=await t.call('POST','/api/v1/sessions?pool=browser&strategy=own',{...as(owner),body:{kind:'browser'}});++ assert.equal(start.status,201,start.text);await t.app.locals.sessions.tick();++ const s=await t.call('GET',`/api/v1/sessions/${start.json.id}?explain=1`,as(owner));++ assert.equal(s.json.compute.selected,machine);assert.equal(s.json.compute.strategy,'own');assert.equal(s.json.compute.hourly,0);++ assert.equal((await t.call('GET',`/api/v1/sessions/${start.json.id}?explain=1`,as(stranger))).status,404);++ assert.equal((await t.call('POST','/api/v1/compute/pools/browser/strategy',{...as(owner),body:{strategy:'cheapest'},headers:{origin:'https://evil.example'}})).status,403);++ console.log('operator auth, CSRF, session strategy and owner-only cost explanation: passed');++ } finally {await t.close();}++})().catch(e=>{console.error(e);process.exit(1);});nnew file mode 100644@@@ -0,0 +1,34 @@++'use strict';++const assert=require('node:assert/strict');++const {testDb}=require('./helpers/db');++const {createBroker}=require('../server/compute/broker');++(async()=>{++ for(const scenario of ['total price','cap fallback','GPU fit','missing key']) {++ const t=await testDb();const now=Date.UTC(2026,10,8);const created=[];++ const config={compute:{enabled:true,dry_run:false,owner:'alice',inventory:{},pools:{workers:{kind:'browser',strategy:'cheapest',currency:'EUR',providers:['hetzner','ovh'],min_warm:1,max_machines:2,gpu_mb:scenario==='GPU fit'?8192:0}},providers:{}}};++ const adapters={};++ for(const [name,hourly,gpu_mb] of [['hetzner',0.1,0],['ovh',0.2,8192]]) {++ config.compute.providers[name]={key:'stub',monthly_cap:'10',currency:'EUR',image:'prepared',costs_bounded:true,extra_hourly:'0'};++ adapters[name]={quote:async()=>[{provider:name,id:name,hourly,gpu_mb,ram_mb:8192,currency:'EUR',region:'eu',tier:'dedicated',setup:0,billing_seconds:3600,expires_at:now+60000}],provision:async()=>{created.push(name);return {id:name};},health:async()=>({state:'running'}),release:async()=>{}};++ }++ if(scenario==='total price')config.compute.providers.hetzner.extra_hourly='0.5';++ if(scenario==='cap fallback')config.compute.providers.hetzner.monthly_cap='0.01';++ if(scenario==='missing key')config.compute.providers.hetzner.key='';++ const broker=createBroker({db:t.db,config,adapters,desks:{enabled:true,hosts:async()=>[]},now:()=>now});++ try {++ await broker.tick();assert.deepEqual(created,['ovh'],scenario);++ if(scenario==='total price') {++ config.compute.inventory.m={pool:'workers',ram_mb:8192};++ const hosts=[{machine_id:'m',trust:'user-owned',online:true,max_browser_desks:4,free:2}];++ config.compute.pools.workers.kind='desktop';++ assert.equal((await broker.inventory(hosts,'alice'))[0].free_ram_mb,3072,'occupied desktops use the 2 GB floor');++ config.compute.pools.workers.measured_peak_ram_mb=3000;++ assert.equal((await broker.inventory(hosts,'alice'))[0].free_ram_mb,1168,'higher measured peaks reduce remaining capacity');++ config.compute.pools.workers.kind='browser';config.compute.pools.workers.measured_peak_ram_mb=900;++ assert.equal((await broker.inventory(hosts,'alice'))[0].free_ram_mb,5368,'browser pools use their measured peak');++ }++ }++ finally {await t.close();}++ }++ console.log('autoscaler selects total cost, falls back on capped or unconfigured providers, and enforces GPU fit: passed');++})().catch(e=>{console.error(e);process.exit(1);});nnew file mode 100644@@@ -0,0 +1,68 @@++'use strict';++const assert=require('node:assert/strict');++const {testDb}=require('./helpers/db');++const {createBroker}=require('../server/compute/broker');++const {loadCompute}=require('../server/compute/config');++(async()=>{++ const t=await testDb(); let time=Date.UTC(2026,10,8); let created=0,removed=0,deleted=false,uncertain=false;++ const q={provider:'hetzner',id:'cx',hourly:1,currency:'EUR',ram_mb:4096,gpu_mb:0,tier:'dedicated',region:'eu',setup:0,billing_seconds:3600,expires_at:time+60000};++ const config={compute:{enabled:true,dry_run:false,owner:'alice',pools:{browsers:{kind:'browser',currency:'EUR',providers:['hetzner'],min_warm:1,strategy:'automatic'}},inventory:{m:{pool:'browsers',tier:'dedicated',ram_mb:4096,hourly:1,currency:'EUR'}},providers:{hetzner:{key:'stub',monthly_cap:'2',currency:'EUR',image:'prepared',costs_bounded:true,extra_hourly:'0'}}}};++ const adapters={hetzner:{quote:async()=>[q],provision:async()=>{created++;if(uncertain)throw new Error('timeout');return {id:'remote'};},health:async()=>({deleted,state:'running'}),release:async()=>{removed++;}}};++ const desks={enabled:true,hosts:async()=>[{machine_id:'m',trust:'first-party',online:true,free:4,max_browser_desks:4}]};++ let broker=createBroker({db:t.db,config,desks,adapters,now:()=>time});++ try {++ assert.equal(loadCompute({}).dry_run,true);assert.equal(loadCompute({}).providers.hetzner.monthly_cap,'0');++ config.compute.providers.hetzner.monthly_cap='0';++ assert.match((await broker.provision('browsers',q)).reason,/quote only/);assert.equal(created,0);++ config.compute.providers.hetzner.monthly_cap='2';++ config.compute.dry_run=true;assert.equal((await broker.provision('browsers',q)).dry_run,true);assert.equal(created,0);++ config.compute.dry_run=false;++ const results=await Promise.all([broker.provision('browsers',q),broker.provision('browsers',q)]);++ assert.equal(created,1);assert.equal(results.filter(r=>r.rental).length,1);assert.ok(results.some(r=>/cap/.test(r.reason)));++ const id=results.find(r=>r.rental).rental;++ await broker.bind(id,'m');++ const choice=await broker.decide({compute_pool:'browsers',owner:'user:alice',kind:'browser'},await desks.hosts());++ assert.equal(choice.selected,'m');assert.equal(choice.hourly,1);++ broker=createBroker({db:t.db,config,desks,adapters,now:()=>time});++ assert.match((await broker.provision('browsers',q)).reason,/cap/,'restart retains reservation');++ time+=3600000;await broker.tick();assert.equal(removed,1);++ let state=await broker.status();assert.equal(state.rentals[0].state,'releasing');assert.equal(state.budgets[0].committed_micros,'2000000');++ q.expires_at=time+60000;++ assert.match((await broker.provision('browsers',q)).reason,/unresolved/,'unconfirmed release blocks new provider spending');++ deleted=true;await broker.tick();state=await broker.status();assert.equal(state.rentals[0].state,'released');assert.equal(state.rentals[0].spent_micros,'1000000');++ assert.ok(state.journal.some(j=>j.event==='release.confirmed'));++ config.compute.providers.hetzner.monthly_cap='10';++ time=Date.UTC(2026,10,30,23,30);q.expires_at=time+60000;++ deleted=false;++ const cross=await broker.provision('browsers',q);++ desks.hosts=async()=>[{machine_id:'auto',name:cross.rental,trust:'community',online:true,free:4,max_browser_desks:4}];++ await broker.tick();assert.equal((await t.db.maybe('SELECT machine FROM run_compute_rentals WHERE id=$1',[cross.rental])).machine,null,'a community name cannot claim a rental');++ desks.hosts=async()=>[{machine_id:'auto',name:cross.rental,trust:'user-owned',online:true,free:4,max_browser_desks:4}];++ await broker.tick();assert.equal((await t.db.maybe('SELECT machine FROM run_compute_rentals WHERE id=$1',[cross.rental])).machine,'auto');++ const bound=await broker.decide({compute_pool:'browsers',owner:'user:alice',kind:'browser'},await desks.hosts());++ assert.equal(bound.hourly,1);assert.equal(bound.tier,'dedicated','rental retains provider tier and cost');++ const ownOnly=await broker.decide({compute_pool:'browsers',compute_strategy:'own',owner:'user:alice',kind:'browser'},await desks.hosts());++ assert.equal(ownOnly.selected,null,'rental is not an own-machine placement');++ time+=3600000;++ deleted=true;++ await broker.release(await t.db.maybe('SELECT * FROM run_compute_rentals WHERE id=$1',[cross.rental]));++ const charges=await t.db.many('SELECT * FROM run_compute_charges WHERE rental=$1 ORDER BY month',[cross.rental]);++ assert.deepEqual(charges.map(c=>[c.month,String(c.amount_micros)]),[['2026-11','500000'],['2026-12','500000']]);++ assert.equal((await broker.status()).budgets[0].spent_micros,'500000');++ q.expires_at=time+60000;deleted=false;++ config.compute.pools.browsers.min_warm=0;desks.hosts=async()=>[];++ const lowered=await broker.provision('browsers',q);++ config.compute.providers.hetzner.monthly_cap='2.25';++ await broker.tick();++ assert.equal((await t.db.maybe('SELECT state FROM run_compute_rentals WHERE id=$1',[lowered.rental])).state,'releasing','lowered cap includes already settled spend, not only live reservations');++ deleted=true;await broker.tick();config.compute.providers.hetzner.monthly_cap='10';++ q.expires_at=time+60000;uncertain=true;++ assert.match((await broker.provision('browsers',q)).reason,/unknown/);++ assert.match((await broker.provision('browsers',q)).reason,/unresolved/);++ time=Date.UTC(2027,0,1);q.expires_at=time+60000;++ assert.match((await broker.provision('browsers',q)).reason,/unresolved/,'month rollover cannot forget an unknown purchase');++ const privateChoice=await broker.decide({compute_pool:'browsers',compute_strategy:'private',owner:'user:alice',kind:'browser'},[{machine_id:'community',trust:'community',online:true,free:10}]);++ assert.equal(privateChoice.selected,null);++ console.log('compute cap, concurrent reservation, dry run, restart, release confirmation, unknown outcome and rollover: passed');++ } finally {await t.close();}++})().catch(e=>{console.error(e);process.exit(1);});
@@ -256,3 +256,7 @@ until Run has a product surface; the vhost below is installed then.
@@ -256,3 +256,7 @@ until Run has a product surface; the vhost below is installed then.
- openvibe-sdk: v0.94.0
- openvibe-sdk: v0.94.0
- openvibe-shared: v4.153.0
- openvibe-shared: v4.153.0
<!-- versions:end -->
<!-- versions:end -->
+
+Compute placement, pool autoscaling, rental caps and the operator view are documented in
+[docs/compute-broker.md](docs/compute-broker.md). Rentals default to quote-only; keys and caps
+are never inferred from available credentials.
@@ -8,11 +8,11 @@
@@ -8,11 +8,11 @@
"port": 4920,
"port": 4920,
"domain": "openvibe.run",
"domain": "openvibe.run",
"contracts": "openvibe-contracts v0.338.0",
"contracts": "openvibe-contracts v0.338.0",
- "sdk": "openvibe-sdk v0.94.0",
+ "sdk": "openvibe-sdk v0.100.0",
"shared": "openvibe-shared v4.153.0",
"shared": "openvibe-shared v4.153.0",
"packages": [
"packages": [
"openvibe-contracts v0.338.0",
"openvibe-contracts v0.338.0",
- "openvibe-sdk v0.94.0",
+ "openvibe-sdk v0.100.0",
"openvibe-shared v4.153.0"
"openvibe-shared v4.153.0"
],
],
"capabilities": [
"capabilities": [
new file mode 100644
new file mode 100644
@@ -0,0 +1,98 @@
@@ -0,0 +1,98 @@
+# Computer pools and compute broker
+
+Run owns scheduling, session leases, autoscaling and rental accounting. Computer is the front door;
+Exchange shows GPU quotes; AI retains GPU inference runtimes and its nano-USD ledger.
+
+## Existing machines and strategies
+
+Set `RUN_COMPUTE_POOLS` to a JSON object keyed by pool name, and `RUN_COMPUTE_INVENTORY` to an
+operator-controlled JSON object keyed by Bot machine id. These are non-secret configuration.
+For example, a browser pool has `kind: "browser"`, `strategy: "automatic"`, `currency: "EUR"`,
+`min_warm: 1`, `max_machines: 3`, `slots_per_machine: 8`, `target_wait_ms: 30000`,
+`idle_grace_ms: 600000`, `max_hourly: 0.10`, `providers: ["hetzner", "ovh"]`.
+A machine entry has `pool`, `tier` (`home`, `dedicated`, `community`, `partner`), `ram_mb`,
+`measured_peak_ram_mb`, `region`, `hourly`, `currency`, optionally measured `latency_ms`,
+`queue_ms`, `gpu_mb`, and `egress: "ovh-tunnel"`. Own-machine identity always comes from Bot.
+Public offers cannot claim an OpenVibe trust tier. Missing RAM/price means ineligible, not free.
+
+`POST /api/v1/sessions?pool=browser&strategy=cheapest` keeps the released request contract.
+Strategies are `automatic`, `cheapest`, `fastest`, `private`, `own`; the request overrides the
+pool default. The pool and strategy are part of idempotency and persist on the session.
+`GET /api/v1/sessions/:id?explain=1` adds the recorded decision, tier, price and reasons for its
+owner. Normal reads remain contract-compatible. Fastest uses measured latency; an unmeasured
+host has no advantage. Region and privacy constraints never relax to satisfy a queue.
+
+Automatic walks own → home → dedicated → community → partner, rejecting health, capacity,
+queue-wait and price failures first. New rentals are ranked with their configured disk/network
+allowance; unavailable keys or exhausted provider caps do not hide eligible alternatives.
+Home public egress requires an OVH tunnel. RAM packing
+uses at least 300 MB/browser and 2048 MB/desktop, subtracts 1 GB for the system, and respects
+Bot's slot ceiling. Configure measured peaks from load tests; these defaults are not measurements.
+
+For jobs, `RUN_COMPUTE_JOB_POOL` applies the pool policy to Fabric offers after the existing
+SDK planner's hard checks. Inventory `free_ram_mb` and `free_slots` cover offers whose public
+projection omits capacity. Job `requirements.objective` selects cheapest, lowest-latency or
+private; an explicit `node` remains pinned. Draining rentals reject both sessions and jobs.
+The dispatch result retains placement reasons and estimated hourly cost.
+
+## Rental switches: all default off
+
+- `RUN_COMPUTE_ENABLED=1` enables pool reconciliation; `RUN_COMPUTE_DRY_RUN=0` permits provision.
+ Dry run defaults on. Cleanup of already-created rentals continues when either switch is disabled.
+- `RUN_COMPUTE_OWNER` is the Network subject whose Bot host list the operator manages.
+- `RUN_COMPUTE_OPERATORS` is a comma-separated allowlist of Network user subject ids.
+- For each provider prefix `RUN_COMPUTE_HETZNER_`, `RUN_COMPUTE_OVH_`, `RUN_COMPUTE_VAST_`,
+ `RUN_COMPUTE_RUNPOD_`, set `KEY`, `MONTHLY_CAP`, `CURRENCY`, and `IMAGE`. Cap is a decimal
+ amount in that currency, default `0`. Image must be a prepared, isolated Link worker image.
+ Never put credentials in the inventory, pool JSON or image. Enrollment is an operator step.
+- Also set `EXTRA_HOURLY` to a conservative bound for disk/egress and `COSTS_BOUNDED=1` only
+ after enforcing that bound in the worker/network. Without both, live provisioning is refused.
+ Public catalogue prices alone do not bound bandwidth or persistent-volume charges.
+- OVH additionally needs `PROJECT`, `APPLICATION_SECRET`, and `CONSUMER_KEY`; `KEY` is the
+ application key. This adapter uses EU API/IE catalogue, EUR including catalogue tax. Hetzner
+ Cloud quotes EUR including tax and provisions IPv6 only. Vast and RunPod quote USD.
+ Do not mix currencies or guess an exchange rate.
+- GPU inference should use AI's existing `ai.gpus` settings and ledger. Leave Run's Vast and
+ RunPod caps at zero unless intentionally operating a separate worker pool with a separate budget.
+
+No-key quotes: auction, OVH public catalogue, Vast, RunPod, and community offer reads, subject to
+upstream availability. Hetzner Cloud needs a token; OVH needs credentials to resolve live project
+flavors. Auction purchase is quote-only pending cancellation/liability implementation. Community
+inference offers are not hourly machine rentals; use AI's existing routing or an existing Bot desk.
+
+The broker reserves a bounded runtime (at most one hour) plus a deletion allowance before calling
+a provider. Integer micro-currency arithmetic and database locks serialize cap checks. Unknown
+creation outcomes block new rentals and retain their reservation across restarts and months.
+A pending release also blocks new purchases from that provider until deletion is confirmed.
+Every provision, release, drain and strategy change is journalled. Deletion must be confirmed by
+provider health before releasing the reservation; a refused deletion continues to accrue cost and
+can exceed a reservation. The journal flags this. No software can guarantee an upstream provider
+will honor deletion while unavailable; the operator must handle failed releases promptly.
+
+## Operator view and lifecycle
+
+`GET /api/v1/compute` shows pools, strategies, machines, sessions, current hourly cost, spend,
+committed budget and cap; `?format=json` gives the same data. Only allowlisted signed-in operators
+can read it. Strategy forms require same-origin writes. Quotes are at
+`GET /api/v1/compute/pools/:pool/quotes`; they never spend.
+
+A prepared worker that enrolls a fresh Link identity under `RUN_COMPUTE_OWNER` and reports the
+provider-assigned rental label as its machine name is bound automatically. Bot must identify it
+as that owner's own machine; community/shared name claims are never accepted. Its quote supplies
+the pool and RAM inventory. The image's enrollment mechanism is an operator prerequisite.
+
+For workers without this naming integration, put the machine id in trusted inventory and bind it
+with `POST /api/v1/compute/rentals/:id/bind`, JSON `{ "machine": "dev_…" }`, as the operator with
+an Origin matching Run. Automatic secure enrollment itself is not implemented here; do not bake reusable Node credentials into an image. Unbound rentals
+count as starting capacity and are deleted at their deadline; never regarded as usable desktops.
+
+Ordinary scale-down marks a host draining, excludes it from placement, waits for zero sessions,
+then requests deletion. At the runtime/budget limit, Run ends sessions before release. Unknown
+host observations never count as zero occupancy. Each rental's stop deadline survives restart.
+Desktop sizing is available, but the current released Bot session API creates browser desks only;
+full Linux desktop execution still needs the Computer/Node runtime work and security review.
+
+Checks: `npm test -- compute sessions dispatch`; the full suite is `OV_ALLOW_SUITE=1 npm test`.
+The SDK dependency is pinned to the reviewed job commit until its PR is merged and released;
+replace that pin with the resulting SDK release before shipping. Do not deploy this PR before
+Opus reviews money, worker enrollment, privacy and cancellation behavior.
new file mode 100644
new file mode 100644
@@ -0,0 +1,22 @@
@@ -0,0 +1,22 @@
+-- phase: expand
+-- Expand only. Operator budgets are configuration; all financial commitments survive process restarts.
+CREATE TABLE IF NOT EXISTS run_compute_budget_locks (provider text PRIMARY KEY);
+CREATE TABLE IF NOT EXISTS run_compute_rentals (
+ id text PRIMARY KEY, pool text NOT NULL, provider text NOT NULL, remote_id text, machine text,
+ state text NOT NULL, month text NOT NULL, currency text NOT NULL, hourly_micros numeric(30,0) NOT NULL,
+ reserved_micros numeric(30,0) NOT NULL, spent_micros numeric(30,0) NOT NULL DEFAULT 0,
+ quote jsonb NOT NULL, created_at bigint NOT NULL, deadline bigint NOT NULL, released_at bigint,
+ idle_since bigint, last_error text
+);
+CREATE INDEX IF NOT EXISTS run_compute_rentals_live ON run_compute_rentals(state, provider);
+CREATE TABLE IF NOT EXISTS run_compute_journal (
+ id bigserial PRIMARY KEY, at bigint NOT NULL, event text NOT NULL, rental text, pool text, detail jsonb NOT NULL
+);
+CREATE TABLE IF NOT EXISTS run_compute_pools (id text PRIMARY KEY, strategy text NOT NULL, last_machine text, last_tier text);
+ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS compute_pool text;
+ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS compute_strategy text;
+ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS compute_decision jsonb;
+CREATE TABLE IF NOT EXISTS run_compute_charges (
+ rental text NOT NULL REFERENCES run_compute_rentals(id), month text NOT NULL,
+ amount_micros numeric(30,0) NOT NULL, PRIMARY KEY (rental, month)
+);
@@ -16,7 +16,7 @@
@@ -16,7 +16,7 @@
"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.338.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.153.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.153.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -1163,9 +1163,9 @@
@@ -1163,9 +1163,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.94.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",
- "integrity": "sha512-p3buZSOSFV81OdZgs5GoSIfMCQNJjW23eAIXIt0RVFIOvh3TByNNEMmuCanxRybFIbzh+9IDMbBwHJrBLl3JeA==",
+ "version": "0.100.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0",
+ "integrity": "sha512-P7Mi0jqC4PACvNZKgZQqGumUYSAin5X1B1ZTpyvcZ/6P9CIg6xDZew6Zmn/Hqp5VafqVKg09Y6VbEBjj2krtQg==",
"license": "MIT",
"license": "MIT",
"bin": {
"bin": {
"openvibe-download": "bin/openvibe-download.js"
"openvibe-download": "bin/openvibe-download.js"
@@ -27,7 +27,7 @@
@@ -27,7 +27,7 @@
"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.338.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.153.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.153.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -80,7 +80,7 @@ function sessionsRouter({ sessions }) {
@@ -80,7 +80,7 @@ function sessionsRouter({ sessions }) {
r.post('/sessions', wrap(async (req, res) => {
r.post('/sessions', wrap(async (req, res) => {
const w = who(req, res, 'run.session.create');
const w = who(req, res, 'run.session.create');
if (!w) return;
if (!w) return;
- const { row, created } = await sessions.create(w.owner, w.via, req.body && typeof req.body === 'object' ? req.body : {});
+ const { row, created } = await sessions.create(w.owner, w.via, req.body && typeof req.body === 'object' ? req.body : {}, { pool: req.query.pool, strategy: req.query.strategy });
res.status(created ? 201 : 200).json(sessions.present(row));
res.status(created ? 201 : 200).json(sessions.present(row));
}));
}));
r.get('/sessions', wrap(async (req, res) => {
r.get('/sessions', wrap(async (req, res) => {
@@ -93,7 +93,7 @@ function sessionsRouter({ sessions }) {
@@ -93,7 +93,7 @@ function sessionsRouter({ sessions }) {
}));
}));
r.get('/sessions/:id', wrap(async (req, res) => {
r.get('/sessions/:id', wrap(async (req, res) => {
const row = await mine(req, res, 'run.session.read');
const row = await mine(req, res, 'run.session.read');
- if (row) res.json(sessions.present(row));
+ if (row) res.json(req.query.explain === '1' ? { ...sessions.present(row), compute: row.compute_decision || null } : sessions.present(row));
}));
}));
r.post('/sessions/:id/commands', wrap(async (req, res) => {
r.post('/sessions/:id/commands', wrap(async (req, res) => {
const row = await mine(req, res, 'run.session.control');
const row = await mine(req, res, 'run.session.control');
@@ -62,7 +62,16 @@ function createApp(opts = {}) {
@@ -62,7 +62,16 @@ function createApp(opts = {}) {
const apiAuth = createApiAuth({ config, keys, userAuth, membership, log });
const apiAuth = createApiAuth({ config, keys, userAuth, membership, log });
// Run sessions (contracts 0.223.0): browser desks Bot leases per session; the tick starts in server/index.js.
// Run sessions (contracts 0.223.0): browser desks Bot leases per session; the tick starts in server/index.js.
const desks = opts.desks || createDeskClient({ config, fetchImpl, now });
const desks = opts.desks || createDeskClient({ config, fetchImpl, now });
- const sessions = opts.sessions || createSessions({ config, db, desks, outbox, now, log });
+ const broker = opts.broker || require('./compute/broker').createBroker({ config, db, desks, now, fetchImpl, log });
+ const sessions = opts.sessions || createSessions({ config, db, desks, outbox, now, log, broker });
+ broker.attachSessions(async (id) => { const row = await sessions.get(id); if (row) await sessions.finish(row, 'ttl', 'compute rental budget exhausted'); });
+ broker.attachJobs(async (machine) => {
+ const jobs = require('./jobs/store');
+ for (const row of await db.many("SELECT * FROM run_jobs WHERE node_id=$1 AND state IN ('placed','running')", [machine])) {
+ const result = await jobs.cancel(db, row.id, row.project_id, now());
+ if (result.row) await dispatch.cancel(result.row);
+ }
+ });
const release = require('openvibe-shared/release').createRelease({ service: 'run', root: path.join(__dirname, '..') });
const release = require('openvibe-shared/release').createRelease({ service: 'run', root: path.join(__dirname, '..') });
const app = express();
const app = express();
@@ -120,6 +129,7 @@ function createApp(opts = {}) {
@@ -120,6 +129,7 @@ function createApp(opts = {}) {
express.json({ limit: JSON_LIMIT }),
express.json({ limit: JSON_LIMIT }),
(req, res, next) => { res.setHeader('Cache-Control', 'no-store'); next(); },
(req, res, next) => { res.setHeader('Cache-Control', 'no-store'); next(); },
apiAuth.middleware,
apiAuth.middleware,
+ require('./compute/routes').computeRouter({ broker, config }),
sessionsRouter({ sessions }),
sessionsRouter({ sessions }),
v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now, log }));
v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now, log }));
@@ -202,7 +212,7 @@ function createApp(opts = {}) {
@@ -202,7 +212,7 @@ function createApp(opts = {}) {
return res.status(500).type('text/plain').send('Something went wrong\n');
return res.status(500).type('text/plain').send('Something went wrong\n');
});
});
- Object.assign(app.locals, { config, db, indexnow, valkey, keys, outbox, events, stream, tickets, dispatch, apiAuth, metrics, desks, sessions });
+ Object.assign(app.locals, { config, db, indexnow, valkey, keys, outbox, events, stream, tickets, dispatch, apiAuth, metrics, desks, sessions, broker });
return app;
return app;
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,237 @@
@@ -0,0 +1,237 @@
+'use strict';
+const { randomUUID }=require('node:crypto');
+const {choose,scale,micros,charge,createProviders,STRATEGIES}=require('openvibe-sdk/compute');
+const obj=v=>typeof v==='string'?JSON.parse(v):v;
+const bare=v=>String(v || '').replace(/^user:/,'');
+function createBroker({db,config,desks,now=Date.now,fetchImpl=globalThis.fetch,adapters,log=console}) {
+ const cfg=config.compute || {pools:{},inventory:{},providers:{},operators:[],dry_run:true};
+ const providers=adapters || createProviders({fetch:fetchImpl,config:cfg.providers,now});
+ let timer=null,running=null,finish=null,finishJobs=null;
+ const month=()=>new Date(now()).toISOString().slice(0,7);
+ const journal=(event,detail={},r={})=>db.query('INSERT INTO run_compute_journal (at,event,rental,pool,detail) VALUES ($1,$2,$3,$4,$5)',[now(),event,r.id || null,r.pool || null,JSON.stringify(detail)]);
+ const live=()=>db.many("SELECT * FROM run_compute_rentals WHERE state <> 'released'");
+ async function poolConfig(id) {
+ const p=Object.hasOwn(cfg.pools,id)?cfg.pools[id]:null; if(!p) throw Object.assign(new Error('unknown compute pool'),{status:422,code:'run.compute.pool'});
+ const saved=await db.maybe('SELECT * FROM run_compute_pools WHERE id=$1',[id]);
+ return {...p,strategy:saved?.strategy || p.strategy || 'automatic',last_machine:saved?.last_machine};
+ }
+ async function inventory(hosts,owner) {
+ const rentals=await live();
+ const counts=await db.many("SELECT machine, count(*)::int AS n FROM run_sessions WHERE state IN ('running','starting') AND machine IS NOT NULL GROUP BY machine");
+ const jobs=await db.many("SELECT node_id,count(*)::int AS n FROM run_jobs WHERE state IN ('placed','running') GROUP BY node_id");
+ return hosts.map(h=>{
+ const rental=rentals.find(r=>r.machine===h.machine_id);
+ const rented=rental?obj(rental.quote):{};
+ const v={...(rental?{pool:rental.pool,ram_mb:rented.ram_mb,region:rented.region,tier:rented.tier}:{}),...(cfg.inventory[h.machine_id] || {})};
+ const ramPerSession=Math.max(cfg.pools[v.pool]?.kind==='desktop'?2048:300,Number(v.measured_peak_ram_mb) || 0,Number(cfg.pools[v.pool]?.measured_peak_ram_mb) || 0);
+ const sessions=Math.max(counts.find(c=>c.machine===h.machine_id)?.n || 0,(h.max_browser_desks || 0)-(h.free || 0))+(jobs.find(j=>j.node_id===h.machine_id)?.n || 0);
+ return {...v,id:h.machine_id,owner:h.trust==='user-owned'?bare(owner):null,
+ tier:rental?(rented.tier || 'partner'):h.trust==='user-owned'?'own':h.trust==='community'?'community':h.trust==='first-party'?(v.tier || 'dedicated'):'partner',
+ hourly:rental?Number(rental.hourly_micros)/1e6:(h.trust==='user-owned'?0:v.hourly ?? null),currency:rental?.currency || v.currency || 'USD',
+ health:h.online?'up':'down',free_slots:Number(h.free) || 0,
+ free_ram_mb:Number(h.free_ram_mb ?? (v.ram_mb!=null?Math.max(0,v.ram_mb-1024-sessions*ramPerSession):0)),
+ sessions,draining:rental?['draining','releasing','uncertain'].includes(rental.state):Boolean(v.draining),managed:Boolean(rental),
+ state:rental?.state || 'running',idle_since:rental?.idle_since==null?undefined:Number(rental.idle_since)};
+ });
+ }
+ async function decide(row,hosts) {
+ const pool=await poolConfig(row.compute_pool);
+ const rentalMachines=await inventory(hosts,row.owner);
+ const allowed=row.requirements?.trust;
+ const filtered=hosts.filter(h=>!allowed || allowed.includes(h.trust)).map(h=>h.machine_id);
+ const candidates=rentalMachines.filter(m=>filtered.includes(m.id) && (!m.pool || m.pool===row.compute_pool));
+ const d=choose({strategy:row.compute_strategy || pool.strategy,owner:bare(row.owner),kind:row.kind,machine:row.machine_asked || undefined,
+ region:row.requirements?.region==='nearest'?undefined:row.requirements?.region,max_hourly:row.requirements?.max_cost_usd == null ? undefined : row.requirements.max_cost_usd / Math.max(1,row.ttl_s || 3600) * 3600,public_egress:pool.public_egress!==false,ram_mb:pool.measured_peak_ram_mb},candidates,pool,{now:now(),current:null});
+ if(d.selected) {
+ const prev=await db.maybe('SELECT * FROM run_compute_pools WHERE id=$1',[row.compute_pool]);
+ if(prev?.last_tier && prev.last_tier!==d.tier) await journal('strategy.switch',{from:prev.last_tier,to:d.tier,strategy:d.strategy,reasons:d.reasons},{pool:row.compute_pool});
+ await db.query('INSERT INTO run_compute_pools (id,strategy,last_machine,last_tier) VALUES ($1,$2,$3,$4) ON CONFLICT (id) DO UPDATE SET last_machine=EXCLUDED.last_machine,last_tier=EXCLUDED.last_tier',[row.compute_pool,pool.strategy,d.selected,d.tier]);
+ }
+ return d;
+ }
+ async function quotes(poolId) {
+ const p=await poolConfig(poolId);
+ const results=await Promise.all((p.providers || []).map(async name=>{
+ if(!providers[name]) return {provider:name,quotes:[],error:'unsupported provider'};
+ try {return {provider:name,quotes:await providers[name].quote({gpu_mb:p.gpu_mb || 0})};}
+ catch {return {provider:name,quotes:[],error:'catalogue unavailable or credentials required'};}
+ }));
+ return results;
+ }
+ async function reserve(pool,q) {
+ const pc=cfg.providers[q.provider] || {};
+ if(!pc.key || micros(pc.monthly_cap || '0')===0n) return {reason:'quote only: key and positive monthly cap required'};
+ if(q.provisionable===false || !pc.image || !pc.costs_bounded || pc.extra_hourly==null) return {reason:'quote only: prepared image and bounded disk/egress costs required'};
+ if(q.currency!==pc.currency || q.expires_at<now() || !Number.isFinite(q.hourly) || q.hourly<=0) return {reason:'invalid or expired quote'};
+ const p=await poolConfig(pool);
+ const hourly=BigInt(Math.ceil(q.hourly*1e6))+micros(pc.extra_hourly);
+ if(p.max_hourly!=null && hourly>micros(p.max_hourly)) return {reason:'total price exceeds pool ceiling'};
+ const runtime=Math.min(3600000,Math.max(60000,p.max_runtime_ms || 3600000));
+ // Reserve one extra billing quantum for asynchronous deletion. Stop before the reservation's deadline.
+ const reserved=charge(hourly,runtime+300000,q.billing_seconds)+micros(q.setup || 0);
+ const cap=micros(pc.monthly_cap);
+ return db.tx(async t=>{
+ await t.query('INSERT INTO run_compute_budget_locks (provider) VALUES ($1) ON CONFLICT DO NOTHING',['pool:'+pool]);
+ await t.maybe('SELECT provider FROM run_compute_budget_locks WHERE provider=$1 FOR UPDATE',['pool:'+pool]);
+ await t.query('INSERT INTO run_compute_budget_locks (provider) VALUES ($1) ON CONFLICT DO NOTHING',[q.provider]);
+ await t.maybe('SELECT provider FROM run_compute_budget_locks WHERE provider=$1 FOR UPDATE',[q.provider]);
+ const poolRows=await t.many("SELECT id FROM run_compute_rentals WHERE pool=$1 AND state <> 'released'",[pool]);
+ if(poolRows.length >= (p.max_machines ?? 10)) return {reason:'pool machine limit reached'};
+ const rows=await t.many("SELECT * FROM run_compute_rentals WHERE provider=$1 AND (month=$2 OR state <> 'released')",[q.provider,month()]);
+ if(rows.some(r=>['uncertain','releasing'].includes(r.state))) return {reason:'an unresolved provider call blocks new rentals'};
+ const settled=await t.value('SELECT COALESCE(SUM(c.amount_micros),0) FROM run_compute_charges c JOIN run_compute_rentals r ON r.id=c.rental WHERE r.provider=$1 AND c.month=$2',[q.provider,month()]);
+ const committed=BigInt(settled)+rows.filter(r=>r.state!=='released').reduce((sum,r)=>sum+BigInt(r.reserved_micros),0n);
+ if(committed+reserved>cap) return {reason:'monthly cap reached'};
+ const r={id:'ov-'+randomUUID(),pool,provider:q.provider,month:month(),currency:q.currency,created_at:now(),deadline:now()+runtime,quote:q};
+ await t.query("INSERT INTO run_compute_rentals (id,pool,provider,state,month,currency,hourly_micros,reserved_micros,quote,created_at,deadline) VALUES ($1,$2,$3,'provisioning',$4,$5,$6,$7,$8,$9,$10)",[r.id,pool,r.provider,r.month,r.currency,hourly.toString(),reserved.toString(),JSON.stringify(q),r.created_at,r.deadline]);
+ await t.query('INSERT INTO run_compute_journal (at,event,rental,pool,detail) VALUES ($1,$2,$3,$4,$5)',[now(),'provision.reserved',r.id,pool,JSON.stringify({reserved_micros:reserved.toString(),hourly_micros:hourly.toString(),quote:q.id})]);
+ return {rental:r};
+ });
+ }
+ async function provision(pool,q) {
+ if(cfg.dry_run || !cfg.enabled) return {dry_run:true,provider:q.provider,quote:q.id,hourly:q.hourly,currency:q.currency};
+ const result=await reserve(pool,q); if(!result.rental) return result;
+ const r=result.rental;
+ try {
+ const remote=await providers[q.provider].provision(q,{label:r.id,image:cfg.providers[q.provider].image});
+ if(!remote?.id) throw new Error('no remote id');
+ await db.query("UPDATE run_compute_rentals SET remote_id=$2,state='starting' WHERE id=$1",[r.id,String(remote.id)]);
+ await journal('provision.accepted',{remote_id:String(remote.id)},r);
+ return {rental:r.id};
+ } catch {
+ await db.query("UPDATE run_compute_rentals SET state='uncertain',last_error='provision outcome unknown; reservation retained' WHERE id=$1",[r.id]);
+ await journal('provision.uncertain',{},r); return {reason:'provider outcome unknown; reservation retained'};
+ }
+ }
+ async function release(r) {
+ if(!r.remote_id) return;
+ await db.query("UPDATE run_compute_rentals SET state='releasing' WHERE id=$1",[r.id]);
+ await journal('release.requested',{},r);
+ try {
+ await providers[r.provider].release(r.remote_id);
+ const h=await providers[r.provider].health(r.remote_id);
+ if(!h.deleted) return; // A stop or accepted DELETE is not proof that billing ended.
+ const q=obj(r.quote);
+ const spent=charge(BigInt(r.hourly_micros),now()-Number(r.created_at),q.billing_seconds)+micros(q.setup || 0);
+ await db.tx(async t=>{
+ const locked=await t.maybe('SELECT state FROM run_compute_rentals WHERE id=$1 FOR UPDATE',[r.id]);
+ if(locked.state==='released')return;
+ const begin=Number(r.created_at),end=now();
+ const setup=micros(q.setup || 0),runtime=spent-setup;
+ let allocated=0n;
+ for(let start=begin;start<end;) {
+ const d=new Date(start),boundary=Math.min(end,Date.UTC(d.getUTCFullYear(),d.getUTCMonth()+1,1));
+ const share=boundary===end?runtime-allocated:runtime*BigInt(boundary-start)/BigInt(end-begin);
+ const value=share+(start===begin?setup:0n);allocated+=share;
+ await t.query('INSERT INTO run_compute_charges (rental,month,amount_micros) VALUES ($1,$2,$3) ON CONFLICT DO NOTHING',[r.id,new Date(start).toISOString().slice(0,7),value.toString()]);
+ start=boundary;
+ }
+ await t.query("UPDATE run_compute_rentals SET state='released',spent_micros=$2,released_at=$3,last_error=NULL WHERE id=$1",[r.id,spent.toString(),now()]);
+ });
+ await journal('release.confirmed',{spent_micros:spent.toString(),over_reservation:spent>BigInt(r.reserved_micros)},r);
+ } catch {await db.query("UPDATE run_compute_rentals SET last_error='release unconfirmed; billing reservation retained' WHERE id=$1",[r.id]);}
+ }
+ async function tickOnce() {
+ const rentals=await live();
+ for(const r of rentals) {
+ const pc=cfg.providers[r.provider] || {};
+ if(r.state==='provisioning') { // another process may still be completing the call; do not retry it
+ if(now()-Number(r.created_at)>60000) await db.query("UPDATE run_compute_rentals SET state='uncertain' WHERE id=$1 AND state='provisioning'",[r.id]);
+ continue;
+ }
+ if(r.state==='releasing') {await release(r); continue;}
+ const settled=BigInt(await db.value('SELECT COALESCE(SUM(c.amount_micros),0) FROM run_compute_charges c JOIN run_compute_rentals p ON p.id=c.rental WHERE p.provider=$1 AND c.month=$2',[r.provider,month()]));
+ const committed=rentals.filter(x=>x.provider===r.provider).reduce((sum,x)=>sum+BigInt(x.reserved_micros),settled);
+ let unhealthy=false;
+ if(r.remote_id) { try { const h=await providers[r.provider].health(r.remote_id); unhealthy=h.deleted || ['error','stopped','off','dead'].includes(String(h.state).toLowerCase()); } catch { /* unknown health keeps the reservation and deadline */ } }
+ const stop=!cfg.enabled || now()>=Number(r.deadline) || committed>micros(pc.monthly_cap || '0') || unhealthy;
+ if(stop && r.remote_id) {
+ await db.query("UPDATE run_compute_rentals SET state='draining' WHERE id=$1",[r.id]);
+ if(r.machine && finish) for(const s of await db.many("SELECT id FROM run_sessions WHERE machine=$1 AND state <> 'ended'",[r.machine])) await finish(s.id);
+ if(r.machine && finishJobs) await finishJobs(r.machine);
+ const jobs=r.machine ? await db.value("SELECT count(*)::int FROM run_jobs WHERE node_id=$1 AND state IN ('placed','running')",[r.machine]):0;
+ const count= r.machine ? await db.value("SELECT count(*)::int FROM run_sessions WHERE machine=$1 AND state <> 'ended'",[r.machine]):0;
+ if(!count && !jobs) await release(r);
+ }
+ }
+ if(!cfg.enabled || !cfg.owner || !desks.enabled) return;
+ let hosts; try {hosts=await desks.hosts(cfg.owner);} catch {return;}
+ // Bot returns a name only for this authenticated owner's own machines. A community seller cannot
+ // claim a paid rental by copying its label. The prepared image must enroll a fresh Link identity.
+ for(const r of await live()) {
+ if(r.state!=='starting' || r.machine || !r.remote_id)continue;
+ const matches=hosts.filter(h=>h.trust==='user-owned' && h.name===r.id && h.online);
+ if(matches.length!==1)continue;
+ const updated=await db.maybe("UPDATE run_compute_rentals SET machine=$2,state='running' WHERE id=$1 AND machine IS NULL AND state='starting' RETURNING id",[r.id,matches[0].machine_id]);
+ if(updated)await journal('machine.bound',{machine:matches[0].machine_id,source:'authenticated owner host label'},r);
+ }
+
+ for(const id of Object.keys(cfg.pools)) {
+ const p=await poolConfig(id);
+ const all=await inventory(hosts,cfg.owner);
+ const rows=(await live()).filter(r=>r.pool===id);
+ const ms=all.filter(m=>m.pool===id || rows.some(r=>r.machine===m.id));
+ for(const r of rows) {
+ const m=ms.find(m=>m.id===r.machine);
+ if(m) {
+ if(m.sessions===0 && r.idle_since==null) await db.query('UPDATE run_compute_rentals SET idle_since=$2 WHERE id=$1',[r.id,now()]);
+ if(m.sessions>0 && r.idle_since!=null) await db.query('UPDATE run_compute_rentals SET idle_since=NULL WHERE id=$1',[r.id]);
+ } else ms.push({id:r.id,state:'provisioning',health:'starting',sessions:null});
+ }
+ const waiting=await db.many("SELECT created_at,compute_strategy FROM run_sessions WHERE compute_pool=$1 AND state='placing' AND machine_asked IS NULL AND (compute_strategy IS NULL OR compute_strategy <> 'own') ORDER BY created_at",[id]);
+ const desired=scale(p,ms,waiting.map(s=>({created_at:Number(s.created_at)})),{now:now()});
+ for(const a of desired.actions) {
+ const r=rows.find(r=>r.machine===a.id || r.id===a.id);
+ if(a.action==='drain' && r && !cfg.dry_run) {await db.query("UPDATE run_compute_rentals SET state='draining' WHERE id=$1",[r.id]);await journal('drain.started',{reason:a.reason},r);}
+ if(a.action==='release' && r && !cfg.dry_run) await release(r);
+ if(a.action==='provision') {
+ const qs=(await quotes(id)).flatMap(x=>x.quotes).filter(q=>q.provisionable!==false && q.hourly!=null && q.currency===(p.currency || 'USD') && q.ram_mb>=((p.kind==='desktop'?2048:300)+1024));
+ let candidates=qs.flatMap(q=>{
+ const pc=cfg.providers[q.provider] || {};
+ if(!cfg.dry_run && (!pc.key || micros(pc.monthly_cap || '0')===0n || !pc.image || !pc.costs_bounded || pc.extra_hourly==null))return [];
+ const extra=pc.extra_hourly==null?0:Number(micros(pc.extra_hourly))/1e6;
+ return [{...q,hourly:q.hourly+extra,id:q.provider+':'+q.id,health:'up',free_ram_mb:q.ram_mb-1024,free_slots:p.slots_per_machine || 1}];
+ });
+ while(true) {
+ const d=choose({kind:p.kind,strategy:waiting[0]?.compute_strategy || p.strategy,region:p.region,gpu_mb:p.gpu_mb},candidates,p,{now:now()});
+ const q=qs.find(q=>q.provider+':'+q.id===d.selected);
+ if(!q){await journal('scale.decision',{reason:a.reason,decision:d,result:{reason:'no eligible rental quote'}},{pool:id});break;}
+ const result=await provision(id,q);
+ await journal('scale.decision',{reason:a.reason,decision:d,result},{pool:id});
+ if(result.rental || result.dry_run || /unknown/.test(result.reason || ''))break;
+ // A provider's exhausted cap or pending release must not hide eligible alternatives.
+ candidates=candidates.filter(c=>c.provider!==q.provider);
+ }
+ break; // one fresh quote/reservation per pool/pass, observes previous launches on next pass
+ }
+ }
+ }
+ }
+ function tick(){if(running)return running;running=tickOnce().finally(()=>{running=null;});return running;}
+ async function status() {
+ const rentals=await db.many('SELECT * FROM run_compute_rentals ORDER BY created_at DESC LIMIT 200');
+ const budgets=[];
+ for(const [provider,pc] of Object.entries(cfg.providers || {})) {
+ const rs=await db.many("SELECT * FROM run_compute_rentals WHERE provider=$1 AND (month=$2 OR state <> 'released')",[provider,month()]);
+ const settled=BigInt(await db.value('SELECT COALESCE(SUM(c.amount_micros),0) FROM run_compute_charges c JOIN run_compute_rentals r ON r.id=c.rental WHERE r.provider=$1 AND c.month=$2',[provider,month()]));
+ let spent=settled,committed=settled,hourly=0n;
+ for(const r of rs.filter(r=>r.state!=='released')){const h=BigInt(r.hourly_micros);spent+=charge(h,now()-Math.max(Number(r.created_at),Date.parse(month()+'-01T00:00:00Z')),obj(r.quote).billing_seconds);committed+=BigInt(r.reserved_micros);hourly+=h;}
+ budgets.push({provider,currency:pc.currency,cap:pc.monthly_cap,configured:Boolean(pc.key),spent_micros:spent.toString(),committed_micros:committed.toString(),hourly_micros:hourly.toString()});
+ }
+ let machines=[];
+ if(cfg.owner && desks.enabled) {try {machines=await inventory(await desks.hosts(cfg.owner),cfg.owner);} catch {machines=Object.entries(cfg.inventory).map(([id,v])=>({...v,id,health:'unknown'}));}}
+ return {enabled:cfg.enabled,dry_run:cfg.dry_run,machines,pools:await Promise.all(Object.keys(cfg.pools).map(async id=>({id,...await poolConfig(id)}))),budgets,rentals,
+ sessions:await db.many("SELECT machine,compute_pool,count(*)::int AS sessions FROM run_sessions WHERE state IN ('starting','running') GROUP BY machine,compute_pool"),
+ journal:await db.many('SELECT * FROM run_compute_journal ORDER BY id DESC LIMIT 100')};
+ }
+ return {decide,inventory,quotes,reserve,provision,release,tick,status,poolConfig,
+ async available(hosts){const blocked=new Set((await live()).filter(r=>['draining','releasing','uncertain'].includes(r.state)).map(r=>r.machine));return hosts.filter(h=>!blocked.has(h.machine_id));},
+ async setStrategy(id,strategy){await poolConfig(id);if(!STRATEGIES.includes(strategy))throw Object.assign(new Error('invalid strategy'),{status:422});await db.query('INSERT INTO run_compute_pools (id,strategy) VALUES ($1,$2) ON CONFLICT(id) DO UPDATE SET strategy=EXCLUDED.strategy',[id,strategy]);await journal('strategy.changed',{strategy},{pool:id});},
+ async bind(id,machine){const r=await db.maybe("SELECT * FROM run_compute_rentals WHERE id=$1 AND state='starting'",[id]);if(!r)throw new Error('rental not starting');if(!cfg.inventory[machine] || cfg.inventory[machine].pool!==r.pool)throw new Error('machine must be in configured pool inventory');await db.query("UPDATE run_compute_rentals SET machine=$2,state='running' WHERE id=$1",[id,machine]);await journal('machine.bound',{machine},r);},
+ attachSessions(fn){finish=fn;},
+ attachJobs(fn){finishJobs=fn;},
+ start(){if(!timer){timer=setInterval(()=>tick().catch(()=>log.warn('[Run] compute reconcile failed')),cfg.tick_ms || 30000);timer.unref?.();}},
+ async stop(){if(timer)clearInterval(timer);timer=null;if(running)await running;},
+ };
+}
+module.exports={createBroker};
new file mode 100644
new file mode 100644
@@ -0,0 +1,28 @@
@@ -0,0 +1,28 @@
+'use strict';
+const { STRATEGIES, micros } = require('openvibe-sdk/compute');
+function loadCompute(env) {
+ let pools = {}, inventory = {};
+ for (const [name, put] of [['RUN_COMPUTE_POOLS', v => { pools=v; }],['RUN_COMPUTE_INVENTORY', v => { inventory=v; }]]) {
+ if (!env[name]) continue;
+ const v=JSON.parse(env[name]);
+ if (!v || typeof v !== 'object' || Array.isArray(v)) throw new Error(`${name} must be an object`);
+ put(v);
+ }
+ for (const [id,p] of Object.entries(pools)) {
+ if (!/^[a-z0-9-]{1,40}$/.test(id) || !STRATEGIES.includes(p.strategy || 'automatic')) throw new Error('invalid compute pool');
+ if (p.kind && !['browser','desktop'].includes(p.kind)) throw new Error('invalid pool kind');
+ }
+ const providers={};
+ for (const name of ['hetzner','ovh','vast','runpod']) {
+ const prefix='RUN_COMPUTE_'+name.toUpperCase()+'_';
+ const cap=env[prefix+'MONTHLY_CAP'] || '0'; micros(cap);
+ providers[name]={ key:env[prefix+'KEY'] || '', monthly_cap:cap, currency:env[prefix+'CURRENCY'] || (['hetzner','ovh'].includes(name)?'EUR':'USD'),
+ image:env[prefix+'IMAGE'] || '', extra_hourly:env[prefix+'EXTRA_HOURLY'] || null,
+ costs_bounded:env[prefix+'COSTS_BOUNDED']==='1',
+ project:env[prefix+'PROJECT'] || '', application_secret:env[prefix+'APPLICATION_SECRET'] || '', consumer_key:env[prefix+'CONSUMER_KEY'] || '' };
+ }
+ return { enabled:env.RUN_COMPUTE_ENABLED==='1', dry_run:env.RUN_COMPUTE_DRY_RUN!=='0', pools, inventory, providers,
+ operators:(env.RUN_COMPUTE_OPERATORS || '').split(',').map(s=>s.trim()).filter(Boolean),
+ owner:env.RUN_COMPUTE_OWNER || '', job_pool:env.RUN_COMPUTE_JOB_POOL || '', tick_ms:30000 };
+}
+module.exports={loadCompute};
new file mode 100644
new file mode 100644
@@ -0,0 +1,25 @@
@@ -0,0 +1,25 @@
+'use strict';
+const express=require('express');
+const {STRATEGIES}=require('openvibe-sdk/compute');
+const esc=s=>String(s ?? '').replace(/[&<>"']/g,c=>({'&':'&','<':'<','>':'>','"':'"',"'":'''}[c]));
+function computeRouter({broker,config}) {
+ const r=express.Router();
+ const wrap=fn=>(req,res,next)=>Promise.resolve(fn(req,res)).catch(next);
+ r.use('/compute',(req,res,next)=>{
+ if(req.principal?.kind!=='user' || !(config.compute?.operators || []).includes(req.principal.subject))return res.status(403).json({error:'operator access required'});
+ if(req.method!=='GET' && req.get('origin')!==config.baseUrl)return res.status(403).json({error:'same-origin write required'});
+ next();
+ });
+ r.get('/compute',wrap(async(req,res)=>{
+ const s=await broker.status();
+ if(req.query.format==='json')return res.json(s);
+ const money=v=>(Number(v)/1e6).toFixed(4);
+ const machines=[...s.machines.map(m=>({...m,machine:m.id,provider:s.rentals.find(r=>r.machine===m.id)?.provider || 'existing',hourly_micros:m.hourly==null?null:m.hourly*1e6})),...s.rentals.filter(r=>!s.machines.some(m=>m.id===r.machine)).map(r=>({...r,tier:(typeof r.quote==='string'?JSON.parse(r.quote):r.quote).tier}))];
+ res.type('html').send(`<!doctype html><html lang="en"><meta charset="utf-8"><meta name="viewport" content="width=device-width"><title>Compute pools · OpenVibe Run</title><style>:root{color-scheme:light dark;font:16px/1.6 system-ui}body{margin:0;background:light-dark(#f6f8fa,#101519);color:light-dark(#16232c,#e6edf2)}main{max-width:1100px;margin:auto;padding:24px}h1{font-size:2rem;line-height:1.2}h2{margin-top:2rem}section{padding:20px;margin:16px 0;border:1px solid #70809060;border-radius:12px}button,select{font:inherit;min-height:44px;padding:8px 12px;border-radius:6px;border:1px solid #70809080}form{display:flex;flex-wrap:wrap;gap:12px;margin:16px 0}a{color:light-dark(#075a85,#87d6ff)}table{display:block;overflow-x:auto;border-collapse:collapse;max-width:100%;text-align:left}th,td{padding:10px;border-bottom:1px solid #70809060;white-space:nowrap}ol{padding-left:24px}li{overflow-wrap:anywhere;margin:12px 0}@media(max-width:600px){main{padding:16px}h1{font-size:1.7rem}}</style><body><main><h1>Compute pools</h1><p>${s.dry_run?'Dry run — rentals disabled':'Live rentals within configured caps'} · ${s.enabled?'Reconciler enabled':'Reconciler disabled'}</p><h2>Pools</h2>${s.pools.map(p=>`<section><h3>${esc(p.id)}</h3><p>Strategy: ${esc(p.strategy)} · Warm minimum: ${esc(p.min_warm || 0)} · Queue target: ${esc(p.target_wait_ms ?? 30000)} ms</p><form method="post" action="/api/v1/compute/pools/${esc(p.id)}/strategy"><select name="strategy">${STRATEGIES.map(v=>`<option${v===p.strategy?' selected':''}>${v}</option>`).join('')}</select><button>Apply strategy</button></form><a href="/api/v1/compute/pools/${esc(p.id)}/quotes">Quotes and dry-run prices</a></section>`).join('')}<h2>Money</h2><table><tr><th>Provider</th><th>Currency</th><th>Per hour</th><th>This month</th><th>Committed</th><th>Cap</th></tr>${s.budgets.map(b=>`<tr><td>${esc(b.provider)}</td><td>${esc(b.currency)}</td><td>${money(b.hourly_micros)}</td><td>${money(b.spent_micros)}</td><td>${money(b.committed_micros)}</td><td>${esc(b.cap)}</td></tr>`).join('')}</table><h2>Machines</h2><table><tr><th>Pool</th><th>Machine / rental</th><th>Tier / provider</th><th>State</th><th>Sessions</th><th>Per hour</th></tr>${machines.map(m=>`<tr><td>${esc(m.pool)}</td><td>${esc(m.machine || m.id)}</td><td>${esc(m.tier)} / ${esc(m.provider)}</td><td>${esc(m.state || m.health)}</td><td>${s.sessions.find(x=>x.machine && x.machine===m.machine)?.sessions || 0}</td><td>${m.hourly_micros==null?'unknown':money(m.hourly_micros)} ${esc(m.currency)}</td></tr>`).join('')}</table><h2>Last decisions and switches</h2><ol>${s.journal.map(j=>`<li>${esc(new Date(Number(j.at)).toISOString())} ${esc(j.event)} ${esc(JSON.stringify(j.detail))}</li>`).join('')}</ol></main></body></html>`);
+ }));
+ r.get('/compute/pools/:pool/quotes',wrap(async(req,res)=>res.json({dry_run:true,providers:await broker.quotes(req.params.pool)})));
+ r.post('/compute/pools/:pool/strategy',express.urlencoded({extended:false,limit:'2kb'}),wrap(async(req,res)=>{await broker.setStrategy(req.params.pool,req.body.strategy);res.redirect(303,'/api/v1/compute');}));
+ r.post('/compute/rentals/:id/bind',wrap(async(req,res)=>{await broker.bind(req.params.id,String(req.body.machine || ''));res.json({ok:true});}));
+ return r;
+}
+module.exports={computeRouter};
@@ -34,6 +34,7 @@ function loadConfig(env = process.env) {
@@ -34,6 +34,7 @@ function loadConfig(env = process.env) {
const networkUrl = trim(env.OV_NETWORK_URL || 'https://openvibe.network');
const networkUrl = trim(env.OV_NETWORK_URL || 'https://openvibe.network');
const baseUrl = trim(env.BASE_URL || (isProduction ? 'https://openvibe.run' : `http://localhost:${port}`));
const baseUrl = trim(env.BASE_URL || (isProduction ? 'https://openvibe.run' : `http://localhost:${port}`));
return {
return {
+ compute: require('./compute/config').loadCompute(env),
nodeEnv,
nodeEnv,
isProduction,
isProduction,
port,
port,
@@ -53,7 +53,7 @@ function createDispatcher({ config = null, db = null, store = null, events = nul
@@ -53,7 +53,7 @@ function createDispatcher({ config = null, db = null, store = null, events = nul
const learned = outcomes || (db ? createOutcomes({ db, now, log }) : null);
const learned = outcomes || (db ? createOutcomes({ db, now, log }) : null);
const benchPriors = priors || createPriors({ config, fetchImpl, now, log, ttlMs: config && config.bench ? config.bench.priorsTtlMs : undefined });
const benchPriors = priors || createPriors({ config, fetchImpl, now, log, ttlMs: config && config.bench ? config.bench.priorsTtlMs : undefined });
// Provider offers are placement candidates only while the adapter can be called.
// Provider offers are placement candidates only while the adapter can be called.
- const placer = placement || createPlacement({ config, fetchImpl, now, log, providers: adapter.enabled ? { [BROWSER_PROVIDER]: ['browser'] } : {}, outcomes: learned, priors: benchPriors });
+ const placer = placement || createPlacement({ config, db, fetchImpl, now, log, providers: adapter.enabled ? { [BROWSER_PROVIDER]: ['browser'] } : {}, outcomes: learned, priors: benchPriors });
const poller = createPoller({ config, db, store, events, stream, bot: botClient, provider: adapter, usageRelay: relay, placement: placer, outcomes: learned, now, log });
const poller = createPoller({ config, db, store, events, stream, bot: botClient, provider: adapter, usageRelay: relay, placement: placer, outcomes: learned, now, log });
const enabled = Boolean(botClient.enabled || adapter.enabled);
const enabled = Boolean(botClient.enabled || adapter.enabled);
const note = botClient.enabled
const note = botClient.enabled
@@ -39,7 +39,7 @@
@@ -39,7 +39,7 @@
* answer with no eligible candidate is null, which the poller settles run.job.unplaceable.
* answer with no eligible candidate is null, which the poller settles run.job.unplaceable.
*/
*/
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
-const { place, optionKey } = require('openvibe-sdk/placement');
+const { place, plan, optionKey } = require('openvibe-sdk/placement');
const { createServiceTokenClient } = require('openvibe-sdk/auth');
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');
@@ -106,7 +106,7 @@ const personOf = (requester) => (/^user:(usr_[0-9A-HJKMNP-TV-Z]{26})$/.exec(Stri
@@ -106,7 +106,7 @@ const personOf = (requester) => (/^user:(usr_[0-9A-HJKMNP-TV-Z]{26})$/.exec(Stri
* createPlacement({ config, fetchImpl, now, log, providers }) → { choose, offers, status }. `providers` maps
* createPlacement({ config, fetchImpl, now, log, providers }) → { choose, offers, status }. `providers` maps
* the name of each external adapter that is on to the classes it runs ({ 'browser-provider': ['browser'] }).
* the name of each external adapter that is on to the classes it runs ({ 'browser-provider': ['browser'] }).
*/
*/
-function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console, providers = {}, outcomes = null, priors = null } = {}) {
+function createPlacement({ config, db = null, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console, providers = {}, outcomes = null, priors = null } = {}) {
const { url } = config.network;
const { url } = config.network;
const offersUrlOf = (kind) => `${url}/api/v1/offers?kind=${kind}`;
const offersUrlOf = (kind) => `${url}/api/v1/offers?kind=${kind}`;
const offersUrl = offersUrlOf('node');
const offersUrl = offersUrlOf('node');
@@ -192,6 +192,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -192,6 +192,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
return { ok: false, code: 'run.job.node_offline', detail: 'node must be a machine id (dev_…)' };
return { ok: false, code: 'run.job.node_offline', detail: 'node must be a machine id (dev_…)' };
}
}
const age = fresh ? 0 : offersTtlMs;
const age = fresh ? 0 : offersTtlMs;
+ if (db && await db.maybe("SELECT id FROM run_compute_rentals WHERE machine=$1 AND state IN ('draining','releasing','uncertain') LIMIT 1", [nodeId])) return { ok: false, code: 'run.job.node_offline', detail: 'the machine is draining' };
const offer = [...(await offers('node', { maxAgeMs: age })), ...(await ownedOffers({ maxAgeMs: age }))]
const offer = [...(await offers('node', { maxAgeMs: age })), ...(await ownedOffers({ maxAgeMs: age }))]
.find((o) => o && o.kind === 'node' && o.node_id === nodeId);
.find((o) => o && o.kind === 'node' && o.node_id === nodeId);
if (!offer) return { ok: false, code: 'run.job.node_offline', detail: `machine ${nodeId} is not connected to Run` };
if (!offer) return { ok: false, code: 'run.job.node_offline', detail: `machine ${nodeId} is not connected to Run` };
@@ -230,7 +231,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -230,7 +231,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
// names user-owned (contracts 0.161.0). The public list carries none; one that slipped in is never used, since
// 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.
// only the internal read states the owner this rule needs.
const trusted = (o) => (o.capabilities || []).includes(want) && o.trust !== 'user-owned';
const trusted = (o) => (o.capabilities || []).includes(want) && o.trust !== 'user-owned';
- const candidates = (await offers('node')).filter((o) => o.kind === 'node'
+ let 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));
@@ -249,6 +250,31 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -249,6 +250,31 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
if (!candidates.length) return null;
if (!candidates.length) return null;
const req = { ...requirements, capabilities: [...new Set([...(requirements.capabilities || []), want])] };
const req = { ...requirements, capabilities: [...new Set([...(requirements.capabilities || []), want])] };
contracts.assertValid('platform.workload-requirements@1', req);
contracts.assertValid('platform.workload-requirements@1', req);
+ let computeDecision = null;
+ if (db) {
+ const draining = new Set((await db.many("SELECT machine FROM run_compute_rentals WHERE state IN ('draining','releasing','uncertain')")).map(r => r.machine));
+ candidates = candidates.filter(o => !draining.has(o.node_id));
+ }
+ const poolId = config.compute && config.compute.job_pool;
+ const pool = poolId && config.compute.pools[poolId];
+ if (pool) {
+ const saved = db && await db.maybe('SELECT strategy FROM run_compute_pools WHERE id=$1', [poolId]);
+ const hard = plan(req, candidates, { now: now() });
+ const eligible = new Set(hard.candidates.filter(c => c.eligible).map(c => c.id));
+ const ms = candidates.filter(o => eligible.has(o.offer_id)).map(o => {
+ const v = config.compute.inventory[o.node_id] || {};
+ return { ...v, id: o.offer_id, owner: o.owner, tier: o.trust === 'user-owned' ? 'own' : o.trust === 'community' ? 'community' : o.trust === 'first-party' ? (v.tier || 'dedicated') : 'partner',
+ health: o.health.status, region: o.region, currency: v.currency || 'USD', hourly: v.hourly ?? (o.trust === 'user-owned' ? 0 : null),
+ free_ram_mb: o.capacity?.memory?.available_mb ?? v.free_ram_mb ?? 0,
+ free_slots: o.capacity?.workers?.[row.class] ?? v.free_slots ?? 0,
+ latency_ms: o.latency_ms?.p95, gpu_mb: o.capacity?.gpu?.memory_mb ?? v.gpu_mb };
+ });
+ const strategy = { cheapest: 'cheapest', 'lowest-latency': 'fastest', private: 'private', 'first-party-only': 'private' }[req.objective] || saved?.strategy || pool.strategy;
+ computeDecision = require('openvibe-sdk/compute').choose({ strategy, owner: person, kind: row.class,
+ ram_mb: req.resources?.memory_mb, region: req.region, public_egress: pool.public_egress !== false }, ms, pool, { now: now() });
+ candidates = candidates.filter(o => o.offer_id === computeDecision.selected);
+ }
+
// Placement A1: Bench's priors per option key (an hour's cache), the person's learned counts (one read) and the
// Placement A1: Bench's priors per option key (an hour's cache), the person's learned counts (one read) and the
// job's class as the step kind. An unreadable table or a quiet Bench leaves the defaults, never a failed placement.
// job's class as the step kind. An unreadable table or a quiet Bench leaves the defaults, never a failed placement.
const learned = outcomes && person ? await outcomes.forPerson(person) : [];
const learned = outcomes && person ? await outcomes.forPerson(person) : [];
@@ -274,7 +300,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -274,7 +300,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
// A provider offer has no device: its own id stands for where the job went (the read result's node).
// A provider offer has no device: its own id stands for where the job went (the read result's node).
node_id: chosen.kind === 'provider' ? (chosen.node_id || chosen.offer_id) : chosen.node_id,
node_id: chosen.kind === 'provider' ? (chosen.node_id || chosen.offer_id) : chosen.node_id,
provider: chosen.provider ?? null, region: chosen.region ?? null, kind: chosen.kind,
provider: chosen.provider ?? null, region: chosen.region ?? null, kind: chosen.kind,
- offer_id: chosen.offer_id, trust: chosen.trust, reasons: result.reasons,
+ offer_id: chosen.offer_id, trust: chosen.trust, reasons: [...result.reasons, ...(computeDecision ? [...computeDecision.reasons, `estimated ${computeDecision.hourly} ${computeDecision.currency}/hour`] : [])],
option_key: optionKey(chosen), step_kind: row.class,
option_key: optionKey(chosen), step_kind: row.class,
explain: publicExplain({ line: placed.line, ranked: placed.ranked, rejected: placed.rejected }),
explain: publicExplain({ line: placed.line, ranked: placed.ranked, rejected: placed.rejected }),
};
};
@@ -38,7 +38,7 @@ async function main() {
@@ -38,7 +38,7 @@ async function main() {
const accountData = accountDataLib.create({ db, log: console });
const accountData = accountDataLib.create({ db, log: console });
const accountSend = config.oauth.clientSecret ? createNetworkSender({ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret }) : null;
const accountSend = config.oauth.clientSecret ? createNetworkSender({ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret }) : null;
const app = createApp({ config, db, valkey, registry, accountData, accountSend });
const app = createApp({ config, db, valkey, registry, accountData, accountSend });
- const { keys, outbox, stream, tickets, dispatch, indexnow, sessions } = app.locals;
+ const { keys, outbox, stream, tickets, dispatch, indexnow, sessions, broker } = app.locals;
keys.start();
keys.start();
const ticketKey = tickets.start();
const ticketKey = tickets.start();
@@ -50,6 +50,7 @@ async function main() {
@@ -50,6 +50,7 @@ async function main() {
dispatch.start();
dispatch.start();
// Run sessions (server/sessions/): place, watch and meter the open ones. Off without the Bot token.
// Run sessions (server/sessions/): place, watch and meter the open ones. Off without the Bot token.
sessions.start();
sessions.start();
+ broker.start();
}
}
let accountSubscription = null;
let accountSubscription = null;
@@ -88,6 +89,7 @@ async function main() {
@@ -88,6 +89,7 @@ async function main() {
() => keys.stop(),
() => keys.stop(),
() => tickets.stop(),
() => tickets.stop(),
() => dispatch.stop(),
() => dispatch.stop(),
+ () => broker.stop(),
() => sessions.stop(),
() => sessions.stop(),
() => stream.close(),
() => stream.close(),
() => outbox.stop(),
() => outbox.stop(),
@@ -43,7 +43,7 @@ class SessionError extends Error {
@@ -43,7 +43,7 @@ class SessionError extends Error {
constructor(status, code, detail) { super(detail || code); this.status = status; this.code = code; this.detail = detail; }
constructor(status, code, detail) { super(detail || code); this.status = status; this.code = code; this.detail = detail; }
}
}
-function createSessions({ config, db, desks, outbox, now = () => Date.now(), log = console }) {
+function createSessions({ config, db, desks, outbox, now = () => Date.now(), log = console, broker = null }) {
const cfg = config.sessions || {};
const cfg = config.sessions || {};
const tickMs = cfg.tickMs || 2000;
const tickMs = cfg.tickMs || 2000;
const leaseCheckMs = cfg.leaseCheckMs || 30_000;
const leaseCheckMs = cfg.leaseCheckMs || 30_000;
@@ -88,10 +88,16 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
@@ -88,10 +88,16 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
}
}
// ── create, act, finish ───────────────────────────────────────────────────────────────────────────────────
// ── create, act, finish ───────────────────────────────────────────────────────────────────────────────────
- async function create(owner, via, body) {
+ async function create(owner, via, body, compute = {}) {
+ if (compute.pool && (typeof compute.pool !== 'string' || !/^[a-z0-9-]{1,40}$/.test(compute.pool))) throw new SessionError(422, 'run.compute.pool', 'invalid compute pool');
+ const pool = compute.pool || null;
+ const strategy = compute.strategy || null;
+ if (strategy && !require('openvibe-sdk/compute').STRATEGIES.includes(strategy)) throw new SessionError(422, 'run.compute.strategy', 'unknown compute strategy');
+ if (strategy && !pool) throw new SessionError(422, 'run.compute.pool', 'a strategy requires a compute pool');
+ if (pool) { if (!broker) throw new SessionError(503, 'run.compute.off'); await broker.poolConfig(pool); }
const v = contracts.validate('run.session-create-request@1', body);
const v = contracts.validate('run.session-create-request@1', body);
if (!v.valid) throw new SessionError(422, 'run.session.invalid', `invalid session request: ${(v.errors && v.errors[0] && v.errors[0].message) || 'bad value'}`.slice(0, 300));
if (!v.valid) throw new SessionError(422, 'run.session.invalid', `invalid session request: ${(v.errors && v.errors[0] && v.errors[0].message) || 'bad value'}`.slice(0, 300));
- const digest = store.digestOf(body);
+ const digest = store.digestOf(pool ? { ...body, compute_pool: pool, compute_strategy: strategy } : body);
if (body.idempotency_key) {
if (body.idempotency_key) {
const earlier = await db.maybe('SELECT * FROM run_sessions WHERE owner = $1 AND idempotency_key = $2 AND created_at > $3', [owner, body.idempotency_key, now() - 86_400_000]);
const earlier = await db.maybe('SELECT * FROM run_sessions WHERE owner = $1 AND idempotency_key = $2 AND created_at > $3', [owner, body.idempotency_key, now() - 86_400_000]);
if (earlier) {
if (earlier) {
@@ -104,7 +110,7 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
@@ -104,7 +110,7 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
const row = {
const row = {
id: store.newSessionId(), owner, via: via || null, kind: 'browser', state: 'placing',
id: store.newSessionId(), owner, via: via || null, kind: 'browser', state: 'placing',
width: body.width || 1280, height: body.height || 800, ttl_s: body.ttl_s || 3600, idle_s: body.idle_s || 900,
width: body.width || 1280, height: body.height || 800, ttl_s: body.ttl_s || 3600, idle_s: body.idle_s || 900,
- machine_asked: body.machine || null, requirements: body.requirements || null, idempotency_key: body.idempotency_key || null,
+ compute_pool: pool, compute_strategy: strategy, compute_decision: null, machine_asked: body.machine || null, requirements: body.requirements || null, idempotency_key: body.idempotency_key || null,
request_digest: digest, reasons: [], task_id: body.task_id || null, curtain: false, created_at: at, updated_at: at,
request_digest: digest, reasons: [], task_id: body.task_id || null, curtain: false, created_at: at, updated_at: at,
};
};
const out = await store.insert(db, row);
const out = await store.insert(db, row);
@@ -219,14 +225,20 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
@@ -219,14 +225,20 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
if (now() - row.created_at >= row.ttl_s * 1000) return finish(row, 'unplaceable', 'no machine had room before the session\'s time ran out', { release: false });
if (now() - row.created_at >= row.ttl_s * 1000) return finish(row, 'unplaceable', 'no machine had room before the session\'s time ran out', { release: false });
let hosts;
let hosts;
try { hosts = await hostsFor(bare(row.owner)); } catch (e) { if (!e.retryable) log.warn(`[Run] session ${row.id}: desk hosts: ${e.code}`); return row; }
try { hosts = await hostsFor(bare(row.owner)); } catch (e) { if (!e.retryable) log.warn(`[Run] session ${row.id}: desk hosts: ${e.code}`); return row; }
- const list = candidates(row, hosts);
+ if (broker) hosts = await broker.available(hosts);
+ let list = candidates(row, hosts);
+ let decision = null;
+ if (row.compute_pool && broker) {
+ decision = await broker.decide(row, hosts);
+ list = decision.selected ? hosts.filter(h => h.machine_id === decision.selected) : [];
+ }
if (!list.length) {
if (!list.length) {
if (row.machine_asked && !hosts.some((h) => h.machine_id === row.machine_asked)) return finish(row, 'unplaceable', 'that machine is not open to your agent (or it runs no browser desks)', { release: false });
if (row.machine_asked && !hosts.some((h) => h.machine_id === row.machine_asked)) return finish(row, 'unplaceable', 'that machine is not open to your agent (or it runs no browser desks)', { release: false });
return row;
return row;
}
}
const desk = store.deskOf(row.id);
const desk = store.deskOf(row.id);
for (const h of list) {
for (const h of list) {
- await store.update(db, row.id, { state: 'starting', machine: h.machine_id, desk, trust: h.trust, offer_id: h.shared ? `desk:${h.machine_id}` : null, reasons: reasonsOf(row, h), updated_at: now() });
+ await store.update(db, row.id, { state: 'starting', machine: h.machine_id, desk, trust: h.trust, offer_id: h.shared ? `desk:${h.machine_id}` : null, reasons: [...reasonsOf(row, h), ...(decision ? [...decision.reasons, `estimated ${decision.hourly} ${decision.currency}/hour`] : [])], compute_decision: decision, updated_at: now() });
let lease;
let lease;
try {
try {
lease = await desks.lease(h.machine_id, bare(row.owner), { session: row.id, desk, kind: 'browser', width: row.width, height: row.height });
lease = await desks.lease(h.machine_id, bare(row.owner), { session: row.id, desk, kind: 'browser', width: row.width, height: row.height });
@@ -13,9 +13,9 @@ const crypto = require('crypto');
@@ -13,9 +13,9 @@ const crypto = require('crypto');
const { ids } = require('openvibe-contracts');
const { ids } = require('openvibe-contracts');
const COLUMNS = ['id', 'owner', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'idempotency_key',
const COLUMNS = ['id', 'owner', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'idempotency_key',
- 'request_digest', 'machine', 'desk', 'trust', 'offer_id', 'reasons', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason',
+ 'compute_pool', 'compute_strategy', 'compute_decision', 'request_digest', 'machine', 'desk', 'trust', 'offer_id', 'reasons', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason',
'end_detail', 'metered_minutes', 'lease_checked_at', 'task_id', 'curtain', 'updated_at'];
'end_detail', 'metered_minutes', 'lease_checked_at', 'task_id', 'curtain', 'updated_at'];
-const JSON_COLUMNS = new Set(['requirements', 'reasons']);
+const JSON_COLUMNS = new Set(['requirements', 'reasons', 'compute_decision']);
const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
const json = (v, dflt) => { if (v == null) return dflt; if (typeof v === 'object') return v; try { return JSON.parse(v); } catch { return dflt; } };
const json = (v, dflt) => { if (v == null) return dflt; if (typeof v === 'object') return v; try { return JSON.parse(v); } catch { return dflt; } };
new file mode 100644
new file mode 100644
@@ -0,0 +1,23 @@
@@ -0,0 +1,23 @@
+'use strict';
+const assert=require('node:assert/strict');
+const {boot}=require('./helpers/app');
+const {ids}=require('openvibe-contracts');
+(async()=>{
+ const owner=ids.newId('user'), stranger=ids.newId('user');
+ const machine=ids.newId('device');
+ const desks={enabled:true,note:'stub',hosts:async()=>[{machine_id:machine,trust:'user-owned',online:true,free:2,max_browser_desks:2}],lease:async()=>({ok:true}),get:async()=>({ok:true})};
+ const t=await boot({appOpts:{desks},env:{RUN_COMPUTE_OPERATORS:owner,RUN_COMPUTE_POOLS:JSON.stringify({browser:{strategy:'own',currency:'USD'}}),RUN_COMPUTE_INVENTORY:JSON.stringify({[machine]:{pool:'browser',ram_mb:4096,region:'us'}})}});
+ try {
+ const as=u=>({token:t.network.signUser({subject:u})});
+ assert.equal((await t.call('GET','/api/v1/compute?format=json',as(stranger))).status,403);
+ assert.equal((await t.call('GET','/api/v1/compute?format=json',as(owner))).status,200);
+ const form=await t.call('GET','/api/v1/compute',as(owner));assert.equal(form.status,200);assert.match(form.text,/Compute pools/);
+ const start=await t.call('POST','/api/v1/sessions?pool=browser&strategy=own',{...as(owner),body:{kind:'browser'}});
+ assert.equal(start.status,201,start.text);await t.app.locals.sessions.tick();
+ const s=await t.call('GET',`/api/v1/sessions/${start.json.id}?explain=1`,as(owner));
+ assert.equal(s.json.compute.selected,machine);assert.equal(s.json.compute.strategy,'own');assert.equal(s.json.compute.hourly,0);
+ assert.equal((await t.call('GET',`/api/v1/sessions/${start.json.id}?explain=1`,as(stranger))).status,404);
+ assert.equal((await t.call('POST','/api/v1/compute/pools/browser/strategy',{...as(owner),body:{strategy:'cheapest'},headers:{origin:'https://evil.example'}})).status,403);
+ console.log('operator auth, CSRF, session strategy and owner-only cost explanation: passed');
+ } finally {await t.close();}
+})().catch(e=>{console.error(e);process.exit(1);});
new file mode 100644
new file mode 100644
@@ -0,0 +1,34 @@
@@ -0,0 +1,34 @@
+'use strict';
+const assert=require('node:assert/strict');
+const {testDb}=require('./helpers/db');
+const {createBroker}=require('../server/compute/broker');
+(async()=>{
+ for(const scenario of ['total price','cap fallback','GPU fit','missing key']) {
+ const t=await testDb();const now=Date.UTC(2026,10,8);const created=[];
+ const config={compute:{enabled:true,dry_run:false,owner:'alice',inventory:{},pools:{workers:{kind:'browser',strategy:'cheapest',currency:'EUR',providers:['hetzner','ovh'],min_warm:1,max_machines:2,gpu_mb:scenario==='GPU fit'?8192:0}},providers:{}}};
+ const adapters={};
+ for(const [name,hourly,gpu_mb] of [['hetzner',0.1,0],['ovh',0.2,8192]]) {
+ config.compute.providers[name]={key:'stub',monthly_cap:'10',currency:'EUR',image:'prepared',costs_bounded:true,extra_hourly:'0'};
+ adapters[name]={quote:async()=>[{provider:name,id:name,hourly,gpu_mb,ram_mb:8192,currency:'EUR',region:'eu',tier:'dedicated',setup:0,billing_seconds:3600,expires_at:now+60000}],provision:async()=>{created.push(name);return {id:name};},health:async()=>({state:'running'}),release:async()=>{}};
+ }
+ if(scenario==='total price')config.compute.providers.hetzner.extra_hourly='0.5';
+ if(scenario==='cap fallback')config.compute.providers.hetzner.monthly_cap='0.01';
+ if(scenario==='missing key')config.compute.providers.hetzner.key='';
+ const broker=createBroker({db:t.db,config,adapters,desks:{enabled:true,hosts:async()=>[]},now:()=>now});
+ try {
+ await broker.tick();assert.deepEqual(created,['ovh'],scenario);
+ if(scenario==='total price') {
+ config.compute.inventory.m={pool:'workers',ram_mb:8192};
+ const hosts=[{machine_id:'m',trust:'user-owned',online:true,max_browser_desks:4,free:2}];
+ config.compute.pools.workers.kind='desktop';
+ assert.equal((await broker.inventory(hosts,'alice'))[0].free_ram_mb,3072,'occupied desktops use the 2 GB floor');
+ config.compute.pools.workers.measured_peak_ram_mb=3000;
+ assert.equal((await broker.inventory(hosts,'alice'))[0].free_ram_mb,1168,'higher measured peaks reduce remaining capacity');
+ config.compute.pools.workers.kind='browser';config.compute.pools.workers.measured_peak_ram_mb=900;
+ assert.equal((await broker.inventory(hosts,'alice'))[0].free_ram_mb,5368,'browser pools use their measured peak');
+ }
+ }
+ finally {await t.close();}
+ }
+ console.log('autoscaler selects total cost, falls back on capped or unconfigured providers, and enforces GPU fit: passed');
+})().catch(e=>{console.error(e);process.exit(1);});
new file mode 100644
new file mode 100644
@@ -0,0 +1,68 @@
@@ -0,0 +1,68 @@
+'use strict';
+const assert=require('node:assert/strict');
+const {testDb}=require('./helpers/db');
+const {createBroker}=require('../server/compute/broker');
+const {loadCompute}=require('../server/compute/config');
+(async()=>{
+ const t=await testDb(); let time=Date.UTC(2026,10,8); let created=0,removed=0,deleted=false,uncertain=false;
+ const q={provider:'hetzner',id:'cx',hourly:1,currency:'EUR',ram_mb:4096,gpu_mb:0,tier:'dedicated',region:'eu',setup:0,billing_seconds:3600,expires_at:time+60000};
+ const config={compute:{enabled:true,dry_run:false,owner:'alice',pools:{browsers:{kind:'browser',currency:'EUR',providers:['hetzner'],min_warm:1,strategy:'automatic'}},inventory:{m:{pool:'browsers',tier:'dedicated',ram_mb:4096,hourly:1,currency:'EUR'}},providers:{hetzner:{key:'stub',monthly_cap:'2',currency:'EUR',image:'prepared',costs_bounded:true,extra_hourly:'0'}}}};
+ const adapters={hetzner:{quote:async()=>[q],provision:async()=>{created++;if(uncertain)throw new Error('timeout');return {id:'remote'};},health:async()=>({deleted,state:'running'}),release:async()=>{removed++;}}};
+ const desks={enabled:true,hosts:async()=>[{machine_id:'m',trust:'first-party',online:true,free:4,max_browser_desks:4}]};
+ let broker=createBroker({db:t.db,config,desks,adapters,now:()=>time});
+ try {
+ assert.equal(loadCompute({}).dry_run,true);assert.equal(loadCompute({}).providers.hetzner.monthly_cap,'0');
+ config.compute.providers.hetzner.monthly_cap='0';
+ assert.match((await broker.provision('browsers',q)).reason,/quote only/);assert.equal(created,0);
+ config.compute.providers.hetzner.monthly_cap='2';
+ config.compute.dry_run=true;assert.equal((await broker.provision('browsers',q)).dry_run,true);assert.equal(created,0);
+ config.compute.dry_run=false;
+ const results=await Promise.all([broker.provision('browsers',q),broker.provision('browsers',q)]);
+ assert.equal(created,1);assert.equal(results.filter(r=>r.rental).length,1);assert.ok(results.some(r=>/cap/.test(r.reason)));
+ const id=results.find(r=>r.rental).rental;
+ await broker.bind(id,'m');
+ const choice=await broker.decide({compute_pool:'browsers',owner:'user:alice',kind:'browser'},await desks.hosts());
+ assert.equal(choice.selected,'m');assert.equal(choice.hourly,1);
+ broker=createBroker({db:t.db,config,desks,adapters,now:()=>time});
+ assert.match((await broker.provision('browsers',q)).reason,/cap/,'restart retains reservation');
+ time+=3600000;await broker.tick();assert.equal(removed,1);
+ let state=await broker.status();assert.equal(state.rentals[0].state,'releasing');assert.equal(state.budgets[0].committed_micros,'2000000');
+ q.expires_at=time+60000;
+ assert.match((await broker.provision('browsers',q)).reason,/unresolved/,'unconfirmed release blocks new provider spending');
+ deleted=true;await broker.tick();state=await broker.status();assert.equal(state.rentals[0].state,'released');assert.equal(state.rentals[0].spent_micros,'1000000');
+ assert.ok(state.journal.some(j=>j.event==='release.confirmed'));
+ config.compute.providers.hetzner.monthly_cap='10';
+ time=Date.UTC(2026,10,30,23,30);q.expires_at=time+60000;
+ deleted=false;
+ const cross=await broker.provision('browsers',q);
+ desks.hosts=async()=>[{machine_id:'auto',name:cross.rental,trust:'community',online:true,free:4,max_browser_desks:4}];
+ await broker.tick();assert.equal((await t.db.maybe('SELECT machine FROM run_compute_rentals WHERE id=$1',[cross.rental])).machine,null,'a community name cannot claim a rental');
+ desks.hosts=async()=>[{machine_id:'auto',name:cross.rental,trust:'user-owned',online:true,free:4,max_browser_desks:4}];
+ await broker.tick();assert.equal((await t.db.maybe('SELECT machine FROM run_compute_rentals WHERE id=$1',[cross.rental])).machine,'auto');
+ const bound=await broker.decide({compute_pool:'browsers',owner:'user:alice',kind:'browser'},await desks.hosts());
+ assert.equal(bound.hourly,1);assert.equal(bound.tier,'dedicated','rental retains provider tier and cost');
+ const ownOnly=await broker.decide({compute_pool:'browsers',compute_strategy:'own',owner:'user:alice',kind:'browser'},await desks.hosts());
+ assert.equal(ownOnly.selected,null,'rental is not an own-machine placement');
+ time+=3600000;
+ deleted=true;
+ await broker.release(await t.db.maybe('SELECT * FROM run_compute_rentals WHERE id=$1',[cross.rental]));
+ const charges=await t.db.many('SELECT * FROM run_compute_charges WHERE rental=$1 ORDER BY month',[cross.rental]);
+ assert.deepEqual(charges.map(c=>[c.month,String(c.amount_micros)]),[['2026-11','500000'],['2026-12','500000']]);
+ assert.equal((await broker.status()).budgets[0].spent_micros,'500000');
+ q.expires_at=time+60000;deleted=false;
+ config.compute.pools.browsers.min_warm=0;desks.hosts=async()=>[];
+ const lowered=await broker.provision('browsers',q);
+ config.compute.providers.hetzner.monthly_cap='2.25';
+ await broker.tick();
+ assert.equal((await t.db.maybe('SELECT state FROM run_compute_rentals WHERE id=$1',[lowered.rental])).state,'releasing','lowered cap includes already settled spend, not only live reservations');
+ deleted=true;await broker.tick();config.compute.providers.hetzner.monthly_cap='10';
+ q.expires_at=time+60000;uncertain=true;
+ assert.match((await broker.provision('browsers',q)).reason,/unknown/);
+ assert.match((await broker.provision('browsers',q)).reason,/unresolved/);
+ time=Date.UTC(2027,0,1);q.expires_at=time+60000;
+ assert.match((await broker.provision('browsers',q)).reason,/unresolved/,'month rollover cannot forget an unknown purchase');
+ const privateChoice=await broker.decide({compute_pool:'browsers',compute_strategy:'private',owner:'user:alice',kind:'browser'},[{machine_id:'community',trust:'community',online:true,free:10}]);
+ assert.equal(privateChoice.selected,null);
+ console.log('compute cap, concurrent reservation, dry run, restart, release confirmation, unknown outcome and rollover: passed');
+ } finally {await t.close();}
+})().catch(e=>{console.error(e);process.exit(1);});