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

ddiff --git a/README.md b/README.md
iindex 50441c4..c5181ce 100644
---- a/README.md
++++ b/README.md
@@@ -262,6 +262,10 @@ envelopes validated as `events.event-envelope@1` and payloads against their rele
  watch-lifecycle events (`watch.watch.created`, `watch.watch.updated`, `watch.watch.paused`,
  `watch.watch.removed`) are emitted on the registry's writes; they have no payload contract yet.
  
++`openvibe-sdk/events` owns the transactional outbox and relay. Pending rows live in `service_outbox`
++(migration 0004); the migration copies pending rows from `event_outbox`, which remains through the
++rollback window. When `EVENTS_URL` or the OAuth client secret is unset, rows wait for the relay.
++
  A watch owned by a service (`svc:…`) still records every observation and trigger in its own tables,
  but the person-facing payloads name a `usr_…`/`agt_…` owner (and a trigger a `usr_…` recipient), so
  those two events are logged as not emitted rather than sent with a payload the contract refuses.
ddiff --git a/STATUS.json b/STATUS.json
iindex 7d77fc3..67f06fa 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -45,7 +45,7 @@
      "the http, feed (RSS/Atom) and api carriers",
      "extraction (text, html, css subset, json/jsonpath, regex) and conditions (changed/threshold/string, for_sec)",
      "observations with value hashes, capped snapshots and retention; check runs with the full state set",
--    "the loopback-only API (one capability per route), watch.* events through the transactional outbox, per-actor limits",
++    "the loopback-only API (one capability per route), watch.* events through openvibe-sdk's transactional service_outbox, per-actor limits",
      "the public site: sign-in with OpenVibe.Network (OAuth2 + PKCE S256, sessions in Watch's database), the home page and four starting points, /watches (paged), the create and edit forms (server-side fieldsets, validation errors that keep the input), one watch with pause/resume/delete/check-now, the observation history and check runs, /how-it-works, /updates, and robots/sitemap/llms.txt",
      "the site calls the registry in-process with the API's ownership rules (another person's watch is 404 on every page and form) and refuses cross-site writes"
    ],
@@@ -68,7 +68,7 @@
      "check runs with the full terminal-state set, exponential backoff and Retry-After",
      "POST /api/v1/watches/:id/check plus the observations and checks reads",
      "GET /api/v1/resources and GET /api/v1/resources/:ovrn (watch.resource.read)",
--    "watch.* events through the outbox in the same transaction as the state change",
++    "watch.* events through openvibe-sdk's service_outbox in the same transaction as the state change",
      "per-actor limits (openvibe-sdk/limits)",
      "the public site: Network sign-in (PKCE S256, web_sessions with only the token hash), server-rendered pages complete without JavaScript, the four create templates, watch management, observation history and check runs, how-it-works and the changelog, robots/sitemap/llms.txt"
    ],
ddiff --git a/migrations/0004_sdk_outbox.sql b/migrations/0004_sdk_outbox.sql
nnew file mode 100644
iindex 0000000..1dd654a
---- /dev/null
++++ b/migrations/0004_sdk_outbox.sql
@@@ -0,0 +1,24 @@
++-- phase: expand
++-- OpenVibe.Watch: openvibe-sdk/events service outbox. The old event_outbox remains for rollback;
++-- a contract migration drops it after the N-1 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;
ddiff --git a/server/app.js b/server/app.js
iindex 53ea502..c21df8c 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, relay, 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, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null }) {
      const app = express();
      app.disable('x-powered-by');
      app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');
@@@ -94,7 +94,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
  
      // Readiness (openvibe-shared/ready): 503 only when the database fails; the Network key and the
      // check worker (running, queue keeping up) are optional and degrade it (see observability.js).
--    const readiness = createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, relay, now, release: release.release });
++    const readiness = createWatchReadiness({ db, config, registry, check, observations, 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 });
ddiff --git a/server/check.js b/server/check.js
iindex a852683..4c817e4 100644
---- a/server/check.js
++++ b/server/check.js
@@@ -25,6 +25,7 @@
  const { ids: contractIds, validate } = require('openvibe-contracts');
  const { nextDue } = require('./registry');
  const { ExtractError } = require('./extract');
