openvibers/openvibe.watch Commits

Watch: event sources: a watch woken by OpenVibe.Events, never polled (plan T18 step 5) (#40)

OP OpenVibers committed 11 Oct 2026 · 2a8bc0d246 · 14 files, +464 −28

Plan T18 step 5, event sources: a watch can be woken by OpenVibe.Events instead of polling ("never polls when an event will do"). The contract already had the shape (`{ kind: 'event', pattern, filter }`, `cadence: null`); the carrier did not exist, so every event watch answered `watch.carrier_unavailable`.

**How it works** (`server/event-sources.js`, `server/carriers/event.js`):
- **Delivery:** a matching event arrives on the signed loopback route Watch already has (`POST /internal/events`, the account-data consumer's `onEvent` hook for every non-account event type). The check runs with the event as its input, through the existing pipeline: extraction, comparison, condition, observations, events and actions, in one commit as before. The run is `trigger: 'event'`, `carrier: 'event'`.
- **The document:** the extraction reads `{ event_type, event_id, timestamp, subject, payload }`, so `value_path: 'payload.version'` takes a release's version.
- **Matching:** `*` in a pattern is one or more whole segments, as Events matches topics. A filter is equality on `subject.id`, `subject.type` or a top-level payload field.
- **Safety and cost:**
  - Only `public` events wake a watch: anyone may own one, and an internal or private event is not theirs.
  - Each `event_id` is handled once (new table `event_source_receipts`, expand migration 0006, pruned after 7 days), so Events' at-least-once redelivery is a `duplicate`.
  - Each watch has `WATCH_EVENT_MAX_PER_HOUR` (60) event checks an hour; the rest are dropped and counted.
  - A paused or deleted watch sleeps through its events.
  - The scheduler never picks an event watch (it has no cadence).
- **Subscriptions:** one Events subscription per distinct pattern of an active event watch, to the same endpoint and secret as the account subscriptions.
  - The sync runs at boot, every `WATCH_EVENT_SYNC_MS` (10 min), and 2 s after an event watch is created, changed or removed (a registry hook).
  - It creates missing subscriptions, re-enables one a watch needs again, and disables unused ones (Events has no delete).
  - It never touches the account subscriptions or another endpoint's, and it is off without `EVENTS_URL`, `WATCH_EVENTS_SECRET` or the OAuth client secret.
- **Check now** on an event watch answers 422 `watch.event_awaited` (there is no event to read) instead of `carrier_unavailable`.

**Tests:**
- New `test/event-sources.test.js` (9/9):
  - matching rules;
  - a public matching event fires the watch and queues its notification;
  - a duplicate delivery is a no-op;
  - internal/private, a wrong filter or type, account events and paused watches wake nothing;
  - the hourly budget;
  - the signed route end to end, with a bad signature refused;
  - Check now's 422;
  - the subscription sync against a stand-in Events (create, enable, disable, and the account and foreign subscriptions left alone; off without config).
- `check` and `site` expectations follow (webhook, run and node are still later steps).
- `ov test`: 15 of 15 covering files pass.

README: a new "Event sources" section, the carrier table, and the check route's 422s.

Changes

ddiff --git a/README.md b/README.md
iindex 46fe5f0..1bed21e 100644
---- a/README.md
++++ b/README.md
@@@ -89,9 +89,10 @@ every page and every form, never 403. A signed-in write must also come from this
  The cadence is a select (15 minutes, 1 hour, 6 hours, 1 day); 15 minutes is the shortest a person may
  choose here (the contract's floor is 30 seconds, for a service). A validation error re-renders the
  form with the person's own input and the registry's own message next to the field it names. The
--kinds whose carrier does not exist yet (event, webhook, run, node) and the `ai` extraction are **not
++kinds whose carrier does not exist yet (webhook, run, node) and the `ai` extraction are **not
  offered** — and a watch of such a kind created through the API still answers `watch.carrier_unavailable`
--when checked from the site, fetching nothing.
++when checked from the site, fetching nothing. An event watch (made through the API) is checked when its event
++arrives, so Check now answers `watch.event_awaited`.
  
  Check-now from the page respects the API's limits and its refusals: a second check while one runs or
  a check of a paused watch is the API's 409, said in words; a kind without a carrier is the API's 422;
@@@ -121,14 +122,14 @@ exists. A service principal that does not name a person owns what it creates.
      webhook/event → ETag/Last-Modified → feed → API → and only then browser polling through Run
  
  The rungs run from cheapest to most expensive, and `source.kind` fixes which one a watch uses.
--This release carries the pull rungs:
++This release carries the pull rungs and event sources:
  
  | Kind | Carrier | State |
  |---|---|---|
  | `http` | one URL, conditional | implemented |
  | `feed` | RSS 2.0 / RSS 1.0 (RDF) / Atom 1.0 → `{ items, latest, count }` | implemented |
  | `api` | JSON/XML with `items_path` + `fields`, or the body itself | implemented |
--| `event` | delivered by OpenVibe.Events | step 5 |
++| `event` | delivered by OpenVibe.Events (see [Event sources](#event-sources)) | implemented |
  | `webhook` | delivered to `/internal/hooks/…` | step 5 |
  | `run` | browser/code polling through OpenVibe.Run | step 6 |
  | `node` | a probe through OpenVibe.Node | step 7 |
@@@ -137,6 +138,31 @@ A kind whose carrier does not exist yet is **accepted by the registry but never
  check is recorded `skipped` with `watch.carrier_unavailable` and the reason names the step. That is
  what keeps a half-built rung from silently turning into polling.
  
++## Event sources
++
++A watch whose source is `{ "kind": "event", "pattern": "host.release.published", "filter": { "repo": "OpenVibe.Bot" } }`
++(and `cadence: null`) is checked when a matching OpenVibe.Events event arrives, and never on a cadence, so it costs
++nothing while nothing happens (`server/event-sources.js`, `server/carriers/event.js`).
++
++- **Matching.** The pattern matches the event type as Events matches topics: `*` stands for one or more whole
++  segments (`media.*.ready`). Every `filter` key must equal the event's value: `subject.id`, `subject.type`, or a
++  top-level `payload` field.
++- **What the extraction reads.** The event is the check's document: `{ event_type, event_id, timestamp, subject,
++  payload }`, so `{ "kind": "json", "value_path": "payload.version" }` takes the released version. Comparison,
++  condition, observations and actions work exactly as for a fetched page.
++- **Only public events.** Anyone may own a watch, and an internal or private event is not theirs to read, so only
++  events with `visibility: "public"` wake a watch.
++- **Once per event.** Events delivers at least once; each `event_id` is handled once (`event_source_receipts`, kept
++  seven days), so a redelivery is answered `duplicate`.
++- **A budget.** At most `WATCH_EVENT_MAX_PER_HOUR` (60) event checks per watch an hour. More are dropped and counted,
++  so a busy pattern cannot turn one watch into a write storm.
++- **Subscriptions.** Watch keeps one Events subscription per distinct pattern of an active event watch, delivered to
++  the same signed loopback route as the account events (`POST /internal/events`, `WATCH_EVENTS_SECRET`). A sync at
++  boot, every `WATCH_EVENT_SYNC_MS` (10 minutes) and two seconds after an event watch changes creates the missing ones,
++  re-enables one a watch needs again, and disables the ones no watch needs. It never touches the account
++  subscriptions or another endpoint's. It is off without `EVENTS_URL`, the secret or the OAuth client secret.
++- **Check now** on an event watch answers 422 `watch.event_awaited`: there is no event to read.
++
  ## A check
  
  An in-process scheduler ([server/scheduler.js](server/scheduler.js)) starts checks for active
@@@ -236,7 +262,7 @@ an app or module may not.
  | `watch.watch.read` | `GET /api/v1/watches?status=&project_id=&before=&limit=`, `GET /api/v1/watches/:id` |
  | `watch.watch.manage` | `POST /api/v1/watches`, `PATCH /api/v1/watches/:id`, `POST /api/v1/watches/:id/pause`, `POST /api/v1/watches/:id/resume`, `DELETE /api/v1/watches/:id` |
  | `watch.observation.read` | `GET /api/v1/watches/:id/observations?before=&limit=`, `GET /api/v1/watches/:id/checks?before=&limit=` |
--| `watch.check.run` | `POST /api/v1/watches/:id/check` (404 unknown, 409 `watch.busy` while one runs, 409 `watch.disabled` for a paused watch, 422 `watch.carrier_unavailable`) |
++| `watch.check.run` | `POST /api/v1/watches/:id/check` (404 unknown, 409 `watch.busy` while one runs, 409 `watch.disabled` for a paused watch, 422 `watch.carrier_unavailable`, or 422 `watch.event_awaited` for an event source) |
  | `watch.resource.read` | `GET /api/v1/resources?project=&kind=&cursor=&limit=`, `GET /api/v1/resources/:ovrn` (first-party service token only) |
  
  The loopback-only authority index lists non-deleted `watch.watch` resources as
@@@ -408,8 +434,10 @@ Called elsewhere, as the service principal `watch` (client credentials from Open
    422), cross-site writes refused, and hostile names and values escaped on every page
  
  Not yet demonstrated: a check against a real site from a deployed service (nothing is deployed
--yet), the event/webhook carriers (step 5), the Run rung (step 6) and Node probes (step 7) — each answers
--`watch.carrier_unavailable` and fetches nothing today — and the action kinds beyond notifications.
++yet), the webhook carrier (step 5), the Run rung (step 6) and Node probes (step 7) — each answers
++`watch.carrier_unavailable` and fetches nothing today — and the action kinds beyond notifications. Event sources are
++covered end to end by `test/event-sources.test.js` (matching, visibility, once per event, the budget, the signed route
++and the subscription sync against a stand-in Events).
  
  ## Deploy
  
ddiff --git a/migrations/0006_event_source_receipts.sql b/migrations/0006_event_source_receipts.sql
nnew file mode 100644
iindex 0000000..fdb275c
---- /dev/null
++++ b/migrations/0006_event_source_receipts.sql
@@@ -0,0 +1,12 @@
++-- phase: expand
++-- OpenVibe.Watch: event sources (plan T18 step 5, server/event-sources.js). OpenVibe.Events delivers at least once, so
++-- each event a watch's event source handled is recorded once by its id: a second delivery of it is a duplicate and
++-- wakes no watch again. Pruned after seven days (Events' replay window is shorter).
++
++CREATE TABLE event_source_receipts (
++    event_id     text COLLATE "C" PRIMARY KEY,
++    event_type   text NOT NULL,
++    received_at  bigint NOT NULL,                -- ms since the epoch
++    watches      integer NOT NULL DEFAULT 0      -- the watches it woke
++);
++CREATE INDEX event_source_receipts_received ON event_source_receipts (received_at);
ddiff --git a/server/api/checks.js b/server/api/checks.js
iindex f1694ff..6e803d7 100644
---- a/server/api/checks.js
++++ b/server/api/checks.js
@@@ -10,7 +10,8 @@
   *
   * A manual check of a watch that is paused, disabled or failed is refused with 409: it is recorded
   * as `disabled` in the check runs, and nothing is fetched. A source whose carrier is a later step
-- * (event, webhook, node, run) answers 422 watch.carrier_unavailable — also without fetching.
++ * (webhook, node, run) answers 422 watch.carrier_unavailable, and an event source (checked when its event arrives,
++ * server/event-sources.js) 422 watch.event_awaited — also without fetching.
   */
  const express = require('express');
  const { http } = require('openvibe-contracts');
ddiff --git a/server/app.js b/server/app.js
iindex 400654e..94ddf23 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -58,7 +58,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 }) {
++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 }) {
      const app = express();
      app.disable('x-powered-by');
      app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');
@@@ -84,7 +84,8 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
      // parser runs before it; a request that came through a proxy is refused.
      if (accountData) {
          const send = accountSend || (async () => { throw new Error('OV_OAUTH_CLIENT_SECRET is not set: Watch cannot answer account events'); });
--        app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets || [], send, log }));
++        // Every other event type is an event source's (plan T18 step 5, server/event-sources.js).
++        app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets || [], send, log, onEvent: eventSources ? eventSources.onEvent : null }));
      }
      app.use('/api', express.json({ limit: '256kb', type: ['application/json', 'application/*+json'] }));
  
ddiff --git a/server/carriers/event.js b/server/carriers/event.js
nnew file mode 100644
iindex 0000000..95d1666
---- /dev/null
++++ b/server/carriers/event.js
@@@ -0,0 +1,31 @@
++'use strict';
++/**
++ * The event carrier (plan T18 step 5): an event source is pushed to, never polled. OpenVibe.Events delivers an event
++ * that matches the watch (server/event-sources.js), and the check runs with it as its input. The extraction reads the
++ * event as its document: { event_type, event_id, timestamp, subject, payload } (`payload.version`, for example).
++ *
++ * Without an event (Check now) there is nothing to read: carriers/index.js answers watch.event_awaited before this runs.
++ */
++const crypto = require('crypto');
++
++function createEventCarrier() {
++    return {
++        name: 'event',
++        async run({ input }) {
++            const document = {
++                event_type: input.event_type,
++                event_id: input.event_id,
++                timestamp: input.timestamp || null,
++                subject: input.subject || null,
++                payload: input.payload === undefined ? null : input.payload,
++            };
++            const body = JSON.stringify(document);
++            return {
++                state: 'ok', document, body, content_type: 'application/json',
++                bytes: Buffer.byteLength(body), raw_body_hash: crypto.createHash('sha256').update(body).digest('hex'),
++            };
++        },
++    };
++}
++
++module.exports = { createEventCarrier };
ddiff --git a/server/carriers/index.js b/server/carriers/index.js
iindex 9e38184..f07803b 100644
---- a/server/carriers/index.js
++++ b/server/carriers/index.js
@@@ -9,10 +9,10 @@
   * ordinary conditional GETs, and a Run check is an expensive browser/code execution. `source.kind`
   * fixes the carrier (migration 0001); PREFERENCE records the order the service is built around.
   *
-- * This release carries the pull rungs only:
-- *
   *   http  feed  api   implemented here (step 3)
-- *   event webhook     step 5
++ *   event             step 5: OpenVibe.Events delivers a matching event and the check reads it (carriers/event.js,
++ *                     server/event-sources.js); a check with no event (Check now) answers watch.event_awaited
++ *   webhook           later in step 5
   *   run               step 6
   *   node              step 7
   *
@@@ -24,13 +24,13 @@
  const { createHttpCarrier } = require('./http');
  const { createFeedCarrier } = require('./feed');
  const { createApiCarrier } = require('./api');
++const { createEventCarrier } = require('./event');
  
  /** The documented order. A source kind fixes its carrier; this is the order a watch would prefer. */
  const PREFERENCE = ['event', 'webhook', 'http', 'feed', 'api'];
  
  /** The step of plan T18 that carries each kind this release does not. */
  const LATER = {
--    event: { step: 5, note: 'an event source is delivered by OpenVibe.Events, never polled' },
      webhook: { step: 5, note: 'a webhook source is delivered to /internal/hooks/, never polled' },
      run: { step: 6, note: 'browser polling goes through OpenVibe.Run only after every cheaper rung was unavailable' },
      node: { step: 7, note: 'a node probe observes a local/LAN thing through OpenVibe.Node' },
@@@ -52,14 +52,18 @@ function createCarriers({ fetcher, config, log = console }) {
          http: createHttpCarrier({ fetcher, config, log }),
          feed: createFeedCarrier({ fetcher, config, log }),
          api: createApiCarrier({ fetcher, config, log }),
++        event: createEventCarrier(),
      };
  
      /**
--     * choose(watch) → { name, run } | { unavailable: { code, carrier, detail } }.
--     * Never fetches, never throws.
++     * choose(watch, { input }) → { name, run } | { unavailable: { code, carrier, detail } }.
++     * Never fetches, never throws. An event source runs only with the event that woke it.
       */
--    function choose(watch) {
++    function choose(watch, { input = null } = {}) {
          const kind = watch && watch.source ? watch.source.kind : null;
++        if (kind === 'event' && !input) {
++            return { unavailable: { code: 'watch.event_awaited', carrier: 'event', detail: `an event source is checked when a ${watch.source.pattern} event arrives from OpenVibe.Events. Nothing was fetched.` } };
++        }
          const carrier = carriers[kind];
          if (!carrier) return { unavailable: unavailable(kind) };
          return { name: carrier.name, run: carrier.run };
ddiff --git a/server/check.js b/server/check.js
iindex 0c2c360..826d65e 100644
---- a/server/check.js
++++ b/server/check.js
@@@ -181,7 +181,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
  
      // ── One check ────────────────────────────────────────────────────────────────────────────
  
--    async function runOnce(row, trigger) {
++    async function runOnce(row, trigger, input = null) {
          const watch = registry.fromRow(row);
          const startedAt = now();
          const carrier = watch.source && watch.source.kind ? watch.source.kind : null;
@@@ -207,7 +207,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
              return { unavailable: { code: 'watch.extraction_unavailable' }, ...out };
          }
  
--        const choice = carriers.choose(watch);
++        const choice = carriers.choose(watch, { input });
          if (choice.unavailable) {
              const out = await skip('skipped', choice.unavailable.code, choice.unavailable.detail);
              return { unavailable: choice.unavailable, ...out };
@@@ -217,7 +217,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
          const state = url ? (await st.endpoint.get(watch.id, url)) || {} : {};
          let out;
          try {
--            out = await choice.run({ watch, state, now });
++            out = await choice.run({ watch, state, now, input });
          } catch (err) {
              log.error(`[check] ${watch.id}: carrier ${carrier} crashed: ${err.stack || err}`);
              out = { state: 'http_error', error_code: 'internal', detail: 'the carrier failed unexpectedly' };
@@@ -374,7 +374,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
       *   { unavailable: { code, detail } }     the carrier (or extraction) is a later step; nothing fetched
       *   { run, observation, met, value }      the check ran
       */
--    async function run(id, { trigger = 'schedule' } = {}) {
++    async function run(id, { trigger = 'schedule', input = null } = {}) {
          const key = String(id);
          // Claimed before the first await, so two callers racing on one watch cannot both proceed.
          if (inflight.has(key)) return { busy: true };
@@@ -382,7 +382,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
          try {
              const row = await registry.get(key);
              if (!row) return null;
--            return await runOnce(row, trigger);
++            return await runOnce(row, trigger, input);
          } finally {
              inflight.delete(key);
          }
ddiff --git a/server/config.js b/server/config.js
iindex df59381..5904515 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -80,6 +80,13 @@ function load(env = process.env) {
              secrets: String(env.WATCH_EVENTS_SECRET || '').split(',').map((x) => x.trim()).filter(Boolean),
          },
  
++        // Event sources (server/event-sources.js): at most WATCH_EVENT_MAX_PER_HOUR event checks per watch, and the
++        // Events subscriptions re-synced every WATCH_EVENT_SYNC_MS (and soon after an event watch changes).
++        eventSources: {
++            maxPerWatchPerHour: int(env.WATCH_EVENT_MAX_PER_HOUR, 60),
++            syncMs: int(env.WATCH_EVENT_SYNC_MS, 10 * 60_000),
++        },
++
          // Actions (server/actions.js): every WATCH_ACTIONS_TICK_MS deliver up to WATCH_ACTIONS_BATCH queued actions of fired
          // watches. Notifications need the Network OAuth client secret (a token for network.notifications.push).
          actions: {
ddiff --git a/server/event-sources.js b/server/event-sources.js
nnew file mode 100644
iindex 0000000..5626c07
---- /dev/null
++++ b/server/event-sources.js
@@@ -0,0 +1,182 @@
++'use strict';
++/**
++ * Event sources (plan T18 step 5): a watch whose source is { kind: 'event', pattern, filter } is checked when a matching
++ * OpenVibe.Events event arrives, and never on a cadence: it costs nothing while nothing happens.
++ *
++ *   Subscriptions  one per distinct pattern of an active event watch, delivered to POST /internal/events (the route the
++ *                  account subscriptions use, signed with WATCH_EVENTS_SECRET). sync() creates the missing ones,
++ *                  re-enables one a watch needs again and disables the ones no watch needs (Events keeps a subscription;
++ *                  it has no delete). It runs at boot, every WATCH_EVENT_SYNC_MS, and shortly after an event watch changes.
++ *   Delivery       onEvent(ev), the account-data consumer's hook for every other event type. Events delivers at least
++ *                  once, so each event_id is handled once (event_source_receipts). Only `public` events wake a watch:
++ *                  anyone may own a watch, and an internal or private event is not theirs to read.
++ *   Matching       the pattern matches the event type as Events matches topics (`*` stands for one or more whole
++ *                  segments), and every filter key equals the event's value for it: `subject.id`, `subject.type`, or
++ *                  a top-level `payload` field.
++ *   Budget         at most WATCH_EVENT_MAX_PER_HOUR event checks per watch (default 60); more in the hour are counted
++ *                  and dropped, so a busy pattern cannot turn one watch into a write storm.
++ */
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++
++const ACCOUNT_TOPICS = new Set(['network.account.export_requested', 'network.account.deleted']);
++const RECEIPT_DAYS = 7;
++
++/** A topic pattern (`media.*.ready`) as a RegExp over event types: `*` is one or more whole segments. */
++function patternRegExp(pattern) {
++    const parts = String(pattern || '').split('.').map((seg) => (seg === '*' ? '[a-z0-9_-]+(?:\\.[a-z0-9_-]+)*' : seg.replace(/[^a-z0-9_-]/g, '\\$&')));
++    return new RegExp(`^${parts.join('\\.')}$`);
++}
++
++/** Does the event carry every filter value? Keys: subject.id, subject.type, or a top-level payload field. */
++function filterMatches(filter, ev) {
++    if (!filter || typeof filter !== 'object') return true;
++    for (const [key, want] of Object.entries(filter)) {
++        let have;
++        if (key === 'subject.id') have = ev.subject ? ev.subject.id : undefined;
++        else if (key === 'subject.type') have = ev.subject ? ev.subject.type : undefined;
++        else have = ev.payload && typeof ev.payload === 'object' ? ev.payload[key] : undefined;
++        if (have === undefined) return false;
++        if (have !== want) return false;
++    }
++    return true;
++}
++
++function createEventSources({ db, check, config, fetchImpl = globalThis.fetch, log = console, now = () => Date.now() }) {
++    const es = config.eventSources || {};
++    const maxPerHour = Number.isInteger(es.maxPerWatchPerHour) && es.maxPerWatchPerHour > 0 ? es.maxPerWatchPerHour : 60;
++    const endpoint = `http://127.0.0.1:${config.port}/internal/events`;
++    const secret = (config.events.secrets || [])[0] || '';
++    const enabled = !!(config.events.url && secret && config.oauth && config.oauth.clientSecret);
++    const counts = new Map();   // watch id → { hour, n }
++    let dropped = 0;
++
++    const st = {
++        receipt: db.prepare(`INSERT INTO event_source_receipts (event_id, event_type, received_at) VALUES (?, ?, ?)
++            ON CONFLICT (event_id) DO NOTHING RETURNING event_id`),
++        woke: db.prepare('UPDATE event_source_receipts SET watches = ? WHERE event_id = ?'),
++        prune: db.prepare('DELETE FROM event_source_receipts WHERE received_at < ?'),
++        watches: db.prepare(`SELECT * FROM watches WHERE status = 'active' AND deleted_at IS NULL AND source->>'kind' = 'event'`),
++    };
++
++    function underBudget(id) {
++        const hour = Math.floor(now() / 3600_000);
++        const c = counts.get(id);
++        if (!c || c.hour !== hour) { counts.set(id, { hour, n: 1 }); return true; }
++        if (c.n >= maxPerHour) { dropped++; return false; }
++        c.n++;
++        return true;
++    }
++
++    /** The account-data consumer's hook: an event that is not an account event. Returns the outcome Events records. */
++    async function onEvent(ev) {
++        if (!ev || typeof ev.event_type !== 'string' || typeof ev.event_id !== 'string') return 'ignored:shape';
++        if (ACCOUNT_TOPICS.has(ev.event_type)) return 'ignored:type';
++        const fresh = await st.receipt.get(ev.event_id, ev.event_type, now());
++        if (!fresh) return 'duplicate';
++        if (ev.visibility !== 'public') return 'ignored:visibility';
++        const rows = await st.watches.all();
++        let woke = 0;
++        for (const row of rows) {
++            const source = typeof row.source === 'string' ? JSON.parse(row.source) : row.source;
++            if (!source || source.kind !== 'event') continue;
++            if (!patternRegExp(source.pattern).test(ev.event_type) || !filterMatches(source.filter, ev)) continue;
++            if (!underBudget(row.id)) continue;
++            try {
++                const out = await check.run(row.id, { trigger: 'event', input: ev });
++                if (out && !out.busy) woke++;
++            } catch (err) {
++                // One watch failing never costs the others their check (the receipt is already written).
++                log.error(`[event-sources] ${row.id} on ${ev.event_type} ${ev.event_id}: ${err.message}`);
++            }
++        }
++        if (woke) await st.woke.run(woke, ev.event_id);
++        return woke ? `applied:${woke}` : 'applied:none';
++    }
++
++    // ── Subscriptions at OpenVibe.Events ────────────────────────────────────────────────────────────
++    const tokens = enabled ? createServiceTokenClient({
++        tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
++        audience: 'openvibe.events', scope: 'events.subscription.manage', fetch: fetchImpl,
++    }) : null;
++    async function call(method, path, body) {
++        const res = await fetchImpl(`${config.events.url}${path}`, {
++            method,
++            headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) },
++            body: body ? JSON.stringify(body) : undefined,
++            signal: AbortSignal.timeout(15000),
++        });
++        if (res.status === 401) tokens.invalidate();
++        return { status: res.status, ok: res.ok, body: await res.json().catch(() => ({})) };
++    }
++
++    /** The patterns active event watches use now. */
++    async function wanted() {
++        const out = new Set();
++        for (const row of await st.watches.all()) {
++            const source = typeof row.source === 'string' ? JSON.parse(row.source) : row.source;
++            if (source && source.kind === 'event' && source.pattern) out.add(source.pattern);
++        }
++        return out;
++    }
++
++    let syncing = null;
++    /** Make Events' subscriptions for this endpoint match the patterns in use. Returns what it did, or null when off. */
++    async function sync() {
++        if (!enabled) return null;
++        if (syncing) return syncing;
++        syncing = (async () => {
++            const want = await wanted();
++            const listed = await call('GET', '/api/v1/subscriptions');
++            if (!listed.ok) throw new Error(`listing subscriptions: ${listed.status}`);
++            const mine = (listed.body.subscriptions || []).filter((s) => s.endpoint === endpoint && !ACCOUNT_TOPICS.has(s.topic_pattern));
++            const done = { created: [], enabled: [], disabled: [] };
++            for (const pattern of want) {
++                const have = mine.find((s) => s.topic_pattern === pattern);
++                if (!have) {
++                    const r = await call('POST', '/api/v1/subscriptions', { topic_pattern: pattern, endpoint, secret });
++                    if (!r.ok && r.status !== 409) throw new Error(`subscribing to ${pattern}: ${r.status} ${r.body.code || ''}`.trim());
++                    if (r.ok) done.created.push(pattern);
++                } else if (have.enabled === false) {
++                    const r = await call('POST', `/api/v1/subscriptions/${encodeURIComponent(have.id)}/enable`);
++                    if (r.ok) done.enabled.push(pattern);
++                }
++            }
++            for (const s of mine) {
++                if (!want.has(s.topic_pattern) && s.enabled !== false) {
++                    const r = await call('POST', `/api/v1/subscriptions/${encodeURIComponent(s.id)}/disable`);
++                    if (r.ok) done.disabled.push(s.topic_pattern);
++                }
++            }
++            if (done.created.length || done.enabled.length || done.disabled.length) log.log(`[event-sources] subscriptions: ${JSON.stringify(done)}`);
++            return done;
++        })().finally(() => { syncing = null; });
++        return syncing;
++    }
++
++    let soon = null;
++    /** A watch with an event source changed: sync in a moment (several changes, one sync). */
++    function kick() {
++        if (!enabled || soon) return;
++        soon = setTimeout(() => { soon = null; sync().catch((err) => log.warn(`[event-sources] sync: ${err.message}`)); }, 2000);
++        soon.unref?.();
++    }
++
++    let timer = null;
++    function start() {
++        if (!enabled || timer) return;
++        sync().catch((err) => log.warn(`[event-sources] sync: ${err.message}`));
++        timer = setInterval(() => sync().catch((err) => log.warn(`[event-sources] sync: ${err.message}`)), es.syncMs || 10 * 60_000);
++        timer.unref?.();
++    }
++    function stop() {
++        if (timer) clearInterval(timer);
++        if (soon) clearTimeout(soon);
++        timer = soon = null;
++    }
++
++    async function prune() { await st.prune.run(now() - RECEIPT_DAYS * 86400_000); }
++
++    return { onEvent, sync, kick, start, stop, prune, enabled, stats: () => ({ dropped, maxPerHour }) };
++}
++
++module.exports = { createEventSources, patternRegExp, filterMatches, ACCOUNT_TOPICS };
ddiff --git a/server/index.js b/server/index.js
iindex ca0548c..43a8e83 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -23,6 +23,7 @@ const { extract } = require('./extract');
  const condition = require('./condition');   // { evaluate, hasChanged, holds, … }
  const { createCheck } = require('./check');
  const { createScheduler } = require('./scheduler');
++const { createEventSources } = require('./event-sources');
  const { createServiceOutbox } = require('openvibe-sdk/events');
  const { validate } = require('openvibe-contracts');
  const { jwksClient, createServiceTokenClient } = require('openvibe-sdk/auth');
@@@ -75,6 +76,9 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
      const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, now, log });
      const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });
      const scheduler = createScheduler({ db, check, config, now, log });
