openvibers/openvibe.watch Commits

D11 owner settings: watch.provider_rates (#206)

OP OpenVibers committed 8 days ago · 5b2750f444 · 20 files, +1777 −22

* Provider-rate evidence: DeepInfra and Featherless prices, a change a person approves (P6, flag off)

Watch now records the two official provider price documents as supply rate evidence, raises a
change when a rate differs, and lets a person approve it. Nothing here is billable and no rate
card is written: an approval is dated evidence for OpenVibe.AI's catalog operator, who alone
changes the supply rate card.

- Source kind `provider-rate`, two sources: DeepInfra /models/list (documented in its OpenAPI)
  and Featherless /v1/models (OpenAI-compatible), both keyless JSON in the provider matrix.
  Featherless is the one provider whose terms permit resale (Scale), so its supply price matters
  most; DeepInfra is an existing adapter target with an exact keyless list. Each row keeps its
  terms and robots notes and starts disabled; a person enables it.
- Observations: provider, model, token class, nano-USD per million as an integer string (D1),
  currency, region, effective date, the fetched URL and the fetch time. DeepInfra cents/token
  -> x1e13, its cached field is a multiplier and its active discount is applied; Featherless
  USD/token strings -> x1e15. No floats, no missing price defaulted to zero.
- Changed value: one pending change row and a watch.rate.changed event. Unchanged: no new row,
  no change. Malformed: parse_error and no observation; stale/failing health is computed.
- POST .../changes/:id/approve|reject requires a person (usr_...), records who/when, and answers
  applied:false. A service or agent is 403. With WATCH_PROVIDER_RATES=0 no route exists.
- Sources stays the publication ingestion registry; its own docs scope it to what products
  publish, and observed rate changes are Watch's (PLAN B). AI remains the rate-card authority.

Tests: test/provider-rates.test.js (10 cases; recorded fixtures, no live fetch), plus api,
outbox, registry and site tests. ov test green.

* provider-rates test: read the CI schema, not hard-coded public

The PostgreSQL CI store puts each test boot in its own schema, so information_schema with
table_schema = 'public' found nothing. current_schema() is the schema the tables were migrated
into on both PGlite and the PgBouncer store.

* Provider rates: pin openvibe-contracts v0.286.0 and validate the watch.rate.* events against it

v0.286.0 publishes watch.rate.changed@1, watch.rate.approved@1 and watch.rate.rejected@1, the payloads this
branch emits; the test now checks the emitted ones against the release.

* D11 owner settings: watch.provider_rates (shared 4.110.0, contracts 0.288.0)

Changes

ddiff --git a/.env.example b/.env.example
iindex 44f5c06..296006e 100644
---- a/.env.example
++++ b/.env.example
@@@ -119,3 +119,12 @@ VALKEY_PREFIX=
#OV_BILLING_AUDIENCE=
#WATCH_BILLING_TIMEOUT_MS=
#WATCH_BILLING_INTERVAL_MS=
++
++# Provider-rate evidence (server/provider-rates.js, plan P6). 0 (the default) is off: the module is not
++# created, no route is mounted and nothing is seeded. 1 mounts the API and the worker, but a source still
++# starts disabled until a person reads its terms/robots notes and enables it (PATCH .../sources/:key).
++#WATCH_PROVIDER_RATES=0
++#WATCH_PROVIDER_RATES_TICK_MS=60000
++#WATCH_PROVIDER_RATES_MAX_BYTES=8388608
++#WATCH_PROVIDER_RATES_POLL_SEC=21600
++#WATCH_PROVIDER_RATES_STALE_SEC=43200
ddiff --git a/README.md b/README.md
iindex cf2720c..430ab1e 100644
---- a/README.md
++++ b/README.md
@@@ -245,6 +245,71 @@ product (`server/webhook-sources.js`, `server/carriers/webhook.js`).
previous check still runs.
- **Check now** on a webhook watch answers 422 `watch.webhook_awaited`: there is no delivery to read.
++## Provider-rate evidence (`WATCH_PROVIDER_RATES`, off)
++
++PLAN P6 gives Watch/Sources the permitted, provenance-backed market data and observed rate changes; the supply
++rate card itself stays with OpenVibe.AI. This is that observation feed, behind `WATCH_PROVIDER_RATES=1` (off by
++default): two official, machine-readable provider price documents become supply rate observations, a difference
++raises a change, and a person approves it. **The approval is evidence; nothing here is billable and no rate card is
++written.** AI's catalog operator applies an approved change to the supply rate card through AI's own path.
++
++**Why Watch and not Sources.** Sources' own README scopes it to the material the publication products ingest and
++publish ("Products read items with provenance and decide — under their own review and indexability rules — what, if
++anything, to publish"), and `sources.source@1`'s `type` is a released enum. A provider price that may become a
++billable supply rate is not publication content; the observation, the change, the condition and the human decision
++are Watch's model, and PLAN B assigns observed rate/model changes to Watch/Sources jointly. So the source kind
++(`provider-rate`) lives here, and the one ingestion registry stays Sources for product content.
++
++**The two providers** (both in OpenVibe.AI `docs/provider-matrix.md`, both keyless and JSON — no page is scraped):
++
++| Source kind | Provider | Official document | Access in the matrix |
++|---|---|---|---|
++| `provider-rate` | DeepInfra | `https://api.deepinfra.com/models/list` (documented in its OpenAPI) | `RESTRICTED` for resale; evidence only |
++| `provider-rate` | Featherless AI | `https://api.featherless.ai/v1/models` (OpenAI-compatible) | `AUTHORIZED_WHOLESALE` (Scale only); terms §3 permit resale, subject to an agreement |
++
++They are the first two because they publish a price per model in a document a program can read without an account
++or an HTML parse, and they bracket the matrix: Featherless is the one provider whose terms permit resale, so its
++supply price is the one the exchange will most want to know; DeepInfra is an existing OpenAI-compatible adapter
++target with an exact keyless list. Each source's row records its terms and robots notes; **it starts disabled**, and a
++person enables it after reading them even when the flag is on.
++
++**What an observation is** ([server/provider-rates.js](server/provider-rates.js), migration
++`0018_provider_rates.sql`): `provider`, `model`, `token_class` (`input`, `output`, and `cached_read` where the
++document states it), `nano_usd_per_m` as an **integer string** (D1: never a float), `currency`, `region`, the
++`effective_date` (the first day the value was observed; an unchanged value keeps its date, so the rows show effective
++periods), `source_url` (the URL actually fetched) and `fetched_at`.
++
++- **Units.** DeepInfra states cents per token: `× 1e13` is nano-USD per million. Its `rate_per_input_token_cached`
++ is a multiplier (the pricing page renders input × it), not a price, and `discount` is applied while active
++ (`price × (1 − discount)`). Cache-write is not recorded: the document states it only as a retention-dependent
++ multiplier (5m/1h), so no single cache_write rate is invented.
++ Featherless states USD per token as decimal strings: `× 1e15` is nano-USD per million; its per-million display
++ numbers are a fallback only. `region` is `global`: neither document scopes its price by region, so no region is invented.
++- **Changes.** A successful parse writes one immutable observation per `(model, token_class)` and updates the
++ source's validators. When the value (or currency/region) differs from the last observation, the same transaction
++ writes a `provider_rate_changes` row in state `pending` and a `watch.rate.changed` event with
++ `approval_required: true`. An unchanged value is not written again (its first observation already records it and
++ keeps the effective date); the fetch row's `stated` and `observations` counts say what the document said versus what
++ was new, and no change is raised. A new model is recorded but raises no change (there is no previous value to differ
++ from).
++- **Failures.** A malformed document (invalid JSON, a price that is not a finite non-negative decimal, no priced
++ model) records a `parse_error` fetch and **no observation at all** — a missing price is never zero. A 304 is
++ `not_modified`; an HTTP error is `http_error`/`rate_limited`; stale means the last success is older than
++ `stale_after_sec`. The health (`healthy | stale | failing | never_fetched | disabled`) is computed, never guessed,
++ and the last good observations stay visible with their labels.
++- **Approval.** `POST /api/v1/provider-rates/changes/:id/approve` (or `/reject`) requires a **person** (`usr_…`),
++ not a service or an agent, and records who and when; the answer says `applied: false` and the change's `effect` is
++ `evidence`. A second decision is `409`. A `watch.rate.approved`/`watch.rate.rejected` event carries the decision,
++ so AI's operator can apply it, but Watch itself has no rate-card table and no writer for one.
++
++API (mounted only with the flag on; `watch.watch.read` to read, `watch.watch.manage` to enable/approve,
++`watch.check.run` to refresh): `GET /api/v1/provider-rates`, `GET .../sources[/:key]`,
++`GET .../observations?latest=1&source=&model=&token_class=`, `GET .../changes[?state=pending]`,
++`GET .../changes/:id`, `PATCH .../sources/:key` `{enabled}`, `POST .../sources/:key/refresh`,
++`POST .../changes/:id/approve|reject` `{note?}`. The worker fetches enabled sources on
++`WATCH_PROVIDER_RATES_TICK_MS` (60 s) every `poll_interval_sec` (6 h); with the flag off the module is not created,
++no route exists and nothing is seeded.
++
## A check
An in-process scheduler ([server/scheduler.js](server/scheduler.js)) starts checks for active
@@@ -364,6 +429,13 @@ signed-in site holds (`aud: openvibe.network`); a watch is theirs, owned by the
| `watch.resource.read` | `GET /api/v1/resources?project=&kind=&cursor=&limit=`, `GET /api/v1/resources/:ovrn` (first-party service token only) |
| `watch.resource.control` | `POST /api/v1/resources/control` (first-party service token only; ADR-048, [server/api/resource-control.js](server/api/resource-control.js)): a `common.resource-control-request@1` for a watch from OpenVibe.Services acting for its owner (`on_behalf_of`) — `suspend` (pause), `resume`, `delete`; other actions refused `control.unsupported_action`; only the owner's watch in the OVRN's project (otherwise `resources.unknown_resource`); `dry_run` applies nothing; the first answer per (caller, project, `idempotency_key`) is kept (`watch_control_log`, migration 0009) and a key reused for another request is 409. Loopback only: the public vhost 404s `/api/v1/resources*`. |
++Provider-rate evidence (`WATCH_PROVIDER_RATES=1` only; off by default) adds
++`GET /api/v1/provider-rates`, `.../sources[/:key]`, `.../observations`,
++`.../changes[?state=]` and `.../changes/:id` on `watch.watch.read`;
++`PATCH .../sources/:key` and `POST .../changes/:id/approve|reject` on `watch.watch.manage`; and
++`POST .../sources/:key/refresh` on `watch.check.run`. Approve and reject require a person (`usr_…`)
++and answer `applied: false`; no route writes a rate card. With the flag off none of these paths exists.
++
The loopback-only authority index lists non-deleted `watch.watch` resources as
`common.resource-list-result@1`. It orders watches by id, pages with an opaque cursor, and
includes an OVRN only for a watch with a valid project id.
@@@ -475,6 +547,9 @@ Reporting: [SECURITY.md](SECURITY.md). The rules the code keeps:
- checks: `check_runs`, `observations`, the scheduler, conditional-fetch state, backoff
- the public site's sign-in sessions (`web_sessions`) and its pages
- `watch.*` events
++- provider-rate evidence (plan P6, `WATCH_PROVIDER_RATES` off): the DeepInfra and Featherless official price
++ documents as immutable observations, the pending change when a rate differs, and a person's dated approval —
++ never a rate card
## Does not own
@@@ -484,6 +559,8 @@ Reporting: [SECURITY.md](SECURITY.md). The rules the code keeps:
step 7) — Watch asks them, they run
- interpreting a trigger and acting on it (OpenVibe.Actor, OpenVibe.Codes, actions = step 4)
- delivery of a notification (OpenVibe.Network consumes `watch.watch.triggered`)
++- the supply rate card (OpenVibe.AI's catalog): an approved provider-rate change is evidence AI's operator applies
++ through AI's own path, so Watch is never a second authority for a billable price
## Depends on
@@@ -579,7 +656,7 @@ POSTs have their own per-address limit zones (nginx refuses `limit_req` inside `
`sudo ovhost deploy watch`.
<!-- versions:start -->
--- openvibe-contracts: v0.222.0
++- openvibe-contracts: v0.286.0
- openvibe-sdk: v0.82.0
- openvibe-shared: v4.109.2
<!-- versions:end -->
ddiff --git a/STATUS.json b/STATUS.json
iindex 69e2b9b..ee3ac1e 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,7 +7,7 @@
"runtime": "node 22, express, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9, fast-xml-parser",
"port": 4730,
"domain": "openvibe.watch (a public server-rendered site + a loopback-only API; vhost in deploy/nginx; README \"Public site\" and \"Public host\")",
-- "contracts": "openvibe-contracts v0.222.0 (actor-agent requires v0.143.0)",
++ "contracts": "openvibe-contracts v0.286.0 (watch.rate.* events; actor-agent requires v0.143.0)",
"sdk": "openvibe-sdk v0.82.0",
"shared": "openvibe-shared v4.109.2",
"capabilities": [
@@@ -80,7 +80,8 @@
"watch.* events through openvibe-sdk's service_outbox in the same transaction as the state change",
"per-actor limits (openvibe-sdk/limits)",
"usage readings (check operations): one platform.usage-sample@1 check-operation reading per completed check run, relayed to OpenVibe.Billing (billing.usage.record) by openvibe-sdk/usage; off by default (WATCH_BILLING_INTERVAL_MS)",
-- "the public site: Network sign-in (PKCE S256, web_sessions with only the token hash), server-rendered pages complete without JavaScript, the four create templates, watch management, observation history and check runs, how-it-works and the changelog, robots/sitemap/llms.txt"
++ "the public site: Network sign-in (PKCE S256, web_sessions with only the token hash), server-rendered pages complete without JavaScript, the four create templates, watch management, observation history and check runs, how-it-works and the changelog, robots/sitemap/llms.txt",
++ "provider-rate evidence (plan P6, WATCH_PROVIDER_RATES off by default): DeepInfra and Featherless AI official price documents parsed into integer nano-USD-per-million supply rate observations (D1), a pending Watch change when a rate differs, and a person's approval recorded as evidence only — no rate card is written and nothing is billable"
],
-- "updated": "2026-11-03"
++ "updated": "2026-11-04"
}
ddiff --git a/migrations/0018_provider_rates.sql b/migrations/0018_provider_rates.sql
nnew file mode 100644
iindex 0000000..fa07e40
---- /dev/null
++++ b/migrations/0018_provider_rates.sql
@@@ -0,0 +1,110 @@
++-- phase: expand
++-- Provider-rate evidence (plan P6; flag WATCH_PROVIDER_RATES, off by default): the official price documents of the
++-- providers OpenVibe may buy inference from, the rates each document stated, and the changes a person approves.
++--
++-- This is EVIDENCE, not a rate card. There is no billable price here and no writer for one: an approved change is a
++-- dated record of what the provider published and who accepted it, which OpenVibe.AI's operator then applies to the
++-- catalog's supply rate card through AI's own path. "Without a second authority."
++--
++-- nano_usd_per_m is an integer string (D1: integer nano-USD, never a float) per million tokens, for one token class.
++-- A malformed document writes no observation at all — never a zero.
++
++CREATE TABLE provider_rate_sources (
++ key text COLLATE "C" PRIMARY KEY,
++ kind text COLLATE "C" NOT NULL DEFAULT 'provider-rate' CHECK (kind = 'provider-rate'),
++ provider text COLLATE "C" NOT NULL,
++ name text COLLATE "C" NOT NULL,
++ url text COLLATE "C" NOT NULL, -- the official document fetched
++ homepage_url text COLLATE "C",
++ currency text COLLATE "C" NOT NULL, -- the document's currency (USD)
++ region text COLLATE "C" NOT NULL, -- the document's scope (global: not region-gated)
++ terms_note text COLLATE "C" NOT NULL, -- why reading the document is permitted; a person re-reads it
++ robots_note text COLLATE "C" NOT NULL, -- the host's robots.txt at the recorded check
++ parser_version text COLLATE "C" NOT NULL,
++ enabled bigint NOT NULL DEFAULT 0, -- off until a person enables a source, even with the flag on
++ poll_interval_sec bigint NOT NULL,
++ stale_after_sec bigint NOT NULL,
++ -- runtime state
++ next_due_at bigint NOT NULL DEFAULT 0,
++ not_before bigint NOT NULL DEFAULT 0, -- Retry-After
++ last_fetch_at bigint,
++ last_success_at bigint,
++ last_state text COLLATE "C",
++ consecutive_failures bigint NOT NULL DEFAULT 0,
++ etag text COLLATE "C",
++ last_modified text COLLATE "C",
++ last_body_hash text COLLATE "C",
++ created_at bigint NOT NULL,
++ updated_at bigint NOT NULL,
++ updated_by text COLLATE "C"
++);
++
++-- One row per fetch attempt, whatever happened. A failed fetch writes no observation and no change.
++CREATE TABLE provider_rate_fetches (
++ rid bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ id text COLLATE "C" NOT NULL UNIQUE, -- prf_<ULID>
++ source_key text COLLATE "C" NOT NULL,
++ trigger text COLLATE "C" NOT NULL CHECK (trigger IN ('schedule','manual')),
++ started_at bigint NOT NULL,
++ finished_at bigint NOT NULL,
++ state text COLLATE "C" NOT NULL CHECK (state IN ('ok','not_modified','http_error','timeout','parse_error','rate_limited','disabled')),
++ http_status bigint,
++ error_code text COLLATE "C",
++ detail text COLLATE "C",
++ bytes bigint,
++ raw_body_hash text COLLATE "C",
++ parser_version text COLLATE "C",
++ stated bigint NOT NULL DEFAULT 0, -- rates the document stated
++ observations bigint NOT NULL DEFAULT 0, -- new observations stored (new or changed values)
++ changed bigint NOT NULL DEFAULT 0
++);
++CREATE INDEX idx_provider_rate_fetches_source ON provider_rate_fetches (source_key, rid DESC);
++
++-- One immutable observation per (model, token class) per successful fetch that stated it.
++CREATE TABLE provider_rate_observations (
++ rid bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ id text COLLATE "C" NOT NULL UNIQUE, -- pro_<ULID>
++ source_key text COLLATE "C" NOT NULL,
++ fetch_id text COLLATE "C" NOT NULL,
++ provider text COLLATE "C" NOT NULL,
++ model text COLLATE "C" NOT NULL,
++ token_class text COLLATE "C" NOT NULL CHECK (token_class IN ('input','cached_read','cache_write','output')),
++ nano_usd_per_m text COLLATE "C" NOT NULL CHECK (nano_usd_per_m ~ '^[0-9]+$'), -- D1: integer string
++ currency text COLLATE "C" NOT NULL,
++ region text COLLATE "C" NOT NULL,
++ effective_date text COLLATE "C" NOT NULL, -- YYYY-MM-DD; the first day this value was observed
++ source_url text COLLATE "C" NOT NULL,
++ fetched_at bigint NOT NULL,
++ value_hash text COLLATE "C" NOT NULL, -- hashValue({nano_usd_per_m,currency,region})
++ changed bigint NOT NULL DEFAULT 0,
++ previous_id text COLLATE "C",
++ previous_nano_usd_per_m text COLLATE "C"
++);
++CREATE INDEX idx_provider_rate_observations_key ON provider_rate_observations (source_key, model, token_class, rid DESC);
++CREATE INDEX idx_provider_rate_observations_rid ON provider_rate_observations (rid DESC);
++
++-- A change Watch raised for a person to approve: one per observation that differs from the last one.
++CREATE TABLE provider_rate_changes (
++ rid bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ id text COLLATE "C" NOT NULL UNIQUE, -- prc_<ULID>
++ source_key text COLLATE "C" NOT NULL,
++ observation_id text COLLATE "C" NOT NULL,
++ previous_id text COLLATE "C",
++ provider text COLLATE "C" NOT NULL,
++ model text COLLATE "C" NOT NULL,
++ token_class text COLLATE "C" NOT NULL,
++ from_nano_usd_per_m text COLLATE "C",
++ to_nano_usd_per_m text COLLATE "C" NOT NULL CHECK (to_nano_usd_per_m ~ '^[0-9]+$'),
++ currency text COLLATE "C" NOT NULL,
++ region text COLLATE "C" NOT NULL,
++ effective_date text COLLATE "C" NOT NULL,
++ source_url text COLLATE "C" NOT NULL,
++ fetched_at bigint NOT NULL,
++ state text COLLATE "C" NOT NULL DEFAULT 'pending' CHECK (state IN ('pending','approved','rejected')),
++ created_at bigint NOT NULL,
++ decided_at bigint,
++ decided_by text COLLATE "C", -- a person (usr_…): only a person approves
++ note text COLLATE "C"
++);
++CREATE INDEX idx_provider_rate_changes_state ON provider_rate_changes (state, rid DESC);
++CREATE INDEX idx_provider_rate_changes_source ON provider_rate_changes (source_key, rid DESC);
ddiff --git a/migrations/0019_config.sql b/migrations/0019_config.sql
nnew file mode 100644
iindex 0000000..49d9764
---- /dev/null
++++ b/migrations/0019_config.sql
@@@ -0,0 +1,28 @@
++-- phase: expand
++-- D11 owner settings: openvibe-shared/config (config_snapshots, config_keys). Applied migrations are immutable.
++CREATE TABLE IF NOT EXISTS config_snapshots (
++ namespace text COLLATE "C" NOT NULL,
++ revision bigint NOT NULL,
++ service text COLLATE "C" NOT NULL,
++ previous_revision bigint,
++ state text COLLATE "C" NOT NULL CHECK (state IN ('proposed', 'active', 'superseded', 'rejected', 'rolled_back')),
++ values_json text COLLATE "C" NOT NULL, -- the full values; secret-class values are envelopes (never their plaintext)
++ classification_json text COLLATE "C" NOT NULL,
++ values_checksum text COLLATE "C" NOT NULL, -- sha256 of the values as served (secrets as markers)
++ created_at text COLLATE "C" NOT NULL,
++ created_by text COLLATE "C" NOT NULL,
++ activated_at text COLLATE "C",
++ activated_by text COLLATE "C",
++ reason text COLLATE "C",
++ error text COLLATE "C",
++ copied_from bigint,
++ good bigint NOT NULL DEFAULT 0, -- 1 once it activated successfully (last-known-good)
++ PRIMARY KEY (namespace, revision)
++);
++CREATE UNIQUE INDEX IF NOT EXISTS config_snapshots_one_active ON config_snapshots (namespace) WHERE state = 'active';
++-- The fingerprint key of each namespace. Never served, never logged.
++CREATE TABLE IF NOT EXISTS config_keys (
++ namespace text COLLATE "C" PRIMARY KEY,
++ hmac_key bytea NOT NULL, -- the 32-byte fingerprint key, or that key wrapped under the current KEK
++ created_at text COLLATE "C" NOT NULL
++);
ddiff --git a/package-lock.json b/package-lock.json
iindex 40c49c1..76de6ae 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,9 +12,9 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.288.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.82.0",
-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.109.2",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.110.0",
"pg": "^8.23.0"
},
"devDependencies": {
@@@ -1044,9 +1044,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.222.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.0",
-- "integrity": "sha512-wdUQCilq9glRMwfxEswbIj3l0QQqKDXZAi2+CR5Mlh34rZjI1Q+7GlAoR0UtNwZAgtLwXhPxFyJ/aPk8lcDG1Q==",
++ "version": "0.288.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.288.0",
++ "integrity": "sha512-AKRP1yXACYNu7okpf4kCDq4ds0PhsoNvD92VlvBWBCjcSlNimSTQhLWIvSa3D31/kQYz91fodo06ZnwuTanOqg==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
@@@ -1088,9 +1088,9 @@
}
},
"node_modules/openvibe-shared": {
-- "version": "4.109.2",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.109.2",
-- "integrity": "sha512-t9K1yHr6GOJsfamqmgERRTTtiWYWAVmjWBEm4ykTcYjEenbTiqQTwX6CJsZW4lYnqS1IlqEvn+0jG4vXmHSJ3Q==",
++ "version": "4.110.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.110.0",
++ "integrity": "sha512-ZNOSraSgE89jhLTI//lFJALvKCiBJIBwf+V0aCkiIk0VCrPPAytsfsgOgCINU8kXDVdCEnuWWfcA9hZUc54QVA==",
"license": "MIT",
"dependencies": {
"jsonwebtoken": "^9.0.0",
ddiff --git a/package.json b/package.json
iindex 8ef825f..4d127d7 100644
---- a/package.json
++++ b/package.json
@@@ -24,9 +24,9 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.288.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.82.0",
-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.109.2",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.110.0",
"pg": "^8.23.0"
},
"engines": {
ddiff --git a/server/api/provider-rates.js b/server/api/provider-rates.js
nnew file mode 100644
iindex 0000000..94c2821
---- /dev/null
++++ b/server/api/provider-rates.js
@@@ -0,0 +1,172 @@
++'use strict';
++/**
++ * Provider-rate evidence API (plan P6; only mounted when WATCH_PROVIDER_RATES=1, off by default).
++ *
++ * watch.watch.read
++ * GET /api/v1/provider-rates the two sources, their health and the pending count
++ * GET /api/v1/provider-rates/sources
++ * GET /api/v1/provider-rates/sources/:key
++ * GET /api/v1/provider-rates/observations?latest=&source=&model=&token_class=&before=&limit=
++ * GET /api/v1/provider-rates/changes?state=&source=&before=&limit=
++ * GET /api/v1/provider-rates/changes/:id
++ * watch.watch.manage
++ * PATCH /api/v1/provider-rates/sources/:key {enabled} — a person enables a source a person reviewed
++ * POST /api/v1/provider-rates/changes/:id/approve {note?} — a PERSON approves; records evidence only
++ * POST /api/v1/provider-rates/changes/:id/reject {note?}
++ * watch.check.run
++ * POST /api/v1/provider-rates/sources/:key/refresh fetch one source now (same code the worker runs)
++ *
++ * No route writes a rate card. An approval answers `applied: false`: OpenVibe.AI's operator applies the approved
++ * observation to the catalog's supply rate card through AI's own path, so Watch never becomes a second authority.
++ */
++const express = require('express');
++const { http } = require('openvibe-contracts');
++const { CAPS } = require('../auth');
++
++const STATES = ['pending', 'approved', 'rejected'];
++
++function providerRatesRouter({ providerRates, auth, limits }) {
++ const router = express.Router();
++ // Owner settings (D11): the module is always constructed now, so the switch is read live. Off, its routes are
++ // the API's ordinary 404 and no source row is written; the first enabled request seeds the two sources.
++ router.use(async (req, res, next) => {
++ if (!providerRates.enabled) return next('router');
++ try { await providerRates.init(); next(); } catch (err) { next(err); }
++ });
++ const read = [auth.requireCap(CAPS.read), limits.reads('watch.read')];
++ const manage = [auth.requireCap(CAPS.manage), limits.budget('watch.write')];
++ const runNow = [auth.requireCap(CAPS.run), limits.budget('watch.check.run')];
++
++ const notFound = (res, req, what = 'provider-rate source') => http.sendProblem(res, 404, 'watch.rate_not_found', { detail: `no such ${what}`, ctx: req.ov });
++ const pageLimit = (q, d = 100, max = 500) => Math.min(Math.max(parseInt(q, 10) || d, 1), max);
++ const before = (q) => (/^\d{1,18}$/.test(String(q || '')) ? Number(q) : null);
++
++ /** A decision is a person's: a service or an agent may watch and fetch, never approve. */
++ function personOnly(req, res, next) {
++ if (!req.principal || req.principal.kind !== 'person' || !/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(req.principal.sub || ''))) {
++ return http.sendProblem(res, 403, 'watch.rate_person_only', {
++ detail: 'a person approves or rejects a provider-rate change; a service or an agent may not', ctx: req.ov,
++ });
++ }
++ next();
++ }
++
++ function noteOf(body) {
++ if (body === undefined || body === null || body === '') return null;
++ if (typeof body !== 'object' || Array.isArray(body)) return { error: 'the body must be an object with an optional note' };
++ for (const k of Object.keys(body)) if (k !== 'note') return { error: `unknown field ${k}` };
++ if (body.note === undefined || body.note === null) return { note: null };
++ if (typeof body.note !== 'string' || body.note.length > 2000) return { error: 'note must be a string of at most 2000 characters' };
++ return { note: body.note || null };
++ }
++
++ router.get('/api/v1/provider-rates', read, async (req, res) => {
++ const rows = await providerRates.sources();
++ res.json({
++ sources: rows.map((r) => providerRates.sourceView(r)),
++ pending_changes: await providerRates.pendingCount(),
++ });
++ });
++
++ router.get('/api/v1/provider-rates/sources', read, async (req, res) => {
++ res.json({ sources: (await providerRates.sources()).map((r) => providerRates.sourceView(r)) });
++ });
++
++ router.get('/api/v1/provider-rates/sources/:key', read, async (req, res) => {
++ const row = await providerRates.source(req.params.key);
++ if (!row) return notFound(res, req);
++ res.json({ source: providerRates.sourceView(row) });
++ });
++
++ router.patch('/api/v1/provider-rates/sources/:key', manage, async (req, res) => {
++ const body = req.body;
++ if (!body || typeof body !== 'object' || Array.isArray(body)) {
++ return http.sendProblem(res, 422, 'watch.rate_invalid', { detail: 'body must be an object with enabled', ctx: req.ov });
++ }
++ for (const k of Object.keys(body)) {
++ if (k !== 'enabled') return http.sendProblem(res, 422, 'watch.rate_invalid', { detail: `unknown field ${k}`, ctx: req.ov });
++ }
++ try {
++ const row = await providerRates.setEnabled(req.params.key, body.enabled, req.principal);
++ if (!row) return notFound(res, req);
++ res.json({ source: providerRates.sourceView(row) });
++ } catch (err) {
++ if (err && err.code === 'watch.rate_invalid') return http.sendProblem(res, 422, err.code, { detail: err.message, ctx: req.ov });
++ throw err;
++ }
++ });
++
++ router.get('/api/v1/provider-rates/observations', read, async (req, res) => {
++ const limit = pageLimit(req.query.limit);
++ const latest = ['1', 'true'].includes(String(req.query.latest || ''));
++ const rows = await providerRates.observations({
++ latest,
++ source: req.query.source ? String(req.query.source) : null,
++ model: req.query.model ? String(req.query.model) : null,
++ tokenClass: req.query.token_class ? String(req.query.token_class) : null,
++ before: before(req.query.before),
++ limit,
++ });
++ res.json({
++ observations: rows.map(providerRates.observationView),
++ next_before: !latest && rows.length === limit ? rows[rows.length - 1].rid : null,
++ });
++ });
++
++ router.get('/api/v1/provider-rates/changes', read, async (req, res) => {
++ const state = req.query.state ? String(req.query.state) : null;
++ if (state && !STATES.includes(state)) {
++ return http.sendProblem(res, 400, 'watch.rate_bad_query', { detail: `state must be ${STATES.join('|')}`, ctx: req.ov });
++ }
++ const limit = pageLimit(req.query.limit);
++ const rows = await providerRates.changes({
++ state, source: req.query.source ? String(req.query.source) : null, before: before(req.query.before), limit,
++ });
++ res.json({ changes: rows.map(providerRates.changeView), next_before: rows.length === limit ? rows[rows.length - 1].rid : null });
++ });
++
++ router.get('/api/v1/provider-rates/changes/:id', read, async (req, res) => {
++ const row = await providerRates.change(req.params.id);
++ if (!row) return notFound(res, req, 'rate change');
++ res.json({ change: providerRates.changeView(row) });
++ });
++
++ router.post('/api/v1/provider-rates/sources/:key/refresh', runNow, async (req, res) => {
++ const out = await providerRates.fetchNow(req.params.key, { trigger: 'manual' });
++ if (!out) return notFound(res, req);
++ if (out.busy) return http.sendProblem(res, 409, 'watch.rate_busy', { detail: 'a fetch of this source is already running', ctx: req.ov });
++ if (out.disabled) return http.sendProblem(res, 409, 'watch.rate_disabled', { detail: 'the source is disabled: enable it first', ctx: req.ov });
++ if (out.unavailable) return http.sendProblem(res, 422, 'watch.rate_unavailable', { detail: out.unavailable, ctx: req.ov });
++ res.json(out);
++ });
++
++ /** Approve or reject one pending change. Both are the same decision path; only the state and event type differ. */
++ function decision(choice) {
++ return async (req, res) => {
++ const body = noteOf(req.body);
++ if (body.error) return http.sendProblem(res, 422, 'watch.rate_invalid', { detail: body.error, ctx: req.ov });
++ try {
++ const out = await providerRates.decide(req.params.id, choice, req.principal, { note: body.note });
++ if (!out) return notFound(res, req, 'rate change');
++ if (out.conflict) {
++ return http.sendProblem(res, 409, 'watch.rate_decided', {
++ detail: `this change was already ${out.conflict}`, ctx: req.ov,
++ });
++ }
++ res.json(out);
++ } catch (err) {
++ if (err && err.code === 'watch.rate_decided') return http.sendProblem(res, 409, err.code, { detail: err.message, ctx: req.ov });
++ if (err && err.code === 'watch.rate_person_only') return http.sendProblem(res, 403, err.code, { detail: err.message, ctx: req.ov });
++ if (err && err.code === 'watch.rate_invalid') return http.sendProblem(res, 422, err.code, { detail: err.message, ctx: req.ov });
++ throw err;
++ }
++ };
++ }
++
++ router.post('/api/v1/provider-rates/changes/:id/approve', manage, personOnly, decision('approved'));
++ router.post('/api/v1/provider-rates/changes/:id/reject', manage, personOnly, decision('rejected'));
++
++ return router;
++}
++
++module.exports = { providerRatesRouter };
ddiff --git a/server/app.js b/server/app.js
iindex 35e972e..a653a33 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -24,6 +24,8 @@ const { watchesRouter } = require('./api/watches');
const { resourceIndexRouter } = require('./api/resource-index');
const { resourceControlRouter } = require('./api/resource-control');
const { checksRouter } = require('./api/checks');
++const { providerRatesRouter } = require('./api/provider-rates');
++const createProviderRatesConfig = require('./provider-rates-config').create;
const { createActorLimits } = require('./api/actor-limits');
const { assetVersion, send, setRelease, SITE_NAME, TAGLINE } = require('./render/layout');
const { html } = require('./render/html');
@@@ -59,7 +61,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv
+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "
+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";
--function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null, webhookSources = null, billing = null }) {
++async function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null, webhookSources = null, billing = null, providerRates = null, env = process.env, onConfigActivate = null }) {
const app = express();
// GET /openapi.json (SEO-3): this service's public API as OpenAPI 3.1, generated from the contracts
// (openvibe-contracts); the operations they mark internal-only are left out.
@@@ -89,6 +91,11 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
const metrics = instrument(app, { service: 'watch', release: release.release });
app.use(require('openvibe-sdk/service').telemetryMiddleware); // request telemetry: a no-op until server/index.js calls telemetry.init()
registerWatchGauges(metrics.registry, { db, watches: registry, check, observations, outbox, now });
++ // D11 owner settings (watch.provider_rates): the switch is read through the store, env pins winning.
++ const providerRatesConfig = await createProviderRatesConfig({ db, auth, config, env, log, onActivate: onConfigActivate });
++ Object.defineProperty(config.providerRates, 'enabled', { enumerable: true, configurable: true, get: () => providerRatesConfig.get('enabled') });
++ providerRatesConfig.mount(app);
++
app.use(http.middleware());
// One W3C trace across services (openvibe-shared/trace): calls made while serving a request carry its traceparent.
require('openvibe-shared/trace').install(app);
@@@ -170,6 +177,9 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
app.use(resourceControlRouter({ db, auth, registry, now }));
app.use(watchesRouter({ registry, auth, limits, webhookSources }));
app.use(checksRouter({ registry, check, observations, auth, limits }));
++ // Provider-rate evidence (plan P6, server/api/provider-rates.js): mounted only with WATCH_PROVIDER_RATES=1. With
++ // the flag off every path below is the API's ordinary 404 and no timer exists.
++ if (providerRates) app.use(providerRatesRouter({ providerRates, auth, limits }));
app.use('/api', (req, res) => http.sendProblem(res, 404, 'watch.not_found', { detail: `no route ${req.method} ${req.path}`, ctx: req.ov }));
// ── Pages ───────────────────────────────────────────────────────────────────────────────
ddiff --git a/server/config.js b/server/config.js
iindex 4c00796..cf91dcf 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -173,6 +173,18 @@ function load(env = process.env) {
},
},
++ // Provider-rate evidence (server/provider-rates.js, plan P6): the official price documents of DeepInfra and
++ // Featherless AI, fetched only when the flag is on and a person has enabled the source. Off by default; the
++ // module is not created when off. 8 MiB covers Featherless' ~3.2 MB model list; a fetch every 6 hours is
++ // polite, and a source is stale after twice that. No rate card is written here (README).
++ providerRates: {
++ enabled: env.WATCH_PROVIDER_RATES === '1',
++ tickMs: Math.max(1000, int(env.WATCH_PROVIDER_RATES_TICK_MS, 60_000)),
++ maxBytes: Math.min(32 * 1024 * 1024, Math.max(1024, int(env.WATCH_PROVIDER_RATES_MAX_BYTES, 8 * 1024 * 1024))),
++ pollIntervalSec: Math.max(300, int(env.WATCH_PROVIDER_RATES_POLL_SEC, 6 * 3600)),
++ staleAfterSec: Math.max(600, int(env.WATCH_PROVIDER_RATES_STALE_SEC, 12 * 3600)),
++ },
++
// Usage readings → OpenVibe.Billing (server/billing.js, plan T5 step 14): one check-operation reading per
// completed check, relayed every WATCH_BILLING_INTERVAL_MS with Watch's own client-credentials token. 0 (the
// default) is off: nothing is queued and no timer starts. Without OV_BILLING_INTERNAL_URL or the client secret,
ddiff --git a/server/index.js b/server/index.js
iindex eeae3cf..012a464 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -29,6 +29,7 @@ const { createWatchBilling } = require('./billing');
const { createScheduler } = require('./scheduler');
const { createEventSources } = require('./event-sources');
const { createWebhookSources } = require('./webhook-sources');
++const { createProviderRates } = require('./provider-rates');
const { createServiceOutbox } = require('openvibe-sdk/events');
const { validate } = require('openvibe-contracts');
const { jwksClient, createServiceTokenClient } = require('openvibe-sdk/auth');
@@@ -45,7 +46,7 @@ const { createApp } = require('./app');
const accountDataLib = require('./account-data');
const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null, actor: givenActor = null, run: givenRun = null, codes: givenCodes = null, projects: givenProjects = null, machines: givenMachines = null, projectSecrets: givenProjectSecrets = null } = {}) {
++async function start({ config, env = process.env, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null, actor: givenActor = null, run: givenRun = null, codes: givenCodes = null, projects: givenProjects = null, machines: givenMachines = null, projectSecrets: givenProjectSecrets = null, providerRateProviders = null } = {}) {
config = config || load();
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
const db = givenDb || await openDb(config, { log });
@@@ -54,7 +55,8 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
intervalMs: config.events.relayIntervalMs, fetch: fetchImpl, log, now, table: 'service_outbox',
eventTypes: ['watch.watch.created', 'watch.watch.updated', 'watch.watch.paused', 'watch.watch.removed',
-- 'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed'],
++ 'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed',
++ 'watch.rate.changed', 'watch.rate.approved', 'watch.rate.rejected'],
validate: (env) => validate('events.event-envelope@1', env),
});
// Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own
@@@ -94,6 +96,12 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const billing = createWatchBilling({ db, config: config.billing, networkUrl: config.networkInternalUrl, clock: { now }, fetchImpl, log });
const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, billing, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
++ // Provider-rate evidence (plan P6): created only behind WATCH_PROVIDER_RATES; the rows are seeded disabled and a
++ // person enables a source after reading its terms. The flag alone never fetches anything.
++ // D11 owner settings: always constructed so the store's switch applies live; off, it seeds nothing until an
++ // enabled request (server/api/provider-rates.js) or an owner change (onConfigActivate below) needs it.
++ const providerRates = createProviderRates({ db, config, fetcher, spacer, outbox, now, log, ...(providerRateProviders ? { providers: providerRateProviders } : {}) });
++ if (providerRates.enabled) await providerRates.init();
// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.
const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });
registry.onEventSourceChange(() => eventSources.kick());
@@@ -111,7 +119,13 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const accountSend = givenSend || (config.oauth.clientSecret
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
: null);
-- const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources, webhookSources, billing });
++ const app = await createApp({ config, env, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources, webhookSources, billing, providerRates,
++ onConfigActivate: async (values) => {
++ if (!providerRates) return;
++ if (!values.enabled) return providerRates.stop();
++ await providerRates.init();
++ if (config.worker.enabled) providerRates.start();
++ } });
// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background
// on an unref'd timer, keeping the last good keys through a Network outage.
keys.start();
@@@ -120,6 +134,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
if (config.worker.enabled) scheduler.start();
actions.start();
if (billing.enabled) billing.start();
++ if (providerRates && config.worker.enabled) providerRates.start();
const pruneTimer = setInterval(async () => {
try { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }
try { await eventSources.prune(); } catch (err) { log.error(`[event-sources] prune: ${err.message}`); }
@@@ -157,6 +172,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
eventSources.stop();
keys.stop();
await scheduler.stop();
++ if (providerRates) await providerRates.stop();
await actions.stop();
await billing.stop();
await outbox.stop();
@@@ -167,7 +183,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
if (!givenDb) await db.close();
}
-- return { config, db, registry, eventSources, webhookSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, billing, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
++ return { config, db, registry, eventSources, webhookSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, billing, providerRates, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
}
/**
ddiff --git a/server/provider-rates-config.js b/server/provider-rates-config.js
nnew file mode 100644
iindex 0000000..344a406
---- /dev/null
++++ b/server/provider-rates-config.js
@@@ -0,0 +1,89 @@
++'use strict';
++/**
++ * The `watch.provider_rates` configuration namespace (D11 owner settings): the one switch the owner changes from
++ * openvibe.network's AI tab. WATCH_PROVIDER_RATES stays the pin: while it is set it wins, is read-only in the
++ * summary and a write answers 409 config.pinned_in_env.
++ *
++ * /internal/config is openvibe-shared/config's adminRoutes, guarded by watch.config.manage (Network's service
++ * token only). actor is the owner Network names in X-OV-Subject; X-OV-Fresh: 1 gates a secret or x-danger write
++ * (this namespace has neither). Every activation writes one audit line with the actor and the keys, never a value.
++ */
++const express = require('express');
++const shared = require('openvibe-shared/config');
++const { capabilities, http, ids } = require('openvibe-contracts');
++
++const SERVICE = 'watch';
++const NAMESPACE = 'watch.provider_rates';
++const CAPABILITY = 'watch.config.manage';
++const GROUP = 'Provider-rate evidence';
++
++const SCHEMA = {
++ $schema: 'https://json-schema.org/draft/2020-12/schema',
++ title: 'Provider-rate evidence',
++ description: 'Read DeepInfra\'s and Featherless\' published price documents and record what they stated, with the change a person approves.',
++ type: 'object',
++ additionalProperties: false,
++ required: ['enabled'],
++ properties: {
++ enabled: {
++ type: 'boolean', title: 'Provider-rate evidence', 'x-group': GROUP, 'x-env': 'WATCH_PROVIDER_RATES',
++ description: 'Offer /api/v1/provider-rates and run the observation worker. A source still starts disabled until a person enables it after reading its terms.',
++ },
++ },
++};
++const DEFAULTS = Object.freeze({ enabled: false });
++
++function bearer(req) {
++ const h = String((req.headers && req.headers.authorization) || '');
++ return h.startsWith('Bearer ') ? h.slice(7).trim() : null;
++}
++
++function capabilityGuard({ auth, log }) {
++ return async function guard(req, res, next) {
++ try {
++ const token = bearer(req);
++ if (!token) return http.sendProblem(res, 401, 'token.missing', { detail: 'a Network service token is required', ctx: req.ov });
++ const r = await auth.verifyService(token);
++ if (!r.ok) return http.sendProblem(res, r.code === 'token.unavailable' ? 503 : 401, r.code || 'token.invalid', { detail: r.reason || 'the token does not verify', ctx: req.ov });
++ const sub = String((r.claims && r.claims.sub) || '');
++ if (!sub.startsWith('svc:')) return http.sendProblem(res, 403, 'capability.denied', { detail: `${CAPABILITY} is first-party; an app may not change Watch configuration`, ctx: req.ov });
++ const decision = capabilities.check(r.claims, CAPABILITY);
++ if (!decision.allowed) return http.sendProblem(res, 403, decision.code || 'capability.denied', { detail: decision.reason || `${CAPABILITY} not granted`, ctx: req.ov });
++ req.configClaims = r.claims;
++ return next();
++ } catch (err) {
++ log.error(`[config] watch.provider_rates guard failed: ${(err && err.message) || err}`);
++ return http.sendProblem(res, 500, 'internal.error', { detail: 'the configuration request failed', ctx: req.ov });
++ }
++ };
++}
++
++function actor(req) {
++ const named = String((req.headers && req.headers['x-ov-subject']) || '').trim();
++ return ids.isSubjectId('user', named) ? { type: 'user', id: named } : null;
++}
++
++const requireFresh = (req) => String((req.headers && req.headers['x-ov-fresh']) || '') === '1';
++
++async function create({ db, auth, config, env = process.env, log = console, onActivate = null }) {
++ const store = await shared.createConfigStore({
++ db, service: SERVICE, namespace: NAMESPACE, schema: SCHEMA, defaults: DEFAULTS, env, log,
++ onActivate: async (values) => { if (onActivate) await onActivate(values); },
++ onChange: ({ namespace, keys, actor: who, revision }) => {
++ log.info(`[config] ${SERVICE} ${namespace} revision ${revision} by ${who && who.id ? `${who.type}:${who.id}` : 'unknown'}: ${keys.join(', ')}`);
++ },
++ });
++ if (env.OV_CONFIG_KEKS) {
++ try { await store.reseal(); } catch (err) { log.warn(`[config] ${NAMESPACE}: reseal failed: ${(err && err.message) || err}`); }
++ }
++ const router = shared.adminRoutes([store], {
++ basePath: '/internal/config', requireAdmin: capabilityGuard({ auth, log }), actor, requireFresh, log,
++ });
++ return {
++ store, router,
++ get: (key) => store.get(key),
++ mount(app) { app.use('/internal/config', express.json({ limit: '128kb' })); router.mount(app); },
++ };
++}
++
++module.exports = { create, SCHEMA, NAMESPACE, CAPABILITY };
ddiff --git a/server/provider-rates.js b/server/provider-rates.js
nnew file mode 100644
iindex 0000000..0ceab6f
---- /dev/null
++++ b/server/provider-rates.js
@@@ -0,0 +1,674 @@
++'use strict';
++/**
++ * Provider-rate evidence (plan P6; flag WATCH_PROVIDER_RATES, off by default).
++ *
++ * Two providers publish their prices as a machine-readable document that needs no account: DeepInfra's
++ * `/models/list` (documented in its OpenAPI) and Featherless AI's OpenAI-compatible `/v1/models`. Both are in
++ * OpenVibe.AI's provider matrix; Featherless is the one provider whose terms permit resale (Scale, terms §3), so
++ * its supply price is the one the exchange will most want to know; DeepInfra is an existing OpenAI-compatible
++ * adapter target with an exact, keyless price list. This module records what those documents state as evidence:
++ *
++ * provider, model, token class, nano-USD per million (an integer STRING, D1), currency, region, effective date,
++ * the URL fetched and the time it was fetched.
++ *
++ * Rate cards are NOT written here and nothing here is billable. When a rate differs from the last observation,
++ * Watch raises a `provider_rate_changes` row (a change) and a `watch.rate.changed` event; a person approves or
++ * rejects it. The approval is dated evidence for OpenVibe.AI's catalog operator, who alone changes the supply rate
++ * card through AI's own path — no second authority. An approval response says `applied: false` for that reason.
++ *
++ * Units, exactly:
++ * DeepInfra cents_per_input_token cents/token × 1e13 = nano-USD per million tokens
++ * cents_per_output_token
++ * rate_per_input_token_cached a multiplier (the pricing page renders input × it), not a price
++ * discount, discount_ends_at a fraction off, while active: price × (1 - discount)
++ * Featherless pricing.prompt/completion USD/token as decimal strings × 1e15 = nano-USD per million tokens
++ * pricing.input/output USD per million numbers, read only when the string is absent
++ *
++ * A malformed document (invalid JSON, a price that is not a finite non-negative decimal, no priced model) records
++ * a `parse_error` fetch and NO observation at all — a missing or unreadable price never becomes zero. An unchanged
++ * value is not written again: its first observation keeps the effective date, and the fetch row's stated/observations
++ * counts say what the document stated versus what was new. An unchanged document raises no change. The effective date
++ * of a rate is the first day it was observed at that value, so the record shows effective periods.
++ */
++const crypto = require('crypto');
++const { ids: contractIds } = require('openvibe-contracts');
++const { hashValue, ParseError, decodeBody } = require('./util');
++const { envelope } = require('./events/envelope');
++
++/** The token classes a document may state (D1: input, cached read, cache write, output). */
++const TOKEN_CLASSES = ['input', 'cached_read', 'cache_write', 'output'];
++/** Fetch states that are failures (health `failing`, backoff). */
++const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_limited', 'disabled']);
++/** A provider's published document is one of these. `kind` is the source kind this module registers. */
++const KIND = 'provider-rate';
++
++class ProviderRateError extends Error {
++ constructor(detail, code = 'watch.rate_invalid') { super(detail); this.code = code; }
++}
++
++// ── Decimal arithmetic (D1: never a float) ────────────────────────────────────────────────────────
++
++const DECIMAL = /^([0-9]+)(?:\.([0-9]+))?(?:[eE]([+-]?[0-9]{1,6}))?$/;
++const TEN = 10n;
++
++/** A non-negative decimal string or number → its exact value × 10^scale, rounded half-up to an integer. */
++function scaled(value, scale, what) {
++ let s;
++ if (typeof value === 'number') {
++ if (!Number.isFinite(value) || value < 0) throw new ParseError(`${what} is not a finite non-negative number`);
++ s = String(value);
++ } else if (typeof value === 'string') {
++ s = value.trim();
++ } else {
++ throw new ParseError(`${what} is not a decimal`);
++ }
++ const m = DECIMAL.exec(s);
++ if (!m) throw new ParseError(`${what} is not a finite non-negative decimal: ${JSON.stringify(s).slice(0, 40)}`);
++ const digits = m[1] + (m[2] || '');
++ let n = BigInt(digits);
++ if (n === 0n) return 0n;
++ const exp = (m[3] ? Number(m[3]) : 0) - (m[2] ? m[2].length : 0) + scale;
++ if (exp >= 0) return n * TEN ** BigInt(exp);
++ const div = TEN ** BigInt(-exp);
++ return (n + div / 2n) / div;
++}
++
++/** `base` (an integer) × a fraction `frac` already scaled by 10^scale, rounded half-up. */
++const SCALE = 18;
++const ONE = TEN ** BigInt(SCALE);
++function mulFraction(base, fracScaled) {
++ return (base * fracScaled + ONE / 2n) / ONE;
++}
++
++/** A fraction in [0,1] → its scaled integer, refusing anything outside the interval. */
++function fraction(value, what) {
++ const f = scaled(value, SCALE, what);
++ if (f > ONE) throw new ParseError(`${what} is greater than 1: ${JSON.stringify(value)}`);
++ return f;
++}
++
++/** A provider's active discount as a scaled fraction off: 0 when absent or its end has passed. */
++function activeDiscount(price, fetchedAt, what) {
++ if (price == null || price.discount == null) return 0n;
++ if (price.discount_ends_at != null) {
++ const ends = Number(price.discount_ends_at);
++ if (!Number.isFinite(ends)) throw new ParseError(`${what}.discount_ends_at is not a Unix timestamp`);
++ if (ends * 1000 <= fetchedAt) return 0n;
++ }
++ return fraction(price.discount, `${what}.discount`);
++}
++
++/** USD per million → nano-USD per million (×1e9), and the two provider scales. */
++const USD_PER_M = 9;
++const CENTS_PER_TOKEN = 13;
++const USD_PER_TOKEN = 15;
++
++// ── Parsers ──────────────────────────────────────────────────────────────────────────────────────
++
++const isObject = (v) => v !== null && typeof v === 'object' && !Array.isArray(v);
++const modelName = (v, what) => {
++ if (typeof v !== 'string' || !v.trim()) throw new ParseError(`${what} is not a model name`);
++ return v.trim();
++};
++
++/** One observation, before its effective date and hashes are filled in. */
++function observation(provider, model, tokenClass, nano, fetchedAt) {
++ return {
++ provider,
++ model,
++ token_class: tokenClass,
++ nano_usd_per_m: nano.toString(),
++ // Currency and region are the document's own scope; each provider definition replaces both below.
++ currency: 'USD',
++ region: 'global',
++ fetched_at: fetchedAt,
++ };
++}
++
++/** DeepInfra `/models/list`: an array of model records; `pricing.type = tokens` carries the rates. */
++function parseDeepInfra(body, { fetchedAt }) {
++ if (!Array.isArray(body)) throw new ParseError('DeepInfra /models/list is not a JSON array');
++ const observations = [];
++ const seen = new Set();
++ let skipped = 0;
++ for (const [i, model] of body.entries()) {
++ if (!isObject(model)) throw new ParseError(`DeepInfra model ${i} is not an object`);
++ const name = modelName(model.model_name, `DeepInfra model ${i}.model_name`);
++ if (model.private === 1 || model.private === true) { skipped++; continue; }
++ const price = model.pricing;
++ if (!isObject(price) || price.type !== 'tokens') { skipped++; continue; }
++ if (seen.has(name)) throw new ParseError(`DeepInfra lists ${name} twice`);
++ seen.add(name);
++ const what = `DeepInfra ${name}`;
++ const off = activeDiscount(price, fetchedAt, what);
++ const factor = ONE - off;
++ // cents/token × 1e13 = nano-USD per million; the discount is a fraction off while active.
++ const input = mulFraction(scaled(price.cents_per_input_token, CENTS_PER_TOKEN, `${what}.cents_per_input_token`), factor);
++ const output = mulFraction(scaled(price.cents_per_output_token, CENTS_PER_TOKEN, `${what}.cents_per_output_token`), factor);
++ observations.push(observation('deepinfra', name, 'input', input, fetchedAt));
++ observations.push(observation('deepinfra', name, 'output', output, fetchedAt));
++ if (price.rate_per_input_token_cached != null) {
++ // The field is a multiplier: the pricing page renders input × it as the cached price.
++ const cached = mulFraction(input, fraction(price.rate_per_input_token_cached, `${what}.rate_per_input_token_cached`));
++ observations.push(observation('deepinfra', name, 'cached_read', cached, fetchedAt));
++ }
++ // A cache-write price depends on the cache retention the document states as multipliers (5m/1h); it is not
++ // one published rate, so no cache_write observation is invented from it.
++ }
++ if (!observations.length) throw new ParseError('DeepInfra stated no token-priced model');
++ return { observations, models: seen.size, skipped };
++}
++
++/** Featherless `/v1/models`: values under pricing are USD per token as decimal strings. */
++function parseFeatherless(body, { fetchedAt }) {
++ if (!isObject(body) || !Array.isArray(body.data)) throw new ParseError('Featherless /v1/models has no data array');
++ const observations = [];
++ const seen = new Set();
++ let skipped = 0;
++ for (const [i, model] of body.data.entries()) {
++ if (!isObject(model)) throw new ParseError(`Featherless model ${i} is not an object`);
++ const name = modelName(model.id, `Featherless model ${i}.id`);
++ const price = model.pricing;
++ if (!isObject(price)) { skipped++; continue; }
++ if (seen.has(name)) throw new ParseError(`Featherless lists ${name} twice`);
++ seen.add(name);
++ const what = `Featherless ${name}`;
++ // The document states both the exact per-token strings and their per-million display numbers; the string
++ // is authoritative, the number a fallback only. A gated model is kept: its published price is still evidence.
++ const inNano = price.prompt != null ? scaled(price.prompt, USD_PER_TOKEN, `${what}.pricing.prompt`)
++ : price.input != null ? scaled(price.input, USD_PER_M, `${what}.pricing.input`) : null;
++ const outNano = price.completion != null ? scaled(price.completion, USD_PER_TOKEN, `${what}.pricing.completion`)
++ : price.output != null ? scaled(price.output, USD_PER_M, `${what}.pricing.output`) : null;
++ if (inNano == null && outNano == null) { skipped++; continue; }
++ if (inNano != null) observations.push(observation('featherless', name, 'input', inNano, fetchedAt));
++ if (outNano != null) observations.push(observation('featherless', name, 'output', outNano, fetchedAt));
++ }
++ if (!observations.length) throw new ParseError('Featherless stated no token-priced model');
++ return { observations, models: seen.size, skipped };
++}
++
++/**
++ * The two providers, their official documents and why reading them is permitted. `enabled` is NEVER true here:
++ * the flag turns the feature on, and a person still has to enable a source (the row starts at 0).
++ */
++const PROVIDERS = Object.freeze({
++ deepinfra: {
++ provider: 'deepinfra',
++ name: 'DeepInfra prices',
++ url: 'https://api.deepinfra.com/models/list',
++ homepage_url: 'https://deepinfra.com/pricing',
++ docs_url: 'https://api.deepinfra.com/openapi.json',
++ currency: 'USD',
++ region: 'global',
++ parser_version: 'deepinfra@1',
++ parse: parseDeepInfra,
++ terms_note: 'DeepInfra is RESTRICTED for managed resale in OpenVibe.AI docs/provider-matrix.md, so this source only records its published rates as evidence; it never resells or routes a call. GET /models/list is the keyless public listing documented in DeepInfra\'s own OpenAPI (https://api.deepinfra.com/openapi.json), not an HTML page, so nothing is scraped. DeepInfra\'s terms (https://deepinfra.com/terms) restrict resale, not reading the public price list. Checked 2026-11-04; a person reviews it before enabling.',
++ robots_note: 'https://api.deepinfra.com/robots.txt answers 404 (the API host publishes no robots rules). Watch\'s fetcher reads only the documented JSON endpoint above, not a page. Checked 2026-11-04.',
++ },
++ featherless: {
++ provider: 'featherless',
++ name: 'Featherless AI prices',
++ url: 'https://api.featherless.ai/v1/models',
++ homepage_url: 'https://featherless.ai/pricing',
++ docs_url: 'https://featherless.ai/terms',
++ currency: 'USD',
++ region: 'global',
++ parser_version: 'featherless@1',
++ parse: parseFeatherless,
++ terms_note: 'Featherless is AUTHORIZED_WHOLESALE (Scale only) in OpenVibe.AI docs/provider-matrix.md: its terms §3 say Scale plans "may be used in arbitrary applications, including inference resale" (https://featherless.ai/terms, checked 2026-11-04), subject to an executed agreement. This source records the published rate as evidence only; it neither accepts the agreement nor routes a call. /v1/models is the documented OpenAI-compatible model list and needs no key (checked 2026-11-04), so nothing is scraped. A person reviews the terms and enables it.',
++ robots_note: 'https://api.featherless.ai/robots.txt answers 404 (the API host publishes no robots rules); https://featherless.ai/robots.txt allows its public pages and disallows /dashboard, /login and /signup, none of which is read. The fetcher reads only /v1/models. Checked 2026-11-04.',
++ },
++});
++
++/** Parse one provider document. Exposed for tests: no fetch, no database. */
++function parse(provider, body, { fetchedAt = Date.now() } = {}) {
++ const def = typeof provider === 'string' ? PROVIDERS[provider] : provider;
++ if (!def) throw new ProviderRateError(`no provider-rate parser for ${provider}`, 'watch.rate.unknown_provider');
++ const at = Number(fetchedAt);
++ if (!Number.isFinite(at) || at <= 0) throw new ProviderRateError('fetchedAt must be a time in ms');
++ const out = def.parse(body, { fetchedAt: at });
++ return {
++ ...out,
++ observations: out.observations.map((o) => ({
++ ...o,
++ provider: def.provider,
++ currency: def.currency,
++ region: def.region,
++ source_url: def.url,
++ fetched_at: at,
++ effective_date: new Date(at).toISOString().slice(0, 10),
++ })),
++ };
++}
++
++// ── The service module ──────────────────────────────────────────────────────────────────────────
++
++const iso = (v) => (v == null ? null : new Date(Number(v)).toISOString());
++const dateOf = (v) => new Date(Number(v)).toISOString().slice(0, 10);
++const note = (v) => (v == null ? null : String(v).slice(0, 2000));
++
++/** Retry-After in seconds or an HTTP date → ms from `at`, or 0. */
++function retryAfterMs(value, at) {
++ if (value == null) return 0;
++ const s = String(value).trim();
++ if (/^\d{1,7}$/.test(s)) return Number(s) * 1000;
++ const t = Date.parse(s);
++ return Number.isFinite(t) ? Math.max(0, t - at) : 0;
++}
++
++function createProviderRates({ db, config, fetcher, spacer = null, outbox = null, now = () => Date.now(), log = console, ids = contractIds, providers = PROVIDERS }) {
++ const settings = (config && config.providerRates) || {};
++ const enabledNow = () => Boolean(settings.enabled);
++ const maxBytes = Number.isSafeInteger(settings.maxBytes) && settings.maxBytes > 0 ? settings.maxBytes : 8 * 1024 * 1024;
++ const pollSec = Number.isSafeInteger(settings.pollIntervalSec) && settings.pollIntervalSec > 0 ? settings.pollIntervalSec : 6 * 3600;
++ const staleSec = Number.isSafeInteger(settings.staleAfterSec) && settings.staleAfterSec > 0 ? settings.staleAfterSec : 2 * pollSec;
++ const tickMs = Number.isSafeInteger(settings.tickMs) && settings.tickMs > 0 ? settings.tickMs : 60_000;
++ const inflight = new Set();
++ let timer = null;
++
++ const st = {
++ seed: db.prepare(`INSERT INTO provider_rate_sources (key, kind, provider, name, url, homepage_url, currency, region,
++ terms_note, robots_note, parser_version, enabled, poll_interval_sec, stale_after_sec, created_at, updated_at)
++ VALUES (@key, 'provider-rate', @provider, @name, @url, @homepage_url, @currency, @region, @terms_note, @robots_note,
++ @parser_version, 0, @poll_interval_sec, @stale_after_sec, @at, @at) ON CONFLICT (key) DO NOTHING`),
++ sources: db.prepare('SELECT * FROM provider_rate_sources ORDER BY key'),
++ source: db.prepare('SELECT * FROM provider_rate_sources WHERE key = ?'),
++ enable: db.prepare('UPDATE provider_rate_sources SET enabled = @enabled, next_due_at = 0, updated_at = @at, updated_by = @by WHERE key = @key'),
++ due: db.prepare(`SELECT * FROM provider_rate_sources WHERE enabled = 1 AND next_due_at <= @now AND not_before <= @now
++ ORDER BY next_due_at, key LIMIT @limit`),
++ success: db.prepare(`UPDATE provider_rate_sources SET last_fetch_at = @at, last_success_at = @at, last_state = @state,
++ consecutive_failures = 0, next_due_at = @next_due, not_before = 0, etag = @etag, last_modified = @lm,
++ last_body_hash = @hash WHERE key = @key`),
++ failure: db.prepare(`UPDATE provider_rate_sources SET last_fetch_at = @at, last_state = @state,
++ consecutive_failures = @failures, next_due_at = @next_due, not_before = @not_before WHERE key = @key`),
++ insertFetch: db.prepare(`INSERT INTO provider_rate_fetches (id, source_key, trigger, started_at, finished_at, state, http_status,
++ error_code, detail, bytes, raw_body_hash, parser_version, stated, observations, changed)
++ VALUES (@id, @source_key, @trigger, @started_at, @finished_at, @state, @http_status, @error_code, @detail, @bytes,
++ @raw_body_hash, @parser_version, @stated, @observations, @changed)`),
++ latest: db.prepare(`SELECT DISTINCT ON (model, token_class) * FROM provider_rate_observations
++ WHERE source_key = ? ORDER BY model, token_class, rid DESC`),
++ insertObservation: db.prepare(`INSERT INTO provider_rate_observations (id, source_key, fetch_id, provider, model, token_class,
++ nano_usd_per_m, currency, region, effective_date, source_url, fetched_at, value_hash, changed, previous_id, previous_nano_usd_per_m)
++ VALUES (@id, @source_key, @fetch_id, @provider, @model, @token_class, @nano_usd_per_m, @currency, @region, @effective_date,
++ @source_url, @fetched_at, @value_hash, @changed, @previous_id, @previous_nano_usd_per_m)`),
++ insertChange: db.prepare(`INSERT INTO provider_rate_changes (id, source_key, observation_id, previous_id, provider, model, token_class,
++ from_nano_usd_per_m, to_nano_usd_per_m, currency, region, effective_date, source_url, fetched_at, state, created_at)
++ VALUES (@id, @source_key, @observation_id, @previous_id, @provider, @model, @token_class, @from_nano_usd_per_m,
++ @to_nano_usd_per_m, @currency, @region, @effective_date, @source_url, @fetched_at, 'pending', @created_at)`),
++ change: db.prepare('SELECT * FROM provider_rate_changes WHERE id = ?'),
++ decide: db.prepare(`UPDATE provider_rate_changes SET state = @state, decided_at = @at, decided_by = @by, note = @note
++ WHERE id = @id AND state = 'pending'`),
++ pendingCount: db.prepare("SELECT COUNT(*) AS n FROM provider_rate_changes WHERE state = 'pending'"),
++ };
++
++ /** The first enabled boot writes the two sources disabled; an existing row is never overwritten (a person may have notes). */
++ let initDone = false;
++ async function init() {
++ if (initDone) return;
++ const at = now();
++ for (const def of Object.values(providers)) {
++ await st.seed.run({
++ key: def.provider, provider: def.provider, name: def.name, url: def.url, homepage_url: def.homepage_url || null,
++ currency: def.currency, region: def.region, terms_note: def.terms_note, robots_note: def.robots_note,
++ parser_version: def.parser_version, poll_interval_sec: pollSec, stale_after_sec: staleSec, at,
++ });
++ }
++ initDone = true;
++ }
++
++ const sourceRow = (key) => st.source.get(String(key || ''));
++ async function sources() { return await st.sources.all(); }
++
++ /** {status: disabled|never_fetched|failing|stale|healthy, times and the last state}; never invented. */
++ function health(row, at = now()) {
++ const out = {
++ status: null,
++ last_fetch_at: iso(row.last_fetch_at),
++ last_success_at: iso(row.last_success_at),
++ last_state: row.last_state == null ? null : row.last_state,
++ consecutive_failures: Number(row.consecutive_failures) || 0,
++ next_due_at: Number(row.enabled) ? iso(Math.max(Number(row.next_due_at) || 0, Number(row.not_before) || 0)) : null,
++ };
++ if (!Number(row.enabled)) out.status = 'disabled';
++ else if (row.last_fetch_at == null) out.status = 'never_fetched';
++ else if (row.last_state && FAILURE_STATES.has(row.last_state) && Number(row.consecutive_failures) > 0) out.status = 'failing';
++ else if (row.last_success_at == null || at - Number(row.last_success_at) > Number(row.stale_after_sec) * 1000) out.status = 'stale';
++ else out.status = 'healthy';
++ return out;
++ }
++
++ function sourceView(row, at = now()) {
++ const h = health(row, at);
++ return {
++ key: row.key, kind: row.kind, provider: row.provider, name: row.name, url: row.url, homepage_url: row.homepage_url,
++ currency: row.currency, region: row.region, terms_note: row.terms_note, robots_note: row.robots_note,
++ parser_version: row.parser_version, enabled: Boolean(Number(row.enabled)),
++ poll_interval_sec: Number(row.poll_interval_sec), stale_after_sec: Number(row.stale_after_sec),
++ health: h,
++ };
++ }
++
++ function fetchView(row) {
++ return {
++ id: row.id, source_key: row.source_key, trigger: row.trigger,
++ started_at: iso(row.started_at), finished_at: iso(row.finished_at), state: row.state,
++ http_status: row.http_status == null ? null : Number(row.http_status),
++ error_code: row.error_code == null ? null : row.error_code,
++ detail: row.detail == null ? null : row.detail,
++ bytes: row.bytes == null ? null : Number(row.bytes),
++ raw_body_hash: row.raw_body_hash == null ? null : row.raw_body_hash,
++ parser_version: row.parser_version == null ? null : row.parser_version,
++ stated: Number(row.stated), observations: Number(row.observations), changed: Number(row.changed),
++ };
++ }
++
++ function observationView(row) {
++ return {
++ id: row.id, source_key: row.source_key, fetch_id: row.fetch_id, provider: row.provider, model: row.model,
++ token_class: row.token_class, nano_usd_per_m: row.nano_usd_per_m, currency: row.currency, region: row.region,
++ effective_date: row.effective_date, source_url: row.source_url, fetched_at: iso(row.fetched_at),
++ changed: Boolean(Number(row.changed)), previous_id: row.previous_id == null ? null : row.previous_id,
++ previous_nano_usd_per_m: row.previous_nano_usd_per_m == null ? null : row.previous_nano_usd_per_m,
++ value_hash: row.value_hash,
++ };
++ }
++
++ function changeView(row) {
++ return {
++ id: row.id, state: row.state, approval_required: row.state === 'pending',
++ source_key: row.source_key, provider: row.provider, model: row.model, token_class: row.token_class,
++ from_nano_usd_per_m: row.from_nano_usd_per_m == null ? null : row.from_nano_usd_per_m,
++ to_nano_usd_per_m: row.to_nano_usd_per_m, currency: row.currency, region: row.region,
++ effective_date: row.effective_date, source_url: row.source_url, fetched_at: iso(row.fetched_at),
++ observation_id: row.observation_id, previous_id: row.previous_id == null ? null : row.previous_id,
++ created_at: iso(row.created_at), decided_at: iso(row.decided_at),
++ decided_by: row.decided_by == null ? null : row.decided_by,
++ note: row.note == null ? null : row.note,
++ // Evidence only: no billable rate is written by Watch, approved or not.
++ effect: 'evidence',
++ };
++ }
++
++ const changeEvent = (row) => ({
++ change_id: row.id, source_key: row.source_key, provider: row.provider, model: row.model, token_class: row.token_class,
++ from_nano_usd_per_m: row.from_nano_usd_per_m == null ? null : row.from_nano_usd_per_m,
++ to_nano_usd_per_m: row.to_nano_usd_per_m, currency: row.currency, region: row.region,
++ effective_date: row.effective_date, source_url: row.source_url, fetched_at: iso(row.fetched_at),
++ approval_required: true,
++ });
++
++ /**
++ * Fetch and parse one source now. The fetch row and, on success, every observation and change commit together;
++ * a failure commits only the fetch row and the source's failure state. One fetch of a source at a time.
++ */
++ async function fetchSource(key, { trigger = 'schedule' } = {}) {
++ if (inflight.has(key)) return { busy: true };
++ inflight.add(key);
++ try {
++ const row = await sourceRow(key);
++ if (!row) return null;
++ if (!Number(row.enabled)) return { disabled: true, source: sourceView(row) };
++ const def = providers[row.provider];
++ if (!def) return { unavailable: `no parser for provider ${row.provider}` };
++ const startedAt = now();
++ const run = {
++ id: `prf_${ids.ulid(startedAt)}`, source_key: row.key, trigger, started_at: startedAt, finished_at: startedAt,
++ state: null, http_status: null, error_code: null, detail: null, bytes: null, raw_body_hash: null,
++ parser_version: def.parser_version, stated: 0, observations: 0, changed: 0,
++ };
++ let res = null;
++ let failure = null;
++ let notBefore = 0;
++ try {
++ if (spacer) await spacer.space(new URL(row.url).host);
++ res = await fetcher.fetchUrl(row.url, {
++ headers: { Accept: 'application/json' },
++ etag: row.etag || null,
++ lastModified: row.last_modified || null,
++ capBytes: maxBytes,
++ });
++ } catch (err) {
++ failure = { state: err.code === 'timeout' ? 'timeout' : 'http_error', error_code: err.code || 'fetch_error', detail: String(err.message || err).slice(0, 500) };
++ }
++ const finishedAt = now();
++ run.finished_at = finishedAt;
++
++ if (!failure) {
++ run.http_status = res.status;
++ run.bytes = res.bytes == null ? null : Number(res.bytes);
++ if (res.status === 304) {
++ run.state = 'not_modified';
++ } else if (res.status === 429) {
++ run.state = 'rate_limited';
++ run.error_code = 'http_429';
++ run.detail = 'the provider answered 429 Too Many Requests';
++ notBefore = finishedAt + retryAfterMs(res.headers && res.headers['retry-after'], finishedAt);
++ } else if (res.status !== 200) {
++ run.state = 'http_error';
++ run.error_code = `http_${res.status}`;
++ run.detail = `the provider answered HTTP ${res.status}`;
++ notBefore = finishedAt + retryAfterMs(res.headers && res.headers['retry-after'], finishedAt);
++ } else {
++ run.raw_body_hash = crypto.createHash('sha256').update(res.body || Buffer.alloc(0)).digest('hex');
++ let parsed = null;
++ try {
++ const text = decodeBody(res.body || Buffer.alloc(0), res.headers && res.headers['content-type']);
++ let body;
++ try { body = JSON.parse(text); } catch { throw new ParseError('the body is not JSON'); }
++ parsed = def.parse(body, { fetchedAt: finishedAt });
++ } catch (err) {
++ failure = { state: 'parse_error', error_code: err instanceof ParseError ? 'parse_error' : 'unreadable', detail: String(err.message || err).slice(0, 500) };
++ }
++ if (!failure) {
++ // Every observation and the changes it raises commit with the fetch row and the source's state.
++ const previous = new Map();
++ for (const p of await st.latest.all(row.key)) previous.set(`${p.model}\u0000${p.token_class}`, p);
++ const observations = [];
++ const changes = [];
++ for (const o of parsed.observations) {
++ const prev = previous.get(`${o.model}\u0000${o.token_class}`) || null;
++ const value = { nano_usd_per_m: o.nano_usd_per_m, currency: def.currency, region: def.region };
++ const valueHash = hashValue(value);
++ const changed = Boolean(prev) && prev.value_hash !== valueHash;
++ const effectiveDate = prev && !changed ? prev.effective_date : dateOf(finishedAt);
++ const obs = {
++ id: `pro_${ids.ulid(finishedAt)}`, source_key: row.key, fetch_id: run.id,
++ provider: def.provider, model: o.model, token_class: o.token_class,
++ nano_usd_per_m: o.nano_usd_per_m, currency: def.currency, region: def.region,
++ effective_date: effectiveDate, source_url: row.url, fetched_at: finishedAt,
++ value_hash: valueHash, changed: changed ? 1 : 0,
++ previous_id: prev ? prev.id : null,
++ previous_nano_usd_per_m: prev ? prev.nano_usd_per_m : null,
++ };
++ // An unchanged value is not written again: its first observation keeps the effective date and
++ // the fetch row says how many rates the document stated. A new model is evidence, not a change.
++ if (!prev || changed) observations.push(obs);
++ if (changed) {
++ changes.push({
++ id: `prc_${ids.ulid(finishedAt)}`, source_key: row.key, observation_id: obs.id,
++ previous_id: prev.id, provider: def.provider, model: o.model, token_class: o.token_class,
++ from_nano_usd_per_m: prev.nano_usd_per_m, to_nano_usd_per_m: o.nano_usd_per_m,
++ currency: def.currency, region: def.region, effective_date: effectiveDate,
++ source_url: row.url, fetched_at: finishedAt, created_at: finishedAt,
++ });
++ }
++ }
++ run.state = 'ok';
++ run.stated = parsed.observations.length;
++ run.observations = observations.length;
++ run.changed = changes.length;
++ const events = changes.map((c) => ({
++ event_type: 'watch.rate.changed', subject: { type: KIND, id: row.key }, visibility: 'internal', priority: 'important',
++ payload: changeEvent(c),
++ }));
++ await db.tx(async () => {
++ await st.insertFetch.run(run);
++ await st.success.run({
++ key: row.key, at: finishedAt, state: 'ok', next_due: finishedAt + Number(row.poll_interval_sec) * 1000,
++ // A 200 states the current validators; carrying an old one forward could make the next
++ // conditional read a 304 for a representation that has no validator.
++ etag: (res.headers && res.headers.etag) || null,
++ lm: (res.headers && res.headers['last-modified']) || null,
++ hash: run.raw_body_hash,
++ });
++ for (const o of observations) await st.insertObservation.run(o);
++ for (const c of changes) await st.insertChange.run(c);
++ if (outbox) for (const e of events) await outbox.emit(envelope(e, config.serviceId || 'watch', now));
++ });
++ if (outbox && events.length) outbox.kick().catch(() => {});
++ return { fetch: fetchView(run), source: sourceView(await sourceRow(row.key)) };
++ }
++ }
++ }
++
++ // A 304 is a successful read that states no new value; a failure writes no observation.
++ if (!failure && run.state === 'not_modified') {
++ await db.tx(async () => {
++ await st.insertFetch.run(run);
++ await st.success.run({
++ key: row.key, at: finishedAt, state: 'not_modified', next_due: finishedAt + Number(row.poll_interval_sec) * 1000,
++ etag: (res.headers && res.headers.etag) || row.etag || null,
++ lm: (res.headers && res.headers['last-modified']) || row.last_modified || null,
++ hash: row.last_body_hash,
++ });
++ });
++ return { fetch: fetchView(run), source: sourceView(await sourceRow(row.key)) };
++ }
++
++ const state = failure ? failure.state : run.state;
++ Object.assign(run, {
++ state,
++ error_code: failure ? failure.error_code : run.error_code,
++ detail: failure ? failure.detail : run.detail,
++ });
++ const failures = Number(row.consecutive_failures) + 1;
++ const every = Number(row.poll_interval_sec) * 1000;
++ const backoff = Math.min(every * 2 ** Math.min(failures - 1, 10), 6 * 3600 * 1000);
++ const nextDue = finishedAt + Math.max(backoff, notBefore > finishedAt ? notBefore - finishedAt : 0);
++ await db.tx(async () => {
++ await st.insertFetch.run(run);
++ await st.failure.run({ key: row.key, at: finishedAt, state, failures, next_due: nextDue, not_before: notBefore });
++ });
++ return { fetch: fetchView(run), source: sourceView(await sourceRow(row.key)) };
++ } finally {
++ inflight.delete(key);
++ }
++ }
++
++ /** One scheduler tick: fetch every enabled source whose turn has come. */
++ async function tick() {
++ if (!enabledNow() || !fetcher) return [];
++ const due = await st.due.all({ now: now(), limit: 4 });
++ for (const row of due) {
++ fetchSource(row.key, { trigger: 'schedule' }).catch((err) => log.error(`[provider-rates] ${row.key}: ${err.stack || err}`));
++ }
++ return due.map((r) => r.key);
++ }
++
++ function start() {
++ if (!enabledNow() || timer) return;
++ timer = setInterval(async () => { try { await tick(); } catch (err) { log.error(`[provider-rates] tick: ${err.message}`); } }, tickMs);
++ timer.unref?.();
++ }
++
++ async function stop() {
++ if (timer) clearInterval(timer);
++ timer = null;
++ const deadline = Date.now() + 15000;
++ while (inflight.size && Date.now() < deadline) await new Promise((r) => setTimeout(r, 50));
++ }
++
++ async function setEnabled(key, want, principal) {
++ if (typeof want !== 'boolean') throw new ProviderRateError('enabled must be true or false');
++ const row = await sourceRow(key);
++ if (!row) return null;
++ if (want && (!row.terms_note || !row.robots_note)) throw new ProviderRateError('a provider-rate source needs its terms and robots notes before it can be enabled');
++ await st.enable.run({ key: row.key, enabled: want ? 1 : 0, at: now(), by: principal && principal.sub ? principal.sub : null });
++ return await sourceRow(key);
++ }
++
++ /** Latest observations first; `latest` returns the newest row per (model, token class). */
++ async function observations({ latest = false, source = null, model = null, tokenClass = null, before = null, limit = 100 } = {}) {
++ const l = Math.min(Math.max(Number(limit) || 100, 1), 500);
++ if (latest) {
++ return await db.prepare(`SELECT * FROM (
++ SELECT DISTINCT ON (model, token_class) * FROM provider_rate_observations
++ WHERE (@source::text IS NULL OR source_key = @source) AND (@model::text IS NULL OR model = @model)
++ AND (@class::text IS NULL OR token_class = @class)
++ ORDER BY model, token_class, rid DESC
++ ) t ORDER BY model, token_class LIMIT @limit`).all({ source, model, class: tokenClass, limit: l });
++ }
++ const b = /^\d{1,18}$/.test(String(before || '')) ? Number(before) : Number.MAX_SAFE_INTEGER;
++ return await db.prepare(`SELECT * FROM provider_rate_observations
++ WHERE rid < @before AND (@source::text IS NULL OR source_key = @source)
++ AND (@model::text IS NULL OR model = @model) AND (@class::text IS NULL OR token_class = @class)
++ ORDER BY rid DESC LIMIT @limit`).all({ before: b, source, model, class: tokenClass, limit: l });
++ }
++
++ async function changes({ state = null, source = null, before = null, limit = 100 } = {}) {
++ if (state != null && !['pending', 'approved', 'rejected'].includes(state)) throw new ProviderRateError('state must be pending, approved or rejected');
++ const l = Math.min(Math.max(Number(limit) || 100, 1), 500);
++ const b = /^\d{1,18}$/.test(String(before || '')) ? Number(before) : Number.MAX_SAFE_INTEGER;
++ return await db.prepare(`SELECT * FROM provider_rate_changes WHERE rid < @before
++ AND (@state::text IS NULL OR state = @state) AND (@source::text IS NULL OR source_key = @source)
++ ORDER BY rid DESC LIMIT @limit`).all({ before: b, state, source, limit: l });
++ }
++
++ async function change(id) { return await st.change.get(String(id || '')); }
++ async function pendingCount() { return Number((await st.pendingCount.get()).n) || 0; }
++ async function fetch(id) {
++ const row = await db.prepare('SELECT * FROM provider_rate_fetches WHERE id = ?').get(String(id || ''));
++ return row ? fetchView(row) : null;
++ }
++
++ /**
++ * A person approves or rejects a pending change. The decision is recorded and a `watch.rate.approved` /
++ * `watch.rate.rejected` event is emitted; no billable rate is written and `applied` is always false.
++ */
++ async function decide(id, decision, principal, { note: text = null } = {}) {
++ if (!['approved', 'rejected'].includes(decision)) throw new ProviderRateError('decision must be approved or rejected');
++ if (!principal || principal.kind !== 'person' || !/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(principal.sub || ''))) {
++ throw new ProviderRateError('a person approves a provider-rate change; a service or an agent may not', 'watch.rate.person_only');
++ }
++ const at = now();
++ const row = await st.change.get(String(id || ''));
++ if (!row) return null;
++ if (row.state !== 'pending') return { conflict: row.state, change: changeView(row) };
++ let updated = null;
++ await db.tx(async () => {
++ const r = await st.decide.run({ id: row.id, state: decision, at, by: principal.sub, note: note(text) });
++ if (r.changes !== 1) {
++ const raced = new ProviderRateError('this change was already decided', 'watch.rate.decided');
++ raced.status = 409;
++ throw raced;
++ }
++ updated = { ...row, state: decision, decided_at: at, decided_by: principal.sub, note: note(text) };
++ if (outbox) {
++ await outbox.emit(envelope({
++ event_type: decision === 'approved' ? 'watch.rate.approved' : 'watch.rate.rejected',
++ subject: { type: KIND, id: row.source_key },
++ visibility: 'internal', priority: 'important',
++ payload: { ...changeEvent(updated), approval_required: false, state: decision, decided_by: principal.sub, decided_at: iso(at), applied: false },
++ }, config.serviceId || 'watch', now));
++ }
++ });
++ if (outbox) outbox.kick().catch(() => {});
++ return { change: changeView(updated), applied: false };
++ }
++
++ return {
++ kind: KIND,
++ get enabled() { return enabledNow(); },
++ providers,
++ init, tick, start, stop,
++ sources, source: sourceRow, sourceView, health,
++ setEnabled, fetchNow: fetchSource, fetch, fetchView,
++ observations, observationView, changes, change, changeView, pendingCount, decide,
++ parse,
++ inflight: () => [...inflight],
++ };
++}
++
++module.exports = {
++ createProviderRates, parse, scaled, PROVIDERS, TOKEN_CLASSES, KIND, FAILURE_STATES,
++};
ddiff --git a/test/account-data.test.js b/test/account-data.test.js
iindex f8005d6..d917488 100644
---- a/test/account-data.test.js
++++ b/test/account-data.test.js
@@@ -60,7 +60,7 @@ t('boot, and the person\'s watches, checks and sessions', async () => {
t('every table that names a person is covered or exempted with a reason', async () => {
const missing = await uncovered(svc.db, {
tables: TABLES,
-- exempt: { account_data_events: 'the ledger of requests itself: which export and deletion Watch applied' },
++ exempt: { account_data_events: 'the ledger of requests itself: which export and deletion Watch applied', config_snapshots: 'D11 configuration revisions: created_by names the owner who changed a setting, not a person\'s account data' },
});
assert.deepStrictEqual(missing, [], `a table with a person column is neither covered nor exempted: ${missing.join(', ')}`);
});
ddiff --git a/test/fixtures/provider-rates/README.md b/test/fixtures/provider-rates/README.md
nnew file mode 100644
iindex 0000000..b664342
---- /dev/null
++++ b/test/fixtures/provider-rates/README.md
@@@ -0,0 +1,13 @@
++# Recorded provider price documents
++
++These are the two official price documents trimmed to a few records, recorded on 2026-11-04. The tests parse
++them and never fetch: `test/provider-rates.test.js`.
++
++| File | Live source | What was kept |
++|---|---|---|
++| `deepinfra-models.json` | `https://api.deepinfra.com/models/list` (documented in `https://api.deepinfra.com/openapi.json`) | three token-priced models (one with an active `discount` and a `rate_per_input_token_cached` multiplier) and one embeddings model with a different pricing type, so the parser's skip path is real |
++| `featherless-models.json` | `https://api.featherless.ai/v1/models` | four models, including a gated one (its published price is still evidence) and an audio-output model |
++
++The live documents are public and keyless. Nothing was changed except dropping the other records and fields the
++parser does not read; no prices were edited. The malformed case is built in the test by changing one price to `"n/a"`,
++so there is no recorded document that could be mistaken for a real one.
ddiff --git a/test/fixtures/provider-rates/deepinfra-models.json b/test/fixtures/provider-rates/deepinfra-models.json
nnew file mode 100644
iindex 0000000..d72a12c
---- /dev/null
++++ b/test/fixtures/provider-rates/deepinfra-models.json
@@@ -0,0 +1,88 @@
++[
++ {
++ "model_name": "Qwen/Qwen3-Coder-30B-A3B-Instruct",
++ "type": "text-generation",
++ "reported_type": "text-generation",
++ "private": 0,
++ "max_tokens": 262144,
++ "pricing": {
++ "discount": null,
++ "discount_ends_at": null,
++ "short": null,
++ "full": null,
++ "table": null,
++ "type": "tokens",
++ "cents_per_input_token": 0.000007,
++ "cents_per_output_token": 0.000026,
++ "rate_per_input_token_cached": null,
++ "rate_per_input_token_cache_write": null,
++ "rate_per_service_tier_priority": null,
++ "rate_per_service_tier_flex": null,
++ "rate_per_explicit_cache_write_token": null,
++ "explicit_cache_granularity_tokens": null
++ }
++ },
++ {
++ "model_name": "stepfun-ai/Step-3.7-Flash",
++ "type": "text-generation",
++ "reported_type": "text-generation",
++ "private": 0,
++ "max_tokens": 262144,
++ "pricing": {
++ "discount": 0.2,
++ "discount_ends_at": null,
++ "short": null,
++ "full": null,
++ "table": null,
++ "type": "tokens",
++ "cents_per_input_token": 0.00002,
++ "cents_per_output_token": 0.000115,
++ "rate_per_input_token_cached": 0.2,
++ "rate_per_input_token_cache_write": null,
++ "rate_per_service_tier_priority": null,
++ "rate_per_service_tier_flex": 0.8,
++ "rate_per_explicit_cache_write_token": null,
++ "explicit_cache_granularity_tokens": null
++ }
++ },
++ {
++ "model_name": "Qwen/Qwen3-VL-235B-A22B-Instruct",
++ "type": "text-generation",
++ "reported_type": "text-generation",
++ "private": 0,
++ "max_tokens": 262144,
++ "pricing": {
++ "discount": null,
++ "discount_ends_at": null,
++ "short": null,
++ "full": null,
++ "table": null,
++ "type": "tokens",
++ "cents_per_input_token": 0.00002,
++ "cents_per_output_token": 0.000088,
++ "rate_per_input_token_cached": 0.55,
++ "rate_per_input_token_cache_write": null,
++ "rate_per_service_tier_priority": 1.5,
++ "rate_per_service_tier_flex": 0.8,
++ "rate_per_explicit_cache_write_token": null,
++ "explicit_cache_granularity_tokens": null
++ }
++ },
++ {
++ "model_name": "sentence-transformers/all-MiniLM-L6-v2",
++ "type": "embeddings",
++ "reported_type": "embeddings",
++ "private": 0,
++ "max_tokens": 512,
++ "pricing": {
++ "discount": null,
++ "discount_ends_at": null,
++ "short": null,
++ "full": null,
++ "table": null,
++ "type": "input_tokens",
++ "cents_per_input_token": 5e-7,
++ "cents_per_image_token": null
++ }
++ }
++]
ddiff --git a/test/fixtures/provider-rates/featherless-models.json b/test/fixtures/provider-rates/featherless-models.json
nnew file mode 100644
iindex 0000000..5bfdcfe
---- /dev/null
++++ b/test/fixtures/provider-rates/featherless-models.json
@@@ -0,0 +1,80 @@
++{
++ "object": "list",
++ "data": [
++ {
++ "id": "Qwen/Qwen3-0.6B",
++ "is_gated": false,
++ "created": 1751086102,
++ "model_class": "qwen3-0b6",
++ "owned_by": "Feather",
++ "context_length": 32768,
++ "pricing": {
++ "prompt": "0.00000004",
++ "completion": "0.00000008",
++ "image": "0",
++ "request": "0",
++ "input": 0.04,
++ "output": 0.08
++ },
++ "features": {
++ "tool_use": true
++ }
++ },
++ {
++ "id": "Qwen/Qwen3-14B",
++ "is_gated": false,
++ "created": 1747094414,
++ "model_class": "qwen3-14b",
++ "owned_by": "Feather",
++ "context_length": 32768,
++ "pricing": {
++ "prompt": "0.00000012",
++ "completion": "0.00000024",
++ "image": "0",
++ "request": "0",
++ "input": 0.12,
++ "output": 0.24
++ },
++ "features": {
++ "tool_use": true
++ }
++ },
++ {
++ "id": "hexgrad/Kokoro-82M",
++ "is_gated": false,
++ "created": 1785442329,
++ "model_class": "tts-kokoro-82m",
++ "owned_by": "Feather",
++ "pricing": {
++ "prompt": "0.0000003",
++ "completion": "0.0000012",
++ "image": "0",
++ "request": "0",
++ "input": 0.3,
++ "output": 1.2
++ },
++ "features": {
++ "audio_output": true
++ }
++ },
++ {
++ "id": "meta-llama/Meta-Llama-3.1-8B-Instruct",
++ "is_gated": true,
++ "created": 1726585673,
++ "model_class": "llama31-8b",
++ "owned_by": "Feather",
++ "context_length": 32768,
++ "pricing": {
++ "prompt": "0.00000005",
++ "completion": "0.00000008",
++ "image": "0",
++ "request": "0",
++ "input": 0.05,
++ "output": 0.08
++ },
++ "features": {
++ "tool_use": true
++ }
++ }
++ ]
++}
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 194f7ec..c51c183 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
*/
--async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null, run = null, codes = null, projects = null, machines = null, projectSecrets = null } = {}) {
++async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null, run = null, codes = null, projects = null, machines = null, projectSecrets = null, providerRateProviders = null } = {}) {
// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real
// to fetch and verify against (no test touches the internet). Its URL is overridable via env.
const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });
@@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l
});
// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
const testdb = await require('./db').testDb();
-- const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(run ? { run } : {}), ...(codes ? { codes } : {}), ...(projects ? { projects } : {}), ...(machines ? { machines } : {}), ...(projectSecrets ? { projectSecrets } : {}), ...(now ? { now } : {}) });
++ const h = await start({ config, env, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(run ? { run } : {}), ...(codes ? { codes } : {}), ...(projects ? { projects } : {}), ...(machines ? { machines } : {}), ...(projectSecrets ? { projectSecrets } : {}), ...(providerRateProviders ? { providerRateProviders } : {}), ...(now ? { now } : {}) });
const base = `http://127.0.0.1:${h.server.address().port}`;
/** A session cookie for a person, as /auth/callback would have set it. */
ddiff --git a/test/internal-config.test.js b/test/internal-config.test.js
nnew file mode 100644
iindex 0000000..59ca96c
---- /dev/null
++++ b/test/internal-config.test.js
@@@ -0,0 +1,103 @@
++'use strict';
++/**
++ * D11 owner settings: the watch.provider_rates namespace and /internal/config.
++ *
++ * - watch.config.manage is required; a service token without it is 403;
++ * - WATCH_PROVIDER_RATES is a pin: it wins, is read-only, and a write is 409 config.pinned_in_env;
++ * - an owner change applies live: the API appears and the module starts reading the new value;
++ * - with no KEK the store works and reports sealed:false; with one it is sealed;
++ * - a secret-class value never leaves a summary, history or snapshot.
++ */
++const assert = require('assert');
++const { boot, serviceToken, request, suite } = require('./helpers');
++const test = suite('internal config');
++
++const CAP = ['watch.config.manage'];
++const OWNER = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';
++const SECRET = 'sk_watch_d11_toppsecret_123456';
++const admin = () => serviceToken('network', CAP);
++
++test('the summary: schema, the env pin, sealed:false without a KEK', async () => {
++ const svc = await boot({ env: { WATCH_PROVIDER_RATES: '1' } });
++ try {
++ const r = await request(svc.base, 'GET', '/internal/config/watch.provider_rates', { token: admin() });
++ assert.strictEqual(r.status, 200, r.text.slice(0, 300));
++ assert.strictEqual(r.body.namespace, 'watch.provider_rates');
++ assert.strictEqual(r.body.sealed, false);
++ assert.strictEqual(r.body.values.enabled, true);
++ assert.deepStrictEqual(r.body.pins.enabled, { env: 'WATCH_PROVIDER_RATES', pinned: true });
++ assert.ok(!r.text.includes(SECRET));
++ } finally { await svc.stop(); }
++ });
++
++ test('a service token without watch.config.manage is refused', async () => {
++ const svc = await boot();
++ try {
++ const r = await request(svc.base, 'GET', '/internal/config/watch.provider_rates', { token: serviceToken('services', ['watch.watch.read']) });
++ assert.strictEqual(r.status, 403);
++ assert.strictEqual(r.body.code, 'capability.denied');
++ } finally { await svc.stop(); }
++ });
++
++ test('an env-pinned key is read-only', async () => {
++ const svc = await boot({ env: { WATCH_PROVIDER_RATES: '1' } });
++ try {
++ const r = await request(svc.base, 'POST', '/internal/config/watch.provider_rates', {
++ token: admin(), headers: { 'x-ov-subject': OWNER }, body: { values: { enabled: false }, reason: 'off' },
++ });
++ assert.strictEqual(r.status, 409, r.text.slice(0, 300));
++ assert.strictEqual(r.body.code, 'config.pinned_in_env');
++ } finally { await svc.stop(); }
++ });
++
++ test('an owner change applies live: the API appears and the worker switch is on', async () => {
++ const lines = [];
++ const log = { log: (m) => lines.push(String(m)), info: (m) => lines.push(String(m)), warn: () => {}, error: () => {} };
++ const svc = await boot({ log });
++ try {
++ assert.strictEqual(svc.config.providerRates.enabled, false);
++ assert.strictEqual((await request(svc.base, 'GET', '/api/v1/provider-rates', { token: serviceToken('network', ['watch.watch.read']) })).status, 404);
++ const r = await request(svc.base, 'POST', '/internal/config/watch.provider_rates', {
++ token: admin(), headers: { 'x-ov-subject': OWNER }, body: { values: { enabled: true }, reason: 'owner turned it on' },
++ });
++ assert.strictEqual(r.status, 200, r.text.slice(0, 300));
++ assert.strictEqual(r.body.revision, 2);
++ assert.strictEqual(svc.config.providerRates.enabled, true);
++ assert.strictEqual(svc.providerRates.enabled, true);
++ const got = await request(svc.base, 'GET', '/api/v1/provider-rates', { token: serviceToken('network', ['watch.watch.read']) });
++ assert.strictEqual(got.status, 200, got.text.slice(0, 300));
++ assert.ok(lines.join('\n').includes('by user:' + OWNER), 'the audit line names the actor');
++ assert.ok(!lines.join('\n').includes(SECRET));
++ } finally { await svc.stop(); }
++ });
++
++ test('with a KEK the store is sealed and boot reseals', async () => {
++ const svc = await boot({ env: { OV_CONFIG_KEKS: 'k1:MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=', OV_CONFIG_KEK_CURRENT: 'k1' } });
++ try {
++ const r = await request(svc.base, 'GET', '/internal/config/watch.provider_rates', { token: admin() });
++ assert.strictEqual(r.status, 200, r.text.slice(0, 300));
++ assert.strictEqual(r.body.sealed, true);
++ assert.strictEqual(r.body.kek, 'k1');
++ } finally { await svc.stop(); }
++ });
++
++ test('a secret is write-only and never appears in a summary, snapshot or history', async () => {
++ const svc = await boot();
++ try {
++ const shared = require('openvibe-shared/config');
++ const extra = await shared.createConfigStore({
++ db: svc.db, service: 'watch', namespace: 'watch.test_secrets',
++ schema: { type: 'object', additionalProperties: false, properties: { api_key: { type: 'string' } } },
++ classify: () => 'secret',
++ });
++ const marker = await extra.setSecret('api_key', SECRET, { actor: { type: 'user', id: OWNER }, reason: 'test' });
++ assert.strictEqual(marker.redacted, true);
++ assert.ok(!JSON.stringify(marker).includes(SECRET));
++ const summary = await extra.summary();
++ assert.strictEqual(summary.values.api_key.redacted, true);
++ assert.ok(!JSON.stringify(summary).includes(SECRET));
++ assert.ok(!JSON.stringify(await extra.history()).includes(SECRET));
++ assert.ok(!JSON.stringify(await extra.snapshot(1)).includes(SECRET));
++ } finally { await svc.stop(); }
++ });
++test.run().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/provider-rates.test.js b/test/provider-rates.test.js
nnew file mode 100644
iindex 0000000..630d753
---- /dev/null
++++ b/test/provider-rates.test.js
@@@ -0,0 +1,273 @@
++'use strict';
++/**
++ * Provider-rate evidence (plan P6): the two official machine-readable price documents parsed without a network,
++ * a changed rate raising one Watch change, an unchanged one raising none, a malformed one labelled parse_error and
++ * never becoming zero, a person's approval recorded as evidence (never a rate card), and the flag off changing
++ * nothing.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const { boot, request, serviceToken, userToken, newUser, site, suite } = require('./helpers');
++const { parse, scaled, PROVIDERS } = require('../server/provider-rates');
++const { validate } = require('openvibe-contracts');
++
++const t = suite('provider-rates');
++const fixture = (name) => JSON.parse(fs.readFileSync(path.join(__dirname, 'fixtures', 'provider-rates', name), 'utf8'));
++const AT = Date.parse('2026-11-04T03:00:00Z');
++const deepinfraDoc = fixture('deepinfra-models.json');
++const featherlessDoc = fixture('featherless-models.json');
++const byKey = (observations, model, cls) => observations.find((o) => o.model === model && o.token_class === cls);
++
++/** A provider definition pointed at a stub; everything else (parser, units, notes) stays the real one. */
++const pointed = (provider, url) => ({ ...PROVIDERS[provider], url });
++
++t('DeepInfra /models/list fixture: exact nano-USD per million, integer strings, cache multiplier and discount', () => {
++ const out = parse('deepinfra', deepinfraDoc, { fetchedAt: AT });
++ assert.strictEqual(out.models, 3, 'the embeddings model is not token-priced and is skipped');
++ assert.strictEqual(out.skipped, 1);
++ assert.strictEqual(out.observations.length, 8, 'input + output for three models, cached read for the two that state it');
++ for (const o of out.observations) {
++ assert.strictEqual(typeof o.nano_usd_per_m, 'string');
++ assert.match(o.nano_usd_per_m, /^\d+$/, `${o.model} ${o.token_class} is an integer string`);
++ assert.strictEqual(o.currency, 'USD');
++ assert.strictEqual(o.region, 'global');
++ assert.strictEqual(o.effective_date, '2026-11-04');
++ assert.strictEqual(o.source_url, PROVIDERS.deepinfra.url);
++ assert.strictEqual(o.fetched_at, AT);
++ }
++ // 0.000007 cents/token = $0.07/M = 70,000,000 nano-USD/M.
++ assert.strictEqual(byKey(out.observations, 'Qwen/Qwen3-Coder-30B-A3B-Instruct', 'input').nano_usd_per_m, '70000000');
++ assert.strictEqual(byKey(out.observations, 'Qwen/Qwen3-Coder-30B-A3B-Instruct', 'output').nano_usd_per_m, '260000000');
++ // 0.00002 cents/token = $0.20/M, less the active 20% promotion → $0.16/M; output 1.15e-4 → $0.92/M.
++ assert.strictEqual(byKey(out.observations, 'stepfun-ai/Step-3.7-Flash', 'input').nano_usd_per_m, '160000000');
++ assert.strictEqual(byKey(out.observations, 'stepfun-ai/Step-3.7-Flash', 'output').nano_usd_per_m, '920000000');
++ // The cached field is a multiplier (0.2 of the discounted input), not a price: $0.032/M.
++ assert.strictEqual(byKey(out.observations, 'stepfun-ai/Step-3.7-Flash', 'cached_read').nano_usd_per_m, '32000000');
++ // 0.55 of $0.20/M, no discount.
++ assert.strictEqual(byKey(out.observations, 'Qwen/Qwen3-VL-235B-A22B-Instruct', 'cached_read').nano_usd_per_m, '110000000');
++});
++
++t('Featherless /v1/models fixture: USD/token strings, gated models kept, exact conversions', () => {
++ const out = parse('featherless', featherlessDoc, { fetchedAt: AT });
++ assert.strictEqual(out.models, 4);
++ assert.strictEqual(out.skipped, 0);
++ assert.strictEqual(out.observations.length, 8);
++ for (const o of out.observations) {
++ assert.match(o.nano_usd_per_m, /^\d+$/);
++ assert.strictEqual(o.source_url, PROVIDERS.featherless.url);
++ }
++ // 0.00000004 USD/token = $0.04/M = 40,000,000 nano-USD/M.
++ assert.strictEqual(byKey(out.observations, 'Qwen/Qwen3-0.6B', 'input').nano_usd_per_m, '40000000');
++ assert.strictEqual(byKey(out.observations, 'Qwen/Qwen3-0.6B', 'output').nano_usd_per_m, '80000000');
++ assert.strictEqual(byKey(out.observations, 'Qwen/Qwen3-14B', 'input').nano_usd_per_m, '120000000');
++ assert.strictEqual(byKey(out.observations, 'hexgrad/Kokoro-82M', 'output').nano_usd_per_m, '1200000000');
++ // Gated is a gate, not a missing price: the published rate is still evidence for a person to read.
++ assert.strictEqual(byKey(out.observations, 'meta-llama/Meta-Llama-3.1-8B-Instruct', 'input').nano_usd_per_m, '50000000');
++});
++
++t('the decimal arithmetic is exact and refuses a malformed value (never a zero)', () => {
++ assert.strictEqual(scaled(0.000012, 13, 'x').toString(), '120000000');
++ assert.strictEqual(scaled('0.00000004', 15, 'x').toString(), '40000000');
++ assert.strictEqual(scaled('0.011999999999999997', 9, 'x').toString(), '12000000', 'round half-up at the nano, not float truncation');
++ assert.strictEqual(scaled('5e-7', 13, 'x').toString(), '5000000');
++ assert.throws(() => scaled('free', 13, 'x'), /not a finite non-negative decimal/);
++ assert.throws(() => scaled(-1, 13, 'x'), /not a finite non-negative number/);
++ assert.throws(() => scaled(null, 13, 'x'), /not a decimal/);
++ // A price the document does not state is never defaulted to zero.
++ const missing = deepinfraDoc.map((m) => (m.model_name === 'Qwen/Qwen3-Coder-30B-A3B-Instruct'
++ ? { ...m, pricing: { ...m.pricing, cents_per_input_token: null } } : m));
++ assert.throws(() => parse('deepinfra', missing, { fetchedAt: AT }), /cents_per_input_token/);
++ assert.throws(() => parse('deepinfra', [], { fetchedAt: AT }), /no token-priced model/);
++ assert.throws(() => parse('featherless', { data: [] }, { fetchedAt: AT }), /no token-priced model/);
++ assert.throws(() => parse('featherless', { data: [{ id: 'x', pricing: { prompt: 'n/a', completion: '1' } }] }, { fetchedAt: AT }), /pricing\.prompt/);
++ // An explicit zero is a stated price (a real observation), unlike a missing one.
++ const zero = parse('featherless', { data: [{ id: 'free/tier', pricing: { prompt: '0', completion: '0' } }] }, { fetchedAt: AT });
++ assert.deepStrictEqual(zero.observations.map((o) => o.nano_usd_per_m), ['0', '0']);
++});
++
++// ── The service: a stub provider document, no live fetch ─────────────────────────────────────────
++let svc;
++let stub;
++let deepinfraBody = JSON.parse(JSON.stringify(deepinfraDoc));
++let featherlessBody = JSON.parse(JSON.stringify(featherlessDoc));
++let deepinfraEtag = null;
++const reader = () => serviceToken('services', ['watch.watch.read']);
++const admin = () => serviceToken('network', ['watch.*']);
++const person = { sub: newUser() };
++const personToken = () => userToken(person.sub);
++
++async function booted(flag = '1') {
++ stub = await site({
++ '/deepinfra': (req) => {
++ if (deepinfraEtag && req.headers['if-none-match'] === deepinfraEtag) return { status: 304, headers: { etag: deepinfraEtag } };
++ return { headers: { 'content-type': 'application/json', ...(deepinfraEtag ? { etag: deepinfraEtag } : {}) }, body: JSON.stringify(deepinfraBody) };
++ },
++ '/featherless': () => ({ headers: { 'content-type': 'application/json' }, body: JSON.stringify(featherlessBody) }),
++ });
++ svc = await boot({
++ env: { WATCH_PROVIDER_RATES: flag },
++ providerRateProviders: {
++ deepinfra: pointed('deepinfra', `${stub.origin}/deepinfra`),
++ featherless: pointed('featherless', `${stub.origin}/featherless`),
++ },
++ });
++}
++
++t('boot with the flag on: two sources, disabled until a person enables one, nothing fetched yet', async () => {
++ await booted('1');
++ const got = await request(svc.base, 'GET', '/api/v1/provider-rates', { token: reader() });
++ assert.strictEqual(got.status, 200);
++ assert.strictEqual(got.body.sources.length, 2);
++ assert.ok(got.body.sources.every((s) => s.kind === 'provider-rate' && s.enabled === false));
++ assert.strictEqual(got.body.sources[0].health.status, 'disabled');
++ assert.strictEqual(got.body.pending_changes, 0);
++ assert.deepStrictEqual(stub.hits('/deepinfra'), [], 'no fetch before a person enables it');
++ // A refresh of a disabled source is refused, and still fetches nothing.
++ const refused = await request(svc.base, 'POST', '/api/v1/provider-rates/sources/deepinfra/refresh', { token: admin() });
++ assert.strictEqual(refused.status, 409);
++ assert.deepStrictEqual(stub.hits('/deepinfra'), []);
++});
++
++t('enable both sources, then parse each fixture into observations (changed=0 on the first reading)', async () => {
++ for (const key of ['deepinfra', 'featherless']) {
++ const on = await request(svc.base, 'PATCH', `/api/v1/provider-rates/sources/${key}`, { token: admin(), body: { enabled: true } });
++ assert.strictEqual(on.status, 200, JSON.stringify(on.body));
++ assert.strictEqual(on.body.source.enabled, true);
++ }
++ const di = await request(svc.base, 'POST', '/api/v1/provider-rates/sources/deepinfra/refresh', { token: admin() });
++ assert.strictEqual(di.status, 200, JSON.stringify(di.body));
++ assert.deepStrictEqual([di.body.fetch.state, di.body.fetch.stated, di.body.fetch.observations, di.body.fetch.changed], ['ok', 8, 8, 0]);
++ assert.strictEqual(di.body.source.health.status, 'healthy');
++ const fe = await request(svc.base, 'POST', '/api/v1/provider-rates/sources/featherless/refresh', { token: admin() });
++ assert.strictEqual(fe.status, 200, JSON.stringify(fe.body));
++ assert.deepStrictEqual([fe.body.fetch.state, fe.body.fetch.stated, fe.body.fetch.observations], ['ok', 8, 8]);
++ assert.strictEqual(stub.hits('/deepinfra').length, 1);
++ assert.strictEqual(stub.hits('/featherless').length, 1);
++
++ const latest = await request(svc.base, 'GET', '/api/v1/provider-rates/observations?latest=1', { token: reader() });
++ assert.strictEqual(latest.status, 200);
++ assert.strictEqual(latest.body.observations.length, 16);
++ const obs = latest.body.observations.find((o) => o.model === 'Qwen/Qwen3-Coder-30B-A3B-Instruct' && o.token_class === 'input');
++ assert.strictEqual(obs.nano_usd_per_m, '70000000');
++ assert.strictEqual(obs.source_url, `${stub.origin}/deepinfra`, 'the URL actually fetched is recorded');
++ assert.match(obs.fetched_at, /^\d{4}-\d{2}-\d{2}T/);
++ assert.strictEqual(obs.changed, false);
++ assert.strictEqual((await request(svc.base, 'GET', '/api/v1/provider-rates/changes', { token: reader() })).body.changes.length, 0);
++});
++
++t('an unchanged document raises no change; a 304 records not_modified and no new observation', async () => {
++ const again = await request(svc.base, 'POST', '/api/v1/provider-rates/sources/deepinfra/refresh', { token: admin() });
++ assert.strictEqual(again.status, 200, JSON.stringify(again.body));
++ assert.deepStrictEqual([again.body.fetch.state, again.body.fetch.stated, again.body.fetch.observations, again.body.fetch.changed], ['ok', 8, 0, 0]);
++ assert.strictEqual((await request(svc.base, 'GET', '/api/v1/provider-rates/changes', { token: reader() })).body.changes.length, 0);
++
++ // With a validator the next read is conditional: 304, no body, no new observation, no change.
++ deepinfraEtag = '"v1"';
++ const stored = await request(svc.base, 'POST', '/api/v1/provider-rates/sources/deepinfra/refresh', { token: admin() });
++ assert.strictEqual(stored.body.fetch.state, 'ok');
++ const notModified = await request(svc.base, 'POST', '/api/v1/provider-rates/sources/deepinfra/refresh', { token: admin() });
++ assert.deepStrictEqual([notModified.body.fetch.state, notModified.body.fetch.stated, notModified.body.fetch.observations, notModified.body.fetch.changed], ['not_modified', 0, 0, 0]);
++ assert.strictEqual(stub.hits('/deepinfra').at(-1).headers['if-none-match'], '"v1"');
++ assert.strictEqual((await request(svc.base, 'GET', '/api/v1/provider-rates/changes', { token: reader() })).body.changes.length, 0);
++});
++
++t('a changed price raises exactly one Watch change with both integer values, and a person approves it', async () => {
++ deepinfraEtag = '"v2"';
++ deepinfraBody[0].pricing.cents_per_input_token = 0.000009; // $0.07/M → $0.09/M
++ const res = await request(svc.base, 'POST', '/api/v1/provider-rates/sources/deepinfra/refresh', { token: admin() });
++ assert.strictEqual(res.status, 200, JSON.stringify(res.body));
++ assert.deepStrictEqual([res.body.fetch.state, res.body.fetch.stated, res.body.fetch.observations, res.body.fetch.changed], ['ok', 8, 1, 1]);
++
++ const pending = await request(svc.base, 'GET', '/api/v1/provider-rates/changes?state=pending', { token: reader() });
++ assert.strictEqual(pending.body.changes.length, 1);
++ const change = pending.body.changes[0];
++ assert.strictEqual(change.model, 'Qwen/Qwen3-Coder-30B-A3B-Instruct');
++ assert.strictEqual(change.token_class, 'input');
++ assert.strictEqual(change.from_nano_usd_per_m, '70000000');
++ assert.strictEqual(change.to_nano_usd_per_m, '90000000');
++ assert.strictEqual(change.approval_required, true);
++ assert.strictEqual(change.effect, 'evidence');
++ assert.strictEqual(change.state, 'pending');
++
++ // The change is a Watch event, not a private table row.
++ const events = await svc.db.many("SELECT envelope FROM service_outbox WHERE envelope->>'event_type' = 'watch.rate.changed'");
++ assert.strictEqual(events.length, 1);
++ assert.strictEqual(events[0].envelope.payload.change_id, change.id);
++ assert.strictEqual(events[0].envelope.payload.approval_required, true);
++ // The released contract (openvibe-contracts v0.286.0) describes exactly what Watch sends.
++ const changedCheck = validate('watch.rate.changed@1', events[0].envelope.payload);
++ assert.ok(changedCheck.valid, JSON.stringify(changedCheck.errors));
++
++ // A service may fetch and read, never approve.
++ const service = await request(svc.base, 'POST', `/api/v1/provider-rates/changes/${change.id}/approve`, { token: admin() });
++ assert.strictEqual(service.status, 403);
++ assert.strictEqual(service.body.code, 'watch.rate_person_only');
++
++ const approved = await request(svc.base, 'POST', `/api/v1/provider-rates/changes/${change.id}/approve`, {
++ token: personToken(), body: { note: 'Reviewed the provider page; this is the list rate.' },
++ });
++ assert.strictEqual(approved.status, 200, JSON.stringify(approved.body));
++ assert.strictEqual(approved.body.applied, false, 'an approval writes no billable rate');
++ assert.strictEqual(approved.body.change.state, 'approved');
++ assert.strictEqual(approved.body.change.decided_by, person.sub);
++ assert.strictEqual(approved.body.change.note, 'Reviewed the provider page; this is the list rate.');
++ assert.match(approved.body.change.decided_at, /^\d{4}-\d{2}-\d{2}T/);
++
++ const again = await request(svc.base, 'POST', `/api/v1/provider-rates/changes/${change.id}/approve`, { token: personToken() });
++ assert.strictEqual(again.status, 409);
++ assert.strictEqual(again.body.code, 'watch.rate_decided');
++
++ const approvedEvents = await svc.db.many("SELECT envelope FROM service_outbox WHERE envelope->>'event_type' = 'watch.rate.approved'");
++ assert.strictEqual(approvedEvents.length, 1);
++ assert.strictEqual(approvedEvents[0].envelope.payload.applied, false);
++ assert.strictEqual(approvedEvents[0].envelope.payload.decided_by, person.sub);
++ const approvedCheck = validate('watch.rate.approved@1', approvedEvents[0].envelope.payload);
++ assert.ok(approvedCheck.valid, JSON.stringify(approvedCheck.errors));
++
++ // There is no rate-card table in Watch, and no route wrote a billable price.
++ const tables = await svc.db.many("SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema() AND table_name LIKE '%rate%'");
++ assert.deepStrictEqual(tables.map((r) => r.table_name).sort(), ['provider_rate_changes', 'provider_rate_fetches', 'provider_rate_observations', 'provider_rate_sources']);
++});
++
++t('a malformed document is parse_error, writes no observation, and never becomes zero', async () => {
++ const before = (await request(svc.base, 'GET', '/api/v1/provider-rates/changes', { token: reader() })).body.changes.length;
++ deepinfraEtag = '"v3"';
++ deepinfraBody[0].pricing.cents_per_input_token = 'n/a';
++ const res = await request(svc.base, 'POST', '/api/v1/provider-rates/sources/deepinfra/refresh', { token: admin() });
++ assert.strictEqual(res.status, 200, JSON.stringify(res.body));
++ assert.deepStrictEqual([res.body.fetch.state, res.body.fetch.observations, res.body.fetch.changed], ['parse_error', 0, 0]);
++ assert.match(res.body.fetch.detail, /not a finite non-negative decimal/);
++ assert.strictEqual(res.body.source.health.status, 'failing');
++ assert.strictEqual((await request(svc.base, 'GET', '/api/v1/provider-rates/changes', { token: reader() })).body.changes.length, before);
++
++ const latest = await request(svc.base, 'GET', '/api/v1/provider-rates/observations?source=deepinfra&latest=1', { token: reader() });
++ const obs = latest.body.observations.find((o) => o.model === 'Qwen/Qwen3-Coder-30B-A3B-Instruct' && o.token_class === 'input');
++ assert.strictEqual(obs.nano_usd_per_m, '90000000', 'the last good observation is still there');
++ assert.ok(latest.body.observations.every((o) => o.nano_usd_per_m !== '0'), 'nothing was defaulted to zero');
++});
++
++t('an old success is labelled stale, not a current price', async () => {
++ await svc.db.query("UPDATE provider_rate_sources SET last_success_at = $1, last_state = 'ok', consecutive_failures = 0 WHERE key = 'featherless'", [Date.now() - 13 * 3600 * 1000]);
++ const got = await request(svc.base, 'GET', '/api/v1/provider-rates/sources/featherless', { token: reader() });
++ assert.strictEqual(got.status, 200);
++ assert.strictEqual(got.body.source.health.status, 'stale');
++});
++
++t('with the flag off the routes are absent and nothing is fetched or seeded', async () => {
++ await svc.stop();
++ await stub.close();
++ const off = await boot({ env: {} });
++ try {
++ assert.strictEqual(off.config.providerRates.enabled, false, 'the module exists so the owner switch can turn it on live; off, it is inert');
++ assert.strictEqual(off.providerRates.enabled, false);
++ const got = await request(off.base, 'GET', '/api/v1/provider-rates', { token: reader() });
++ assert.strictEqual(got.status, 404);
++ assert.strictEqual(got.body.code, 'watch.not_found');
++ assert.strictEqual(await off.db.value('SELECT COUNT(*) FROM provider_rate_sources'), 0);
++ } finally {
++ await off.stop();
++ }
++});
++
++t.run();