++const { envelope } = require('./events/envelope');
  
  /** States that count as a failure, back off, and emit watch.check.failed. */
  const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_limited', 'budget_exceeded', 'skipped', 'disabled']);
@@@ -33,7 +34,7 @@ const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_li
  const PERSON_OWNER = /^(usr|agt)_[0-9A-HJKMNP-TV-Z]{26}$/;
  const USER_OWNER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
  
--function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, now = () => Date.now(), log = console, relay = null, ids = contractIds }) {
++function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, now = () => Date.now(), log = console, ids = contractIds }) {
      const inflight = new Set();
  
      const st = {
@@@ -170,9 +171,9 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
                  id: watch.id, at: run.finished_at, success_at: successAt, state: run.state,
                  failures, next_due: armed, not_before: notBefore,
              });
--            for (const e of events || []) await outbox.enqueue(e);
++            for (const e of events || []) await outbox.emit(envelope(e, config.serviceId, now));
          });
--        if (relay) relay.flush().catch(() => { });
++        outbox.kick().catch(() => { });
          return { run, observation, met: run.triggers > 0, failures };
      }
  
ddiff --git a/server/events/envelope.js b/server/events/envelope.js
nnew file mode 100644
iindex 0000000..7f7cd22
---- /dev/null
++++ b/server/events/envelope.js
@@@ -0,0 +1,16 @@
++'use strict';
++/** The defaults Watch used for its event envelopes before the SDK outbox owned delivery. */
++const { ids } = require('openvibe-contracts');
++
++function envelope({ 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 = { envelope };
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 65849b8..91a62f7 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -21,7 +21,8 @@ const { extract } = require('./extract');
  const condition = require('./condition');   // { evaluate, hasChanged, holds, … }
  const { createCheck } = require('./check');
  const { createScheduler } = require('./scheduler');
--const { createOutbox, createRelay } = require('./events/outbox');
++const { createServiceOutbox } = require('openvibe-sdk/events');
++const { validate } = require('openvibe-contracts');
  const { jwksClient } = require('openvibe-sdk/auth');
  const { gracefulStop } = require('openvibe-sdk/service');
  const { createAuth } = require('./auth');
@@@ -32,17 +33,19 @@ const { createApp } = require('./app');
  const accountDataLib = require('./account-data');
  const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
  
--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, tokenClient, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null } = {}) {
++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = 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, fetch: fetchImpl, log, now, table: 'service_outbox',
++        eventTypes: ['watch.watch.created', 'watch.watch.updated', 'watch.watch.paused', 'watch.watch.removed',
++            'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed'],
++        validate: (env) => validate('events.event-envelope@1', env),
      });
--    const registry = createRegistry({ db, now, outbox, maxWatchesPerOwner: config.watches.maxPerOwner });
++    const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner });
      const observations = createObservations({ db, now });
      // The only way out to the internet: SSRF guard (ports, private hosts, every redirect hop), byte
      // cap, one deadline, conditional GET; per-host spacing in front of every request.
@@@ -51,7 +54,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
      const spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });
      // Every carrier goes out through the spacer, so two watches over one host never hammer it.
      const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });
--    const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, now, log, relay });
++    const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, now, log });
      const scheduler = createScheduler({ db, check, config, now, log });
      const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
      const auth = createAuth({ config, log });
@@@ -66,19 +69,19 @@ 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, relay, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend });
++    const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend });
      // 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 observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }
          try { await sessions.prune(); } catch (err) { log.error(`[sessions] prune: ${err.message}`); }
      }, 3600 * 1000);
      pruneTimer.unref?.();