++    // 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());
      const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
      const auth = createAuth({ config, log });
      // The public site: sessions in Watch's own database, the Network key to verify a sign-in token
@@@ -88,7 +92,7 @@ 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 });
++    const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources });
      // 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();
@@@ -98,6 +102,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
      actions.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}`); }
          try { await sessions.prune(); } catch (err) { log.error(`[sessions] prune: ${err.message}`); }
      }, 3600 * 1000);
      pruneTimer.unref?.();
@@@ -118,11 +123,13 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
          eventsUrl: config.events.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`, secret: (config.events.secrets || [])[0],
          networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl, log,
      }) : null;
++    if (listen) eventSources.start();
  
      async function close() {
          clearInterval(pruneTimer);
          clearInterval(outboxPruneTimer);
          if (subscriptions) subscriptions.stop();
++        eventSources.stop();
          keys.stop();
          await scheduler.stop();
          await actions.stop();
@@@ -134,7 +141,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
          if (!givenDb) await db.close();
      }
  
--    return { config, db, registry, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
++    return { config, db, registry, eventSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
  }
  
  /**
ddiff --git a/server/registry.js b/server/registry.js
iindex 470a051..d3fc468 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -239,6 +239,11 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
          return row;
      }
  
++    // Event sources (server/event-sources.js) re-sync their Events subscriptions when an event watch comes or goes.
++    let onEventSource = null;
++    const touchesEvents = (...sources) => sources.some((s) => s && s.kind === 'event');
++    const eventSourceChanged = () => { if (onEventSource) { try { onEventSource(); } catch { /* the periodic sync catches up */ } } };
++
      async function create(body, principal) {
          const req = validateRequest(body, { create: true });
          const owner = ownerOf(principal);
@@@ -289,6 +294,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
              await st.insert.run(row);
              if (outbox) await outbox.emit(envelope(watchEvent('watch.watch.created', row, owner, { name: rec.name, status: rec.status }), source, now));
          });
