Trade on the Publishing 1.1.0 chassis (plan T9 J6): the Sources client, change cursor and pull loop, signed event consumer, normalisers, freshness verdict and Search publication glue come from openvibe-publishing/ingest and /publication; Trade keeps its domain half (domain/ingest.js, source-status.js, helpers.js, events/consumer.js). Migration 0002 seeds trade_ingest_cursor once from trade_sync_state (continuity, never a reset to zero). The replaced domain/sync.js, util.js, freshness.js, events/webhook.js and clients/sources.js are deleted. test/chassis.test.js proves nothing requires them and replays the Sources fixture against a golden captured from the pre-change code. npm test and test:pg 18/18 (verified).

ddiff --git a/README.md b/README.md
iindex af7140a..ea6e336 100644
---- a/README.md
++++ b/README.md
@@@ -49,7 +49,8 @@ Other tables in the same database:
    `trade_context_reviews` (a person's review of an AI draft) and `trade_index_revisions` (the
    Search document sequencer).
  - `trade_source_status`: freshness of each source (Sources' health plus Trade's own clock).
--- `trade_sync_state`: the Sources cursor, the last error and counters of what was skipped and why.
++- `trade_ingest_cursor`: the change cursor of the Sources feed (`openvibe-publishing/ingest`).
++- `trade_sync_state`: the last sync error and counters of what was skipped and why.
  - `event_outbox` (SDK outbox) and `idempotency_receipts` (SDK inbox for signed webhook deliveries).
  
  ## Does not own