--    const outboxPruneTimer = setInterval(async () => { try { await outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);
++    const outboxPruneTimer = setInterval(async () => { try { await outbox.outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);
      outboxPruneTimer.unref?.();
  
      let server = null;
@@@ -102,7 +105,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
          if (subscriptions) subscriptions.stop();
          keys.stop();
          await scheduler.stop();
--        await relay.stop();
++        await outbox.stop();
          if (server) {
              server.closeAllConnections?.();
              await new Promise(resolve => server.close(() => resolve()));
@@@ -110,7 +113,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, relay, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
++    return { config, db, registry, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
  }
  
  /**
ddiff --git a/server/observability.js b/server/observability.js
iindex 7622b38..033c695 100644
---- a/server/observability.js
++++ b/server/observability.js
@@@ -48,7 +48,7 @@ function readers(db, { startedAt }) {
      };
  }
  
--function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, relay, now, release = null }) {
++function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, now, release = null }) {
      const read = readers(db, { startedAt: now() });
      const iso = (v) => (v == null ? null : new Date(v).toISOString());
      return createReadiness({
@@@ -97,7 +97,7 @@ function createWatchReadiness({ db, config, registry, check, observations, sched
                  watches,
                  checks_in_flight: check.inflight().length,
                  observations: dbOk ? await read.observations() : null,
--                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,
              };
          },
      });
@@@ -125,7 +125,7 @@ function registerWatchGauges(registry, { db, watches, check, outbox, now }) {
      // Left out until there is a check: a timestamp of 0 would read as "1970", not "never".
      registry.gauge({ name: 'watch_last_check_timestamp_seconds', help: 'When the last check finished (any outcome), Unix seconds', collect: async () => seconds(await read.lastCheckAt()) });
      registry.gauge({ name: 'watch_last_success_timestamp_seconds', help: 'When a watch was last checked successfully, Unix seconds', collect: async () => seconds(await read.lastSuccessAt()) });
--    registry.gauge({ name: 'watch_outbox_pending', help: 'Events waiting in the outbox', collect: async () => await outbox.pending() });
++    registry.gauge({ name: 'watch_outbox_pending', help: 'Events waiting in the outbox', collect: async () => (await outbox.status()).pending });
  }
  
  
ddiff --git a/server/registry.js b/server/registry.js
iindex 0dd900c..9d50c72 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -19,6 +19,7 @@
   */
  const { validate, ids: contractIds } = require('openvibe-contracts');
  const safe = require('./safe-regex');
++const { envelope } = require('./events/envelope');
  
  /** Source kinds that reach out on a cadence; event and webhook sources are pushed to instead. */
  const POLLING = new Set(['http', 'feed', 'api', 'node', 'run']);
@@@ -121,7 +122,7 @@ function validateView(view) {
      return view;
  }
  
--function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, maxWatchesPerOwner = 50 }) {
++function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50 }) {
      const st = {
          get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
          countOwned: db.prepare('SELECT COUNT(*) AS n FROM watches WHERE owner_sub = ? AND deleted_at IS NULL'),
@@@ -278,7 +279,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
          // The row and its event commit together: an event exists exactly when its effect does.
          await db.tx(async () => {
              await st.insert.run(row);
--            if (outbox) await outbox.enqueue(watchEvent('watch.watch.created', row, owner, { name: rec.name, status: rec.status }));
++            if (outbox) await outbox.emit(envelope(watchEvent('watch.watch.created', row, owner, { name: rec.name, status: rec.status }), source, now));
          });
          return await st.get.get(row.id);
      }
@@@ -314,7 +315,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
          await db.tx(async () => {
              await db.prepare(`UPDATE watches SET ${sets.join(', ')} WHERE id = @id AND deleted_at IS NULL`).run(args);
              if (outbox && touched.length) {
--                await outbox.enqueue(watchEvent(type, row, row.owner_sub, { name: rec.name, status: rec.status, fields: touched.sort() }));
++                await outbox.emit(envelope(watchEvent(type, row, row.owner_sub, { name: rec.name, status: rec.status, fields: touched.sort() }), source, now));
              }
          });
          return await st.get.get(row.id);
@@@ -327,7 +328,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
          const t = now();
          await db.tx(async () => {
              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.enqueue(watchEvent('watch.watch.removed', row, row.owner_sub, { name: row.name, status: row.status }));
++            if (outbox) await outbox.emit(envelope(watchEvent('watch.watch.removed', row, row.owner_sub, { name: row.name, status: row.status }), source, now));
          });
          return { id: row.id, deleted: true };
      }
ddiff --git a/test/check.test.js b/test/check.test.js
iindex 3329ff2..1f2ee8f 100644
---- a/test/check.test.js
++++ b/test/check.test.js
@@@ -33,8 +33,9 @@ const check = async (id, trigger = 'schedule') => await svc.check.run(id, { trig
  const row = async (id) => await svc.db.prepare('SELECT * FROM watches WHERE id = ?').get(id);
  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);
--const payloads = async (type) => (await svc.outbox.all()).filter(e => e.event_type === type).map(e => e.payload);
--const envelopes = async (type) => (await svc.outbox.all()).filter(e => e.event_type === type);
++const allEnvelopes = async () => (await svc.db.many('SELECT envelope FROM service_outbox ORDER BY id')).map(r => r.envelope);
++const payloads = async (type) => (await allEnvelopes()).filter(e => e.event_type === type).map(e => e.payload);
++const envelopes = async (type) => (await allEnvelopes()).filter(e => e.event_type === type);
  
  const valid = (ref, value) => {
      const r = validate(ref, value);
@@@ -292,7 +293,7 @@ t('an owned-by-a-service watch records everything and emits no person-facing eve
      const run = (await runs(w.id)).pop();
      assert.strictEqual(Number(run.triggers), 1);
      assert.strictEqual((await observations(w.id)).length, 1);
--    const all = await svc.outbox.all();
++    const all = await allEnvelopes();
      const mine = all.filter(e => e.subject.id === w.id);
      // the contract names usr_/agt_ owners for an observation and a usr_ recipient for a trigger, so
      // a service-owned watch emits neither — it keeps the record in its own tables
@@@ -303,7 +304,7 @@ t('an owned-by-a-service watch records everything and emits no person-facing eve
  t('an agent-owned watch emits observation.recorded but never a trigger (recipient is usr_ only)', async () => {
      const w = await create(agent, { condition: { op: 'lt', value: 100 } });
      await check(w.id);
--    const all = (await svc.outbox.all()).filter(e => e.subject.id === w.id);
++    const all = (await allEnvelopes()).filter(e => e.subject.id === w.id);
      const observed = all.filter(e => e.event_type === 'watch.observation.recorded');
      assert.strictEqual(observed.length, 1);
      valid('watch.observation.recorded@1', observed[0].payload);
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 06833a8..1c4b565 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
   * the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
   * log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
   */
--async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null } = {}) {
++async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = 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] }) }) });
@@@ -77,7 +77,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now, lo
      });
      // One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
      const testdb = await require('./db').testDb();
