Outbox: openvibe-sdk/events instead of the hand-rolled copy (#29)

ddiff --git a/README.md b/README.md
iindex cfc2538..885153c 100644
---- a/README.md
++++ b/README.md
@@@ -155,6 +155,9 @@ scheduler, fetches, the outbox relay) run in-process and never pass through HTTP
  - `sources.index_document.upserted|deleted` for sources with `search_visibility: "members"`: raw items indexed in OpenVibe.Search for staff only (`acl.groups: role:admin, role:global_mod`), always `noindex` (`third_party_content`).
  
  All with `visibility: "internal"`. Envelopes validate as `events.event-envelope@1`.
++The openvibe-sdk service outbox stores pending events in `service_outbox` and relays them with the
++Sources service token. The expand migration copies pending legacy rows; `event_outbox` remains
++for the N-1 rollback window.
  
  ## Seeds
  
ddiff --git a/STATUS.json b/STATUS.json
iindex 0cc3748..b158a08 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -40,7 +40,7 @@
      "failed fetches recorded, never fabricated",
      "item revisions and sticky removals with provenance",
      "cursor item API by capability",
--    "transactional outbox to Events",
++    "openvibe-sdk transactional outbox to Events (service_outbox)",
      "staff-only index documents for Search",
      "per-actor limits (openvibe-sdk/limits)"
    ],
ddiff --git a/migrations/0002_sdk_outbox.sql b/migrations/0002_sdk_outbox.sql
nnew file mode 100644
iindex 0000000..0a0a72e
---- /dev/null
++++ b/migrations/0002_sdk_outbox.sql
@@@ -0,0 +1,24 @@
++-- phase: expand
++-- openvibe-sdk/events outboxSchema('service_outbox'); keep event_outbox through the N-1 rollback window.
++CREATE TABLE IF NOT EXISTS service_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 service_outbox_due ON service_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
++CREATE INDEX IF NOT EXISTS service_outbox_sent ON service_outbox (sent_at) WHERE sent_at IS NOT NULL;
++
++INSERT INTO service_outbox (event_id, envelope, created_at, attempts, next_attempt_at)
++SELECT event_id, envelope::jsonb, created_at, attempts, next_attempt_at
++FROM event_outbox WHERE sent_at IS NULL AND rejected_at IS NULL
++ON CONFLICT (event_id) DO NOTHING;
++
++-- A later contract migration drops event_outbox after the N-1 window.
ddiff --git a/server/api/items.js b/server/api/items.js
iindex 8dc5e58..8e687fa 100644
---- a/server/api/items.js
++++ b/server/api/items.js
@@@ -47,7 +47,7 @@ function manualItem(body) {
      };
  }
  