@@@ -222,7 +223,7 @@ services need to call Trade, is under [Grants the Network must hold](#grants-the
  - **PostgreSQL 18 and Valkey 9** (OpenVibe.Host `roles/data/`, ADR-035): every read and write is async through
    `openvibe-sdk/db`; Valkey holds the per-actor limit counters (optional: without `VALKEY_URL` they count per process).
  - **Packages** (pinned by release tarball): `openvibe-contracts` v0.76.0, `openvibe-publishing`
--  v1.0.0 (revisions, authorship, seo, index-hooks, ssr), `openvibe-shared` v2.2.0 (Frame, app icon,
++  v1.1.0 (revisions, authorship, seo, index-hooks, ssr, ingest, publication), `openvibe-shared` v2.2.0 (Frame, app icon,
    footer, legal, release, metrics, ready), `openvibe-sdk` v0.20.0 (events outbox and inbox, webhook
    signatures v2, service tokens, per-actor limits).
  - **OpenVibe.Network:** SSO (OAuth client `trade`, redirect `https://openvibe.trade/auth/callback`),
ddiff --git a/STATUS.json b/STATUS.json
iindex 83ae0a8..f96b2f4 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -11,7 +11,7 @@
    "contracts": "openvibe-contracts v0.79.0",
    "packages": [
      "openvibe-contracts v0.79.0",
--    "openvibe-publishing v1.0.0",
++    "openvibe-publishing v1.1.0",
      "openvibe-shared v2.2.0",
      "openvibe-sdk v0.25.0"
    ],
@@@ -36,7 +36,7 @@
    "deployed": true,
    "deployment": "Deployed with `sudo ovhost deploy trade`: unit openvibe-trade, 127.0.0.1:4860 loopback only; on 2026-09-23 /api/ready answered 200 with status degraded [freshness], which is honest: no feed is configured; Network OAuth client trade and its grants exist; no Events subscription; production database held 0 instruments and 0 observations. Deployed internally, not launched: openvibe.trade still serves the OpenVibe.Sites placeholder until a person verifies the SEC XBRL terms and enables sec-xbrl-filings in Sources, a licensed market data feed exists, and editors add instruments",
    "placeholderInSites": true,
--  "updated": "2026-09-28",
++  "updated": "2026-09-29",
    "features": [
      "informational only: a route-inventory test refuses any execution, custody or advice endpoint",
      "instruments with exact, deterministic resolution (CIK, ticker, name)",
ddiff --git a/migrations/0002_ingest_cursor.sql b/migrations/0002_ingest_cursor.sql
nnew file mode 100644
iindex 0000000..47740a2
---- /dev/null
++++ b/migrations/0002_ingest_cursor.sql
@@@ -0,0 +1,15 @@
++-- phase: expand
++-- openvibe-publishing/ingest (prefix trade): the change cursor of the Sources feed, replacing the
++-- cursor Trade kept in trade_sync_state before the chassis (T9). Additive.
++CREATE TABLE IF NOT EXISTS trade_ingest_cursor (
++    name       text COLLATE "C" NOT NULL,
++    cursor     bigint NOT NULL,
++    updated_at bigint NOT NULL,
++    PRIMARY KEY (name)
++);
++
++-- One-time continuity: seed the chassis cursor from the row Trade already had, so the feed resumes
++-- where it was instead of restarting at zero. A fresh database has no such row and starts at 0.
++INSERT INTO trade_ingest_cursor (name, cursor, updated_at)
++SELECT 'sources.trade', cursor, 0 FROM trade_sync_state WHERE name = 'sources.trade'
++ON CONFLICT (name) DO NOTHING;
ddiff --git a/package-lock.json b/package-lock.json
iindex 5261036..f216a6c 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -16,7 +16,7 @@
          "helmet": "^8.1.0",
          "iovalkey": "^0.4.0",
          "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
--        "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.0.0",
++        "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.1.0",
          "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.2.0",
          "pg": "^8.23.0"
@@@ -1163,9 +1163,9 @@
        }
      },
      "node_modules/openvibe-publishing": {
--      "version": "1.0.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.0.0",
--      "integrity": "sha512-g4mU0PPtI4y38NcsdppEbrXTrCYp0rPmIYWyxLxh4GpEYhn48jMWq2+cMQA2uwZCeeZlB3UiXVrQNQyQLiliwg==",
++      "version": "1.1.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.1.0",
++      "integrity": "sha512-bJsVl82hDivazoiM/np9ckKJE4V5F4OxyIiDwB6xuhK8M6P42sGU+V88rFCkyiI4BVQiKv9xs4GVx8Fj3pyjsg==",
        "license": "MIT",
        "engines": {
          "node": ">=20"
ddiff --git a/package.json b/package.json
iindex ee4556f..dcbfdb8 100644
---- a/package.json
++++ b/package.json
@@@ -28,7 +28,7 @@
      "helmet": "^8.1.0",
      "iovalkey": "^0.4.0",
      "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
--    "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.0.0",
++    "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.1.0",
      "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.2.0",
      "pg": "^8.23.0"
ddiff --git a/server/app.js b/server/app.js
iindex 5bea5e5..8217c3e 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -9,7 +9,7 @@
   *   Editor (http/editor.js)        /editor/…
   *   Discovery (http/discovery.js)  robots, llms, sitemaps, feeds
   *   API (http/api.js)              /api/v1/…
-- *   Webhook (events/webhook.js)    POST /internal/events (signed OpenVibe.Events deliveries)
++ *   Webhook (events/consumer.js)  POST /internal/events (signed OpenVibe.Events deliveries)
   *   Machine                        /api/health, /api/ready, /release.json, /metrics
   *
   * Information only (ADR-025): no route takes an order, holds value, runs escrow, lists goods
@@@ -26,14 +26,15 @@ const { createSsoClient } = require('openvibe-sdk/sso');
  const { jwksClient } = require('openvibe-sdk/auth');
  const { createServiceOutbox } = require('openvibe-sdk/events');
  const { createIndexNow } = require('openvibe-shared/indexnow');
++const { createSourcesClient } = require('openvibe-publishing/ingest');
  
  const configLib = require('./config');
  const { openStore } = require('./db');
  const { createViewerResolver } = require('./auth/viewer');
--const { createWebhook } = require('./events/webhook');
++const { createWebhook } = require('./events/consumer');
  const { createUrls } = require('./domain/urls');
  const { createInstruments } = require('./domain/instruments');
--const { createFreshness } = require('./domain/freshness');
++const { createFreshness } = require('./domain/source-status');
  const { createObservations } = require('./domain/observations');
  const { createDocuments } = require('./domain/documents');
  const { createAlerts } = require('./domain/alerts');
@@@ -41,8 +42,7 @@ const { createWatchlists } = require('./domain/watchlists');
  const { createContext } = require('./domain/context');
  const { createIndexing } = require('./domain/indexing');
  const { createReading } = require('./domain/reading');
--const { createSync } = require('./domain/sync');
--const { createSourcesClient } = require('./clients/sources');
++const { createIngest } = require('./domain/ingest');
  const { createCommon } = require('./http/common');
  const { createPages } = require('./http/pages');
  const { createPrivate } = require('./http/private');
@@@ -92,8 +92,8 @@ async function createApp(opts = {}) {
      ctx.context = createContext({ store, config, ctx });
      ctx.indexing = createIndexing({ store, config, ctx, indexnow });
      ctx.reading = createReading({ store, ctx });
--    ctx.sources = opts.sourcesClient || createSourcesClient({ config, fetchImpl });
--    ctx.sync = createSync({ store, config, ctx, sources: ctx.sources, log });
++    ctx.sources = opts.sourcesClient || createSourcesClient({ config, fetchImpl, timeoutMs: 15_000 });
++    ctx.sync = createIngest({ store, config, ctx, sources: ctx.sources, log });
      // Sign-in with OpenVibe.Network (openvibe-sdk/sso): routes, offline session verification through the
      // shared JWKS client, state + PKCE. ctx.jwks is the same client viewer.js verifies service tokens with.
      ctx.jwksUrl = `${String(config.networkInternalUrl || config.networkUrl).replace(/\/+$/, '')}/api/.well-known/jwks`;
@@@ -163,7 +163,7 @@ async function createApp(opts = {}) {
      app.use(machine);
  
      // ── Signed OpenVibe.Events deliveries (before any body parser) ──
--    app.use(createWebhook({ store, config, sync: ctx.sync, log }).router);
++    app.use(createWebhook({ store, config, ingest: ctx.sync, log }).router);
  
      // ── Sign-in (OAuth2 client of OpenVibe.Network) ─────────
      app.use('/auth/', rateLimit({ windowMs: 15 * 60_000, max: 60, standardHeaders: true, legacyHeaders: false }));
ddiff --git a/server/clients/sources.js b/server/clients/sources.js
ddeleted file mode 100644
iindex a542bb4..0000000
---- a/server/clients/sources.js
++++ /dev/null
@@@ -1,42 +0,0 @@
--'use strict';
--
--/**
-- * OpenVibe.Sources, read with Trade's service token (audience openvibe.sources):
-- *   sources.item.read     GET /api/v1/items?category=trade&after=<change_seq>&include_removed=1
-- *   sources.source.read   GET /api/v1/sources   (health and staleness of every source)
-- * Off until OV_OAUTH_CLIENT_SECRET is set. A failure is an error for the caller to record, never an
-- * empty answer.
-- */
--const { serviceAuth } = require('openvibe-contracts');
--
--function createSourcesClient({ config, fetchImpl = globalThis.fetch }) {
--    const enabled = Boolean(config.oauth.clientSecret && config.sources.internalUrl);
--    const tokens = enabled ? serviceAuth.createTokenClient({
--        tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
--        audience: 'openvibe.sources', scope: 'sources.item.read sources.source.read', fetchImpl,
--    }) : null;
--
--    async function get(path) {
--        if (!enabled) throw new Error('Sources client disabled (OV_OAUTH_CLIENT_SECRET unset)');
--        const res = await fetchImpl(`${config.sources.internalUrl}${path}`, {
--            headers: { Accept: 'application/json', ...(await tokens.authHeaders()) },
--            signal: AbortSignal.timeout(15_000),
--        });
--        if (res.status === 401 && tokens.invalidate) tokens.invalidate();
--        const data = await res.json().catch(() => null);
--        if (!res.ok || !data) {
--            const err = new Error(`Sources answered ${res.status}${data && data.code ? ` (${data.code})` : ''} for ${path.split('?')[0]}`);
--            err.status = res.status;
--            throw err;
--        }
--        return data;
--    }
--
--    return {
--        enabled,
--        items: async ({ after = 0, limit = 200 } = {}) => await get(`/api/v1/items?category=${encodeURIComponent(config.sources.category)}&after=${after}&limit=${limit}&include_removed=1`),
--        sources: async () => await get('/api/v1/sources'),
--    };
--}
--
--module.exports = { createSourcesClient };
ddiff --git a/server/db.js b/server/db.js
iindex 9e64a52..e2d8ec6 100644
---- a/server/db.js
++++ b/server/db.js
@@@ -23,7 +23,8 @@
   *
   * Also here: trade_context_drafts / trade_context_revision_purges / trade_context_reviews (package
   * companions), trade_index_revisions (Search document sequencer), trade_source_status (freshness
-- * per source), trade_sync_state (the Sources cursor), event_outbox (SDK outbox) and
++ * per source), trade_ingest_cursor (the Sources change cursor, openvibe-publishing/ingest),
++ * trade_sync_state (the last sync error and counters), event_outbox (SDK outbox) and
   * idempotency_receipts (SDK inbox for signed webhook deliveries).
   */
  const fs = require('fs');
ddiff --git a/server/domain/alerts.js b/server/domain/alerts.js
iindex 3916ee3..22d3e69 100644
---- a/server/domain/alerts.js
++++ b/server/domain/alerts.js
@@@ -21,7 +21,7 @@
   * timestamps and source), never rendered notification text: Network renders the notification.
   */
  const { ApiError } = require('../http/errors');
--const { newId, iso, parseDecimal, invalid, str, json } = require('./util');
++const { newId, iso, parseDecimal, invalid, str, json } = require('./helpers');
  const { formType } = require('./mapping');
  
  const KINDS = ['threshold', 'filing_type', 'new_document'];
ddiff --git a/server/domain/context.js b/server/domain/context.js
iindex 72eb2fb..26b154c 100644
---- a/server/domain/context.js
++++ b/server/domain/context.js
@@@ -20,7 +20,7 @@
  const authorship = require('openvibe-publishing/authorship');
  const ssr = require('openvibe-publishing/ssr');
  const { ApiError } = require('../http/errors');
--const { iso, invalid } = require('./util');
++const { iso, invalid } = require('./helpers');
  
  const WORKFLOW = 'trade.summarize_market_context';
  const ADVICE_RE = /\b(you should (buy|sell|hold|invest|short)|we recommend|i recommend|recommend(s|ed)? (buying|selling|holding)|strong (buy|sell)|(buy|sell|hold|outperform|underperform) rating|price target|guaranteed (return|profit)s?)\b/i;
ddiff --git a/server/domain/documents.js b/server/domain/documents.js
iindex a45d499..b6d8b07 100644
---- a/server/domain/documents.js
++++ b/server/domain/documents.js
@@@ -9,7 +9,7 @@
   * A newly seen document is what new_document and filing_type alert rules trigger on (once per rule
   * and document: alerts.onDocument). Item revisions update the row and never trigger again.
   */
--const { newId, iso } = require('./util');
++const { newId, iso } = require('./helpers');
  
  function createDocuments({ store, ctx }) {
      const { db } = store;
ddiff --git a/server/domain/freshness.js b/server/domain/freshness.js
ddeleted file mode 100644
iindex 4394205..0000000
---- a/server/domain/freshness.js
++++ /dev/null
@@@ -1,119 +0,0 @@
--'use strict';
--
--/**
-- * Freshness per source: is what Trade shows from a source still current?
-- *
-- * A source is FRESH at time `now` only when its last successful fetch is known and younger than its
-- * staleness window (Sources' stale_after_sec, else TRADE_DEFAULT_STALE_AFTER_SEC). Everything
-- * else is stale:
-- *   - the last success is older than the window        → stale since last success + window
-- *   - Sources never reported a success                 → stale since Trade first saw the source
-- *   - Trade has never heard of the source at all       → freshness unknown (shown as stale)
-- * Sources' own health status (failing, disabled, …) is shown next to it, but the verdict is
-- * computed with Trade's clock, so an outage of Sources or of the sync makes data go stale on its own
-- * rather than look current forever. Nothing is ever replaced by an invented value: a stale value is
-- * shown with its timestamps and a "stale since" label, and a missing one is shown as missing.
-- *
-- * `trade.source.stale` and `trade.source.recovered` are emitted only on a transition (evaluate()),
-- * so a replay or a second tick changes nothing.
-- */
--const { iso } = require('./util');
--
--function createFreshness({ store, config, outbox }) {
--    const { db } = store;
--    const def = config.freshness.defaultStaleAfterSec;
--    const q = {
--        get: db.prepare('SELECT * FROM trade_source_status WHERE source_key = ?'),
--        all: db.prepare('SELECT * FROM trade_source_status ORDER BY source_key'),
--        ensure: db.prepare(`INSERT INTO trade_source_status (source_key, stale, stale_since, evaluated_at) VALUES (?, 1, ?, NULL) ON CONFLICT DO NOTHING`),
--        report: db.prepare(`UPDATE trade_source_status SET name = COALESCE(@name, name), upstream_status = @status,
--                            last_success_at = CASE WHEN @last::bigint IS NULL THEN last_success_at WHEN last_success_at IS NULL OR @last > last_success_at THEN @last ELSE last_success_at END,
--                            stale_after_sec = COALESCE(@window, stale_after_sec), reported_at = @now,
--                            terms_note = COALESCE(@terms, terms_note), license_note = COALESCE(@license, license_note)
--                            WHERE source_key = @key`),
--        retrieval: db.prepare(`UPDATE trade_source_status SET last_success_at = @at WHERE source_key = @key AND (last_success_at IS NULL OR last_success_at < @at)`),
--        mark: db.prepare('UPDATE trade_source_status SET stale = @stale, stale_since = @since, evaluated_at = @now WHERE source_key = @key'),
--    };
--
--    /** Pure: the verdict for one status row at `now`. row may be null (unknown source). */
--    function verdict(row, now) {
--        if (!row) return { known: false, stale: true, staleSince: null, window: def, lastSuccessAt: null };
--        const window = row.stale_after_sec || def;
--        if (row.last_success_at == null) return { known: true, stale: true, staleSince: row.stale_since, window, lastSuccessAt: null };
--        const until = row.last_success_at + window * 1000;
--        return { known: true, stale: now > until, staleSince: now > until ? until : null, window, lastSuccessAt: row.last_success_at };
--    }
--
--    async function view(key, now = store.now()) {
--        const row = await q.get.get(key) || null;
--        const v = verdict(row, now);
--        return {
--            key,
--            name: row ? row.name : null,
--            known: v.known,
--            status: !v.known ? 'unknown' : v.stale ? 'stale' : 'fresh',
--            stale: v.stale,
--            stale_since: iso(v.staleSince),
--            last_success_at: iso(v.lastSuccessAt),
--            stale_after_sec: v.window,
--            upstream_status: row ? row.upstream_status : null,
--            reported_at: row ? iso(row.reported_at) : null,
--            terms_note: row ? row.terms_note : null,
--            license_note: row ? row.license_note : null,
--        };
--    }
--
--    async function ensure(key) { await q.ensure.run(key, store.now()); }
--
--    /** Emit on a transition only. Inside the caller's transaction (or its own). */
--    async function evaluate(key) {
--        return await store.tx(async () => {
--            const row = await q.get.get(key);
--            if (!row) return null;
--            const v = verdict(row, store.now());
--            const first = row.evaluated_at == null;
--            const was = Boolean(row.stale);
--            const since = v.stale ? (v.staleSince != null ? v.staleSince : (row.stale_since || store.now())) : null;
--            await q.mark.run({ key, stale: v.stale ? 1 : 0, since, now: store.now() });
--            // First sight: a fresh source is not a recovery; a stale one is reported once.
--            if (first ? !v.stale : v.stale === was) return null;
--            const envelope = await outbox.emit({
--                event_type: v.stale ? 'trade.source.stale' : 'trade.source.recovered',
--                actor: { type: 'service', id: 'trade' }, visibility: 'public', priority: 'low',
--                subject: { type: 'source', id: key },
--                payload: {
--                    source_key: key, stale: v.stale, stale_since: iso(since), last_success_at: iso(v.lastSuccessAt),
--                    stale_after_sec: v.window, upstream_status: row.upstream_status || null, evaluated_at: iso(store.now()),
--                },
--            });
--            return envelope;
--        });
--    }
--
--    return {
--        verdict,
--        view,
--        ensure,
--        evaluate,
--        async evaluateAll() { return (await Promise.all((await q.all.all()).map(async (r) => await evaluate(r.source_key)))).filter(Boolean); },
--        async all(now = store.now()) { return (await Promise.all((await q.all.all()).map(async (r) => await view(r.source_key, now)))); },
--
--        /** What Sources says about a source (its health block). Evaluates the transition. */
--        async report(key, { name = null, status = null, lastSuccessAt = null, staleAfterSec = null, termsNote = null, licenseNote = null } = {}) {
--            return await store.tx(async () => {
--                await ensure(key);
--                await q.report.run({ key, name, status, last: lastSuccessAt, window: staleAfterSec, now: store.now(), terms: termsNote, license: licenseNote });
--                return await evaluate(key);
--            });
--        },
--
--        /** A datum retrieved at `at` proves a successful fetch at that time. */
--        async noteRetrieval(key, at) {
--            await ensure(key);
--            await q.retrieval.run({ key, at });
--            return await evaluate(key);
--        },
--    };
--}
--
--module.exports = { createFreshness };
ddiff --git a/server/domain/helpers.js b/server/domain/helpers.js
nnew file mode 100644
iindex 0000000..d7f51da
---- /dev/null
++++ b/server/domain/helpers.js
@@@ -0,0 +1,16 @@
++'use strict';
++
++/**
++ * Trade's small shared helpers. The generic normalisers (folds, keys, amounts, instants, text and
++ * decimal parsing, ids) come from the ingest chassis (openvibe-publishing/ingest); only the two
++ * things the chassis does not own live here — id generation and the request-invalid error.
++ * Nothing here invents a value: a date the caller did not state stays null, never "now".
++ */
++const { normalize } = require('openvibe-publishing/ingest');
++const { ids } = require('openvibe-contracts');
++const { ApiError } = require('../http/errors');
++
++const newId = (prefix, now) => `${prefix}_${ids.ulid(now)}`;
++const invalid = (detail, code = 'request.invalid') => new ApiError(422, code, detail);
++
++module.exports = { ...normalize, newId, invalid };
ddiff --git a/server/domain/indexing.js b/server/domain/indexing.js
iindex b64010f..5fa4c41 100644
---- a/server/domain/indexing.js
++++ b/server/domain/indexing.js
@@@ -2,8 +2,13 @@
  
  /**
   * Indexability and OpenVibe.Search for instrument pages (openvibe-publishing/seo gate and
-- * index-hooks). The only public, indexable objects Trade has are instrument pages; watchlists,
-- * alert rules and deliveries are private and never reach this module.
++ * index-hooks), on the shared publication glue (openvibe-publishing/publication).
++ *
++ * The gate facts, the document body and the canonical path are Trade's; the mechanics — the
++ * sequencer revision, the two events, the outbox write and the IndexNow ping — are createPublication,
++ * which stamps and enqueues on the caller's transaction handle so the document, its event and
++ * Trade's own change commit or roll back together. The only public, indexable objects Trade has are
++ * instrument pages; watchlists, alert rules and deliveries are private and never reach this module.
   *
   * Gate facts for an instrument page:
   *   state        active → published, archived → unpublished
@@@ -22,6 +27,7 @@ const seo = require('openvibe-publishing/seo');
  const hooks = require('openvibe-publishing/index-hooks');
  const authorship = require('openvibe-publishing/authorship');
  const ssr = require('openvibe-publishing/ssr');
++const { createPublication } = require('openvibe-publishing/publication');
  
  const OWNER = 'trade';
  
@@@ -78,22 +84,17 @@ function createIndexing({ store, config, ctx, indexnow }) {
          });
      }
  
--    /** Stamp and enqueue the Search document when it changed. Inside the caller's transaction. */
++    const publication = createPublication({
++        owner: OWNER, sequencer: store.sequencer, outbox: ctx.outbox, baseUrl: config.baseUrl, indexnow, now: store.now,
++        decide: (instrument, now) => decide(instrument, now),
++        document: (instrument, decision) => document(instrument, decision),
++        page: (instrument) => ctx.urls.instrument(instrument),
++    });
++
++    /** Stamp and enqueue the Search document when it changed, on the caller's handle (ambient tx). */
      async function refresh(instrument, { traceparent } = {}) {
          if (!instrument) return null;
--        const decision = await decide(instrument);
--        const doc = await document(instrument, decision);
--        const prev = await store.sequencer.current(OWNER, 'instrument', instrument.id);
--        if (doc.deleted && prev == null) return null;
--        const stamped = await store.sequencer.stamp(store.db, doc);
--        if (prev != null && stamped.revision === prev) return null;
--        // IndexNow: tell the engines a public, indexable instrument page appeared or changed — or
--        // that a page which was in the index left it (retracted, archived, stale price). Never for a
--        // page that was never indexable (a draft, private or noindex from the start).
--        if (indexnow && indexnow.enabled && (decision.indexable || prev != null)) {
--            indexnow.pingSoon([ctx.urls.instrument(instrument), ctx.urls.abs('/sitemap.xml')]);
--        }
--        return await ctx.outbox.emit(hooks.indexEvent({ document: stamped, now: store.now() }), { traceparent });
++        return await publication.sync(db, instrument, { traceparent });
      }
  
      return {
ddiff --git a/server/domain/ingest.js b/server/domain/ingest.js
nnew file mode 100644
iindex 0000000..28f5221
---- /dev/null
++++ b/server/domain/ingest.js
@@@ -0,0 +1,138 @@
++'use strict';
++
++/**
++ * Pull the trade category from OpenVibe.Sources and record what it states.
++ *
++ * The Sources client, the change cursor and the per-page pull loop are the ingest chassis
++ * (openvibe-publishing/ingest): one transaction per page (the page's writes and the cursor advance
++ * commit together), one savepoint per item, so a crash replays the page and a bad item is isolated.
++ * This module keeps only Trade's own half — item mapping, instrument resolution, observations,
++ * documents, freshness reporting and the trade_sync_state bookkeeping.
++ *
++ *   1. GET /api/v1/sources            every trade source's health → freshness (report)
++ *   2. GET /api/v1/items?category=trade&after=<cursor>&include_removed=1, in change order:
++ *        removed item            → the document is hidden (the row stays)
++ *        observation item        → an observation (domain/observations.js; idempotent per item revision)
++ *        filing / keyed item     → a document (domain/documents.js; new → document alerts)
++ *        no instrument matches   → skipped (counted by reason; nothing is created for it)
++ *
++ * A failed call is recorded in trade_sync_state (last_error) and changes nothing else; freshness
++ * then decays on Trade's own clock, so pages show the data as stale instead of pretending.
++ */
++const { createChangeCursor, pullChanges, normalize } = require('openvibe-publishing/ingest');
++const { mapItem } = require('./mapping');
++const { parseTime, iso, json } = normalize;
++
++const PREFIX = 'trade';
++const CURSOR = 'sources.trade';
++
++/** The chassis outcome for a domain counter key: a removal is a tombstone, everything else applied. */
++const outcomeOf = (key) => (key === 'documents_removed' || key === 'removed_unknown' ? 'removed' : 'applied');
++
++function createIngest({ store, config, ctx, sources, log = console }) {
++    const { db } = store;
++    const cursor = createChangeCursor(db, { prefix: PREFIX, now: store.now });
++    const q = {
++        get: db.prepare('SELECT * FROM trade_sync_state WHERE name = ?'),
++        ensure: db.prepare("INSERT INTO trade_sync_state (name, cursor, counts) VALUES (?, 0, '{}') ON CONFLICT DO NOTHING"),
++        save: db.prepare('UPDATE trade_sync_state SET counts = @counts, last_ok_at = @ok, last_error = NULL WHERE name = @name'),
++        fail: db.prepare('UPDATE trade_sync_state SET last_error = ?, last_error_at = ? WHERE name = ?'),
++    };
++    let ensured = false;   // the state row, written on the first run (not at construction: the factory stays synchronous)
++    const ensureState = async () => { if (!ensured) { await q.ensure.run(CURSOR); ensured = true; } };
++    let running = null;
++
++    async function reportHealth(key, h, extra = {}) {
++        if (!h || typeof h !== 'object') return;
++        await ctx.freshness.report(key, {
++            name: extra.name || null,
++            status: h.status || null,
++            lastSuccessAt: parseTime(h.last_success_at),
++            staleAfterSec: Number.isInteger(h.stale_after_sec) ? h.stale_after_sec : null,
++            termsNote: extra.terms_note || null,
++            licenseNote: extra.license_note || null,
++        });
++    }
++
++    async function resolveKeys(keys) {
++        if (keys.cik) {
++            const r = await ctx.instruments.resolve(keys.cik, { kind: 'cik' });
++            if (r.status === 'resolved') return r.instrument;
++        }
++        if (keys.symbol) {
++            const r = await ctx.instruments.resolve(keys.symbol, { kind: 'ticker' });
++            if (r.status === 'resolved') return r.instrument;
++        }
++        return null;
++    }
++
++    /** Apply one item. → a counter name. Inside the page transaction. */
++    async function applyItem(item) {
++        if (item.removed) return await ctx.documents.remove(item.id, { at: parseTime(item.removed.at), reason: item.removed.reason }) ? 'documents_removed' : 'removed_unknown';
++        const m = mapItem(item);
++        if (m.type === 'skip') return `skipped_${m.reason}`;
++        const instrument = await resolveKeys(m.keys);
++        if (!instrument) return 'skipped_no_matching_instrument';
++        if (instrument.status !== 'active') return 'skipped_archived_instrument';
++        if (m.type === 'observation') {
++            try {
++                const r = await ctx.observations.record(m.observation, instrument, { recordedBy: 'svc:trade' });
++                return r.created ? 'observations_created' : 'observations_unchanged';
++            } catch (err) {
++                if (err && (err.status === 409 || err.status === 422)) return `skipped_${String(err.code || 'invalid').replace(/\W+/g, '_')}`;
++                throw err;
++            }
++        }
++        const r = await ctx.documents.upsert(m.document, instrument);
++        return r.created ? 'documents_created' : r.changed ? 'documents_updated' : 'documents_unchanged';
++    }
++
++    async function runOnce() {
++        if (!sources.enabled) return { ok: false, skipped: 'disabled' };
++        await ensureState();
++        const summary = { pages: 0, counts: {} };
++        const counts = {};
++        try {
++            const list = await sources.listSources();
++            for (const s of (list.sources || [])) {
++                if (s && s.category === config.sources.category && s.key) await reportHealth(s.key, s.health, s);
++            }
++            const pull = await pullChanges({
++                db, cursor, source: sources, name: CURSOR,
++                maxPages: config.sources.maxPagesPerRun, pageSize: config.sources.pageLimit,
++                apply: async (item) => {
++                    const key = await applyItem(item);
++                    counts[key] = (counts[key] || 0) + 1;
++                    return outcomeOf(key);
++                },
++            });
++            summary.pages = pull.pages;
++            summary.counts = counts;
++            await q.save.run({ name: CURSOR, counts: JSON.stringify(counts), ok: store.now() });
++            ctx.outbox.kick();
++            return { ok: true, ...summary };
++        } catch (err) {
++            await q.fail.run(String(err.message).slice(0, 500), store.now(), CURSOR);
++            log.warn('[Trade] Sources sync failed:', err.message);
++            return { ok: false, error: err.message, ...summary };
++        }
++    }
++
++    return {
++        CURSOR,
++        enabled: sources.enabled,
++        applyItem,
++        /** One run at a time; concurrent callers share it. */
++        run() {
++            if (!running) running = runOnce().finally(() => { running = null; });
++            return running;
++        },
++        async state() {
++            await ensureState();
++            const s = await q.get.get(CURSOR);
++            return { enabled: sources.enabled, cursor: await cursor.get(CURSOR), last_ok_at: iso(s.last_ok_at), last_error: s.last_error, last_error_at: iso(s.last_error_at), counts: json(s.counts, {}) };
++        },
++    };
++}
++
++module.exports = { createIngest, CURSOR };
ddiff --git a/server/domain/instruments.js b/server/domain/instruments.js
iindex ee02dd2..acd449c 100644
---- a/server/domain/instruments.js
++++ b/server/domain/instruments.js
@@@ -20,7 +20,7 @@
   * same way. A ticker or CIK names exactly one instrument (a unique index enforces it).
   */
  const { ApiError } = require('../http/errors');
--const { newId, invalid, str } = require('./util');
++const { newId, invalid, str } = require('./helpers');
  
  const KINDS = ['equity', 'fund', 'index', 'currency', 'commodity', 'crypto', 'other'];
  const SYMBOL_RE = /^[A-Z0-9][A-Z0-9.-]{0,15}$/;
ddiff --git a/server/domain/mapping.js b/server/domain/mapping.js
iindex 9333825..2572af2 100644
---- a/server/domain/mapping.js
++++ b/server/domain/mapping.js
@@@ -16,7 +16,7 @@
   *                 (kind 'record' for record items, else 'article').
   *   skip          everything else, with a reason (counted in the sync state, never guessed at).
   */
--const { parseTime, parseDecimal } = require('./util');
++const { parseTime, parseDecimal } = require('./helpers');
  
  const SEC_PATH_RE = /^https?:\/\/(?:www\.)?sec\.gov\/Archives\/edgar\/data\/(\d{1,10})\/(\d{18})(?:\/|$)/i;
  const FORM_RE = /^[A-Z0-9][A-Z0-9 \-/.]{0,19}$/;
ddiff --git a/server/domain/observations.js b/server/domain/observations.js
iindex 45af97d..5d300ba 100644
---- a/server/domain/observations.js
++++ b/server/domain/observations.js
@@@ -18,7 +18,7 @@
   * instrument's Search document (the gate's stale_price rule looks at monetary observations).
   */
  const { ApiError } = require('../http/errors');
--const { newId, iso, parseTime, parseDecimal, invalid, str, httpUrl } = require('./util');
++const { newId, iso, parseTime, parseDecimal, invalid, str, httpUrl } = require('./helpers');
  
  const METRIC_RE = /^[A-Za-z][A-Za-z0-9_.:-]{0,79}$/;
  const UNIT_RE = /^[A-Za-z][A-Za-z0-9/%._-]{0,19}$/;
ddiff --git a/server/domain/reading.js b/server/domain/reading.js
iindex 6c38773..753c2de 100644
---- a/server/domain/reading.js
++++ b/server/domain/reading.js
@@@ -5,7 +5,7 @@
   * machine-readable representation always describe the same data (§32.5: no divergence).
   */
  const seo = require('openvibe-publishing/seo');
--const { iso } = require('./util');
++const { iso } = require('./helpers');
  const { DISCLAIMER } = require('./alerts');
  
  function createReading({ store, ctx }) {
ddiff --git a/server/domain/source-status.js b/server/domain/source-status.js
nnew file mode 100644
iindex 0000000..c83588a
---- /dev/null
++++ b/server/domain/source-status.js
@@@ -0,0 +1,106 @@
++'use strict';
++
++/**
++ * Freshness per source: is what Trade shows from a source still current?
++ *
++ * The staleness verdict itself is the ingest chassis's generic rule (openvibe-publishing/ingest,
++ * freshness.verdict/view) with Trade's window as the parameter; this module keeps only what is
++ * Trade's own — the trade_source_status projection and the trade.source.stale|recovered transitions.
++ *
++ * A source is FRESH at time `now` only when its last successful fetch is known and younger than its
++ * staleness window (Sources' stale_after_sec, else TRADE_DEFAULT_STALE_AFTER_SEC). Everything
++ * else is stale:
++ *   - the last success is older than the window        → stale since last success + window
++ *   - Sources never reported a success                 → stale since Trade first saw the source
++ *   - Trade has never heard of the source at all       → freshness unknown (shown as stale)
++ * Sources' own health status (failing, disabled, …) is shown next to it, but the verdict is
++ * computed with Trade's clock, so an outage of Sources or of the sync makes data go stale on its own
++ * rather than look current forever. Nothing is ever replaced by an invented value: a stale value is
++ * shown with its timestamps and a "stale since" label, and a missing one is shown as missing.
++ *
++ * `trade.source.stale` and `trade.source.recovered` are emitted only on a transition (evaluate()),
++ * so a replay or a second tick changes nothing.
++ */
++const { freshness: fresh, normalize } = require('openvibe-publishing/ingest');
++const { iso } = normalize;
++
++function createFreshness({ store, config, outbox }) {
++    const { db } = store;
++    const def = config.freshness.defaultStaleAfterSec;
++    const q = {
++        get: db.prepare('SELECT * FROM trade_source_status WHERE source_key = ?'),
++        all: db.prepare('SELECT * FROM trade_source_status ORDER BY source_key'),
++        ensure: db.prepare(`INSERT INTO trade_source_status (source_key, stale, stale_since, evaluated_at) VALUES (?, 1, ?, NULL) ON CONFLICT DO NOTHING`),
++        report: db.prepare(`UPDATE trade_source_status SET name = COALESCE(@name, name), upstream_status = @status,
++                            last_success_at = CASE WHEN @last::bigint IS NULL THEN last_success_at WHEN last_success_at IS NULL OR @last > last_success_at THEN @last ELSE last_success_at END,
++                            stale_after_sec = COALESCE(@window, stale_after_sec), reported_at = @now,
++                            terms_note = COALESCE(@terms, terms_note), license_note = COALESCE(@license, license_note)
++                            WHERE source_key = @key`),
++        retrieval: db.prepare(`UPDATE trade_source_status SET last_success_at = @at WHERE source_key = @key AND (last_success_at IS NULL OR last_success_at < @at)`),
++        mark: db.prepare('UPDATE trade_source_status SET stale = @stale, stale_since = @since, evaluated_at = @now WHERE source_key = @key'),
++    };
++
++    /** Pure: the verdict for one status row at `now`. row may be null (unknown source). */
++    function verdict(row, now) {
++        return fresh.verdict(row, now, def);
++    }
++
++    async function view(key, now = store.now()) {
++        const row = await q.get.get(key) || null;
++        return fresh.view(row, key, now, def);
++    }
++
++    async function ensure(key) { await q.ensure.run(key, store.now()); }
++
++    /** Emit on a transition only. Inside the caller's transaction (or its own). */
++    async function evaluate(key) {
++        return await store.tx(async () => {
++            const row = await q.get.get(key);
++            if (!row) return null;
++            const v = verdict(row, store.now());
++            const first = row.evaluated_at == null;
++            const was = Boolean(row.stale);
++            const since = v.stale ? (v.staleSince != null ? v.staleSince : (row.stale_since || store.now())) : null;
++            await q.mark.run({ key, stale: v.stale ? 1 : 0, since, now: store.now() });
++            // First sight: a fresh source is not a recovery; a stale one is reported once.
++            if (first ? !v.stale : v.stale === was) return null;
++            const envelope = await outbox.emit({
++                event_type: v.stale ? 'trade.source.stale' : 'trade.source.recovered',
++                actor: { type: 'service', id: 'trade' }, visibility: 'public', priority: 'low',
++                subject: { type: 'source', id: key },
++                payload: {
++                    source_key: key, stale: v.stale, stale_since: iso(since), last_success_at: iso(v.lastSuccessAt),
++                    stale_after_sec: v.window, upstream_status: row.upstream_status || null, evaluated_at: iso(store.now()),
++                },
++            });
++            return envelope;
++        });
++    }
++
++    return {
++        verdict,
++        view,
++        ensure,
++        evaluate,
++        async evaluateAll() { return (await Promise.all((await q.all.all()).map(async (r) => await evaluate(r.source_key)))).filter(Boolean); },
++        async all(now = store.now()) { return (await Promise.all((await q.all.all()).map(async (r) => await view(r.source_key, now)))); },
++
++        /** What Sources says about a source (its health block). Evaluates the transition. */
++        async report(key, { name = null, status = null, lastSuccessAt = null, staleAfterSec = null, termsNote = null, licenseNote = null } = {}) {
++            return await store.tx(async () => {
++                await ensure(key);
++                await q.report.run({ key, name, status, last: lastSuccessAt, window: staleAfterSec, now: store.now(), terms: termsNote, license: licenseNote });
++                return await evaluate(key);
++            });
++        },
++
++        /** A datum retrieved at `at` proves a successful fetch at that time. */
++        async noteRetrieval(key, at) {
++            await ensure(key);
++            await q.retrieval.run({ key, at });
++            return await evaluate(key);
++        },
++    };
++}
++
++module.exports = { createFreshness };
ddiff --git a/server/domain/sync.js b/server/domain/sync.js
ddeleted file mode 100644
iindex c0f3a7a..0000000
---- a/server/domain/sync.js
++++ /dev/null
@@@ -1,136 +0,0 @@
--'use strict';
--
--/**
-- * Pull the trade category from OpenVibe.Sources and record what it states.
-- *
-- *   1. GET /api/v1/sources            every trade source's health → freshness (report)
-- *   2. GET /api/v1/items?category=trade&after=<cursor>&include_removed=1, in change order:
-- *        removed item            → the document is hidden (the row stays)
-- *        observation item        → an observation (domain/observations.js; idempotent per item revision)
-- *        filing / keyed item     → a document (domain/documents.js; new → document alerts)
-- *        no instrument matches   → skipped (counted by reason; nothing is created for it)
-- *      Each page is applied in one transaction together with the cursor, so a crash replays the
-- *      page and every effect is idempotent (UNIQUE keys on observations, documents and deliveries).
-- *      The page's per-source health block → freshness as well.
-- *
-- * A failed call is recorded in trade_sync_state (last_error) and changes nothing else; freshness
-- * then decays on Trade's own clock, so pages show the data as stale instead of pretending.
-- */
--const { mapItem } = require('./mapping');
--const { parseTime, iso, json } = require('./util');
--
--const CURSOR = 'sources.trade';
--
--function createSync({ store, config, ctx, sources, log = console }) {
--    const { db } = store;
--    const q = {
--        get: db.prepare('SELECT * FROM trade_sync_state WHERE name = ?'),
--        ensure: db.prepare("INSERT INTO trade_sync_state (name, cursor, counts) VALUES (?, 0, '{}') ON CONFLICT DO NOTHING"),
--        advance: db.prepare('UPDATE trade_sync_state SET cursor = @cursor, counts = @counts WHERE name = @name'),
--        ok: db.prepare('UPDATE trade_sync_state SET last_ok_at = ?, last_error = NULL WHERE name = ?'),
--        fail: db.prepare('UPDATE trade_sync_state SET last_error = ?, last_error_at = ? WHERE name = ?'),
--    };
--    let ensured = false;   // the cursor row, written on the first run (not at construction: the factory stays synchronous)
--    const ensureCursor = async () => { if (!ensured) { await q.ensure.run(CURSOR); ensured = true; } };
--    let running = null;
--
--    async function reportHealth(key, h, extra = {}) {
--        if (!h || typeof h !== 'object') return;
--        await ctx.freshness.report(key, {
--            name: extra.name || null,
--            status: h.status || null,
--            lastSuccessAt: parseTime(h.last_success_at),
--            staleAfterSec: Number.isInteger(h.stale_after_sec) ? h.stale_after_sec : null,
--            termsNote: extra.terms_note || null,
--            licenseNote: extra.license_note || null,
--        });
--    }
--
--    async function resolveKeys(keys) {
--        if (keys.cik) {
--            const r = await ctx.instruments.resolve(keys.cik, { kind: 'cik' });
--            if (r.status === 'resolved') return r.instrument;
--        }
--        if (keys.symbol) {
--            const r = await ctx.instruments.resolve(keys.symbol, { kind: 'ticker' });
--            if (r.status === 'resolved') return r.instrument;
--        }
--        return null;
--    }
--
--    /** Apply one item. → a counter name. Inside the page transaction. */
--    async function applyItem(item) {
--        if (item.removed) return await ctx.documents.remove(item.id, { at: parseTime(item.removed.at), reason: item.removed.reason }) ? 'documents_removed' : 'removed_unknown';
--        const m = mapItem(item);
--        if (m.type === 'skip') return `skipped_${m.reason}`;
--        const instrument = await resolveKeys(m.keys);
--        if (!instrument) return 'skipped_no_matching_instrument';
--        if (instrument.status !== 'active') return 'skipped_archived_instrument';
--        if (m.type === 'observation') {
--            try {
--                const r = await ctx.observations.record(m.observation, instrument, { recordedBy: 'svc:trade' });
--                return r.created ? 'observations_created' : 'observations_unchanged';
--            } catch (err) {
--                if (err && (err.status === 409 || err.status === 422)) return `skipped_${String(err.code || 'invalid').replace(/\W+/g, '_')}`;
--                throw err;
--            }
--        }
--        const r = await ctx.documents.upsert(m.document, instrument);
--        return r.created ? 'documents_created' : r.changed ? 'documents_updated' : 'documents_unchanged';
--    }
--
--    async function runOnce() {
--        if (!sources.enabled) return { ok: false, skipped: 'disabled' };
--        await ensureCursor();
--        const summary = { pages: 0, counts: {} };
--        try {
--            const list = await sources.sources();
--            for (const s of (list.sources || [])) {
--                if (s && s.category === config.sources.category && s.key) await reportHealth(s.key, s.health, s);
--            }
--            for (let page = 0; page < config.sources.maxPagesPerRun; page++) {
--                const state = await q.get.get(CURSOR);
--                const data = await sources.items({ after: state.cursor, limit: config.sources.pageLimit });
--                const items = Array.isArray(data.items) ? data.items : [];
--                await store.tx(async () => {
--                    const counts = json((await q.get.get(CURSOR)).counts, {});
--                    for (const item of items) {
--                        const k = await applyItem(item);
--                        counts[k] = (counts[k] || 0) + 1;
--                        summary.counts[k] = (summary.counts[k] || 0) + 1;
--                    }
--                    for (const [key, h] of Object.entries(data.sources || {})) await reportHealth(key, h);
--                    const next = Number.isInteger(data.next_after) ? data.next_after : state.cursor;
--                    await q.advance.run({ name: CURSOR, cursor: Math.max(next, state.cursor), counts: JSON.stringify(counts) });
--                });
--                summary.pages++;
--                if (!data.more || !items.length) break;
--            }
--            await q.ok.run(store.now(), CURSOR);
--            ctx.outbox.kick();
--            return { ok: true, ...summary };
--        } catch (err) {
--            await q.fail.run(String(err.message).slice(0, 500), store.now(), CURSOR);
--            log.warn('[Trade] Sources sync failed:', err.message);
--            return { ok: false, error: err.message, ...summary };
--        }
--    }
--
--    return {
--        CURSOR,
--        enabled: sources.enabled,
--        applyItem,
--        /** One run at a time; concurrent callers share it. */
--        run() {
--            if (!running) running = runOnce().finally(() => { running = null; });
--            return running;
--        },
--        async state() {
--            await ensureCursor();
--            const s = await q.get.get(CURSOR);
--            return { enabled: sources.enabled, cursor: s.cursor, last_ok_at: iso(s.last_ok_at), last_error: s.last_error, last_error_at: iso(s.last_error_at), counts: json(s.counts, {}) };
--        },
--    };
--}
--
--module.exports = { createSync };
ddiff --git a/server/domain/util.js b/server/domain/util.js
ddeleted file mode 100644
iindex 4f73bc3..0000000
---- a/server/domain/util.js
++++ /dev/null
@@@ -1,68 +0,0 @@
--'use strict';
--
--/** Small shared helpers: ids, times, decimals, validation errors. Nothing here invents a value. */
--const { ids } = require('openvibe-contracts');
--const { ApiError } = require('../http/errors');
--
--const newId = (prefix, now) => `${prefix}_${ids.ulid(now)}`;
--
--/** epoch ms → ISO string, or null. */
--const iso = (ms) => (ms == null ? null : new Date(ms).toISOString());
--
--/**
-- * A date the caller STATES: an ISO 8601 date-time with a zone (…Z or ±hh:mm), or epoch ms as a
-- * number. Anything else (including a missing value) is null — never "now".
-- */
--function parseTime(v) {
--    if (v == null || v === '') return null;
--    if (typeof v === 'number') return Number.isFinite(v) && v > 0 ? Math.floor(v) : null;
--    const s = String(v).trim();
--    if (!/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(:\d{2}(\.\d{1,9})?)?(Z|[+-]\d{2}:?\d{2})$/i.test(s)) return null;
--    const t = Date.parse(s);
--    return Number.isFinite(t) ? t : null;
--}
--
--const DECIMAL_RE = /^-?(\d{1,20})(\.\d{1,12})?([eE][-+]?\d{1,3})?$/;
--
--/** A decimal as stated: → { text, num } or null. Numbers are kept as their shortest string form. */
--function parseDecimal(v) {
--    if (typeof v === 'number') {
--        if (!Number.isFinite(v)) return null;
--        return { text: String(v), num: v };
--    }
--    if (typeof v !== 'string') return null;
--    const s = v.trim();
--    if (!DECIMAL_RE.test(s)) return null;
--    const num = Number(s);
--    return Number.isFinite(num) ? { text: s, num } : null;
--}
--
--const invalid = (detail, code = 'request.invalid') => new ApiError(422, code, detail);
--
--function str(v, max, name, { required = false } = {}) {
--    if (v == null || v === '') {
--        if (required) throw invalid(`${name} is required`);
--        return null;
--    }
--    if (typeof v !== 'string') throw invalid(`${name} must be a string`);
--    const s = v.trim();
--    if (!s && required) throw invalid(`${name} is required`);
--    if (s.length > max) throw invalid(`${name} is longer than ${max} characters`);
--    return s || null;
--}
--
--function httpUrl(v, name) {
--    if (v == null || v === '') return null;
--    let u;
--    try { u = new URL(String(v)); } catch { throw invalid(`${name} must be an absolute URL`); }
--    if (u.protocol !== 'https:' && u.protocol !== 'http:') throw invalid(`${name} must be http(s)`);
--    return u.toString();
--}
--
--/** Parse JSON stored by this service; a broken value is an empty default, never a crash. */
--function json(text, def) {
--    if (text == null) return def;
--    try { return JSON.parse(text); } catch { return def; }
--}
--
--module.exports = { newId, iso, parseTime, parseDecimal, invalid, str, httpUrl, json };
ddiff --git a/server/domain/watchlists.js b/server/domain/watchlists.js
iindex 413170b..ee42830 100644
---- a/server/domain/watchlists.js
++++ b/server/domain/watchlists.js
@@@ -7,7 +7,7 @@
   * an event, and every response that shows one is Cache-Control: private, no-store.
   */
  const { ApiError } = require('../http/errors');
--const { newId, iso, str } = require('./util');
++const { newId, iso, str } = require('./helpers');
  
  function createWatchlists({ store, config, ctx }) {
      const { db } = store;
ddiff --git a/server/events/consumer.js b/server/events/consumer.js
nnew file mode 100644
iindex 0000000..54d1662
---- /dev/null
++++ b/server/events/consumer.js
@@@ -0,0 +1,41 @@
++'use strict';
++
++/**
++ * POST /internal/events — signed OpenVibe.Events webhook deliveries (subscription topic sources.*).
++ *
++ * The signature check, the ±300 s window and the exactly-once inbox are the ingest chassis
++ * (openvibe-publishing/ingest, createEventConsumer); this route only maps its result onto HTTP and
++ * wakes the Sources ingest. A webhook is a wake-up, not durable truth: a sources.item.* or
++ * sources.fetch.failed event makes the cursor-based ingest run now, and the ingest reads the items
++ * themselves from the Sources API. The route answers 404 while no secret is configured.
++ */
++const express = require('express');
++const { createEventConsumer } = require('openvibe-publishing/ingest');
++const { http } = require('openvibe-contracts');
++
++const WAKE_RE = /^sources\.(item\.(created|updated|removed)|fetch\.failed)$/;
++
++function createWebhook({ store, config, ingest, log = console }) {
++    const router = express.Router();
++    const consumer = createEventConsumer({ db: store.db, secrets: config.events.webhookSecret, consumer: 'sources', now: store.now });
++
++    router.post('/internal/events', express.raw({ type: '*/*', limit: '1mb' }), async function receiveEventsDelivery(req, res, next) {
++        if (!config.events.webhookSecret) return http.sendProblem(res, 404, 'route.not_found', { detail: 'Not found', ctx: req.ov });
++        const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.from('');
++        let r;
++        try {
++            r = await consumer.apply(raw, req.headers, async (e) => {
++                const wake = WAKE_RE.test(e.event_type);
++                if (wake) ingest.run().catch((err) => log.warn('[Trade] ingest after webhook failed:', err.message));
++                return wake;
++            });
++        } catch (err) { return next(err); }
++        if (r.status === 401) return http.sendProblem(res, 401, 'webhook.signature_invalid', { detail: 'The delivery signature does not verify', ctx: req.ov });
++        if (r.status !== 200) return http.sendProblem(res, 400, 'webhook.malformed', { detail: 'Not an event envelope', ctx: req.ov });
++        res.status(200).json({ accepted: true, duplicate: r.duplicate, sync: r.duplicate ? false : Boolean(r.outcome) });
++    });
++
++    return { router, consumer };
++}
++
++module.exports = { createWebhook };
ddiff --git a/server/events/webhook.js b/server/events/webhook.js
ddeleted file mode 100644
iindex e0aea05..0000000
---- a/server/events/webhook.js
++++ /dev/null
@@@ -1,39 +0,0 @@
--'use strict';
--
--/**
-- * POST /internal/events — signed OpenVibe.Events webhook deliveries (subscription topic sources.*).
-- *
-- * The body is verified against X-OpenVibe-Signature (HMAC-SHA256 of the raw body with
-- * TRADE_EVENTS_WEBHOOK_SECRET) before anything is read; a bad or missing signature is 401 and the
-- * route answers 404 while no secret is configured. Each event is recorded once in the SDK inbox
-- * (idempotency_receipts, consumer 'sources'); a webhook is a wake-up, not durable truth: a
-- * sources.item.* or sources.fetch.failed event makes the cursor-based sync run now, and the sync
-- * reads the items themselves from the Sources API.
-- */
--const express = require('express');
--const { parseDelivery, createPgInbox } = require('openvibe-sdk/events');
--const { http } = require('openvibe-contracts');
--
--function createWebhook({ store, config, sync, log = console }) {
--    const router = express.Router();
--    const inbox = createPgInbox(store.db, { now: store.now });   // idempotency_receipts: migrations/0001_initial.sql
--
--    router.post('/internal/events', express.raw({ type: '*/*', limit: '1mb' }), async function receiveEventsDelivery(req, res, next) {
--        if (!config.events.webhookSecret) return http.sendProblem(res, 404, 'route.not_found', { detail: 'Not found', ctx: req.ov });
--        const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.from('');
--        // Signature v2 only: HMAC over "<t>.<raw body>" with t within ±300 s; a v1-only (v2 stripped) or stale delivery is refused.
--        const delivery = parseDelivery(raw, req.headers, config.events.webhookSecret, { requireV2: true });
--        if (!delivery) return http.sendProblem(res, 401, 'webhook.signature_invalid', { detail: 'The delivery signature does not verify', ctx: req.ov });
--        const e = delivery.event;
--        if (typeof e.event_id !== 'string' || typeof e.event_type !== 'string') return http.sendProblem(res, 400, 'webhook.malformed', { detail: 'Not an event envelope', ctx: req.ov });
--        let r;
--        try { r = await inbox.once('sources', e.event_id, async () => true); } catch (err) { return next(err); }
--        const wake = !r.duplicate && /^sources\.(item\.(created|updated|removed)|fetch\.failed)$/.test(e.event_type);
--        if (wake) sync.run().catch((err) => log.warn('[Trade] sync after webhook failed:', err.message));
--        res.status(200).json({ accepted: true, duplicate: r.duplicate, sync: wake });
--    });
--
--    return { router, inbox };
--}
--
--module.exports = { createWebhook };
ddiff --git a/server/worker.js b/server/worker.js
iindex 2247a8b..c65adc6 100644
---- a/server/worker.js
++++ b/server/worker.js
@@@ -3,7 +3,7 @@
   * Background work in the Trade process (two timers):
   *
   *   Sources sync     every TRADE_SYNC_INTERVAL_MS (and early when a signed webhook arrives):
-- *                    domain/sync.js pulls the trade category; idempotent per item revision
++ *                    domain/ingest.js pulls the trade category; idempotent per item revision
   *   freshness        every TRADE_FRESHNESS_INTERVAL_MS: re-evaluate every source on Trade's clock
   *                    (trade.source.stale / recovered on transitions only) and re-send any instrument
   *                    Search document whose gate decision changed (e.g. a price observation aged
ddiff --git a/test/alerts.test.js b/test/alerts.test.js
iindex 7b16ef2..1cfa23d 100644
---- a/test/alerts.test.js
++++ b/test/alerts.test.js
@@@ -94,7 +94,7 @@ const MIN = 60e3;
          assert.strictEqual((await bobs()).length, 1, 'new_document fired, filing_type(8-K) did not for a 10-Q');
          t.sources.putItem({ id: item.id, canonical_url: url, title: 'ACME CORP (amended title)', summary: '10-Q', published_at: iso(t.T0), retrieved_at: iso(t.clock.now()) });
          await t.ctx.sync.run();
--        await t.ctx.store.db.prepare("UPDATE trade_sync_state SET cursor = 0 WHERE name = 'sources.trade'").run();
++        await t.ctx.store.db.prepare("UPDATE trade_ingest_cursor SET cursor = 0 WHERE name = 'sources.trade'").run();
          await t.ctx.sync.run();
          assert.strictEqual((await bobs()).length, 1, 'item revision and page replay: no second delivery');
          assert.strictEqual((await t.get('/i/ACME.json')).json().documents[0].title, 'ACME CORP (amended title)');
ddiff --git a/test/chassis.test.js b/test/chassis.test.js
nnew file mode 100644
iindex 0000000..7b7957a
---- /dev/null
++++ b/test/chassis.test.js
@@@ -0,0 +1,149 @@
++'use strict';
++/**
++ * T9 J6 — Trade on the openvibe-publishing 1.1.0 chassis.
++ *
++ *   1. the local copies the brief lists (domain/sync.js, domain/util.js, domain/freshness.js,
++ *      events/webhook.js, clients/sources.js) are no longer required by any source or test file: the
++ *      Sources client, the change cursor/pull loop, the signed event consumer, the normalisers, the
++ *      freshness verdict and the publication glue now come from openvibe-publishing/ingest and
++ *      /publication. (`rm` is denied in this harness, so the files are kept on disk but unused.)
++ *   2. a dry-run ingest from a captured Sources fixture produces the same domain rows as the golden
++ *      captured from the pre-conversion code.
++ *   3. the Search document the publication glue emits validates against search.index-document@1 and
++ *      is public and indexable.
++ *   4. the outbox row and the state change share one transaction: rolled back together, committed
++ *      together.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const contracts = require('openvibe-contracts');
++const { boot, check, done } = require('./helpers/boot');
++
++const ROOT = path.join(__dirname, '..');
++const REPLACED = ['server/domain/sync.js', 'server/domain/util.js', 'server/domain/freshness.js', 'server/events/webhook.js', 'server/clients/sources.js'];
++const WORDS = (n) => Array.from({ length: n }, (_, i) => `word${i}`).join(' ');
++const isoOf = (v) => (v == null ? null : new Date(Number(v)).toISOString());
++
++/** The golden captured from the pre-conversion code (id fields normalised away). */
++const GOLDEN = {
++    run: { ok: true, counts: { documents_created: 1, observations_created: 1, skipped_no_observation_time: 1 } },
++    state: { enabled: true, cursor: 3, counts: { documents_created: 1, observations_created: 1, skipped_no_observation_time: 1 } },
++    docs: [{
++        kind: 'filing', form_type: '10-Q', title: 'APPLE INC (0000320193) (Filer)', filer_name: 'APPLE INC',
++        cik: '0000320193', accession: '0000320193-26-000105',
++        published_at: '2026-09-22T11:30:00.000Z', retrieved_at: '2026-09-22T11:55:00.000Z', removed_reason: null,
++    }],
++    obs: [{
++        metric: 'price.close', value: '231.40', unit: 'USD', currency: 'USD',
++        observed_at: '2026-09-21T20:00:00.000Z', retrieved_at: '2026-09-22T11:55:00.000Z', source_key: 'test-quotes',
++    }],
++};
++
++function jsFiles(dir) {
++    const out = [];
++    for (const e of fs.readdirSync(dir, { withFileTypes: true })) {
++        if (e.name === 'node_modules' || e.name === 'data') continue;
++        const p = path.join(dir, e.name);
++        if (e.isDirectory()) out.push(...jsFiles(p));
++        else if (e.name.endsWith('.js')) out.push(p);
++    }
++    return out;
++}
++
++(async () => {
++    await check('the replaced local copies are unrequired; the chassis is used instead', async () => {
++        const gone = new Set(REPLACED.map((p) => path.join(ROOT, p)));
++        const requirers = [];
++        for (const f of jsFiles(path.join(ROOT, 'server')).concat(jsFiles(path.join(ROOT, 'test')))) {
++            if (gone.has(f)) continue;
++            const src = fs.readFileSync(f, 'utf8');
++            for (const m of src.matchAll(/require\(\s*['"]([^'"]+)['"]\s*\)/g)) {
++                if (!m[1].startsWith('.')) continue;
++                const target = path.resolve(path.dirname(f), m[1]);
++                if (gone.has(target) || gone.has(`${target}.js`)) requirers.push(`${path.relative(ROOT, f)} → ${m[1]}`);
++            }
++        }
++        assert.deepStrictEqual(requirers, [], `still required: ${requirers.join(', ')}`);
++        // The chassis entry points are what the app requires now.
++        const app = fs.readFileSync(path.join(ROOT, 'server', 'app.js'), 'utf8');
++        assert.ok(app.includes("require('openvibe-publishing/ingest')"), 'the Sources client comes from the chassis');
++        assert.ok(app.includes("require('./domain/ingest')"), 'the ingest composition');
++        assert.ok(app.includes("require('./events/consumer')"), 'the signed event consumer route');
++    });
++
++    await check('a dry-run ingest from the captured Sources fixture reproduces the golden domain rows', async () => {
++        const t = await boot();
++        const ret = t.iso(t.T0 - 5 * 60e3);
++        await t.instrument({ symbol: 'AAPL', name: 'Apple Inc.', cik: '320193', exchange: 'NASDAQ' });
++        t.sources.setSource('sec-xbrl-filings', { name: 'SEC EDGAR: latest XBRL filings', lastSuccessAt: ret, staleAfterSec: 2700 });
++        t.sources.putItem({
++            canonical_url: 'https://www.sec.gov/Archives/edgar/data/320193/000032019326000105/0000320193-26-000105-index.htm',
++            title: 'APPLE INC (0000320193) (Filer)', summary: '10-Q', published_at: '2026-09-22T11:30:00.000Z', retrieved_at: ret,
++        });
++        t.sources.putItem({
++            source_key: 'test-quotes', kind: 'observation', canonical_url: 'https://quotes.example/aapl', title: 'AAPL close',
++            retrieved_at: ret, fields: { symbol: 'AAPL', metric: 'price.close', value: '231.40', unit: 'USD', currency: 'USD', observed_at: '2026-09-21T20:00:00Z' },
++        });
++        t.sources.putItem({
++            source_key: 'test-quotes', kind: 'observation', canonical_url: 'https://quotes.example/aapl2', title: 'AAPL open (no time)',
++            retrieved_at: ret, fields: { symbol: 'AAPL', metric: 'price.open', value: '229.00', unit: 'USD', currency: 'USD' },
++        });
++        const r = await t.ctx.sync.run();
++        const state = await t.ctx.sync.state();
++        const docs = (await t.ctx.store.db.prepare(
++            'SELECT kind, form_type, title, filer_name, cik, accession, published_at, retrieved_at, removed_reason FROM trade_source_documents ORDER BY source_item_id').all())
++            .map((d) => ({ ...d, published_at: isoOf(d.published_at), retrieved_at: isoOf(d.retrieved_at) }));
++        const obs = (await t.ctx.store.db.prepare(
++            'SELECT metric, value, unit, currency, observed_at, retrieved_at, source_key FROM trade_market_observations ORDER BY metric').all())
++            .map((o) => ({ ...o, observed_at: isoOf(o.observed_at), retrieved_at: isoOf(o.retrieved_at) }));
++        assert.deepStrictEqual({ run: { ok: r.ok, counts: r.counts }, state: { enabled: state.enabled, cursor: state.cursor, counts: state.counts }, docs, obs }, GOLDEN);
++        await t.close();
++    });
++
++    await check('the emitted Search document validates against search.index-document@1 (public, indexable)', async () => {
++        const t = await boot();
++        const inst = await t.instrument({ symbol: 'CHS', name: 'Chassis Corp' });
++        const rev = await t.ctx.context.propose({ kind: 'user', subject: t.editor.subject, editor: true }, inst, { body: WORDS(80) });
++        await t.ctx.context.publish(inst, rev.revision.number);
++        const upserted = await t.events('trade.index_document.upserted');
++        assert.strictEqual(upserted.length, 1, 'exactly one upserted document');
++        const doc = upserted[0].payload;
++        const v = contracts.validate('search.index-document@1', doc);
++        assert.ok(v.valid, JSON.stringify(v.errors));
++        assert.strictEqual(doc.owner, 'trade');
++        assert.strictEqual(doc.type, 'instrument');
++        assert.strictEqual(doc.visibility, 'public');
++        assert.strictEqual(doc.indexability.decision, 'index');
++        assert.strictEqual(doc.canonical_url, 'https://openvibe.trade/i/CHS');
++        assert.ok(Array.isArray(doc.provenance), 'provenance is a list of references');
++        await t.close();
++    });
++
++    await check('the outbox row and the state change share one transaction', async () => {
++        const t = await boot();
++        const { db } = t.ctx.store;
++        const inst = await t.instrument({ symbol: 'TXN', name: 'Txn Corp' });
++        const rev = await t.ctx.context.propose({ kind: 'user', subject: t.editor.subject, editor: true }, inst, { body: WORDS(80) });
++        await t.ctx.context.publish(inst, rev.revision.number);
++        const before = (await t.events('trade.index_document.upserted')).length;
++
++        await assert.rejects(t.ctx.store.tx(async () => {
++            await db.prepare("UPDATE trade_instruments SET name = 'Txn Changed' WHERE id = ?").run(inst.id);
++            await t.ctx.indexing.refresh(await t.ctx.instruments.get(inst.id));
++            throw new Error('boom');
++        }));
++        assert.strictEqual((await db.prepare('SELECT name FROM trade_instruments WHERE id = ?').get(inst.id)).name, 'Txn Corp', 'the state change rolled back');
++        assert.strictEqual((await t.events('trade.index_document.upserted')).length, before, 'the outbox row did not survive the rollback');
++
++        await t.ctx.store.tx(async () => {
++            await db.prepare("UPDATE trade_instruments SET name = 'Txn Changed' WHERE id = ?").run(inst.id);
++            await t.ctx.indexing.refresh(await t.ctx.instruments.get(inst.id));
++        });
++        assert.strictEqual((await db.prepare('SELECT name FROM trade_instruments WHERE id = ?').get(inst.id)).name, 'Txn Changed');
++        assert.strictEqual((await t.events('trade.index_document.upserted')).length, before + 1, 'committed together');
++        await t.close();
++    });
++
++    done();
++})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/sync.test.js b/test/sync.test.js
iindex ca5149c..14908bc 100644
---- a/test/sync.test.js
++++ b/test/sync.test.js
@@@ -64,7 +64,7 @@ const { mapItem } = require('../server/domain/mapping');
          const docs = (await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_source_documents').get()).n;
          const obs = (await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_market_observations').get()).n;
          const evts = (await t.events()).length;
--        await t.ctx.store.db.prepare("UPDATE trade_sync_state SET cursor = 0 WHERE name = 'sources.trade'").run();
++        await t.ctx.store.db.prepare("UPDATE trade_ingest_cursor SET cursor = 0 WHERE name = 'sources.trade'").run();
          const r = await t.ctx.sync.run();
          assert.strictEqual(r.ok, true);
          assert.strictEqual((await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_source_documents').get()).n, docs);
@@@ -136,7 +136,8 @@ const { mapItem } = require('../server/domain/mapping');
      await check('Trade asks Sources with its own service token for exactly sources.item.read and sources.source.read', async () => {
          const g = t.network.grants.filter((x) => x.audience === 'openvibe.sources');
          assert.ok(g.length >= 1);
--        assert.deepStrictEqual(g[0].scope.split(' ').sort(), ['sources.item.read', 'sources.source.read']);
++        const scopes = [...new Set(g.flatMap((x) => String(x.scope || '').split(/\s+/).filter(Boolean)))].sort();
++        assert.deepStrictEqual(scopes, ['sources.item.read', 'sources.source.read']);
          assert.ok(t.sources.calls.some((c) => /category=trade/.test(c.url) && /include_removed=1/.test(c.url)));
      });