--    const h = await start({ config, db: testdb.db, log, lookupImpl, tokenClient, limitsNow, accountSend, ...(now ? { now } : {}) });
++    const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(now ? { now } : {}) });
      const base = `http://127.0.0.1:${h.server.address().port}`;
  
      /** A session cookie for a person, as /auth/callback would have set it. */
ddiff --git a/test/outbox.test.js b/test/outbox.test.js
nnew file mode 100644
iindex 0000000..7f5037c
---- /dev/null
++++ b/test/outbox.test.js
@@@ -0,0 +1,110 @@
++'use strict';
++/** The expand copy, transaction boundary, disabled relay and SDK delivery through Watch's wiring. */
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const { createDb } = require('openvibe-sdk/db');
++const { load } = require('../server/config');
++const { start } = require('../server/index');
++const { envelope } = require('../server/events/envelope');
++const { suite, silent, watchDef, newUser } = require('./helpers');
++
++const t = suite('outbox');
++const migrations = path.join(__dirname, '..', 'migrations');
++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'watch-outbox-'));
++const db = createDb({ pglite: true, service: 'watch-outbox-test', log: silent });
++let svc;
++const ms = 1750000000000;
++const event = envelope({
++    event_type: 'watch.watch.created', subject: { type: 'watch', id: 'wch_01J0000000000000000000000' },
++    payload: { watch_id: 'wch_01J0000000000000000000000', owner: 'usr_01J0000000000000000000000', name: 'old', status: 'active' },
++    priority: 'low',
++}, 'watch', () => ms);
++
++const config = (eventsUrl) => load({
++    NODE_ENV: 'test', WATCH_WORKER: 'off', EVENTS_URL: eventsUrl,
++    OV_NETWORK_INTERNAL_URL: 'http://network.test', OV_NETWORK_JWKS_URL: 'http://network.test/jwks',
++    OV_OAUTH_CLIENT_ID: 'watch', OV_OAUTH_CLIENT_SECRET: 'test-secret',
++});
++const calls = [];
++async function fakeFetch(url, opts = {}) {
++    calls.push({ url: String(url), opts });
++    if (String(url).endsWith('/oauth/token')) {
++        const body = new URLSearchParams(opts.body);
++        assert.strictEqual(body.get('audience'), 'openvibe.events');
++        assert.strictEqual(body.get('scope'), 'events.event.publish');
++        return Response.json({ access_token: 'service-token', token_type: 'Bearer', expires_in: 300 });
++    }
++    if (String(url).endsWith('/api/v1/events')) {
++        assert.strictEqual(new Headers(opts.headers).get('authorization'), 'Bearer service-token');
++        return Response.json({ event_id: event.event_id, seq: 12, duplicate: false });
++    }
++    if (String(url).endsWith('/.well-known/openvibe')) {
++        return Response.json({ issuer: 'http://network.test', contracts: { version: '0.129.0' }, services: [{ id: 'events', origin: 'http://events.test' }] });
++    }
++    if (String(url).endsWith('/jwks')) return Response.json({ keys: [] });
++    throw new Error(`unexpected request ${url}`);
++}
++
++t('expand copies only pending legacy rows into the exact SDK table', async () => {
++    try {
++        for (const name of fs.readdirSync(migrations).filter(n => /^000[1-3]_/.test(n))) fs.copyFileSync(path.join(migrations, name), path.join(dir, name));
++        await db.migrate({ dir, log: silent });
++        await db.query('INSERT INTO event_outbox (event_id, event_type, envelope, created_at, attempts, next_attempt_at) VALUES ($1, $2, $3, $4, 2, 0)', [event.event_id, event.event_type, JSON.stringify(event), ms]);
++        const sent = { ...event, event_id: 'evt_01J0000000000000000000001' };
++        await db.query('INSERT INTO event_outbox (event_id, event_type, envelope, created_at, sent_at) VALUES ($1, $2, $3, $4, $5)', [sent.event_id, sent.event_type, JSON.stringify(sent), ms, ms]);
++        fs.copyFileSync(path.join(migrations, '0004_sdk_outbox.sql'), path.join(dir, '0004_sdk_outbox.sql'));
++        await db.migrate({ dir, log: silent });
++        const rows = await db.many('SELECT event_id, envelope, attempts, next_attempt_at FROM service_outbox');
++        assert.strictEqual(rows.length, 1);
++        assert.strictEqual(rows[0].event_id, event.event_id);
++        assert.deepStrictEqual(rows[0].envelope, event);
++        assert.strictEqual(rows[0].attempts, 2);
++        assert.strictEqual(await db.value('SELECT count(*) FROM event_outbox'), 2);
++    } finally { fs.rmSync(dir, { recursive: true, force: true }); }
++});
++
++t('rows wait with the relay off and an aborted change has no event', async () => {
++    svc = await start({ config: config(''), db, fetchImpl: fakeFetch, log: silent, listen: false });
++    assert.deepStrictEqual(await svc.outbox.status(), { enabled: false, pending: 1, rejected: 0, last_error: null });
++    await assert.rejects(db.tx(async () => {
++        await db.query('INSERT INTO counters (name, value) VALUES ($1, 1)', ['aborted']);
++        await svc.outbox.emit(envelope({ event_type: 'watch.watch.updated', subject: event.subject, payload: event.payload }, 'watch', () => ms + 1));
++        throw new Error('abort');
++    }), /abort/);
++    assert.strictEqual(await db.value('SELECT count(*) FROM counters WHERE name = $1', ['aborted']), 0);
++    assert.strictEqual((await svc.outbox.status()).pending, 1);
++    assert.strictEqual(calls.filter(c => c.url.endsWith('/api/v1/events')).length, 0);
++    await svc.close();
++});
++
++t('the SDK relay posts the migrated row with a service token', async () => {
++    svc = await start({ config: config('http://events.test'), db, fetchImpl: fakeFetch, log: silent, listen: false });
++    await svc.outbox.outbox.flush();
++    const posts = calls.filter(c => c.url.endsWith('/api/v1/events'));
++    assert.strictEqual(posts.length, 1);
++    assert.strictEqual(JSON.parse(posts[0].opts.body).event_id, event.event_id);
++    assert.strictEqual((await svc.outbox.status()).pending, 0);
++    assert.strictEqual(await db.value('SELECT count(*) FROM service_outbox WHERE sent_at IS NOT NULL'), 1);
++});
++
++t('a committed registry change and its event appear together', async () => {
++    await svc.outbox.stop();
++    const principal = { sub: newUser() };
++    const row = await svc.registry.create(watchDef(), principal);
++    const events = await db.many('SELECT envelope FROM service_outbox WHERE envelope->>\'event_type\' = $1', ['watch.watch.created']);
++    assert.ok(await db.maybe('SELECT id FROM watches WHERE id = $1', [row.id]));
++    assert.ok(events.some(r => r.envelope.subject.id === row.id));
++    const before = await db.value('SELECT count(*) FROM service_outbox');
++    await assert.rejects(db.tx(async () => {
++        await svc.registry.create(watchDef(), principal);
++        throw new Error('abort registry');
++    }), /abort registry/);
++    assert.strictEqual(await db.value('SELECT count(*) FROM service_outbox'), before);
++    assert.strictEqual(await db.value('SELECT count(*) FROM watches'), 1);
++    await svc.close();
++    await db.close();
++});
++
++t.run();
ddiff --git a/test/registry.test.js b/test/registry.test.js
iindex b34065d..aedf949 100644
---- a/test/registry.test.js
++++ b/test/registry.test.js
@@@ -10,6 +10,7 @@ const { boot, suite, newUser, newAgent } = require('./helpers');
  
  const t = suite('registry');
  let svc;
