openvibers/openvibe.vision Commits

Vision's events get a durable outbox (#13)

OP OpenVibers committed 14 days ago · 06783aeaee · 12 files, +378 −74

Vision publishes vision.result.ready and vision.alert to OpenVibe.Events. A publish that failed
because Events was down or the network blipped was lost: the failure was logged and the event was
gone.

The envelopes now go into Vision's own event_outbox (migration 0004_event_outbox.sql) and the
openvibe-sdk outbox relay sends them, so a failed publish waits in the table and goes out with the
next drain, with backoff (1s, 5s, 30s, 2 min, 10 min). Delivery is at least once and the event's own
id is the idempotency key Events dedupes on, so a redelivery is a duplicate, not a second event.

- server/events/outbox.js: createServiceOutbox (openvibe-sdk 0.72.0, the helper nine services use),
  the contracts check for the envelope and the payload, and the relay's lifecycle
- server/events/publisher.js: the envelopes are written to the outbox instead of published inline, so
  publishing is never on the request's critical path
- server/index.js: the relay starts with the process and settles before the store closes; a drill
  never publishes and never writes a row
- test/events-outbox.test.js: a failed publish is kept and sent on the next drain, a successful one is
  removed, and the drain stays off in a drill and without Events configured

Changes

ddiff --git a/.env.example b/.env.example
iindex 79790ea..7191a34 100644
---- a/.env.example
++++ b/.env.example
@@@ -35,6 +35,8 @@ VISION_EVENTS_SECRET=
VISION_EVENTS_URL=
#VISION_EVENTS_ENDPOINT=http://127.0.0.1:5130/internal/events
#VISION_EVENTS_INTERVAL_MS=5000
++# The event outbox relay: how often the drain looks for events that are due (the retry backoff is 1s, 5s, 30s, 2min, 10min).
++#VISION_EVENTS_OUTBOX_INTERVAL_MS=2000
# OpenVibe.AI: the only model gateway. A first-party service token for client `vision` (Network grant
# ai.run.create, namespace ai.*) makes the run; AI's routing, quotas, cache and cost accounting apply. Vision
ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex c26c4cd..152edb5 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -10,6 +10,10 @@ What changed in OpenVibe.Vision, newest first.
tools remain open. Verification gives the same refusal before and after consent until enrollment is complete.
- Publishes `vision.result.ready` and `vision.alert` through OpenVibe.Events when Events is configured. Event
payloads contain provenance and class names, never image bytes, answer text or face templates.
++- Publishes through a durable outbox (`event_outbox`, migration `0004_event_outbox.sql`): the event waits in the
++ table until OpenVibe.Events takes it, so a publish that fails because Events is down or the network blipped is
++ retried with backoff instead of lost. Delivery is at least once and the event's own id is the idempotency key
++ Events dedupes on. The relay is off in a drill.
## 0.1.0 — 2026-10-28
ddiff --git a/README.md b/README.md
iindex 515f490..a3ed40c 100644
---- a/README.md
++++ b/README.md
@@@ -78,6 +78,7 @@ outbound capability; AI's quotas and its own route decide whether a run is allow
| `face_media` | owner, AES-GCM ciphertext of an opt-in Vision upload and its face vectors, media type | `0003_faces.sql` |
| `face_tags` | consenting subject, channel, private media reference and time; removed on cast revocation | `0003_faces.sql` |
| `face_clusters` | owner, media, encrypted vector, group id, optional label (null while unknown), excluded flag | `0003_faces.sql` |
++| `event_outbox` | the events Vision publishes wait here until OpenVibe.Events takes them: the envelope, its own event id (the idempotency key), the attempts and the last error; sent rows are pruned after 7 days | `migrations/0004_event_outbox.sql` |
General answer tables store no image bytes, token or provider key. The opt-in face library stores encrypted uploads and vectors.
ddiff --git a/STATUS.json b/STATUS.json
iindex f7f3852..b454b14 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -56,7 +56,7 @@
"Encrypted face templates, opt-in uploads and vectors (AES-256-GCM); immediate template/tag erasure on opt-out and a daily three-year retention job",
"Network tools: routed alt text, owner-scoped CLIP index/find, Tesseract-first receipts, product vectors, DETR watch and moments, each with a receipt",
"MCP on /mcp with RFC 9728 protected-resource metadata, token challenges and route-equivalent capability checks",
-- "vision.result.ready and vision.alert published through OpenVibe.Events when configured",
++ "vision.result.ready and vision.alert published through OpenVibe.Events when configured, through a durable outbox: a failed publish waits in event_outbox and is retried with backoff",
"Answers, compare samples, private index and UI checks stored without image bytes; account export/deletion over every table including faces and the private index",
"Network sign-in as OAuth client vision (PKCE S256, vision_at/vision_rt/vision_oauth), truthful readiness, loopback metrics",
"Server-rendered pages with stable vi- ids/classes, discovery artifacts, deploy unit and nginx vhost, CI"
ddiff --git a/migrations/0004_event_outbox.sql b/migrations/0004_event_outbox.sql
nnew file mode 100644
iindex 0000000..5b0c436
---- /dev/null
++++ b/migrations/0004_event_outbox.sql
@@@ -0,0 +1,20 @@
++-- phase: expand
++-- Vision's own event outbox: the events Vision publishes to OpenVibe.Events wait here until Events takes them.
++-- A publish that fails because Events is down or the network blipped is not lost: the row stays and the relay
++-- sends it later, with backoff. The DDL is openvibe-sdk/events' outboxSchema() (openvibe-sdk 0.72.0); the
++-- runtime role cannot create tables.
++CREATE TABLE IF NOT EXISTS event_outbox (
++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ event_id text NOT NULL UNIQUE,
++ envelope jsonb NOT NULL,
++ traceparent text,
++ created_at bigint NOT NULL,
++ attempts integer NOT NULL DEFAULT 0,
++ next_attempt_at bigint NOT NULL DEFAULT 0,
++ sent_at bigint,
++ seq bigint,
++ rejected_at bigint,
++ last_error text
++);
++CREATE INDEX IF NOT EXISTS event_outbox_due ON event_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
++CREATE INDEX IF NOT EXISTS event_outbox_sent ON event_outbox (sent_at) WHERE sent_at IS NOT NULL;
ddiff --git a/server/app.js b/server/app.js
iindex d996df0..d8263ee 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -44,6 +44,7 @@ const { createFacesModel } = require('./vision/faces-model');
const { createFaces } = require('./vision/faces');
const { createNetworkSender } = require('openvibe-sdk/account-data');
const { createPublisher } = require('./events/publisher');
++const { createVisionOutbox } = require('./events/outbox');
const PUBLIC_DIR = path.join(__dirname, '..', 'public');
const VERSION = require('../package.json').version;
@@@ -79,8 +80,11 @@ async function createApp(opts = {}) {
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
: async () => { throw new Error('OV_OAUTH_CLIENT_SECRET is not set: Vision cannot answer account events'); });
-- const eventPublisher = opts.eventPublisher || createPublisher({ config, fetchImpl, log });
-- const ctx = { config, s, keys, sso, principal, store, ocr, ai, vision, faces, models, networkTools, accountData, accountSend, eventPublisher, log, fetchImpl };
++ // Vision's own event outbox (server/events/outbox.js): the events Vision publishes wait in event_outbox
++ // until OpenVibe.Events takes them, so a failed publish is not lost. The relay starts in server/index.js.
++ const eventOutbox = opts.eventOutbox || createVisionOutbox({ s, config, log, fetchImpl });
++ const eventPublisher = opts.eventPublisher || createPublisher({ outbox: eventOutbox, log });
++ const ctx = { config, s, keys, sso, principal, store, ocr, ai, vision, faces, models, networkTools, accountData, accountSend, eventOutbox, eventPublisher, log, fetchImpl };
const app = express();
app.disable('x-powered-by');
ddiff --git a/server/config.js b/server/config.js
iindex b90c38c..f60ffcd 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -58,6 +58,9 @@ function load(env = process.env) {
url: trim(env.VISION_EVENTS_URL || env.EVENTS_URL || ''),
endpoint: env.VISION_EVENTS_ENDPOINT || '',
intervalMs: Math.max(500, int(env.VISION_EVENTS_INTERVAL_MS, 5000)),
++ // The outbox relay: how often the drain looks for events that are due (the retry backoff is the
++ // SDK outbox's own: 1s, 5s, 30s, 2 min, 10 min).
++ outboxIntervalMs: Math.max(250, int(env.VISION_EVENTS_OUTBOX_INTERVAL_MS, 2000)),
scope: env.VISION_EVENTS_SCOPE || 'events.event.publish',
},
ddiff --git a/server/events/outbox.js b/server/events/outbox.js
nnew file mode 100644
iindex 0000000..fc49c52
---- /dev/null
++++ b/server/events/outbox.js
@@@ -0,0 +1,80 @@
++'use strict';
++
++/**
++ * Vision's event outbox: the events Vision publishes to OpenVibe.Events wait in `event_outbox`
++ * (migrations/0004_event_outbox.sql) until Events takes them. A publish that fails because Events is down or
++ * the network blipped is not lost: the row stays and the relay sends it later, with backoff.
++ *
++ * openvibe-sdk/events' createServiceOutbox (openvibe-sdk 0.72.0) is the helper nine services already use
++ * (Trade, News, Host, Codes, Blog, Deals, Coupons, VIP, Tips). It writes the envelope with its own `event_id`
++ * — the idempotency key Events dedupes on — and the relay publishes the due rows with Vision's service token
++ * (events.event.publish, audience openvibe.events).
++ *
++ * Delivery is at least once: a crash between the publish and the row being marked sent republishes the same
++ * event_id, and Events answers that as a duplicate. A permanent refusal (a 4xx other than 401/408/409/425/429)
++ * marks the row rejected and never retries it; anything else is retried with backoff.
++ *
++ * The relay is off in a drill (VISION_DRILL=1): a drill publishes nothing and writes no row. Without the events
++ * URL or the OAuth client secret the relay is off too, and Vision publishes no event at all.
++ */
++const contracts = require('openvibe-contracts');
++const { createServiceOutbox } = require('openvibe-sdk/events');
++
++/** The event types Vision publishes (contracts 0.240.0). */
++const EVENT_TYPES = ['vision.result.ready', 'vision.alert'];
++
++function createVisionOutbox({ s, config, log = console, fetchImpl = null }) {
++ const drill = Boolean(config.drill);
++ const out = createServiceOutbox({
++ db: s.db,
++ source: 'vision',
++ eventsUrl: config.events.url || null,
++ networkInternalUrl: config.networkInternalUrl,
++ clientId: config.oauth.clientId,
++ clientSecret: config.oauth.clientSecret,
++ intervalMs: config.events.outboxIntervalMs,
++ log,
++ eventTypes: EVENT_TYPES,
++ ...(fetchImpl ? { fetch: fetchImpl } : {}),
++ validate: (env) => {
++ const envelope = contracts.validate('events.event-envelope@1', env);
++ if (!envelope.valid) return envelope;
++ return contracts.validate(`${env.event_type}@${env.version}`, env.payload);
++ },
++ });
++ // The SDK's own switch (the events URL and the client secret are set); a drill turns the relay off on top.
++ const enabled = Boolean(out.enabled) && !drill;
++
++ return {
++ EVENT_TYPES,
++ /** Whether the relay may publish. False in a drill and without the events URL or the client secret. */
++ enabled,
++ /** The SDK relay itself (flush, pending, rejected, prune): the drain a test drives by hand. */
++ outbox: out.outbox,
++ /**
++ * Write one envelope into the outbox. The row is durable before the caller answers, and the relay takes
++ * it from there: publishing is never on the request's critical path.
++ *
++ * Nothing is written when the relay could not publish anyway (a drill, or Events not configured): Vision
++ * publishes no event at all then, exactly as it did before the outbox.
++ */
++ async emit(envelope, opts = {}) {
++ if (!enabled) return null;
++ const env = await out.emit(envelope, opts);
++ // Wake the relay now: a queued event does not wait for the next tick.
++ if (enabled) Promise.resolve(out.kick()).catch(() => {});
++ return env;
++ },
++ /** Start the relay (the background drain): due rows go out, a failure backs off. */
++ start() { if (enabled) out.start(); },
++ stop: () => out.stop(),
++ /** Wake the relay now (a queued event does not wait for the next tick). */
++ async kick() { if (enabled) await out.kick(); },
++ async status() {
++ const status = await out.status();
++ return { ...status, enabled };
++ },
++ };
++}
++
++module.exports = { createVisionOutbox, EVENT_TYPES };
ddiff --git a/server/events/publisher.js b/server/events/publisher.js
iindex d9da5e4..8381bdb 100644
---- a/server/events/publisher.js
++++ b/server/events/publisher.js
@@@ -1,23 +1,28 @@
'use strict';
--const { createClient } = require('openvibe-sdk/core');
--const { createServiceTokenClient } = require('openvibe-sdk/auth');
--const { createEventsClient } = require('openvibe-sdk/events');
--
++/**
++ * Vision → OpenVibe.Events (contracts 0.240.0): the events Vision publishes.
++ *
++ * vision.result.ready a visual result is ready for its owner or caller (never the answer text, never an image)
++ * vision.alert an owner-granted camera alert (a watched source saw something)
++ *
++ * The envelope goes into Vision's own outbox (server/events/outbox.js) and the relay sends it, so a publish
++ * that fails because Events is down or the network blipped is not lost: the row waits in the table and goes out
++ * with the next drain, the event's own id as the idempotency key. Publishing is never on the request's
++ * critical path.
++ */
const TASKS = Object.freeze({
'/ask': 'ask', '/describe': 'describe', '/ocr': 'ocr', '/compare': 'compare', '/alt': 'alt',
'/find': 'find', '/receipt': 'receipt', '/product': 'product', '/watch': 'watch',
'/moments': 'moments', '/ui-check': 'ui_check',
});
--function createPublisher({ config, fetchImpl = globalThis.fetch, log = console } = {}) {
-- if (config.drill || !config.events.url || !config.oauth.clientSecret) return { result: async () => {}, alert: async () => {} };
-- const tokenProvider = createServiceTokenClient({
-- tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId,
-- clientSecret: config.oauth.clientSecret, audience: 'openvibe.events', scope: config.events.scope, fetch: fetchImpl,
-- });
-- const client = createClient({ baseUrls: { events: config.events.url }, tokenProvider, fetch: fetchImpl, retries: 1 });
-- const events = createEventsClient(client, { source: 'vision' });
++/**
++ * The publisher: result(req, out) and alert(req, out) write one envelope each into the outbox.
++ * The outbox (Vision's own, server/events/outbox.js) owns the relay, the event id and the retry.
++ */
++function createPublisher({ outbox, log = console } = {}) {
++ if (!outbox || typeof outbox.emit !== 'function') throw new TypeError('createPublisher: the Vision event outbox is required');
function subject(req) {
const p = req.principal;
@@@ -27,10 +32,18 @@ function createPublisher({ config, fetchImpl = globalThis.fetch, log = console }
}
async function publish(req, type, payload) {
const ref = subject(req);
-- if (!ref) return;
++ if (!ref) return null;
try {
-- await events.publish({ event_type: type, actor: ref, subject: ref, visibility: ref.type === 'user' ? 'subject' : 'internal', priority: 'important', payload }, { traceparent: req.ov && req.ov.traceparent });
-- } catch (err) { log.warn('[OpenVibe.Vision] event publish failed:', String(err && err.message || err).slice(0, 180)); }
++ return await outbox.emit(
++ { event_type: type, actor: ref, subject: ref, visibility: ref.type === 'user' ? 'subject' : 'internal', priority: 'important', payload },
++ { traceparent: req.ov && req.ov.traceparent },
++ );
++ } catch (err) {
++ // The row could not be written (the database is unhappy, the envelope broke a contract): say so,
++ // but never fail the person's request over an event.
++ log.warn('[OpenVibe.Vision] event outbox write failed:', String((err && err.message) || err).slice(0, 180));
++ return null;
++ }
}
async function result(req, out) {
const task = req.visionTask || TASKS[req.path];
ddiff --git a/server/index.js b/server/index.js
iindex 92f7252..2de9a45 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -21,6 +21,7 @@ function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }
() => ctx.keys.client.stop(),
async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); },
() => clearInterval(ctx.faceRetentionTick),
++ () => { if (ctx.eventOutbox) ctx.eventOutbox.stop(); },
() => ctx.s.close(),
...extra,
],
@@@ -49,6 +50,11 @@ async function start() {
}
ctx.keys.client.start();
++ // The event relay: Vision's own outbox → OpenVibe.Events. The rows wait in event_outbox until the drain
++ // sends them, with backoff after a failure. Off in a drill (a drill publishes nothing) and without the
++ // events URL or the OAuth client secret (the rows wait for the next boot).
++ ctx.eventOutbox.start();
++
// The two account topics (export_requested, deleted) at OpenVibe.Events: idempotent; off without the URL, the
// delivery secret and the client secret. The consumer itself is mounted in server/app.js.
const subs = config.drill ? null : startSubscriptions({
ddiff --git a/test/events-outbox.test.js b/test/events-outbox.test.js
nnew file mode 100644
iindex 0000000..6670002
---- /dev/null
++++ b/test/events-outbox.test.js
@@@ -0,0 +1,154 @@
++'use strict';
++
++/**
++ * Vision's event outbox (server/events/outbox.js): the durable retry queue behind every event Vision publishes.
++ *
++ * a failed publish is kept and sent on the next drain Events is down: the row stays, the next drain sends it
++ * a successful one is removed a sent event leaves the pending queue and is pruned
++ * the drain stays off in a drill VISION_DRILL=1: nothing is published and no row is written
++ *
++ * Delivery is at least once and the event's own id is the idempotency key Events dedupes on.
++ */
++const assert = require('assert');
++const { check, done } = require('./helpers/boot');
++const { testDb } = require('./helpers/db');
++const { createVisionOutbox } = require('../server/events/outbox');
++const { createPublisher } = require('../server/events/publisher');
++
++const USER = { kind: 'user', subject: 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2A4' };
++const RESULT_ID = 'vis_01J8Z3Q4R5S6T7V8W9X0Y1Z2A4';
++
++const quiet = { warn() {}, error() {}, log() {} };
++
++function config(over = {}) {
++ return {
++ drill: false,
++ networkInternalUrl: 'https://network.example',
++ oauth: { clientId: 'vision', clientSecret: 'vision-secret' },
++ events: { url: 'https://events.example', scope: 'events.event.publish', outboxIntervalMs: 50 },
++ ...over,
++ };
++}
++
++/** The token and the publish request; state.fail makes Events answer 503 (Events is down). */
++function eventsFetch(state) {
++ return async (url, options) => {
++ const u = String(url);
++ if (u.endsWith('/oauth/token')) return Response.json({ access_token: 'test-token', expires_in: 300 });
++ if (u.endsWith('/api/v1/events')) {
++ if (state.fail) return Response.json({ error: { code: 'events.unavailable', message: 'down' } }, { status: 503 });
++ const body = JSON.parse(options.body);
++ const headers = options.headers || {};
++ const key = typeof headers.get === 'function' ? headers.get('idempotency-key') : headers['Idempotency-Key'];
++ state.sent.push({ body, key });
++ return Response.json({ event_id: body.event_id, cursor: 'cursor' });
++ }
++ throw new Error('unexpected URL: ' + u);
++ };
++}
++
++const request = (path, body) => ({ path, body, principal: USER, ov: {} });
++const answer = { answer: { id: RESULT_ID, text: 'A red bicycle.', model: 'model', provider: 'ai', route: 'routed', cost_usd: 0.01, latency_ms: 120 } };
++
++(async () => {
++ const testdb = await testDb();
++ const db = testdb.db;
++ try {
++ await check('a failed publish is kept and sent on the next drain', async () => {
++ await db.exec('DELETE FROM event_outbox');
++ const state = { fail: true, sent: [] };
++ const outbox = createVisionOutbox({ s: { db }, config: config(), log: quiet, fetchImpl: eventsFetch(state) });
++ const publisher = createPublisher({ outbox, log: quiet });
++
++ await publisher.result(request('/ask', { image: 'data:image/png;base64,AAAA' }), answer);
++
++ // The event is durable before the answer went out: it waits in the table.
++ const row = await db.maybe('SELECT event_id, envelope, sent_at FROM event_outbox ORDER BY id DESC LIMIT 1');
++ assert.ok(row, 'the event waits in the outbox');
++ assert.strictEqual(row.sent_at, null);
++ assert.strictEqual(row.envelope.event_type, 'vision.result.ready');
++ assert.strictEqual(row.envelope.source, 'vision');
++ assert.ok(row.event_id, 'the event carries its own id');
++
++ // The drain tries and Events is down: the row stays, with the failure recorded.
++ await outbox.outbox.flush();
++ const failed = await db.maybe('SELECT * FROM event_outbox WHERE event_id = $1', [row.event_id]);
++ assert.ok(failed, 'a failed publish is kept');
++ assert.strictEqual(failed.sent_at, null);
++ assert.strictEqual(Number(failed.attempts), 1);
++ assert.ok(failed.last_error, 'the failure is recorded on the row');
++ assert.strictEqual(Number(await outbox.outbox.pending()), 1);
++
++ // The next drain sends it: the same event id is the idempotency key Events dedupes on.
++ state.fail = false;
++ await db.exec('UPDATE event_outbox SET next_attempt_at = 0 WHERE event_id = $1', [row.event_id]);
++ await outbox.outbox.flush();
++ assert.strictEqual(state.sent.length, 1, 'the kept event goes out with the next drain');
++ assert.strictEqual(state.sent[0].body.event_id, row.event_id);
++ assert.strictEqual(state.sent[0].key, row.event_id, 'the idempotency key is the event id');
++ const sent = await db.maybe('SELECT sent_at FROM event_outbox WHERE event_id = $1', [row.event_id]);
++ assert.ok(sent.sent_at, 'the row is marked sent');
++ assert.strictEqual(Number(await outbox.outbox.pending()), 0);
++ await outbox.stop();
++ });
++
++ await check('a successful one is removed', async () => {
++ await db.exec('DELETE FROM event_outbox');
++ const state = { fail: false, sent: [] };
++ const outbox = createVisionOutbox({ s: { db }, config: config(), log: quiet, fetchImpl: eventsFetch(state) });
++ const publisher = createPublisher({ outbox, log: quiet });
++
++ await publisher.result(request('/ask', {}), answer);
++ await outbox.outbox.flush();
++ assert.strictEqual(state.sent.length, 1);
++ assert.strictEqual(Number(await outbox.outbox.pending()), 0, 'a sent event is no longer pending');
++
++ // The relay prunes the sent rows once they are old enough (every 6 hours while it runs).
++ await db.exec('UPDATE event_outbox SET sent_at = sent_at - 10000');
++ await outbox.outbox.prune(0);
++ assert.strictEqual(Number(await db.value('SELECT count(*) FROM event_outbox')), 0, 'a sent event is removed');
++ await outbox.stop();
++ });
++
++ await check('the drain stays off in a drill', async () => {
++ await db.exec('DELETE FROM event_outbox');
++ let calls = 0;
++ const outbox = createVisionOutbox({
++ s: { db }, config: config({ drill: true }), log: quiet,
++ fetchImpl: async () => { calls++; return Response.json({}); },
++ });
++ const publisher = createPublisher({ outbox, log: quiet });
++
++ await publisher.result(request('/ask', {}), answer);
++ outbox.start();
++ await outbox.outbox.flush();
++ await outbox.stop();
++
++ assert.strictEqual(calls, 0, 'a drill publishes nothing');
++ assert.strictEqual(Number(await db.value('SELECT count(*) FROM event_outbox')), 0, 'a drill writes no row');
++ assert.strictEqual(outbox.enabled, false);
++ });
++
++ await check('without Events configured nothing is published and no row is written', async () => {
++ await db.exec('DELETE FROM event_outbox');
++ let calls = 0;
++ const outbox = createVisionOutbox({
++ s: { db }, config: config({ events: { url: '', scope: 'events.event.publish', outboxIntervalMs: 50 } }), log: quiet,
++ fetchImpl: async () => { calls++; return Response.json({}); },
++ });
++ const publisher = createPublisher({ outbox, log: quiet });
++
++ await publisher.result(request('/ask', {}), answer);
++ outbox.start();
++ await outbox.outbox.flush();
++ await outbox.stop();
++
++ assert.strictEqual(calls, 0, 'nothing is published');
++ assert.strictEqual(Number(await db.value('SELECT count(*) FROM event_outbox')), 0, 'no row is written');
++ assert.strictEqual(outbox.enabled, false);
++ });
++ } finally {
++ await testdb.close();
++ }
++ done();
++})();
ddiff --git a/test/events-publisher.test.js b/test/events-publisher.test.js
iindex 7373599..4edf87b 100644
---- a/test/events-publisher.test.js
++++ b/test/events-publisher.test.js
@@@ -1,76 +1,93 @@
'use strict';
++/**
++ * The events Vision publishes (server/events/publisher.js): the envelope lands in Vision's own outbox
++ * (server/events/outbox.js) and the relay sends it from there.
++ */
const assert = require('assert');
const { boot, check, done } = require('./helpers/boot');
++const { testDb } = require('./helpers/db');
++const { createVisionOutbox } = require('../server/events/outbox');
const { createPublisher } = require('../server/events/publisher');
++const principal = { kind: 'user', subject: 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2A4' };
++const req = (path, body) => ({ path, body, principal, ov: {} });
++const quiet = { warn() {}, error() {}, log() {} };
++
(async () => {
-- const sent = [];
++ const testdb = await testDb();
++ const db = testdb.db;
const config = {
drill: false, networkInternalUrl: 'https://network.example',
oauth: { clientId: 'vision', clientSecret: 'test-secret' },
-- events: { url: 'https://events.example', scope: 'events.event.publish' },
++ events: { url: 'https://events.example', scope: 'events.event.publish', outboxIntervalMs: 50 },
};
-- const fetchImpl = async (url, options) => {
-- if (String(url).endsWith('/oauth/token')) return Response.json({ access_token: 'test-token', expires_in: 300 });
-- if (String(url).endsWith('/api/v1/events')) {
-- sent.push(JSON.parse(options.body));
-- return Response.json({ event_id: sent.at(-1).event_id, cursor: 'cursor' });
-- }
-- throw new Error('unexpected URL');
++ // The relay is off in this test: the envelope is only written, never sent.
++ const outbox = createVisionOutbox({ s: { db }, config, log: quiet, fetchImpl: async () => { throw new Error('the relay must not send here'); } });
++ const publisher = createPublisher({ outbox, log: quiet });
++ const lastEvent = async () => {
++ const row = await db.maybe('SELECT envelope FROM event_outbox ORDER BY id DESC LIMIT 1');
++ return row ? row.envelope : null;
};
-- const publisher = createPublisher({ config, fetchImpl, log: { warn() { throw new Error('event publish failed'); } } });
-- const principal = { kind: 'user', subject: 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2A4' };
-- const req = (path, body) => ({ path, body, principal, ov: {} });
-- await check('ready event carries provenance but no answer text or image', async () => {
-- await publisher.result(req('/ask', { image: 'data:image/png;base64,AAAA' }), {
-- answer: { id: 'vis_01J8Z3Q4R5S6T7V8W9X0Y1Z2A4', text: 'A red bicycle.', model: 'model', provider: 'ai', route: 'routed', cost_usd: 0.01, latency_ms: 120 },
++ try {
++ await check('ready event carries provenance but no answer text or image', async () => {
++ await publisher.result(req('/ask', { image: 'data:image/png;base64,AAAA' }), {
++ answer: { id: 'vis_01J8Z3Q4R5S6T7V8W9X0Y1Z2A4', text: 'A red bicycle.', model: 'model', provider: 'ai', route: 'routed', cost_usd: 0.01, latency_ms: 120 },
++ });
++ const env = await lastEvent();
++ assert.ok(env, 'the event waits in the outbox');
++ assert.strictEqual(env.event_type, 'vision.result.ready');
++ assert.deepStrictEqual(env.subject, { type: 'user', id: principal.subject });
++ assert.strictEqual(env.payload.task, 'ask');
++ assert.strictEqual(env.payload.source, 'data');
++ assert.ok(!JSON.stringify(env).includes('red bicycle'));
++ assert.ok(!JSON.stringify(env).includes('AAAA'));
});
-- assert.strictEqual(sent.length, 1);
-- assert.strictEqual(sent[0].event_type, 'vision.result.ready');
-- assert.deepStrictEqual(sent[0].subject, { type: 'user', id: principal.subject });
-- assert.strictEqual(sent[0].payload.task, 'ask');
-- assert.strictEqual(sent[0].payload.source, 'data');
-- assert.ok(!JSON.stringify(sent[0]).includes('red bicycle'));
-- assert.ok(!JSON.stringify(sent[0]).includes('AAAA'));
-- });
-- await check('watch event names only the owner-granted source and detected classes', async () => {
-- await publisher.alert(req('/watch', { source_ref: 'cam:front-door' }), { answer: { detections: [{ class: 'animal', confidence: 0.92 }] } });
-- assert.strictEqual(sent[1].event_type, 'vision.alert');
-- assert.strictEqual(sent[1].payload.source_ref, 'cam:front-door');
-- assert.deepStrictEqual(sent[1].payload.classes, ['animal']);
-- await publisher.alert(req('/watch', {}), { answer: { detections: [{ class: 'animal', confidence: 0.92 }] } });
-- assert.strictEqual(sent.length, 2);
-- });
++ await check('watch event names only the owner-granted source and detected classes', async () => {
++ await publisher.alert(req('/watch', { source_ref: 'cam:front-door' }), { answer: { detections: [{ class: 'animal', confidence: 0.92 }] } });
++ const env = await lastEvent();
++ assert.strictEqual(env.event_type, 'vision.alert');
++ assert.strictEqual(env.payload.source_ref, 'cam:front-door');
++ assert.deepStrictEqual(env.payload.classes, ['animal']);
++ await publisher.alert(req('/watch', {}), { answer: { detections: [{ class: 'animal', confidence: 0.92 }] } });
++ const after = await lastEvent();
++ assert.strictEqual(after.payload.source_ref, 'cam:front-door', 'a request without a source ref publishes nothing');
++ });
-- await check('anonymous results create no subject event', async () => {
-- await publisher.result({ ...req('/ocr', {}), principal: { kind: 'anonymous' } }, { answer: { text: 'x' } });
-- assert.strictEqual(sent.length, 2);
-- });
++ await check('anonymous results create no subject event', async () => {
++ const before = Number(await db.value('SELECT count(*) FROM event_outbox'));
++ await publisher.result({ ...req('/ocr', {}), principal: { kind: 'anonymous' } }, { answer: { text: 'x' } });
++ assert.strictEqual(Number(await db.value('SELECT count(*) FROM event_outbox')), before);
++ });
-- await check('successful API results call the publisher with the authenticated owner', async () => {
-- const calls = [];
-- const t = await boot({ eventPublisher: {
-- result: async (request, out) => calls.push({ path: request.path, subject: request.principal.subject, id: out.check && out.check.id }),
-- alert: async () => {},
-- } });
-- try {
-- const user = t.network.addUser('event-owner');
-- const r = await t.get('/api/v1/ui-check', { as: user, json: { html: '<p>hello</p>' } });
-- assert.strictEqual(r.status, 201, r.text);
-- assert.deepStrictEqual(calls, [{ path: '/ui-check', subject: user.subject, id: r.json().check.id }]);
-- const mcp = await t.get('/mcp', { bearer: t.network.userToken(user), json: {
-- jsonrpc: '2.0', id: 1, method: 'tools/call',
-- params: { name: 'vision.check_ui', arguments: { html: '<p>hello</p>' } },
++ await check('successful API results call the publisher with the authenticated owner', async () => {
++ const calls = [];
++ const t = await boot({ eventPublisher: {
++ result: async (request, out) => calls.push({ path: request.path, subject: request.principal.subject, id: out.check && out.check.id }),
++ alert: async () => {},
} });
-- assert.strictEqual(mcp.status, 200, mcp.text);
-- assert.strictEqual(mcp.json().result.isError, false, mcp.text);
-- assert.strictEqual(calls.length, 2);
-- assert.strictEqual(calls[1].path, '/ui-check');
-- assert.strictEqual(calls[1].subject, user.subject);
-- } finally { await t.close(); }
-- });
++ try {
++ // The drill mode of the test boot keeps the relay off.
++ assert.strictEqual(t.ctx.eventOutbox.enabled, false, 'the drain is off in a drill');
++ const user = t.network.addUser('event-owner');
++ const r = await t.get('/api/v1/ui-check', { as: user, json: { html: '<p>hello</p>' } });
++ assert.strictEqual(r.status, 201, r.text);
++ assert.deepStrictEqual(calls, [{ path: '/ui-check', subject: user.subject, id: r.json().check.id }]);
++ const mcp = await t.get('/mcp', { bearer: t.network.userToken(user), json: {
++ jsonrpc: '2.0', id: 1, method: 'tools/call',
++ params: { name: 'vision.check_ui', arguments: { html: '<p>hello</p>' } },
++ } });
++ assert.strictEqual(mcp.status, 200, mcp.text);
++ assert.strictEqual(mcp.json().result.isError, false, mcp.text);
++ assert.strictEqual(calls.length, 2);
++ assert.strictEqual(calls[1].path, '/ui-check');
++ assert.strictEqual(calls[1].subject, user.subject);
++ } finally { await t.close(); }
++ });
++ } finally {
++ await testdb.close();
++ }
done();
})();