++        if (touchesEvents(rec.source)) eventSourceChanged();
          return await st.get.get(row.id);
      }
  
@@@ -327,6 +333,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
                  await outbox.emit(envelope(watchEvent(type, row, row.owner_sub, { name: rec.name, status: rec.status, fields: touched.sort() }), source, now));
              }
          });
++        if (touchesEvents(current.source, rec.source)) eventSourceChanged();
          return await st.get.get(row.id);
      }
  
@@@ -339,6 +346,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
              await db.prepare('UPDATE watches SET deleted_at = ?, updated_at = ?, next_due_at = 0 WHERE id = ?').run(t, t, row.id);
              if (outbox) await outbox.emit(envelope(watchEvent('watch.watch.removed', row, row.owner_sub, { name: row.name, status: row.status }), source, now));
          });
++        if (touchesEvents(fromRow(row).source)) eventSourceChanged();
          return { id: row.id, deleted: true };
      }
  
@@@ -378,7 +386,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
          };
      }
  
--    return { validate: validateRequest, validateView, view, health, fromRow, create, get, all, patch, remove, due };
++    return { validate: validateRequest, validateView, view, health, fromRow, create, get, all, patch, remove, due, onEventSourceChange: (fn) => { onEventSource = fn; } };
  }
  
  module.exports = { createRegistry, validateRequest, validateView, RegistryError, defaultName, nextDue, POLLING, PUSHED, RUN_STATES, STATUSES };