++const allEnvelopes = async () => (await svc.db.many('SELECT envelope FROM service_outbox ORDER BY id')).map(r => r.envelope);
  const me = { sub: newUser() };
  const someoneElse = { sub: newUser() };
  
@@@ -97,7 +98,7 @@ t('create: an id, an owner, a name, the arm and the event', async () => {
      assert.strictEqual(row.name, 'example.org/page', 'a watch without a name is named after its source');
      assert.ok(row.next_due_at >= before && row.next_due_at <= before + 1000, 'a new watch is due on the next tick (its first reading)');
      assert.strictEqual(row.updated_by, me.sub);
--    const events = (await svc.outbox.all()).filter(e => e.subject.id === row.id);
++    const events = (await allEnvelopes()).filter(e => e.subject.id === row.id);
      assert.deepStrictEqual(events.map(e => e.event_type), ['watch.watch.created']);
      assert.strictEqual(events[0].payload.owner, me.sub);
      const agentOwned = await svc.registry.create(def(), { sub: newAgent() });
@@@ -163,7 +164,7 @@ t('patch re-arms the cadence, and pause/resume through status', async () => {
      await assert.rejects(() => svc.registry.patch(w.id, { cadence: null }, me), (err) => err.code === 'watch.invalid');
      const after = await svc.registry.get(w.id);
      assert.deepStrictEqual(after.cadence, { every_sec: 3600 });
--    const events = (await svc.outbox.all()).filter(e => e.subject.id === w.id).map(e => e.event_type);
++    const events = (await allEnvelopes()).filter(e => e.subject.id === w.id).map(e => e.event_type);
      assert.deepStrictEqual(events, ['watch.watch.created', 'watch.watch.updated', 'watch.watch.paused', 'watch.watch.updated', 'watch.watch.updated']);
  });