openvibers/openvibe.run Commits

Run: jobs run — placement on Network's node offers (SDK plan), dispatch through Bot's jobs API, mirrored state/events/stdout; unplaceable jobs wait until their ttl (T14 steps 6-7). (#2)

OP OpenVibers committed 7 Oct 2026 · c8566807b4 · 19 files, +1639 −130

* Run: jobs run — placement on Network's node offers (SDK plan), dispatch through Bot's jobs API, mirrored state/events/stdout; unplaceable jobs wait until their ttl (T14 steps 6-7).

Everything passes. Final state of the work:

## Changes

**New**
- `server/dispatch/bot.js:50` — the Bot client (`send` :96 / `cancel` :103 / `state` :105); token minted with `openvibe-sdk/auth`'s `createServiceTokenClient` (audience `openvibe.bot`, scope `bot.job.dispatch`) from `OV_OAUTH_CLIENT_ID/SECRET` against `OV_NETWORK_INTERNAL_URL`, cached to 60 s before expiry, 401 → invalidate + retry once; 8 s timeout, no token/header/job body ever logged; 404 `bot.job_not_found` → `null`, other 4xx → `DispatchError` carrying Bot's own code, 5xx/timeout → retryable.
- `server/dispatch/placement.js:51` — offers read from Network's public registry (`fetch` :58, cached `RUN_OFFERS_TTL_MS`), each validated against `platform.resource-offer@1` (:75), then SDK `plan()` (:100) on the job's own `requirements` + `worker:<class>`; user-owned offers are dropped unless the requirements name them (:92-96); returns `node_id/provider/region/offer_id`.
- `server/jobs/poller.js:45` — the loop: ttl sweep + cancel-on-Bot (:144), placement and single send (:156) with the cancel-race guard (:200: a row that left `queued` while the frame was in flight is never placed, and Bot is told to stop it), adoption on `bot.job_id_reused` (:222), bounded 404 failure (:237), mirror + one `run.job.*` event + stdout (:245), one `job_cancel` per job (:323), `tick`/`start`/`stop` (:361/:396/:403).
- `test/dispatch.test.js` — 18 checks against a fake Bot that is a real HTTP server verifying Run's minted token (`bot.job.dispatch`, `svc:run`) and a stub registry.

**Changed**
- `server/dispatch/index.js:35` — the no-op seam replaced by the assembled bridge; `enabled` = Run can mint its Bot token; `place`/`cancel`/`poll`/`sweep`/`start`/`stop`/`status` kept.
- `server/index.js:42-43` — the poller is started by the process (nothing at module load), the two dead no-op intervals removed, boot log updated (:48).
- `server/config.js:113-120` — `RUN_POLL_MS`, `RUN_OFFERS_TTL_MS`, `RUN_DISPATCH_UNKNOWN_POLLS`, `RUN_DISPATCH_MAX_PER_TICK`; `RUN_JOBS_INTERVAL_MS`/`RUN_DISPATCH_INTERVAL_MS` gone.
- `server/jobs/store.js:177-183` — `waiting`/`active`/`dueTtl`; `:202` cancel marks a **placed** job `cancel_requested` instead of settling it (a node may hold it); `:231` `mirror` can no longer revive an end state.
- `server/api/v1.js:150` — submit no longer places eagerly (the poller owns it, so a submit never waits on Network/Bot); `:185` cancel reaches Bot for placed jobs too; `:209-227` `GET /jobs/:id/stream` also accepts a Bearer service token holding `run.job.stream` for the job's project.
- `server/app.js:57` — `fetchImpl` injected into the dispatcher; `server/observability.js:34` bot probe reason, `:49-52` `/api/ready` now carries `jobs_active` and the poller's status, `:61` `active()`.
- `test/helpers/app.js` — the stub Network gained `/oauth/token` (call-counted, so token reuse is assertable) and the offers registry (`t.setOffers`); `test/app.test.js:24,73` updated for the bridge's off state.
- Docs: README ("How a job runs" 1-5, cancel/usage/placement/security, env, tests), `.env.example`, `STATUS.json`, `CHANGELOG.md`, `docs/README.md`, `deploy/systemd/openvibe-run.service`.

## Tests

`nice -n 15 ov test --no-cache test/dispatch.test.js test/jobs.test.js test/app.test.js` → **3 passed in 18 s** (app 8.5 s, dispatch 4.7 s, jobs 4.5 s). Plus a boot smoke of `server/index.js` (PGlite + poller at 300 ms + unreachable Network and Bot): the poller ticked, the queued job stayed `queued` rather

* dispatch test: poll until the job is sent (waiting jobs from earlier checks can fill a tick).

* tests: no offers cache (RUN_OFFERS_TTL_MS=0), so t.setOffers takes effect at once.

* poller: a tick that finds another in flight waits for it instead of skipping (on PostgreSQL a test poll could land mid-tick).

Changes

ddiff --git a/.env.example b/.env.example
iindex e6e6717..54a774a 100644
---- a/.env.example
++++ b/.env.example
@@@ -37,8 +37,8 @@ OV_OAUTH_CLIENT_SECRET=
#EVENTS_RELAY_INTERVAL_MS=2000
# ── Jobs ────────────────────────────────────────────────────
++# Background jobs: the events relay and the dispatcher's poller. RUN_JOBS=off runs the API alone.
#RUN_JOBS=on
--#RUN_JOBS_INTERVAL_MS=5000
# ── Project caps on a job's limits (run.job-create-request@1); over one is 422 run.limits.exceeded ──
#RUN_LIMITS_MAX_WALL_MS=3600000
@@@ -58,10 +58,19 @@ OV_OAUTH_CLIENT_SECRET=
#RUN_STREAM_MAX_PER_JOB=64
#RUN_STREAM_HEARTBEAT_MS=25000
--# ── The dispatcher bridge to Bot (plan T14 step 6; a no-op seam until then) ──
++# ── The dispatcher bridge to Bot (plan T14 step 6, R1c) ──
++# Bot holds the node link and the meter; Run places each job on a node from the Fabric offers and mirrors
++# what Bot answers. Needs OV_OAUTH_CLIENT_SECRET above (Run mints its token for audience openvibe.bot);
++# without it nothing is sent and every job stays queued.
#RUN_BOT_URL=http://127.0.0.1:4630
#RUN_BOT_AUDIENCE=openvibe.bot
--#RUN_DISPATCH_INTERVAL_MS=1000
++#RUN_POLL_MS=1000
#RUN_DISPATCH_TIMEOUT_MS=8000
++# How long the offers Run places from are cached (one read serves a whole tick).
++#RUN_OFFERS_TTL_MS=5000
++# Polls Bot may answer 404 for a job Run sent before that job fails.
++#RUN_DISPATCH_UNKNOWN_POLLS=5
++# 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).
ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 6c6be5d..87d6b86 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -20,6 +20,25 @@ Notable changes to OpenVibe.Run. The service is unreleased; everything below is
monotonic `seq`, `Last-Event-ID`).
- `server/dispatch/index.js`: the dispatcher-bridge seam (plan T14 step 6) as a no-op, so every job stays `queued`.
++- The dispatcher bridge to OpenVibe.Bot (plan T14 step 6, R1c) and placement (step 7, R3): `server/dispatch/bot.js`
++ (the client — `send`/`cancel`/`state` with Run's Network client-credentials token for audience `openvibe.bot`
++ holding `bot.job.dispatch`, cached until 60 s before expiry), `server/dispatch/placement.js` (the Fabric offers
++ plus `openvibe-sdk/placement` `plan()`, with `user-owned` eligible only when the job's requirements name it) and
++ `server/jobs/poller.js` (`RUN_POLL_MS`, default 1000 ms): the ttl sweep, the send, the mirror of Bot's state with
++ 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.
++
++### Changed
++
++- Cancelling a `placed` job no longer settles it at once: a job Run has handed to Bot is marked `cancel_requested`,
++ Bot is asked for `job_cancel` and the job ends when Bot reports it — Run never says `cancelled` for work a node may
++ still be doing. A `queued` job (sent nowhere) is still cancelled at once.
++- `POST /api/v1/jobs` no longer places eagerly: the poller sends each queued job, so a submit never waits on Network
++ or Bot and a job whose lifetime ran out is never sent.
++- `RUN_JOBS_INTERVAL_MS`/`RUN_DISPATCH_INTERVAL_MS` are replaced by `RUN_POLL_MS` (the poller's own interval).
++- `usage.wall_ms`/`usage.seconds` are Bot's metered wall clock: 0 until the job's `job_exit` (Bot is the meter).
++
### Notes
- Run holds no Billing token: OpenVibe.Bot owns the node link and the per-second metering (plan T14 L1).
ddiff --git a/README.md b/README.md
iindex 4acff73..de82a62 100644
---- a/README.md
++++ b/README.md
@@@ -10,8 +10,33 @@ its end. A job belongs to a **project**, is billed per wall-clock second it runs
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).
--This repository is at plan **T14 step 5 (R1b)**: the API is complete and every job stays `queued`. The dispatcher
--bridge to Bot is plan T14 step 6 (R1c) and lives behind the seam in `server/dispatch/index.js`.
++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
++`server/index.js`) sends it to OpenVibe.Bot, mirrors Bot's state and the node's stdout back, and settles it. Without
++`OV_OAUTH_CLIENT_SECRET` Run cannot mint the Network token Bot requires, so the bridge is off and every job stays
++`queued`.
++
++## How a job runs
++
++1. **Submit** — `POST /api/v1/jobs` (capability `run.job.submit`). Run validates the request against
++ `run.job-create-request@1`, mints the `job_<ULID>` id, stores the row `queued` and writes `run.job.queued` in the
++ same transaction. Nothing leaves the process here: a submit never waits on Network or Bot.
++2. **Place** — the poller (`RUN_POLL_MS`, default 1000 ms) reads the Fabric's offers for the job's class
++ (`GET {OV_NETWORK_URL}/api/v1/offers?kind=node`), and `openvibe-sdk/placement`'s `plan()` chooses the node the
++ job's own `requirements` allow. Run records the node, provider and region on its row (`placed_at` included).
++ No eligible offer: the job waits queued (a node may be briefly offline) and every tick tries again; if its
++ `ttl_ms` runs out first it ends `expired` with `run.job.unplaceable` (ADR-036 section 2).
++3. **Send** — Run hands the `platform.job@1` body to Bot (`POST {RUN_BOT_URL}/api/v1/jobs`, capability
++ `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
++ 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.
++5. **End** — `succeeded`, `failed`, `cancelled` or `expired`, from Bot's `job_exit` (or from the cancel request and
++ the ttl sweep). An end state never changes. Bot's answer is never trusted over Run's own row: it can neither
++ rename the node nor the project of a job Run placed.
## Owns
@@@ -29,11 +54,19 @@ bridge to Bot is plan T14 step 6 (R1c) and lives behind the seam in `server/disp
`seq`, plus the two-minute single-use stream ticket (`server/jobs/ticket.js`).
- The `platform.workload-requirements@1` default (`{ kind: run.<class>, mobility: job, latency_class: background,
objective: balanced }`, the contract's own default) and the project caps on a job's limits.
++- Placement (plan T14 R3, `server/dispatch/placement.js`): the job's requirements are stated by Run and the choice
++ among offers is `openvibe-sdk/placement` `plan()` — Run writes no placer of its own (ADR-036). The job's
++ `worker:<class>` capability is what a candidate must advertise, and a user-owned offer is eligible only when the
++ job's requirements name `user-owned` (the contract's own rule; the SDK's fallback trust list would otherwise
++ admit it).
++- The poller (`server/jobs/poller.js`): the ttl sweep, the send, the mirror, the cancel-on-Bot and the job's stdout
++ into the stream.
## Does not own
--- The node link, the device credentials and the per-second metering: OpenVibe.Bot (`bot.job.dispatch`), which places
-- the job on a node advertising `worker:<class>`.
++- The node link, the device credentials and the per-second metering: OpenVibe.Bot (`bot.job.dispatch`), which gives
++ the job to a node advertising `worker:<class>` and meters every wall-clock second into `platform.usage-sample@1`
++ readings keyed `run:<job id>:<n>`.
- 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).
@@@ -43,10 +76,13 @@ bridge to Bot is plan T14 step 6 (R1c) and lives behind the seam in `server/disp
- PostgreSQL 18: `run_jobs` and the `run_events_outbox`; `migrations/` applies on boot.
- Valkey (`VALKEY_URL`, optional): shared state between processes; unset, nothing is shared.
--- OpenVibe.Network: the RS256 signing key (JWKS) every token is verified against; Run's client `run` mints the
-- events relay's service token.
++- OpenVibe.Network: the RS256 signing key (JWKS) every token is verified against; the offers Run places from
++ (`GET /api/v1/offers`, public); Run's client `run` mints the events relay's service token and its Bot token.
- OpenVibe.Events: `run.job.*` events go to `events.event.publish`; unset, they wait in `run_events_outbox`.
--- OpenVibe.Bot (plan T14 step 6): the dispatcher bridge the jobs wait for.
++- OpenVibe.Bot (`RUN_BOT_URL`, `RUN_POLL_MS`): the node link and the meter. Run sends each job to it
++ (`bot.job.dispatch`, audience `openvibe.bot`, minted from `OV_OAUTH_CLIENT_ID`/`OV_OAUTH_CLIENT_SECRET` against
++ `OV_NETWORK_INTERNAL_URL` and cached until 60 s before expiry) and mirrors its answer. Unset the client secret and
++ the bridge is off: `/api/ready` skips the `bot` check and every job stays `queued`.
- `openvibe-contracts` v0.106.0, `openvibe-sdk` v0.34.0 and `openvibe-shared` v2.11.0 (package.json).
## Capabilities and routes
@@@ -62,7 +98,7 @@ refused like any other token without the grant.
| `GET /api/v1/jobs?state=&limit=&cursor=` | `run.job.list` | `run.job-list-query@1` → `run.job-list-result@1` |
| `POST /api/v1/jobs/:id/cancel` | `run.job.cancel` | `common.no-body@1` → `run.job-read-result@1` |
| `POST /api/v1/jobs/:id/stream/ticket` | `run.job.stream` | `common.no-body@1` → `run.job-stream-ticket-result@1` |
--| `GET /api/v1/jobs/:id/stream?ticket=` | `run.job.stream` | a ticket (never a service token) → `text/event-stream` of `run.job-stream-event@1` |
++| `GET /api/v1/jobs/:id/stream?ticket=` | `run.job.stream` | a ticket in the query, **or** a Bearer service token with the grant → `text/event-stream` of `run.job-stream-event@1` |
| `GET /api/v1/admin/jobs` | `run.job.admin` | `run.job-list-query@1` (with `project_id`) → `run.job-list-result@1` |
`run.job.admin` additionally reads and cancels any project's job and is the only way to pass `project_id` to a list
@@@ -79,18 +115,21 @@ every answer before it is sent; errors are RFC 9457 `application/problem+json`.
reused. `limits` are capped by the project's (`RUN_LIMITS_*`, `422 run.limits.exceeded`) — the lifetime too, whether
named or defaulted — `args` over 512 KiB are refused `413 run.args_too_large`, and `ttl_ms` defaults to
`limits.wall_ms + 600000` exactly as the contract says.
--- **States.** `queued → placed → running → succeeded | failed | cancelled | expired`. Run moves a job to `queued`
-- and, on a cancel, to `cancelled`; the rest arrives from Bot (plan T14 step 6). An end state never changes, so a
-- replay of a cancel is a no-op rather than a second event.
++- **States.** `queued → placed → running → succeeded | failed | cancelled | expired`. Run stores `queued` and, once
++ the job is on a node, records `placed` itself; `running` and the end states arrive from Bot. An end state never
++ changes, so a replay of a cancel is a no-op rather than a second event.
- **Events.** `run.job.queued|started|succeeded|failed|cancelled|expired`, payloads being the
`run.job-read-result@1` projection their contracts describe. The outbox row is keyed `run:<job id>:<state>` (one
row per job and state, so a replay is one row); the published envelope carries a contract-valid `evt_<ULID>`
document id, because `events.event-envelope@1` pins that pattern and OpenVibe.Events validates it on publish.
- **Stream.** `POST /jobs/:id/stream/ticket` mints a two-minute single-use RS256 ticket for that one job (audience
`openvibe.run`, `typ run-stream`, its own key — never Network's signing key, so it is never a session anywhere).
-- `GET /jobs/:id/stream?ticket=…` replays what is kept (the last 1 MiB) and follows; `Last-Event-ID` resumes after
-- the last `seq` seen; an ended job replays what is kept and closes with `type: 'end'`. Reconnecting never re-runs a
-- job. The ring is this process's memory (reclaimed once it ended or went idle, and rebuilt from the job's row when
++ `GET /jobs/:id/stream?ticket=…` replays what is kept (the last 1 MiB) and follows; a service that can send a header
++ opens the same stream with a Bearer token holding `run.job.stream` for the job's project (the manifest allows
++ both). `Last-Event-ID` resumes after the last `seq` seen; an ended job replays what is kept and closes with
++ `type: 'end'`. Reconnecting never re-runs a job. The `output` events are the node's `job_stdout` chunks as Bot
++ holds them (best effort: a chunk lost with the node's link is never resent, and a gap Bot dropped is dropped here
++ rather than streamed twice). The ring is this process's memory (reclaimed once it ended or went idle, and rebuilt from the job's row when
a stream opens again), so `seq` is this process's counter: a resume across a restart gets the current state rather
than a gap, and output is best effort either way.
- **The `job` record.** `run_jobs.job` holds the `platform.job@1` body Run will send (id, class, artifact, args,
@@@ -98,28 +137,47 @@ every answer before it is sent; errors are RFC 9457 `application/problem+json`.
for: `requirements` (required by `run.job-read-result@1` and by placement), `placed_at` (`timings.placed_at`) and
`cancel_requested` (a cancel of a placed or running job). `store.frameOf(record)` builds the frame body from the
`platform.job@1` keys alone, and the dispatcher bridge validates it before anything is sent.
--- **Cancelling.** A queued or placed job is cancelled at once (`exit` stays null, usage stays 0); a running job is
-- marked `cancel_requested` and ends when its `job_exit` arrives (step 6). Cancelling an ended job answers it as it
-- is.
++- **Cancelling.** A **queued** job was never sent anywhere: cancelled at once (`exit` stays null, usage stays 0). A
++ **placed or running** job may be on a node this instant, so it is marked `cancel_requested`, Bot is asked for its
++ `job_cancel` (once per job — Bot repeats it on every reconnect until the job ends — and the poller retries the
++ 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.
++- **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
++ queued until an offer appears or its `ttl_ms` ends it `expired` with `run.job.unplaceable`. Offers are cached for `RUN_OFFERS_TTL_MS`; a Network that does not answer
++ is retried, never a reason to fail a job.
## Tests
```
npm test # every file in test/, PGlite (in-process PostgreSQL)
npm run test:pg # the same on the production-shaped containers (openvibe-sdk scripts/test-services.sh up)
--node test/jobs.test.js # the job API against the released contracts
--node test/app.test.js # the skeleton: health, readiness, release.json, /metrics, the dispatcher seam
++node test/jobs.test.js # the job API against the released contracts
++node test/dispatch.test.js # the dispatcher bridge: offers → placement → Bot → mirror → stream, cancel, ttl
++node test/app.test.js # the skeleton: health, readiness, release.json, /metrics, the bridge's off state
```
--`test/helpers/app.js` boots Run against an in-process OpenVibe.Network stub (its JWKS and its tokens), so nothing
--needs the network, OpenVibe.Bot or a running service.
++`test/helpers/app.js` boots Run against an in-process OpenVibe.Network stub (its JWKS, its tokens, the token
++endpoint Run mints from and the offers registry), so nothing needs the network, OpenVibe.Bot or a running service.
++`test/dispatch.test.js` runs a fake OpenVibe.Bot as a real HTTP server that verifies the Network service token Run
++mints for it (audience `openvibe.bot`, capability `bot.job.dispatch`), so the bridge is exercised over the wire
++rather than through a fetch stub.
## Security
- Every route checks one capability; the project and the requester come from the token, never the body.
- Another project's job is `404 run.job_not_found`, never `403`: an id's existence is never confirmed.
--- The stream route accepts only a valid, unexpired, unused ticket for that job — a service token in the query string
-- is refused like any other non-ticket.
++- The stream route accepts a valid, unexpired, unused ticket for that job, or a Bearer service token holding
++ `run.job.stream` for the job's project — a token in the query string is refused like any other non-ticket.
++- Bot is a trusted peer but never an authority: its answer cannot rename the node or the project of a job Run
++ placed, cannot move a job's state backwards, and cannot make Run carry out work the caller did not ask for. Run's
++ own row, its offers and the job's requirements decide.
++- Run's Bot token is minted from OV_OAUTH_CLIENT_ID/SECRET, cached until 60 s before expiry, never logged and never
++ put in an error; the job body is validated against `platform.job@1` before it is sent.
- Limits are bounded by the contract and the project's caps; nothing is silently lowered.
- No secret is logged (no `Authorization` header, no ticket, no job output), and nothing starts at module load.
- See SECURITY.md to report a problem.
ddiff --git a/STATUS.json b/STATUS.json
iindex 3da10ba..39d97ed 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -40,9 +40,13 @@
"the run.job.queued|started|succeeded|failed|cancelled|expired events written in the same transaction as the state change, the outbox row keyed run:<job id>:<state> (a replay is one row) with a contract-valid evt_ document id on the published envelope, relayed from run_events_outbox by the openvibe-sdk outbox",
"the job stream: a two-minute single-use RS256 stream ticket (audience openvibe.run, typ run-stream, its own key — never Network's) opens text/event-stream of run.job-stream-event@1, replayed from the last 1 MiB with monotonic seq as the SSE id and Last-Event-ID resume",
"keyset paging on (created_at, id) with next_cursor null on the last page; run.job.admin lists any project's jobs (project_id in the query) and reads and cancels any job",
-- "run_jobs (migrations/0001_jobs.sql) holds every job, its state, placement, timings, exit, result and the metered wall clock; jobs stay queued until the dispatcher bridge to Bot exists (plan T14 step 6, server/dispatch/index.js — a no-op seam today)",
++ "the dispatcher bridge to OpenVibe.Bot (plan T14 step 6, server/dispatch/index.js + server/jobs/poller.js): a queued job is placed on a node from the Fabric offers (server/dispatch/placement.js, openvibe-sdk/placement plan(), the job's own platform.workload-requirements@1, no eligible offer → run.job.unplaceable) and handed to Bot once (server/dispatch/bot.js POST /api/v1/jobs, Network client-credentials token audience openvibe.bot holding bot.job.dispatch, cached until 60 s before expiry, never logged); every placed or running job is mirrored from Bot's own row (state, placement, timings, exit, result, metered wall clock) by a poller on RUN_POLL_MS (default 1000 ms, started by server/index.js, nothing at module load), each state change emitting its run.job.* event once in the same transaction and feeding the job's SSE stream with the node's stdout as output events; a job Bot answers 404 for after Run sent it fails run.job.worker_failed after RUN_DISPATCH_UNKNOWN_POLLS polls, and Bot can neither rename the node or the project of a job Run placed nor move its state backwards",
++ "cancel reaches the node: a queued job is cancelled at once, a placed or running one is marked cancel_requested and Bot is asked for job_cancel (once per job, retried by the poller) — the job settles cancelled when Bot reports it",
++ "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)",
-- "truthful readiness (GET /api/ready: db required; valkey, Network JWKS, Events and Bot reported, never claimed) and the loopback-only /metrics gauges (jobs queued, active, ended, outbox backlog, streams)"
++ "truthful readiness (GET /api/ready: db required; valkey, Network JWKS, Events, Bot reachability and the poller's own status reported, never claimed) and the loopback-only /metrics gauges (jobs queued, active, ended, outbox backlog, streams)"
],
"updated": "2026-10-07"
}
ddiff --git a/deploy/systemd/openvibe-run.service b/deploy/systemd/openvibe-run.service
iindex bfa3b79..74faff0 100644
---- a/deploy/systemd/openvibe-run.service
++++ b/deploy/systemd/openvibe-run.service
@@@ -1,9 +1,10 @@
[Unit]
Description=OpenVibe.Run — the job service (jobs, placement, streaming; plan T14, ADR-034)
--# Network (sign-in, JWKS, OAuth) should be up first, and Run's own PostgreSQL (pooled runtime role plus
--# the direct owner role for migrations) and Valkey (state shared between processes) are required to
--# serve; the events relay waits in the outbox and the dispatcher bridge to Bot (plan T14 step 6) is a
--# no-op until it exists. Run degrades honestly without any of them: /api/ready says which (STATUS).
++# Network (sign-in, JWKS, OAuth for the job offers and the Bot token) should be up first, and Run's own
++# PostgreSQL (pooled runtime role plus the direct owner role for migrations) and Valkey (state shared
++# between processes) are required to serve; the events relay waits in the outbox and OpenVibe.Bot is
++# needed for a job to run at all (RUN_BOT_URL; without OV_OAUTH_CLIENT_SECRET jobs stay queued). Run
++# degrades honestly without any of them: /api/ready says which, and the poller's own status is in it.
After=network.target postgresql.service pgbouncer.service valkey-server.service
Wants=openvibe-network.service postgresql.service pgbouncer.service valkey-server.service
ddiff --git a/docs/README.md b/docs/README.md
iindex a7ee464..2da43c1 100644
---- a/docs/README.md
++++ b/docs/README.md
@@@ -7,9 +7,10 @@ capability manifests: `contracts/generated/openapi/run.json` once `manifests/ser
What does live here:
--- `protocol.md` — how Run and Bot talk when the dispatcher bridge lands (plan T14 step 6): the
-- `platform.job@1` body it sends, the `platform.job-frame@1` answers it mirrors, and the state mapping. Not written
-- yet: the bridge does not exist.
++The dispatcher bridge itself needs no page here: `server/dispatch/index.js` (the seam), `server/dispatch/bot.js`
++(Run → Bot over `bot.job.dispatch`), `server/dispatch/placement.js` (offers → `platform.placement-result@1`) and
++`server/jobs/poller.js` (the loop) carry their own headers, and the README's "How a job runs" is the end-to-end
++account. Bot's side is `OpenVibe.Bot/docs/protocol.md` (the device link and the job frames).
Nothing else is a document here. The README's route table is the API as it is served today, and `STATUS.json` is
what this repository is.
ddiff --git a/server/api/v1.js b/server/api/v1.js
iindex 8c0fe49..ba894f2 100644
---- a/server/api/v1.js
++++ b/server/api/v1.js
@@@ -12,7 +12,7 @@
* | GET /jobs?state=&limit=&cursor= | run.job.list | run.job-list-query@1 → run.job-list-result@1 |
* | POST /jobs/:id/cancel | run.job.cancel | common.no-body@1 → run.job-read-result@1 |
* | POST /jobs/:id/stream/ticket | run.job.stream | common.no-body@1 → run.job-stream-ticket-result@1 |
-- * | GET /jobs/:id/stream | run.job.stream | a ticket (never a service token) → text/event-stream of run.job-stream-event@1 |
++ * | GET /jobs/:id/stream | run.job.stream | a ticket in the query or a Bearer token → text/event-stream of run.job-stream-event@1 |
* | GET /admin/jobs | run.job.admin | run.job-list-query@1 (with project_id) → run.job-list-result@1 |
*
* The project and the requester always come from the token (claims.project_id, claims.sub), NEVER from the
@@@ -23,6 +23,11 @@
* Errors are RFC 9457 problem+json: 400 run.invalid_request (the body does not match its contract), 401
* token.required / ticket.*, 403 capability.denied / run.project_required, 404 run.job_not_found, 409
* run.idempotency.conflict, 413 run.args_too_large, 422 run.limits.exceeded / run.invalid_input.
++ *
++ * Cancelling is a request, not a state change of its own (plan T14 step 6): a queued job — sent to no node —
++ * is cancelled at once, while a placed or running one is marked cancel_requested, Bot is asked for its
++ * job_cancel (and asked again by the poller until the job ends), and the job reaches `cancelled` when Bot
++ * reports it. Run never says cancelled for work a node may still be doing.
*/
const express = require('express');
const contracts = require('openvibe-contracts');
@@@ -142,8 +147,8 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
if (made.created) await events.queued(t, made.row, { traceparent: req.ov && req.ov.traceparent });
return made;
});
-- // The job stays queued: the dispatcher bridge is plan T14 step 6 (server/dispatch/index.js).
-- if (out.created && dispatch.enabled) await dispatch.place(out.row);
++ // The job stays queued here; the dispatcher's poller places and sends it (server/dispatch/index.js
++ // → server/jobs/poller.js), so a submit never waits on Network or Bot.
if (out.created) stream.seed(out.row);
res.status(out.created ? 201 : 200).json({ created: out.created, job: store.readResult(out.row) });
}));
@@@ -175,8 +180,10 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
if (moved.cancelled) await events.cancelled(t, moved.row, { traceparent: req.ov && req.ov.traceparent }); // run:<id>:cancelled — a replay is the same row
return moved;
});
-- // A running job is only marked: the dispatcher bridge sends job_cancel and its job_exit ends it (step 6).
-- if (out.cancelRequested && dispatch.enabled) await dispatch.cancel(out.row);
++ // A placed or running job is only marked cancel_requested: Bot may already have it on a node, so the
++ // dispatcher sends job_cancel (here, and again on the next tick if this call did not land) and the
++ // job reaches `cancelled` when Bot reports it (step 6).
++ if (out.cancelRequested) await dispatch.cancel(out.row);
if (out.cancelled) stream.state(out.row.id, out.row.state, at);
res.json(store.readResult(out.row));
}));
@@@ -194,17 +201,34 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
}));
}));
-- // ── GET /jobs/:id/stream — run.job.stream, ticket only ─────────────────────
-- // A browser cannot send a header on an EventSource, and a service token must never ride in a query
-- // string (it would land in logs): this route accepts the two-minute ticket the POST above mints, and
-- // nothing else. The ticket names the job and the project, so the stream is scoped without any query.
++ // ── GET /jobs/:id/stream — run.job.stream: a ticket, or a Bearer service token ─────
++ // A browser cannot send a header on an EventSource, so the POST above mints a two-minute ticket and this
++ // route accepts it in the query. A service that can send a header uses its own Network service token
++ // holding run.job.stream instead (the manifest: "GET /api/v1/jobs/:id/stream?ticket= (or with a Bearer
++ // token)"); a token is never accepted in the query string, where it would land in logs.
r.get('/jobs/:id/stream', wrap(async (req, res) => {
const token = Array.isArray(req.query.ticket) ? null : req.query.ticket;
-- if (token == null || token === '') fail(401, 'ticket.required', 'open the stream with ?ticket= from POST /api/v1/jobs/:id/stream/ticket');
-- const verified = tickets.verify(token); // single use: the jti is consumed here, so one ticket opens one stream
-- if (!verified.ok) fail(401, verified.code, verified.reason);
-- if (verified.claims.sub !== req.params.id) fail(403, 'run.stream.ticket_mismatch', 'this ticket is for another job');
-- const row = await store.get(db, req.params.id, verified.claims.project);
++ const bearer = String(req.headers.authorization || '').startsWith('Bearer ');
++ let row = null;
++ if (token) {
++ const verified = tickets.verify(token); // single use: the jti is consumed here, so one ticket opens one stream
++ if (!verified.ok) fail(401, verified.code, verified.reason);
++ if (verified.claims.sub !== req.params.id) fail(403, 'run.stream.ticket_mismatch', 'this ticket is for another job');
++ row = await store.get(db, req.params.id, verified.claims.project);
++ } else if (bearer) {
++ // The same capability the other routes' guards enforce, checked here because this route's other
++ // door is a query string a middleware cannot see.
++ const p = req.principal;
++ if (p.kind !== 'service') fail(403, 'capability.denied', 'run.job.stream is granted to service principals only');
++ const decision = contracts.capabilities.check(p.claims, 'run.job.stream');
++ if (!decision.allowed) fail(403, decision.code || 'capability.denied', decision.reason);
++ const project = apiAuth.projectOf(p);
++ if (!project) fail(403, 'run.project_required', 'this token names no project: a stream is scoped to the project the token names');
++ if (!isJobId(req.params.id)) fail(404, 'run.job_not_found', 'no such job');
++ row = await store.get(db, req.params.id, project); // another project's job: 404, never 403
++ } else {
++ fail(401, 'ticket.required', 'open the stream with ?ticket= from POST /api/v1/jobs/:id/stream/ticket, or with a Bearer token holding run.job.stream');
++ }
if (!row) fail(404, 'run.job_not_found', 'no such job');
if (!stream.canAttach(row.id)) fail(503, 'run.stream.busy', 'too many streams on this job; try again shortly');
ddiff --git a/server/app.js b/server/app.js
iindex 275fc8e..184f972 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -51,9 +51,10 @@ function createApp(opts = {}) {
const events = opts.events || createRunEvents({ outbox, now, log });
const stream = opts.stream || createJobStream({ config, now, log });
const tickets = opts.tickets || createStreamTickets({ config, now, log });
-- // The dispatcher bridge to Bot: a no-op until plan T14 step 6 (server/dispatch/index.js), so a job
-- // stays queued and no node ever sees it.
-- const dispatch = opts.dispatch || createDispatcher({ config, db, store: require('./jobs/store'), events, stream, log, now });
++ // The dispatcher bridge to Bot (plan T14 step 6, server/dispatch/index.js): placement from the Fabric
++ // offers and the job loop that sends, mirrors and cancels. It is off without OV_OAUTH_CLIENT_SECRET —
++ // Run cannot mint the Bot token — and nothing runs until server/index.js calls its start().
++ const dispatch = opts.dispatch || createDispatcher({ config, db, store: require('./jobs/store'), events, stream, log, now, fetchImpl });
const apiAuth = createApiAuth({ config, keys, userAuth });
const release = require('openvibe-shared/release').createRelease({ service: 'run', root: path.join(__dirname, '..') });
ddiff --git a/server/config.js b/server/config.js
iindex 2213572..6b96f99 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -71,9 +71,10 @@ function loadConfig(env = process.env) {
intervalMs: int(env.EVENTS_RELAY_INTERVAL_MS, 2000),
},
++ // The background jobs (the events relay and the dispatcher poller); RUN_JOBS=off runs the API alone,
++ // which is what tests do (they drive dispatch.poll() by hand).
jobs: {
enabled: env.RUN_JOBS !== 'off',
-- intervalMs: int(env.RUN_JOBS_INTERVAL_MS, 5000),
},
// What a project's token may ask for (run.job-create-request@1 $defs.limits): a request over one of
@@@ -103,13 +104,20 @@ function loadConfig(env = process.env) {
},
// The dispatcher bridge to Bot (plan T14 step 6, R1c): Bot holds the node link and the per-second
-- // metering; Run calls it as the service principal `svc:run` for audience openvibe.bot. Until step 6
-- // server/dispatch/index.js is a no-op and every job stays queued.
++ // metering; Run calls it as the service principal `svc:run` for audience openvibe.bot. Without
++ // OV_OAUTH_CLIENT_SECRET Run cannot mint that token and nothing is sent (jobs stay queued).
dispatch: {
botUrl: trim(env.RUN_BOT_URL || 'http://127.0.0.1:4630'),
botAudience: env.RUN_BOT_AUDIENCE || 'openvibe.bot',
-- intervalMs: int(env.RUN_DISPATCH_INTERVAL_MS, 1000),
++ // One poll a second: place what is queued, mirror what is on a node (server/jobs/poller.js).
++ pollMs: Math.max(100, int(env.RUN_POLL_MS, 1000)),
timeoutMs: int(env.RUN_DISPATCH_TIMEOUT_MS, 8000),
++ // The offers Run places from are cached this long, so one tick places many jobs off one read.
++ offersTtlMs: Math.max(0, int(env.RUN_OFFERS_TTL_MS, 5000)),
++ // Polls Bot answers 404 for a job Run sent before that job fails (never mirrored forever).
++ unknownPolls: Math.max(1, int(env.RUN_DISPATCH_UNKNOWN_POLLS, 5)),
++ // Jobs one tick places and mirrors; the next tick takes the rest.
++ maxPerTick: Math.max(1, int(env.RUN_DISPATCH_MAX_PER_TICK, 200)),
},
};
}
ddiff --git a/server/dispatch/bot.js b/server/dispatch/bot.js
nnew file mode 100644
iindex 0000000..f37c5d9
---- /dev/null
++++ b/server/dispatch/bot.js
@@@ -0,0 +1,110 @@
++'use strict';
++
++/**
++ * 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).
++ *
++ * 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
++ * state(jobId) GET …/api/v1/jobs/{id} → { job, stdout } | null
++ *
++ * Bot's answer's `job` is its own run_jobs record (id, node_id, state, sent_at, started_ms, finished_at,
++ * exit_reason, exit_code, wall_ms, usage_read, result, project_id, subject, provider, fault_code) and
++ * `stdout` is the last 1 MiB it holds of the job's output. None of it is trusted as authority: the poller
++ * validates it against Run's own row (server/jobs/poller.js) and never lets Bot name another node or
++ * project.
++ *
++ * The token is a Network client-credentials service token for audience `openvibe.bot` holding
++ * `bot.job.dispatch` (contracts/manifests/capabilities/bot.job.dispatch.json), minted from
++ * OV_OAUTH_CLIENT_ID/OV_OAUTH_CLIENT_SECRET against OV_NETWORK_INTERNAL_URL and cached until 60 s before
++ * expiry by openvibe-sdk/auth's createServiceTokenClient; a 401 from Bot drops the cached token and the
++ * call is tried once more. Unset client secret: the bridge is off and nothing is sent. The token, the
++ * Authorization header and the job body are never logged, and never put in an error.
++ *
++ * Errors: a DispatchError carrying Bot's own RFC 9457 code (409 bot.class_unadvertised, …) is permanent;
++ * a transport failure, a timeout or a 5xx is `retryable` (the poller tries again next tick); a 404 is
++ * `null` — Bot does not know the job (bot.job_not_found), which for state() is how the poller learns a job
++ * it sent has disappeared.
++ */
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++
++/** Bot's own capability this client holds (contracts/manifests/capabilities/bot.job.dispatch.json). */
++const BOT_DISPATCH = 'bot.job.dispatch';
++
++class DispatchError extends Error {
++ constructor(status, code, detail, { retryable = false } = {}) {
++ super(detail || code);
++ this.name = 'DispatchError';
++ this.status = status;
++ this.code = code;
++ this.detail = detail;
++ this.retryable = retryable;
++ }
++}
++
++/**
++ * createBotClient({ config, fetchImpl, now, log }) → the three calls plus status().
++ * enabled is false without OV_OAUTH_CLIENT_SECRET: Run cannot mint a Bot token, so nothing is sent.
++ */
++function createBotClient({ config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console } = {}) {
++ const { botUrl, botAudience, timeoutMs } = config.dispatch;
++ const { clientId, clientSecret } = config.oauth;
++ const tokenUrl = `${config.network.internalUrl}/oauth/token`;
++ const tokens = clientSecret
++ ? createServiceTokenClient({ tokenUrl, clientId, clientSecret, audience: botAudience, scope: BOT_DISPATCH, fetch: fetchImpl, now })
++ : null;
++ const enabled = Boolean(tokens && botUrl);
++ const note = enabled
++ ? `the dispatcher bridge to Bot: ${botUrl}, audience ${botAudience}, capability ${BOT_DISPATCH}`
++ : 'the dispatcher bridge to Bot is off: OV_OAUTH_CLIENT_SECRET is unset, so Run cannot mint its Network client-credentials token';
++
++ /** One call: mint (or reuse) the token, fetch, and turn Bot's answer into a value or a DispatchError. */
++ async function call(method, path, { body } = {}) {
++ if (!enabled) throw new DispatchError(503, 'run.dispatch.disabled', note, { retryable: true });
++ for (let attempt = 0; ; attempt++) {
++ const headers = { Accept: 'application/json', ...(body !== undefined ? { 'Content-Type': 'application/json' } : {}) };
++ try { Object.assign(headers, await tokens.authHeaders()); }
++ catch (e) { throw new DispatchError(503, 'run.dispatch.no_token', 'Run could not mint its Bot token from Network', { retryable: true }); }
++ let res;
++ try {
++ res = await fetchImpl(`${botUrl}${path}`, {
++ method, headers, ...(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', `Bot did not answer (${why})`, { retryable: true });
++ }
++ // A rotated or clock-skewed token Bot no longer accepts: drop the cache and try once more.
++ if (res.status === 401 && attempt === 0) { tokens.invalidate(); continue; }
++ const doc = await res.json().catch(() => null);
++ if (res.ok) return doc || {};
++ const code = doc && typeof doc.code === 'string' ? doc.code : null;
++ const detail = doc && typeof doc.detail === 'string' ? doc.detail.slice(0, 300) : null;
++ if (res.status === 404 && code === 'bot.job_not_found') return null;
++ const why = `Bot answered HTTP ${res.status}${code ? ` ${code}` : ''}${detail ? `: ${detail}` : ''}`;
++ if (res.status >= 500) throw new DispatchError(503, code || 'run.dispatch.unavailable', why, { retryable: true });
++ throw new DispatchError(502, code || 'run.dispatch.refused', why, { retryable: false });
++ }
++ }
++
++ return {
++ enabled,
++ note,
++ /** Hand one job to one device (POST /api/v1/jobs); Bot validates the platform.job@1 body itself. */
++ async send({ nodeId, job, project, subject = null, provider = null }) {
++ if (!nodeId) throw new DispatchError(422, 'run.dispatch.no_node', 'send: Bot needs a node_id (a device) to place the job on');
++ const out = await call('POST', '/api/v1/jobs', { body: { node_id: nodeId, job, project_id: project, subject, provider } });
++ if (!out || !out.job || typeof out.job.id !== 'string') throw new DispatchError(502, 'run.dispatch.refused', 'Bot answered without a job record');
++ return { job: out.job, sent: Boolean(out.sent) };
++ },
++ /** Ask Bot to stop a job; null when Bot has no such job (nothing to stop). Idempotent at Bot. */
++ cancel: (jobId) => call('POST', `/api/v1/jobs/${encodeURIComponent(jobId)}/cancel`),
++ /** Bot's own record of a job and the stdout it holds; null when Bot has never seen it. */
++ state: (jobId) => call('GET', `/api/v1/jobs/${encodeURIComponent(jobId)}`),
++ status: () => ({ enabled, note, bot_url: botUrl, audience: botAudience, capability: BOT_DISPATCH, timeout_ms: timeoutMs, token_url: tokenUrl }),
++ };
++}
++
++module.exports = { createBotClient, DispatchError, BOT_DISPATCH };
ddiff --git a/server/dispatch/index.js b/server/dispatch/index.js
iindex 16ecca0..cacfb68 100644
---- a/server/dispatch/index.js
++++ b/server/dispatch/index.js
@@@ -1,66 +1,84 @@
'use strict';
/**
-- * The dispatcher bridge Run → Bot — SEAM ONLY: plan T14 step 6 (R1c) builds this, and every method here
-- * is a no-op until then. Run accepts jobs, scopes them to a project and queues them (plan T14 step 5,
-- * R1b); it does not yet hand them to a node, so every job stays `queued` and no node ever sees it.
-- *
-- * Bot owns the node link and the per-second metering (OpenVibe.Bot server/jobs: dispatch(), cancel(),
-- * createJobFrames(); migrations/0004_run_jobs.sql). Step 6 replaces this module with the real bridge:
++ * The dispatcher bridge Run → Bot (plan T14 step 6, R1c; step 7, R3). Run accepts jobs, scopes them to a
++ * project and queues them (plan T14 step 5, R1b); this module is what makes a queued job run on a node:
*
* place(row) the platform.job@1 body — store.frameOf(row.job), the job record's own platform keys,
-- * validated against platform.job@1, never the requirements/placed_at/cancel_requested Run
-- * keeps beside it — sent to Bot with Run's service token `svc:run` for audience
-- * openvibe.bot holding bot.job.dispatch, which places it on a node advertising
-- * worker:<class>
-- * poll() what Bot answered since the last poll; store.mirror() copies state, placement, timings,
-- * exit, result and the metered wall clock back, store.settle() reaches the end state, and
-- * jobs/events.js emits run.job.started|succeeded|failed|cancelled|expired in the same
-- * transaction; stream.js carries the job's stdout
-- * cancel(row) a job whose cancel_requested marker is set: Bot's job_cancel
-- * sweep() ttl_ms ran out before placement: settle expired with run.job.unplaceable or run.job.ttl
++ * never the requirements/placed_at/cancel_requested Run keeps beside it — is placed on a
++ * node from OpenVibe.Network's offers (server/dispatch/placement.js, openvibe-sdk/placement
++ * plan()) advertising worker:<class>, and sent to Bot with Run's service token `svc:run` for
++ * audience openvibe.bot holding bot.job.dispatch (server/dispatch/bot.js). Run records the
++ * node, provider and region on its own row (store.mirror).
++ * poll() one tick of the loop (server/jobs/poller.js, RUN_POLL_MS): the ttl sweep, the placement of
++ * every queued job, and the mirror of every placed or running job — state, placement,
++ * timings, exit, result and the metered wall clock — with run.job.started|succeeded|failed|
++ * cancelled|expired emitted once each (jobs/events.js, in the state change's transaction) and
++ * the job's stdout carried into stream.js.
++ * cancel(row) a job whose cancel_requested marker is set: Bot's job_cancel, once per job (Bot repeats it
++ * on every reconnect until the job ends). The job reaches `cancelled` when Bot says so.
++ * 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.
*
-- * The interface is fixed here so server/index.js and the API already call it the way the bridge will be
-- * called; only this file changes in step 6. The seam is never enabled by configuration: RUN_BOT_URL and
-- * RUN_BOT_AUDIENCE (server/config.js) are read only by the real bridge.
++ * `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 runtime class a job needs a node to advertise (platform.resource-offer@1 `worker:<class>`). */
--const workerCapability = (runtimeClass) => `worker:${runtimeClass}`;
++const { createBotClient } = require('./bot');
++const { createPlacement, workerCapability } = require('./placement');
++const { createPoller } = require('../jobs/poller');
/**
-- * createDispatcher({ config, db, store, events, stream, log, now }) → the dispatcher interface.
-- * `enabled` is false until plan T14 step 6: nothing is sent to a node, nothing is mirrored.
++ * 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.
*/
--function createDispatcher({ config = null, db = null, store = null, events = null, stream = null, log = console, now = () => Date.now() } = {}) {
-- const enabled = false;
-- const note = 'the dispatcher bridge to Bot is not built yet (plan T14 step 6, R1c): every job stays queued';
++function createDispatcher({ config = null, db = null, store = null, events = null, stream = null, log = console,
++ now = () => Date.now(), fetchImpl = globalThis.fetch, bot = 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)`;
-- /** Send a queued job to a node. → { placed, reason }. No-op until step 6. */
++ /** Send a queued job to a node. → { placed, reason }. Never throws: the poller retries. */
async function place(row) {
-- return { placed: false, reason: 'dispatcher.noop', job_id: row && row.id, worker: row && workerCapability(row.class) };
++ if (!enabled) return { placed: false, reason: 'dispatch.disabled', job_id: row && row.id, worker: row && workerCapability(row.class) };
++ try {
++ const answer = await poller.placeOne(row, now());
++ // placed means a node holds the job now (a job that ended here — no offer, a refusal — has none).
++ return { placed: Boolean(answer.node_id), reason: answer.reason || answer.error || null, job_id: answer.job_id, node_id: answer.node_id || null, worker: workerCapability(row.class) };
++ } catch (e) {
++ log.warn(`[Run] job ${row && row.id}: placement failed (${e.code || e.message})`);
++ return { placed: false, reason: e.code || 'dispatch.error', job_id: row && row.id, worker: row && workerCapability(row.class) };
++ }
}
/** Cancel a placed or running job on its node (after the API marked cancel_requested). → { cancelled, reason }. */
async function cancel(row) {
-- return { cancelled: false, reason: 'dispatcher.noop', job_id: row && row.id };
++ if (!enabled) return { cancelled: false, reason: 'dispatch.disabled', job_id: row && row.id };
++ const out = await poller.cancelNow(row);
++ return { ...out, job_id: row && row.id };
}
-- /** What Bot answered since the last poll. → [ { job_id, state, … } ]. Empty until step 6. */
-- async function poll() { return []; }
--
-- /** The ttl sweep: jobs whose ttl_ms ran out before placement (run.job.unplaceable / run.job.ttl). → [ { job_id, state } ]. */
-- async function sweep() { return []; }
--
-- function start() { /* step 6 starts the poller here; nothing runs at module load */ }
-- function stop() {}
++ const poll = () => (enabled ? poller.tick() : Promise.resolve([]));
++ const sweep = () => (enabled ? poller.sweep() : Promise.resolve([]));
return {
-- enabled, note, place, cancel, poll, sweep, start, stop,
-- status: () => ({ enabled, note, bot_url: config ? config.dispatch.botUrl : null, audience: config ? config.dispatch.botAudience : null }),
++ enabled, note, place, cancel, poll, sweep,
++ start: () => poller.start(),
++ stop: () => poller.stop(),
++ status: () => ({
++ enabled, note,
++ bot_url: config ? config.dispatch.botUrl : null,
++ audience: config ? config.dispatch.botAudience : null,
++ 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 },
++ deps: { config, db, store, events, stream, log, bot: botClient, placement: placer, poller },
};
}
ddiff --git a/server/dispatch/placement.js b/server/dispatch/placement.js
nnew file mode 100644
iindex 0000000..067b0f7
---- /dev/null
++++ b/server/dispatch/placement.js
@@@ -0,0 +1,120 @@
++'use strict';
++
++/**
++ * Placement — which node runs a job (plan T14 step 7, R3): Run states the job's
++ * platform.workload-requirements@1 and reads Fabric offers; openvibe-sdk/placement's plan() does the
++ * choosing (ADR-036: "Run writes no placer of its own. A service states requirements and reads results; it
++ * does not rank offers.").
++ *
++ * choose(row) → { node_id, provider, region, offer_id, trust, reasons } | null
++ *
++ * The candidate offers come from OpenVibe.Network's public resource registry
++ * (GET {OV_NETWORK_URL}/api/v1/offers?kind=node), where a paired Node's runtime classes are published as
++ * platform.resource-offer@1 with the reserved `worker:<class>` capability (ADR-036 §3: "device node_id →
++ * resource-offer.node_id"), so an offer's node_id is the device Bot dispatches to. The public projection
++ * leaves `capacity` out (Network redacts it), so the `worker:<class>` free-slot check inside plan() cannot
++ * fire here; Run holds no network.resource.report grant and must not invent one — see For Opus in the PR.
++ *
++ * The trust policy is the job's own: plan() filters on requirements.trust, kind, latency_class, objective,
++ * residency and resources, and Run only adds the job's `worker:<class>` capability (a candidate must
++ * advertise the class the job runs). One rule Run applies before the placer sees anything, because the
++ * released contract states it and the SDK's current default does not: **user-owned is never a default**
++ * (platform.workload-requirements@1: "when absent, first-party, partner, community and external"). plan()
++ * falls back to its TRUST_ORDER, which lists user-owned, so a job that names no trust would otherwise be
++ * 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).
++ *
++ * 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
++ * answer with no eligible candidate is null, which the poller settles run.job.unplaceable.
++ */
++const contracts = require('openvibe-contracts');
++const { plan } = require('openvibe-sdk/placement');
++const { requirementsDefault } = require('../util');
++
++/** The reserved Fabric capability of a runtime class (platform.runtime-class@1; ADR-046). */
++const workerCapability = (runtimeClass) => `worker:${runtimeClass}`;
++
++class PlacementError extends Error {
++ constructor(code, detail, { retryable = true } = {}) {
++ super(detail || code);
++ this.name = 'PlacementError';
++ this.code = code;
++ this.detail = detail;
++ this.retryable = retryable;
++ }
++}
++
++const obj = (v) => (v && typeof v === 'object' && !Array.isArray(v) ? v : {});
++
++function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console } = {}) {
++ const { url } = config.network;
++ const offersUrl = `${url}/api/v1/offers?kind=node`;
++ const { offersTtlMs, timeoutMs } = config.dispatch;
++ let cache = { at: 0, offers: [], fetched: 0 };
++
++ /** The published node offers, cached briefly. A Network that does not answer is a retryable error. */
++ async function offers() {
++ if (cache.at && now() - cache.at < offersTtlMs) return cache.offers;
++ let res;
++ try {
++ res = await fetchImpl(offersUrl, { 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})`);
++ }
++ if (!res.ok) throw new PlacementError('run.dispatch.no_offers', `Network answered HTTP ${res.status} for the resource offers`);
++ const doc = await res.json().catch(() => null);
++ const list = doc && Array.isArray(doc.offers) ? doc.offers : null;
++ if (!list) throw new PlacementError('run.dispatch.no_offers', 'the resource offers answer carries no offers array');
++ const good = [];
++ for (const offer of list) {
++ // Validate what Network published against the released contract before plan() ever sees it; an
++ // offer that does not match is dropped, never placed on.
++ if (!contracts.validate('platform.resource-offer@1', offer).valid) {
++ log.warn(`[Run] offer ${obj(offer).offer_id || '?'} does not match platform.resource-offer@1: dropped`);
++ continue;
++ }
++ good.push(offer);
++ }
++ cache = { at: now(), offers: good, fetched: good.length };
++ return good;
++ }
++
++ /**
++ * The node to run this job on, or null when no offer is eligible. The requirements are the job's own
++ * (the contract's default was filled in at submit), plus the class it runs.
++ */
++ async function choose(row) {
++ 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'
++ && typeof o.node_id === 'string' && o.node_id.length > 0
++ && (o.capabilities || []).includes(want)
++ && (o.trust !== 'user-owned' || userOwnedNamed));
++ if (!candidates.length) return null;
++ const req = { ...requirements, capabilities: [...new Set([...(requirements.capabilities || []), want])] };
++ contracts.assertValid('platform.workload-requirements@1', req);
++ const result = plan(req, candidates, { now: now() });
++ const chosen = result.selected ? candidates.find((o) => o.offer_id === result.selected) : null;
++ if (!chosen) {
++ if (log && log.debug) log.debug(`[Run] job ${row.id}: no eligible offer for ${want} (${result.reasons.join('; ')})`);
++ return null;
++ }
++ return {
++ node_id: chosen.node_id, provider: chosen.provider ?? null, region: chosen.region ?? null,
++ offer_id: chosen.offer_id, trust: chosen.trust, reasons: result.reasons,
++ };
++ }
++
++ return {
++ choose,
++ offers,
++ workerCapability,
++ status: () => ({ offers_url: offersUrl, cached: cache.fetched, fetched_at: cache.at ? new Date(cache.at).toISOString() : null, ttl_ms: offersTtlMs }),
++ };
++}
++
++module.exports = { createPlacement, PlacementError, workerCapability };
ddiff --git a/server/index.js b/server/index.js
iindex cdb6a4c..3c7f3aa 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -10,9 +10,10 @@
* refresher, the stream key, the events outbox relay and the dispatcher poller all start here, and the
* signal handlers stop them in order.
*
-- * Background jobs (RUN_JOBS=off disables them): the events outbox relay and the job loop. The job loop
-- * calls the dispatcher's poll() and sweep() — a no-op until the dispatcher bridge to Bot exists (plan T14
-- * step 6, server/dispatch/index.js) — so today every job stays queued and no node ever sees one.
++ * 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.
*/
const { loadConfig } = require('./config');
const { openDb, migrate } = require('./db');
@@@ -34,23 +35,17 @@ async function main() {
keys.start();
const ticketKey = tickets.start();
-- const timers = [];
if (config.jobs.enabled) {
-- const every = (ms, fn) => { const t = setInterval(() => { Promise.resolve().then(fn).catch((e) => console.warn('[Run] job:', e.message)); }, ms); t.unref(); timers.push(t); };
-- // The dispatcher bridge is a no-op until plan T14 step 6: poll() and sweep() answer [] and nothing
-- // changes, but the loop is wired the way the real bridge will run it.
-- every(config.dispatch.intervalMs, async () => {
-- const answers = await dispatch.poll();
-- if (answers.length) console.warn(`[Run] ${answers.length} dispatcher answer(s) ignored: ${dispatch.note}`);
-- });
-- every(config.jobs.intervalMs, () => dispatch.sweep());
++ // The dispatcher's poller owns its own interval (RUN_POLL_MS): one tick sweeps the ttl, places the
++ // queued jobs and mirrors every placed or running one (server/jobs/poller.js). Off without Bot
++ // credentials: start() then does nothing and every job stays queued.
outbox.start();
dispatch.start();
}
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 ? `→ ${config.dispatch.botUrl}` : 'no-op until plan T14 step 6 (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 → ${config.dispatch.botUrl} (poll ${config.dispatch.pollMs} ms)` : 'off: no OV_OAUTH_CLIENT_SECRET, jobs stay queued'}`);
});
server.keepAliveTimeout = 65_000;
@@@ -65,7 +60,6 @@ async function main() {
deadlineMs: 5000,
deadlineExitCode: 0,
stop: [
-- () => timers.forEach(clearInterval),
() => keys.stop(),
() => tickets.stop(),
() => dispatch.stop(),
ddiff --git a/server/jobs/poller.js b/server/jobs/poller.js
nnew file mode 100644
iindex 0000000..9a30742
---- /dev/null
++++ b/server/jobs/poller.js
@@@ -0,0 +1,432 @@
++'use strict';
++
++/**
++ * The job loop (plan T14 step 6, R1c; RUN_POLL_MS, default 1000). One tick does four things, and nothing
++ * here runs at module load — server/index.js starts it (server/dispatch/index.js owns the instance):
++ *
++ * sweep a non-terminal job whose ttl_ms ran out reaches `expired` (run.job.ttl when a node may hold it,
++ * run.job.unplaceable when it never left `queued`), and Bot is asked to stop it
++ * place a `queued` job is placed — the node comes from OpenVibe.Network's offers through
++ * openvibe-sdk/placement (server/dispatch/placement.js) — and sent to Bot once (server/dispatch/
++ * bot.js; Bot's dispatch is idempotent by job id, so a retry after a crash is safe). No eligible
++ * offer fails the job run.job.unplaceable; a class the chosen node does not advertise fails it
++ * run.job.worker_failed (Bot 409 bot.class_unadvertised)
++ * mirror every `placed` or `running` job is mirrored from Bot: state, placement, timings, exit, result
++ * and the metered wall clock. A state that moves emits its run.job.* event once (the outbox row
++ * key run:<id>:<state> is what makes a replay one row) and updates the job's SSE stream
++ * cancel a job whose cancel_requested marker is set gets one job_cancel to Bot (Bot repeats it on every
++ * reconnect until the job ends); the job reaches `cancelled` when Bot's answer says so
++ *
++ * What is deliberately not trusted: Bot's answer never names the node, the project or the job id Run keeps
++ * (a mismatch is refused, not adopted), and a state Bot reports behind the one Run already holds is
++ * ignored. A job Bot answers 404 for after Run sent it fails after RUN_DISPATCH_UNKNOWN_POLLS polls — it
++ * is never mirrored forever.
++ *
++ * 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');
++
++/** 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 };
++const END_STATES = new Set(['succeeded', 'failed', 'cancelled', 'expired']);
++const STATES = ['queued', 'placed', 'running', 'succeeded', 'failed', 'cancelled', 'expired'];
++
++const obj = (v) => (v && typeof v === 'object' && !Array.isArray(v) ? v : {});
++const cancelling = (row) => obj(row.job).cancel_requested === true;
++
++/**
++ * 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.
++ */
++function createPoller({ config, db, store, events, stream, bot, 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
++ const warned = new Map(); // job id → when its placement problem was last logged
++ let noOfferWarnedAt = 0;
++ // 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();
++ 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 };
++ let timer = null;
++ let busy = false;
++ let inflight = null; // the tick in progress, so a caller can wait for it instead of skipping
++
++ const note = (id, state, extra = {}) => ({ job_id: id, state, ...extra });
++
++ /**
++ * A placement problem repeats every tick for as long as it lasts (a Network or a Bot that is down with
++ * queued jobs waiting), so it is logged once a minute per job rather than once a second per job. What is
++ * skipped on the log is still in the answer's `reason` and in status().last_error.
++ */
++ function warnSlowly(row, at, message) {
++ if (at - (warned.get(row.id) || 0) < 60_000) return;
++ warned.set(row.id, at);
++ log.warn(message);
++ }
++
++ /** 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 }) {
++ 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);
++ return s;
++ });
++ if (settled.changed) {
++ if (state === 'failed') stat.failed++;
++ if (state === 'expired') stat.expired++;
++ forget(row.id);
++ stream.state(row.id, state, at);
++ }
++ return settled;
++ }
++
++ const fail = (row, at, errorCode, detail) => end(row, at, { state: 'failed', errorCode, errorDetail: detail ? String(detail).slice(0, 1024) : null })
++ .then((s) => note(row.id, 'failed', { changed: s.changed, error: errorCode }));
++
++ /** 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); }
++
++ /**
++ * 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
++ * which first_seq that count belongs to); a slide that no longer lines up is re-anchored on the tail Run
++ * last sent, and a gap Bot dropped is dropped here too — never streamed twice.
++ */
++ function stdoutOf(row, ring, at) {
++ if (!ring || typeof ring.text !== 'string' || ring.last_seq == null) return;
++ const cur = cursors.get(row.id) || { seq: 0, chars: 0, first: null, tail: '' };
++ if (ring.last_seq <= cur.seq) return;
++ let chunk;
++ if (!cur.seq) chunk = ring.text; // first look: all Bot holds
++ else if (ring.first_seq === cur.first) chunk = ring.text.slice(cur.chars); // the window did not move
++ else if (ring.first_seq == null || ring.first_seq > cur.seq) chunk = ring.text; // all of it is new
++ else {
++ const i = cur.tail ? ring.text.lastIndexOf(cur.tail) : -1;
++ if (i < 0) {
++ log.warn(`[Run] job ${row.id}: Bot dropped the stdout Run had not streamed yet; the gap is dropped`);
++ chunk = '';
++ } else chunk = ring.text.slice(i + cur.tail.length);
++ }
++ cursors.set(row.id, { seq: ring.last_seq, chars: ring.text.length, first: ring.first_seq ?? null, tail: ring.text.slice(-256) });
++ if (chunk) stream.output(row.id, chunk, at);
++ }
++
++ /**
++ * Bot's answer's job record, checked against Run's own row: the id, the class and (when Bot names one)
++ * the project must be Run's. Anything else is refused — a job id is Run's, and Bot never names the
++ * project a job is billed to. → the record, or null.
++ */
++ function botRowOf(row, raw) {
++ if (!raw || typeof raw !== 'object' || Array.isArray(raw)) return null;
++ if (raw.id !== row.id) return null;
++ if (typeof raw.class === 'string' && raw.class !== row.class) return null;
++ if (raw.project_id != null && raw.project_id !== row.project_id) return null;
++ if (!STATES.includes(raw.state)) return null;
++ return raw;
++ }
++
++ /** The node Bot reports must be the node Run placed on; a different one is refused, never adopted. */
++ function nodeAgrees(row, raw) {
++ if (raw.node_id == null || row.node_id == null || raw.node_id === row.node_id) return true;
++ log.error(`[Run] job ${row.id}: Bot reports node ${raw.node_id}, Run placed it on ${row.node_id}: the answer is refused`);
++ return false;
++ }
++
++ // ── The three steps a tick takes ───────────────────────────────────────────
++
++ /**
++ * TTL: a job whose lifetime ran out ends `expired` (exit reason ttl → run.job.ttl; a job that never ran
++ * carries code null, which exit allows), and a node that may hold it is told to stop.
++ */
++ async function expireOne(row, at) {
++ 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})`); }
++ }
++ // 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}` })
++ : 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' });
++ noOffer.delete(row.id);
++ return note(row.id, 'expired', { changed: settled.changed });
++ }
++
++ /** Placement: a queued job is placed on a node advertising worker:<class> and sent to Bot once. */
++ async function placeOne(row, at) {
++ let chosen;
++ try { chosen = await placement.choose(row); }
++ catch (e) {
++ // Network unreachable: nothing is decided, the job stays queued and the next tick tries again.
++ // The message is the same for every queued job (the read is one call per tick), so it is logged
++ // slowly rather than once per job per second.
++ if (at - offersWarnedAt >= 60_000) { offersWarnedAt = at; log.warn(`[Run] placement could not read the offers (${e.code || e.message})`); }
++ stat.last_error = e.message;
++ return note(row.id, 'queued', { reason: e.code || 'no_offers' });
++ }
++ if (!chosen) {
++ // No eligible offer right now: the job waits for capacity (a node may be briefly offline) and every
++ // tick tries again; the ttl sweep ends it `expired` with run.job.unplaceable when ttl_ms runs out
++ // first (ADR-036 section 2), so a waiting job never fails on a transient absence.
++ if (at - noOfferWarnedAt >= 60_000) { noOfferWarnedAt = at; log.warn(`[Run] no node offers ${workerCapability(row.class)} yet; queued jobs wait until their ttl`); }
++ noOffer.add(row.id);
++ return note(row.id, 'queued', { reason: 'unplaceable' });
++ }
++ let frame;
++ try {
++ frame = store.frameOf(row.job);
++ contracts.assertValid('platform.job@1', frame);
++ } catch (e) {
++ return fail(row, at, 'run.job.worker_failed', `Run's own job record is not a valid platform.job@1: ${e.message}`);
++ }
++ let sent;
++ try {
++ 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 (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`);
++ 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());
++ }
++ // 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
++ // frame was in flight: an end state never changes, so such a job is not placed — and Bot, which just
++ // took the frame, is told to stop it rather than left running work its payer cancelled.
++ 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);
++ });
++ 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'}`);
++ // 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 });
++ }
++ stat.placed++;
++ if (sent.sent) stat.sent++;
++ stream.state(row.id, 'placed', at);
++ return note(row.id, 'placed', { node_id: chosen.node_id, sent: sent.sent });
++ }
++
++ /**
++ * Bot already holds this job id on a node Run has no record of (Run crashed between the send and its own
++ * mirror, or another Run process placed it). Run's row has no placement, so Bot's own is adopted rather
++ * than the job being killed and its work orphaned; a record that names no usable node fails the job.
++ */
++ async function adopt(row, at, err) {
++ let answer = null;
++ try { answer = await bot.state(row.id); } catch { /* the failure below names the conflict */ }
++ const b = answer && botRowOf(row, answer.job);
++ 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,
++ 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`);
++ return note(row.id, updated.state, { node_id: nodeId, adopted: true });
++ }
++
++ /** Bot has no such job (404). Run sent it, so it is counted and the job fails after a bounded number. */
++ async function unknownJob(row, at) {
++ 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`);
++ }
++
++ /** Mirror: what Bot answers about a job Run placed. */
++ async function mirrorOne(row, at) {
++ let answer;
++ try { answer = await bot.state(row.id); }
++ catch (e) {
++ stat.last_error = e.message;
++ if (!e.retryable) warnSlowly(row, at, `[Run] job ${row.id}: Bot's state was refused (${e.code || e.message})`);
++ return note(row.id, row.state, { reason: e.code || 'unavailable' });
++ }
++ if (!answer) return unknownJob(row, at);
++ unknown.delete(row.id);
++ const b = botRowOf(row, answer.job);
++ if (!b || !nodeAgrees(row, b)) return note(row.id, row.state, { refused: true });
++
++ // A state Bot reports behind the one Run holds is not a change (Bot's row is behind, e.g. Run placed
++ // the job before the device acked it): the meter is mirrored, the state is not.
++ const moved = RANK[b.state] > RANK[row.state] ? b.state : null;
++
++ if (moved && END_STATES.has(moved)) {
++ // The last output is streamed before the job's end: `end` closes the ring, so anything after it
++ // would never reach a subscriber.
++ stdoutOf(row, answer.stdout, at);
++ const settled = await db.tx(async (t) => {
++ // Placement, timings and the meter first, then the end state with its event; the error code is
++ // derived from the job_exit reason (store.settle: limit, stopped, ttl, exit_nonzero, …).
++ await store.mirror(t, row.id, {
++ state: null, node_id: row.node_id, provider: null, region: null,
++ started_ms: b.started_ms ?? null, wall_ms: b.wall_ms ?? null, usage_read: b.usage_read ?? null,
++ }, at);
++ const s = await store.settle(t, row.id, {
++ state: moved, exit: b.exit_reason ?? null, code: b.exit_code ?? null,
++ result: b.result ?? null, wallMs: b.wall_ms ?? null,
++ errorDetail: b.fault_code ? `bot:${String(b.fault_code).slice(0, 64)}` : null, at,
++ });
++ if (s.changed) await events.ofState(t, s.row);
++ return s;
++ });
++ if (settled.changed) {
++ stat.mirrored++;
++ if (moved === 'failed') stat.failed++;
++ forget(row.id);
++ stream.state(row.id, moved, at);
++ }
++ return note(row.id, moved, { changed: settled.changed, exit: b.exit_reason ?? null });
++ }
++
++ 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: b.started_ms ?? null, wall_ms: b.wall_ms ?? null, usage_read: b.usage_read ?? null,
++ }, at);
++ // store.mirror refuses a state the row cannot back (running without a started_ms), so the event
++ // follows the row, never Bot's word: an event whose payload contradicts the row is unpublishable.
++ 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);
++ }
++ // The job's output follows its state in the stream: running, then what it printed.
++ stdoutOf(row, answer.stdout, at);
++ // A cancel Run recorded but has not delivered yet: Bot gets it once per job and repeats it on every
++ // reconnect until the job ends (bot.job.dispatch), so one call is enough.
++ if (cancelling(row) && !cancelSent.has(row.id)) {
++ try {
++ const out = await bot.cancel(row.id);
++ cancelSent.add(row.id);
++ stat.cancels_sent++;
++ if (out === null) log.warn(`[Run] job ${row.id}: Bot has no such job to cancel`);
++ } catch (e) {
++ stat.last_error = e.message;
++ log.warn(`[Run] job ${row.id}: job_cancel did not reach Bot (${e.code || e.message})`);
++ }
++ }
++ return note(row.id, updated.state, { moved, sent: b.sent_at ? true : undefined });
++ }
++
++ /** 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' };
++ if (cancelSent.has(row.id)) return { cancelled: true, reason: 'already_sent' };
++ try {
++ await bot.cancel(row.id);
++ cancelSent.add(row.id);
++ stat.cancels_sent++;
++ return { cancelled: true, reason: 'sent' };
++ } catch (e) {
++ stat.last_error = e.message;
++ // The marker stays on the row, so the next tick retries; the caller answers what it knows.
++ return { cancelled: false, reason: e.code || 'unavailable' };
++ }
++ }
++
++ /** One step of the loop, per job: one job's failure never stops the tick. */
++ async function one(row, fn) {
++ try { return await fn(); }
++ catch (e) {
++ stat.errors++;
++ stat.last_error = `${e.code || e.name}: ${e.message}`;
++ log.warn(`[Run] job ${row.id}: ${e.message}`);
++ return null;
++ }
++ }
++
++ /** The ttl sweep alone (the runner calls it inside tick(); exported for the dispatcher's seam). */
++ async function sweep() {
++ const at = now();
++ const out = [];
++ for (const row of await store.dueTtl(db, at)) {
++ stat.swept++;
++ out.push(await one(row, () => expireOne(row, at)));
++ }
++ return out.filter(Boolean);
++ }
++
++ /** One poll: the sweep, then placement, then the mirror. Never two ticks at once. */
++ async function tick() {
++ if (!bot.enabled) return [];
++ if (busy) return (await inflight.catch(() => null), []); // one tick at a time; the timer's next one runs later
++ busy = true;
++ let release;
++ inflight = new Promise((r) => { release = r; });
++ const answers = [];
++ try {
++ stat.ticks++;
++ answers.push(...await sweep());
++ const at = now();
++ const waitingRows = await store.waiting(db, config.dispatch.maxPerTick);
++ for (const row of waitingRows) answers.push(await one(row, () => placeOne(row, at)));
++ const activeRows = await store.active(db, config.dispatch.maxPerTick);
++ const stillCancelling = new Set();
++ for (const row of activeRows) {
++ if (cancelling(row)) stillCancelling.add(row.id);
++ answers.push(await one(row, () => mirrorOne(row, at)));
++ }
++ // 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));
++ 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);
++ for (const key of [...warned.keys()]) if (!live.has(key)) warned.delete(key);
++ } catch (e) {
++ stat.errors++;
++ stat.last_error = e.message;
++ log.warn(`[Run] poller: ${e.message}`);
++ } finally {
++ busy = false;
++ stat.last_tick_at = now();
++ release();
++ }
++ stream.prune();
++ return answers.filter(Boolean);
++ }
++
++ function start() {
++ if (timer || !bot.enabled) 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?.();
++ return status();
++ }
++ function stop() { if (timer) clearInterval(timer); timer = null; }
++
++ function status() {
++ return {
++ running: Boolean(timer), interval_ms: config.dispatch.pollMs, ...stat,
++ cancel_pending: cancelSent.size, unknown_jobs: unknown.size, stdout_cursors: cursors.size,
++ };
++ }
++
++ return { tick, sweep, placeOne, cancelNow, start, stop, status };
++}
++
++module.exports = { createPoller, RANK, STATES };
ddiff --git a/server/jobs/store.js b/server/jobs/store.js
iindex 585c33e..7d3a8db 100644
---- a/server/jobs/store.js
++++ b/server/jobs/store.js
@@@ -166,6 +166,22 @@ function list(db, projectId, { state = null, limit = 50, cursor = null } = {}) {
return page(db, where, values, limit, cursor ? readCursor(cursor) : null);
}
++/**
++ * The dispatcher bridge's three reads (plan T14 step 6, server/jobs/poller.js). Row access only: every
++ * state change happens in the poller's transaction, so the run.job.* event exists if and only if it
++ * committed. No migration adds a state index (0001_jobs.sql is the final schema), so each is a plain scan
++ * of a small table; the poller bounds how much of it one tick takes.
++ */
++
++/** Jobs still waiting for a node, oldest first: what the poller places and sends. */
++const waiting = (db, limit) => db.many(`SELECT * FROM run_jobs WHERE state = 'queued' ORDER BY created_at, id LIMIT $1`, [limit]);
++
++/** Jobs a node may hold — placed or running — oldest first: what the poller mirrors from Bot. */
++const active = (db, limit) => db.many(`SELECT * FROM run_jobs WHERE state IN ('placed', 'running') ORDER BY created_at, id LIMIT $1`, [limit]);
++
++/** Jobs whose lifetime ran out (created_at + ttl_ms <= at) and that have not ended, oldest first: the sweeper. */
++const dueTtl = (db, at, limit = 500) => db.many(`SELECT * FROM run_jobs WHERE state IN ('queued', 'placed', 'running') AND created_at + ttl_ms <= $1 ORDER BY created_at, id LIMIT $2`, [at, limit]);
++
/** Every project's jobs: GET /api/v1/admin/jobs (run.job.admin); project_id narrows it. */
function adminList(db, { project_id: projectId = null, state = null, limit = 50, cursor = null } = {}) {
const where = [];
@@@ -176,20 +192,23 @@ function adminList(db, { project_id: projectId = null, state = null, limit = 50,
}
/**
-- * Cancel one job (POST /api/v1/jobs/:id/cancel, run.job.cancel). A queued or placed job never starts:
-- * cancelled now. A running one is marked cancel_requested and stays running until its job_exit arrives
-- * (the dispatcher bridge sends job_cancel in plan T14 step 6). An ended job changes nothing.
++ * Cancel one job (POST /api/v1/jobs/:id/cancel, run.job.cancel). A queued job was never sent anywhere:
++ * cancelled now. A placed or running job may be on a node this instant (Run marked it placed as soon as
++ * Bot took the frame, before the node's job_started arrived), so it is only marked cancel_requested: the
++ * dispatcher bridge sends job_cancel and the job reaches `cancelled` when Bot reports it (plan T14 step 6,
++ * job_exit reason cancelled). An ended job changes nothing.
* → { row, cancelled, cancelRequested } — cancelled means the row moved to 'cancelled' now.
*/
async function cancel(db, id, projectId, at) {
const moved = await db.maybe(
`UPDATE run_jobs SET state = 'cancelled', updated_at = $3, finished_ms = $3
-- WHERE id = $1 AND project_id = $2 AND state IN ('queued', 'placed') RETURNING *`,
++ WHERE id = $1 AND project_id = $2 AND state = 'queued' RETURNING *`,
[id, projectId, at]);
if (moved) return { row: moved, cancelled: true, cancelRequested: false };
const asked = await db.maybe(
`UPDATE run_jobs SET job = jsonb_set(job, '{cancel_requested}', 'true'::jsonb), updated_at = $3
-- WHERE id = $1 AND project_id = $2 AND state = 'running' AND job->>'cancel_requested' IS DISTINCT FROM 'true' RETURNING *`,
++ WHERE id = $1 AND project_id = $2 AND state IN ('placed', 'running')
++ AND job->>'cancel_requested' IS DISTINCT FROM 'true' RETURNING *`,
[id, projectId, at]);
if (asked) return { row: asked, cancelled: false, cancelRequested: true };
return { row: await get(db, id, projectId), cancelled: false, cancelRequested: false };
@@@ -202,12 +221,14 @@ async function cancel(db, id, projectId, at) {
*/
async function mirror(db, id, bot = {}, at) {
const placed = bot.node_id ? at : null;
-- // Bot's answer moves the row only into a state the row can back: run.job-read-result@1 requires a
-- // placement for placed/running/succeeded and started_at for running, so an answer missing them leaves
-- // the state as it is (the bridge sends the placement with the state it describes).
++ // Bot's answer moves the row only into a state the row can back: an end state never changes (the answer
++ // of a job the API cancelled or the ttl swept while the poll was in flight cannot revive it), Run's
++ // requirements (run.job-read-result@1: a placement for placed/running/succeeded, started_at for running)
++ // decide the rest, and an answer missing them leaves the state as it is.
return db.maybe(
`UPDATE run_jobs SET
state = CASE
++ WHEN state IN ('succeeded', 'failed', 'cancelled', 'expired') THEN state
WHEN $2::text IS NULL THEN state
WHEN $2::text IN ('placed', 'running', 'succeeded') AND COALESCE($3, node_id) IS NULL THEN state
WHEN $2::text = 'running' AND COALESCE($6, started_ms) IS NULL THEN state
@@@ -306,4 +327,6 @@ function readResult(row) {
module.exports = {
// The store surface the API and the dispatcher bridge (plan T14 step 6) use.
put, mint, get, list, adminList, cancel, mirror, settle, toJob, readResult, frameOf,
++ // The poller's reads (waiting → placed → active → ended).
++ waiting, active, dueTtl,
};
ddiff --git a/server/observability.js b/server/observability.js
iindex 4ccec76..e3a35df 100644
---- a/server/observability.js
++++ b/server/observability.js
@@@ -7,10 +7,11 @@
* valkey optional a real PING; without it nothing is shared between processes
* network_jwks optional the Network signing key has loaded; without it no token can be verified
* events optional OpenVibe.Events answers /api/health; without it run.job.* events wait in the outbox
-- * bot optional OpenVibe.Bot answers /api/health; skipped until the dispatcher bridge exists
++ * bot optional OpenVibe.Bot answers /api/health; skipped when Run cannot call it (no OV_OAUTH_CLIENT_SECRET)
*
* Gauges: jobs queued, placed or running, ended, and the outbox backlog. Counts only, never subjects, and
-- * never a job's payload; a scrape whose read failed leaves them out.
++ * never a job's payload; a scrape whose read failed leaves them out. The details also carry the poller's
++ * own status (server/jobs/poller.js) so a job that is not moving can be told from one that is mid-flight.
*/
const { sql } = require('openvibe-sdk/db');
const { createReadiness, skip } = require('openvibe-shared/ready');
@@@ -32,7 +33,7 @@ function createRunReadiness({ db, valkey = null, keys, config, outbox, stream =
: () => skip('events relay off (EVENTS_URL or OV_OAUTH_CLIENT_SECRET unset): run.job.* events wait in the outbox');
const bot = dispatch && dispatch.enabled
? probe(`${config.dispatch.botUrl}/api/health`, fetchImpl)
-- : () => skip('the dispatcher bridge to Bot is not built yet (plan T14 step 6): jobs stay queued');
++ : () => skip('the dispatcher bridge to Bot is off (no OV_OAUTH_CLIENT_SECRET): no job leaves queued');
return createReadiness({
service: 'run',
release,
@@@ -44,8 +45,11 @@ function createRunReadiness({ db, valkey = null, keys, config, outbox, stream =
{ name: 'bot', required: false, cacheMs: dispatch && dispatch.enabled ? PING_TTL_MS : 0, timeoutMs: 2500, check: bot },
],
details: async (body) => (body.checks.db.status === 'ok'
-- ? { jobs_queued: await queued(db), events_outbox: await outbox.status(), streams: stream ? stream.status().streams : null }
-- : { jobs_queued: null, events_outbox: null, streams: null }),
++ ? {
++ jobs_queued: await queued(db), jobs_active: await active(db), events_outbox: await outbox.status(),
++ streams: stream ? stream.status().streams : null, poller: dispatch ? dispatch.status().poller : null,
++ }
++ : { jobs_queued: null, jobs_active: null, events_outbox: null, streams: null, poller: null }),
});
}
@@@ -53,6 +57,11 @@ async function queued(db) {
return Number(await db.value(`SELECT count(*)::int FROM run_jobs WHERE state = 'queued'`) || 0);
}
++/** Jobs on a node: placed or running (what the poller mirrors from Bot). */
++async function active(db) {
++ return Number(await db.value(`SELECT count(*)::int FROM run_jobs WHERE state IN ('placed', 'running')`) || 0);
++}
++
/** Run's gauges on the openvibe-shared/metrics registry; refresh() reads them all in one query. */
function registerRunGauges(registry, { db, stream = null, now = () => Date.now(), log = console }) {
let snap = null;
ddiff --git a/test/app.test.js b/test/app.test.js
iindex 3d31988..e89b3e0 100644
---- a/test/app.test.js
++++ b/test/app.test.js
@@@ -21,8 +21,10 @@ async function main() {
assert.strictEqual(r.json.version, require('../package.json').version);
assert.strictEqual(r.json.streams, 0);
assert.strictEqual(r.json.events.enabled, false); // EVENTS_URL unset in the test
-- assert.strictEqual(r.json.dispatcher.enabled, false); // plan T14 step 6 is not built
-- assert.match(r.json.dispatcher.note, /T14 step 6/);
++ // The dispatcher bridge exists (plan T14 step 6), but without OV_OAUTH_CLIENT_SECRET Run cannot
++ // mint the Bot token it needs: nothing is sent and every job stays queued.
++ assert.strictEqual(r.json.dispatcher.enabled, false);
++ assert.match(r.json.dispatcher.note, /OV_OAUTH_CLIENT_SECRET/);
});
await check('readiness: GET /api/ready reports the database, and honestly skips what is absent', async () => {
@@@ -38,8 +40,10 @@ async function main() {
assert.strictEqual(r.json.checks.valkey.status, 'skipped'); // VALKEY_URL unset
assert.strictEqual(r.json.checks.events.status, 'skipped'); // the relay is off
assert.strictEqual(r.json.checks.bot.status, 'skipped');
-- assert.match(r.json.checks.bot.reason, /T14 step 6/);
++ assert.match(r.json.checks.bot.reason, /OV_OAUTH_CLIENT_SECRET/); // not probed: Run cannot call Bot
assert.strictEqual(r.json.jobs_queued, 0);
++ assert.strictEqual(r.json.jobs_active, 0);
++ assert.strictEqual(r.json.poller.running, false); // nothing starts in the app factory
assert.strictEqual(r.json.events_outbox.pending, 0);
assert.strictEqual(r.json.checks.network_jwks.status, 'ok'); // the stub Network's key loaded
});
@@@ -66,7 +70,7 @@ async function main() {
assert.strictEqual(proxied.status, 404);
});
-- await check('a job that stays queued: the dispatcher seam hands nothing to a node', async () => {
++ await check('with no Bot credentials the bridge is off: a submitted job stays queued', async () => {
const made = await t.call('POST', '/api/v1/jobs', { body: t.request(), cap: ['run.job.submit'], project, sub: 'svc:builder' });
assert.strictEqual(made.status, 201, made.text);
assert.strictEqual(made.json.job.state, 'queued');
ddiff --git a/test/dispatch.test.js b/test/dispatch.test.js
nnew file mode 100644
iindex 0000000..f9e62e8
---- /dev/null
++++ b/test/dispatch.test.js
@@@ -0,0 +1,626 @@
++'use strict';
++/**
++ * The dispatcher bridge (plan T14 steps 6 and 7): a submitted job is placed on a node from the Fabric offers
++ * (the stub Network's registry + openvibe-sdk/placement), sent to OpenVibe.Bot once, mirrored from Bot's own
++ * record, streamed, cancelled and expired — all against an in-process fake Bot that verifies the Network
++ * service token Run mints for it (audience openvibe.bot, capability bot.job.dispatch).
++ *
++ * node test/dispatch.test.js
++ *
++ * The fake Bot is a real HTTP server (not a fetch stub), so Run's token client, its Authorization header,
++ * its timeouts and Bot's 404/409 answers travel the wire the way they will in production, and the token
++ * cache is observable (the stub Network counts its /oauth/token calls). The job loop is driven by hand
++ * (RUN_JOBS=off) except where a test starts it on purpose.
++ */
++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 DEVICE = 'dev_01J8Z4M2Q0R7T9YV3K6N8P1W2X'; // the device Bot dispatches to (offer.node_id)
++const OTHER_DEVICE = 'dev_01J8Z4M2Q0R7T9YV3K6N8P1W2XQ';
++const PROJECT = () => ids.newId('project');
++
++/** A contract-valid platform.resource-offer@1 for a paired Node advertising worker:<class>. */
++function offer(over = {}) {
++ return {
++ offer_id: 'off_node_01',
++ kind: 'node',
++ node_id: DEVICE,
++ region: 'eu-west',
++ cell: 'wnam-1',
++ trust: 'first-party',
++ capabilities: ['node:http', 'worker:function'],
++ health: { status: 'up', checked_at: new Date().toISOString() },
++ pricing: { model: 'prepaid' },
++ updated_at: new Date().toISOString(),
++ ...over,
++ };
++}
++
++/**
++ * A fake OpenVibe.Bot: its three job routes, a device registry and a job record shaped like
++ * server/jobs/index.js presentJob(). Every call is verified to carry Run's service token for audience
++ * openvibe.bot holding bot.job.dispatch — the real auth path, not a bypass. The Network keys it verifies
++ * against are handed to it after boot() (useNetwork).
++ */
++async function startFakeBot() {
++ const keys = { publicPem: null, issuer: null };
++ const devices = new Map([[DEVICE, { classes: ['function'] }]]);
++ const jobs = new Map();
++ const calls = [];
++ const mode = { exitOnCancel: true };
++ const chunkSeq = new Map();
++
++ const present = (row) => ({
++ id: row.id, node_id: row.node_id, class: row.class, state: row.state,
++ project_id: row.project_id, subject: row.subject, provider: row.provider,
++ cancel_requested: row.cancel_requested, fault_code: row.fault_code ?? null,
++ sent_at: row.sent_at ?? null, started_ms: row.started_ms ?? null, finished_at: row.finished_at ?? null,
++ exit_reason: row.exit_reason ?? null, exit_code: row.exit_code ?? null, wall_ms: row.wall_ms ?? null,
++ usage_read: row.usage_read ?? 0, result: row.result ?? null, job: row.job,
++ created_at: row.created_at, updated_at: row.updated_at,
++ });
++ const stdoutOf = (row) => {
++ const chunks = row.chunks || [];
++ return {
++ text: chunks.map((c) => c.chunk).join(''), first_seq: chunks.length ? chunks[0].seq : null,
++ last_seq: chunks.length ? chunks[chunks.length - 1].seq : null, truncated: false,
++ };
++ };
++ 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 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://bot.test');
++ if (url.pathname === '/api/health') return answer(200, { ok: true, service: 'bot' });
++ const parts = url.pathname.split('/').filter(Boolean); // api v1 jobs [id] [cancel]
++ const auth = String(req.headers.authorization || '');
++ if (!auth.startsWith('Bearer ')) return answer(401, { code: 'bot.sign_in', detail: 'sign in to do that' });
++ const verified = serviceAuth.verifyServiceToken(auth.slice(7).trim(), { publicKey: keys.publicPem, issuer: keys.issuer, audience: 'openvibe.bot' });
++ if (!verified.ok) return answer(401, { code: verified.code, detail: verified.reason });
++ if (!capabilities.grants(verified.claims.cap, 'bot.job.dispatch')) return answer(403, { code: 'capability.denied', detail: 'bot.job.dispatch not granted' });
++ if (verified.claims.sub !== 'svc:run') return answer(403, { code: 'bot.forbidden', detail: 'job dispatch is Run\'s' });
++
++ if (req.method === 'POST' && url.pathname === '/api/v1/jobs') {
++ const doc = await body(req);
++ calls.push({ method: 'POST', path: url.pathname, body: doc, claims: verified.claims });
++ if (!doc || !doc.project_id) return answer(422, { code: 'bot.invalid_input', detail: 'project_id is required (Run is the payer)' });
++ const device = devices.get(doc.node_id);
++ if (!device) return answer(404, { code: 'bot.device_not_found', detail: 'no such device, or it is revoked' });
++ if (!device.classes.includes(doc.job && doc.job.class)) return answer(409, { code: 'bot.class_unadvertised', detail: `the device does not advertise the runtime class ${doc.job && doc.job.class}` });
++ const known = jobs.get(doc.job.id);
++ if (known) return answer(201, { job: present(known), sent: false }); // Bot's dispatch is idempotent by job id
++ const now = Date.now();
++ const row = { ...doc, id: doc.job.id, class: doc.job.class, state: 'queued', cancel_requested: false, created_at: now, updated_at: now, chunks: [] };
++ row.sent_at = device.connected === false ? null : now;
++ jobs.set(row.id, row);
++ // A test hook: what happens on the wire between Bot taking the frame and Run recording it.
++ if (mode.onDispatch) await mode.onDispatch(doc);
++ return answer(201, { job: present(row), sent: row.sent_at != null });
++ }
++ if (parts[0] === 'api' && parts[1] === 'v1' && parts[2] === 'jobs' && parts[3]) {
++ const row = jobs.get(parts[3]);
++ if (req.method === 'POST' && parts[4] === 'cancel') {
++ calls.push({ method: 'POST', path: url.pathname, claims: verified.claims });
++ if (!row) return answer(404, { code: 'bot.job_not_found', detail: 'no such job' });
++ row.cancel_requested = true;
++ row.updated_at = Date.now();
++ const sent = row.sent_at != null && row.exit_reason == null;
++ // A killed process has no exit status: job_exit cancelled carries code null.
++ if (mode.exitOnCancel && row.exit_reason == null) finish(row, { reason: 'cancelled', code: null });
++ return answer(200, { job: present(row), sent });
++ }
++ if (req.method === 'GET' && !parts[4]) {
++ calls.push({ method: 'GET', path: url.pathname, claims: verified.claims });
++ if (!row) return answer(404, { code: 'bot.job_not_found', detail: 'no such job' });
++ return answer(200, { job: present(row), stdout: stdoutOf(row) });
++ }
++ }
++ return answer(404, { code: 'bot.not_found', detail: 'no such route' });
++ });
++
++ /** job_started: the node's process started (Bot keeps the first started_ms). */
++ function start(id, startedMs = Date.now()) {
++ const row = jobs.get(id);
++ row.started_ms = row.started_ms ?? startedMs;
++ row.state = row.exit_reason == null ? 'running' : row.state;
++ row.updated_at = Date.now();
++ }
++ /** job_stdout: one chunk, the job's own chunk_seq from 1. */
++ function stdout(id, chunk) {
++ const row = jobs.get(id);
++ const seq = (chunkSeq.get(id) || 0) + 1;
++ chunkSeq.set(id, seq);
++ row.chunks.push({ seq, chunk });
++ }
++ /** job_exit: the end state and the metered wall clock (Bot is the meter). */
++ function finish(row, { reason, code = 0, wallMs = 0, result = null }) {
++ row.state = reason === 'exited' && code === 0 ? 'succeeded' : reason === 'cancelled' ? 'cancelled' : reason === 'ttl' ? 'expired' : 'failed';
++ row.exit_reason = reason; row.exit_code = code; row.wall_ms = wallMs; row.result = result;
++ row.usage_read = Math.ceil(wallMs / 1000);
++ row.finished_at = Date.now(); row.updated_at = Date.now();
++ }
++ const exit = (id, opts) => finish(jobs.get(id), opts);
++ /** Bot restarted without this job (or lost it): the job it once took is gone. */
++ const forget = (id) => jobs.delete(id);
++ const url = await new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve(`http://127.0.0.1:${server.address().port}`)));
++
++ return {
++ url, calls, devices, jobs, mode, start, stdout, exit, forget, present,
++ /** The Network stub this Bot verifies Run's token against (after boot). */
++ useNetwork: (network) => { keys.publicPem = network.publicPem; keys.issuer = network.issuer; },
++ callsOf: (method, path) => calls.filter((c) => c.method === method && (!path || c.path === path)),
++ close: () => new Promise((r) => { server.closeAllConnections(); server.close(r); }),
++ };
++}
++
++async function main() {
++ const bot = await startFakeBot();
++ const t = await boot({
++ env: {
++ OV_OAUTH_CLIENT_SECRET: crypto.randomBytes(12).toString('hex'),
++ RUN_BOT_URL: bot.url,
++ RUN_OFFERS_TTL_MS: '0', // every tick reads the registry; the stub is local
++ RUN_DISPATCH_UNKNOWN_POLLS: '3',
++ RUN_JOBS: 'off', // the tests drive dispatch.poll() themselves
++ },
++ });
++ bot.useNetwork(t.network);
++ const submit = (over = {}, opts = {}) => t.submit(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' });
++
++ try {
++ await check('the bridge is on: health and readiness say so', async () => {
++ const health = await t.call('GET', '/api/health', { token: null });
++ assert.strictEqual(health.json.dispatcher.enabled, true, health.text);
++ assert.match(health.json.dispatcher.note, /openvibe\.bot/);
++ assert.strictEqual(health.json.dispatcher.bot_url, bot.url);
++ const ready = await t.call('GET', '/api/ready', { token: null });
++ assert.strictEqual(ready.status, 200, ready.text);
++ assert.strictEqual(ready.json.checks.bot.status, 'ok', ready.text); // Bot answers, and it is not required
++ assert.strictEqual(ready.json.checks.bot.required, false);
++ assert.ok(ready.json.poller, 'readiness reports the poller');
++ assert.strictEqual(ready.json.poller.running, false); // nothing starts before start()
++ assert.strictEqual(ready.json.jobs_active, 0);
++ });
++
++ await check('the poller starts and stops; nothing runs at module load', async () => {
++ assert.strictEqual(t.dispatch.status().poller.ticks, 0);
++ assert.strictEqual(t.dispatch.start().running, true);
++ t.dispatch.stop();
++ assert.strictEqual(t.dispatch.status().poller.running, false);
++ });
++
++ await check('a queued job is placed from the offers and sent to Bot exactly once', async () => {
++ t.setOffers([offer()]);
++ const project = PROJECT();
++ const sentBefore = bot.callsOf('POST', '/api/v1/jobs').length;
++ const job = await submit({}, { project });
++ assert.strictEqual(job.state, 'queued');
++ const first = await t.dispatch.poll();
++ const mine = first.find((a) => a.job_id === job.id);
++ assert.strictEqual(mine.state, 'placed');
++ const sent = bot.callsOf('POST', '/api/v1/jobs');
++ assert.strictEqual(sent.length, sentBefore + 1, JSON.stringify(sent.map((c) => c.body && c.body.job.id)));
++ const call = sent[sent.length - 1];
++ // The platform.job@1 body, the device the offer names, and Run's own project and requester.
++ assert.strictEqual(call.body.node_id, DEVICE);
++ assert.strictEqual(call.body.project_id, project);
++ assert.strictEqual(call.body.subject, 'service:builder');
++ assert.deepStrictEqual(call.claims.aud, ['openvibe.bot']); // the audience Bot checks
++ assert.deepStrictEqual(call.claims.cap, ['bot.job.dispatch']);
++ assert.deepStrictEqual(call.body.job, {
++ id: job.id, class: 'function', artifact: { name: 'thumbnail', version: '1.2.0' }, args: { width: 320 },
++ ttl_ms: 30_000 + 600_000, limits: { wall_ms: 30_000, cpu_ms: 30_000, mem_bytes: 268_435_456 }, net: 'none', inputs: [],
++ });
++ assert.ok(contracts.validate('platform.job@1', call.body.job).valid);
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.state, 'placed');
++ assert.deepStrictEqual(read1.json.placement, { node: DEVICE, provider: null, region: 'eu-west' });
++ assert.ok(read1.json.timings.placed_at);
++ assert.strictEqual(read1.json.timings.started_at, null);
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ assert.strictEqual((await t.outboxRows('run.job.queued', job.id)).length, 1); // placed has no event of its own
++ assert.strictEqual((await t.outboxRows('run.job.started', job.id)).length, 0);
++ // It is sent once: the next polls only mirror what Bot answers.
++ await t.dispatch.poll();
++ await t.dispatch.poll();
++ assert.strictEqual(bot.callsOf('POST', '/api/v1/jobs').length, sentBefore + 1, 'a placed job is never sent again');
++ });
++
++ await check('placed→running→succeeded mirrors with exactly one event each and Bot\'s meter', async () => {
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ await t.dispatch.poll();
++ bot.start(job.id, Date.now());
++ const running = (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++ assert.strictEqual(running.state, 'running');
++ const started = await t.outboxRows('run.job.started', job.id);
++ assert.strictEqual(started.length, 1);
++ assert.strictEqual(started[0].outbox_key, `run:${job.id}:running`);
++ assert.strictEqual(started[0].payload.state, 'running');
++ assert.strictEqual(started[0].payload.placement.node, DEVICE);
++ assert.strictEqual(started[0].payload.timings.finished_at, null);
++ assert.ok(contracts.validate('run.job.started@1', started[0].payload).valid, JSON.stringify(started[0].payload));
++
++ const result = { thumb: 'med_01JAB2C3D4E5F6G7H8J9K0MNPS' };
++ bot.exit(job.id, { reason: 'exited', code: 0, wallMs: 2350, result });
++ const ended = (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++ assert.strictEqual(ended.state, 'succeeded');
++ const read1 = await read(job.id, project);
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ assert.strictEqual(read1.json.state, 'succeeded');
++ assert.deepStrictEqual(read1.json.exit, { reason: 'exited', code: 0 });
++ assert.deepStrictEqual(read1.json.result, result);
++ assert.deepStrictEqual(read1.json.usage, { seconds: 2.35, wall_ms: 2350, cpu_ms: null, mem_peak_bytes: null });
++ assert.strictEqual(read1.json.error, null);
++ const ok = await t.outboxRows('run.job.succeeded', job.id);
++ assert.strictEqual(ok.length, 1);
++ assert.strictEqual(ok[0].outbox_key, `run:${job.id}:succeeded`);
++ assert.strictEqual(ok[0].payload.result, undefined); // the payload never carries the result
++ assert.ok(contracts.validate('run.job.succeeded@1', ok[0].payload).valid);
++ // Later polls change nothing: an end state never changes and its event is one row.
++ await t.dispatch.poll();
++ await t.dispatch.poll();
++ assert.strictEqual((await t.outboxRows('run.job.succeeded', job.id)).length, 1);
++ assert.strictEqual((await read(job.id, project)).json.state, 'succeeded');
++ });
++
++ await check('a non-zero exit mirrors failed with run.job.exit_nonzero; a killed job with run.job.limit', async () => {
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ await t.dispatch.poll();
++ bot.start(job.id);
++ await t.dispatch.poll();
++ bot.exit(job.id, { reason: 'exited', code: 3, wallMs: 1200 });
++ assert.strictEqual((await t.dispatch.poll()).find((a) => a.job_id === job.id).state, 'failed');
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.state, 'failed');
++ assert.deepStrictEqual(read1.json.exit, { reason: 'exited', code: 3 });
++ assert.strictEqual(read1.json.error.code, 'run.job.exit_nonzero');
++ assert.strictEqual(read1.json.usage.wall_ms, 1200);
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ const rows = await t.outboxRows('run.job.failed', job.id);
++ assert.strictEqual(rows.length, 1);
++ assert.ok(contracts.validate('run.job.failed@1', rows[0].payload).valid);
++ assert.strictEqual(rows[0].payload.error.code, 'run.job.exit_nonzero');
++
++ const limited = await submit({}, { project: PROJECT() });
++ await t.dispatch.poll();
++ bot.start(limited.id);
++ await t.dispatch.poll();
++ bot.exit(limited.id, { reason: 'limit', code: null, wallMs: 5000 });
++ await t.dispatch.poll();
++ const read2 = await read(limited.id, await t.db.value('SELECT project_id FROM run_jobs WHERE id = $1', [limited.id]));
++ assert.strictEqual(read2.json.state, 'failed');
++ assert.strictEqual(read2.json.error.code, 'run.job.limit'); // job_exit reason limit
++ assert.deepStrictEqual(read2.json.exit, { reason: 'limit', code: null });
++ });
++
++ await check('the job\'s stdout reaches the SSE stream in order, with monotonic seq', async () => {
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ const s = await t.openStream(job.id, { project });
++ const seed = await s.waitFor((e) => e.event === 'state');
++ assert.strictEqual(seed.data.state, 'queued');
++ await t.dispatch.poll(); // placed
++ const placed = await s.waitFor((e) => e.event === 'state' && e.data.state === 'placed');
++ assert.ok(placed.data.seq > seed.data.seq);
++ bot.start(job.id);
++ bot.stdout(job.id, 'resized 1 of 1\n');
++ await t.dispatch.poll();
++ bot.stdout(job.id, 'done\n');
++ await t.dispatch.poll();
++ const out1 = await s.waitFor((e) => e.event === 'output' && e.data.chunk === 'resized 1 of 1\n');
++ const out2 = await s.waitFor((e) => e.event === 'output' && e.data.chunk === 'done\n');
++ assert.ok(contracts.validate('run.job-stream-event@1', out1.data).valid, JSON.stringify(out1.data));
++ assert.strictEqual(out1.data.seq + 1, out2.data.seq);
++ assert.strictEqual(out1.id, String(out1.data.seq));
++ assert.strictEqual(out1.data.job_id, job.id);
++ // Output is never streamed twice: a poll with no new chunk says nothing.
++ const before = s.events.filter((e) => e.event === 'output').length;
++ await t.dispatch.poll();
++ await t.wait(50);
++ assert.strictEqual(s.events.filter((e) => e.event === 'output').length, before);
++ bot.exit(job.id, { reason: 'exited', code: 0, wallMs: 1000 });
++ await t.dispatch.poll();
++ const end = await s.waitFor((e) => e.event === 'end');
++ assert.strictEqual(end.data.state, 'succeeded');
++ assert.ok(end.data.seq > out2.data.seq);
++ assert.strictEqual(await s.waitForClose(5000), true);
++ });
++
++ await check('a class the node does not advertise fails the job run.job.worker_failed', async () => {
++ // The offer advertises worker:function (a stale report); the device Bot knows advertises code.
++ t.setOffers([offer({ offer_id: 'off_stale' })]);
++ bot.devices.set(DEVICE, { classes: ['code'] });
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ const answer = (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++ assert.strictEqual(answer.state, 'failed');
++ assert.strictEqual(answer.error, 'run.job.worker_failed');
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.state, 'failed');
++ assert.strictEqual(read1.json.error.code, 'run.job.worker_failed');
++ assert.match(read1.json.error.detail, /bot\.class_unadvertised/);
++ assert.strictEqual(read1.json.usage.wall_ms, 0);
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ const failed = await t.outboxRows('run.job.failed', job.id);
++ assert.strictEqual(failed.length, 1);
++ assert.ok(contracts.validate('run.job.failed@1', failed[0].payload).valid);
++ bot.devices.set(DEVICE, { classes: ['function'] });
++ });
++
++ await check('no offer for worker:<class>: the job waits queued, and expires as run.job.unplaceable at its ttl', async () => {
++ t.setOffers([]);
++ const project = PROJECT();
++ const sentBefore = bot.callsOf('POST', '/api/v1/jobs').length;
++ const job = await submit({}, { project });
++ const answer = (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++ assert.strictEqual(answer.state, 'queued'); // waits for capacity
++ assert.strictEqual(bot.callsOf('POST', '/api/v1/jobs').length, sentBefore); // nothing reached Bot
++ assert.strictEqual((await read(job.id, project)).json.state, 'queued');
++ // Its ttl runs out with no offer: expired, run.job.unplaceable (ADR-036 section 2).
++ await t.db.exec('UPDATE run_jobs SET created_at = created_at - 100000000 WHERE id = $1', [job.id]);
++ assert.strictEqual((await t.dispatch.sweep()).find((a) => a.job_id === job.id).state, 'expired');
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.state, 'expired');
++ assert.strictEqual(read1.json.error.code, 'run.job.unplaceable');
++ assert.match(read1.json.error.detail, /worker:function/);
++ assert.strictEqual(read1.json.placement, null);
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ assert.ok(contracts.validate('run.job.expired@1', (await t.outboxRows('run.job.expired', job.id))[0].payload).valid);
++
++ // An offer for another class is not a candidate either: the filter is this job's own class.
++ t.setOffers([offer({ offer_id: 'off_code', capabilities: ['worker:code'] })]);
++ bot.devices.set(DEVICE, { classes: ['function', 'code'] });
++ const other = await submit({}, { project: PROJECT() });
++ assert.strictEqual((await t.dispatch.poll()).find((a) => a.job_id === other.id).state, 'queued');
++ await t.db.exec('UPDATE run_jobs SET created_at = created_at - 100000000 WHERE id = $1', [other.id]);
++ await t.dispatch.sweep(); // it waited; now it is out of the way
++
++ // And the same class with an offer is placed, provider and region included.
++ t.setOffers([offer({ offer_id: 'off_both', provider: 'ovh', capabilities: ['worker:function', 'worker:code'] })]);
++ const ok = await submit({}, { project: PROJECT() });
++ const placed = (await t.dispatch.poll()).find((a) => a.job_id === ok.id);
++ assert.strictEqual(placed.state, 'placed');
++ assert.strictEqual(placed.node_id, DEVICE);
++ const row = await t.db.maybe('SELECT provider, region FROM run_jobs WHERE id = $1', [ok.id]);
++ assert.deepStrictEqual({ provider: row.provider, region: row.region }, { provider: 'ovh', region: 'eu-west' });
++ bot.devices.set(DEVICE, { classes: ['function'] });
++ });
++
++ await check('a user-owned node is eligible only when the job\'s requirements name it', async () => {
++ bot.devices.set(OTHER_DEVICE, { classes: ['function'] });
++ t.setOffers([offer({ offer_id: 'off_user', trust: 'user-owned', node_id: OTHER_DEVICE })]);
++ const projectA = PROJECT();
++ const plain = await submit({}, { project: projectA });
++ const denied = (await t.dispatch.poll()).find((a) => a.job_id === plain.id);
++ assert.strictEqual(denied.state, 'queued'); // user-owned is never a default: no eligible node
++ const readA = await read(plain.id, projectA);
++ assert.strictEqual(readA.json.state, 'queued');
++ assert.strictEqual(readA.json.placement, null);
++ await t.db.exec('UPDATE run_jobs SET created_at = created_at - 100000000 WHERE id = $1', [plain.id]);
++ await t.dispatch.sweep();
++ const projectB = PROJECT();
++ const named = await submit({ requirements: { kind: 'run.function', mobility: 'job', latency_class: 'background', objective: 'balanced', trust: ['user-owned', 'first-party'] } }, { project: projectB });
++ const placed = (await t.dispatch.poll()).find((a) => a.job_id === named.id);
++ assert.strictEqual(placed.state, 'placed');
++ assert.strictEqual(placed.node_id, OTHER_DEVICE);
++ bot.devices.delete(OTHER_DEVICE);
++ });
++
++ await check('a job cancelled while its frame is in flight is not placed, and Bot is told to stop it', async () => {
++ t.setOffers([offer()]);
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ // The cancel lands after Bot took the frame and before Run recorded the placement.
++ bot.mode.onDispatch = async (doc) => {
++ if (doc.job.id !== job.id) return;
++ bot.mode.onDispatch = null;
++ 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(asked.json.state, 'cancelled'); // queued: cancelled at once, never sent anywhere
++ };
++ const answer = (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++ bot.mode.onDispatch = null;
++ assert.strictEqual(answer.raced, true);
++ assert.strictEqual(answer.stopped, true);
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.state, 'cancelled'); // an end state never changes
++ assert.strictEqual(read1.json.placement, null);
++ assert.strictEqual(bot.jobs.get(job.id).cancel_requested, true); // the frame Run had sent is cancelled
++ assert.strictEqual((await t.outboxRows('run.job.cancelled', job.id)).length, 1);
++ assert.strictEqual((await t.outboxRows('run.job.started', job.id)).length, 0);
++ });
++
++ await check('Bot losing a job it was sent: the job fails after a bounded number of polls', async () => {
++ t.setOffers([offer()]);
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ await t.dispatch.poll();
++ bot.forget(job.id);
++ const first = (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++ assert.strictEqual(first.bot_404, 1);
++ assert.strictEqual((await read(job.id, project)).json.state, 'placed'); // not failed on the first miss
++ const second = (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++ assert.strictEqual(second.bot_404, 2);
++ const third = (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++ assert.strictEqual(third.state, 'failed');
++ assert.strictEqual(third.error, 'run.job.worker_failed');
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.state, 'failed');
++ assert.match(read1.json.error.detail, /Bot has no job/);
++ assert.strictEqual((await t.outboxRows('run.job.failed', job.id)).length, 1);
++ await t.dispatch.poll();
++ assert.strictEqual((await read(job.id, project)).json.state, 'failed'); // never mirrored forever
++ assert.strictEqual((await t.outboxRows('run.job.failed', job.id)).length, 1);
++ });
++
++ await check('cancel of a running job reaches Bot and settles cancelled when Bot confirms', async () => {
++ t.setOffers([offer()]);
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ await t.dispatch.poll();
++ bot.start(job.id);
++ await t.dispatch.poll();
++ assert.strictEqual((await read(job.id, project)).json.state, 'running');
++ const cancelsBefore = bot.callsOf('POST', `/api/v1/jobs/${job.id}/cancel`).length;
++ 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);
++ // Run does not say cancelled for work a node may still be doing: Bot was asked, and its answer decides.
++ assert.strictEqual(asked.json.state, 'running');
++ assert.strictEqual(bot.callsOf('POST', `/api/v1/jobs/${job.id}/cancel`).length, cancelsBefore + 1);
++ const answer = (await t.dispatch.poll()).find((a) => a.job_id === job.id);
++ assert.strictEqual(answer.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.strictEqual(read1.json.error, null);
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ const rows = await t.outboxRows('run.job.cancelled', job.id);
++ assert.strictEqual(rows.length, 1);
++ assert.strictEqual(rows[0].outbox_key, `run:${job.id}:cancelled`);
++ assert.ok(contracts.validate('run.job.cancelled@1', rows[0].payload).valid);
++ await t.dispatch.poll();
++ assert.strictEqual(bot.callsOf('POST', `/api/v1/jobs/${job.id}/cancel`).length, cancelsBefore + 1, 'one job_cancel per job');
++ });
++
++ await check('ttl_ms running out ends the job expired, and cancels on Bot when a node holds it', async () => {
++ t.setOffers([offer()]);
++ const project = PROJECT();
++ const sentBefore = bot.callsOf('POST', '/api/v1/jobs').length;
++ const job = await submit({ ttl_ms: 1000 }, { project }); // the contract's floor for ttl_ms
++ await t.db.exec('UPDATE run_jobs SET created_at = created_at - 1000000 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(swept.changed, true);
++ assert.strictEqual(bot.callsOf('POST', '/api/v1/jobs').length, sentBefore); // the sweep runs before placement
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.state, 'expired');
++ assert.deepStrictEqual(read1.json.exit, { reason: 'ttl', code: null });
++ assert.strictEqual(read1.json.error.code, 'run.job.ttl');
++ assert.ok(contracts.validate('run.job-read-result@1', read1.json).valid, read1.text);
++ const rows = await t.outboxRows('run.job.expired', job.id);
++ assert.strictEqual(rows.length, 1);
++ assert.strictEqual(rows[0].outbox_key, `run:${job.id}:expired`);
++ assert.ok(contracts.validate('run.job.expired@1', rows[0].payload).valid);
++
++ // A job a node holds: expired, and Bot asked to stop it.
++ const held = await submit({}, { project: PROJECT() });
++ await t.dispatch.poll();
++ bot.start(held.id);
++ await t.dispatch.poll();
++ const before = bot.callsOf('POST', `/api/v1/jobs/${held.id}/cancel`).length;
++ await t.db.exec('UPDATE run_jobs SET created_at = created_at - 10000000 WHERE id = $1', [held.id]);
++ const second = (await t.dispatch.sweep()).find((a) => a.job_id === held.id);
++ assert.strictEqual(second.state, 'expired');
++ assert.strictEqual(bot.callsOf('POST', `/api/v1/jobs/${held.id}/cancel`).length, before + 1);
++ const read2 = await read(held.id, await t.db.value('SELECT project_id FROM run_jobs WHERE id = $1', [held.id]));
++ assert.strictEqual(read2.json.state, 'expired');
++ assert.strictEqual(read2.json.error.code, 'run.job.ttl');
++ assert.strictEqual(read2.json.placement.node, DEVICE); // the placement it had is kept
++ });
++
++ await check('the Bot token is minted once and reused, never per call', async () => {
++ assert.strictEqual(t.network.tokenRequests.n, 1, JSON.stringify(t.network.tokenRequests));
++ assert.deepStrictEqual(t.network.tokenRequests.audiences, ['openvibe.bot']);
++ assert.deepStrictEqual(t.network.tokenRequests.scopes, ['bot.job.dispatch']);
++ });
++
++ await check('the stream accepts a Bearer service token holding run.job.stream, and only that', async () => {
++ t.setOffers([offer()]);
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ await t.dispatch.poll();
++ const token = t.network.signService({ cap: ['run.job.stream'], project_id: project, sub: 'svc:watcher' });
++ const s = await t.sse(`/api/v1/jobs/${job.id}/stream`, { headers: { Authorization: `Bearer ${token}` } });
++ assert.strictEqual(s.status, 200, s.text);
++ const state = await s.waitFor((e) => e.event === 'state');
++ assert.strictEqual(state.data.job_id, job.id);
++ bot.start(job.id);
++ bot.stdout(job.id, 'from the token stream\n');
++ await t.dispatch.poll();
++ const out = await s.waitFor((e) => e.event === 'output');
++ assert.strictEqual(out.data.chunk, 'from the token stream\n');
++ s.close();
++
++ const weak = t.network.signService({ cap: ['run.job.read'], project_id: project, sub: 'svc:watcher' });
++ const denied = await t.sse(`/api/v1/jobs/${job.id}/stream`, { headers: { Authorization: `Bearer ${weak}` } });
++ assert.strictEqual(denied.status, 403, denied.text);
++ assert.strictEqual(denied.json.code, 'capability.denied');
++ const stranger = t.network.signService({ cap: ['run.job.stream'], project_id: PROJECT(), sub: 'svc:watcher' });
++ const hidden = await t.sse(`/api/v1/jobs/${job.id}/stream`, { headers: { Authorization: `Bearer ${stranger}` } });
++ assert.strictEqual(hidden.status, 404, hidden.text); // another project's job: 404, never 403
++ const noProject = t.network.signService({ cap: ['run.job.stream'], sub: 'svc:watcher' });
++ const unowned = await t.sse(`/api/v1/jobs/${job.id}/stream`, { headers: { Authorization: `Bearer ${noProject}` } });
++ assert.strictEqual(unowned.status, 403, unowned.text);
++ assert.strictEqual(unowned.json.code, 'run.project_required');
++ const anonymous = await t.sse(`/api/v1/jobs/${job.id}/stream`);
++ assert.strictEqual(anonymous.status, 401, anonymous.text);
++ assert.strictEqual(anonymous.json.code, 'ticket.required');
++ });
++
++ await check('the loop runs on its own between start() and stop()', async () => {
++ t.setOffers([offer()]);
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ t.config.dispatch.pollMs = 50; // the loop's own interval (RUN_POLL_MS)
++ t.dispatch.start();
++ const deadline = Date.now() + 5000;
++ let state = 'queued';
++ while (Date.now() < deadline) {
++ state = await t.db.value('SELECT state FROM run_jobs WHERE id = $1', [job.id]);
++ if (state !== 'queued') break;
++ await t.wait(25);
++ }
++ const ticks = t.dispatch.status().poller.ticks;
++ t.dispatch.stop();
++ assert.strictEqual(state, 'placed', `the poller placed the job by itself (saw ${state})`);
++ assert.ok(ticks > 0, 'the loop ticked');
++ assert.strictEqual(t.dispatch.status().poller.running, false);
++ });
++
++ await check('Bot never moves Run\'s own placement or project', async () => {
++ t.setOffers([offer()]);
++ const project = PROJECT();
++ const job = await submit({}, { project });
++ // Other checks may have left waiting jobs ahead of this one in the per-tick budget: poll until it is sent.
++ for (let i = 0; i < 10 && !bot.jobs.get(job.id); i++) await t.dispatch.poll();
++ const row = bot.jobs.get(job.id);
++ assert.ok(row, `the job reached Bot (${JSON.stringify((await read(job.id, project)).json.state)})`);
++ row.node_id = OTHER_DEVICE; // Bot claims another node
++ await t.dispatch.poll();
++ const read1 = await read(job.id, project);
++ assert.strictEqual(read1.json.placement.node, DEVICE);
++ row.node_id = DEVICE;
++ row.project_id = PROJECT(); // and another payer
++ await t.dispatch.poll();
++ const read2 = await read(job.id, project);
++ assert.strictEqual(read2.json.project_id, project);
++ assert.strictEqual(read2.json.state, 'placed');
++ row.project_id = project;
++ });
++
++ await check('no secret and no job output is ever logged by the bridge', async () => {
++ const secret = t.config.oauth.clientSecret;
++ assert.ok(secret);
++ assert.ok(!t.logs.some((l) => l.includes(secret)), t.logs.join('\n'));
++ assert.ok(!t.logs.some((l) => /from the token stream|resized 1 of 1/.test(l)), t.logs.join('\n'));
++ });
++ } finally {
++ t.dispatch.stop();
++ await t.close();
++ await bot.close();
++ }
++ done();
++}
++
++main().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/helpers/app.js b/test/helpers/app.js
iindex 22a3434..855e86b 100644
---- a/test/helpers/app.js
++++ b/test/helpers/app.js
@@@ -27,23 +27,67 @@ function listen(server) {
return new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve(`http://127.0.0.1:${server.address().port}`)));
}
--/** OpenVibe.Network's two routes Run uses: the JWKS the key provider loads, and the tokens it verifies. */
++/**
++ * OpenVibe.Network's three routes Run uses: the JWKS the key provider loads, the tokens it verifies, and
++ * the /oauth/token endpoint its own client credentials mint Run's outgoing service tokens from (the
++ * dispatcher's token for audience openvibe.bot). tokenRequests counts the mint calls, so a test can see
++ * that the client caches its token instead of fetching one per call.
++ */
async function startNetwork() {
const { publicKey, privateKey } = crypto.generateKeyPairSync('rsa', { modulusLength: 2048 });
const publicPem = publicKey.export({ type: 'spki', format: 'pem' });
const privatePem = privateKey.export({ type: 'pkcs8', format: 'pem' });
const issuer = 'http://network.test';
++ const tokenRequests = { n: 0, audiences: [], scopes: [] };
++ const offers = []; // what the registry publishes; a test replaces its contents (t.setOffers)
const server = http.createServer((req, res) => {
if (req.url === '/api/.well-known/jwks') {
res.writeHead(200, { 'Content-Type': 'application/json' });
return res.end(JSON.stringify({ keys: [{ kty: 'RSA', kid: 'n1', ...publicKey.export({ format: 'jwk' }) }], public_key: publicPem }));
}
++ // The public resource registry Run places from (server/dispatch/placement.js): the offers a test
++ // publishes with t.setOffers(), filtered the way Network filters them.
++ if (req.method === 'GET' && req.url.startsWith('/api/v1/offers')) {
++ const query = new URL(req.url, 'http://network.test').searchParams;
++ const kind = query.get('kind');
++ const docs = kind ? offers.filter((o) => o.kind === kind) : offers;
++ res.writeHead(200, { 'Content-Type': 'application/json' });
++ return res.end(JSON.stringify({ offers: docs, count: docs.length, generated_at: new Date().toISOString(), filters: kind ? { kind } : {} }));
++ }
++ if (req.method === 'POST' && req.url === '/oauth/token') {
++ let body = '';
++ req.on('data', (chunk) => { body += chunk; });
++ req.on('end', () => {
++ const form = new URLSearchParams(body);
++ const audience = form.get('audience') || 'openvibe.run';
++ const scope = form.get('scope') || '';
++ if (form.get('grant_type') !== 'client_credentials' || !form.get('client_id') || !form.get('client_secret')) {
++ res.writeHead(400, { 'Content-Type': 'application/json' });
++ return res.end(JSON.stringify({ error: 'invalid_client' }));
++ }
++ tokenRequests.n++;
++ tokenRequests.audiences.push(audience);
++ tokenRequests.scopes.push(scope);
++ const now = Math.floor(Date.now() / 1000);
++ const access_token = serviceAuth.signServiceToken({
++ iss: issuer, sub: 'svc:run', actor_type: 'service', aud: [audience],
++ cap: scope ? scope.split(/\s+/).filter(Boolean) : ['bot.job.dispatch'],
++ iat: now, exp: now + 300, jti: crypto.randomBytes(8).toString('hex'),
++ }, privatePem);
++ res.writeHead(200, { 'Content-Type': 'application/json' });
++ res.end(JSON.stringify({ access_token, token_type: 'Bearer', expires_in: 300, scope }));
++ });
++ return undefined;
++ }
res.writeHead(404, { 'Content-Type': 'application/json' });
return res.end('{}');
});
const url = await listen(server);
return {
-- url, issuer, publicPem, privatePem, close: () => new Promise((r) => server.close(r)),
++ url, issuer, publicPem, privatePem, tokenRequests, offers,
++ /** Publish exactly these platform.resource-offer@1 documents (the array keeps its identity). */
++ setOffers(list) { offers.splice(0, offers.length, ...list); },
++ close: () => new Promise((r) => server.close(r)),
/** A Network service token (identity.service-token-claims@1). */
signService({ sub = 'svc:live', aud = ['openvibe.run'], cap = ['run.*'], project_id, actor_type = 'service', expSec = 300, env: tokenEnv } = {}) {
const now = Math.floor(Date.now() / 1000);
@@@ -71,6 +115,8 @@ async function boot(opts = {}) {
OV_OAUTH_CLIENT_ID: 'run',
OV_OAUTH_CLIENT_SECRET: '',
RUN_JOBS: 'off',
++ // No offers cache in tests: a check that changes the published offers (t.setOffers) must see them at once.
++ RUN_OFFERS_TTL_MS: '0',
RUN_STREAM_PRIVATE_KEY: crypto.generateKeyPairSync('rsa', { modulusLength: 2048 }).privateKey.export({ type: 'pkcs8', format: 'pem' }).replace(/\n/g, '\\n'),
...(opts.env || {}),
};
@@@ -208,6 +254,8 @@ async function boot(opts = {}) {
return {
app, base, call, sse, openStream, submit, request, db: app.locals.db, store, config, clock, network, logs, dir, stream, outbox, tickets, runEvents, dispatch,
++ // What the stub registry publishes for placement (server/dispatch/placement.js).
++ setOffers: (list) => network.setOffers(list),
outboxRows, streamState: (jobId) => stream.ringOf(jobId),
wait: (ms) => new Promise((r) => setTimeout(r, ms)),
close: async () => {