openvibers/openvibe.run Commits

Send browser jobs to the managed-browser provider and bill its clamped…

OP OpenVibers committed 14 Oct 2026 · 4e39cf3ad1 · 16 files, +1144 −74

Asked: Part of a cross-repository initiative (init-t14-step-2-run-r4a-the-manag-1013T2351): T14 step 2 = Run R4a, the managed-browser provider adapter (dossier /home/workstation/openvibe/agents/ds/prep/T14.md section 'T14 step 2'; Run R1a-R3 are on Run main, Run is NOT deployed and this initiative deploys…
Harness job init-t14-step-2-run-r-2351-s2-003348 (plan T14).

Changes

ddiff --git a/.env.example b/.env.example
iindex 54a774a..0aa061a 100644
---- a/.env.example
++++ b/.env.example
@@@ -73,4 +73,20 @@ OV_OAUTH_CLIENT_SECRET=
# Jobs one tick places and mirrors; the next tick takes the rest.
#RUN_DISPATCH_MAX_PER_TICK=200
--# Run holds no Billing token: Bot owns the node link and the per-second metering (plan T14 L1).
++# Bot owns the node link and the per-second metering of the jobs it carries (plan T14 L1).
++
++# ── The managed-browser provider (plan T14 R4a) ──
++# Class browser jobs placed on a kind=provider offer named browser-provider go to this API instead of Bot.
++# Off while the URL or the key is unset (browser jobs then stay queued). The key lives here only: never in an
++# offer, a job or a log. Outside development the URL must be https (or loopback).
++#RUN_BROWSER_PROVIDER_URL=https://browser-provider.example/v1
++#RUN_BROWSER_PROVIDER_API_KEY=
++#RUN_BROWSER_PROVIDER_TIMEOUT_MS=8000
++
++# ── Run's usage relay to Billing (the provider's jobs only; Bot meters its own) ──
++# Run mints its own token for audience openvibe.billing (billing.usage.record) from OV_OAUTH_CLIENT_SECRET.
++# Unset: the readings wait in run_usage_outbox, nothing is sent, and the jobs still run.
++#RUN_BILLING_URL=http://127.0.0.1:4600
++#RUN_BILLING_AUDIENCE=openvibe.billing
++#RUN_BILLING_INTERVAL_MS=5000
++#RUN_BILLING_TIMEOUT_MS=8000
ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 87d6b86..487cdcd 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -28,6 +28,16 @@ Notable changes to OpenVibe.Run. The service is unreleased; everything below is
one `run.job.*` event per state change, the node's stdout as stream `output` events, and the job_cancel.
- `GET /api/v1/jobs/:id/stream` also opens with a Bearer service token holding `run.job.stream` for the job's
project, beside the ticket.
++- The managed-browser provider adapter (plan T14 R4a, `server/dispatch/browser-provider.js`): class `browser` jobs
++ placed on a `kind=provider` offer named `browser-provider` are sent to the provider's API
++ (`browser-provider.job-create-request@1`, the frame alone: no project, subject or requester), mirrored, cancelled
++ and expired there; every answer is validated against `job-create-result@1` / `job-state@1` and a malformed one is
++ never adopted. Configured by `RUN_BROWSER_PROVIDER_URL`, `RUN_BROWSER_PROVIDER_API_KEY` (Run's env only, never
++ logged) and `RUN_BROWSER_PROVIDER_TIMEOUT_MS` (8000); off while the URL or the key is unset.
++- Run's own metering of those jobs: the provider's claimed wall time is clamped to Run's observed duration + 5 s
++ (a larger claim is logged and never billed), queued as `platform.usage-sample@1` readings in `run_usage_outbox`
++ (`migrations/0003_usage_outbox.sql`) with the end state, and relayed to Billing (`RUN_BILLING_URL`) with a Run
++ token for audience `openvibe.billing` holding `billing.usage.record` — used for nothing else.
### Changed
@@@ -41,6 +51,7 @@ Notable changes to OpenVibe.Run. The service is unreleased; everything below is
### Notes
--- Run holds no Billing token: OpenVibe.Bot owns the node link and the per-second metering (plan T14 L1).
++- OpenVibe.Bot owns the node link and the per-second metering of the jobs it carries (plan T14 L1); Run's Billing
++ token is for the external provider's jobs only.
- The repo is not deployed; the first deploy needs `ovhost`'s inventory, the `ov_run` database,
`/etc/openvibe/run.env` and `openvibe.run` DNS (see STATUS.json).
ddiff --git a/README.md b/README.md
iindex 4501198..e2df66b 100644
---- a/README.md
++++ b/README.md
@@@ -8,7 +8,9 @@ plan track T14.
OpenVibe.Run accepts jobs, places each one on a node that advertises the runtime class it needs, and follows it to
its end. A job belongs to a **project**, is billed per wall-clock second it runs, and is visible only to the token
that project presents. Run owns the API, the project scope and the placement; OpenVibe.Bot owns the node link and the
--per-second metering, so Run holds no Billing token at all (plan T14 L1).
++per-second metering of the jobs it carries (plan T14 L1). The one exception is the external managed-browser provider
++(plan T14 R4a): Run meters its jobs itself and posts their readings with a Billing token of its own, used for nothing
++else.
This repository is at plan **T14 steps 5 (R1b), 6 (R1c) and 7 (R3)**: the API is complete, and a queued job is
placed from the Fabric's offers and actually runs — the poller in `server/jobs/poller.js` (started by
@@@ -30,7 +32,7 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs
`bot.job.dispatch`, audience `openvibe.bot`) and Bot gives it to the node over its device link. Bot's dispatch is
idempotent by job id, so a retry after a crash or a timeout is safe.
4. **Mirror** — every tick mirrors each `placed` or `running` job from `GET {RUN_BOT_URL}/api/v1/jobs/{id}`: state,
-- timings, exit, result and the metered wall clock (Bot is the meter; Run never posts a usage record). The state
++ timings, exit, result and the metered wall clock (Bot is the meter; Run never posts a usage record for a job Bot carries). The state
change and its `run.job.started|succeeded|failed|cancelled|expired` event are one transaction, and the job's
stdout is carried into its SSE stream as `output` events. A job Bot answers 404 for after Run sent it fails
`run.job.worker_failed` after `RUN_DISPATCH_UNKNOWN_POLLS` polls — it is never mirrored forever.
@@@ -79,7 +81,8 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs
- Identity, tokens and projects (OpenVibe.Network); Run only verifies the tokens it is the audience of.
- The artifacts a job runs and the media its inputs come from (OpenVibe.Media; Run never fetches a caller-chosen
URL).
--- Money: Run writes no usage record and holds no Billing token.
++- Money for the jobs Bot carries: Bot writes their usage records. Run writes only the external provider's
++ (`billing.usage.record`, clamped to Run's own observed duration + 5 s).
## Depends on
@@@ -153,8 +156,11 @@ every answer before it is sent; errors are RFC 9457 `application/problem+json`.
request until it lands), and the job reaches `cancelled` when Bot says so. Run never answers `cancelled` for work a
node may still be doing. Cancelling an ended job answers it as it is.
- **Usage.** `usage.wall_ms` and `usage.seconds` are Bot's metered wall clock (Bot writes the
-- `run:<job id>:<n>` readings; Run holds no Billing token and never posts one). A running job therefore shows 0
-- until its `job_exit` carries the wall clock: Run does not estimate a number it cannot back.
++ `run:<job id>:<n>` readings; Run never posts those). A running job therefore shows 0
++ until its `job_exit` carries the wall clock: Run does not estimate a number it cannot back. A job the managed-browser
++ provider ran is metered by Run: the provider's claimed `wall_ms`, never above Run's own observed duration (its send
++ to the answer that showed the end) + 5 s — a larger claim is clamped and logged — queued as `run:<job id>:<n>`
++ readings in `run_usage_outbox` with the end state and relayed to Billing (`RUN_BILLING_URL`; unset, they wait).
- **Placement.** The node is chosen per tick from the offers Network publishes
(`GET {OV_NETWORK_URL}/api/v1/offers?kind=node`) by `openvibe-sdk/placement` `plan()`, on the job's own
`requirements`; `node_id`, `provider` and `region` are recorded on Run's row. A class no node offers keeps the job
ddiff --git a/STATUS.json b/STATUS.json
iindex 8ced5c5..2b646ca 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,11 +7,11 @@
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db) + Valkey 9",
"port": 4920,
"domain": "openvibe.run",
-- "contracts": "openvibe-contracts v0.143.0",
++ "contracts": "openvibe-contracts v0.155.0",
"sdk": "openvibe-sdk v0.44.0",
"shared": "openvibe-shared v4.1.3",
"packages": [
-- "openvibe-contracts v0.143.0",
++ "openvibe-contracts v0.155.0",
"openvibe-sdk v0.44.0",
"openvibe-shared v4.1.3"
],
@@@ -47,8 +47,9 @@
"ttl_ms expiry settles the job expired (run.job.ttl) and asks Bot to stop a job a node may hold",
"the job stream also opens with a Bearer service token holding run.job.stream for the job's project, besides the two-minute ticket",
"run_jobs (migrations/0001_jobs.sql) holds every job, its state, placement, timings, exit, result and the metered wall clock",
-- "Run holds no Billing token at all: Bot owns the node link and the per-second metering (plan T14 L1)",
++ "Bot owns the node link and the per-second metering of the jobs it carries (plan T14 L1); Run holds a Billing token of its own (billing.usage.record, audience openvibe.billing) used only for the external managed-browser provider's jobs",
++ "plan T14 R4a built, not deployed: the managed-browser provider adapter (server/dispatch/browser-provider.js; RUN_BROWSER_PROVIDER_URL / _API_KEY / _TIMEOUT_MS, off while unset) places class browser jobs on kind=provider offers named browser-provider, validates every answer against browser-provider.job-create-result@1 / job-state@1 and never adopts a malformed one; Run meters those jobs itself, the provider's claimed wall time clamped to Run's observed duration + 5 s, and relays the readings from run_usage_outbox (migrations/0003_usage_outbox.sql) to Billing (RUN_BILLING_URL; unset, they wait and the jobs still run)",
"truthful readiness (GET /api/ready: db required; valkey, Network JWKS, Events, Bot reachability and the poller's own status reported, never claimed) and the loopback-only /metrics gauges (jobs queued, active, ended, outbox backlog, streams)"
],
-- "updated": "2026-10-12"
++ "updated": "2026-10-13"
}
ddiff --git a/migrations/0003_usage_outbox.sql b/migrations/0003_usage_outbox.sql
nnew file mode 100644
iindex 0000000..86aaa4f
---- /dev/null
++++ b/migrations/0003_usage_outbox.sql
@@@ -0,0 +1,20 @@
++-- phase: expand
++-- Usage readings Run meters itself (plan T14 R4a): the external managed-browser provider's jobs only, since Bot meters
++-- every job it carries. platform.usage-sample@1 readings waiting for OpenVibe.Billing (billing.usage.record), the
++-- openvibe-sdk outbox shape (outboxSchema('run_usage_outbox')), as OpenVibe.Bot's own run_usage_outbox. event_id is
++-- the reading's idempotency_key run:<job id>:<second>, so a second is queued once however often it is written.
++CREATE TABLE IF NOT EXISTS run_usage_outbox (
++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ event_id text NOT NULL UNIQUE,
++ envelope jsonb NOT NULL,
++ traceparent text,
++ created_at bigint NOT NULL,
++ attempts integer NOT NULL DEFAULT 0,
++ next_attempt_at bigint NOT NULL DEFAULT 0,
++ sent_at bigint,
++ seq bigint,
++ rejected_at bigint,
++ last_error text
++);
++CREATE INDEX IF NOT EXISTS run_usage_outbox_due ON run_usage_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
++CREATE INDEX IF NOT EXISTS run_usage_outbox_sent ON run_usage_outbox (sent_at) WHERE sent_at IS NOT NULL;
ddiff --git a/package-lock.json b/package-lock.json
iindex d980bab..da4fe27 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -15,7 +15,7 @@
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.155.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
@@@ -872,9 +872,8 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.143.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
-- "integrity": "sha512-Cdbf+9RoPAUBibpM2zXQvtsiKMugb55dKubTt8pFZvAmxsLtGvX+zaC++yL1vOSub8FX3+GgqcGtoJS6tklrUg==",
++ "version": "0.155.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.155.0",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex e81fafa..420fdb1 100644
---- a/package.json
++++ b/package.json
@@@ -26,7 +26,7 @@
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.155.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
ddiff --git a/server/config.js b/server/config.js
iindex 6b96f99..e2f0988 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -5,8 +5,11 @@
* /etc/openvibe/run.env in production). loadConfig(env) is pure so tests build their own.
*
* Run is the job service of the network (plan T14, ADR-034): it owns the API, the project scope and the
-- * placement, on its own PostgreSQL (ADR-035). Bot owns the node link and the per-second metering, so Run
-- * holds no Billing token at all — Run never moves money (plan T14 L1).
++ * placement, on its own PostgreSQL (ADR-035). Bot owns the node link and the per-second metering of the jobs it
++ * carries (plan T14 L1). The one exception is an external managed-browser provider (plan T14 R4a): it is not a
++ * paired Node and writes no usage of its own, so Run meters those jobs itself and posts their readings with a
++ * Billing token of its own (billing.usage.record, audience openvibe.billing), used for nothing else. The
++ * provider's API key lives in Run's environment only: never in an offer, a job or a log.
*/
require('dotenv').config();
@@@ -119,6 +122,26 @@ function loadConfig(env = process.env) {
// Jobs one tick places and mirrors; the next tick takes the rest.
maxPerTick: Math.max(1, int(env.RUN_DISPATCH_MAX_PER_TICK, 200)),
},
++
++ // 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.
++ browserProvider: {
++ url: trim(String(env.RUN_BROWSER_PROVIDER_URL || '').trim()),
++ apiKey: String(env.RUN_BROWSER_PROVIDER_API_KEY || '').trim(),
++ timeoutMs: Math.max(100, int(env.RUN_BROWSER_PROVIDER_TIMEOUT_MS, 8000)),
++ },
++
++ // Run's own usage relay to OpenVibe.Billing (billing.usage.record, POST <url>/api/v1/usage), for the
++ // external provider's jobs only (Bot meters its own). The token is minted from Run's Network client for
++ // audience openvibe.billing. Unset URL or no OV_OAUTH_CLIENT_SECRET: readings wait in run_usage_outbox,
++ // nothing is sent, and the jobs still run.
++ billing: {
++ url: trim(String(env.RUN_BILLING_URL || '').trim()),
++ audience: env.RUN_BILLING_AUDIENCE || 'openvibe.billing',
++ intervalMs: Math.max(100, int(env.RUN_BILLING_INTERVAL_MS, 5000)),
++ timeoutMs: Math.max(100, int(env.RUN_BILLING_TIMEOUT_MS, 8000)),
++ },
};
}
ddiff --git a/server/dispatch/bot.js b/server/dispatch/bot.js
iindex f37c5d9..50619c4 100644
---- a/server/dispatch/bot.js
++++ b/server/dispatch/bot.js
@@@ -3,7 +3,7 @@
/**
* The Bot client — Run → OpenVibe.Bot over Bot's internal jobs API (plan T14 step 6, R1c). Bot owns the
* node link, the job row and the per-second metering; Run owns run.job.*, the project scope and the
-- * placement, so this client never writes a usage record (Run holds no Billing token, plan T14 L1).
++ * placement, so this client never writes a usage record (Bot meters every job it carries, plan T14 L1).
*
* send({ nodeId, job, project, subject, provider }) POST {RUN_BOT_URL}/api/v1/jobs → { job, sent }
* cancel(jobId) POST …/api/v1/jobs/{id}/cancel → { job, sent } | null
ddiff --git a/server/dispatch/browser-provider.js b/server/dispatch/browser-provider.js
nnew file mode 100644
iindex 0000000..bfb969b
---- /dev/null
++++ b/server/dispatch/browser-provider.js
@@@ -0,0 +1,219 @@
++'use strict';
++
++/**
++ * The managed-browser provider adapter — Run → an external managed-browser provider's job API (plan T14 R4a,
++ * ADR-034; Run's own outbound capability browser_provider.job.dispatch, contracts/manifests/capabilities). A
++ * class `browser` job placed on a kind=provider offer named `browser-provider` goes here instead of Bot: the
++ * provider is not a paired Node, so there is no Bot in between and Run meters the job itself.
++ *
++ * send({ job }) POST {url}/jobs browser-provider.job-create-request@1 → job-create-result@1
++ * state(jobId) GET {url}/jobs/{id} → browser-provider.job-state@1 | null
++ * cancel(jobId) POST {url}/jobs/{id}/cancel → browser-provider.job-state@1 | null
++ * meter({ jobId, claimedMs, observedMs }) → the wall clock Run bills: the provider's claim, clamped
++ *
++ * The request is built from the platform.job@1 frame alone (id, class, args, inputs, limits, egress) and
++ * validated before it is sent: it never carries the project, the subject, the requester or any credential, so
++ * the provider never learns who pays. The job id is the Idempotency-Key, so a resend after a crash answers
++ * the first job (200, created false) and never starts a second.
++ *
++ * The key: RUN_BROWSER_PROVIDER_API_KEY from Run's own environment, sent as `Authorization: Bearer <key>` and
++ * nowhere else — never read from an offer (platform.resource-offer@1 has no field for it), never logged,
++ * never in an error, never in status(). Provider text that ends up in an error is scrubbed of it, and a
++ * redirect is refused rather than followed with the header. Outside development the URL must be https (or
++ * loopback), else the adapter stays off.
++ *
++ * Nothing the provider answers is adopted unchecked: every 2xx body is validated against its contract, its
++ * `id` must be the job's, a result over 256 KiB is refused, and a body over MAX_BODY is not read. Such an
++ * answer is a DispatchError with retryable false (code run.dispatch.malformed): the poller treats it like a
++ * Bot answer botRowOf() refuses — the job stays as Run last saw it. A 404 on state or cancel is null (the
++ * provider does not know the job); a 5xx, a 401/403/408/429, a timeout or a transport failure is retryable
++ * (the next tick tries again); any other 4xx is a refusal (the job cannot run as asked).
++ *
++ * meter(): the provider's wall_ms is a claim (browser-provider.job-state@1). Run bills at most its own
++ * observed duration plus 5 s (decision B of plan T14 R4a): a larger claim is clamped and logged, never
++ * billed above the cap.
++ */
++const contracts = require('openvibe-contracts');
++const { DispatchError } = require('./bot');
++
++/** The offer `provider` name that routes a job to this adapter (the job row stores it as its provider). */
++const PROVIDER = 'browser-provider';
++/** Run's own outbound capability (planned in Contracts; no token grants it). */
++const CAPABILITY = 'browser_provider.job.dispatch';
++const REQUEST = 'browser-provider.job-create-request@1';
++const RESULT = 'browser-provider.job-create-result@1';
++const STATE = 'browser-provider.job-state@1';
++/** What Run bills above its own observed duration at most. */
++const METER_SLACK_MS = 5000;
++/** The largest answer read: 1 MiB of stdout (up to 4 bytes a character) and a 256 KiB result, with room. */
++const MAX_BODY = 6 * 1024 * 1024;
++const MAX_RESULT = 256 * 1024;
++/** Answers that say nothing about the job: a key, a grant or a rate the operator fixes. Retried. */
++const RETRY_4XX = new Set([401, 403, 408, 429]);
++
++const LOOPBACK = new Set(['localhost', '127.0.0.1', '[::1]', '::1']);
++
++/** The URL Run may send its key to: https, or loopback in any environment, or anything in development. */
++function urlProblem(url, isProduction) {
++ let u;
++ try { u = new URL(url); } catch { return 'RUN_BROWSER_PROVIDER_URL is not a URL'; }
++ if (u.username || u.password) return 'RUN_BROWSER_PROVIDER_URL carries credentials (the key belongs in RUN_BROWSER_PROVIDER_API_KEY)';
++ if (u.protocol === 'https:' || LOOPBACK.has(u.hostname) || (!isProduction && u.protocol === 'http:')) return null;
++ return 'RUN_BROWSER_PROVIDER_URL must be https (the key travels in a header)';
++}
++
++/** A body read with a ceiling, so a provider cannot make Run buffer without bound. → parsed JSON or undefined. */
++async function readJson(res) {
++ if (!res.body || typeof res.body.getReader !== 'function') {
++ const text = await res.text().catch(() => '');
++ if (Buffer.byteLength(text) > MAX_BODY) return undefined;
++ try { return text ? JSON.parse(text) : null; } catch { return undefined; }
++ }
++ const reader = res.body.getReader();
++ const parts = [];
++ let size = 0;
++ for (;;) {
++ const { done, value } = await reader.read();
++ if (done) break;
++ size += value.byteLength;
++ if (size > MAX_BODY) { await reader.cancel().catch(() => {}); return undefined; }
++ parts.push(value);
++ }
++ const text = Buffer.concat(parts).toString('utf8');
++ try { return text ? JSON.parse(text) : null; } catch { return undefined; }
++}
++
++const obj = (v) => (v && typeof v === 'object' && !Array.isArray(v) ? v : null);
++
++/**
++ * createBrowserProviderAdapter({ config, fetchImpl, now, log }) → { enabled, note, send, state, cancel, meter,
++ * status }. enabled is false without RUN_BROWSER_PROVIDER_URL or RUN_BROWSER_PROVIDER_API_KEY.
++ */
++function createBrowserProviderAdapter({ config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console } = {}) {
++ const { url, apiKey, timeoutMs } = (config && config.browserProvider) || {};
++ const problem = url ? urlProblem(url, Boolean(config && config.isProduction)) : null;
++ const enabled = Boolean(url && apiKey && !problem);
++ // Where the jobs go, for status and logs: the origin only (a path or query is the provider's business).
++ const origin = (() => { try { return new URL(url).origin; } catch { return null; } })();
++ const note = enabled
++ ? `the managed-browser provider: ${origin}, capability ${CAPABILITY}`
++ : !url ? 'the managed-browser provider is off: RUN_BROWSER_PROVIDER_URL is unset'
++ : !apiKey ? 'the managed-browser provider is off: RUN_BROWSER_PROVIDER_API_KEY is unset'
++ : `the managed-browser provider is off: ${problem}`;
++ const stat = { sent: 0, refused: 0, malformed: 0, clamped: 0 };
++
++ /** Provider text is never trusted to be free of what Run sent it. */
++ const scrub = (s) => (apiKey ? String(s).split(apiKey).join('[redacted]') : String(s));
++ const malformed = (why) => {
++ stat.malformed++;
++ return new DispatchError(502, 'run.dispatch.malformed', scrub(`the browser provider's answer was not adopted: ${why}`).slice(0, 500));
++ };
++
++ /** One call → the parsed 2xx body ({ status, doc }), null for a 404, or a DispatchError. */
++ async function call(method, path, { body, idempotencyKey } = {}) {
++ if (!enabled) throw new DispatchError(503, 'run.dispatch.disabled', note, { retryable: true });
++ const headers = { Accept: 'application/json', Authorization: `Bearer ${apiKey}` };
++ if (body !== undefined) headers['Content-Type'] = 'application/json';
++ if (idempotencyKey) headers['Idempotency-Key'] = idempotencyKey;
++ let res;
++ try {
++ res = await fetchImpl(`${url}${path}`, {
++ method, headers, redirect: 'error',
++ ...(body !== undefined ? { body: JSON.stringify(body) } : {}),
++ signal: AbortSignal.timeout(timeoutMs),
++ });
++ } catch (e) {
++ const why = e && (e.name === 'TimeoutError' || e.name === 'AbortError') ? `no answer within ${timeoutMs} ms` : 'unreachable';
++ throw new DispatchError(503, 'run.dispatch.unavailable', `the browser provider did not answer (${why})`, { retryable: true });
++ }
++ if (res.status === 404) { await res.body?.cancel?.().catch(() => {}); return null; }
++ const doc = await readJson(res);
++ if (res.ok) {
++ if (doc === undefined) throw malformed('the body is not JSON, or larger than Run reads');
++ return { status: res.status, doc };
++ }
++ const problemDoc = obj(doc) || {};
++ const code = typeof problemDoc.code === 'string' ? problemDoc.code.slice(0, 64) : null;
++ const detail = typeof problemDoc.detail === 'string' ? problemDoc.detail.slice(0, 300) : null;
++ const why = scrub(`the browser provider answered HTTP ${res.status}${code ? ` ${code}` : ''}${detail ? `: ${detail}` : ''}`);
++ if (res.status >= 500 || RETRY_4XX.has(res.status)) throw new DispatchError(503, 'run.dispatch.unavailable', why, { retryable: true });
++ stat.refused++;
++ throw new DispatchError(502, 'run.dispatch.refused', why, { retryable: false });
++ }
++
++ /** A job-state the provider answered, checked: the contract, the job's own id, the result's size. */
++ function stateOf(jobId, doc) {
++ const v = contracts.validate(STATE, doc);
++ if (!v.valid) throw malformed(`not a valid ${STATE}`);
++ if (doc.id !== jobId) throw malformed(`it names job ${String(doc.id).slice(0, 40)}, not ${jobId}`);
++ if (doc.result != null && Buffer.byteLength(JSON.stringify(doc.result)) > MAX_RESULT) throw malformed('its result is over 256 KiB');
++ return doc;
++ }
++
++ /** The provider's request, from the platform.job@1 frame's own keys only. */
++ function requestOf(job) {
++ const limits = obj(job.limits) || {};
++ const out = {
++ id: job.id,
++ class: job.class,
++ args: obj(job.args) || {},
++ inputs: Array.isArray(job.inputs) ? job.inputs : [],
++ limits: {
++ wall_ms: limits.wall_ms,
++ ...(limits.cpu_ms != null ? { cpu_ms: limits.cpu_ms } : {}),
++ ...(limits.mem_bytes != null ? { mem_bytes: limits.mem_bytes } : {}),
++ },
++ // platform.job@1 spells "no network" deny as well as none; the provider's contract only knows none.
++ egress: !job.net || job.net === 'deny' ? 'none' : job.net,
++ };
++ const v = contracts.validate(REQUEST, out);
++ if (!v.valid) throw new DispatchError(422, 'run.dispatch.invalid_job', `the job is not a valid ${REQUEST}`);
++ return out;
++ }
++
++ return {
++ enabled,
++ note,
++ provider: PROVIDER,
++ /** Hand one job to the provider. → { job: job-state@1, created }. Idempotent by job id at the provider. */
++ async send({ job }) {
++ const body = requestOf(obj(job) || {});
++ const out = await call('POST', '/jobs', { body, idempotencyKey: body.id });
++ if (!out) throw new DispatchError(502, 'run.dispatch.refused', 'the browser provider answered 404 to the job POST (is RUN_BROWSER_PROVIDER_URL its API base?)');
++ if (!contracts.validate(RESULT, out.doc).valid) throw malformed(`not a valid ${RESULT}`);
++ const state = stateOf(body.id, out.doc.job);
++ stat.sent++;
++ return { job: state, created: out.doc.created === true };
++ },
++ /** The provider's state of a job; null when it does not know the job. */
++ async state(jobId) {
++ const out = await call('GET', `/jobs/${encodeURIComponent(jobId)}`);
++ return out ? stateOf(jobId, out.doc) : null;
++ },
++ /** Ask the provider to stop a job; null when it does not know the job (nothing to stop). */
++ async cancel(jobId) {
++ const out = await call('POST', `/jobs/${encodeURIComponent(jobId)}/cancel`, { idempotencyKey: `${jobId}:cancel` });
++ return out ? stateOf(jobId, out.doc) : null;
++ },
++ /**
++ * The wall clock Run bills for a provider job: the provider's claim, never above Run's own observed
++ * duration plus 5 s. → { wallMs, claimedMs, capMs, clamped }
++ */
++ meter({ jobId = null, claimedMs, observedMs }) {
++ const claimed = Number.isInteger(claimedMs) && claimedMs > 0 ? claimedMs : 0;
++ const capMs = Math.max(0, Math.floor(Number(observedMs) || 0)) + METER_SLACK_MS;
++ const clamped = claimed > capMs;
++ if (clamped) {
++ stat.clamped++;
++ log.warn(`[Run] job ${jobId || '?'}: the browser provider claims ${claimed} ms, Run observed ${capMs - METER_SLACK_MS} ms: billed ${capMs} ms (clamped to observed + ${METER_SLACK_MS} ms)`);
++ }
++ return { wallMs: clamped ? capMs : claimed, claimedMs: claimed, capMs, clamped };
++ },
++ status: () => ({
++ enabled, note, provider: PROVIDER, capability: CAPABILITY, url: origin, timeout_ms: timeoutMs ?? null,
++ key_set: Boolean(apiKey), ...stat,
++ }),
++ };
++}
++
++module.exports = { createBrowserProviderAdapter, PROVIDER, CAPABILITY, METER_SLACK_MS, MAX_RESULT };
ddiff --git a/server/dispatch/index.js b/server/dispatch/index.js
iindex cacfb68..fda3c74 100644
---- a/server/dispatch/index.js
++++ b/server/dispatch/index.js
@@@ -20,27 +20,41 @@
* sweep() ttl_ms ran out before placement: settle expired with run.job.unplaceable (never placed) or
* run.job.ttl (a node may hold it), and cancel on Bot.
*
-- * `enabled` is false when Run has no Network client credentials (OV_OAUTH_CLIENT_SECRET): it cannot mint the
-- * token Bot requires, so nothing is sent and every job stays queued — the same no-op the seam was before
-- * step 6, now for a stated reason. Nothing starts at module load: server/index.js calls start().
++ * The managed-browser provider (plan T14 R4a, server/dispatch/browser-provider.js): with RUN_BROWSER_PROVIDER_URL
++ * and its key set, class browser jobs may be placed on a kind=provider offer named `browser-provider` and sent to
++ * the provider instead of Bot; Run meters those jobs itself (clamped to its own observed duration + 5 s) and its
++ * usage relay (server/jobs/usage.js) posts the readings to Billing. The rest of the bridge is unchanged.
++ *
++ * `enabled` is false when neither Bot nor the provider can be called: without Network client credentials
++ * (OV_OAUTH_CLIENT_SECRET) Run cannot mint the token Bot requires, and without the provider's URL and key the
++ * adapter is off, so nothing is sent and every job stays queued — the same no-op the seam was before step 6,
++ * now for a stated reason. Nothing starts at module load: server/index.js calls start().
*/
const { createBotClient } = require('./bot');
++const { createBrowserProviderAdapter, PROVIDER: BROWSER_PROVIDER } = require('./browser-provider');
const { createPlacement, workerCapability } = require('./placement');
const { createPoller } = require('../jobs/poller');
++const { createUsageRelay } = require('../jobs/usage');
/**
-- * createDispatcher({ config, db, store, events, stream, log, now, fetchImpl, bot, placement }) → the
-- * dispatcher interface. Every dependency is injectable so tests build their own Bot, offers and clock.
++ * createDispatcher({ config, db, store, events, stream, log, now, fetchImpl, bot, provider, usageRelay,
++ * placement }) → the dispatcher interface. Every dependency is injectable so tests build their own Bot,
++ * 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, placement = null } = {}) {
++ now = () => Date.now(), fetchImpl = globalThis.fetch, bot = null, provider = null, usageRelay = null, placement = null } = {}) {
const botClient = bot || createBotClient({ config, fetchImpl, now, log });
-- const placer = placement || createPlacement({ config, fetchImpl, now, log });
-- const poller = createPoller({ config, db, store, events, stream, bot: botClient, placement: placer, now, log });
-- const enabled = botClient.enabled;
-- const note = enabled
-- ? botClient.note
-- : `${botClient.note}: every job stays queued (plan T14 step 6 is built, but Run cannot call Bot)`;
++ const adapter = provider || createBrowserProviderAdapter({ config, fetchImpl, now, log });
++ const relay = usageRelay || createUsageRelay({ db, config, fetchImpl, now, log });
++ // 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 enabled = Boolean(botClient.enabled || adapter.enabled);
++ const note = botClient.enabled
++ ? (adapter.enabled ? `${botClient.note}; ${adapter.note}` : botClient.note)
++ : adapter.enabled
++ ? `${adapter.note}; ${botClient.note}: only browser jobs on the provider leave queued`
++ : `${botClient.note}: every job stays queued (plan T14 step 6 is built, but Run cannot call Bot)`;
/** Send a queued job to a node. → { placed, reason }. Never throws: the poller retries. */
async function place(row) {
@@@ -67,18 +81,20 @@ function createDispatcher({ config = null, db = null, store = null, events = nul
return {
enabled, note, place, cancel, poll, sweep,
-- start: () => poller.start(),
-- stop: () => poller.stop(),
++ start: () => { relay.start(); return poller.start(); },
++ stop: () => { poller.stop(); return relay.stop(); },
status: () => ({
enabled, note,
bot_url: config ? config.dispatch.botUrl : null,
audience: config ? config.dispatch.botAudience : null,
++ browser_provider: adapter.status(),
++ usage_relay: relay.status(),
placement: placer.status(),
poller: poller.status(),
}),
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, placement: placer, poller },
++ deps: { config, db, store, events, stream, log, bot: botClient, provider: adapter, usageRelay: relay, placement: placer, poller },
};
}
ddiff --git a/server/dispatch/placement.js b/server/dispatch/placement.js
iindex 067b0f7..9e19fe7 100644
---- a/server/dispatch/placement.js
++++ b/server/dispatch/placement.js
@@@ -24,6 +24,15 @@
* placed on a stranger's personal Node; Run drops user-owned offers unless the requirements name
* user-owned (see For Opus: the SDK's TRUST_ORDER and the contract disagree).
*
++ * External providers (plan T14 R4a): when an adapter is on (server/dispatch/browser-provider.js), the offers of
++ * kind=provider (GET …/api/v1/offers?kind=provider) whose `provider` names it and that advertise worker:<class>
++ * for a class the adapter runs are candidates beside the nodes, and plan() chooses among both. Such an offer has
++ * no device, so the placement's node is its node_id when it names one and its offer_id otherwise; the chosen
++ * `provider` is what the poller routes on (the job row stores it). A node offer cannot borrow that name: a kind
++ * node offer naming an external provider is dropped, so a job Bot carries is never mirrored from a provider.
++ * The provider's API key is Run's own environment's and never an offer's (platform.resource-offer@1 has no field
++ * for one).
++ *
* Every offer is validated against platform.resource-offer@1 before it reaches plan(); the offers answer is
* cached for RUN_OFFERS_TTL_MS, so one tick places many jobs off one read. A Network that does not answer
* throws PlacementError(retryable) — the poller tries again next tick and never fails a job for it. An
@@@ -32,6 +41,10 @@
const contracts = require('openvibe-contracts');
const { plan } = require('openvibe-sdk/placement');
const { requirementsDefault } = require('../util');
++const { PROVIDER: BROWSER_PROVIDER } = require('./browser-provider');
++
++/** The offer `provider` names that route to an external adapter, never to Bot. */
++const EXTERNAL_PROVIDERS = new Set([BROWSER_PROVIDER]);
/** The reserved Fabric capability of a runtime class (platform.runtime-class@1; ADR-046). */
const workerCapability = (runtimeClass) => `worker:${runtimeClass}`;
@@@ -48,18 +61,26 @@ class PlacementError extends Error {
const obj = (v) => (v && typeof v === 'object' && !Array.isArray(v) ? v : {});
--function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console } = {}) {
++/**
++ * 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 = {} } = {}) {
const { url } = config.network;
-- const offersUrl = `${url}/api/v1/offers?kind=node`;
++ const offersUrlOf = (kind) => `${url}/api/v1/offers?kind=${kind}`;
++ const offersUrl = offersUrlOf('node');
const { offersTtlMs, timeoutMs } = config.dispatch;
-- let cache = { at: 0, offers: [], fetched: 0 };
++ const caches = new Map(); // kind → { at, offers, fetched }
++ const external = Object.entries(providers).filter(([name, classes]) => EXTERNAL_PROVIDERS.has(name) && Array.isArray(classes) && classes.length);
-- /** The published node offers, cached briefly. A Network that does not answer is a retryable error. */
-- async function offers() {
++ /** The published offers of one kind, cached briefly. A Network that does not answer is a retryable error. */
++ async function offers(kind = 'node') {
++ const cache = caches.get(kind) || { at: 0, offers: [], fetched: 0 };
if (cache.at && now() - cache.at < offersTtlMs) return cache.offers;
++ const href = offersUrlOf(kind);
let res;
try {
-- res = await fetchImpl(offersUrl, { headers: { Accept: 'application/json' }, signal: AbortSignal.timeout(timeoutMs) });
++ res = await fetchImpl(href, { headers: { Accept: 'application/json' }, signal: AbortSignal.timeout(timeoutMs) });
} catch (e) {
const why = e && (e.name === 'TimeoutError' || e.name === 'AbortError') ? `no answer within ${timeoutMs} ms` : 'unreachable';
throw new PlacementError('run.dispatch.no_offers', `Network did not answer for the resource offers (${why})`);
@@@ -78,7 +99,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
}
good.push(offer);
}
-- cache = { at: now(), offers: good, fetched: good.length };
++ caches.set(kind, { at: now(), offers: good, fetched: good.length });
return good;
}
@@@ -90,10 +111,18 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
const want = workerCapability(row.class);
const requirements = { ...obj(row.job).requirements || requirementsDefault(row.class) };
const userOwnedNamed = Array.isArray(requirements.trust) && requirements.trust.includes('user-owned');
-- const candidates = (await offers()).filter((o) => o.kind === 'node'
++ const trusted = (o) => (o.capabilities || []).includes(want) && (o.trust !== 'user-owned' || userOwnedNamed);
++ const candidates = (await offers('node')).filter((o) => o.kind === 'node'
&& typeof o.node_id === 'string' && o.node_id.length > 0
-- && (o.capabilities || []).includes(want)
-- && (o.trust !== 'user-owned' || userOwnedNamed));
++ && !EXTERNAL_PROVIDERS.has(o.provider)
++ && trusted(o));
++ // An external provider that runs this class: its kind=provider offers are candidates too.
++ const names = new Set(external.filter(([, classes]) => classes.includes(row.class)).map(([name]) => name));
++ if (names.size) {
++ candidates.push(...(await offers('provider')).filter((o) => o.kind === 'provider' && names.has(o.provider)
++ && typeof o.offer_id === 'string' && o.offer_id.length > 0 && o.offer_id.length <= 128
++ && trusted(o)));
++ }
if (!candidates.length) return null;
const req = { ...requirements, capabilities: [...new Set([...(requirements.capabilities || []), want])] };
contracts.assertValid('platform.workload-requirements@1', req);
@@@ -104,7 +133,9 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
return null;
}
return {
-- node_id: chosen.node_id, provider: chosen.provider ?? null, region: chosen.region ?? null,
++ // 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,
};
}
@@@ -113,8 +144,14 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
choose,
offers,
workerCapability,
-- status: () => ({ offers_url: offersUrl, cached: cache.fetched, fetched_at: cache.at ? new Date(cache.at).toISOString() : null, ttl_ms: offersTtlMs }),
++ status: () => {
++ const cache = caches.get('node') || { at: 0, fetched: 0 };
++ return {
++ offers_url: offersUrl, cached: cache.fetched, fetched_at: cache.at ? new Date(cache.at).toISOString() : null, ttl_ms: offersTtlMs,
++ providers: external.map(([name, classes]) => ({ provider: name, classes, cached: (caches.get('provider') || { fetched: 0 }).fetched })),
++ };
++ },
};
}
--module.exports = { createPlacement, PlacementError, workerCapability };
++module.exports = { createPlacement, PlacementError, workerCapability, EXTERNAL_PROVIDERS };
ddiff --git a/server/index.js b/server/index.js
iindex 3c7f3aa..b70acbe 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -13,7 +13,9 @@
* Background jobs (RUN_JOBS=off disables them): the events outbox relay and the dispatcher's poller
* (plan T14 step 6, server/jobs/poller.js; one tick every RUN_POLL_MS), which places queued jobs on a node
* from the Fabric offers, sends them to Bot and mirrors what Bot answers. Without OV_OAUTH_CLIENT_SECRET
-- * Run cannot mint its Bot token and every job stays queued.
++ * Run cannot mint its Bot token and every job stays queued. With RUN_BROWSER_PROVIDER_URL and its key, browser
++ * jobs may go to the managed-browser provider instead (plan T14 R4a); the dispatcher also runs Run's usage relay
++ * for those jobs (RUN_BILLING_URL; unset, their readings wait in run_usage_outbox).
*/
const { loadConfig } = require('./config');
const { openDb, migrate } = require('./db');
@@@ -45,7 +47,7 @@ async function main() {
const server = app.listen(config.port, config.host, () => {
console.log(`[Run] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (store ${db.store}; valkey ${valkey ? 'on' : 'off: this process only'})`);
-- console.log(`[Run] jobs ${config.jobs.enabled ? 'on' : 'off (RUN_JOBS=off)'}; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (run.job.* events wait in the outbox)'}; stream tickets ${ticketKey.enabled ? `on (${config.stream.keyId}${ticketKey.ephemeral ? ', ephemeral key' : ''})` : 'off (no stream key)'}; dispatcher ${dispatch.enabled ? `on → ${config.dispatch.botUrl} (poll ${config.dispatch.pollMs} ms)` : 'off: no OV_OAUTH_CLIENT_SECRET, jobs stay queued'}`);
++ console.log(`[Run] jobs ${config.jobs.enabled ? 'on' : 'off (RUN_JOBS=off)'}; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (run.job.* events wait in the outbox)'}; stream tickets ${ticketKey.enabled ? `on (${config.stream.keyId}${ticketKey.ephemeral ? ', ephemeral key' : ''})` : 'off (no stream key)'}; dispatcher ${dispatch.enabled ? `on (poll ${config.dispatch.pollMs} ms): ${dispatch.note}` : `off: ${dispatch.note}`}; ${dispatch.status().usage_relay.note}`);
});
server.keepAliveTimeout = 65_000;
ddiff --git a/server/jobs/poller.js b/server/jobs/poller.js
iindex 9a30742..ef08c0a 100644
---- a/server/jobs/poller.js
++++ b/server/jobs/poller.js
@@@ -22,13 +22,26 @@
* ignored. A job Bot answers 404 for after Run sent it fails after RUN_DISPATCH_UNKNOWN_POLLS polls — it
* is never mirrored forever.
*
++ * External providers (plan T14 R4a): a job placed on a kind=provider offer named `browser-provider` is sent to the
++ * managed-browser adapter (server/dispatch/browser-provider.js) instead of Bot, and every later call for it —
++ * mirror, cancel, the ttl's stop — goes to the adapter too: the route is the provider Run stored on its own row,
++ * never anything an answer says (Bot's adopt() cannot hand a job that name). The adapter's answers are validated
++ * before they reach here; one that is not is refused exactly as botRowOf() refuses Bot's. The provider names no
++ * node and no project. Its wall_ms is a claim: at the end state Run bills at most its own observed duration —
++ * from its own send (placed_at) to the answer that showed the end — plus 5 s, writes that as the job's wall
++ * clock and queues the per-second readings (server/jobs/usage.js) in the same transaction as the end state.
++ * A send that may have reached the provider without Run recording the placement (a timeout, a malformed
++ * answer) is remembered, and if the job then ends without being placed the provider is asked to stop it.
++ *
* stdout: Bot holds the last 1 MiB of a job's output as `{ text, first_seq, last_seq, truncated }`, so each
* tick streams only what Run has not streamed yet into the job's ring as `output` events with the ring's
* own monotonic seq. Output is best effort by contract (a chunk lost with the node's link is never resent):
* 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 } = require('../dispatch/placement');
++const { workerCapability, EXTERNAL_PROVIDERS } = require('../dispatch/placement');
++const { PROVIDER: BROWSER_PROVIDER, METER_SLACK_MS } = require('../dispatch/browser-provider');
++const usage = require('./usage');
/** Where a state sits in the job's life, so a Bot answer can never move a job backwards. */
const RANK = { queued: 0, placed: 1, running: 2, succeeded: 3, failed: 3, cancelled: 3, expired: 3 };
@@@ -37,12 +50,16 @@ const STATES = ['queued', 'placed', 'running', 'succeeded', 'failed', 'cancelled
const obj = (v) => (v && typeof v === 'object' && !Array.isArray(v) ? v : {});
const cancelling = (row) => obj(row.job).cancel_requested === true;
++/** The provider state a Run state stands for: the provider's queued is a job Run already placed. */
++const RUN_STATE_OF = { queued: 'placed', running: 'running', succeeded: 'succeeded', failed: 'failed', cancelled: 'cancelled', expired: 'expired' };
/**
-- * createPoller({ config, db, store, events, stream, bot, placement, now, log }) → { tick, sweep, placeOne,
-- * cancelNow, start, stop, status }. Every dependency is injected; nothing here starts at module load.
++ * createPoller({ config, db, store, events, stream, bot, provider, usageRelay, placement, now, log }) → { tick,
++ * sweep, placeOne, cancelNow, start, stop, status }. `provider` is the managed-browser adapter (absent: no job
++ * 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, placement, now = () => Date.now(), log = console } = {}) {
++function createPoller({ config, db, store, events, stream, bot, provider = null, usageRelay = null, placement, 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
@@@ -51,8 +68,18 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
// Jobs whose last placement attempt found no eligible offer (process memory: after a restart a waiting
// job is simply re-tried; at its ttl it then ends with the plain ttl outcome).
const noOffer = new Set();
++ // Jobs sent to the provider whose placement Run has not recorded (a timeout, a malformed answer): if one ends
++ // without being placed, the provider is asked to stop it (process memory, like the sets above).
++ const unconfirmed = new Set();
let offersWarnedAt = 0;
-- const stat = { ticks: 0, swept: 0, placed: 0, sent: 0, mirrored: 0, cancels_sent: 0, failed: 0, expired: 0, errors: 0, last_tick_at: null, last_error: null };
++ /** Where a job's calls go: the provider Run stored on its row decides, never an answer. */
++ const external = (row) => row.provider === BROWSER_PROVIDER;
++ const anyEnabled = () => Boolean(bot.enabled || (provider && provider.enabled));
++ const absent = async () => { throw Object.assign(new Error('the browser provider adapter is absent'), { code: 'run.dispatch.disabled', retryable: true }); };
++ const disabled = { enabled: false, state: absent, cancel: absent };
++ const workerOf = (row) => (external(row) ? provider || disabled : bot);
++ const nameOf = (row) => (external(row) ? 'the browser provider' : 'Bot');
++ const stat = { ticks: 0, swept: 0, placed: 0, sent: 0, mirrored: 0, cancels_sent: 0, failed: 0, expired: 0, errors: 0, metered: 0, clamped: 0, last_tick_at: null, last_error: null };
let timer = null;
let busy = false;
let inflight = null; // the tick in progress, so a caller can wait for it instead of skipping
@@@ -71,14 +98,17 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
}
/** Take one job to its end state with the run.job.* event of that state, in one transaction. */
-- async function end(row, at, { state, exit = null, code = null, result = null, wallMs = null, errorCode = null, errorDetail = null }) {
++ async function end(row, at, { state, exit = null, code = null, result = null, wallMs = null, errorCode = null, errorDetail = null, readings = null }) {
const settled = await db.tx(async (t) => {
const s = await store.settle(t, row.id, { state, exit, code, result, wallMs, errorCode, errorDetail, at });
// run.job.<state> exists if and only if the state change committed; a replay is the row already
// there (event_id run:<job id>:<state>), so emitting again is one row, never a second event.
if (s.changed) await events.ofState(t, s.row);
++ // A provider job's usage: queued with the end state, or not at all.
++ if (s.changed && readings && readings.length) await usage.enqueue(t, readings, at);
return s;
});
++ if (settled.changed && readings && readings.length) { stat.metered++; usageRelay?.kick?.(); }
if (settled.changed) {
if (state === 'failed') stat.failed++;
if (state === 'expired') stat.expired++;
@@@ -94,6 +124,26 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
/** Time to forget everything this process held about a job: it has ended. */
function forget(jobId) { cancelSent.delete(jobId); unknown.delete(jobId); cursors.delete(jobId); }
++ /** When Run sent a provider job (its own placed_at, Run's clock): where Run's observed duration starts. */
++ const sentAtOf = (row) => {
++ const placedAt = Number(obj(row.job).placed_at);
++ return Number.isFinite(placedAt) && placedAt > 0 ? placedAt : Number(row.created_at);
++ };
++
++ /**
++ * What Run bills for a provider job that ended: the claim clamped to Run's observed duration + 5 s, and the
++ * per-second readings of exactly that. Project and subject are Run's own row's, never the provider's.
++ */
++ function meterOf(row, claimedMs, seenAt) {
++ const m = provider.meter({ jobId: row.id, claimedMs, observedMs: seenAt - sentAtOf(row) });
++ if (m.clamped) stat.clamped++;
++ const readings = usage.readingsOf({
++ jobId: row.id, wallMs: m.wallMs, startMs: sentAtOf(row), node: row.node_id || undefined,
++ provider: BROWSER_PROVIDER, project: row.project_id, subject: row.requester,
++ });
++ return { ...m, readings };
++ }
++
/**
* One job's stdout, appended to Run's stream ring. Bot's ring is a suffix of the job's chunks, so the
* cursor remembers (the last chunk_seq streamed, how many characters of that ring's text they were, and
@@@ -147,15 +197,19 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
* carries code null, which exit allows), and a node that may hold it is told to stop.
*/
async function expireOne(row, at) {
++ let meter = null;
if (row.node_id) {
-- try { await bot.cancel(row.id); } catch (e) { log.warn(`[Run] job ${row.id}: job_cancel did not reach Bot (${e.code || e.message})`); }
++ let answer;
++ try { answer = await workerOf(row).cancel(row.id); } catch (e) { log.warn(`[Run] job ${row.id}: job_cancel did not reach ${nameOf(row)} (${e.code || e.message})`); }
++ // A provider job may have run before its ttl ran out: what it ran is billed, clamped like any other.
++ if (external(row) && answer && answer.wall_ms > 0) meter = meterOf(row, answer.wall_ms, now());
}
// A job no node ever took ends with run.job.unplaceable (ADR-036 section 2): it waited for an offer of
// worker:<class> until its ttl. A placed job that ran out carries no error code (exit reason ttl).
// Only a job whose last placement attempt found no eligible offer is unplaceable; one that simply ran out
// of time (Bot or Network unreachable, a very short ttl) keeps the plain ttl outcome.
const settled = row.node_id
-- ? await end(row, at, { state: 'expired', exit: 'ttl', errorDetail: `ttl_ms ran out with the job placed on ${row.node_id}` })
++ ? await end(row, at, { state: 'expired', exit: 'ttl', errorDetail: `ttl_ms ran out with the job placed on ${row.node_id}`, ...(meter ? { wallMs: meter.wallMs, readings: meter.readings } : {}) })
: noOffer.has(row.id)
? await end(row, at, { state: 'expired', exit: 'ttl', errorCode: 'run.job.unplaceable', errorDetail: `ttl_ms ran out before any node offering ${workerCapability(row.class)} took the job` })
: await end(row, at, { state: 'expired', exit: 'ttl', errorDetail: 'ttl_ms ran out before a node was placed' });
@@@ -190,19 +244,36 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
} catch (e) {
return fail(row, at, 'run.job.worker_failed', `Run's own job record is not a valid platform.job@1: ${e.message}`);
}
++ // An external provider's offer goes to its adapter (the frame alone: no project, no subject); a node's to Bot.
++ const toProvider = chosen.kind === 'provider' && chosen.provider === BROWSER_PROVIDER;
++ const worker = toProvider ? (provider || disabled) : bot;
++ const name = toProvider ? 'the browser provider' : 'Bot';
++ if (!worker.enabled) return note(row.id, 'queued', { reason: 'dispatch.disabled' });
++ // Run's own clock at the send: where a provider job's observed duration starts.
++ const sentAt = toProvider ? now() : at;
let sent;
try {
-- sent = await bot.send({ nodeId: chosen.node_id, job: frame, project: row.project_id, subject: row.requester, provider: chosen.provider });
++ if (toProvider) {
++ unconfirmed.add(row.id);
++ const out = await provider.send({ job: frame });
++ sent = { sent: true, created: out.created };
++ } else {
++ sent = await bot.send({ nodeId: chosen.node_id, job: frame, project: row.project_id, subject: row.requester, provider: chosen.provider });
++ }
} catch (e) {
-- if (e.code === 'bot.job_id_reused') return adopt(row, at, e);
++ if (!toProvider && e.code === 'bot.job_id_reused') return adopt(row, at, e);
if (e.retryable) {
stat.last_error = e.message;
-- warnSlowly(row, at, `[Run] job ${row.id}: Bot did not take the job (${e.code || e.message}); the next poll tries again`);
++ warnSlowly(row, at, `[Run] job ${row.id}: ${name} did not take the job (${e.code || e.message}); the next poll tries again`);
return note(row.id, 'queued', { reason: e.code || 'unavailable' });
}
-- // Bot refused the frame itself (a class the node does not advertise, an unknown device, a body
-- // it will not accept): the job cannot run as asked and fails rather than waiting forever.
-- return fail(row, at, 'run.job.worker_failed', `${e.code || 'bot refused the job'}: ${e.detail || ''}`.trim());
++ // A plain refusal took nothing; a malformed answer may stand for a job the provider holds, so the
++ // provider is asked to stop it (tick() does, once the job has ended here).
++ if (toProvider && e.code !== 'run.dispatch.malformed') unconfirmed.delete(row.id);
++ // The worker refused the frame itself (a class the node does not advertise, an unknown device, a
++ // body it will not accept, an answer Run cannot adopt): the job cannot run as asked and fails rather
++ // than waiting forever.
++ return fail(row, at, 'run.job.worker_failed', `${e.code || 'the worker refused the job'}: ${e.detail || ''}`.trim());
}
// Run's own record of where the job went: Bot's answer is not where the placement comes from. The row
// is re-read inside the transaction, because a cancel (or the ttl sweep) may have landed while the
@@@ -211,12 +282,13 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
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 }, at);
++ return store.mirror(t, row.id, { state: 'placed', node_id: chosen.node_id, provider: chosen.provider, region: chosen.region }, sentAt);
});
++ unconfirmed.delete(row.id);
if (!placed) {
let stopped = false;
-- try { stopped = (await bot.cancel(row.id)) !== undefined; } catch { stopped = false; }
-- log.warn(`[Run] job ${row.id}: it left queued while Run was sending it to ${chosen.node_id}; ${stopped ? 'Bot was asked to stop it' : 'Bot could NOT be asked to stop it'}`);
++ try { stopped = (await worker.cancel(row.id)) !== undefined; } catch { stopped = false; }
++ log.warn(`[Run] job ${row.id}: it left queued while Run was sending it to ${chosen.node_id}; ${stopped ? `${name} was asked to stop it` : `${name} could NOT be asked to stop it`}`);
// No node_id in the answer: the job is not placed (its end state was reached first) even though
// Bot briefly held the frame.
return note(row.id, 'cancelled', { raced: true, stopped, node: chosen.node_id });
@@@ -239,7 +311,8 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
const nodeId = b && typeof b.node_id === 'string' ? b.node_id : null;
if (!nodeId || nodeId.length > 128) return fail(row, at, 'run.job.worker_failed', `Bot already holds this job id (${err.detail})`);
const updated = await db.tx((t) => store.mirror(t, row.id, {
-- state: null, node_id: nodeId, provider: b.provider ?? null, region: b.region ?? null,
++ // Bot cannot hand a job an external provider's name: that would route its later calls away from Bot.
++ state: null, node_id: nodeId, provider: EXTERNAL_PROVIDERS.has(b.provider) ? null : (b.provider ?? null), region: b.region ?? null,
started_ms: b.started_ms ?? null, wall_ms: b.wall_ms ?? null, usage_read: b.usage_read ?? null,
}, at));
log.warn(`[Run] job ${row.id}: Bot already held it on ${nodeId} (${err.detail}); Run adopted Bot's placement instead of failing the job`);
@@@ -251,11 +324,12 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
const n = (unknown.get(row.id) || 0) + 1;
unknown.set(row.id, n);
if (n < config.dispatch.unknownPolls) return note(row.id, row.state, { bot_404: n });
-- return fail(row, at, 'run.job.worker_failed', `Bot has no job ${row.id} (it was sent to ${row.node_id || 'no node'}): giving up after ${n} polls`);
++ return fail(row, at, 'run.job.worker_failed', `${nameOf(row)} has no job ${row.id} (it was sent to ${row.node_id || 'no node'}): giving up after ${n} polls`);
}
-- /** Mirror: what Bot answers about a job Run placed. */
++ /** Mirror: what Bot (or, for a provider job, the provider) answers about a job Run placed. */
async function mirrorOne(row, at) {
++ if (external(row)) return mirrorProvider(row, at);
let answer;
try { answer = await bot.state(row.id); }
catch (e) {
@@@ -332,12 +406,95 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
return note(row.id, updated.state, { moved, sent: b.sent_at ? true : undefined });
}
++ /**
++ * Mirror a provider job: the adapter's validated browser-provider.job-state@1. The provider names no node and
++ * no project, its clock never sets Run's timings past what Run saw, and its wall_ms is billed only at the end
++ * state, clamped (meterOf). A running job shows no wall clock until then, as a Bot job does.
++ */
++ async function mirrorProvider(row, at) {
++ const worker = workerOf(row);
++ let p;
++ try { p = await worker.state(row.id); }
++ catch (e) {
++ stat.last_error = e.message;
++ if (!e.retryable) warnSlowly(row, at, `[Run] job ${row.id}: the browser provider's state was refused (${e.code || e.message})`);
++ return note(row.id, row.state, { reason: e.code || 'unavailable', ...(e.retryable ? {} : { refused: true }) });
++ }
++ if (!p) return unknownJob(row, at);
++ unknown.delete(row.id);
++ const seenAt = now();
++ const sentAt = sentAtOf(row);
++ const state = RUN_STATE_OF[p.state];
++ const moved = RANK[state] > RANK[row.state] ? state : null;
++ // The provider's start, never before Run sent the job nor after Run saw it; Run's own clock when it gives
++ // none. The first start Run records stays (store.mirror would overwrite it).
++ const startedMs = row.started_ms != null ? null
++ : p.started_ms == null ? (state === 'running' ? seenAt : null)
++ : Math.min(Math.max(p.started_ms, sentAt), seenAt);
++
++ if (moved && END_STATES.has(moved)) {
++ stdoutOf(row, p.stdout, at);
++ const m = meterOf(row, p.wall_ms, seenAt);
++ const settled = await db.tx(async (t) => {
++ await store.mirror(t, row.id, {
++ state: null, node_id: row.node_id, provider: null, region: null,
++ started_ms: startedMs, wall_ms: m.wallMs, usage_read: m.readings.length,
++ }, at);
++ const s = await store.settle(t, row.id, {
++ state: moved, exit: p.exit_reason ?? null, code: p.exit_code ?? null,
++ result: moved === 'succeeded' ? p.result ?? null : null, wallMs: m.wallMs,
++ errorDetail: m.clamped ? `browser-provider claimed ${m.claimedMs} ms; billed ${m.wallMs} ms (observed + ${METER_SLACK_MS} ms)` : null, at,
++ });
++ if (s.changed) {
++ await events.ofState(t, s.row);
++ if (m.readings.length) await usage.enqueue(t, m.readings, seenAt);
++ }
++ return s;
++ });
++ if (settled.changed) {
++ stat.mirrored++;
++ if (m.readings.length) { stat.metered++; usageRelay?.kick?.(); }
++ if (moved === 'failed') stat.failed++;
++ forget(row.id);
++ stream.state(row.id, moved, at);
++ }
++ return note(row.id, moved, { changed: settled.changed, exit: p.exit_reason ?? null, wall_ms: m.wallMs, clamped: m.clamped || undefined });
++ }
++
++ const updated = await db.tx(async (t) => {
++ const r = await store.mirror(t, row.id, {
++ state: moved, node_id: row.node_id, provider: null, region: null,
++ started_ms: moved === 'running' || row.state === 'running' ? startedMs : null, wall_ms: null, usage_read: null,
++ }, at);
++ if (moved === 'running' && r.state === 'running') await events.started(t, r);
++ return r;
++ });
++ if (updated.state !== row.state) {
++ stat.mirrored++;
++ stream.state(row.id, updated.state, at);
++ }
++ stdoutOf(row, p.stdout, at);
++ if (cancelling(row) && !cancelSent.has(row.id)) {
++ try {
++ const out = await worker.cancel(row.id);
++ cancelSent.add(row.id);
++ stat.cancels_sent++;
++ if (out === null) log.warn(`[Run] job ${row.id}: the browser provider has no such job to cancel`);
++ } catch (e) {
++ stat.last_error = e.message;
++ log.warn(`[Run] job ${row.id}: the cancel did not reach the browser provider (${e.code || e.message})`);
++ }
++ }
++ return note(row.id, updated.state, { moved });
++ }
++
/** The cancel route's own call (POST /jobs/:id/cancel): the same once-per-job job_cancel the tick sends. */
async function cancelNow(row) {
-- if (!bot.enabled || !row || !row.id) return { cancelled: false, reason: 'dispatch.disabled' };
++ const worker = row ? workerOf(row) : bot;
++ if (!worker.enabled || !row || !row.id) return { cancelled: false, reason: 'dispatch.disabled' };
if (cancelSent.has(row.id)) return { cancelled: true, reason: 'already_sent' };
try {
-- await bot.cancel(row.id);
++ await worker.cancel(row.id);
cancelSent.add(row.id);
stat.cancels_sent++;
return { cancelled: true, reason: 'sent' };
@@@ -372,7 +529,7 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
/** One poll: the sweep, then placement, then the mirror. Never two ticks at once. */
async function tick() {
-- if (!bot.enabled) return [];
++ if (!anyEnabled()) return [];
if (busy) return (await inflight.catch(() => null), []); // one tick at a time; the timer's next one runs later
busy = true;
let release;
@@@ -393,6 +550,16 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
// What this process remembers is only about the jobs in flight: a job that ended (or left both
// sets) is forgotten here, so the maps stay bounded by the work that is actually open.
const live = new Set([...waitingRows, ...activeRows].map((r) => r.id));
++ // A provider send Run could not confirm, for a job that has since ended without a placement (cancelled
++ // while queued, or its ttl ran out): the provider may hold it, so it is asked to stop.
++ for (const id of [...unconfirmed]) {
++ if (live.has(id)) continue;
++ const fresh = await store.get(db, id);
++ if (fresh && fresh.state === 'queued') continue;
++ unconfirmed.delete(id);
++ if (fresh && fresh.node_id) continue;
++ try { await (provider || disabled).cancel(id); } catch (e) { log.warn(`[Run] job ${id}: the browser provider could not be asked to stop it (${e.code || e.message})`); }
++ }
for (const id of [...cancelSent]) if (!stillCancelling.has(id)) cancelSent.delete(id);
for (const key of [...unknown.keys()]) if (!live.has(key)) unknown.delete(key);
for (const key of [...cursors.keys()]) if (!live.has(key)) cursors.delete(key);
@@@ -411,7 +578,7 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
}
function start() {
-- if (timer || !bot.enabled) return status();
++ if (timer || !anyEnabled()) return status();
// Nothing runs at module load: server/index.js starts this, and stop() clears it.
timer = setInterval(() => { tick().catch((e) => log.warn(`[Run] poller: ${e.message}`)); }, config.dispatch.pollMs);
timer.unref?.();
@@@ -422,7 +589,7 @@ function createPoller({ config, db, store, events, stream, bot, placement, now =
function status() {
return {
running: Boolean(timer), interval_ms: config.dispatch.pollMs, ...stat,
-- cancel_pending: cancelSent.size, unknown_jobs: unknown.size, stdout_cursors: cursors.size,
++ cancel_pending: cancelSent.size, unknown_jobs: unknown.size, stdout_cursors: cursors.size, unconfirmed_sends: unconfirmed.size,
};
}
ddiff --git a/server/jobs/usage.js b/server/jobs/usage.js
nnew file mode 100644
iindex 0000000..c9b75ac
---- /dev/null
++++ b/server/jobs/usage.js
@@@ -0,0 +1,136 @@
++'use strict';
++
++/**
++ * Run's own metering (plan T14 R4a, decision B): the usage of a job an external managed-browser provider ran.
++ * Bot meters every job it carries and Run never posts those; a provider is not a paired Node and writes no usage,
++ * so for its jobs — and only for them — Run queues the readings and posts them to OpenVibe.Billing's
++ * billing.usage.record (POST <RUN_BILLING_URL>/api/v1/usage, one reading per request), the pattern of OpenVibe.Bot's
++ * server/jobs/metering.js.
++ *
++ * The readings: one platform.usage-sample@1 per wall-clock second Run bills — the provider's claim clamped to Run's
++ * own observed duration plus 5 s (server/dispatch/browser-provider.js meter()) — written in the transaction that
++ * settles the job, so they exist if and only if the end state committed. id = idempotency_key = run:<job id>:<n>
++ * (the key platform.job@1 reserves; a provider job never reaches Bot, so the two never meter one job), service run,
++ * operation browser.invoke, unit s, quantity 1 (the partial last second (wall_ms mod 1000)/1000), at = Run's own
++ * placed_at + n*1000, node the offer the job was placed on, provider browser-provider, source
++ * openvibe-run.browser-provider, project and subject from Run's row — never from the provider. `resource` is left
++ * out: platform.usage-sample@1 makes it a rate-card metric, never a job id, and no card names one for this yet.
++ *
++ * The relay (openvibe-sdk createPgOutbox on run_usage_outbox, migrations/0003_usage_outbox.sql) runs while
++ * RUN_BILLING_URL and OV_OAUTH_CLIENT_SECRET are set: Run mints a Network service token for audience
++ * openvibe.billing holding billing.usage.record and uses it for nothing else. Unset: readings wait in the table,
++ * nothing is sent, and the jobs still run. Billing down, 401/403/404/429/5xx: retried with backoff across
++ * restarts, never dropped. Billing refusing the reading itself (400, 409, 413, 422): the row is kept with its
++ * error and never sent again, so nothing is billed twice. The token is never logged.
++ */
++const contracts = require('openvibe-contracts');
++const { createPgOutbox } = require('openvibe-sdk/events');
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++
++const TABLE = 'run_usage_outbox';
++const SCHEMA = 'platform.usage-sample@1';
++const SERVICE = 'run';
++const OPERATION = 'browser.invoke';
++const UNIT = 's';
++const SOURCE = 'openvibe-run.browser-provider';
++const BILLING_SCOPE = 'billing.usage.record';
++const REFUSED = new Set([400, 409, 413, 422]);
++
++/**
++ * The readings a billed wall clock stands for: seconds 0 … floor(wallMs/1000)-1 with quantity 1 and, when
++ * wallMs mod 1000 > 0, second floor(wallMs/1000) with quantity (wallMs mod 1000)/1000. Nothing for 0 ms.
++ */
++function readingsOf({ jobId, wallMs, startMs, node, provider, project, subject }) {
++ if (!Number.isInteger(wallMs) || wallMs < 0) throw new TypeError(`wall_ms must be an integer >= 0, not ${wallMs}`);
++ if (!Number.isInteger(startMs) || startMs < 0) throw new TypeError('the start must be an integer >= 0');
++ const one = (n, quantity) => {
++ const key = `run:${jobId}:${n}`;
++ const r = {
++ id: key, idempotency_key: key, service: SERVICE, operation: OPERATION, unit: UNIT, quantity,
++ at: new Date(startMs + n * 1000).toISOString(), source: SOURCE,
++ };
++ if (node) r.node = node;
++ if (provider) r.provider = provider;
++ if (project) r.project = project;
++ if (subject) r.subject = subject;
++ return r;
++ };
++ const full = Math.floor(wallMs / 1000);
++ const out = [];
++ for (let n = 0; n < full; n++) out.push(one(n, 1));
++ if (wallMs % 1000 > 0) out.push(one(full, (wallMs % 1000) / 1000));
++ return out;
++}
++
++/** Queue readings inside the caller's transaction; a key already queued changes nothing. → how many were new. */
++async function enqueue(t, readings, at) {
++ let added = 0;
++ for (const reading of readings) {
++ contracts.assertValid(SCHEMA, reading);
++ const row = await t.maybe(`INSERT INTO ${TABLE} (event_id, envelope, created_at) VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING RETURNING id`,
++ [reading.idempotency_key, JSON.stringify(reading), at]);
++ if (row) added++;
++ }
++ return added;
++}
++
++// createPgOutbox's `events` seam pointed at Billing. isPermanent() reads err.status: 422 rejects the row, 503 retries.
++function billingSink({ url, authorization, onUnauthorized, fetchImpl, timeoutMs }) {
++ return {
++ prepare: (reading) => reading,
++ async publish(reading) {
++ const res = await fetchImpl(`${url}/api/v1/usage`, {
++ method: 'POST', redirect: 'error',
++ headers: { Authorization: await authorization(), 'Content-Type': 'application/json', Accept: 'application/json' },
++ body: JSON.stringify(reading), signal: AbortSignal.timeout(timeoutMs),
++ });
++ await res.body?.cancel?.().catch(() => {});
++ // A minted token Billing no longer accepts (rotated key, clock): drop it so the next sweep mints afresh.
++ if (res.status === 401 && onUnauthorized) onUnauthorized();
++ // 201 written, 200 an identical reading replayed: either way Billing holds it.
++ if (!res.ok) { const err = new Error(`billing.usage.record answered ${res.status}`); err.status = REFUSED.has(res.status) ? 422 : 503; throw err; }
++ return { event_id: reading.id, seq: null };
++ },
++ };
++}
++
++/**
++ * createUsageRelay({ db, config, fetchImpl, now, log }) → { enabled, note, start, stop, kick, flush, pending,
++ * rejected, status }. Off (every call a no-op) without RUN_BILLING_URL or OV_OAUTH_CLIENT_SECRET.
++ */
++function createUsageRelay({ db, config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console }) {
++ const { url, audience, intervalMs, timeoutMs } = config.billing;
++ const { clientId, clientSecret } = config.oauth;
++ const count = (where) => (db ? db.value(`SELECT count(*)::int FROM ${TABLE} WHERE ${where}`).then(Number) : Promise.resolve(0));
++ const pending = () => count('sent_at IS NULL AND rejected_at IS NULL');
++ const rejected = () => count('rejected_at IS NOT NULL');
++ const enabled = Boolean(db && url && clientSecret);
++ const note = enabled
++ ? `the usage relay to Billing: ${url}, audience ${audience}, capability ${BILLING_SCOPE} (external-provider jobs only)`
++ : !url ? 'the usage relay is off: RUN_BILLING_URL is unset (external-provider readings wait in run_usage_outbox)'
++ : 'the usage relay is off: OV_OAUTH_CLIENT_SECRET is unset, so Run cannot mint its Billing token (readings wait)';
++ if (!enabled) {
++ const idle = async () => ({ sent: 0, failed: 0, rejected: 0 });
++ return { enabled, note, start() {}, stop: async () => {}, kick() {}, flush: idle, pending, rejected, status: () => ({ enabled, note }) };
++ }
++ const tokens = createServiceTokenClient({
++ tokenUrl: `${config.network.internalUrl}/oauth/token`, clientId, clientSecret, audience, scope: BILLING_SCOPE, fetch: fetchImpl, now,
++ });
++ const authorization = async () => {
++ try { return (await tokens.authHeaders()).Authorization; }
++ catch (e) { const err = new Error('Run could not mint its Billing token from Network'); err.status = 503; throw err; }
++ };
++ let lastError = null;
++ const relay = createPgOutbox(db, {
++ events: billingSink({ url, authorization, onUnauthorized: () => tokens.invalidate(), fetchImpl, timeoutMs }),
++ table: TABLE, batchSize: 1, intervalMs, now,
++ onError: (err) => { const m = err && err.message; if (m !== lastError) log.warn(`[Run] usage reading not delivered: ${m}`); lastError = m; },
++ });
++ return {
++ enabled, note, pending, rejected,
++ start: relay.start, stop: relay.stop, kick: relay.kick, flush: relay.flush,
++ status: () => ({ enabled, note, billing_url: url, audience, capability: BILLING_SCOPE, interval_ms: intervalMs, last_error: lastError }),
++ };
++}
++
++module.exports = { readingsOf, enqueue, createUsageRelay, TABLE, SERVICE, OPERATION, UNIT, SOURCE };
ddiff --git a/test/dispatch-browser-provider.test.js b/test/dispatch-browser-provider.test.js
nnew file mode 100644
iindex 0000000..6bebae1
---- /dev/null
++++ b/test/dispatch-browser-provider.test.js
@@@ -0,0 +1,417 @@
++'use strict';
++/**
++ * The managed-browser provider adapter (plan T14 R4a, server/dispatch/browser-provider.js): a class browser job
++ * placed on a kind=provider offer named `browser-provider` is sent to the provider's own API instead of Bot,
++ * mirrored from its validated browser-provider.job-state@1, cancelled and expired there — and metered by Run
++ * itself, the provider's claimed wall time clamped to Run's observed duration + 5 s, its readings relayed to
++ * Billing with Run's own token (audience openvibe.billing, billing.usage.record).
++ *
++ * node test/dispatch-browser-provider.test.js
++ *
++ * The fake provider and the fake Billing are real HTTP servers, so the key header, the Idempotency-Key, the
++ * timeouts and the 404/5xx answers travel the wire. The job loop is driven by hand (RUN_JOBS=off).
++ */
++const assert = require('assert');
++const crypto = require('crypto');
++const http = require('http');
++const contracts = require('openvibe-contracts');
++const { serviceAuth, capabilities, ids } = require('openvibe-contracts');
++const { boot, check, done } = require('./helpers/app');
++
++const KEY = `bpk_${crypto.randomBytes(16).toString('hex')}`;
++const PROJECT = () => ids.newId('project');
++const OFFER_ID = 'off_browser_01';
++const LIMITS = { wall_ms: 60_000, cpu_ms: 60_000, mem_bytes: 1_073_741_824 };
++
++/** A contract-valid platform.resource-offer@1 of the managed-browser provider. */
++function providerOffer(over = {}) {
++ return {
++ offer_id: OFFER_ID,
++ kind: 'provider',
++ provider: 'browser-provider',
++ region: 'eu-west',
++ trust: 'first-party',
++ capabilities: ['worker:browser'],
++ health: { status: 'up', checked_at: new Date().toISOString() },
++ pricing: { model: 'prepaid' },
++ updated_at: new Date().toISOString(),
++ ...over,
++ };
++}
++
++const body = (req) => new Promise((resolve) => {
++ let raw = '';
++ req.on('data', (c) => { raw += c; });
++ req.on('end', () => { try { resolve(raw ? JSON.parse(raw) : null); } catch { resolve(null); } });
++});
++const listen = (server) => new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve(`http://127.0.0.1:${server.address().port}`)));
++const closer = (server) => () => new Promise((r) => { server.closeAllConnections(); server.close(r); });
++
++/**
++ * A fake managed-browser provider under /v1: POST /jobs (idempotent by job id), GET /jobs/:id and POST
++ * /jobs/:id/cancel, every call checked for `Authorization: Bearer <KEY>`. Every request it receives is
++ * validated against browser-provider.job-create-request@1, and every answer it gives is a valid
++ * job-state@1 unless a test asks for a broken one (mode.broken).
++ */
++async function startFakeProvider() {
++ const jobs = new Map();
++ const calls = [];
++ const mode = { fail5xx: new Map(), missing: new Set(), broken: new Map() }; // fail5xx: job id → answers left to fail
++ const stateOf = (row) => ({
++ id: row.id, state: row.state, started_ms: row.started_ms, finished_ms: row.finished_ms, wall_ms: row.wall_ms,
++ exit_reason: row.exit_reason, exit_code: row.exit_code, result: row.result, stdout: null,
++ });
++ const server = http.createServer(async (req, res) => {
++ const answer = (status, doc) => { res.writeHead(status, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(doc)); };
++ const url = new URL(req.url, 'http://provider.test');
++ const parts = url.pathname.split('/').filter(Boolean); // v1 jobs [id] [cancel]
++ const doc = req.method === 'POST' ? await body(req) : null;
++ calls.push({ method: req.method, path: url.pathname, headers: req.headers, body: doc });
++ if (req.headers.authorization !== `Bearer ${KEY}`) return answer(401, { code: 'provider.unauthorized', detail: 'bad key' });
++ // A provider that echoes the key back in its error text: Run must never log it.
++ if (parts[0] !== 'v1' || parts[1] !== 'jobs') return answer(404, { code: 'provider.not_found' });
++ const target = parts[2] || (doc && doc.id);
++ if (mode.fail5xx.get(target) > 0) {
++ mode.fail5xx.set(target, mode.fail5xx.get(target) - 1);
++ return answer(503, { code: 'provider.down', detail: `overloaded (key ${KEY})` });
++ }
++ if (req.method === 'POST' && parts.length === 2) {
++ if (!contracts.validate('browser-provider.job-create-request@1', doc).valid) return answer(422, { code: 'provider.invalid', detail: 'not a job-create-request@1' });
++ if (req.headers['idempotency-key'] !== doc.id) return answer(400, { code: 'provider.idempotency', detail: 'Idempotency-Key must be the job id' });
++ const known = jobs.get(doc.id);
++ if (known) return answer(200, { job: stateOf(known), created: false });
++ const row = { id: doc.id, request: doc, state: 'queued', started_ms: null, finished_ms: null, wall_ms: 0, exit_reason: null, exit_code: null, result: null };
++ jobs.set(row.id, row);
++ return answer(201, { job: stateOf(row), created: true });
++ }
++ const row = jobs.get(parts[2]);
++ if (!row || mode.missing.has(parts[2])) return answer(404, { code: 'provider.job_not_found' });
++ if (mode.broken.has(row.id)) return answer(200, mode.broken.get(row.id)(stateOf(row)));
++ if (req.method === 'GET' && parts.length === 3) return answer(200, stateOf(row));
++ if (req.method === 'POST' && parts[3] === 'cancel') {
++ if (!row.exit_reason) Object.assign(row, { state: 'cancelled', exit_reason: 'cancelled', exit_code: null, finished_ms: Date.now() });
++ return answer(200, stateOf(row));
++ }
++ return answer(404, { code: 'provider.not_found' });
++ });
++ const url = await listen(server);
++ return {
++ url, jobs, calls, mode,
++ callsOf: (method, path) => calls.filter((c) => c.method === method && c.path === path),
++ start(id, startedMs = Date.now()) { Object.assign(jobs.get(id), { state: 'running', started_ms: startedMs }); },
++ finish(id, { state = 'succeeded', exit = 'exited', code = 0, wallMs, result = null }) {
++ Object.assign(jobs.get(id), { state, exit_reason: exit, exit_code: code, wall_ms: wallMs, result, finished_ms: Date.now() });
++ },
++ close: closer(server),
++ };
++}
++
++/** A fake OpenVibe.Billing: POST /api/v1/usage, verifying Run's own token (openvibe.billing, billing.usage.record). */
++async function startFakeBilling() {
++ const keys = { publicPem: null, issuer: null };
++ const readings = [];
++ const server = http.createServer(async (req, res) => {
++ const answer = (status, doc) => { res.writeHead(status, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(doc)); };
++ const auth = String(req.headers.authorization || '');
++ const verified = serviceAuth.verifyServiceToken(auth.slice(7).trim(), { publicKey: keys.publicPem, issuer: keys.issuer, audience: 'openvibe.billing' });
++ if (!verified.ok) return answer(401, { code: verified.code });
++ if (!capabilities.grants(verified.claims.cap, 'billing.usage.record')) return answer(403, { code: 'capability.denied' });
++ if (req.method !== 'POST' || req.url !== '/api/v1/usage') return answer(404, {});
++ const doc = await body(req);
++ readings.push({ doc, claims: verified.claims });
++ return answer(201, { ok: true });
++ });
++ const url = await listen(server);
++ return {
++ url, readings,
++ useNetwork(network) { keys.publicPem = network.publicPem; keys.issuer = network.issuer; },
++ close: closer(server),
++ };
++}
++
++async function main() {
++ const provider = await startFakeProvider();
++ const billing = await startFakeBilling();
++ const providerEnv = { RUN_BROWSER_PROVIDER_URL: `${provider.url}/v1/`, RUN_BROWSER_PROVIDER_API_KEY: KEY, RUN_BROWSER_PROVIDER_TIMEOUT_MS: '2000' };
++ const t = await boot({
++ env: {
++ OV_OAUTH_CLIENT_SECRET: crypto.randomBytes(12).toString('hex'),
++ RUN_BOT_URL: 'http://127.0.0.1:9', // no node offer is published: nothing goes to Bot
++ RUN_BILLING_URL: billing.url,
++ RUN_DISPATCH_UNKNOWN_POLLS: '2',
++ ...providerEnv,
++ },
++ });
++ billing.useNetwork(t.network);
++ const submit = (over = {}, opts = {}) => t.submit({ class: 'browser', artifact: undefined, args: { url: 'https://example.com' }, limits: LIMITS, ...over },
++ { sub: 'svc:builder', ...opts, project: opts.project || PROJECT() });
++ const read = (id, project) => t.call('GET', `/api/v1/jobs/${id}`, { cap: ['run.job.read'], project, sub: 'svc:builder' });
++ const projectOf = (id) => t.db.value('SELECT project_id FROM run_jobs WHERE id = $1', [id]);
++ const readingsOf = (id) => t.db.many('SELECT event_id, envelope, sent_at FROM run_usage_outbox WHERE event_id LIKE $1 ORDER BY id', [`run:${id}:%`]);
++ const pollOne = async (id) => (await t.dispatch.poll()).find((a) => a.job_id === id);
++ const adapter = t.dispatch.deps.provider;
++ t.setOffers([providerOffer()]);
++
++ try {
++ await check('config: the provider and the Billing relay come from Run\'s env; the timeout has a default and a floor', async () => {
++ const { loadConfig } = require('../server/config');
++ const off = loadConfig({});
++ assert.deepStrictEqual(off.browserProvider, { url: '', apiKey: '', timeoutMs: 8000 });
++ assert.strictEqual(off.billing.url, '');
++ const on = loadConfig({ RUN_BROWSER_PROVIDER_URL: ' https://browser.example/v1/ ', RUN_BROWSER_PROVIDER_API_KEY: 'k', RUN_BROWSER_PROVIDER_TIMEOUT_MS: '5' });
++ assert.deepStrictEqual(on.browserProvider, { url: 'https://browser.example/v1', apiKey: 'k', timeoutMs: 100 });
++ });
++
++ await check('the adapter is on: status names the provider and never its key', async () => {
++ const health = await t.call('GET', '/api/health', { token: null });
++ const bp = health.json.dispatcher.browser_provider;
++ assert.strictEqual(bp.enabled, true, health.text);
++ assert.strictEqual(bp.key_set, true);
++ assert.strictEqual(bp.url, provider.url); // the origin only
++ assert.strictEqual(health.json.dispatcher.usage_relay.enabled, true);
++ assert.ok(!health.text.includes(KEY));
++ assert.ok(!JSON.stringify(t.dispatch.status()).includes(KEY));
++ });
++
++ await check('a browser job is placed on the provider offer and sent once, as the frame alone', async () => {
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ const placed = await pollOne(job.id);
++ assert.strictEqual(placed.state, 'placed', JSON.stringify(placed));
++ const posts = provider.callsOf('POST', '/v1/jobs').filter((c) => c.body && c.body.id === job.id);
++ assert.strictEqual(posts.length, 1);
++ const call = posts[0];
++ assert.strictEqual(call.headers.authorization, `Bearer ${KEY}`);
++ assert.strictEqual(call.headers['idempotency-key'], job.id);
++ // No project, no subject, no requester, no credential: the provider never learns who pays.
++ assert.deepStrictEqual(call.body, { id: job.id, class: 'browser', args: { url: 'https://example.com' }, inputs: [], limits: LIMITS, egress: 'none' });
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.state, 'placed');
++ assert.deepStrictEqual(read1.json.placement, { node: OFFER_ID, provider: 'browser-provider', region: 'eu-west' });
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ await t.dispatch.poll();
++ await t.dispatch.poll();
++ assert.strictEqual(provider.callsOf('POST', '/v1/jobs').filter((c) => c.body && c.body.id === job.id).length, 1, 'a placed job is never sent again');
++ });
++
++ await check('running → succeeded mirrors once; the claim within the cap is billed and relayed to Billing', async () => {
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ await t.dispatch.poll();
++ provider.start(job.id);
++ assert.strictEqual((await pollOne(job.id)).state, 'running');
++ assert.strictEqual((await t.outboxRows('run.job.started', job.id)).length, 1);
++ t.clock.offset += 4000; // Run observes 4 s pass
++ provider.finish(job.id, { wallMs: 2500, result: { title: 'Example Domain' } });
++ assert.strictEqual((await pollOne(job.id)).state, 'succeeded');
++ const read1 = await read(job.id, project);
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ assert.deepStrictEqual(read1.json.exit, { reason: 'exited', code: 0 });
++ assert.deepStrictEqual(read1.json.result, { title: 'Example Domain' });
++ assert.strictEqual(read1.json.usage.wall_ms, 2500);
++ assert.strictEqual(read1.json.usage.seconds, 2.5);
++ assert.strictEqual((await t.outboxRows('run.job.succeeded', job.id)).length, 1);
++ // One reading per billed second, the partial last one as a fraction; project and subject are Run's own.
++ const rows = await readingsOf(job.id);
++ assert.deepStrictEqual(rows.map((r) => r.event_id), [0, 1, 2].map((n) => `run:${job.id}:${n}`));
++ const docs = rows.map((r) => (typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope));
++ assert.deepStrictEqual(docs.map((d) => d.quantity), [1, 1, 0.5]);
++ for (const d of docs) {
++ assert.ok(contracts.validate('platform.usage-sample@1', d).valid, JSON.stringify(d));
++ assert.strictEqual(d.project, project);
++ assert.strictEqual(d.subject, 'service:builder');
++ assert.strictEqual(d.service, 'run');
++ assert.strictEqual(d.provider, 'browser-provider');
++ }
++ // Later polls bill nothing twice.
++ await t.dispatch.poll();
++ assert.strictEqual((await readingsOf(job.id)).length, 3);
++ // The relay posts them with Run's own Billing token, used for nothing else.
++ await t.dispatch.deps.usageRelay.flush();
++ const got = billing.readings.filter((r) => r.doc.id.startsWith(`run:${job.id}:`));
++ assert.strictEqual(got.length, 3);
++ assert.deepStrictEqual(got[0].claims.aud, ['openvibe.billing']);
++ assert.deepStrictEqual(got[0].claims.cap, ['billing.usage.record']);
++ assert.ok((await readingsOf(job.id)).every((r) => r.sent_at != null));
++ assert.ok(t.network.tokenRequests.audiences.includes('openvibe.billing'));
++ });
++
++ await check('an over-claimed meter is clamped to Run\'s observed duration + 5 s and logged', async () => {
++ const direct = adapter.meter({ jobId: 'job_x', claimedMs: 20_000, observedMs: 1000 });
++ assert.deepStrictEqual(direct, { wallMs: 6000, claimedMs: 20_000, capMs: 6000, clamped: true });
++ assert.strictEqual(adapter.meter({ claimedMs: 3000, observedMs: 1000 }).wallMs, 3000);
++ assert.strictEqual(adapter.meter({ claimedMs: -5, observedMs: 1000 }).wallMs, 0);
++
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ await t.dispatch.poll();
++ provider.start(job.id);
++ await t.dispatch.poll();
++ t.clock.offset += 2000;
++ provider.finish(job.id, { wallMs: 60_000 });
++ const ended = await pollOne(job.id);
++ assert.strictEqual(ended.state, 'succeeded');
++ assert.strictEqual(ended.clamped, true);
++ const wall = (await read(job.id, project)).json.usage.wall_ms;
++ assert.ok(wall >= 7000 && wall < 9000, `billed ${wall} ms, not observed (~2 s) + 5 s`);
++ const docs = (await readingsOf(job.id)).map((r) => (typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope));
++ assert.strictEqual(docs.length, Math.ceil(wall / 1000));
++ assert.strictEqual(Math.round(docs.reduce((s, d) => s + d.quantity, 0) * 1000), wall);
++ assert.ok(t.logs.some((l) => l.includes(job.id) && /claims 60000 ms/.test(l) && /clamped/.test(l)), t.logs.join('\n'));
++ assert.ok(adapter.status().clamped >= 2);
++ });
++
++ await check('cancel: a placed provider job is stopped at the provider and ends cancelled', async () => {
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ await t.dispatch.poll();
++ provider.start(job.id);
++ await t.dispatch.poll();
++ const asked = await t.call('POST', `/api/v1/jobs/${job.id}/cancel`, { cap: ['run.job.cancel'], project, sub: 'svc:builder' });
++ assert.strictEqual(asked.status, 200, asked.text);
++ assert.strictEqual(provider.callsOf('POST', `/v1/jobs/${job.id}/cancel`).length, 1);
++ assert.strictEqual((await pollOne(job.id)).state, 'cancelled');
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.state, 'cancelled');
++ assert.deepStrictEqual(read1.json.exit, { reason: 'cancelled', code: null });
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ assert.strictEqual((await t.outboxRows('run.job.cancelled', job.id)).length, 1);
++ await t.dispatch.poll();
++ assert.strictEqual(provider.callsOf('POST', `/v1/jobs/${job.id}/cancel`).length, 1, 'one cancel per job');
++ });
++
++ await check('expire: a provider job whose ttl runs out is stopped at the provider and ends expired', async () => {
++ const job = await submit();
++ await t.dispatch.poll();
++ provider.start(job.id);
++ await t.dispatch.poll();
++ await t.db.exec('UPDATE run_jobs SET created_at = created_at - 10000000 WHERE id = $1', [job.id]);
++ const swept = (await t.dispatch.sweep()).find((a) => a.job_id === job.id);
++ assert.strictEqual(swept.state, 'expired');
++ assert.strictEqual(provider.callsOf('POST', `/v1/jobs/${job.id}/cancel`).length, 1);
++ const read1 = await read(job.id, await projectOf(job.id));
++ assert.strictEqual(read1.json.state, 'expired');
++ assert.strictEqual(read1.json.error.code, 'run.job.ttl');
++ assert.strictEqual(read1.json.placement.node, OFFER_ID);
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ });
++
++ await check('a provider that does not know the job (404) fails it with run.job.worker_failed', async () => {
++ const job = await submit();
++ await t.dispatch.poll();
++ provider.mode.missing.add(job.id);
++ assert.strictEqual((await pollOne(job.id)).state, 'placed'); // one 404 is not yet the end
++ assert.strictEqual((await pollOne(job.id)).state, 'failed');
++ const read1 = await read(job.id, await projectOf(job.id));
++ assert.strictEqual(read1.json.error.code, 'run.job.worker_failed');
++ assert.match(read1.json.error.detail, /browser provider has no job/);
++ assert.strictEqual((await readingsOf(job.id)).length, 0);
++ });
++
++ await check('a 5xx is retried on the next tick, for the send and for the state', async () => {
++ const job = await submit();
++ provider.mode.fail5xx.set(job.id, 1);
++ const first = await pollOne(job.id);
++ assert.strictEqual(first.state, 'queued');
++ assert.strictEqual(first.reason, 'run.dispatch.unavailable');
++ assert.strictEqual((await pollOne(job.id)).state, 'placed');
++ provider.start(job.id);
++ provider.mode.fail5xx.set(job.id, 1);
++ assert.strictEqual((await pollOne(job.id)).state, 'placed'); // the state answer was a 503
++ assert.strictEqual((await pollOne(job.id)).state, 'running');
++ });
++
++ await check('a malformed or foreign state is never adopted', async () => {
++ const job = await submit();
++ await t.dispatch.poll();
++ provider.start(job.id);
++ await t.dispatch.poll();
++ provider.finish(job.id, { wallMs: 1000 });
++ for (const broken of [
++ (s) => ({ ...s, wall_ms: 'lots' }), // not a job-state@1
++ (s) => ({ ...s, project_id: PROJECT() }), // a field the contract has no place for
++ (s) => ({ ...s, id: ids.newId('job') }), // another job's state
++ ]) {
++ provider.mode.broken.set(job.id, broken);
++ const answer = await pollOne(job.id);
++ assert.strictEqual(answer.state, 'running', JSON.stringify(answer));
++ assert.strictEqual(answer.refused, true);
++ }
++ assert.strictEqual((await readingsOf(job.id)).length, 0);
++ provider.mode.broken.delete(job.id);
++ assert.strictEqual((await pollOne(job.id)).state, 'succeeded');
++ assert.ok(adapter.status().malformed >= 3);
++ });
++
++ await check('the provider\'s key is never in a log, a status or a read result', async () => {
++ assert.ok(t.logs.some((l) => /browser provider/.test(l)), 'the bridge logged its provider calls');
++ assert.ok(!t.logs.some((l) => l.includes(KEY)), t.logs.join('\n'));
++ assert.ok(!JSON.stringify(t.dispatch.status()).includes(KEY));
++ const rows = await t.db.many('SELECT job::text AS job FROM run_jobs');
++ assert.ok(!rows.some((r) => r.job.includes(KEY)));
++ });
++ } finally {
++ t.dispatch.stop();
++ await t.close();
++ }
++
++ // Without RUN_BILLING_URL the provider's jobs still run; their readings wait and nothing is sent.
++ const quiet = await boot({ env: { OV_OAUTH_CLIENT_SECRET: crypto.randomBytes(12).toString('hex'), RUN_BOT_URL: 'http://127.0.0.1:9', ...providerEnv } });
++ try {
++ await check('without RUN_BILLING_URL a provider job still runs and its readings wait unsent', async () => {
++ assert.strictEqual(quiet.dispatch.status().usage_relay.enabled, false);
++ quiet.setOffers([providerOffer()]);
++ const job = await quiet.submit({ class: 'browser', artifact: undefined, args: {}, limits: LIMITS });
++ await quiet.dispatch.poll();
++ provider.start(job.id);
++ await quiet.dispatch.poll();
++ quiet.clock.offset += 3000;
++ provider.finish(job.id, { wallMs: 1500 });
++ assert.strictEqual((await quiet.dispatch.poll()).find((a) => a.job_id === job.id).state, 'succeeded');
++ const rows = await quiet.db.many('SELECT sent_at FROM run_usage_outbox WHERE event_id LIKE $1', [`run:${job.id}:%`]);
++ assert.strictEqual(rows.length, 2);
++ assert.ok(rows.every((r) => r.sent_at == null));
++ assert.ok(!billing.readings.some((r) => r.doc.id.startsWith(`run:${job.id}:`)));
++ });
++ } finally {
++ await quiet.close();
++ }
++
++ // Without the provider's URL (or its key) the adapter is off: a browser job stays queued, nothing is sent.
++ for (const [name, env] of [['RUN_BROWSER_PROVIDER_URL', { RUN_BROWSER_PROVIDER_API_KEY: KEY }], ['RUN_BROWSER_PROVIDER_API_KEY', { RUN_BROWSER_PROVIDER_URL: `${provider.url}/v1` }]]) {
++ const off = await boot({ env: { OV_OAUTH_CLIENT_SECRET: crypto.randomBytes(12).toString('hex'), RUN_BOT_URL: 'http://127.0.0.1:9', ...env } });
++ try {
++ await check(`without ${name} the adapter is off and a browser job stays queued`, async () => {
++ const status = off.dispatch.status().browser_provider;
++ assert.strictEqual(status.enabled, false);
++ assert.match(status.note, new RegExp(`${name} is unset`));
++ // A node offer cannot borrow the provider's name to reach Bot either.
++ off.setOffers([providerOffer(), providerOffer({ offer_id: 'off_node_x', kind: 'node', node_id: 'dev_01J8Z4M2Q0R7T9YV3K6N8P1W2X' })]);
++ const before = provider.calls.length;
++ const job = await off.submit({ class: 'browser', artifact: undefined, args: {}, limits: LIMITS });
++ assert.strictEqual((await off.dispatch.poll()).find((a) => a.job_id === job.id).state, 'queued');
++ assert.strictEqual((await off.dispatch.poll()).find((a) => a.job_id === job.id).state, 'queued');
++ assert.strictEqual(provider.calls.length, before);
++ assert.strictEqual(await off.db.value('SELECT state FROM run_jobs WHERE id = $1', [job.id]), 'queued');
++ });
++ } finally {
++ await off.close();
++ }
++ }
++
++ await check('outside development the provider URL must be https (or loopback): the key never travels in clear', async () => {
++ const { createBrowserProviderAdapter } = require('../server/dispatch/browser-provider');
++ const plain = createBrowserProviderAdapter({ config: { isProduction: true, browserProvider: { url: 'http://browser.example/v1', apiKey: KEY, timeoutMs: 1000 } }, log: { warn() {} } });
++ assert.strictEqual(plain.enabled, false);
++ assert.match(plain.note, /must be https/);
++ assert.ok(!plain.note.includes(KEY));
++ const secure = createBrowserProviderAdapter({ config: { isProduction: true, browserProvider: { url: 'https://browser.example/v1', apiKey: KEY, timeoutMs: 1000 } }, log: { warn() {} } });
++ assert.strictEqual(secure.enabled, true);
++ });
++
++ await provider.close();
++ await billing.close();
++ done();
++}
++
++main().catch((e) => { console.error(e); process.exit(1); });