openvibers/openvibe.bot Commits

Dispatch jobs to paired Nodes and meter each second for Billing

OP OpenVibers committed 4 Oct 2026 · 12de4f6644 · 16 files, +922 −20

Asked: T14 step L1 — OpenVibe.Bot dispatches jobs to a paired Node over the device link and meters them (money: security-critical). ONE pull request from current origin/main (git fetch first; read with `git show origin/main:<path>`, the local checkout may be stale). The attached T14 dossier's section "T14…
Harness job bot-t14-step-l1-openvibe-bot-125030 (plan T14).

Changes

ddiff --git a/.env.example b/.env.example
iindex f526f52..f6c3114 100644
---- a/.env.example
++++ b/.env.example
@@@ -77,3 +77,12 @@ OV_OAUTH_CLIENT_SECRET=
#BOT_OPENRE_URL=
#BOT_OPENRE_TOKEN=
#BOT_OPENRE_TIMEOUT_MS=8000
++
++# ── OpenVibe.Billing: job usage (plan T14 L1) ───────────────
++# Where the per-second readings of jobs run on paired Nodes go: billing.usage.record, POST <url>/api/v1/usage.
++# The token is a Network service token for audience openvibe.billing holding billing.usage.record. Either
++# unset: readings wait in run_usage_outbox (never dropped) and a later relay with both set delivers each once.
++#BOT_BILLING_URL=http://127.0.0.1:4600
++#BOT_BILLING_TOKEN=
++#BOT_BILLING_INTERVAL_MS=2000
++#BOT_BILLING_TIMEOUT_MS=5000
ddiff --git a/README.md b/README.md
iindex 60e08cc..83a9ae9 100644
---- a/README.md
++++ b/README.md
@@@ -13,6 +13,7 @@ OpenVibe.Bot pairs a robot's machines, keeps their state and gates every operato
- Pairing: one-time codes with a short expiry, minted by Bot or, with `BOT_PAIRING_AUTHORITY=network`, by OpenVibe.Network.
- The `/device` and `/control` WebSockets, the command gate and the latched e-stop.
- The command audit (`command_audit`, kept 30 days) and the `bot.*` outbox events.
++- Dispatching `platform.job@1` jobs to a paired Node over `/device` (`server/jobs/dispatch.js`, no HTTP route) and metering them per wall-clock second (`run_jobs`, `run_usage_outbox`; docs/protocol.md §1.2).
## Does not own
@@@ -27,7 +28,8 @@ OpenVibe.Bot pairs a robot's machines, keeps their state and gates every operato
- Valkey (`VALKEY_URL`, optional): the shared per-actor limit counters; unset, limits are counted per process.
- OpenVibe.Network: user and service tokens, node tokens, and pairing when the authority is `network`.
- OpenRe.Stream: the WHIP ingest base (`BOT_WHIP_BASE`) devices publish to.
--- `openvibe-contracts` v0.85.0, `openvibe-sdk` v0.26.0 and `openvibe-shared` v2.5.0 (package.json).
++- OpenVibe.Billing: job usage readings go to `billing.usage.record` (`BOT_BILLING_URL`, `BOT_BILLING_TOKEN`); unset, they wait in `run_usage_outbox`.
++- `openvibe-contracts` v0.92.0, `openvibe-sdk` v0.26.0 and `openvibe-shared` v2.5.0 (package.json).
## Capabilities
ddiff --git a/STATUS.json b/STATUS.json
iindex ed3ed65..2103161 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -5,9 +5,9 @@
"runtime": "node 22, express 4, ws; PostgreSQL 18 via openvibe-sdk/db (pg, or embedded PGlite in tests); Valkey optional; no build step",
"port": 4630,
"domain": "openvibe.bot (www -> apex; /device and /control WebSockets; /install)",
-- "contracts": "openvibe-contracts v0.85.0",
++ "contracts": "openvibe-contracts v0.92.0",
"packages": [
-- "openvibe-contracts v0.85.0",
++ "openvibe-contracts v0.92.0",
"openvibe-sdk v0.26.0",
"openvibe-shared v2.5.0"
],
@@@ -35,7 +35,8 @@
"a latched e-stop only the owner clears, with halt always allowed, and a device-side deadman deadline on the motion kinds",
"command audit kept 30 days, and the bot.robot.online/offline, bot.estop.set/cleared and bot.command.refused outbox events",
"public robot-profile catalogue (GET /profiles, /profiles/:id); robots are private to their owner, with public control only as a timed turn queue",
-- "per-actor limits on capability-boundary writes (openvibe-sdk/limits; shared through Valkey when VALKEY_URL is set)"
++ "per-actor limits on capability-boundary writes (openvibe-sdk/limits; shared through Valkey when VALKEY_URL is set)",
++ "platform.job@1 jobs dispatched to a paired Node over /device (platform.job-frame@1: resent unacked on reconnect, only to an advertised runtime class) and metered per second into run_usage_outbox for billing.usage.record"
],
-- "updated": "2026-10-03"
++ "updated": "2026-10-04"
}
ddiff --git a/docs/cutover-agent-bot-t14-step-l1-openvibe-bot-125030-muttjrzf.md b/docs/cutover-agent-bot-t14-step-l1-openvibe-bot-125030-muttjrzf.md
nnew file mode 100644
iindex 0000000..371af39
---- /dev/null
++++ b/docs/cutover-agent-bot-t14-step-l1-openvibe-bot-125030-muttjrzf.md
@@@ -0,0 +1,106 @@
++# OpenVibe.Bot — cutover runbook for plan T14 step L1 (jobs on the device link)
++
++The harness marks OpenVibe.Bot **risk data** (the repository owns `migrations/`), so a PR that merges into it
++must carry a cutover manifest and a green rehearsal. This PR does touch the schema: it adds one migration. This
++is its runbook: what runs in which order, what is backed up, how the result is verified, and the way back.
++
++## Scope — what this PR actually changes
++
++Bot becomes the dispatcher of plan T14: it hands `platform.job@1` jobs to a paired OpenVibe.Node over the
++`/device` socket (`platform.job-frame@1`) and meters them per wall-clock second for OpenVibe.Billing.
++
++- `migrations/0004_run_jobs.sql` (`-- phase: expand`) — two **new** tables and their indexes:
++ - `run_jobs` — one row per job id: the job body, the device (`node_id` → `devices.id`), its state and how far
++ it is metered (`usage_read`). `project_id` is plain text with no FK (Bot holds no projects).
++ - `run_usage_outbox` — `platform.usage-sample@1` readings waiting for Billing, keyed by
++ `event_id` = `run:<job id>:<second>` (UNIQUE), in the openvibe-sdk outbox shape.
++- `server/jobs/{dispatch,store,metering}.js` and the device switch in `server/realtime.js` — the seven job
++ frames; no HTTP route and no new capability (the Run service calls `dispatch` later).
++- `server/config.js`, `.env.example` — `BOT_BILLING_URL`, `BOT_BILLING_TOKEN`, `BOT_BILLING_INTERVAL_MS`,
++ `BOT_BILLING_TIMEOUT_MS`. Unset, the usage relay is off and readings wait in `run_usage_outbox`.
++- `openvibe-contracts` pin v0.85.0 → v0.92.0 (`platform.job@1`, `platform.job-frame@1`).
++
++The migration is **additive and expand-only**: nothing existing is altered, renamed or dropped, and there is
++**no backfill** — both tables start empty and only fill once a caller dispatches a job, which nothing in
++production does yet. Existing rows of `robots`, `devices`, `command_audit` and `bot_event_outbox` are untouched.
++
++## Cutover manifest (the harness writes it into the PR description)
++
++```cutover
++{"runbook": "docs/cutover-agent-bot-t14-step-l1-openvibe-bot-125030-muttjrzf.md", "rehearsal": "bot-t14-l1"}
++```
++
++- `runbook` — this file, the runbook the rehearsal follows.
++- `rehearsal` — the marker name `bot-t14-l1`; the rehearsal writes `ds/deploy/rehearsals/bot-t14-l1.json` =
++ `{"ok": true}` on the harness when it is green. That marker is written by whoever runs the rehearsal, never by
++ this runbook and never by the PR author.
++
++## What runs, in which order
++
++1. **Merge gate (harness).** Checks green, review SHIP, this runbook on the verified head, and
++ `ds/deploy/rehearsals/bot-t14-l1.json` saying `{"ok": true}` for the pushed head. Only then does the
++ squash-merge run.
++2. **Backup.** Take the standard pre-deploy backup of Bot's PostgreSQL database (pgBackRest) and confirm the
++ latest restore point is current. No deploy starts on an unbacked database.
++3. **Deploy.** `ov deploy bot` queues the guarded deploy on `openvibe-ovh` (unit `openvibe-bot.service`, layout
++ `/opt/openvibe.bot`, port 4630). On boot the service applies `migrations/` with the owner role on
++ `DATABASE_DIRECT_URL` (serialised by an advisory lock): `0004_run_jobs.sql` creates the two tables, then the
++ service serves on the pooled `DATABASE_URL`.
++4. **Configure (optional, any time later).** Set `BOT_BILLING_URL` and `BOT_BILLING_TOKEN` (a Network service
++ token, audience `openvibe.billing`, holding `billing.usage.record`) in `/etc/openvibe/bot.env` once Billing
++ grants Bot. Until then readings queue; nothing is lost and nothing is billed twice when the relay starts.
++5. **Verify.** Run the checks in the next section before the deploy is called good. A failed readiness check
++ rolls the release back automatically.
++
++## The backup
++
++- Command: the standard pgBackRest backup of the Bot database, then `pgbackrest info` to confirm a restore
++ point newer than the deploy start.
++- The migration only creates empty tables, so there is nothing of this PR's to restore; the backup is the guard
++ that lets the deploy proceed, and the way back for any other commit in the same release.
++
++## How the result is verified
++
++1. `openvibe-bot.service` is active after the restart (no crash loop); the start line reads
++ `usage relay off (readings wait)` while `BOT_BILLING_URL`/`BOT_BILLING_TOKEN` are unset.
++2. `GET http://127.0.0.1:4630/api/health` and `GET http://127.0.0.1:4630/api/ready` answer 2xx.
++3. The migrations table lists `0004_run_jobs.sql` as applied — **0 pending** — and `run_jobs` and
++ `run_usage_outbox` exist and are empty.
++4. Row counts of `robots`, `devices`, `command_audit` and `bot_event_outbox` are unchanged.
++5. A paired device still connects on `/device`, and a command (`halt`) still reaches it and is acked.
++
++If any of 1–5 fails, the deploy is not good: roll back (below) and investigate.
++
++## Way back (rollback and restore)
++
++- **Automatic:** a non-2xx `/api/ready` while the deploy waits restores the previous release's sha.
++- **Manual:** `ovhost rollback bot --to <previous sha>` on `openvibe-ovh`, then re-run checks 1, 2, 4 and 5.
++- **Data:** leave the two tables in place. The previous release never reads `run_jobs` or `run_usage_outbox`,
++ so they are inert; they are empty unless a job was dispatched, and any queued readings in
++ `run_usage_outbox` must be kept (they are owed to Billing) and are delivered once a release with this PR
++ returns. Never drop them in a rollback, and never restore the backup for this PR alone.
++
++## Rehearsal on a copy of the data
++
++Whoever runs this runbook rehearses it **on a copy** — never on production — against a scratch PostgreSQL, with
++main's migrations, the repository's fixtures and this PR's head. The fenced block declares the commands the
++harness runs after it has applied the migrations:
++
++```rehearse
++# 0004_run_jobs.sql is additive (two new tables, no backfill). This proves the job frames, the per-second
++# metering and the Billing outbox against the migrated scratch schema.
++node test/run-jobs.test.js
++```
++
++1. The harness starts a scratch database, applies the base branch's migrations and the repository's fixtures,
++ then this PR's migrations: exactly `0004_run_jobs.sql` applies, and a second run applies nothing.
++2. It runs `node test/run-jobs.test.js`: a job reaches a connected Node and only for an advertised class; it is
++ resent until acked; exited 0 writes one reading per second plus the partial last second; a lost `job_usage`
++ is backfilled; a second `job_exit` writes nothing; a cancel before `job_started` bills nothing;
++ `usage_read` never regresses; and with Billing unset or down readings stay queued until a later sweep
++ delivers each key once.
++3. It confirms 0 pending migrations, the two new tables present, and unchanged row counts of the existing tables
++ against the copy.
++4. When all of it is green, record the marker `ds/deploy/rehearsals/bot-t14-l1.json` = `{"ok": true}` on the
++ harness. **This runbook must not write that marker itself** — it is written by whoever performed the
++ rehearsal, from the runbook's exit code.
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 095d2fc..c1e04ed 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -82,8 +82,8 @@ At most 64 frames wait; any more are answered `error` `bot.not_ready`.
| `telemetry` | `battery, voltage, sensors{}, events[]?, …` | at most 2 Hz; extra frames are dropped — except a frame with a non-empty `events` array (a fault, a bump, low battery), which is always delivered and does not count against the samples' window. `battery` is either a 0..1 fraction or an object `{volts, percent}` with `percent` 0..100 (as OpenVibe.Node sends it); `robot_state.battery` is always the fraction 0..1 (or `null`), while `robot_state.telemetry` keeps the raw frame |
| `status` | `firmware, capabilities, faults[], estop_latched, device_kind?, drivers[]?, agent_version?` | on connect and on change. See below |
| `reauth` | `token` | a Network-paired machine only: a fresh node token for the same principal, before the last one expires (300 s); a refused one is answered `error` `bot.reauth_refused` and moves nothing |
--| `ack` | `id` | a command ran |
--| `nack` | `id, fault_code` | a command was refused on the device |
++| `ack` | `id` | a command ran (or, keyed by a job id, a job was accepted: §1.2) |
++| `nack` | `id, fault_code` | a command (or a job, §1.2) was refused on the device |
| `estop_state` | `latched, by, at, robot_id?` | the device's own latch, a **report**; `robot_id` omitted = every robot the device serves. See below |
`kind` is one of `drive`, `actuator`, `ptz`, `say`, `display`, `halt`, and a device only ever gets the
@@@ -238,6 +238,51 @@ heartbeat or any frame brings it back. Commands are never replayed after a recon
---
++### 1.2 Jobs (platform.job-frame@1)
++
++Bot is the dispatcher of plan T14: it hands a `platform.job@1` job to a paired OpenVibe.Node over this socket
++and meters it. There is no HTTP route; the Run service (which owns `run.job.*`) calls
++`server/jobs/dispatch.js` `dispatch(db, nodeId, job, { link, project, subject, provider })` and
++`cancel(db, jobId, { link })`. Each frame carries the envelope (`v`, `seq`, `ts`, `type`) and is validated
++against `platform.job-frame@1` from the pinned openvibe-contracts; an invalid one is answered `error`
++`bot.bad_frame`, and one naming a job of another device `error` `bot.unknown_job` (nothing changes).
++
++| direction | type | fields | Bot's part |
++|---|---|---|---|
++| server → device | `job` | `job` (platform.job@1) | sent by `dispatch` when the device is connected, and resent on every reconnect while unacked. Never sent for a `class` missing from the device's stored `capabilities.worker.runtime_classes` (dispatch answers `409 bot.class_unadvertised`; a queued job whose class is no longer advertised fails with `fault_code` `bot.class_unadvertised`) |
++| server → device | `job_cancel` | `id` | sent by `cancel` if the job was sent and has not ended; resent on a reconnect and when a `job_started` or `ack` crosses it. A job not yet started is `cancelled` at once; a cancelled job is never sent again |
++| server → device | `job_exit_ack` | `id` | after every reading of the job is committed to `run_usage_outbox`; the Node then forgets the job and stops resending its `job_exit` |
++| device → server | `ack` / `nack` | `id`, `fault_code` | the Node accepted (`placed`) or refused (`failed`, `fault_code` kept) the job |
++| device → server | `job_started` | `id, started_ms` | `running`; the first `started_ms` anchors every second and never changes |
++| device → server | `job_stdout` | `id, chunk_seq, chunk` | the last 1 MiB of each job, in memory; a `chunk_seq` already held is dropped. Never metered |
++| device → server | `job_usage` | `id, started_ms, second, cpu_ms?` | one reading for `second` (quantity 1); ignored once `job_exit` was taken |
++| device → server | `job_exit` | `id, reason, code, result, usage{started_ms?, wall_ms, …}` | authoritative: writes (or finds) every second it stands for, sets the final state, then `job_exit_ack`. A second `job_exit` writes nothing and is acked again |
++
++States (`run_jobs.state`): `queued` → `placed` (ack) → `running` (`job_started`) → `succeeded` (exited, code 0),
++`failed` (exited ≠ 0, `limit`, `stopped`, `failed`, or a `nack`), `cancelled`, `expired` (`ttl`).
++
++**Metering keys.** Each wall-clock second `n` of a job is one `platform.usage-sample@1` reading:
++
++| field | value |
++|---|---|
++| `id`, `idempotency_key` | `run:<job id>:<n>` |
++| `service` / `operation` / `unit` | `run` / `function.invoke` / `s` |
++| `quantity` | `1`; the partial last second `(wall_ms mod 1000)/1000`, never 0 |
++| `at` | `new Date(started_ms + n*1000).toISOString()` |
++| `resource` / `node` / `source` | the job id / the device id (`dev_…`) / `openvibe-node.worker` |
++| `project`, `subject`, `provider` | from Bot's record of the job (what `dispatch` was given); left out when unset |
++
++No field depends on when or how often a frame arrived, and no rating field is set. `job_exit` writes seconds
++0 … floor(wall_ms/1000) − 1 and the partial last second, so a lost `job_usage` is backfilled; a resend, a
++`job_usage` and the backfill of one second are one key, kept once in `run_usage_outbox` (`ON CONFLICT DO
++NOTHING`). `run_jobs.usage_read` (seconds queued) only moves forward. Every reason is billed for the seconds the
++process held, never beyond the job's own `limits.wall_ms` (a later second is not metered and `wall_ms` is capped
++there). The relay posts each reading to OpenVibe.Billing's `billing.usage.record` (`POST <BOT_BILLING_URL>/api/v1/usage`,
++`Authorization: Bearer <BOT_BILLING_TOKEN>`), one per request; 201 (written) and 200 (identical replay) mark it
++sent. 400/409/413/422 mark it rejected (kept, never resent). Anything else, and `BOT_BILLING_URL` or
++`BOT_BILLING_TOKEN` unset, leaves it queued and retried with backoff: a reading is never dropped and never billed
++twice.
++
## 2. `/control` — the operator WebSocket
### Authentication
ddiff --git a/migrations/0004_run_jobs.sql b/migrations/0004_run_jobs.sql
nnew file mode 100644
iindex 0000000..2bc8cf3
---- /dev/null
++++ b/migrations/0004_run_jobs.sql
@@@ -0,0 +1,53 @@
++-- phase: expand
++--
++-- Jobs on the device link (plan T14 step L1; platform.job@1 inside platform.job-frame@1): Bot is the dispatcher
++-- that hands a job to a paired Node and meters it per wall-clock second for Billing. Additive only: two new
++-- tables, nothing existing changes, nothing to backfill.
++--
++-- run_jobs one row per job id: what was sent, to which device, its state and how far it is metered
++-- run_usage_outbox platform.usage-sample@1 readings waiting for Billing (the openvibe-sdk outboxSchema shape).
++-- event_id is the reading's idempotency_key run:<job id>:<second>, so a second is queued once
++-- however often the Node reports it.
++
++CREATE TABLE IF NOT EXISTS run_jobs (
++ id text PRIMARY KEY CHECK (id ~ '^job_[0-9A-HJKMNP-TV-Z]{26}$'), -- platform.job@1 id
++ node_id text NOT NULL REFERENCES devices (id), -- dev_…: the device whose Node runs it
++ project_id text, -- prj_… of the payer; no FK, Bot holds no projects
++ subject text, -- who the readings are for (user:usr_…)
++ provider text,
++ class text NOT NULL,
++ job jsonb NOT NULL, -- the platform.job@1 body, resent verbatim until acked
++ state text NOT NULL CHECK (state IN ('queued', 'placed', 'running', 'succeeded', 'failed', 'cancelled', 'expired')),
++ cancel_requested boolean NOT NULL DEFAULT false,
++ fault_code text, -- the Node's nack, or why Bot stopped sending it
++ sent_at bigint, -- first time the job frame went out (ms)
++ started_ms bigint, -- the Node's started_ms: anchors every usage second
++ wall_ms bigint, -- job_exit.usage.wall_ms as billed (capped at limits.wall_ms)
++ usage_read integer NOT NULL DEFAULT 0 CHECK (usage_read >= 0), -- seconds queued for Billing; never moves back
++ exit_reason text, -- job_exit.reason; set once, by the first job_exit
++ exit_code integer,
++ result jsonb,
++ created_at bigint NOT NULL,
++ updated_at bigint NOT NULL,
++ finished_at bigint
++);
++-- Resend on reconnect: a device's unacked jobs, oldest first; the cancel sweep of the same device.
++CREATE INDEX IF NOT EXISTS run_jobs_node_state ON run_jobs (node_id, state, created_at);
++
++-- ── Usage readings for Billing: the openvibe-sdk outbox (outboxSchema('run_usage_outbox')) ─────────
++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;
++CREATE INDEX IF NOT EXISTS run_usage_outbox_rejected ON run_usage_outbox (id) WHERE rejected_at IS NOT NULL;
ddiff --git a/package-lock.json b/package-lock.json
iindex 3293ad5..75af0a8 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -14,7 +14,7 @@
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.85.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0",
@@@ -844,9 +844,8 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.85.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.85.0",
-- "integrity": "sha512-Xo54hG4kiL1kOTcxwbnH8cmt2zZhvp/gqCs+egPBLLSBg+SjVuhXIPaX8EJPQux5x1YxwO7wjDDywcHvIaGoqA==",
++ "version": "0.92.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex 705c88e..feff7c5 100644
---- a/package.json
++++ b/package.json
@@@ -26,7 +26,7 @@
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.85.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0",
ddiff --git a/server/app.js b/server/app.js
iindex 99c327f..5ceb4da 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -9,7 +9,7 @@
* /auth/* Network SSO session for people (the control WS accepts the ov_token cookie)
* GET / a one-line text placeholder; the panel and the site are designed separately
*
-- * createApp({ config, db, valkey, registry, keys, hub, outbox, nodes, openre, now, fetchImpl, log }) — everything
++ * createApp({ config, db, valkey, registry, keys, hub, outbox, usage, nodes, openre, now, fetchImpl, log }) — everything
* injectable. `db` is an openvibe-sdk/db handle with the schema migrated; `hub` the realtime WebSocket
* hub (server/realtime.js), whose handleUpgrade the caller attaches to the HTTP server.
*/
@@@ -23,6 +23,7 @@ const { openDb } = require('./db');
const { createKeyProvider, createUserAuth, createNodePrincipals } = require('./network');
const { createOpenRe } = require('./openre/client');
const { createBotOutbox } = require('./events/outbox');
++const { createUsageRelay } = require('./jobs/metering');
const { createDomain } = require('./domain');
const { createRealtime } = require('./realtime');
const { createApiAuth } = require('./api/auth');
@@@ -45,7 +46,9 @@ function createApp(opts = {}) {
const keys = opts.keys || createKeyProvider(config, { fetchImpl, log });
const userAuth = createUserAuth(config, keys);
const outbox = opts.outbox || createBotOutbox({ db, config, fetchImpl: opts.eventsFetch, now, log });
-- const hub = opts.hub || createRealtime({ config, keys, userAuth, log, now });
++ // Job usage readings → Billing (server/jobs/metering.js); started by server/index.js with the other jobs.
++ const usage = opts.usage || createUsageRelay({ db, config, fetchImpl: opts.billingFetch || fetchImpl, now, log });
++ const hub = opts.hub || createRealtime({ config, keys, userAuth, usage, log, now });
const nodes = opts.nodes || createNodePrincipals(config, { fetchImpl });
const openre = opts.openre !== undefined ? opts.openre : createOpenRe(config, { fetchImpl });
const domain = createDomain({ db, config, outbox, link: hub, nodes, openre, now, log });
@@@ -101,7 +104,7 @@ function createApp(opts = {}) {
return res.status(500).type('text/plain').send('Something went wrong\n');
});
-- Object.assign(app.locals, { config, db, valkey, domain, keys, outbox, hub, userAuth, metrics });
++ Object.assign(app.locals, { config, db, valkey, domain, keys, outbox, usage, hub, userAuth, metrics });
return app;
}
ddiff --git a/server/config.js b/server/config.js
iindex 0181c4c..bea1000 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -139,6 +139,15 @@ function loadConfig(env = process.env) {
token: String(env.BOT_OPENRE_TOKEN || '').trim(),
timeoutMs: Math.max(100, int(env.BOT_OPENRE_TIMEOUT_MS, 8000)),
},
++ // OpenVibe.Billing (plan T14 L1): job usage readings go to billing.usage.record (POST <url>/api/v1/usage).
++ // token is a Network service token for audience openvibe.billing holding billing.usage.record; it is never
++ // logged or returned. Either unset: readings wait in run_usage_outbox (never dropped) until both are set.
++ billing: {
++ url: trim(env.BOT_BILLING_URL || ''),
++ token: String(env.BOT_BILLING_TOKEN || '').trim(),
++ intervalMs: Math.max(100, int(env.BOT_BILLING_INTERVAL_MS, 2000)),
++ timeoutMs: Math.max(100, int(env.BOT_BILLING_TIMEOUT_MS, 5000)),
++ },
// The one-line installer the owner copies next to the pairing code. scriptUrl is what the command
// prints; sourceUrl is where GET /install sends the client (302). Bot keeps no copy of the script:
// it redirects to OpenVibe.Node's canonical one, and only this config (never a query parameter)
ddiff --git a/server/index.js b/server/index.js
iindex 58d6fb4..1ba614e 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -10,7 +10,8 @@
* migration run is serialised by an advisory lock.
*
* Background jobs (BOT_JOBS=off disables them): the events outbox relay, the 30-day audit prune, the
-- * pairing-code sweep and the turn-queue sweep (expire a turn, promote the next waiting person).
++ * pairing-code sweep, the turn-queue sweep (expire a turn, promote the next waiting person) and the job usage
++ * relay to Billing (BOT_BILLING_URL and BOT_BILLING_TOKEN; unset, readings wait in run_usage_outbox).
*/
const { loadConfig } = require('./config');
const { openDb, migrate } = require('./db');
@@@ -30,7 +31,7 @@ async function main() {
const valkey = createValkey({ url: config.valkey.url, prefix: config.valkey.prefix });
const app = createApp({ config, db, valkey, registry });
-- const { domain, keys, outbox, hub } = app.locals;
++ const { domain, keys, outbox, usage, hub } = app.locals;
keys.start();
const timers = [];
@@@ -40,11 +41,12 @@ async function main() {
every(30 * 60 * 1000, () => domain.pairing.prune());
every(1000, async () => { for (const robotId of await domain.queue.sweep()) hub.broadcast(robotId); });
outbox.start();
++ usage.start();
}
const server = app.listen(config.port, config.host, () => {
console.log(`[Bot] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (store ${db.store}; valkey ${valkey ? 'on' : 'off: per process'})`);
-- console.log(`[Bot] profiles from server/profiles; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (outbox accumulates)'}; heartbeat ${config.device.heartbeatMs} ms, offline after ${config.device.heartbeatMs * config.device.offlineMisses + config.device.offlineGraceMs} ms`);
++ console.log(`[Bot] profiles from server/profiles; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (outbox accumulates)'}; usage relay ${usage.enabled ? `→ ${config.billing.url}` : 'off (readings wait)'}; heartbeat ${config.device.heartbeatMs} ms, offline after ${config.device.heartbeatMs * config.device.offlineMisses + config.device.offlineGraceMs} ms`);
});
server.keepAliveTimeout = 65_000;
server.on('upgrade', (req, socket, head) => hub.handleUpgrade(req, socket, head));
@@@ -66,6 +68,7 @@ async function main() {
() => keys.stop(),
() => hub.close().catch(() => {}),
() => outbox.stop(),
++ () => usage.stop(),
],
close: [
() => db.close().catch(() => {}),
ddiff --git a/server/jobs/dispatch.js b/server/jobs/dispatch.js
nnew file mode 100644
iindex 0000000..2166f63
---- /dev/null
++++ b/server/jobs/dispatch.js
@@@ -0,0 +1,214 @@
++'use strict';
++
++/**
++ * The dispatcher (plan T14 step L1): Bot hands platform.job@1 jobs to a paired Node over the device link and
++ * meters them (platform.job-frame@1, docs/protocol.md "Jobs"). There is no HTTP route: the Run service, which
++ * owns run.job.*, calls this module later.
++ *
++ * dispatch(db, nodeId, job, { link, project, subject, provider }) validate, store, send (`job`)
++ * cancel(db, jobId, { link }) `job_cancel`
++ * createJobFrames({ db, usage, log, now }) the device side: ack/nack of a job,
++ * job_started, job_stdout, job_usage,
++ * job_exit (→ job_exit_ack), and the resend
++ * on reconnect
++ *
++ * `link` is the realtime hub (sendToDevice). A job is sent only to a device whose stored
++ * capabilities.worker.runtime_classes lists its class. An unacked job is resent on every reconnect (the Node acks
++ * a job it already holds again, or answers with its job_exit, and never runs one twice). job_exit_ack goes out
++ * once every reading of the job is committed to run_usage_outbox, from which the relay delivers it to Billing.
++ */
++const { validate } = require('openvibe-contracts');
++const store = require('./store');
++const metering = require('./metering');
++const { fail, json } = require('../util');
++
++const JOB = 'platform.job@1';
++const FRAME = 'platform.job-frame@1';
++const READING = 'platform.usage-sample@1';
++const STDOUT_RING_BYTES = 1024 * 1024; // the last 1 MiB of each job's stdout
++const STDOUT_JOBS = 1024; // rings kept, oldest dropped first
++
++const brief = (errors) => (errors || []).slice(0, 3).map((e) => `${e.path} ${e.message}`).join('; ');
++/** The classes a device's stored capabilities advertise (OpenVibe.Node status.capabilities.worker). */
++function advertisedClasses(capabilities) {
++ const worker = json(capabilities, {}).worker;
++ return worker && Array.isArray(worker.runtime_classes) ? worker.runtime_classes.filter((c) => typeof c === 'string') : [];
++}
++const canonical = (v) => JSON.stringify(v, (k, x) => (x && typeof x === 'object' && !Array.isArray(x) ? Object.fromEntries(Object.keys(x).sort().map((key) => [key, x[key]])) : x));
++const liveDevice = (db, id) => db.maybe('SELECT id, capabilities FROM devices WHERE id = $1 AND revoked_at IS NULL', [id]);
++
++/**
++ * Validate `job` against platform.job@1, store it for device `nodeId` and send it if the device is connected
++ * (else it goes out when the device connects). The same job again is idempotent; another body under the same
++ * id is refused. → { job: row, sent }
++ */
++async function dispatch(db, nodeId, job, { link, project = null, subject = null, provider = null, now = () => Date.now() } = {}) {
++ const v = validate(JOB, job);
++ if (!v.valid) fail(422, 'bot.invalid_job', `the job is not a valid ${JOB}: ${brief(v.errors)}`);
++ // The attribution every reading of the job will carry, checked now so a reading is never refused later.
++ const probe = validate(READING, metering.secondReading({ jobId: job.id, n: 0, startedMs: 0, nodeId: String(nodeId), project, subject, provider }));
++ if (!probe.valid) fail(422, 'bot.invalid_job', `project, subject or provider cannot attribute a reading: ${brief(probe.errors)}`);
++ const device = await liveDevice(db, nodeId);
++ if (!device) fail(404, 'bot.device_not_found', 'no such device, or it is revoked');
++ if (!advertisedClasses(device.capabilities).includes(job.class)) fail(409, 'bot.class_unadvertised', `the device does not advertise the runtime class ${job.class}`);
++ const { row, created } = await store.put(db, { job, nodeId, project, subject, provider, at: now() });
++ if (!created && (row.node_id !== nodeId || canonical(row.job) !== canonical(job)
++ || row.project_id !== project || row.subject !== subject || row.provider !== provider)) {
++ fail(409, 'bot.job_id_reused', 'this job id already names another job');
++ }
++ let sent = false;
++ if (row.state === 'queued' && !row.cancel_requested) {
++ sent = !!(link && link.sendToDevice(nodeId, { type: 'job', job }));
++ if (sent) await store.markSent(db, job.id, now());
++ }
++ return { job: await store.get(db, job.id), sent };
++}
++
++/** Ask the device to stop the job. A job not yet started is cancelled at once; job_exit still settles it. → { job, sent } */
++async function cancel(db, jobId, { link, now = () => Date.now() } = {}) {
++ const row = await store.requestCancel(db, String(jobId), now());
++ if (!row) fail(404, 'bot.job_not_found', 'no such job');
++ let sent = false;
++ if (row.sent_at != null && row.exit_reason == null && row.finished_at == null) sent = !!(link && link.sendToDevice(row.node_id, { type: 'job_cancel', id: row.id }));
++ return { job: row, sent };
++}
++
++/**
++ * The device side of the job frames. Every handler takes `link` = { deviceId, send(type, fields), error(code,
++ * detail) } for the device's socket; a frame naming a job of another device is answered bot.unknown_job and
++ * changes nothing.
++ */
++function createJobFrames({ db: getDb, usage = null, log = console, now = () => Date.now() }) {
++ const rings = new Map(); // job id → { nodeId, chunks, bytes, lastSeq, dropped }
++ const kick = () => { if (usage) usage.kick(); };
++
++ /** `ack`/`nack` keyed by a job id. */
++ async function onAck(link, msg) {
++ const db = getDb();
++ const row = await store.ofNode(db, String(msg.id || ''), link.deviceId);
++ if (!row) return;
++ if (msg.type === 'nack') {
++ await store.markRefused(db, row.id, String(msg.fault_code || 'nack').slice(0, 64), now());
++ return;
++ }
++ await store.markPlaced(db, row.id, now());
++ if (row.cancel_requested && row.exit_reason == null) link.send('job_cancel', { id: row.id });
++ }
++
++ /** After a device authenticates: resend its unacked jobs (still advertised) and the cancels it may not have seen. */
++ async function onConnect(link) {
++ const db = getDb();
++ const device = await liveDevice(db, link.deviceId);
++ if (!device) return;
++ const classes = advertisedClasses(device.capabilities);
++ for (const row of await store.unacked(db, link.deviceId)) {
++ if (!classes.includes(row.class)) { await store.markRefused(db, row.id, 'bot.class_unadvertised', now()); continue; }
++ if (link.send('job', { job: json(row.job, {}) })) await store.markSent(db, row.id, now());
++ }
++ for (const { id } of await store.cancelPending(db, link.deviceId)) link.send('job_cancel', { id });
++ }
++
++ async function onFrame(link, msg) {
++ const v = validate(FRAME, msg);
++ if (!v.valid) return link.error('bot.bad_frame', `not a valid ${FRAME} ${msg.type}: ${brief(v.errors)}`);
++ switch (msg.type) {
++ case 'job_started': return onStarted(link, msg);
++ case 'job_stdout': return onStdout(link, msg);
++ case 'job_usage': return onUsage(link, msg);
++ case 'job_exit': return onExit(link, msg);
++ default: return link.error('bot.unknown_message', `unknown type ${msg.type}`);
++ }
++ }
++ const unknown = (link, id) => link.error('bot.unknown_job', `no job ${id} on this device`);
++
++ async function onStarted(link, msg) {
++ const db = getDb();
++ const row = await store.ofNode(db, msg.id, link.deviceId);
++ if (!row) return unknown(link, msg.id);
++ await store.markStarted(db, row.id, msg.started_ms, now());
++ // It started although a cancel was asked (the cancel crossed it, or the link dropped): ask again.
++ if (row.cancel_requested && row.exit_reason == null) link.send('job_cancel', { id: row.id });
++ return undefined;
++ }
++
++ /** Best effort, never metered: the last 1 MiB per job; a chunk_seq already held is dropped. */
++ async function onStdout(link, msg) {
++ let ring = rings.get(msg.id);
++ if (!ring) {
++ if (!await store.ofNode(getDb(), msg.id, link.deviceId)) return unknown(link, msg.id);
++ ring = rings.get(msg.id) || { nodeId: link.deviceId, chunks: [], bytes: 0, lastSeq: 0, dropped: false };
++ rings.set(msg.id, ring);
++ if (rings.size > STDOUT_JOBS) rings.delete(rings.keys().next().value);
++ }
++ if (ring.nodeId !== link.deviceId) return unknown(link, msg.id);
++ if (msg.chunk_seq <= ring.lastSeq) return undefined;
++ const bytes = Buffer.byteLength(msg.chunk, 'utf8');
++ ring.chunks.push({ seq: msg.chunk_seq, chunk: msg.chunk, bytes });
++ ring.bytes += bytes; ring.lastSeq = msg.chunk_seq;
++ while (ring.bytes > STDOUT_RING_BYTES) { const old = ring.chunks.shift(); ring.bytes -= old.bytes; ring.dropped = true; }
++ return undefined;
++ }
++
++ /** The reading for one fully elapsed second, queued once; ignored after job_exit (it is authoritative). */
++ async function onUsage(link, msg) {
++ const db = getDb();
++ let queued = false;
++ const known = await db.tx(async (t) => {
++ const row = await store.ofNode(t, msg.id, link.deviceId, { lock: true });
++ if (!row) return false;
++ if (row.exit_reason != null) return true;
++ const job = json(row.job, {});
++ // Never past the wall-clock cap the payer set: the Node kills the job there.
++ if ((msg.second + 1) * 1000 > job.limits.wall_ms) { log.warn(`[Bot] job ${row.id}: job_usage second ${msg.second} beyond limits.wall_ms, not metered`); return true; }
++ const startedMs = row.started_ms != null ? Number(row.started_ms) : msg.started_ms;
++ if (row.started_ms == null) await store.markStarted(t, row.id, startedMs, now());
++ queued = await metering.enqueue(t, metering.secondReading(readingFields(row, startedMs, msg.second)), now());
++ await store.advanceUsage(t, row.id, msg.second + 1, now());
++ return true;
++ });
++ if (!known) return unknown(link, msg.id);
++ if (queued) kick();
++ return undefined;
++ }
++
++ /**
++ * The authoritative usage: write (or find already written) every second job_exit stands for, settle the job,
++ * then job_exit_ack. A second job_exit writes nothing and is acked again.
++ */
++ async function onExit(link, msg) {
++ const db = getDb();
++ const known = await db.tx(async (t) => {
++ const row = await store.ofNode(t, msg.id, link.deviceId, { lock: true });
++ if (!row) return false;
++ if (row.exit_reason != null) return true;
++ const job = json(row.job, {});
++ const wallMs = Math.min(msg.usage.wall_ms, job.limits.wall_ms);
++ const startedMs = row.started_ms != null ? Number(row.started_ms) : (msg.usage.started_ms ?? null);
++ if (wallMs > 0) {
++ for (const r of metering.exitReadings({ wallMs, ...readingFields(row, startedMs) })) await metering.enqueue(t, r, now());
++ await store.advanceUsage(t, row.id, Math.ceil(wallMs / 1000), now());
++ }
++ await store.finish(t, row.id, { reason: msg.reason, code: msg.code, result: msg.result, wallMs, startedMs, at: now() });
++ return true;
++ });
++ if (!known) return unknown(link, msg.id);
++ kick();
++ link.send('job_exit_ack', { id: msg.id });
++ return undefined;
++ }
++
++ function readingFields(row, startedMs, n) {
++ return { jobId: row.id, n, startedMs, nodeId: row.node_id, project: row.project_id, subject: row.subject, provider: row.provider };
++ }
++
++ /** What is held of a job's stdout (the last 1 MiB), or null. */
++ function stdout(jobId) {
++ const ring = rings.get(jobId);
++ if (!ring) return null;
++ return { text: ring.chunks.map((c) => c.chunk).join(''), first_seq: ring.chunks.length ? ring.chunks[0].seq : null, last_seq: ring.lastSeq, truncated: ring.dropped };
++ }
++
++ return { onAck, onConnect, onFrame, stdout };
++}
++
++module.exports = { dispatch, cancel, createJobFrames, advertisedClasses };
ddiff --git a/server/jobs/metering.js b/server/jobs/metering.js
nnew file mode 100644
iindex 0000000..d05fcd0
---- /dev/null
++++ b/server/jobs/metering.js
@@@ -0,0 +1,105 @@
++'use strict';
++
++/**
++ * Job metering (platform.job-frame@1 `job_usage` / `job_exit`): one platform.usage-sample@1 reading per
++ * wall-clock second a job's process held, queued in run_usage_outbox and posted to OpenVibe.Billing's
++ * billing.usage.record (POST /api/v1/usage, one reading per request).
++ *
++ * Every field of a reading is derived from the job and its second: id = idempotency_key = run:<job id>:<n>,
++ * service run, operation function.invoke, unit s, quantity 1 (the partial last second (wall_ms mod 1000)/1000,
++ * never 0), at = started_ms + n*1000, resource the job id, node the device id, source openvibe-node.worker, and
++ * project, subject and provider from Bot's own record of the job. Nothing depends on when a frame arrived, so a
++ * resend, a job_usage and the job_exit backfill of one second are one key: the outbox keeps the first
++ * (ON CONFLICT DO NOTHING) and Billing replays an identical reading (200) and refuses a different one (409).
++ *
++ * The relay (openvibe-sdk createPgOutbox) runs while BOT_BILLING_URL and BOT_BILLING_TOKEN are set. If either is
++ * unset, or Billing is down, has no grant for Bot yet or answers 401/403/404/429/5xx, readings wait and are
++ * retried with backoff, across restarts: never dropped. Only Billing refusing the reading itself (400, 409, 413,
++ * 422) marks a row rejected; it is kept with its error and never sent again, so nothing is billed twice.
++ */
++const { validate } = require('openvibe-contracts');
++const { createPgOutbox } = require('openvibe-sdk/events');
++
++const TABLE = 'run_usage_outbox'; // migrations/0004_run_jobs.sql
++const SCHEMA = 'platform.usage-sample@1';
++const SERVICE = 'run';
++const OPERATION = 'function.invoke';
++const UNIT = 's';
++const SOURCE = 'openvibe-node.worker';
++const REFUSED = new Set([400, 409, 413, 422]);
++
++/** The reading for second `n` of a job; `quantity` is 1 except for the partial last second. */
++function secondReading({ jobId, n, quantity = 1, startedMs, nodeId, project = null, subject = null, provider = null }) {
++ if (!Number.isInteger(n) || n < 0) throw new TypeError(`second must be an integer >= 0, not ${n}`);
++ if (!Number.isInteger(startedMs) || startedMs < 0) throw new TypeError('started_ms must be an integer >= 0');
++ const key = `run:${jobId}:${n}`;
++ const r = {
++ id: key, idempotency_key: key, service: SERVICE, operation: OPERATION, unit: UNIT, quantity,
++ at: new Date(startedMs + n * 1000).toISOString(), resource: jobId, node: nodeId, source: SOURCE,
++ };
++ if (project) r.project = project;
++ if (subject) r.subject = subject;
++ if (provider) r.provider = provider;
++ return r;
++}
++
++/**
++ * The readings job_exit's usage stands for: seconds 0 … floor(wall_ms/1000)-1 with quantity 1 and, when
++ * wall_ms mod 1000 > 0, second floor(wall_ms/1000) with quantity (wall_ms mod 1000)/1000.
++ */
++function exitReadings({ wallMs, ...job }) {
++ const full = Math.floor(wallMs / 1000);
++ const out = [];
++ for (let n = 0; n < full; n++) out.push(secondReading({ ...job, n }));
++ const rest = wallMs % 1000;
++ if (rest > 0) out.push(secondReading({ ...job, n: full, quantity: rest / 1000 }));
++ return out;
++}
++
++/** Queue one reading inside the caller's transaction; a key already queued changes nothing. → inserted? */
++async function enqueue(t, reading, at) {
++ const v = validate(SCHEMA, reading);
++ if (!v.valid) throw new Error(`reading ${reading.id} is not a valid ${SCHEMA}: ${JSON.stringify(v.errors)}`);
++ const n = await t.exec(`INSERT INTO ${TABLE} (event_id, envelope, created_at) VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING`,
++ [reading.idempotency_key, JSON.stringify(reading), at]);
++ return n > 0;
++}
++
++// createPgOutbox's `events` seam pointed at Billing. isPermanent() reads err.status: 422 rejects the row, 503 retries.
++function billingSink({ url, token, fetchImpl, timeoutMs }) {
++ return {
++ prepare: (reading) => reading,
++ async publish(reading) {
++ const res = await fetchImpl(`${url}/api/v1/usage`, {
++ method: 'POST',
++ headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'application/json' },
++ body: JSON.stringify(reading), signal: AbortSignal.timeout(timeoutMs),
++ });
++ // 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 };
++ },
++ };
++}
++
++/**
++ * The relay from run_usage_outbox to Billing. enabled only with config.billing.url and .token; otherwise start(),
++ * kick() and flush() do nothing and readings accumulate until a process with both set sweeps them.
++ */
++function createUsageRelay({ db, config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console }) {
++ const { url, token, intervalMs, timeoutMs } = config.billing;
++ const count = (where) => db.value(`SELECT count(*) FROM ${TABLE} WHERE ${where}`).then(Number);
++ const counts = { pending: () => count('sent_at IS NULL AND rejected_at IS NULL'), rejected: () => count('rejected_at IS NOT NULL') };
++ if (!url || !token) {
++ const idle = async () => ({ sent: 0, failed: 0, rejected: 0 });
++ return { enabled: false, start() {}, stop: async () => {}, kick() {}, flush: idle, ...counts };
++ }
++ let lastError = null;
++ const relay = createPgOutbox(db, {
++ events: billingSink({ url, token, fetchImpl, timeoutMs }), table: TABLE, batchSize: 1, intervalMs, now,
++ onError: (err) => { const m = err && err.message; if (m !== lastError) log.warn(`[Bot] usage reading not delivered: ${m}`); lastError = m; },
++ });
++ return { enabled: true, start: relay.start, stop: relay.stop, kick: relay.kick, flush: relay.flush, ...counts };
++}
++
++module.exports = { secondReading, exitReadings, enqueue, createUsageRelay, TABLE, SERVICE, OPERATION, UNIT, SOURCE };
ddiff --git a/server/jobs/store.js b/server/jobs/store.js
nnew file mode 100644
iindex 0000000..9e2604f
---- /dev/null
++++ b/server/jobs/store.js
@@@ -0,0 +1,61 @@
++'use strict';
++
++/**
++ * run_jobs (migrations/0004_run_jobs.sql): the dispatcher's record of every job it handed to a device.
++ *
++ * queued stored; sent (again on every reconnect) until the Node acks it
++ * placed the Node acked it
++ * running job_started
++ * succeeded / failed / cancelled / expired from job_exit (exited 0 / exited ≠ 0, limit, stopped, failed /
++ * cancelled / ttl); failed also when the Node nacked it
++ *
++ * A cancel before job_started marks the row cancelled at once; job_exit, which is authoritative, still sets the
++ * final state and is still metered. exit_reason is written once, by the first job_exit: after it no usage frame
++ * of the job is metered again. usage_read only moves forward (GREATEST).
++ */
++const STATE_OF_EXIT = { cancelled: 'cancelled', ttl: 'expired' };
++const stateOfExit = (reason, code) => (reason === 'exited' && code === 0 ? 'succeeded' : STATE_OF_EXIT[reason] || 'failed');
++
++/** Insert a new job; the same id again is a conflict unless it is the same job for the same device. → { row, created } */
++async function put(db, { job, nodeId, project = null, subject = null, provider = null, at }) {
++ const row = await db.maybe(
++ `INSERT INTO run_jobs (id, node_id, project_id, subject, provider, class, job, state, created_at, updated_at)
++ VALUES ($1, $2, $3, $4, $5, $6, $7, 'queued', $8, $8) ON CONFLICT (id) DO NOTHING RETURNING *`,
++ [job.id, nodeId, project, subject, provider, job.class, JSON.stringify(job), at]);
++ if (row) return { row, created: true };
++ return { row: await get(db, job.id), created: false };
++}
++
++const get = (db, id) => db.maybe('SELECT * FROM run_jobs WHERE id = $1', [id]);
++/** The job, only if it belongs to this device: a Node never reads or meters another device's job. */
++const ofNode = (db, id, nodeId, { lock = false } = {}) => db.maybe(`SELECT * FROM run_jobs WHERE id = $1 AND node_id = $2${lock ? ' FOR UPDATE' : ''}`, [id, nodeId]);
++
++/** Unacked, uncancelled jobs of a device, oldest first: what a reconnect resends. */
++const unacked = (db, nodeId) => db.many(`SELECT * FROM run_jobs WHERE node_id = $1 AND state = 'queued' AND NOT cancel_requested ORDER BY created_at, id`, [nodeId]);
++/** Jobs a cancel was asked for that the device may still hold (sent, no job_exit yet). */
++const cancelPending = (db, nodeId) => db.many(`SELECT id FROM run_jobs WHERE node_id = $1 AND cancel_requested AND sent_at IS NOT NULL AND exit_reason IS NULL AND finished_at IS NULL ORDER BY created_at, id`, [nodeId]);
++
++const markSent = (db, id, at) => db.exec('UPDATE run_jobs SET sent_at = COALESCE(sent_at, $2), updated_at = $2 WHERE id = $1', [id, at]);
++const markPlaced = (db, id, at) => db.exec(`UPDATE run_jobs SET state = 'placed', updated_at = $2 WHERE id = $1 AND state = 'queued'`, [id, at]);
++/** The Node refused it (nack), or Bot stopped sending it: failed, ended, nothing to meter. */
++const markRefused = (db, id, faultCode, at) => db.exec(
++ `UPDATE run_jobs SET state = CASE WHEN cancel_requested THEN 'cancelled' ELSE 'failed' END, fault_code = $2, finished_at = $3, updated_at = $3
++ WHERE id = $1 AND state IN ('queued', 'placed', 'cancelled') AND exit_reason IS NULL AND finished_at IS NULL`, [id, faultCode, at]);
++/** job_started: the first started_ms is kept (it anchors every second); running unless job_exit already came. */
++const markStarted = (db, id, startedMs, at) => db.exec(
++ `UPDATE run_jobs SET started_ms = COALESCE(started_ms, $2), state = CASE WHEN exit_reason IS NULL THEN 'running' ELSE state END, updated_at = $3
++ WHERE id = $1`, [id, startedMs, at]);
++/** A cancel: a job that has not started is cancelled now; a running one ends with its job_exit. → the row */
++const requestCancel = (db, id, at) => db.maybe(
++ `UPDATE run_jobs SET cancel_requested = true, state = CASE WHEN state IN ('queued', 'placed') THEN 'cancelled' ELSE state END, updated_at = $2
++ WHERE id = $1 RETURNING *`, [id, at]);
++/** Seconds queued so far; never moves back. */
++const advanceUsage = (db, id, upTo, at) => db.exec('UPDATE run_jobs SET usage_read = GREATEST(usage_read, $2), updated_at = $3 WHERE id = $1', [id, upTo, at]);
++/** The first job_exit: final state and exit fields, once. */
++const finish = (db, id, { reason, code, result, wallMs, startedMs, at }) => db.exec(
++ `UPDATE run_jobs SET state = $2, exit_reason = $3, exit_code = $4, result = $5, wall_ms = $6, started_ms = COALESCE(started_ms, $7),
++ finished_at = COALESCE(finished_at, $8), updated_at = $8
++ WHERE id = $1 AND exit_reason IS NULL`,
++ [id, stateOfExit(reason, code), reason, code, result === undefined || result === null ? null : JSON.stringify(result), wallMs, startedMs, at]);
++
++module.exports = { put, get, ofNode, unacked, cancelPending, markSent, markPlaced, markRefused, markStarted, requestCancel, advanceUsage, finish, stateOfExit };
ddiff --git a/server/realtime.js b/server/realtime.js
iindex 87597c3..e217bed 100644
---- a/server/realtime.js
++++ b/server/realtime.js
@@@ -7,7 +7,9 @@
* on the upgrade, never a query string; an unauthenticated socket may only send
* `pair`. A Network-paired machine presents its node token instead (bound on first
* use) and renews it with `reauth` before the 330 s deadline, else close 4002. Heartbeats every 1 s; offline after 2 missed + 3 s grace. Commands are
-- * never queued for an offline device and never replayed after a reconnect.
++ * never queued for an offline device and never replayed after a reconnect. Jobs
++ * (platform.job-frame@1, server/jobs/dispatch.js) are the exception: an unacked `job`
++ * is resent on every reconnect, and the Node never runs one id twice.
* wss://…/control a signed-in person (Network session cookie or Bearer) or a service token holding
* `bot.robot.control` acting for `X-OV-Subject`. Joins a robot, sends commands; the
* gate lives in the domain and every decision is audited.
@@@ -20,6 +22,7 @@ const { capabilities, ids } = require('openvibe-contracts');
const { json, iso, prefixedId } = require('./util');
const { userPrincipal, verifyService, verifyNode, isNodeToken, PRINCIPAL_SUB, decodePayload } = require('./api/auth');
const { getProfile } = require('./profiles');
++const { createJobFrames } = require('./jobs/dispatch');
const OPEN = 1;
const MAX_BACKLOG = 64; // frames a device may send before its authentication completes
@@@ -38,7 +41,7 @@ function parseCookies(header) {
const bearer = (req) => { const h = String(req.headers.authorization || ''); return h.startsWith('Bearer ') ? h.slice(7).trim() : null; };
const isSubject = (v) => typeof v === 'string' && ids.isSubjectId('user', v);
--function createRealtime({ config, keys, userAuth, log = console, now = () => Date.now() }) {
++function createRealtime({ config, keys, userAuth, usage = null, log = console, now = () => Date.now() }) {
let domain = null;
const deviceConns = new Map(); // device_id → conn
const subsByRobot = new Map(); // robot_id → Set<conn>
@@@ -51,12 +54,15 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
const deviceWss = new WebSocketServer({ noServer: true, maxPayload: 64 * 1024, perMessageDeflate: false });
const controlWss = new WebSocketServer({ noServer: true, maxPayload: 64 * 1024, perMessageDeflate: false });
const offlineAfterMs = config.device.heartbeatMs * config.device.offlineMisses + config.device.offlineGraceMs;
++ const jobs = createJobFrames({ db: () => domain.db, usage, log, now });
const nextSeq = (conn) => ++conn.seq;
function send(ws, obj) { if (ws.readyState === OPEN) { ws.send(JSON.stringify(obj)); return true; } return false; }
function sendFrame(conn, type, fields) { return send(conn.ws, { v: 1, seq: nextSeq(conn), ts: now(), type, ...fields }); }
function sendError(conn, code, detail) { return sendFrame(conn, 'error', { code, detail: detail || null }); }
const deviceRobotIds = (device) => json(device.robot_ids, []);
++ /** The job frames' view of one device socket (server/jobs/dispatch.js). */
++ const jobLink = (conn) => ({ deviceId: conn.device.id, send: (type, fields) => sendFrame(conn, type, fields), error: (code, detail) => sendError(conn, code, detail) });
// ── Upgrade routing ───────────────────────────────────────────────────────────────────────────
function handleUpgrade(req, socket, head) {
@@@ -88,6 +94,7 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
await sendConfig(conn);
await bringOnline(conn);
} catch (e) { log.warn(`[Bot] device auth: ${e.message}`); conn.backlog = null; ws.close(4002, 'invalid credential'); return; }
++ await jobs.onConnect(jobLink(conn)).catch((e) => log.warn(`[Bot] job resend to ${conn.device.id}: ${e.message}`));
await drainBacklog(conn);
}
}
@@@ -236,6 +243,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
case 'reauth': if (conn.node) return onReauth(conn, msg); return sendError(conn, 'bot.unknown_message', `unknown type ${msg.type}`);
case 'ack': case 'nack': return onAck(conn, msg);
case 'estop_state': return onEstopState(conn, msg);
++ // Jobs (platform.job-frame@1). Out: job, job_cancel, job_exit_ack (server/jobs/dispatch.js); in:
++ case 'job_started': case 'job_stdout': case 'job_usage': case 'job_exit': return jobs.onFrame(jobLink(conn), msg);
default: return sendError(conn, 'bot.unknown_message', `unknown type ${msg.type}`);
}
}
@@@ -274,6 +283,7 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
async function onAck(conn, msg) {
const id = String(msg.id || '');
++ if (id.startsWith('job_')) return jobs.onAck(jobLink(conn), msg); // a job's ack/nack, never a command's
const p = pendingCmds.get(id);
if (!p || p.deviceId !== conn.device.id) return; // unknown, already answered, or another device's command
pendingCmds.delete(id);
@@@ -517,6 +527,7 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
bindDomain(d) { domain = d; },
handleUpgrade,
isOnline, deviceState, sendToDevice, sendToRobotDevices, broadcast, closeDevice, robotState, refreshConfig, pushEstop,
++ jobStdout: (jobId) => jobs.stdout(jobId),
onlineCount() { let n = 0; for (const c of deviceConns.values()) if (c.online) n++; return n; },
devices() { return [...deviceConns.values()].map((c) => ({ device_id: c.device.id, online: c.online })); },
async close() {
ddiff --git a/test/run-jobs.test.js b/test/run-jobs.test.js
nnew file mode 100644
iindex 0000000..39a58fb
---- /dev/null
++++ b/test/run-jobs.test.js
@@@ -0,0 +1,281 @@
++'use strict';
++// Jobs on the device link (plan T14 L1, platform.job-frame@1): server/jobs/dispatch.js hands a platform.job@1 job
++// to a paired Node, only for a class it advertises, resends it until acked, and meters it into run_usage_outbox
++// one reading per wall-clock second (run:<job id>:<n>), backfilled and deduplicated from job_exit; the relay
++// delivers each reading to Billing once, and keeps it queued while Billing is unset or down.
++const assert = require('assert');
++const { ids, validate } = require('openvibe-contracts');
++const { boot, check, done } = require('./helpers/app');
++
++const PROJECT = `prj_${ids.ulid()}`;
++
++(async () => {
++ const t = await boot();
++ const alex = t.network.newUser('alex');
++ const SUBJECT = `user:${alex.subject}`;
++ // Required after boot(): it reloads server/, and these must be the same modules the app runs.
++ const { dispatch, cancel } = require('../server/jobs/dispatch');
++ const { createUsageRelay } = require('../server/jobs/metering');
++
++ let seq = 0;
++ const frame = (type, fields) => ({ v: 1, seq: ++seq, ts: Date.now(), type, ...fields });
++ const newJob = (over = {}) => ({
++ id: `job_${ids.ulid()}`, class: 'function', artifact: { name: 'thumbnail', version: '1.2.0' }, args: { width: 320 },
++ ttl_ms: 60000, limits: { wall_ms: 30000, cpu_ms: 30000, mem_bytes: 268435456 }, ...over,
++ });
++ const row = (id) => t.db.maybe('SELECT * FROM run_jobs WHERE id = $1', [id]);
++ const readings = async (id) => (await t.db.many('SELECT envelope FROM run_usage_outbox WHERE event_id LIKE $1 ORDER BY id', [`run:${id}:%`])).map((r) => r.envelope);
++ const send = (deviceId, job, opts = {}) => dispatch(t.db, deviceId, job, { link: t.hub, project: PROJECT, subject: SUBJECT, ...opts });
++ const sawJob = (c, id) => c.messages.some((m) => m.type === 'job' && m.job.id === id);
++
++ /** A Network-paired Node advertising `classes`, connected; its capabilities are persisted before it returns. */
++ async function node(classes = ['function']) {
++ const { robot } = await t.robot(alex);
++ const principal = t.network.addNode({ owner: alex.subject, ref: robot.id });
++ const n = { principal, c: null, deviceId: null };
++ n.connect = async () => {
++ n.c = await t.ws('/device', { headers: { Authorization: `Bearer ${t.network.signNode(principal)}` } });
++ n.deviceId = (await n.c.waitFor((m) => m.type === 'hello')).device_id;
++ await n.c.waitFor((m) => m.type === 'config');
++ return n.c;
++ };
++ await n.connect();
++ n.c.send(frame('status', { capabilities: { worker: { capabilities: classes.map((x) => `worker:${x}`), runtime_classes: classes } } }));
++ await poll(async () => { const d = await t.db.maybe('SELECT capabilities FROM devices WHERE id = $1', [n.deviceId]); return d && d.capabilities.worker; });
++ /** A frame from the Node, then the Bot frame that answers it (or the settle of a no-answer frame). */
++ n.say = (type, fields) => n.c.send(frame(type, fields));
++ n.exit = async (id, fields) => {
++ const before = n.c.messages.filter((m) => m.type === 'job_exit_ack' && m.id === id).length;
++ n.say('job_exit', { id, code: null, result: null, ...fields });
++ await n.c.waitFor(() => n.c.messages.filter((m) => m.type === 'job_exit_ack' && m.id === id).length > before);
++ };
++ /** A running job: dispatched, acked, started at `startedMs`. */
++ n.running = async (startedMs, over = {}) => {
++ const job = newJob(over);
++ await send(n.deviceId, job);
++ await n.c.waitFor((m) => m.type === 'job' && m.job.id === job.id);
++ n.say('ack', { id: job.id });
++ n.say('job_started', { id: job.id, started_ms: startedMs });
++ await poll(async () => (await row(job.id)).state === 'running');
++ return job;
++ };
++ return n;
++ }
++
++ await check('a job reaches its connected Node as a valid job frame and the ack places it', async () => {
++ const n = await node();
++ const job = newJob();
++ const r = await send(n.deviceId, job);
++ assert.strictEqual(r.sent, true);
++ assert.strictEqual(r.job.state, 'queued');
++ const f = await n.c.waitFor((m) => m.type === 'job' && m.job.id === job.id);
++ assert.deepStrictEqual(f.job, job);
++ assert.ok(validate('platform.job-frame@1', f).valid, JSON.stringify(validate('platform.job-frame@1', f).errors));
++ n.say('ack', { id: job.id });
++ await poll(async () => (await row(job.id)).state === 'placed');
++ const again = await send(n.deviceId, job);
++ assert.strictEqual(again.sent, false, 'the same job again is idempotent and not resent once acked');
++ await assert.rejects(send(n.deviceId, { ...job, args: { width: 1 } }), (e) => e.code === 'bot.job_id_reused');
++ await assert.rejects(send(n.deviceId, { ...newJob(), ttl_ms: 0 }), (e) => e.code === 'bot.invalid_job');
++ await assert.rejects(send(n.deviceId, newJob(), { project: 'not-a-project' }), (e) => e.code === 'bot.invalid_job');
++ });
++
++ await check('a class the Node does not advertise is never sent, at dispatch or on a reconnect', async () => {
++ const n = await node(['function']);
++ const code = newJob({ class: 'code' });
++ await assert.rejects(send(n.deviceId, code), (e) => e.code === 'bot.class_unadvertised');
++ assert.strictEqual(await row(code.id), null);
++ // Stored while the Node is away; it comes back advertising another class only.
++ n.c.close(); await n.c.waitForClose();
++ const fn = newJob();
++ assert.strictEqual((await send(n.deviceId, fn)).sent, false);
++ await t.db.query('UPDATE devices SET capabilities = $2 WHERE id = $1', [n.deviceId, JSON.stringify({ worker: { runtime_classes: ['code'] } })]);
++ await n.connect();
++ await poll(async () => (await row(fn.id)).state === 'failed');
++ assert.strictEqual((await row(fn.id)).fault_code, 'bot.class_unadvertised');
++ assert.ok(!sawJob(n.c, code.id) && !sawJob(n.c, fn.id), 'no job frame for a class the Node lacks');
++ });
++
++ await check('an unacked job is resent on every reconnect until acked, then never again', async () => {
++ const n = await node();
++ const job = newJob();
++ await send(n.deviceId, job);
++ await n.c.waitFor((m) => m.type === 'job' && m.job.id === job.id);
++ n.c.close(); await n.c.waitForClose();
++ await n.connect();
++ const f = await n.c.waitFor((m) => m.type === 'job' && m.job.id === job.id);
++ assert.deepStrictEqual(f.job, job);
++ n.say('ack', { id: job.id });
++ await poll(async () => (await row(job.id)).state === 'placed');
++ n.c.close(); await n.c.waitForClose();
++ await n.connect();
++ await t.wait(300);
++ assert.ok(!sawJob(n.c, job.id), 'an acked job is not resent');
++ });
++
++ await check('exited 0 writes one reading of quantity 1 per second plus the partial last second', async () => {
++ const n = await node();
++ const S = 1790935200250;
++ const job = await n.running(S);
++ n.say('job_usage', { id: job.id, started_ms: S, second: 0, cpu_ms: 640 });
++ n.say('job_usage', { id: job.id, started_ms: S, second: 1 });
++ await n.exit(job.id, { reason: 'exited', code: 0, result: { url: 'media:object-7' }, usage: { started_ms: S, wall_ms: 2350, cpu_ms: 1800 } });
++ const got = await readings(job.id);
++ assert.deepStrictEqual(got.map((r) => [r.idempotency_key, r.quantity, r.at]), [
++ [`run:${job.id}:0`, 1, new Date(S).toISOString()],
++ [`run:${job.id}:1`, 1, new Date(S + 1000).toISOString()],
++ [`run:${job.id}:2`, 0.35, new Date(S + 2000).toISOString()],
++ ]);
++ for (const r of got) {
++ assert.ok(validate('platform.usage-sample@1', r).valid);
++ assert.deepStrictEqual({ ...r, idempotency_key: undefined, id: undefined, quantity: undefined, at: undefined }, {
++ idempotency_key: undefined, id: undefined, quantity: undefined, at: undefined,
++ service: 'run', operation: 'function.invoke', unit: 's', resource: job.id, node: n.deviceId,
++ source: 'openvibe-node.worker', project: PROJECT, subject: SUBJECT,
++ });
++ assert.strictEqual(r.id, r.idempotency_key);
++ }
++ const j = await row(job.id);
++ assert.strictEqual(j.state, 'succeeded');
++ assert.strictEqual(j.usage_read, 3);
++ assert.deepStrictEqual(j.result, { url: 'media:object-7' });
++ const ack = n.c.messages.find((m) => m.type === 'job_exit_ack' && m.id === job.id);
++ assert.ok(validate('platform.job-frame@1', ack).valid);
++ });
++
++ await check('a lost job_usage is backfilled from job_exit.usage', async () => {
++ const n = await node();
++ const S = 1790935300000;
++ const job = await n.running(S);
++ n.say('job_usage', { id: job.id, started_ms: S, second: 0 });
++ n.say('job_usage', { id: job.id, started_ms: S, second: 2 }); // second 1 was lost with the link
++ await n.exit(job.id, { reason: 'limit', usage: { started_ms: S, wall_ms: 3000 } });
++ const got = await readings(job.id);
++ assert.deepStrictEqual(got.map((r) => [r.idempotency_key, r.quantity]), [0, 2, 1].map((k) => [`run:${job.id}:${k}`, 1]), 'second 1 written from job_exit; no partial second when wall_ms is whole');
++ assert.strictEqual((await row(job.id)).state, 'failed');
++ });
++
++ await check('job_exit twice writes one set and is acked twice', async () => {
++ const n = await node();
++ const S = 1790935400000;
++ const job = await n.running(S);
++ const exit = { reason: 'exited', code: 3, usage: { started_ms: S, wall_ms: 1500 } };
++ await n.exit(job.id, exit);
++ const first = await readings(job.id);
++ await n.exit(job.id, exit);
++ await n.exit(job.id, { ...exit, usage: { started_ms: S, wall_ms: 9000 } }); // a different resend changes nothing
++ assert.deepStrictEqual(await readings(job.id), first);
++ assert.strictEqual(first.length, 2);
++ const j = await row(job.id);
++ assert.strictEqual(j.state, 'failed');
++ assert.strictEqual(Number(j.wall_ms), 1500);
++ assert.strictEqual(j.exit_code, 3);
++ });
++
++ await check('job_cancel before job_started: the job is cancelled, never started, and nothing is billed', async () => {
++ const n = await node();
++ const job = newJob();
++ await send(n.deviceId, job);
++ await n.c.waitFor((m) => m.type === 'job' && m.job.id === job.id);
++ n.say('ack', { id: job.id });
++ await poll(async () => (await row(job.id)).state === 'placed');
++ const r = await cancel(t.db, job.id, { link: t.hub });
++ assert.strictEqual(r.sent, true);
++ assert.strictEqual(r.job.state, 'cancelled');
++ const f = await n.c.waitFor((m) => m.type === 'job_cancel' && m.id === job.id);
++ assert.ok(validate('platform.job-frame@1', f).valid);
++ await n.exit(job.id, { reason: 'cancelled', usage: { wall_ms: 0 } });
++ const j = await row(job.id);
++ assert.strictEqual(j.state, 'cancelled');
++ assert.strictEqual(j.started_ms, null);
++ assert.deepStrictEqual(await readings(job.id), []);
++ // Cancelled while the Node is away and never sent: it is not sent on the reconnect either.
++ n.c.close(); await n.c.waitForClose();
++ const away = newJob();
++ await send(n.deviceId, away);
++ await cancel(t.db, away.id, { link: t.hub });
++ await n.connect();
++ await t.wait(300);
++ assert.ok(!sawJob(n.c, away.id));
++ assert.strictEqual((await row(away.id)).state, 'cancelled');
++ });
++
++ await check('usage_read never regresses and a reading is never written twice', async () => {
++ const n = await node();
++ const S = 1790935500000;
++ const job = await n.running(S);
++ for (let k = 0; k < 5; k++) n.say('job_usage', { id: job.id, started_ms: S, second: k });
++ await poll(async () => (await row(job.id)).usage_read === 5);
++ n.say('job_usage', { id: job.id, started_ms: S, second: 1 }); // a replay after a reconnect
++ await n.exit(job.id, { reason: 'stopped', usage: { started_ms: S, wall_ms: 2000 } }); // smaller than what was read
++ const j = await row(job.id);
++ assert.strictEqual(j.usage_read, 5);
++ assert.deepStrictEqual((await readings(job.id)).map((r) => r.idempotency_key), [0, 1, 2, 3, 4].map((k) => `run:${job.id}:${k}`));
++ });
++
++ await check('a Node never meters another device\'s job, nor seconds past limits.wall_ms', async () => {
++ const a = await node();
++ const b = await node();
++ const S = 1790935600000;
++ const job = await a.running(S, { limits: { wall_ms: 2500, cpu_ms: 30000, mem_bytes: 268435456 } });
++ b.say('job_usage', { id: job.id, started_ms: S, second: 0 });
++ assert.strictEqual((await b.c.waitFor((m) => m.type === 'error')).code, 'bot.unknown_job');
++ b.say('job_exit', { id: job.id, reason: 'exited', code: 0, result: null, usage: { started_ms: S, wall_ms: 5000 } });
++ await b.c.waitFor(() => b.c.messages.filter((m) => m.type === 'error').length === 2);
++ assert.ok(!b.c.messages.some((m) => m.type === 'job_exit_ack'));
++ a.say('job_usage', { id: job.id, started_ms: S, second: 2 }); // [2000, 3000) is past the 2500 ms cap
++ await a.exit(job.id, { reason: 'limit', usage: { started_ms: S, wall_ms: 2600 } });
++ assert.deepStrictEqual((await readings(job.id)).map((r) => [r.idempotency_key, r.quantity]), [[`run:${job.id}:0`, 1], [`run:${job.id}:1`, 1], [`run:${job.id}:2`, 0.5]]);
++ assert.strictEqual(Number((await row(job.id)).wall_ms), 2500);
++ });
++
++ await check('Billing unset or down keeps readings queued; a later sweep delivers each key once', async () => {
++ assert.strictEqual(t.app.locals.usage.enabled, false, 'BOT_BILLING_* unset: no relay');
++ const queued = (await t.db.many('SELECT event_id FROM run_usage_outbox ORDER BY id')).map((r) => r.event_id);
++ assert.ok(queued.length >= 10);
++ assert.strictEqual(await t.app.locals.usage.pending(), queued.length, 'nothing dropped while Billing is unset');
++
++ // A Billing stand-in with POST /api/v1/usage's idempotency: 201 new, 200 identical replay, 409 a different reading.
++ const billed = new Map(); const posts = []; let up = false; let clock = Date.now();
++ const fetchImpl = async (url, init) => {
++ posts.push({ url, auth: init.headers.Authorization, body: JSON.parse(init.body) });
++ if (!up) return new Response('{}', { status: 503 });
++ const r = JSON.parse(init.body); const prev = billed.get(r.idempotency_key);
++ if (prev) return new Response('{}', { status: prev === init.body ? 200 : 409 });
++ billed.set(r.idempotency_key, init.body);
++ return new Response('{}', { status: 201 });
++ };
++ const config = { ...t.config, billing: { url: 'http://billing.test', token: 'svc-token', intervalMs: 100, timeoutMs: 2000 } };
++ const relay = createUsageRelay({ db: t.db, config, fetchImpl, now: () => clock, log: { warn() {} } });
++ assert.strictEqual(relay.enabled, true);
++ const down = await relay.flush();
++ assert.strictEqual(down.sent, 0);
++ assert.strictEqual(await relay.pending(), queued.length, 'Billing down: still queued');
++ assert.strictEqual(await relay.rejected(), 0);
++
++ up = true; clock += 10 * 60 * 1000;
++ const whileDown = posts.length;
++ await relay.flush();
++ await relay.flush();
++ assert.strictEqual(await relay.pending(), 0);
++ assert.deepStrictEqual([...billed.keys()].sort(), [...queued].sort());
++ assert.deepStrictEqual(posts.slice(whileDown).map((p) => p.body.idempotency_key), queued, 'each key posted once, in queue order');
++ assert.ok(posts.every((p) => p.url === 'http://billing.test/api/v1/usage' && p.auth === 'Bearer svc-token'));
++ const sentNow = posts.length;
++ await relay.flush();
++ assert.strictEqual(posts.length, sentNow, 'a later sweep sends nothing again');
++ await relay.stop();
++ });
++
++ await t.close();
++ done();
++})().catch((e) => { console.error(e); process.exit(1); });
++
++async function poll(fn, ms = 20000) {
++ const until = Date.now() + ms;
++ for (;;) {
++ if (await fn()) return;
++ if (Date.now() > until) throw new Error('condition not met in time');
++ await new Promise((r) => setTimeout(r, 25));
++ }
++}