openvibers/openvibe.run Commits

Placement A1: place() each job, answer its line, learn from the end (#122)

OP OpenVibers committed 13 days ago · 78dce8b634 · 15 files, +377 −19

* Placement A1: place() each job, answer its line, learn from the end

* Placement review: display names and a sanitized explain

Changes

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 487cdcd..aa98768 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -6,6 +6,10 @@ Notable changes to OpenVibe.Run. The service is unreleased; everything below is
### Added
++- Placement A1 (openvibe-sdk 0.76.0): `server/dispatch/placement.js` calls `place()` — Bench priors (an hour's
++ cache), the person's learned counts and one plain line — and the same answer is stored on the job. `GET
++ /api/v1/jobs/:id?explain=1` answers the line beside the contract body. `placement_outcomes` (migration 0007) learns
++ one outcome per ended job: exit 0 is ok, a failure or a timeout is not; a cancelled job is not counted.
- The repository skeleton (plan T14 step 4, R1a): `server/config.js` on port 4920, `server/db.js`,
`server/app.js`, `server/index.js`, `server/observability.js` (`/api/health`, `/api/ready`, `/release.json`,
loopback-only `/metrics`), `server/api/auth.js`, `server/events/outbox.js`, `migrations/0001_jobs.sql`
ddiff --git a/migrations/0007_placement_a1.sql b/migrations/0007_placement_a1.sql
nnew file mode 100644
iindex 0000000..97e3b43
---- /dev/null
++++ b/migrations/0007_placement_a1.sql
@@@ -0,0 +1,29 @@
++-- phase: expand
++-- Placement A1 (agent plan 2026-10-30): Run places each job with openvibe-sdk/placement's place(), answers the one
++-- line saying why with the job, and learns from the job's end.
++--
++-- run_jobs grows the placement's line, the option key and step kind it chose, the ranked/rejected answer for the API,
++-- and placement_recorded, so one ended job is learned exactly once however many times its end is delivered. None of
++-- these holds a job argument, a key or a path.
++--
++-- placement_outcomes is the learned count, one row per (person, option key, step kind): two weights and a time. It
++-- holds no text and no payload. Global counts are a SUM over the rows, read through a five-minute cache. The rows are
++-- exported and erased with the person (server/account-data.js).
++
++ALTER TABLE run_jobs
++ ADD COLUMN IF NOT EXISTS placement_line text,
++ ADD COLUMN IF NOT EXISTS placement_option_key text,
++ ADD COLUMN IF NOT EXISTS placement_step_kind text,
++ ADD COLUMN IF NOT EXISTS placement_explain jsonb,
++ ADD COLUMN IF NOT EXISTS placement_recorded boolean NOT NULL DEFAULT false;
++
++CREATE TABLE IF NOT EXISTS placement_outcomes (
++ person_id text NOT NULL, -- usr_<ULID>
++ option_key text NOT NULL, -- the SDK's optionKey(offer)
++ step_kind text NOT NULL, -- the job's runtime class
++ ok_w double precision NOT NULL DEFAULT 0, -- decayed successes
++ n_w double precision NOT NULL DEFAULT 0, -- decayed tries
++ updated_at timestamptz NOT NULL DEFAULT now(),
++ PRIMARY KEY (person_id, option_key, step_kind)
++);
++CREATE INDEX IF NOT EXISTS placement_outcomes_updated ON placement_outcomes (updated_at);
ddiff --git a/package-lock.json b/package-lock.json
iindex 6525e13..86acc0e 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -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.250.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.76.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.55.0",
"pg": "^8.23.0"
},
@@@ -1163,9 +1163,9 @@
}
},
"node_modules/openvibe-sdk": {
-- "version": "0.74.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
-- "integrity": "sha512-DO9KVPnUriY/I4lX6ZEJtXaUvRDbJyqnqJxvlpl+zkNGFILpmuBXVJogNOCfK071Qci1gg7C4Iw1Xrseklcajg==",
++ "version": "0.76.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.76.0",
++ "integrity": "sha512-jHtxeiX1kpLUhZVKHLBcapdqgCrOaLdXaO7uw5ETCuqNnSwbm+qFG3WaL2tM2thn9QXIPvaU2BZ0//ko1QfwEg==",
"license": "MIT",
"bin": {
"openvibe-download": "bin/openvibe-download.js"
ddiff --git a/package.json b/package.json
iindex 3865363..348a389 100644
---- a/package.json
++++ b/package.json
@@@ -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.250.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.76.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.55.0",
"pg": "^8.23.0"
},
ddiff --git a/server/account-data.js b/server/account-data.js
iindex 1d03182..9fbcb93 100644
---- a/server/account-data.js
++++ b/server/account-data.js
@@@ -7,7 +7,7 @@
* - their open browser sessions (run_sessions, server/sessions/), keyed `owner = user:usr_…`. The export carries each
* session's settings and timing, never its ticket or stream secret; deletion removes them.
* - the jobs they submitted (run_jobs.requester = `user:usr_…`). The export carries the job, its result and its
-- * timings; the row itself stays, because it is the project's record of the work and of what it cost, but the
++ * timings; the placement counts learned about their jobs (placement_outcomes) leave with them too; the row itself stays, because it is the project's record of the work and of what it cost, but the
* requester becomes `user:deleted`, so the job no longer leads to a person.
*
* Telemetry samples that name a person are covered by the SDK itself (telemetry.json, then subject NULL).
@@@ -19,6 +19,9 @@ const TABLES = [
table: 'run_sessions', subject: 'owner', value: (usr) => `user:${usr}`, file: 'sessions.json', order: 'created_at',
columns: ['id', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'machine', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason', 'end_detail', 'metered_minutes', 'updated_at'],
},
++ // Placement A1's learned counts: plain weights per option key and step kind, never text. Exported and erased with
++ // the person.
++ { table: 'placement_outcomes', subject: 'person_id', file: 'placement-outcomes.json', order: 'updated_at' },
// The jobs are handled by the service itself (the requester is tombstoned, not deleted), so they are not in the
// table map; they are listed as covered in the test's guard.
];
ddiff --git a/server/api/v1.js b/server/api/v1.js
iindex 483b52b..344f64c 100644
---- a/server/api/v1.js
++++ b/server/api/v1.js
@@@ -181,7 +181,11 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
r.get('/jobs/:id', guard('run.job.read'), wrap(async (req, res) => {
const row = await getJob(req, req.params.id, { delegated: true });
if (!row) fail(404, 'run.job_not_found', 'no such job');
-- res.json(store.readResult(row));
++ // Placement A1: the default answer is exactly run.job-read-result@1 (its contract forbids extra fields), so the
++ // one line saying why rides beside it when the caller asks with ?explain=1.
++ const job = store.readResult(row);
++ if (req.query.explain !== '1') return res.json(job);
++ res.json({ ...job, explain: row.placement_explain ? { line: row.placement_line || row.placement_explain.line, ranked: row.placement_explain.ranked || [], rejected: row.placement_explain.rejected || [], others_hidden: Number(row.placement_explain.others_hidden) || 0 } : null });
}));
// ── POST /jobs/:id/cancel — run.job.cancel ─────────────────────────────────
ddiff --git a/server/config.js b/server/config.js
iindex 24c8c78..42a2464 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -136,6 +136,14 @@ function loadConfig(env = process.env) {
maxPerTick: Math.max(1, int(env.RUN_DISPATCH_MAX_PER_TICK, 200)),
},
++ // Placement A1: Bench's quality priors, cached an hour, used by place() when the option is a model Bench has
++ // seen. Off in tests unless asked for, so a unit test never reaches the real Bench.
++ bench: {
++ url: trim(env.RUN_BENCH_URL || 'https://bench.openvibe.services'),
++ priors: env.RUN_BENCH_PRIORS != null ? bool(env.RUN_BENCH_PRIORS) : nodeEnv !== 'test',
++ priorsTtlMs: Math.max(0, int(env.RUN_BENCH_PRIORS_TTL_MS, 60 * 60_000)),
++ },
++
// The external managed-browser provider (plan T14 R4a, server/dispatch/browser-provider.js): class browser
// jobs placed on a kind=provider offer named `browser-provider` go to its API instead of Bot. Off while
// RUN_BROWSER_PROVIDER_URL or the key is unset: no provider offer is placed on, so a browser job waits.
ddiff --git a/server/dispatch/index.js b/server/dispatch/index.js
iindex cc742ff..71d79fb 100644
---- a/server/dispatch/index.js
++++ b/server/dispatch/index.js
@@@ -34,6 +34,8 @@ const { createBotClient } = require('./bot');
const { createBrowserProviderAdapter, PROVIDER: BROWSER_PROVIDER } = require('./browser-provider');
const { createPlacement, workerCapability } = require('./placement');
const { createPoller } = require('../jobs/poller');
++const { createOutcomes } = require('./outcomes');
++const { createPriors } = require('./priors');
const { createUsageRelay } = require('../jobs/usage');
/**
@@@ -42,13 +44,17 @@ const { createUsageRelay } = require('../jobs/usage');
* provider, offers and clock.
*/
function createDispatcher({ config = null, db = null, store = null, events = null, stream = null, log = console,
-- now = () => Date.now(), fetchImpl = globalThis.fetch, bot = null, provider = null, usageRelay = null, placement = null } = {}) {
++ now = () => Date.now(), fetchImpl = globalThis.fetch, bot = null, provider = null, usageRelay = null, placement = null,
++ outcomes = null, priors = null } = {}) {
const botClient = bot || createBotClient({ config, fetchImpl, now, log });
const adapter = provider || createBrowserProviderAdapter({ config, fetchImpl, now, log });
const relay = usageRelay || createUsageRelay({ db, config, fetchImpl, now, log });
++ // Placement A1: the learned counts and Bench priors (an hour) are shared by the placer and the poller's end record.
++ 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 });
// 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'] } : {} });
-- const poller = createPoller({ config, db, store, events, stream, bot: botClient, provider: adapter, usageRelay: relay, placement: placer, now, log });
++ const placer = placement || createPlacement({ config, 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
? (adapter.enabled ? `${botClient.note}; ${adapter.note}` : botClient.note)
@@@ -97,7 +103,7 @@ function createDispatcher({ config = null, db = null, store = null, events = nul
}),
now,
// Kept so step 6's bridge has its dependencies in one place without changing the seam's signature.
-- deps: { config, db, store, events, stream, log, bot: botClient, provider: adapter, usageRelay: relay, placement: placer, poller },
++ deps: { config, db, store, events, stream, log, bot: botClient, provider: adapter, usageRelay: relay, placement: placer, poller, outcomes: learned, priors: benchPriors },
};
}
ddiff --git a/server/dispatch/outcomes.js b/server/dispatch/outcomes.js
nnew file mode 100644
iindex 0000000..1b5d826
---- /dev/null
++++ b/server/dispatch/outcomes.js
@@@ -0,0 +1,81 @@
++'use strict';
++
++/**
++ * The learned placement counts (agent plan A1, section 3): one row per (person, option key, step kind) in
++ * placement_outcomes. Plain weights and a time, never text and never a payload.
++ *
++ * createOutcomes({ db, now }) →
++ * forPerson(person) the person's rows plus everyone else's, for place()
++ * record({ person, key, kind, ok, at }) learn() once and upsert the row
++ * global() everyone's rows, cached for five minutes
++ *
++ * The person's own rows are read once per placement; the global read is the five-minute cache. learn() decays at
++ * write and successOf() decays again at read, so an old route fades without a job deleting anything.
++ */
++const { learn } = require('openvibe-sdk/placement');
++
++const GLOBAL_TTL_MS = 5 * 60_000;
++const toRow = (r) => ({
++ person: r.person_id, key: r.option_key, kind: r.step_kind,
++ ok_w: Number(r.ok_w) || 0, n_w: Number(r.n_w) || 0,
++ updated_at: r.updated_at instanceof Date ? r.updated_at.toISOString() : String(r.updated_at),
++});
++
++function createOutcomes({ db, now = Date.now, log = console, globalTtlMs = GLOBAL_TTL_MS } = {}) {
++ let cached = { at: 0, rows: [] };
++
++ async function read(where, args) {
++ return (await db.many(`SELECT person_id, option_key, step_kind, ok_w, n_w, updated_at FROM placement_outcomes ${where}`, args)).map(toRow);
++ }
++
++ async function global() {
++ if (cached.at && now() - cached.at < globalTtlMs) return cached.rows;
++ let rows = [];
++ try { rows = await read('', []); } catch (e) {
++ log.warn(`[Run] placement outcomes could not be read: ${String((e && e.message) || e).slice(0, 200)}`);
++ }
++ cached = { at: now(), rows };
++ return rows;
++ }
++
++ async function forPerson(person) {
++ if (!person) return await global();
++ let mine = [];
++ try { mine = await read('WHERE person_id = $1', [person]); } catch (e) {
++ log.warn(`[Run] placement outcomes for ${person} could not be read: ${String((e && e.message) || e).slice(0, 200)}`);
++ }
++ return [...mine, ...(await global()).filter((r) => r.person !== person)];
++ }
++
++ async function upsert(row) {
++ await db.exec(
++ `INSERT INTO placement_outcomes (person_id, option_key, step_kind, ok_w, n_w, updated_at)
++ VALUES ($1, $2, $3, $4, $5, $6::timestamptz)
++ ON CONFLICT (person_id, option_key, step_kind) DO UPDATE SET
++ ok_w = EXCLUDED.ok_w, n_w = EXCLUDED.n_w, updated_at = EXCLUDED.updated_at`,
++ [row.person, row.key, row.kind, row.ok_w, row.n_w, row.updated_at],
++ );
++ cached.at = 0;
++ }
++
++ async function record({ person, key, kind, ok, at }) {
++ if (!person || !key || !kind) return null;
++ const before = await read('WHERE person_id = $1 AND option_key = $2 AND step_kind = $3', [person, key, kind]);
++ const row = learn(before, { person, key, kind, ok: !!ok, at })
++ .find((r) => r.person === person && r.key === key && r.kind === kind);
++ if (!row) return null;
++ await upsert(row);
++ return row;
++ }
++
++ /** Persist what escalate() returned (it already learned the failure); never record that failure again. */
++ async function recordAll(rows, person) {
++ const mine = (Array.isArray(rows) ? rows : []).filter((r) => r && r.person === person && r.key && r.kind);
++ for (const row of mine) await upsert(row);
++ return mine.length;
++ }
++
++ return { forPerson, global, record, recordAll, GLOBAL_TTL_MS };
++}
++
++module.exports = { createOutcomes, GLOBAL_TTL_MS };
ddiff --git a/server/dispatch/placement.js b/server/dispatch/placement.js
iindex 6735ff0..ed3cf2d 100644
---- a/server/dispatch/placement.js
++++ b/server/dispatch/placement.js
@@@ -39,7 +39,7 @@
* answer with no eligible candidate is null, which the poller settles run.job.unplaceable.
*/
const contracts = require('openvibe-contracts');
--const { plan } = require('openvibe-sdk/placement');
++const { place, optionKey } = require('openvibe-sdk/placement');
const { createServiceTokenClient } = require('openvibe-sdk/auth');
const { requirementsDefault } = require('../util');
const { PROVIDER: BROWSER_PROVIDER } = require('./browser-provider');
@@@ -61,6 +61,44 @@ class PlacementError extends Error {
}
const obj = (v) => (v && typeof v === 'object' && !Array.isArray(v) ? v : {});
++
++/**
++ * The reasons place() gives for excluding a user-owned machine that is not this job's person. Run's public answer
++ * never lists other people's machines, so those entries are dropped and only a count says they existed.
++ */
++const OTHER_MACHINE_REASONS = new Set([
++ "another person's machine",
++ 'a user-owned machine has no owner',
++ 'a user-owned machine needs its owner',
++]);
++
++/** The explanation safe to store and answer: no other person's machine in `rejected`; a count instead. */
++function publicExplain(explain) {
++ if (!explain || typeof explain !== 'object') return null;
++ const rejected = [];
++ let others = 0;
++ for (const entry of (Array.isArray(explain.rejected) ? explain.rejected : [])) {
++ if (OTHER_MACHINE_REASONS.has(String(entry && entry.why))) { others++; continue; }
++ rejected.push(entry);
++ }
++ return {
++ line: explain.line,
++ ranked: Array.isArray(explain.ranked) ? explain.ranked : [],
++ rejected,
++ others_hidden: others,
++ };
++}
++
++/** A candidate's human name for the line: the machine's or provider's name Run has, else a plain fallback. */
++function displayNameOf(offer, person) {
++ const named = (offer.detail && typeof offer.detail.name === 'string' && offer.detail.name)
++ || (typeof offer.name === 'string' && offer.name)
++ || (typeof offer.machine_name === 'string' && offer.machine_name)
++ || null;
++ if (offer.trust === 'user-owned' && offer.owner === person) return named || 'your machine';
++ return named || offer.provider || (offer.detail && offer.detail.provider) || 'the cloud pool';
++}
++
/** The person a job was requested by (`user:usr_…`, as Run stores the requester), or null for a service or an app. */
const personOf = (requester) => (/^user:(usr_[0-9A-HJKMNP-TV-Z]{26})$/.exec(String(requester || '')) || [])[1] || null;
@@@ -68,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 = {} } = {}) {
++function createPlacement({ config, 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');
@@@ -182,6 +220,8 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
return {
node_id: o.node_id, provider: o.provider ?? null, region: o.region ?? null, kind: o.kind,
offer_id: o.offer_id, trust: o.trust, reasons: [`pinned to ${o.node_id}`],
++ option_key: optionKey(o), step_kind: row.class,
++ explain: publicExplain({ line: `On ${displayNameOf(o, personOf(row.requester))}: this job was pinned there.`, ranked: [], rejected: [] }),
};
}
const requirements = { ...obj(row.job).requirements || requirementsDefault(row.class) };
@@@ -209,7 +249,22 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
if (!candidates.length) return null;
const req = { ...requirements, capabilities: [...new Set([...(requirements.capabilities || []), want])] };
contracts.assertValid('platform.workload-requirements@1', req);
-- const result = plan(req, candidates, { now: now() });
++ // 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) : [];
++ const scores = priors ? await priors.forStep(row.class) : {};
++ const priorMap = {};
++ const displayNames = {};
++ for (const o of candidates) {
++ const model = (o.detail && o.detail.model) || o.model;
++ if (model && scores[model] != null) priorMap[optionKey(o)] = scores[model];
++ displayNames[o.offer_id] = displayNameOf(o, person);
++ }
++ const placed = place(req, candidates, {
++ now: now(), person, step: { kind: row.class }, priors: priorMap, outcomes: learned,
++ names: displayNames, current: requirements.authority || null,
++ });
++ const result = placed.result;
const chosen = result.selected ? candidates.find((o) => o.offer_id === result.selected) : null;
if (!chosen) {
if (log && log.debug) log.debug(`[Run] job ${row.id}: no eligible offer for ${want} (${result.reasons.join('; ')})`);
@@@ -220,6 +275,8 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
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,
++ option_key: optionKey(chosen), step_kind: row.class,
++ explain: publicExplain({ line: placed.line, ranked: placed.ranked, rejected: placed.rejected }),
};
}
@@@ -238,4 +295,4 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
};
}
--module.exports = { createPlacement, PlacementError, workerCapability, EXTERNAL_PROVIDERS, personOf };
++module.exports = { createPlacement, PlacementError, workerCapability, EXTERNAL_PROVIDERS, personOf, publicExplain, displayNameOf, OTHER_MACHINE_REASONS };
ddiff --git a/server/dispatch/priors.js b/server/dispatch/priors.js
nnew file mode 100644
iindex 0000000..9567a91
---- /dev/null
++++ b/server/dispatch/priors.js
@@@ -0,0 +1,50 @@
++'use strict';
++
++/**
++ * Bench priors for Run's placement (agent plan A1): Bench's quality per model on the kind of step, cached for an
++ * hour. A job's runtime class maps to a Bench kind when one exists (code → codes.patch), and an offer names its model
++ * in detail.model; an offer with no model (a paired Node) simply has no prior and place() uses its 0.7 default.
++ *
++ * The board is public; a Bench that does not answer leaves the map empty rather than delaying the tick for long.
++ */
++const { createClient } = require('openvibe-sdk/core');
++const { createBenchClient } = require('openvibe-sdk/bench');
++
++const TTL_MS = 60 * 60_000;
++const BENCH_KIND = { code: 'codes.patch', review: 'codes.review' };
++
++function createPriors({ config = {}, fetchImpl = globalThis.fetch, now = Date.now, log = console, ttlMs = TTL_MS } = {}) {
++ const bench = config.bench || {};
++ const enabled = bench.priors !== false && !!bench.url;
++ const client = enabled
++ ? createBenchClient(createClient({ baseUrls: { bench: bench.url }, autoDiscover: false, fetch: fetchImpl, timeoutMs: 4000, retries: 0 }), { baseUrl: bench.url })
++ : null;
++ const cache = new Map();
++
++ async function scoresOf(kind) {
++ const hit = cache.get(kind);
++ if (hit && now() - hit.at < ttlMs) return hit.scores;
++ let scores = {};
++ try {
++ const { board } = await client.boards.get(kind, { type: 'pareto' });
++ for (const row of (board && Array.isArray(board.rows) ? board.rows : [])) {
++ const q = Number(row && row.quality);
++ if (row && typeof row.subject === 'string' && Number.isFinite(q)) scores[row.subject] = Math.max(0, Math.min(1, q));
++ }
++ } catch (e) {
++ log.warn(`[Run] Bench priors for ${kind} unavailable: ${String((e && e.message) || e).slice(0, 160)}`);
++ }
++ cache.set(kind, { at: now(), scores });
++ return scores;
++ }
++
++ async function forStep(runtimeClass) {
++ if (!enabled) return {};
++ const kind = BENCH_KIND[runtimeClass];
++ return kind ? await scoresOf(kind) : {};
++ }
++
++ return { enabled, forStep, BENCH_KIND };
++}
++
++module.exports = { createPriors, BENCH_KIND, TTL_MS };
ddiff --git a/server/jobs/poller.js b/server/jobs/poller.js
iindex ef08c0a..db34db2 100644
---- a/server/jobs/poller.js
++++ b/server/jobs/poller.js
@@@ -39,7 +39,7 @@
* a ring Bot dropped past Run's cursor is a gap Run drops rather than a duplicate Run invents.
*/
const contracts = require('openvibe-contracts');
--const { workerCapability, EXTERNAL_PROVIDERS } = require('../dispatch/placement');
++const { workerCapability, EXTERNAL_PROVIDERS, personOf } = require('../dispatch/placement');
const { PROVIDER: BROWSER_PROVIDER, METER_SLACK_MS } = require('../dispatch/browser-provider');
const usage = require('./usage');
@@@ -59,7 +59,7 @@ const RUN_STATE_OF = { queued: 'placed', running: 'running', succeeded: 'succeed
* is ever routed to it); `usageRelay` is kicked after readings commit. Every dependency is injected; nothing here
* starts at module load.
*/
--function createPoller({ config, db, store, events, stream, bot, provider = null, usageRelay = null, placement, now = () => Date.now(), log = console } = {}) {
++function createPoller({ config, db, store, events, stream, bot, provider = null, usageRelay = null, placement, outcomes = null, now = () => Date.now(), log = console } = {}) {
const cancelSent = new Set(); // job ids whose job_cancel reached Bot in this process
const unknown = new Map(); // job id → consecutive polls Bot answered 404 for
const cursors = new Map(); // job id → how much of Bot's stdout ring Run has streamed
@@@ -114,10 +114,34 @@ function createPoller({ config, db, store, events, stream, bot, provider = null,
if (state === 'expired') stat.expired++;
forget(row.id);
stream.state(row.id, state, at);
++ await recordOutcome(settled.row);
}
return settled;
}
++ /**
++ * Learn one ended job (placement A1): ok means it finished and its exit was 0; a failure, a ttl or a person's
++ * cancel is false; a cancelled job is not counted at all. placement_recorded fences a redelivered end, so one job
++ * is learned once. Only a job a person requested has a person to learn for.
++ */
++ async function recordOutcome(row) {
++ if (!outcomes || !row || !row.placement_option_key || !row.placement_step_kind) return row;
++ const claimed = await db.maybe(
++ 'UPDATE run_jobs SET placement_recorded = true WHERE id = $1 AND placement_recorded = false RETURNING *',
++ [row.id],
++ );
++ if (!claimed) return row;
++ if (row.state === 'cancelled') return row;
++ const person = personOf(row.requester);
++ if (!person) return row;
++ try {
++ await outcomes.record({ person, key: row.placement_option_key, kind: row.placement_step_kind, ok: row.state === 'succeeded', at: now() });
++ } catch (e) {
++ log.warn(`[Run] placement outcome for job ${row.id} was not recorded: ${String((e && e.message) || e).slice(0, 200)}`);
++ }
++ return row;
++ }
++
const fail = (row, at, errorCode, detail) => end(row, at, { state: 'failed', errorCode, errorDetail: detail ? String(detail).slice(0, 1024) : null })
.then((s) => note(row.id, 'failed', { changed: s.changed, error: errorCode }));
@@@ -282,7 +306,13 @@ function createPoller({ config, db, store, events, stream, bot, provider = null,
const placed = await db.tx(async (t) => {
const fresh = await store.get(t, row.id);
if (!fresh || fresh.state !== 'queued') return null;
-- return store.mirror(t, row.id, { state: 'placed', node_id: chosen.node_id, provider: chosen.provider, region: chosen.region }, sentAt);
++ return store.mirror(t, row.id, {
++ state: 'placed', node_id: chosen.node_id, provider: chosen.provider, region: chosen.region,
++ placement_line: chosen.explain ? chosen.explain.line : null,
++ placement_option_key: chosen.option_key || null,
++ placement_step_kind: chosen.step_kind || row.class,
++ placement_explain: chosen.explain || null,
++ }, sentAt);
});
unconfirmed.delete(row.id);
if (!placed) {
@@@ -370,6 +400,7 @@ function createPoller({ config, db, store, events, stream, bot, provider = null,
if (moved === 'failed') stat.failed++;
forget(row.id);
stream.state(row.id, moved, at);
++ await recordOutcome(settled.row);
}
return note(row.id, moved, { changed: settled.changed, exit: b.exit_reason ?? null });
}
@@@ -457,6 +488,7 @@ function createPoller({ config, db, store, events, stream, bot, provider = null,
if (moved === 'failed') stat.failed++;
forget(row.id);
stream.state(row.id, moved, at);
++ await recordOutcome(settled.row);
}
return note(row.id, moved, { changed: settled.changed, exit: p.exit_reason ?? null, wall_ms: m.wallMs, clamped: m.clamped || undefined });
}
ddiff --git a/server/jobs/store.js b/server/jobs/store.js
iindex efa9603..c129182 100644
---- a/server/jobs/store.js
++++ b/server/jobs/store.js
@@@ -262,12 +262,18 @@ async function mirror(db, id, bot = {}, at) {
usage_read = GREATEST(usage_read, COALESCE($12, 0)),
job = CASE WHEN $13::bigint IS NULL OR job ? 'placed_at' THEN job
ELSE jsonb_set(job, '{placed_at}', to_jsonb($13::bigint)) END,
++ placement_line = COALESCE($15, placement_line),
++ placement_option_key = COALESCE($16, placement_option_key),
++ placement_step_kind = COALESCE($17, placement_step_kind),
++ placement_explain = COALESCE($18::jsonb, placement_explain),
updated_at = $14
WHERE id = $1 RETURNING *`,
[id, bot.state ?? null, bot.node_id ?? null, bot.provider ?? null, bot.region ?? null, bot.started_ms ?? null,
bot.finished_at ?? null, bot.exit_reason ?? null, bot.exit_code ?? null,
bot.result === undefined || bot.result === null ? null : JSON.stringify(bot.result),
-- bot.wall_ms ?? null, bot.usage_read ?? null, placed, at]);
++ bot.wall_ms ?? null, bot.usage_read ?? null, placed, at,
++ bot.placement_line ?? null, bot.placement_option_key ?? null, bot.placement_step_kind ?? null,
++ bot.placement_explain === undefined || bot.placement_explain === null ? null : JSON.stringify(bot.placement_explain)]);
}
/**
ddiff --git a/test/account-data.test.js b/test/account-data.test.js
iindex 42b8ffc..6a7e4a1 100644
---- a/test/account-data.test.js
++++ b/test/account-data.test.js
@@@ -28,6 +28,8 @@ const B = 'usr_01JZ0000000000000000000BBB';
VALUES ($1, $2, 'browser', 'ended', 1280, 720, 900, 60, 'digest', 1, 1)`, [`ses_${name}`, `user:${subject}`]);
await d.db.exec(`INSERT INTO run_jobs (id, project_id, requester, idempotency_key, class, job, ttl_ms, state, created_at, updated_at)
VALUES ($1, 'proj_1', $2, $3, 'browser', '{}', 1000, 'succeeded', 1, 1)`, [jobId, `user:${subject}`, `idem-${name}`]);
++ await d.db.exec(`INSERT INTO placement_outcomes (person_id, option_key, step_kind, ok_w, n_w, updated_at)
++ VALUES ($1, $2, 'browser', 1, 1, now())`, [subject, `off_${name}`]);
}
const part = await accountData.exportPart(A);
const file = part.files.find((f) => f.name === 'sessions.json');
@@@ -37,6 +39,9 @@ const B = 'usr_01JZ0000000000000000000BBB';
const jobs = part.files.find((f) => f.name === 'jobs.json');
assert.ok(jobs, 'jobs.json is in the part');
assert.deepStrictEqual(jobs.content.map((j) => j.id), ['job_01JZ0000000000000000000ADA'], 'only their own job');
++ const learned = part.files.find((f) => f.name === 'placement-outcomes.json');
++ assert.ok(learned, 'placement-outcomes.json is in the part');
++ assert.deepStrictEqual(learned.content.map((r) => r.option_key), ['off_ada'], 'only their own placement counts');
const counts = await accountData.erase([A]);
assert.strictEqual(counts.erased.run_sessions, 1);
@@@ -45,6 +50,8 @@ const B = 'usr_01JZ0000000000000000000BBB';
assert.strictEqual(await d.db.value('SELECT requester FROM run_jobs WHERE id = $1', ['job_01JZ0000000000000000000ADA']), 'user:deleted', 'the job stays, without its requester');
assert.strictEqual(await d.db.value('SELECT requester FROM run_jobs WHERE id = $1', ['job_01JZ0000000000000000000BQB']), `user:${B}`, 'Bo\'s job is untouched');
assert.strictEqual(counts.retained.run_jobs_tombstoned, 1);
++ assert.strictEqual(Number(await d.db.value('SELECT count(*) FROM placement_outcomes WHERE person_id = $1', [A])), 0, 'Ada\'s placement counts are erased');
++ assert.strictEqual(Number(await d.db.value('SELECT count(*) FROM placement_outcomes WHERE person_id = $1', [B])), 1, 'Bo\'s stay');
// The route answers: an unsigned delivery is refused (401), not lost (404) and not accepted.
const t = await boot({ db: d.db, env: { RUN_EVENTS_SECRET: 's'.repeat(40) }, appOpts: { accountData } });
ddiff --git a/test/dispatch.test.js b/test/dispatch.test.js
iindex a5bb68b..23b4290 100644
---- a/test/dispatch.test.js
++++ b/test/dispatch.test.js
@@@ -432,6 +432,77 @@ async function main() {
bot.devices.delete(OTHER_DEVICE);
});
++ await check('placement A1: the line, the learned outcome once, and a cancelled job not counted', async () => {
++ const cols = await t.db.many("SELECT column_name FROM information_schema.columns WHERE table_name = 'placement_outcomes' ORDER BY column_name");
++ assert.deepStrictEqual(cols.map((c) => c.column_name), ['n_w', 'ok_w', 'option_key', 'person_id', 'step_kind', 'updated_at']);
++ const ada = ids.newId('user');
++ const bo = ids.newId('user');
++ const P = PROJECT();
++ t.network.members.set(`${P} ${ada}`, { role: 'owner' });
++ t.network.members.set(`${P} ${bo}`, { role: 'owner' });
++ const DEV_A = `dev_${ids.ulid()}`;
++ const DEV_B = `dev_${ids.ulid()}`;
++ bot.devices.set(DEV_A, { classes: ['function'] });
++ bot.devices.set(DEV_B, { classes: ['function'] });
++ t.setOffers([
++ offer({ offer_id: 'off_learn_a', trust: 'user-owned', owner: ada, provider: 'self', node_id: DEV_A }),
++ offer({ offer_id: 'off_learn_b', trust: 'user-owned', owner: bo, provider: 'self', node_id: DEV_B }),
++ ]);
++ const named = { requirements: { kind: 'run.function', mobility: 'job', latency_class: 'background', objective: 'balanced', trust: ['user-owned'] } };
++ const token = t.network.signService({ sub: 'svc:codes', cap: ['run.job.submit', 'run.job.read', 'run.job.cancel'] });
++ const asPerson = async (subject, over) => {
++ const r = await t.call('POST', '/api/v1/jobs', { body: t.request(over), token, headers: { 'X-OV-Subject': subject, 'X-OV-Project': P } });
++ assert.strictEqual(r.status, 201, r.text);
++ return r.json.job;
++ };
++ const settle = async (job) => (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++
++ const mine = await settle(await asPerson(ada, named));
++ assert.strictEqual(mine.state, 'placed');
++ const explained = await t.call('GET', `/api/v1/jobs/${mine.job_id}?explain=1`, { token, headers: { 'X-OV-Subject': ada, 'X-OV-Project': P } });
++ assert.strictEqual(explained.status, 200, explained.text);
++ assert.ok(explained.json.explain && typeof explained.json.explain.line === 'string' && explained.json.explain.line.length > 0, explained.text);
++ assert.ok(explained.json.explain.line.includes('your machine'), explained.json.explain.line);
++ assert.ok(!explained.json.explain.line.includes('off_learn_a'), 'the line never shows a raw offer id');
++ assert.ok(Array.isArray(explained.json.explain.ranked));
++ assert.strictEqual(explained.json.explain.others_hidden, 0);
++ const storedExplain = await t.db.value('SELECT placement_explain FROM run_jobs WHERE id = $1', [mine.job_id]);
++ assert.strictEqual(Number(storedExplain.others_hidden), 0);
++ assert.ok(!JSON.stringify(storedExplain).includes("another person's machine"), 'the stored explain never lists another machine');
++ // Another person's machine is never listed, only counted.
++ const { publicExplain, displayNameOf } = require('../server/dispatch/placement');
++ assert.strictEqual(displayNameOf({ offer_id: 'off_named', trust: 'user-owned', owner: ada, detail: { name: 'Ada laptop' } }, ada), 'Ada laptop');
++ assert.strictEqual(displayNameOf({ offer_id: 'off_named', trust: 'third-party', provider: 'ovh' }, ada), 'ovh');
++ const hidden = publicExplain({ line: 'x', ranked: [], rejected: [
++ { id: 'off_other', why: "another person's machine" },
++ { id: 'off_none', why: 'a user-owned machine has no owner' },
++ { id: 'off_down', why: 'health down' },
++ ] });
++ assert.deepStrictEqual(hidden.rejected, [{ id: 'off_down', why: 'health down' }]);
++ assert.strictEqual(hidden.others_hidden, 2);
++ assert.ok(!JSON.stringify(hidden).includes('off_other') && !JSON.stringify(hidden).includes('off_none'));
++ const { explain: _explain, ...contractBody } = explained.json;
++ assert.ok(contracts.validate('run.job-read-result@1', contractBody).valid, 'the contract body is unchanged beside explain');
++ bot.start(mine.job_id);
++ bot.exit(mine.job_id, { reason: 'exited', code: 0, wallMs: 1000 });
++ await t.dispatch.poll();
++ const rows = await t.db.many('SELECT * FROM placement_outcomes WHERE person_id = $1', [ada]);
++ assert.strictEqual(rows.length, 1, 'one learned row');
++ assert.strictEqual(rows[0].option_key, 'off_learn_a');
++ assert.strictEqual(rows[0].step_kind, 'function');
++ assert.ok(Number(rows[0].ok_w) > 0 && Number(rows[0].n_w) === 1);
++ await t.dispatch.poll();
++ assert.strictEqual(Number((await t.db.one('SELECT n_w FROM placement_outcomes WHERE person_id = $1', [ada])).n_w), 1, 'learned once');
++
++ const cancelled = await settle(await asPerson(bo, named));
++ assert.strictEqual(cancelled.state, 'placed');
++ assert.strictEqual((await t.call('POST', `/api/v1/jobs/${cancelled.job_id}/cancel`, { token, headers: { 'X-OV-Subject': bo, 'X-OV-Project': P } })).status, 200);
++ await t.dispatch.poll();
++ assert.strictEqual(Number(await t.db.value('SELECT count(*) FROM placement_outcomes WHERE person_id = $1', [bo])), 0, 'a cancelled job is not counted');
++ bot.devices.delete(DEV_A);
++ bot.devices.delete(DEV_B);
++ });
++
await check('a pinned machine (Codes C9c): placed only there, refused when it is offline, never moved', async () => {
const ada = ids.newId('user');
const bo = ids.newId('user');