--function itemsRouter({ db, registry, items, auth, relay, now, limits }) {
++function itemsRouter({ db, registry, items, auth, outbox, now, limits }) {
      const router = express.Router();
      // Each guard, then the principal's per-actor limit (api/actor-limits.js), before any work.
      const read = [auth.requireCap(CAPS.items), limits.reads('sources.read')];
@@@ -92,7 +92,7 @@ function itemsRouter({ db, registry, items, auth, relay, now, limits }) {
          const t = now();
          const counts = await db.tx(async () => await items.ingest(source, [m.item], { runId: null, rawBodyHash: null, parserVersion: MANUAL_VERSION, retrievedAt: t, enteredBy: req.principal.sub }));
          await db.prepare('UPDATE sources SET last_success_at = ?, last_run_at = ? WHERE key = ?').run(t, t, source.key);
--        if (relay) relay.flush().catch(() => {});
++        if (outbox) outbox.kick().catch(() => {});
          const saved = await db.prepare('SELECT id FROM items WHERE source_key = ? AND identity = ?').get(source.key, m.item.identity);
          const outcome = counts.created ? 'created' : counts.updated ? 'updated' : counts.skipped ? 'removed' : 'unchanged';
          res.status(counts.created ? 201 : 200).json({ outcome, item: await items.get(saved.id) });
@@@ -105,7 +105,7 @@ function itemsRouter({ db, registry, items, auth, relay, now, limits }) {
          if (!found) return http.sendProblem(res, 404, 'sources.not_found', { detail: 'no such item', ctx: req.ov });
          const source = registry.fromRow(await registry.get(found.source_key));
          const item = await items.remove(source, found.id, reason, req.principal.sub);
--        if (relay) relay.flush().catch(() => {});
++        if (outbox) outbox.kick().catch(() => {});
          res.json({ item });
      });
  
ddiff --git a/server/app.js b/server/app.js
iindex 212219d..63ce5d2 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -63,7 +63,7 @@ Source code: <a href="https://github.com/OpenVibers/OpenVibe.Sources">OpenVibers
  </html>
  `;
  
--function createApp({ config, db, registry, items, ingest, scheduler, auth, outbox, relay, now, log = console, limitsNow = null }) {
++function createApp({ config, db, registry, items, ingest, scheduler, auth, outbox, now, log = console, limitsNow = null }) {
      const app = express();
      app.disable('x-powered-by');
      app.set('trust proxy', 'loopback');
@@@ -88,7 +88,7 @@ function createApp({ config, db, registry, items, ingest, scheduler, auth, outbo
  
      // Readiness (openvibe-shared/ready): 503 only when the database fails; the Network key and the
      // fetcher (worker running, queue keeping up) are optional and degrade it (see observability.js).
--    const readiness = createSourcesReadiness({ db, config, registry, ingest, scheduler, outbox, relay, now, release: release.release });
++    const readiness = createSourcesReadiness({ db, config, registry, ingest, scheduler, outbox, now, release: release.release });
      app.get('/api/ready', readiness.handler);
      // GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).
      release.mount(app, { registry: metrics.registry });
@@@ -97,7 +97,7 @@ function createApp({ config, db, registry, items, ingest, scheduler, auth, outbo
      // passed. limitsNow: the limiter's clock (tests; default the wall clock, not the ingest clock).
      const limits = createActorLimits({ config, now: limitsNow || (() => Date.now()), registry: metrics.registry, log });
      app.use(sourcesRouter({ db, registry, ingest, auth, limits }));
--    app.use(itemsRouter({ db, registry, items, auth, relay, now, limits }));
++    app.use(itemsRouter({ db, registry, items, auth, outbox, now, limits }));
  
      // The public host (sources.openvibe.network) shows only this page, health, readiness and
      // /release.json: Sources is internal. Browsers get a short honest HTML page, other clients the
ddiff --git a/server/events/envelope.js b/server/events/envelope.js
nnew file mode 100644
iindex 0000000..4365489
---- /dev/null
++++ b/server/events/envelope.js
@@@ -0,0 +1,16 @@
++'use strict';
++/** Preserve Sources' event-envelope defaults while the SDK owns storage and delivery. */
++const { ids } = require('openvibe-contracts');
++
++function createEnvelope({ event_type, subject, payload, visibility = 'internal', priority = 'important', actor, trace_id }, { source, now }) {
++    const ms = now();
++    const env = {
++        event_id: ids.newId('event', ms), event_type, version: 1, source,
++        actor: actor || { type: 'service', id: source }, timestamp: new Date(ms).toISOString(),
++        priority, visibility, subject, payload: payload || {},
++    };
++    if (trace_id && /^[0-9a-f]{32}$/.test(trace_id)) env.trace_id = trace_id;
++    return env;
++}
++
++module.exports = { createEnvelope };
ddiff --git a/server/events/outbox.js b/server/events/outbox.js
ddeleted file mode 100644
iindex 26729a8..0000000
---- a/server/events/outbox.js
++++ /dev/null
@@@ -1,163 +0,0 @@
--'use strict';
--/**
-- * Transactional outbox (events.event-envelope@1), OpenVibe.Events client semantics:
-- *
-- *   enqueue(envelope)  INSIDE the transaction that makes the change, so an event exists exactly
-- *                      when its effect does (throws outside a transaction)
-- *   relay              publishes unsent rows to OpenVibe.Events (POST /api/v1/events, service
-- *                      token for audience openvibe.events with events.event.publish), marks them
-- *                      sent, backs off on failure; a permanently refused row (4xx other than
-- *                      401/408/429) is marked rejected so it cannot block the rest
-- *
-- * The relay runs only when EVENTS_URL is set; without it rows wait and are relayed later. Events
-- * deduplicates on event_id, so a retry after a lost response never publishes twice.
-- */
--const { ids, validate, serviceAuth } = require('openvibe-contracts');
--
--const BACKOFF_MS = [1000, 5000, 30000, 120000, 600000];
--
--function createOutbox(db, { source, now = () => Date.now() }) {
--    const insert = db.prepare('INSERT INTO event_outbox (event_id, event_type, envelope, created_at) VALUES (?, ?, ?, ?)');
--
--    /** { event_type, subject, payload, visibility?, priority?, actor?, trace_id? } → envelope */
--    async function enqueue({ event_type, subject, payload, visibility = 'internal', priority = 'important', actor, trace_id }) {
--        if (!db.inTransaction()) throw new Error('outbox.enqueue() must run inside the transaction that makes the change');
--        const ms = now();
--        const env = {
--            event_id: ids.newId('event', ms),
--            event_type,
--            version: 1,
--            source,
--            actor: actor || { type: 'service', id: source },
--            timestamp: new Date(ms).toISOString(),
--            priority,
--            visibility,
--            subject,
--            payload: payload || {},
--        };
--        if (trace_id && /^[0-9a-f]{32}$/.test(trace_id)) env.trace_id = trace_id;
--        const v = validate('events.event-envelope@1', env);
--        if (!v.valid) throw new Error(`outbox: invalid envelope for ${event_type}: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`);
--        await insert.run(env.event_id, event_type, JSON.stringify(env), ms);
--        return env;
--    }
--
--    const q = {
--        due: db.prepare('SELECT seq, event_id, envelope, attempts FROM event_outbox WHERE sent_at IS NULL AND rejected_at IS NULL AND next_attempt_at <= ? ORDER BY seq LIMIT ?'),
--        sent: db.prepare('UPDATE event_outbox SET sent_at = ?, attempts = attempts + 1, last_error = NULL WHERE seq = ?'),
--        failed: db.prepare('UPDATE event_outbox SET attempts = attempts + 1, next_attempt_at = ?, last_error = ? WHERE seq = ?'),
--        rejected: db.prepare('UPDATE event_outbox SET rejected_at = ?, attempts = attempts + 1, last_error = ? WHERE seq = ?'),
--        pending: db.prepare('SELECT COUNT(*) AS n FROM event_outbox WHERE sent_at IS NULL AND rejected_at IS NULL'),
--        rejectedCount: db.prepare('SELECT COUNT(*) AS n FROM event_outbox WHERE rejected_at IS NOT NULL'),
--        list: db.prepare('SELECT envelope FROM event_outbox ORDER BY seq'),
--        prune: db.prepare('DELETE FROM event_outbox WHERE sent_at IS NOT NULL AND sent_at < ?'),
--    };
--
--    return {
--        enqueue,
--        pending: async () => (await q.pending.get()).n,
--        rejected: async () => (await q.rejectedCount.get()).n,
--        /** Every envelope in order (tests and operators). */
--        all: async () => (await q.list.all()).map(r => JSON.parse(r.envelope)),
--        prune: async (olderThanMs = 7 * 24 * 3600 * 1000) => (await q.prune.run(now() - olderThanMs)).changes,
--        _q: q,
--    };
--}
--
--/**
-- * createRelay({ db, outbox, eventsUrl, intervalMs?, tokenClient? | tokenOpts?, fetchImpl?, log?, now? })
-- *   → { start, stop, flush, running }
-- * tokenOpts = { tokenUrl, clientId, clientSecret } builds a client-credentials token client for
-- * audience openvibe.events. Without eventsUrl or a token source the relay does nothing.
-- */
--function createRelay({ outbox, eventsUrl, intervalMs = 2000, tokenClient, tokenOpts, fetchImpl = globalThis.fetch, log = console, now = () => Date.now(), db }) {
--    const q = outbox._q;
--    const tokens = tokenClient || (tokenOpts && serviceAuth.createTokenClient({ ...tokenOpts, audience: 'openvibe.events', scope: 'events.event.publish', fetchImpl }));
--    let timer = null;
--    let flushing = null;
--
--    function permanent(status) {
--        return status >= 400 && status < 500 && ![401, 408, 429].includes(status);
--    }
--
--    async function post(rows) {
--        const body = rows.length === 1 ? JSON.parse(rows[0].envelope) : { events: rows.map(r => JSON.parse(r.envelope)) };
--        const res = await fetchImpl(`${eventsUrl}/api/v1/events`, {
--            method: 'POST',
--            headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...(await tokens.authHeaders()) },
--            body: JSON.stringify(body),
--            signal: AbortSignal.timeout(10000),
--        });
--        if (res.status === 401 && typeof tokens.invalidate === 'function') tokens.invalidate();
--        const text = await res.text().catch(() => '');
--        return { ok: res.ok, status: res.status, text };
--    }
--
--    async function mark(rows, fn) { await db.tx(async () => { for (const r of rows) await fn(r); }); }
--
--    async function publishRows(rows) {
--        let r;
--        try {
--            r = await post(rows);
--        } catch (err) {
--            await mark(rows, async row => await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], String(err.message).slice(0, 500), row.seq));
--            return { sent: 0, failed: rows.length, rejected: 0 };
--        }
--        if (r.ok) {
--            await mark(rows, async row => await q.sent.run(now(), row.seq));
--            return { sent: rows.length, failed: 0, rejected: 0 };
--        }
--        if (permanent(r.status) && rows.length > 1) {
--            // One bad envelope refuses the batch: isolate it.
--            const total = { sent: 0, failed: 0, rejected: 0 };
--            for (const row of rows) {
--                const s = await publishRows([row]);
--                total.sent += s.sent; total.failed += s.failed; total.rejected += s.rejected;
--            }
--            return total;
--        }
--        const msg = `${r.status} ${r.text.slice(0, 300)}`;
--        if (permanent(r.status)) {
--            await mark(rows, async row => await q.rejected.run(now(), msg, row.seq));
--            log.warn(`[outbox] event refused by Events: ${msg}`);
--            return { sent: 0, failed: 0, rejected: rows.length };
--        }
--        await mark(rows, async row => await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], msg, row.seq));
--        return { sent: 0, failed: rows.length, rejected: 0 };
--    }
--
--    async function doFlush() {
--        const total = { sent: 0, failed: 0, rejected: 0 };
--        if (!eventsUrl || !tokens) return total;
--        for (;;) {
--            const rows = await q.due.all(now(), 50);
--            if (!rows.length) break;
--            const s = await publishRows(rows);
--            total.sent += s.sent; total.failed += s.failed; total.rejected += s.rejected;
--            if (s.failed || rows.length < 50) break;
--        }
--        if (total.failed) log.warn(`[outbox] ${total.failed} event(s) not published yet; will retry`);
--        return total;
--    }
--
--    function flush() {
--        if (!flushing) flushing = doFlush().finally(() => { flushing = null; });
--        return flushing;
--    }
--
--    function start() {
--        if (timer || !eventsUrl) return;
--        timer = setInterval(() => { flush().catch(err => log.warn(`[outbox] relay: ${err.message}`)); }, intervalMs);
--        timer.unref?.();
--    }
--
--    function stop() {
--        if (timer) clearInterval(timer);
--        timer = null;
--        return flushing || Promise.resolve();
--    }
--
--    return { start, stop, flush, running: () => Boolean(timer) };
--}
--
--module.exports = { createOutbox, createRelay };
ddiff --git a/server/index.js b/server/index.js
iindex 8910cfa..a90a16d 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -17,24 +17,31 @@ const { createSpacer } = require('./spacer');
  const { createRobots } = require('./robots');
  const { createIngest } = require('./ingest');
  const { createScheduler } = require('./scheduler');
--const { createOutbox, createRelay } = require('./events/outbox');
++const { createServiceOutbox } = require('openvibe-sdk/events');
++const { validate } = require('openvibe-contracts');
++const { createEnvelope } = require('./events/envelope');
++const manifest = require('../docs/service-manifest-proposal.json');
  const { jwksClient } = require('openvibe-sdk/auth');
  const { gracefulStop } = require('openvibe-sdk/service');
  const { createAuth } = require('./auth');
  const { createApp } = require('./app');
  
--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, tokenClient, lookupImpl, log = console, listen = true, limitsNow = null } = {}) {
++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = 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 });
--    const outbox = createOutbox(db, { source: config.serviceId, now });
--    const relay = createRelay({
--        db, outbox, eventsUrl: config.events.url, intervalMs: config.events.relayIntervalMs, fetchImpl, log, now,
--        tokenClient,
--        tokenOpts: config.oauth.clientSecret ? { tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret } : null,
++    const outbox = createServiceOutbox({
++        db, source: config.serviceId, eventsUrl: config.events.url, networkInternalUrl: config.networkInternalUrl,
++        clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
++        intervalMs: config.events.relayIntervalMs, log, table: 'service_outbox', eventTypes: manifest.eventsProduced,
++        validate: (env) => validate('events.event-envelope@1', env), now, fetch: fetchImpl,
      });
++    const emitEvent = (args) => {
++        if (!db.inTransaction()) throw new Error('event emission must run inside the transaction that makes the change');
++        return outbox.emit(createEnvelope(args, { source: config.serviceId, now }));
++    };
      const registry = createRegistry({ db, now, maxItemsCap: config.items.maxPerFetch });
--    const items = createItems({ db, outbox, now });
++    const items = createItems({ db, emitEvent, now });
      const guard = createGuard({ allowPrivateHosts: config.fetch.allowPrivateHosts, allowedPorts: config.fetch.allowedPorts, ...(lookupImpl ? { lookupImpl } : {}) });
      const fetcher = createFetcher({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.fetch.timeoutMs, maxBytes: config.fetch.maxBytes, maxRedirects: config.fetch.maxRedirects });
      const spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });
@@@ -42,18 +49,18 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
          db, fetcher, agent: config.fetch.robotsAgent, ttlMs: config.fetch.robotsTtlMs, maxBytes: config.fetch.robotsMaxBytes, now,
          beforeRequest: (host) => spacer.space(host, 0),
      });
--    const ingest = createIngest({ db, config, registry, items, robots, fetcher, spacer, outbox, now, log, relay });
++    const ingest = createIngest({ db, config, registry, items, robots, fetcher, spacer, emitEvent, now, log, outbox });
      const scheduler = createScheduler({ db, ingest, config, now, log });
      const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
      const auth = createAuth({ config, log });
--    const app = createApp({ config, db, registry, items, ingest, scheduler, auth, outbox, relay, now, log, limitsNow });
++    const app = createApp({ config, db, registry, items, ingest, scheduler, auth, outbox, now, log, limitsNow });
      // 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();
      const keyLoaded = keys.keys().catch(() => null);
--    relay.start();
++    outbox.start();
      if (config.worker.enabled) scheduler.start();
--    const pruneTimer = setInterval(async () => { try { await outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);
++    const pruneTimer = setInterval(async () => { try { await outbox.outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);
      pruneTimer.unref?.();
  
      let server = null;
@@@ -69,7 +76,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
          clearInterval(pruneTimer);
          keys.stop();
          await scheduler.stop();
--        await relay.stop();
++        await outbox.stop();
          if (server) {
              server.closeAllConnections?.();
              await new Promise(resolve => server.close(() => resolve()));
@@@ -77,7 +84,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
          if (!givenDb) await db.close();
      }
  
--    return { config, db, registry, items, guard, fetcher, robots, ingest, scheduler, outbox, relay, keys, keyLoaded, auth, app, server, close };
++    return { config, db, registry, items, guard, fetcher, robots, ingest, scheduler, outbox, keys, keyLoaded, auth, app, server, close };
  }
  
  if (require.main === module) {
ddiff --git a/server/ingest.js b/server/ingest.js
iindex 07b2165..0d8447e 100644
---- a/server/ingest.js
++++ b/server/ingest.js
@@@ -31,7 +31,7 @@ function retryAfterMs(value, now) {
      return Number.isFinite(t) ? Math.max(0, t - now) : null;
  }
  
--function createIngest({ db, config, registry, items, robots, fetcher, spacer, outbox, now = () => Date.now(), log = console, relay = null }) {
++function createIngest({ db, config, registry, items, robots, fetcher, spacer, emitEvent, now = () => Date.now(), log = console, outbox = null }) {
      const inflight = new Set();
      const { space, setCrawlDelay } = spacer;
  
@@@ -73,7 +73,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
      }
  
      async function failedEvent(source, run, failures) {
--        await outbox.enqueue({
++        await emitEvent({
              event_type: 'sources.fetch.failed',
              subject: { type: 'source', id: source.key },
              payload: {
@@@ -266,7 +266,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
                  key, at: finishedAt, state: lastState, failures: newFailures,
                  success_at: anySuccess ? finishedAt : null, next_due: finishedAt + backoff, not_before: notBefore,
              });
--            if (relay) relay.flush().catch(() => {});
++            if (outbox) outbox.kick().catch(() => {});
              return { runs };
          } finally {
              inflight.delete(key);
ddiff --git a/server/items.js b/server/items.js
iindex a1c2c25..8fd9e91 100644
---- a/server/items.js
++++ b/server/items.js
@@@ -25,7 +25,7 @@ function contentHash(it) {
      }));
  }
  
--function createItems({ db, outbox, now = () => Date.now() }) {
++function createItems({ db, emitEvent, now = () => Date.now() }) {
      const st = {
          byIdentity: db.prepare('SELECT * FROM items WHERE source_key = ? AND identity = ?'),
          byId: db.prepare('SELECT * FROM items WHERE id = ?'),
@@@ -85,7 +85,7 @@ function createItems({ db, outbox, now = () => Date.now() }) {
      }
  
      async function itemEvent(type, row, extra = {}) {
--        await outbox.enqueue({
++        await emitEvent({
              event_type: type,
              subject: subjectOf(row),
              payload: {
@@@ -123,7 +123,7 @@ function createItems({ db, outbox, now = () => Date.now() }) {
  
      async function indexEvent(source, row, deleted = false) {
          if (!source.search_visibility) return;
--        await outbox.enqueue({
++        await emitEvent({
              event_type: deleted ? 'sources.index_document.deleted' : 'sources.index_document.upserted',
              subject: subjectOf(row),
              payload: deleted ? { type: 'item', id: row.id, revision: row.revision } : indexDocument(row),
ddiff --git a/server/observability.js b/server/observability.js
iindex 5a54caa..374fabd 100644
---- a/server/observability.js
++++ b/server/observability.js
@@@ -40,7 +40,7 @@ function readers(db, { startedAt }) {
      };
  }
  
--function createSourcesReadiness({ db, config, registry, ingest, scheduler, outbox, relay, now, release = null }) {
++function createSourcesReadiness({ db, config, registry, ingest, scheduler, outbox, now, release = null }) {
      const read = readers(db, { startedAt: now() });
      const iso = (v) => (v == null ? null : new Date(v).toISOString());
      return createReadiness({
@@@ -88,7 +88,7 @@ function createSourcesReadiness({ db, config, registry, ingest, scheduler, outbo
              return {
                  sources,
                  runs_in_flight: ingest.inflight().length,
--                outbox: dbOk ? { pending: await outbox.pending(), rejected: await outbox.rejected(), relay: config.events.url ? (relay.running() ? 'running' : 'stopped') : 'off (EVENTS_URL unset)' } : null,
++                outbox: dbOk ? await outbox.status() : null,
              };
          },
      });
@@@ -116,7 +116,7 @@ function registerSourcesGauges(registry, { db, sources, ingest, outbox, now }) {
      // Left out until there is a fetch: a timestamp of 0 would read as "1970", not "never".
      registry.gauge({ name: 'sources_last_fetch_timestamp_seconds', help: 'When the last fetch run finished (any outcome), Unix seconds', collect: async () => seconds(await read.lastFetchAt()) });
      registry.gauge({ name: 'sources_last_success_timestamp_seconds', help: 'When a source was last fetched successfully, Unix seconds', collect: async () => seconds(await read.lastSuccessAt()) });
--    registry.gauge({ name: 'sources_outbox_pending', help: 'Events waiting in the outbox', collect: async () => await outbox.pending() });
++    registry.gauge({ name: 'sources_outbox_pending', help: 'Events waiting in the outbox', collect: async () => (await outbox.status()).pending });
  }
  
  module.exports = { createSourcesReadiness, registerSourcesGauges, OVERDUE_MS };
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 26bdaee..2dda1f0 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -34,7 +34,7 @@ function serviceToken(slug, cap, { aud = 'openvibe.sources', exp = Math.floor(Da
   * Boot Sources; the scheduler is off unless worker: 'on'. Loopback is allowlisted for the stubs.
   * log: Sources' logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
   */
--async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now, log = silent, limitsNow = null } = {}) {
++async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = 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] }) }) });
@@@ -51,7 +51,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now, lo
      });
      // One database per boot (PGlite, or SOURCES_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, tokenClient, limitsNow, ...(now ? { now } : {}) });
++    const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, ...(now ? { now } : {}) });
      const base = `http://127.0.0.1:${h.server.address().port}`;
      return { ...h, base, jwksSite, async stop() { await h.close(); await testdb.close(); await jwksSite.close(); } };
  }
ddiff --git a/test/ingest.test.js b/test/ingest.test.js
iindex 7f82d60..c59f2e6 100644
---- a/test/ingest.test.js
++++ b/test/ingest.test.js
@@@ -25,7 +25,8 @@ async function create(def) {
  const run = async (key, trigger = 'manual') => await svc.ingest.run(key, { trigger });
  const itemsOf = async (key) => await svc.db.prepare('SELECT * FROM items WHERE source_key = ? ORDER BY identity').all(key);
  const runsOf = async (key) => await svc.db.prepare('SELECT * FROM fetch_runs WHERE source_key = ? ORDER BY rid').all(key);
--const events = async (type) => (await svc.outbox.all()).filter(e => e.event_type === type);
++const allEvents = async (service = svc) => (await service.db.prepare('SELECT envelope FROM service_outbox ORDER BY id').all()).map(r => r.envelope);
++const events = async (type) => (await allEvents()).filter(e => e.event_type === type);
  const reset = async (key) => await svc.db.prepare('UPDATE sources SET last_request_at = NULL, not_before = 0 WHERE key = ?').run(key);
  
  t('boot', async () => {
@@@ -271,7 +272,7 @@ t('a credential is read by name, sent only to its origin, and never stored', asy
          assert.strictEqual(s.hits('/api')[0].headers['x-api-key'], secret);
          assert.strictEqual(other.hits('/landing')[0].headers['x-api-key'], undefined, 'not forwarded across origins');
          const dump = JSON.stringify(await svc2.db.prepare('SELECT * FROM sources').all()) + JSON.stringify(await svc2.db.prepare('SELECT * FROM fetch_runs').all())
--            + JSON.stringify(await svc2.db.prepare('SELECT * FROM items').all()) + JSON.stringify(await svc2.outbox.all());
++            + JSON.stringify(await svc2.db.prepare('SELECT * FROM items').all()) + JSON.stringify(await allEvents(svc2));
          assert.ok(!dump.includes(secret), 'the secret value is nowhere in the database or events');
          assert.ok(dump.includes('SOURCES_CRED_EXAMPLE'), 'the variable name is recorded');
      } finally { await svc2.stop(); await s.close(); await other.close(); }
ddiff --git a/test/observability.test.js b/test/observability.test.js
iindex 57a1125..7cd1a9d 100644
---- a/test/observability.test.js
++++ b/test/observability.test.js
@@@ -48,6 +48,7 @@ t('/api/ready: db is the only required check; a worker that is off degrades it (
      assert.match(r.body.checks.fetcher.error, /SOURCES_WORKER=off/);
      assert.ok(Date.parse(r.body.checks.fetcher.detail.last_fetch_at));
      assert.deepStrictEqual(r.body.sources, { healthy: 1 });
++    assert.deepStrictEqual(r.body.outbox, { enabled: false, pending: 3, rejected: 0, last_error: null });
  });
  
  t('/metrics: 404 through a proxy; route templates and Sources gauges direct', async () => {
ddiff --git a/test/robots-ssrf.test.js b/test/robots-ssrf.test.js
iindex 6a7f281..52f601f 100644
---- a/test/robots-ssrf.test.js
++++ b/test/robots-ssrf.test.js
@@@ -56,7 +56,7 @@ t('a disallowed endpoint is robots_denied and never requested', async () => {
      assert.deepStrictEqual(out.runs.map(r => r.state), ['robots_denied', 'ok']);
      assert.strictEqual(web.hits('/private/feed.xml').length, 0);
      assert.strictEqual(web.hits('/robots.txt').length, 1, 'robots.txt is cached per origin');
--    assert.ok((await svc.outbox.all()).some(e => e.event_type === 'sources.fetch.failed' && e.payload.state === 'robots_denied'));
++    assert.ok((await svc.db.prepare('SELECT envelope FROM service_outbox').all()).some(r => r.envelope.event_type === 'sources.fetch.failed' && r.envelope.payload.state === 'robots_denied'));
  });
  
  t('a redirect into a disallowed path is refused before it is followed', async () => {
ddiff --git a/test/scheduler-events.test.js b/test/scheduler-events.test.js
iindex f454cbf..4e8852e 100644
---- a/test/scheduler-events.test.js
++++ b/test/scheduler-events.test.js
@@@ -52,25 +52,35 @@ t('due, enabled sources are fetched on schedule; disabled and manual ones never
  
  t('events are valid envelopes and reach OpenVibe.Events through the relay', async () => {
      const received = [];
++    const tokenRequests = [];
      const events = http.createServer((req, res) => {
          const chunks = [];
          req.on('data', c => chunks.push(c));
          req.on('end', () => {
++            if (req.url === '/oauth/token') {
++                tokenRequests.push(Object.fromEntries(new URLSearchParams(Buffer.concat(chunks).toString('utf8'))));
++                res.setHeader('Content-Type', 'application/json');
++                res.end(JSON.stringify({ access_token: 'relay', token_type: 'Bearer', expires_in: 300 }));
++                return;
++            }
++            assert.strictEqual(req.headers.authorization, 'Bearer relay');
              const body = JSON.parse(Buffer.concat(chunks).toString('utf8'));
--            received.push(...(body.events || [body]));
++            const sent = body.events || [body];
++            received.push(...sent);
              res.statusCode = 201;
--            res.end('{}');
++            res.setHeader('Content-Type', 'application/json');
++            res.end(JSON.stringify(body.events ? { results: sent.map((e, i) => ({ event_id: e.event_id, seq: i + 1 })) } : { event_id: sent[0].event_id, seq: 1 }));
          });
      });
      await new Promise(r => events.listen(0, '127.0.0.1', r));
      const s = await site({ '/robots.txt': () => ({ status: 404 }), '/f': () => ({ body: FEED }), '/bad': () => ({ status: 502 }) });
--    const tokenClient = { async authHeaders() { return { Authorization: 'Bearer relay' }; }, invalidate() {} };
--    const svc = await boot({ env: { EVENTS_URL: `http://127.0.0.1:${events.address().port}` }, tokenClient });
++    const svc = await boot({ env: { EVENTS_URL: `http://127.0.0.1:${events.address().port}`, OV_NETWORK_INTERNAL_URL: `http://127.0.0.1:${events.address().port}`, OV_OAUTH_CLIENT_SECRET: 'test-secret' } });
      try {
          await svc.registry.create(sourceDef({ key: 'ev', endpoints: [`${s.origin}/f`, `${s.origin}/bad`] }), 'test');
          await svc.ingest.run('ev', { trigger: 'manual' });
--        for (let i = 0; i < 5 && await svc.outbox.pending(); i++) await svc.relay.flush();
--        assert.strictEqual(await svc.outbox.pending(), 0);
++        for (let i = 0; i < 5 && (await svc.outbox.status()).pending; i++) await svc.outbox.outbox.flush();
++        assert.strictEqual((await svc.outbox.status()).pending, 0);
++        assert.ok(tokenRequests.some(r => r.audience === 'openvibe.events' && r.scope === 'events.event.publish' && r.client_secret === 'test-secret'));
          assert.deepStrictEqual(received.map(e => e.event_type).sort(), ['sources.fetch.failed', 'sources.item.created']);
          for (const e of received) {
              const v = validate('events.event-envelope@1', e);
ddiff --git a/test/sdk-outbox.test.js b/test/sdk-outbox.test.js
nnew file mode 100644
iindex 0000000..170da9c
---- /dev/null
++++ b/test/sdk-outbox.test.js
@@@ -0,0 +1,76 @@
++'use strict';
++/** The Sources migration and SDK outbox, without a loopback server. */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const { createServiceOutbox } = require('openvibe-sdk/events');
++const { createMockPlatform } = require('openvibe-sdk/testing');
++const { validate } = require('openvibe-contracts');
++const { createEnvelope } = require('../server/events/envelope');
++const { load } = require('../server/config');
++const { start } = require('../server/index');
++const { testDb } = require('./db');
++const { suite } = require('./helpers');
++
++const t = suite('sdk-outbox');
++const types = require('../docs/service-manifest-proposal.json').eventsProduced;
++const args = { event_type: 'sources.item.created', subject: { type: 'item', id: 'itm_test', revision: 1 }, payload: { item_id: 'itm_test' } };
++const envelope = () => createEnvelope(args, { source: 'sources', now: () => Date.now() });
++const make = (db, platform, secret = 'test-secret') => createServiceOutbox({
++    db, source: 'sources', eventsUrl: 'https://openvibe.events', networkInternalUrl: 'https://openvibe.network',
++    clientId: 'sources', clientSecret: secret, table: 'service_outbox', eventTypes: types,
++    validate: (env) => validate('events.event-envelope@1', env), fetch: platform.fetch,
++    log: { warn() {}, log() {} },
++});
++const platform = () => createMockPlatform({ clients: { sources: { secret: 'test-secret', grants: [['events.event.publish', 'openvibe.events']] } } });
++
++t('an event exists exactly when its change commits; rows wait while delivery is off', async () => {
++    const h = await testDb();
++    let svc;
++    try {
++        const p = platform();
++        const config = load({ NODE_ENV: 'test', SOURCES_WORKER: 'off' });
++        svc = await start({ config, db: h.db, fetchImpl: p.fetch, listen: false, log: { warn() {}, log() {}, error() {} } });
++        const source = { key: 'unit', category: 'news', search_visibility: null, license_note: null, terms_note: 'test' };
++        const item = (identity) => ({ identity, kind: 'article', canonical_url: `https://example.org/${identity}`, title: identity, authors: [], fields: {} });
++        const ctx = { retrievedAt: Date.now(), parserVersion: 'test' };
++        await assert.rejects(h.db.tx(async () => {
++            await svc.items.ingest(source, [item('rollback')], ctx);
++            throw new Error('rollback');
++        }), /rollback/);
++        assert.strictEqual(await h.db.value('SELECT count(*) FROM items'), 0);
++        assert.strictEqual((await svc.outbox.status()).pending, 0);
++        await h.db.tx(async () => {
++            await svc.items.ingest(source, [item('commit')], ctx);
++        });
++        assert.strictEqual(await h.db.value('SELECT count(*) FROM items'), 1);
++        assert.strictEqual((await svc.outbox.status()).pending, 1);
++        assert.strictEqual(p.state.events.length, 0);
++    } finally { if (svc) await svc.close(); await h.close(); }
++});
++
++t('the expand migration copies a legacy pending row and the SDK relays it with a service token', async () => {
++    const h = await testDb();
++    try {
++        const env = envelope();
++        await h.db.query('INSERT INTO event_outbox (event_id, event_type, envelope, created_at) VALUES ($1, $2, $3, $4)',
++            [env.event_id, env.event_type, JSON.stringify(env), Date.now()]);
++        // The harness already applied the migration (as the owner); re-run its copy step, the only DML in it, as the
++        // runtime role (the CI containers give that role no CREATE on the schema).
++        const migration = fs.readFileSync(path.join(__dirname, '..', 'migrations', '0002_sdk_outbox.sql'), 'utf8');
++        const copy = migration.slice(migration.indexOf('INSERT INTO service_outbox'));
++        assert.match(copy, /^INSERT INTO service_outbox[\s\S]*ON CONFLICT \(event_id\) DO NOTHING;/);
++        await h.db.query(copy);
++        const copied = await h.db.one('SELECT envelope FROM service_outbox WHERE event_id = $1', [env.event_id]);
++        assert.deepStrictEqual(copied.envelope, env);
++        const p = platform();
++        const out = make(h.db, p);
++        assert.strictEqual((await out.status()).pending, 1);
++        assert.deepStrictEqual(await out.outbox.flush(), { sent: 1, failed: 0, rejected: 0 });
++        assert.strictEqual((await out.status()).pending, 0);
++        assert.strictEqual(p.state.events[0].event.event_id, env.event_id);
++        assert.ok(p.stats.tokenRequests > 0);
++    } finally { await h.close(); }
++});
++
++t.run();
ddiff --git a/test/security-secrets.test.js b/test/security-secrets.test.js
iindex 3a45826..15db11e 100644
---- a/test/security-secrets.test.js
++++ b/test/security-secrets.test.js
@@@ -77,7 +77,7 @@ t('every route, as every kind of caller, with real and nonsense ids: the client
  });
  
  t('the outbox carries neither', async () => {
--    const rows = JSON.stringify(svc.outbox.all ? await svc.outbox.all() : await svc.db.prepare('SELECT * FROM event_outbox').all());
++    const rows = JSON.stringify(await svc.db.prepare('SELECT * FROM service_outbox').all());
      assert.ok(!rows.includes(SECRET), 'client secret in the outbox');
      assert.ok(!rows.includes(FEED_KEY), 'feed key in the outbox');
  });
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex 8576946..b7dd55a 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -151,7 +151,6 @@ t('a source registered over the API with internal endpoints is stored, and its f
  t('ratchet: every file that makes an outbound request itself is reviewed', () => {
      const REVIEWED = {
          'server/net/fetcher.js': 'every source fetch; the guard (checkUrl up front, lookup at connect, on every redirect hop)',
--        'server/events/outbox.js': 'OpenVibe.Events (configured)',
      };
      const root = path.join(__dirname, '..');
      const found = [];