ddiff --git a/test/check.test.js b/test/check.test.js
iindex bff40a6..46c6a11 100644
---- a/test/check.test.js
++++ b/test/check.test.js
@@@ -186,9 +186,8 @@ t('a pattern that hits its deadline is a recorded check error, never a hang', as
      price = '19.99';
  });
  
--t('an event, webhook, node or run source is accepted but never fetched (steps 5-7)', async () => {
++t('a webhook, node or run source is accepted but never fetched (steps 5-7); an event source waits for its event', async () => {
      const cases = [
--        ['event', { kind: 'event', pattern: 'release.published' }, { op: 'exists' }],
          ['webhook', { kind: 'webhook', endpoint: '/internal/hooks/abc' }, { op: 'exists' }],
          ['node', { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, { op: 'exists' }],
          ['run', { kind: 'run', runtime: 'browser', url: `${web.origin}/price`, steps: [{ goto: 'https://example.org' }] }, { op: 'exists' }],
ddiff --git a/test/event-sources.test.js b/test/event-sources.test.js
nnew file mode 100644
iindex 0000000..b807b7a
---- /dev/null
++++ b/test/event-sources.test.js
@@@ -0,0 +1,156 @@
++'use strict';
++/**
++ * Event sources (plan T18 step 5): a watch whose source is an OpenVibe.Events pattern is checked when a matching public
++ * event arrives, never on a cadence. Each event is handled once; only public events wake a watch; a watch has an hourly
++ * budget of event checks; Check now answers watch.event_awaited; and the Events subscriptions follow the patterns in use.
++ */
++const assert = require('assert');
++const { boot, suite, newUser } = require('./helpers');
++const { createEventSources, patternRegExp, filterMatches } = require('../server/event-sources');
++const { signDeliveryHeaders } = require('openvibe-sdk/events');
++
++const SECRET = 'event-sources-test-secret-0123456789abcdef';
++
++const t = suite('event-sources');
++let svc;
++const person = { sub: newUser() };
++let n = 0;
++const evt = (over = {}) => ({
++    event_id: `evt_${String(++n).padStart(4, '0')}`, event_type: 'host.release.published', visibility: 'public',
++    timestamp: new Date().toISOString(), subject: { type: 'release', id: `rel_${n}` },
++    payload: { repo: 'OpenVibe.Bot', version: `1.${n}.0` }, ...over,
++});
++const def = (over = {}) => ({
++    name: 'bot releases',
++    source: { kind: 'event', pattern: 'host.release.published', filter: { repo: 'OpenVibe.Bot' } },
++    cadence: null,
++    extraction: { kind: 'json', value_path: 'payload.version' },
++    condition: { op: 'changed' },
++    action: [{ kind: 'notification' }],
++    ...over,
++});
++const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);
++const observations = async (id) => await svc.db.prepare('SELECT * FROM observations WHERE watch_id = ? ORDER BY rid').all(id);
++
++t('boot', async () => {
++    svc = await boot({ env: { WATCH_EVENT_MAX_PER_HOUR: '3', WATCH_EVENTS_SECRET: SECRET } });
++});
++
++t('patterns match as Events matches topics; filters read subject.id, subject.type and payload fields', () => {
++    assert.ok(patternRegExp('host.release.published').test('host.release.published'));
++    assert.ok(!patternRegExp('host.release.published').test('host.release.published.extra'));
++    assert.ok(patternRegExp('media.*.ready').test('media.vod.ready'));
++    assert.ok(patternRegExp('media.*.ready').test('media.clip.thumb.ready'), '* is one or more whole segments');
++    assert.ok(!patternRegExp('media.*.ready').test('media.ready'));
++    assert.ok(patternRegExp('live.stream.*').test('live.stream.started'));
++    const ev = { subject: { type: 'release', id: 'rel_1' }, payload: { repo: 'OpenVibe.Bot', n: 2, ok: true } };
++    assert.ok(filterMatches(null, ev) && filterMatches({}, ev));
++    assert.ok(filterMatches({ 'subject.id': 'rel_1', 'subject.type': 'release', repo: 'OpenVibe.Bot', n: 2, ok: true }, ev));
++    assert.ok(!filterMatches({ repo: 'OpenVibe.Live' }, ev));
++    assert.ok(!filterMatches({ missing: null }, ev), 'a field the event does not carry never matches');
++});
++
++t('a matching public event runs the check with the event as its document; the value is recorded', async () => {
++    const w = await svc.registry.create(def(), person);
++    assert.strictEqual(Number((await svc.db.prepare('SELECT next_due_at FROM watches WHERE id = ?').get(w.id)).next_due_at) >= 0, true);
++    const e = evt();
++    assert.strictEqual(await svc.eventSources.onEvent(e), 'applied:1');
++    const [r] = await runs(w.id);
++    assert.deepStrictEqual([r.trigger, r.carrier, r.state], ['event', 'event', 'condition_met'], 'a first value is a change: the watch fires');
++    assert.strictEqual(Number(r.triggers), 1);
++    const [o] = await observations(w.id);
++    assert.strictEqual(o.value, e.payload.version);
++    const queued = await svc.db.prepare('SELECT kind, state FROM action_runs WHERE watch_id = ?').all(w.id);
++    assert.strictEqual(queued.length, 1, 'its notification is queued like any firing');
++    // The same delivery again (Events is at least once): a duplicate, and no second run.
++    assert.strictEqual(await svc.eventSources.onEvent(e), 'duplicate');
++    assert.strictEqual((await runs(w.id)).length, 1);
++    await svc.registry.remove(w.id, person);
++});
++
++t('only public events wake a watch, and only the ones its pattern and filter match', async () => {
++    const w = await svc.registry.create(def(), person);
++    assert.strictEqual(await svc.eventSources.onEvent(evt({ visibility: 'internal' })), 'ignored:visibility');
++    assert.strictEqual(await svc.eventSources.onEvent(evt({ visibility: 'private' })), 'ignored:visibility');
++    assert.strictEqual(await svc.eventSources.onEvent(evt({ payload: { repo: 'OpenVibe.Live', version: '9.0.0' } })), 'applied:none');
++    assert.strictEqual(await svc.eventSources.onEvent(evt({ event_type: 'host.release.failed' })), 'applied:none');
++    assert.strictEqual(await svc.eventSources.onEvent(evt({ event_type: 'network.account.deleted' })), 'ignored:type', 'account events are the consumer\'s');
++    assert.strictEqual((await runs(w.id)).length, 0);
++    // A paused watch sleeps through its events.
++    await svc.registry.patch(w.id, { status: 'paused' }, person);
++    assert.strictEqual(await svc.eventSources.onEvent(evt()), 'applied:none');
++    await svc.registry.remove(w.id, person);
++});
++
++t('a watch takes at most WATCH_EVENT_MAX_PER_HOUR event checks an hour; the rest are dropped and counted', async () => {
++    const w = await svc.registry.create(def(), person);
++    const before = svc.eventSources.stats().dropped;
++    const outcomes = [];
++    for (const e of [evt(), evt(), evt(), evt()]) outcomes.push(await svc.eventSources.onEvent(e));
++    assert.deepStrictEqual(outcomes, ['applied:1', 'applied:1', 'applied:1', 'applied:none']);
++    assert.strictEqual((await runs(w.id)).length, 3);
++    assert.strictEqual(svc.eventSources.stats().dropped, before + 1);
++    await svc.registry.remove(w.id, person);
++});
++
++t('through the signed loopback route: a delivery wakes the watch, and Events gets the outcome', async () => {
++    const w = await svc.registry.create(def(), person);
++    const e = evt();
++    const body = JSON.stringify({ event: e });
++    const res = await fetch(`${svc.base}/internal/events`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });
++    assert.strictEqual(res.status, 200);
++    assert.deepStrictEqual(await res.json(), { event_id: e.event_id, outcome: 'applied:1' });
++    const bad = await fetch(`${svc.base}/internal/events`, { method: 'POST', body: JSON.stringify({ event: evt() }), headers: { 'Content-Type': 'application/json', ...signDeliveryHeaders(body, 'x'.repeat(40)) } });
++    assert.strictEqual(bad.status, 401, 'an unsigned or mis-signed delivery wakes nothing');
++    assert.strictEqual((await runs(w.id)).length, 1);
++    await svc.registry.remove(w.id, person);
++});
++
++t('Check now on an event source answers watch.event_awaited and fetches nothing', async () => {
++    const w = await svc.registry.create(def(), person);
++    const out = await svc.check.run(w.id, { trigger: 'manual' });
++    assert.strictEqual(out.run.state, 'skipped');
++    assert.strictEqual(out.run.error_code, 'watch.event_awaited');
++    assert.strictEqual(out.unavailable.code, 'watch.event_awaited');
++    assert.match(out.run.detail, /host\.release\.published event arrives/);
++    await svc.registry.remove(w.id, person);
++});
++
++t('sync: one subscription per pattern in use; re-enables, disables the unused, never touches the account topics', async () => {
++    const endpoint = `http://127.0.0.1:${svc.config.port}/internal/events`;
++    const subs = [
++        { id: 'sub_acct', topic_pattern: 'network.account.deleted', endpoint, enabled: true },
++        { id: 'sub_old', topic_pattern: 'shop.price.changed', endpoint, enabled: true },
++        { id: 'sub_off', topic_pattern: 'media.*.ready', endpoint, enabled: false },
++        { id: 'sub_other', topic_pattern: 'shop.price.changed', endpoint: 'http://127.0.0.1:9/elsewhere', enabled: true },
++    ];
++    const calls = [];
++    const fetchImpl = async (url, opts = {}) => {
++        const u = new URL(url);
++        calls.push(`${opts.method || 'GET'} ${u.pathname}`);
++        const json = (status, body) => ({ status, ok: status < 300, json: async () => body });
++        if (u.pathname === '/oauth/token') return json(200, { access_token: 'tok', token_type: 'Bearer', expires_in: 300 });
++        if (u.pathname === '/api/v1/subscriptions' && opts.method === 'GET') return json(200, { subscriptions: subs });
++        if (u.pathname === '/api/v1/subscriptions' && opts.method === 'POST') { const b = JSON.parse(opts.body); return json(201, { id: 'sub_new', topic_pattern: b.topic_pattern }); }
++        if (/\/(enable|disable)$/.test(u.pathname)) return json(200, {});
++        return json(404, {});
++    };
++    const w1 = await svc.registry.create(def(), person);
++    const w2 = await svc.registry.create(def({ source: { kind: 'event', pattern: 'media.*.ready', filter: null } }), person);
++    const config = { ...svc.config, events: { ...svc.config.events, url: 'http://events.test', secrets: ['s'.repeat(40)] } };
++    const es = createEventSources({ db: svc.db, check: svc.check, config, fetchImpl, log: { log() {}, warn() {}, error() {} } });
++    const done = await es.sync();
++    assert.deepStrictEqual(done, { created: ['host.release.published'], enabled: ['media.*.ready'], disabled: ['shop.price.changed'] });
++    assert.ok(calls.includes('POST /api/v1/subscriptions/sub_old/disable'));
++    assert.ok(!calls.some((c) => c.includes('sub_acct') || c.includes('sub_other')), 'account and foreign subscriptions are left alone');
++    await svc.registry.remove(w1.id, person);
++    await svc.registry.remove(w2.id, person);
++    // Off without Events or the secret: nothing is called.
++    const off = createEventSources({ db: svc.db, check: svc.check, config: { ...svc.config, events: { ...svc.config.events, url: '' } }, fetchImpl });
++    assert.strictEqual(await off.sync(), null);
++    assert.strictEqual(off.enabled, false);
++});
++
++t('stop', async () => { await svc.stop(); });
++
++t.run();
ddiff --git a/test/site.test.js b/test/site.test.js
iindex 84c6859..d7adf37 100644
---- a/test/site.test.js
++++ b/test/site.test.js
@@@ -293,7 +293,7 @@ t('a kind whose carrier is not built is never fetched: check-now is 422 with the
      assert.strictEqual(made.status, 201, made.text);
      const res = await svc.get(`/watches/${made.body.watch.id}/check`, { cookie: session, form: {} });
      assert.strictEqual(res.status, 422);
--    assert.match(res.text, /not checked yet|carrier/i);
++    assert.match(res.text, /event arrives|not checked yet|carrier/i, 'an event source is checked when its event arrives');
  });
  
  t('check-now is limited per person, with a friendly 429 and Retry-After', async () => {