openvibers/openvibe.events Commits

Opaque cursors beside the global seq (plan T7, ADR-042 decision 7; openvibe-contracts v0.81.0): every answer that carries a seq also carries its cursor (publish results, pulls with next_cursor, single events, checkpoints with epoch and carrier), pull accepts after=<cursor>, the SSE id is the cursor and Last-Event-ID accepts a cursor or, for this release, a bare seq; a cursor from another retention epoch answers the gap shape, never a silent restart (migration 0003: store_epoch, consumer_checkpoints.epoch/carrier). seq, next_after_seq, latest_seq and X-OpenVibe-Seq are unchanged this release. test/cursor.test.js; answers validated against the contracts; npm test and test:pg 19/19 (verified).

OP OpenVibers committed 29 Sept 2026 · 2863954778 · 15 files, +388 −79





Changes
ddiff --git a/README.md b/README.md
iindex c1c6d6b..bcd947c 100644
---- a/README.md
++++ b/README.md
@@@ -112,7 +112,7 @@ Each delivery is `POST <endpoint>` with body `{ "event": <envelope>, "seq": n }`
  
  Operators: `GET /api/v1/deliveries?status=dead` is the dead-letter queue; `POST /api/v1/deliveries/replay { subscription_id, event_ids: [...] }` or `{ subscription_id, from_seq }` requeues retained events (that is also how a new subscription catches up on history).
  
--Pull consumers: `GET /api/v1/events?topic=media.vod.*&after_seq=<cursor>&limit=100` returns `{ events: [{ seq, event }], next_after_seq, latest_seq }`, plus `gap: { from_seq, to_seq }` when the cursor is older than retention. `PUT /api/v1/checkpoints { topic, cursor }` stores a consumer's cursor here if it has nowhere better.
++Pull consumers: `GET /api/v1/events?topic=media.vod.*&after_seq=<position>&limit=100` returns `{ events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq }`, plus `gap: { from_seq, to_seq }` when the position is older than retention or from another retention epoch. The opaque cursor ([ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 7) goes back in as `after=<cursor>` beside `after_seq`; `next_cursor` is the cursor for the next page, `seq`/`next_after_seq`/`latest_seq` stay for one release. `PUT /api/v1/checkpoints { topic, cursor }` stores a consumer's position here if it has nowhere better (an opaque cursor string is accepted in place of the integer; the answer carries its `epoch`, `carrier` and `next_cursor`).
  
  ## Developer apps
  
@@@ -198,19 +198,19 @@ re-exports them and keeps no SQLite copies.
  const es = new EventSource('https://events.openvibe.network/realtime/stream?topics=live.stream.*', { withCredentials: true });
  // A signed-in person's own events from any OpenVibe site: a fresh ticket from Network for every (re)connect.
  const { ticket, stream_url, topics } = await (await fetch('https://openvibe.network/api/v1/realtime/ticket', { method: 'POST', headers: { Authorization: `Bearer ${networkJwt}` } })).json();
--const mine = new EventSource(`${stream_url}?topics=${topics.join(',')}&ticket=${ticket}${lastSeq != null ? `&last_event_id=${lastSeq}` : ''}`);
--es.onmessage = (m) => { const { seq, event } = JSON.parse(m.data); };
++const mine = new EventSource(`${stream_url}?topics=${topics.join(',')}&ticket=${ticket}${lastEventId != null ? `&last_event_id=${lastEventId}` : ''}`);
++es.onmessage = (m) => { lastEventId = m.lastEventId; const { seq, event } = JSON.parse(m.data); };
  es.addEventListener('gap', (m) => { /* events were missed: refetch state */ });
  ```
  
  - Auth: a **realtime ticket** (`?ticket=`), the Network `ov_token` cookie or a Bearer user JWT; a service token with `events.event.read`; or nobody (public events only, `REALTIME_ALLOW_ANONYMOUS`). An expired cookie degrades to anonymous; a bad Bearer is a 401.
  - Realtime tickets (ADR-005 amendment 2): a page on any OpenVibe site cannot count on a cookie of events.openvibe.network (third-party there), and an EventSource cannot send a header. So it asks Network for a ticket (`POST https://openvibe.network/api/v1/realtime/ticket`, answering `network.realtime-ticket-result@1`) and opens `/realtime/stream?topics=network.notification.*&ticket=<ticket>` without credentials. The ticket is an RS256 JWT signed with Network's key (`identity.realtime-ticket-claims@1`): `iss <OV_NETWORK_ISSUER>/realtime`, `sub <usr_>`, `aud [openvibe.events]`, `typ` and `purpose` `realtime`, a lifetime of at most 300 s (Network mints 120 s) and `jti rtk_…`.
    - Events accepts each ticket once, never as Bearer or cookie, and never logs it. The refusals are 401: `ticket.invalid`, `ticket.expired` and `ticket.used`.
--  - A reconnect asks for a new ticket and resumes with `last_event_id`.
++  - A reconnect asks for a new ticket and resumes with `last_event_id` (the cursor from the last message's SSE `id`).
    - Conversely, a user JWT that carries `typ` or `purpose` (a ticket, a FedCM assertion) is not a session here.
  - A person's topic: `network.notification.created` (Network's outbox; visibility `subject`, subject the recipient). `topics=network.notification.*` streams a person's own notifications and nobody else's. There is no `user:<id>` topic: it is not a valid pattern (400 `realtime.bad_topic`), and subject visibility already does its job.
  - Visibility: `public` events go to anyone subscribed to the topic; `subject` events only to the user whose subject id (`usr_…`) is the event's `actor.id` or its user `subject.id`; `internal` events never reach a browser. A guessed topic yields nothing.
--- Resume: the SSE `id` is the seq, so the browser's automatic `Last-Event-ID` (or `?last_event_id=`) replays what was missed. A cursor older than retention first gets `event: gap` (`{ reason, from_seq, to_seq }`), as does a replay that hits `REALTIME_REPLAY_MAX`.
++- Resume: the SSE `id` is the cursor (ADR-042 decision 7; a bare seq is accepted for one release), so the browser's automatic `Last-Event-ID` (or `?last_event_id=`) replays what was missed. A cursor older than retention — or from another retention epoch (`reason: "epoch"`) — first gets `event: gap` (`{ reason, from_seq, to_seq }`), as does a replay that hits `REALTIME_REPLAY_MAX`.
  - Public replay window: browsers (signed out or signed in) are replayed `public` events received in the last `REALTIME_PUBLIC_REPLAY_SECONDS` (300; 0 = none), enough to ride out a reconnect. Older public events are not replayed: the stream opens with `event: gap` (`reason: "public_window"`, `window_seconds`), and the client refetches state from the owning service. `subject` events addressed to the viewer, and service viewers, keep the whole retention. This keeps the stream from paging through a month of public history, chat lines included; nothing in the network needs more (no browser surface replays public events today).
  - Redacted events are replayed as their tombstones ([Redaction](#redaction)).
  - Heartbeat comment every 25 s; at most 20 topics per connection and `REALTIME_MAX_CONNECTIONS` (2000) overall; CORS with credentials for `https://*.openvibe.*` only.
ddiff --git a/STATUS.json b/STATUS.json
iindex 9d0aa6a..6015248 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,7 +7,7 @@
    "runtime": "node 22, express, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
    "port": 4300,
    "domain": "events.openvibe.network",
--  "contracts": "openvibe-contracts v0.80.0",
++  "contracts": "openvibe-contracts v0.81.0",
    "capabilities": [
      "events.event.publish",
      "events.subscription.manage",
@@@ -20,7 +20,7 @@
    "capabilitiesRegistered": true,
    "deployed": true,
    "deployment": "Deployed with `sudo ovhost deploy events`: /opt/openvibe.events, unit openvibe-events, 127.0.0.1:4300; public at https://events.openvibe.network (realtime SSE, health, token-guarded /api/v1/events, /api/v1/subscriptions, /api/v1/checkpoints; /api/v1/deliveries host-local); real traffic since 2026-09-23 (Live, Media, Chat, Games, Network and OpenRe publish; on 2026-09-23 8 subscriptions, held by Search, Tips, VIP, News and Reviews); restore drill passed 2026-09-23",
--  "updated": "2026-09-28",
++  "updated": "2026-09-29",
    "features": [
      "durable event store with the canonical envelope",
      "subscriptions with signed webhook delivery (v1 and v2), retry, dead letters and replay",
ddiff --git a/lib/client.js b/lib/client.js
iindex 68ac4fb..d0f927f 100644
---- a/lib/client.js
++++ b/lib/client.js
@@@ -150,7 +150,8 @@ class PublishError extends Error {
  
  /**
   * createPublisher({ eventsUrl, tokenClient, fetchImpl?, timeoutMs? })
-- *   publish(envelope | [envelopes], { traceparent }) -> { event_id, seq, duplicate } | { results: [...] }
++ *   publish(envelope | [envelopes], { traceparent }) -> { event_id, seq, cursor, duplicate } | { results: [...] }
++ *   (`cursor` is the opaque position beside `seq`; ADR-042 decision 7. Hand it back as `after=` or `Last-Event-ID`.)
   * A 401 invalidates the cached token and retries once.
   */
  function createPublisher({ eventsUrl, tokenClient, fetchImpl = globalThis.fetch, timeoutMs = 10000 }) {
ddiff --git a/migrations/0003_cursors.sql b/migrations/0003_cursors.sql
nnew file mode 100644
iindex 0000000..a5ce429
---- /dev/null
++++ b/migrations/0003_cursors.sql
@@@ -0,0 +1,14 @@
++-- phase: expand
++-- ADR-042 decision 7: opaque cursors beside the global seq. A cursor names a position in the hot store — the global
++-- seq — together with the retention epoch the position belongs to. The epoch changes only when the hot store is
++-- truncated or re-imported (store.bumpEpoch), so a cursor from another epoch is answered with the existing gap shape
++-- instead of silently restarting a consumer. Checkpoints keep the numeric position in `cursor` and gain the epoch and
++-- carrier of the position they name.
++CREATE TABLE store_epoch (
++    name  text COLLATE "C" PRIMARY KEY,
++    value bigint NOT NULL
++);
++INSERT INTO store_epoch (name, value) VALUES ('events', 0) ON CONFLICT DO NOTHING;
++
++ALTER TABLE consumer_checkpoints ADD COLUMN epoch bigint NOT NULL DEFAULT 0;
++ALTER TABLE consumer_checkpoints ADD COLUMN carrier text COLLATE "C";
ddiff --git a/package-lock.json b/package-lock.json
iindex 013af04..4349d51 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,7 +12,7 @@
          "dotenv": "^16.4.7",
          "express": "^4.21.2",
          "iovalkey": "^0.4.0",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.80.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.81.0",
          "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.2.0",
          "pg": "^8.23.0"
@@@ -1102,9 +1102,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.80.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.80.0",
--      "integrity": "sha512-7RCNFKxFRey3PquBwyQkSu8Bu2q2Ek0yb248SJWaymcg/g9PrlStFVUDTLMgh8i7BZSaZSngj0nojpwYTLGYDQ==",
++      "version": "0.81.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.81.0",
++      "integrity": "sha512-nOM8WXdxHOHFxt777ST1PUbAwYCp86LNzIRcBHR5qtCRDaRvJnBvkstnZpgZ1eJ9buU2NgifatIFt8E0Y4Ua0g==",
        "license": "AGPL-3.0-only",
        "dependencies": {
          "ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex 4bbd6ae..3ef6a44 100644
---- a/package.json
++++ b/package.json
@@@ -24,7 +24,7 @@
      "dotenv": "^16.4.7",
      "express": "^4.21.2",
      "iovalkey": "^0.4.0",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.80.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.81.0",
      "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.2.0",
      "pg": "^8.23.0"
ddiff --git a/server/api/publish.js b/server/api/publish.js
iindex c1ee51a..1f79dde 100644
---- a/server/api/publish.js
++++ b/server/api/publish.js
@@@ -3,8 +3,11 @@
   * POST /api/v1/events  (service token, audience openvibe.events, capability events.event.publish;
   *                       or a developer app token with events.app.publish)
   *
-- *   body: <envelope>            -> 201 { event_id, seq, duplicate: false } | 200 on a repeat
-- *   body: { events: [...] }     -> 201 { results: [{ event_id, seq, duplicate }] } (<= 100, atomic)
++ *   body: <envelope>            -> 201 { event_id, seq, cursor, duplicate: false } | 200 on a repeat
++ *   body: { events: [...] }     -> 201 { results: [{ event_id, seq, cursor, duplicate }] } (<= 100, atomic)
++ *
++ * `cursor` (ADR-042 decision 7) is the opaque position beside seq; `seq` and the global-order promise
++ * stay for one release, then go.
   *
   * Every envelope must validate against events.event-envelope@1; `source` must be the calling
   * service (svc:live -> 'live'); `event_type` must start with a prefix that source owns. A
ddiff --git a/server/api/read.js b/server/api/read.js
iindex d80ff6d..1d1b9d8 100644
---- a/server/api/read.js
++++ b/server/api/read.js
@@@ -3,12 +3,16 @@
   * Pull consumers (service token with events.event.read):
   *
   *   GET /api/v1/events?topic=media.vod.*[,…]&after_seq=0&limit=100
-- *       -> { events: [{ seq, event }], next_after_seq, latest_seq, gap? }
-- *       `gap` ({ from_seq, to_seq }) means events after after_seq were already pruned by retention.
-- *       Keep next_after_seq as the cursor; it moves past events that did not match.
-- *   GET /api/v1/events/:event_id -> { seq, event }
++ *       -> { events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq, gap? }
++ *       `after=` takes an opaque cursor beside after_seq (ADR-042 decision 7); a cursor from another
++ *       retention epoch answers `gap` — never a silent restart. `gap` ({ from_seq, to_seq }) means
++ *       events after the position were already pruned by retention (or the epoch changed).
++ *       Keep next_cursor as the cursor; it moves past events that did not match.
++ *   GET /api/v1/events/:event_id -> { seq, cursor, event }
   *   GET /api/v1/checkpoints?topic=…  /  PUT /api/v1/checkpoints { topic, cursor }
-- *       a consumer's own stored cursor per topic pattern (consumer = calling principal)
++ *       a consumer's own stored cursor per topic pattern (consumer = calling principal). `cursor` is
++ *       the numeric position; PUT also accepts an opaque cursor string, and both answers carry the
++ *       position's epoch and carrier so the cursor can be rebuilt.
   *
   * Developer apps (events.app.read) use the same routes with an app token: only their project's
   * events in the token's environment plus public first-party events (server/apps.js); every topic
@@@ -24,6 +28,7 @@
  const express = require('express');
  const { http } = require('openvibe-contracts');
  const topics = require('../topics');
++const cursor = require('../cursor');
  const { rowToEnvelope } = require('../store');
  const { CAPS } = require('../auth');
  const apps = require('../apps');
@@@ -67,21 +72,37 @@ function readRouter({ store, auth, worker, limits }) {
          }
          const scopeErr = scopeError(req.principal, patterns);
          if (scopeErr) return http.sendProblem(res, 403, 'events.topic_not_allowed', { detail: scopeErr, ctx });
--        const after = intParam(req.query.after_seq, 0, 0, Number.MAX_SAFE_INTEGER);
          const limit = intParam(req.query.limit, 100, 1, 1000);
--        if (Number.isNaN(after) || Number.isNaN(limit)) {
--            return http.sendProblem(res, 400, 'events.bad_request', { detail: 'after_seq must be >= 0 and limit 1..1000', ctx });
++        if (Number.isNaN(limit)) {
++            return http.sendProblem(res, 400, 'events.bad_request', { detail: 'limit must be 1..1000', ctx });
++        }
++        const epoch = await store.epoch();
++        // `after=` is an opaque cursor; `after_seq` stays for one release. A cursor from another epoch cannot be
++        // mapped onto this store, so it is answered with a gap (never treated as a position here).
++        let after = intParam(req.query.after_seq, 0, 0, Number.MAX_SAFE_INTEGER);
++        let epochMismatch = false;
++        if (req.query.after !== undefined && String(req.query.after) !== '') {
++            const decoded = cursor.decode(String(req.query.after));
++            if (!decoded) return http.sendProblem(res, 400, 'events.bad_request', { detail: 'after must be an opaque cursor', ctx });
++            after = decoded.seq;
++            epochMismatch = decoded.epoch !== epoch;
++        } else if (Number.isNaN(after)) {
++            return http.sendProblem(res, 400, 'events.bad_request', { detail: 'after_seq must be >= 0', ctx });
          }
          const oldest = await store.oldestSeq();
          const out = {};
          let from = after;
--        if (after < oldest - 1) {
++        if (epochMismatch) {
++            out.gap = { from_seq: 1, to_seq: Math.max(oldest - 1, 0) };
++            from = Math.max(oldest - 1, 0);
++        } else if (after < oldest - 1) {
              out.gap = { from_seq: after + 1, to_seq: oldest - 1 };
              from = oldest - 1;
          }
--        const { rows, cursor } = await store.scan(from, { patterns, limit, accept: acceptFor(req.principal, patterns) });
--        out.events = rows.map(r => ({ seq: r.seq, event: rowToEnvelope(r) }));
--        out.next_after_seq = cursor;
++        const { rows, cursor: scanned } = await store.scan(from, { patterns, limit, accept: acceptFor(req.principal, patterns) });
++        out.events = rows.map(r => ({ seq: r.seq, cursor: cursor.encode(r.seq, epoch), event: rowToEnvelope(r) }));
++        out.next_after_seq = scanned;
++        out.next_cursor = cursor.encode(scanned, epoch);
          out.latest_seq = await store.lastSeq();
          res.json(out);
      });
@@@ -90,7 +111,7 @@ function readRouter({ store, auth, worker, limits }) {
          const row = await store.getEvent(String(req.params.id));
          const visible = row && (req.principal.kind === 'app' ? apps.visibleToApp(row, req.principal) : (row.env || 'production') === 'production');
          if (!visible) return http.sendProblem(res, 404, 'events.not_found', { detail: 'no such event (or pruned by retention)', ctx: req.ov });
--        return res.json({ seq: row.seq, event: rowToEnvelope(row) });
++        return res.json({ seq: row.seq, cursor: cursor.encode(row.seq, await store.epoch()), event: rowToEnvelope(row) });
      });
  
      const consumerOf = (req) => req.principal.service || req.principal.sub;
@@@ -101,18 +122,36 @@ function readRouter({ store, auth, worker, limits }) {
          const scopeErr = scopeError(req.principal, [topic]);
          if (scopeErr) return http.sendProblem(res, 403, 'events.topic_not_allowed', { detail: scopeErr, ctx: req.ov });
          const cp = await store.getCheckpoint(consumerOf(req), topic);
--        res.json({ consumer: consumerOf(req), topic, cursor: cp ? cp.cursor : 0, updated_at: cp ? cp.updated_at : null });
++        const epoch = cp ? cp.epoch : await store.epoch();
++        const position = cp ? cp.cursor : 0;
++        res.json({
++            consumer: consumerOf(req), topic, cursor: position, epoch, carrier: cp ? cp.carrier : null,
++            next_cursor: cursor.encode(position, epoch), updated_at: cp ? cp.updated_at : null,
++        });
      });
  
      router.put('/api/v1/checkpoints', canRead, limits('events.checkpoint.write', { minute: 600, hour: 20000 }), async (req, res) => {
          const b = req.body || {};
--        if (!topics.isValidPattern(b.topic) || !Number.isInteger(b.cursor) || b.cursor < 0) {
--            return http.sendProblem(res, 400, 'events.bad_request', { detail: 'topic (pattern) and cursor (integer >= 0) are required', ctx: req.ov });
++        // `cursor` stays the numeric position; an opaque cursor string is accepted in its place and decoded to the
++        // position and epoch it names (ADR-042 decision 7). `carrier` records the carrier the position is on.
++        let position = null;
++        let atEpoch = null;
++        if (typeof b.cursor === 'string') {
++            const decoded = cursor.decode(b.cursor);
++            if (!decoded) return http.sendProblem(res, 400, 'events.bad_request', { detail: 'cursor must be an opaque cursor or an integer >= 0', ctx: req.ov });
++            position = decoded.seq;
++            atEpoch = decoded.epoch;
++        } else if (Number.isInteger(b.cursor) && b.cursor >= 0) {
++            position = b.cursor;
++            if (Number.isInteger(b.epoch) && b.epoch >= 0) atEpoch = b.epoch;
++        }
++        if (!topics.isValidPattern(b.topic) || position === null) {
++            return http.sendProblem(res, 400, 'events.bad_request', { detail: 'topic (pattern) and cursor (integer >= 0 or an opaque cursor) are required', ctx: req.ov });
          }
          const scopeErr = scopeError(req.principal, [b.topic]);
          if (scopeErr) return http.sendProblem(res, 403, 'events.topic_not_allowed', { detail: scopeErr, ctx: req.ov });
--        const cp = await store.setCheckpoint(consumerOf(req), b.topic, b.cursor);
--        res.json({ consumer: consumerOf(req), topic: b.topic, ...cp });
++        const cp = await store.setCheckpoint(consumerOf(req), b.topic, position, { epoch: atEpoch, carrier: typeof b.carrier === 'string' && b.carrier ? b.carrier : null });
++        res.json({ consumer: consumerOf(req), topic: b.topic, ...cp, next_cursor: cursor.encode(cp.cursor, cp.epoch) });
      });
  
      router.get('/api/v1/deliveries', isAdmin, limits('events.delivery.list'), async (req, res) => {
ddiff --git a/server/cursor.js b/server/cursor.js
nnew file mode 100644
iindex 0000000..8cb6d1c
---- /dev/null
++++ b/server/cursor.js
@@@ -0,0 +1,36 @@
++'use strict';
++/**
++ * Opaque cursors (ADR-042 decision 7): a cursor names a position in the hot store — the global seq — together with
++ * the retention epoch that position belongs to. The wire form is `c1.<epoch>.<base64url(seq)>`; it is opaque to
++ * clients, which hand it back as `after=` or `Last-Event-ID` and never parse it.
++ *
++ * decode() never throws on user input: anything malformed, non-canonical or of another version is null, and the
++ * caller answers with its own refusal or gap shape. encode() is strict — it is only ever called with a seq the
++ * store just handed out.
++ */
++const VERSION = 'c1';
++const MAX_LENGTH = 200;   // the contract caps a cursor at 200 characters
++
++/** The cursor for `seq` in `epoch`. The epoch is a small integer, so the string stays short. */
++function encode(seq, epoch = 0) {
++    if (!Number.isSafeInteger(seq) || seq < 0) throw new TypeError('seq must be a safe integer >= 0');
++    if (!Number.isInteger(epoch) || epoch < 0) throw new TypeError('epoch must be an integer >= 0');
++    return `${VERSION}.${epoch}.${Buffer.from(String(seq), 'utf8').toString('base64url')}`;
++}
++
++/** { seq, epoch } for a cursor, or null for anything that is not one. Never throws. */
++function decode(input) {
++    if (typeof input !== 'string' || input.length === 0 || input.length > MAX_LENGTH) return null;
++    const parts = input.split('.');
++    if (parts.length !== 3 || parts[0] !== VERSION) return null;
++    if (!/^\d{1,15}$/.test(parts[1]) || !/^[A-Za-z0-9_-]+$/.test(parts[2])) return null;
++    const epoch = Number(parts[1]);
++    let seq;
++    try { seq = Number(Buffer.from(parts[2], 'base64url').toString('utf8')); } catch { return null; }
++    if (!Number.isSafeInteger(seq) || seq < 0) return null;
++    // Re-encoding rejects non-canonical base64 and leading zeros, so one position has exactly one cursor string.
++    if (encode(seq, epoch) !== input) return null;
++    return { seq, epoch };
++}
++
++module.exports = { encode, decode, VERSION };
ddiff --git a/server/realtime.js b/server/realtime.js
iindex a1c4b88..36c9845 100644
---- a/server/realtime.js
++++ b/server/realtime.js
@@@ -8,8 +8,8 @@
   * minted by Network for the signed-in person: how pages on every OpenVibe site open a stream without a
   * third-party cookie), a Bearer user or service token, the ov_token cookie, or nobody.
   *
-- * Every message is `id: <seq>` + `data: {"seq":n,"event":<envelope>}`. Visibility decides who sees
-- * an event:
++ * Every message is `id: <cursor>` (ADR-042 decision 7; a bare seq is still accepted for one release) +
++ * `data: {"seq":n,"event":<envelope>}`. Visibility decides who sees an event:
   *   public    anyone subscribed to a matching topic (signed-out visitors too, unless disabled)
   *   subject   only the user whose subject id is the event's actor.id or its subject.id (subject
   *             type user); a guessed topic yields nothing for anyone else. A person's topic is an
@@@ -18,9 +18,10 @@
   *   internal  service principals (token with events.event.read) only, never a browser
   * Developer-app events (app.<project_key>.*, events.app.publish) are never streamed here, to anyone.
   *
-- * Resume: `Last-Event-ID` (or ?last_event_id=) is a seq. Missed events are replayed first; when the
-- * cursor is older than retention an `event: gap` message comes first, so the client knows to
-- * refetch state. Heartbeat comment every 25 s.
++ * Resume: `Last-Event-ID` (or ?last_event_id=) is a cursor, or a bare seq for one release. Missed
++ * events are replayed first; when the cursor is older than retention — or belongs to another
++ * retention epoch — an `event: gap` message comes first, so the client knows to refetch state.
++ * Heartbeat comment every 25 s.
   *
   * Public replay window: a browser (signed out or signed in) is replayed `public` events received in
   * the last REALTIME_PUBLIC_REPLAY_SECONDS (300) only, enough to ride out a reconnect. Anything
@@@ -31,6 +32,7 @@
   */
  const { http } = require('openvibe-contracts');
  const topics = require('./topics');
++const cursor = require('./cursor');
  const { rowToEnvelope } = require('./store');
  
  const ORIGIN_RE = /^https:\/\/(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*openvibe\.[a-z]{2,63}$/;
@@@ -87,7 +89,7 @@ function createRealtime({ store, auth, config, clock = { now: () => Date.now() }
  
      function sendEvent(conn, row) {
          conn.lastSeq = row.seq;
--        write(conn, `id: ${row.seq}\ndata: ${JSON.stringify({ seq: row.seq, event: rowToEnvelope(row) })}\n\n`);
++        write(conn, `id: ${cursor.encode(row.seq, conn.epoch)}\ndata: ${JSON.stringify({ seq: row.seq, event: rowToEnvelope(row) })}\n\n`);
      }
  
      function sendGap(conn, gap) {
@@@ -126,8 +128,19 @@ function createRealtime({ store, auth, config, clock = { now: () => Date.now() }
              return http.sendProblem(res, 503, 'realtime.over_capacity', { detail: 'too many realtime connections, retry shortly', ctx });
          }
  
++        const epoch = await store.epoch();
          const rawLast = req.headers['last-event-id'] ?? req.query.last_event_id;
--        const lastId = rawLast != null && /^\d{1,15}$/.test(String(rawLast).trim()) ? Number(String(rawLast).trim()) : null;
++        // A cursor, or — for one release — a bare seq. Anything else is ignored, as before.
++        let lastId = null;
++        let epochMismatch = false;
++        if (rawLast != null) {
++            const raw = String(rawLast).trim();
++            if (/^\d{1,15}$/.test(raw)) lastId = Number(raw);
++            else {
++                const decoded = cursor.decode(raw);
++                if (decoded) { lastId = decoded.seq; epochMismatch = decoded.epoch !== epoch; }
++            }
++        }
  
          res.status(200);
          res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
@@@ -140,7 +153,7 @@ function createRealtime({ store, auth, config, clock = { now: () => Date.now() }
  
          // The connection takes its slot and joins the fan-out before the first await, so the cap holds and nothing
          // published during the catch-up is lost: those rows wait in `pending` and follow the replay, skipping what it sent.
--        const conn = { res, viewer, patterns: wanted, lastSeq: lastId, pending: [], closed: false };
++        const conn = { res, viewer, patterns: wanted, lastSeq: lastId, epoch, pending: [], closed: false };
          conns.add(conn);
          res.on('close', () => close(conn));
          write(conn, `retry: 3000\n: connected ${viewer.kind}\n\n`);
@@@ -149,34 +162,37 @@ function createRealtime({ store, auth, config, clock = { now: () => Date.now() }
              if (lastId != null) {
                  const latest = await store.lastSeq();
                  const oldest = await store.oldestSeq();
--                if (lastId > latest) {
++                let at = lastId;   // where the replay starts
++                if (epochMismatch) {
++                    // The hot store was truncated or re-imported: the cursor's position belongs to a previous epoch
++                    // and cannot be mapped. The gap is the signal; the retained events then replay.
++                    sendGap(conn, { reason: 'epoch', from_seq: 1, to_seq: Math.max(oldest - 1, 0), latest_seq: latest });
++                    at = Math.max(oldest - 1, 0);
++                } else if (lastId > latest) {
                      sendGap(conn, { reason: 'cursor_ahead', from_seq: latest + 1, to_seq: lastId, latest_seq: latest });
--                    conn.lastSeq = latest;
--                } else {
--                    let cursor = lastId;
--                    if (lastId < oldest - 1) {
--                        sendGap(conn, { reason: 'retention', from_seq: lastId + 1, to_seq: oldest - 1, latest_seq: latest });
--                        cursor = oldest - 1;
--                    }
--                    // Browsers: public events older than the window are not replayed (a gap says so).
--                    const publicFrom = viewer.kind === 'service' ? 0
--                        : opts.publicReplaySeconds > 0 ? await store.firstSeqSince(clock.now() - opts.publicReplaySeconds * 1000) : latest + 1;
--                    if (cursor < publicFrom - 1) {
--                        sendGap(conn, { reason: 'public_window', from_seq: cursor + 1, to_seq: publicFrom - 1, latest_seq: latest, window_seconds: opts.publicReplaySeconds });
--                        // Signed out, only public events are visible: nothing before the window to scan.
--                        if (viewer.kind === 'anonymous') cursor = publicFrom - 1;
--                    }
--                    const accept = (row) => visibleTo(viewer, row) && !(row.visibility === 'public' && row.seq < publicFrom);
--                    const { rows, cursor: scanned } = await store.scan(cursor, {
--                        patterns: wanted, limit: opts.replayMax, scanMax: opts.replayMax * 50, accept,
--                    });
--                    for (const row of rows) sendEvent(conn, row);
--                    if (scanned < latest) {
--                        // Replay stopped at its limit: everything after `scanned` is reported as a gap.
--                        sendGap(conn, { reason: 'replay_limit', from_seq: scanned + 1, to_seq: latest, latest_seq: latest });
--                    }
--                    conn.lastSeq = latest;
++                    at = latest;
++                } else if (lastId < oldest - 1) {
++                    sendGap(conn, { reason: 'retention', from_seq: lastId + 1, to_seq: oldest - 1, latest_seq: latest });
++                    at = oldest - 1;
++                }
++                // Browsers: public events older than the window are not replayed (a gap says so).
++                const publicFrom = viewer.kind === 'service' ? 0
++                    : opts.publicReplaySeconds > 0 ? await store.firstSeqSince(clock.now() - opts.publicReplaySeconds * 1000) : latest + 1;
++                if (at < publicFrom - 1) {
++                    sendGap(conn, { reason: 'public_window', from_seq: at + 1, to_seq: publicFrom - 1, latest_seq: latest, window_seconds: opts.publicReplaySeconds });
++                    // Signed out, only public events are visible: nothing before the window to scan.
++                    if (viewer.kind === 'anonymous') at = publicFrom - 1;
++                }
++                const accept = (row) => visibleTo(viewer, row) && !(row.visibility === 'public' && row.seq < publicFrom);
++                const { rows, cursor: scanned } = await store.scan(at, {
++                    patterns: wanted, limit: opts.replayMax, scanMax: opts.replayMax * 50, accept,
++                });
++                for (const row of rows) sendEvent(conn, row);
++                if (scanned < latest) {
++                    // Replay stopped at its limit: everything after `scanned` is reported as a gap.
++                    sendGap(conn, { reason: 'replay_limit', from_seq: scanned + 1, to_seq: latest, latest_seq: latest });
                  }
++                conn.lastSeq = latest;
              }
          } catch (err) {
              // The headers are sent: a failed catch-up can only end the stream (the client reconnects with its Last-Event-ID).
ddiff --git a/server/store.js b/server/store.js
iindex 052e6b6..5f07a58 100644
---- a/server/store.js
++++ b/server/store.js
@@@ -3,7 +3,8 @@
   * The durable store (PostgreSQL through openvibe-sdk/db). An event is committed, together with one
   * delivery row per matching subscription, before anything is sent anywhere: "event persists before
   * consumer delivery". Global order is `seq`, handed out from a counter that never goes backwards,
-- * even after retention has pruned the newest rows.
++ * even after retention has pruned the newest rows. Beside it a cursor (./cursor.js) names a seq plus
++ * the retention epoch — ADR-042 decision 7; seq stays on the wire for one release.
   */
  const fs = require('fs');
  const path = require('path');
@@@ -11,6 +12,7 @@ const { createDb } = require('openvibe-sdk/db');
  const topics = require('./topics');
  const apps = require('./apps');
  const redaction = require('./redaction');
++const cursor = require('./cursor');
  const { createUsage, deliveryCode } = require('./usage');
  
  const PRIORITY_RANK = { critical: 0, important: 1, low: 2 };
@@@ -99,6 +101,8 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
          nextSeq: db.prepare("UPDATE sequences SET value = value + 1 WHERE name = 'events' RETURNING value"),
          lastSeq: db.prepare("SELECT value FROM sequences WHERE name = 'events'"),
          minSeq: db.prepare('SELECT MIN(seq) AS s FROM events'),
++        epoch: db.prepare("SELECT value FROM store_epoch WHERE name = 'events'"),
++        bumpEpoch: db.prepare("UPDATE store_epoch SET value = value + 1 WHERE name = 'events' RETURNING value"),
          // Hop depth counts only rows the publisher's own tenancy can have caused: a first-party
          // publish ignores app events (an app that saw a public trace_id cannot poison that trace), an
          // app publish counts first-party rows plus its own project and environment.
@@@ -144,6 +148,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
       */
      const insertBatch = async (items, { publisher, requestId, project = null }) => await db.tx(async () => {
          const now = clock.now();
++        const epoch = (await q.epoch.get()).value;   // the retention epoch the stored positions belong to (ADR-042)
          const results = [];
          const inserted = [];
          const subs = await q.enabledSubs.all();
@@@ -159,7 +164,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
                  if (existing.source !== env.source || existing.event_type !== env.event_type) {
                      throw new StoreError(409, 'events.id_conflict', `event ${env.event_id} already exists with different content`);
                  }
--                results.push({ event_id: env.event_id, seq: existing.seq, duplicate: true });
++                results.push({ event_id: env.event_id, seq: existing.seq, duplicate: true, cursor: cursor.encode(existing.seq, epoch) });
                  continue;
              }
              if (await q.getReceipt.get(PUBLISH_RECEIPT, env.event_id)) {
@@@ -230,7 +235,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
                  }
              }
              await revocationHook(row, now);
--            results.push({ event_id: env.event_id, seq, duplicate: false });
++            results.push({ event_id: env.event_id, seq, duplicate: false, cursor: cursor.encode(seq, epoch) });
              inserted.push(row);
              if (directive) {
                  const redacted = await applyRedaction(row, directive, now);
@@@ -339,6 +344,17 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
          return m == null ? await lastSeq() + 1 : m;
      }
  
++    /** The retention epoch of the hot store (ADR-042 decision 7): a cursor carries it so a position from a previous
++     *  epoch — after a truncate or a re-import — is answered with a gap, never used as if it were current. */
++    async function epoch() {
++        return (await q.epoch.get()).value;
++    }
++
++    /** Advance the retention epoch. Called only when the hot store is truncated or re-imported. Returns the new epoch. */
++    async function bumpEpoch() {
++        return (await q.bumpEpoch.get()).value;
++    }
++
      /**
       * Events with seq > afterSeq that match any of `patterns`, in order, filtered by `accept(row)`.
       * Scans at most `scanMax` rows; returns { rows, cursor } where cursor is the last seq examined
@@@ -535,14 +551,20 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
  
      // ── Checkpoints ──────────────────────────────────────────
  
++    /**
++     * A checkpoint is a cursor in parts: `cursor` is the numeric position, `epoch` the retention epoch it belongs to
++     * and `carrier` the carrier that position is on (ADR-042 decision 7). `cursor` stays numeric on the wire.
++     */
      async function getCheckpoint(consumer, topicPattern) {
--        const r = await db.prepare('SELECT cursor, updated_at FROM consumer_checkpoints WHERE consumer = ? AND topic_pattern = ?').get(consumer, topicPattern);
--        return r ? { cursor: r.cursor, updated_at: new Date(r.updated_at).toISOString() } : null;
++        const r = await db.prepare('SELECT cursor, epoch, carrier, updated_at FROM consumer_checkpoints WHERE consumer = ? AND topic_pattern = ?').get(consumer, topicPattern);
++        return r ? { cursor: r.cursor, epoch: r.epoch, carrier: r.carrier || null, updated_at: new Date(r.updated_at).toISOString() } : null;
      }
--    async function setCheckpoint(consumer, topicPattern, cursor) {
--        await db.prepare(`INSERT INTO consumer_checkpoints (consumer, topic_pattern, cursor, updated_at) VALUES (?, ?, ?, ?)
--            ON CONFLICT(consumer, topic_pattern) DO UPDATE SET cursor = excluded.cursor, updated_at = excluded.updated_at`)
--            .run(consumer, topicPattern, cursor, clock.now());
++    /** Store a position. `epoch` defaults to the current retention epoch (a legacy numeric cursor). */
++    async function setCheckpoint(consumer, topicPattern, position, { epoch: atEpoch = null, carrier = null } = {}) {
++        const at = atEpoch == null ? await epoch() : atEpoch;
++        await db.prepare(`INSERT INTO consumer_checkpoints (consumer, topic_pattern, cursor, epoch, carrier, updated_at) VALUES (?, ?, ?, ?, ?, ?)
++            ON CONFLICT(consumer, topic_pattern) DO UPDATE SET cursor = excluded.cursor, epoch = excluded.epoch, carrier = excluded.carrier, updated_at = excluded.updated_at`)
++            .run(consumer, topicPattern, position, at, carrier, clock.now());
          return await getCheckpoint(consumer, topicPattern);
      }
  
@@@ -567,7 +589,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
      }
  
      return {
--        db, insertBatch, redact, getEvent, revokedAt, lastSeq, oldestSeq, firstSeqSince, scan,
++        db, insertBatch, redact, getEvent, revokedAt, lastSeq, oldestSeq, epoch, bumpEpoch, firstSeqSince, scan,
          createSubscription, getSubscription, listSubscriptions, countSubscriptions, countProjectSubscriptions, setSubscriptionEnabled, rotateSubscriptionSecret,
          claimDeliveries, recordAttempt, getDelivery, listDeliveries, requeue, deliveryCounts,
          getCheckpoint, setCheckpoint, prune, ping, usage, flushUsage,
ddiff --git a/test/cursor.test.js b/test/cursor.test.js
nnew file mode 100644
iindex 0000000..c587f2f
---- /dev/null
++++ b/test/cursor.test.js
@@@ -0,0 +1,169 @@
++'use strict';
++/**
++ * ADR-042 decision 7: opaque cursors beside the global seq. A cursor is `c1.<epoch>.<base64url(seq)>`;
++ * every answer that carries a seq also carries its cursor, pull accepts `after=<cursor>` beside
++ * `after_seq`, the SSE `id` is the cursor (a bare seq is accepted for one release), and checkpoints
++ * store and return the position's epoch. The contract answers are validated against the pinned
++ * openvibe-contracts (0.81.0, the release that added the cursor fields).
++ */
++const assert = require('assert');
++const { validate } = require('openvibe-contracts');
++const cursor = require('../server/cursor');
++const { boot, request, serviceToken, envelope, sse, suite } = require('./helpers');
++
++const t = suite('cursor');
++let h;
++const live = serviceToken('live', ['events.event.publish']);
++const reader = serviceToken('games', ['events.event.read']);
++const readerSvc = serviceToken('search', ['events.event.read']);
++
++const publish = async (over = {}) => {
++    const r = await request(h.base, 'POST', '/api/v1/events', { token: live, body: envelope('live', { event_type: 'live.vod.ready', ...over }) });
++    assert.strictEqual(r.status, 201, r.text);
++    return r.body;
++};
++
++t('boot', async () => { h = await boot(); });
++
++t('encode/decode round trip (never throws on malformed input)', () => {
++    for (const [seq, epoch] of [[0, 0], [1, 0], [42, 7], [Number.MAX_SAFE_INTEGER, 3]]) {
++        const c = cursor.encode(seq, epoch);
++        assert.match(c, /^c1\.\d+\.[A-Za-z0-9_-]+$/, c);
++        assert.deepStrictEqual(cursor.decode(c), { seq, epoch });
++    }
++    assert.strictEqual(cursor.encode(42, 0), `c1.0.${Buffer.from('42').toString('base64url')}`);
++    for (const bad of [null, undefined, 42, {}, [], '', 'c1', 'c1.0', 'c1.0.MQ.MQ', 'c2.0.MQ', 'c1.x.MQ', 'c1.-1.MQ',
++        'c1.0.!!!', 'c1.0.MA==', 'c1.00.MQ', 'c1.0.' + 'A'.repeat(300), 'c1.99999999999999999999.MQ',
++        'c1.0.' + Buffer.from('01').toString('base64url')]) {   // the last one is non-canonical for seq 1
++        assert.strictEqual(cursor.decode(bad), null, JSON.stringify(bad));
++    }
++});
++
++t('publish and read answers carry cursors, validated against events.publish-result@1 / read-result@1', async () => {
++    const epoch = await h.store.epoch();
++    const one = await publish();
++    const pv = validate('events.publish-result@1', one);
++    assert.ok(pv.valid, JSON.stringify(pv.errors));
++    assert.deepStrictEqual(cursor.decode(one.cursor), { seq: one.seq, epoch });
++
++    const batch = await request(h.base, 'POST', '/api/v1/events', { token: live, body: { events: [envelope('live', { event_type: 'live.vod.ready' }), envelope('live', { event_type: 'live.vod.ready' })] } });
++    assert.ok(validate('events.publish-result@1', batch.body).valid);
++    for (const r of batch.body.results) assert.deepStrictEqual(cursor.decode(r.cursor), { seq: r.seq, epoch });
++
++    const page = await request(h.base, 'GET', '/api/v1/events?topic=live.vod.*', { token: reader });
++    const rv = validate('events.read-result@1', page.body);
++    assert.ok(rv.valid, JSON.stringify(rv.errors));
++    assert.ok(page.body.events.length >= 3);
++    for (const e of page.body.events) assert.deepStrictEqual(cursor.decode(e.cursor), { seq: e.seq, epoch });
++    assert.deepStrictEqual(cursor.decode(page.body.next_cursor), { seq: page.body.next_after_seq, epoch });
++
++    const single = await request(h.base, 'GET', `/api/v1/events/${one.event_id}`, { token: reader });
++    assert.ok(validate('events.read-result@1', single.body).valid);
++    assert.deepStrictEqual(cursor.decode(single.body.cursor), { seq: single.body.seq, epoch });
++});
++
++t('pull accepts after=<cursor> beside after_seq; next_cursor round-trips a page', async () => {
++    let r = await request(h.base, 'GET', '/api/v1/events?topic=live.vod.*&limit=1', { token: reader });
++    assert.strictEqual(r.body.events.length, 1);
++    const first = r.body.events[0];
++    assert.strictEqual(r.body.next_after_seq, first.seq);
++    r = await request(h.base, 'GET', `/api/v1/events?topic=live.vod.*&after=${encodeURIComponent(r.body.next_cursor)}`, { token: reader });
++    assert.strictEqual(r.status, 200, r.text);
++    assert.ok(r.body.events.length >= 1 && r.body.events.every(e => e.seq > first.seq));
++    // after_seq still works, unchanged.
++    r = await request(h.base, 'GET', `/api/v1/events?topic=live.vod.*&after_seq=${first.seq}`, { token: reader });
++    assert.ok(r.body.events.every(e => e.seq > first.seq));
++});
++
++t('a malformed after= is refused, not treated as a position', async () => {
++    for (const bad of ['not-a-cursor', 'c1.0!!!', 'c2.0.MQ']) {
++        const r = await request(h.base, 'GET', `/api/v1/events?after=${encodeURIComponent(bad)}`, { token: reader });
++        assert.strictEqual(r.status, 400, bad);
++        assert.strictEqual(r.body.code, 'events.bad_request');
++    }
++});
++
++t('a cursor from another epoch answers the gap shape, never a silent restart', async () => {
++    const before = await request(h.base, 'GET', '/api/v1/events?topic=live.vod.*', { token: reader });
++    const stale = before.body.next_cursor;
++    const olderEpoch = cursor.decode(stale).epoch;
++    await h.store.bumpEpoch();
++    assert.strictEqual(await h.store.epoch(), olderEpoch + 1);
++    const r = await request(h.base, 'GET', `/api/v1/events?topic=live.vod.*&after=${encodeURIComponent(stale)}`, { token: reader });
++    assert.strictEqual(r.status, 200);
++    assert.ok(r.body.gap, JSON.stringify(r.body));
++    assert.ok(Number.isInteger(r.body.gap.from_seq) && Number.isInteger(r.body.gap.to_seq));
++    assert.ok(r.body.events.length >= 1, 'the retained events are still served after the gap');
++    assert.strictEqual(cursor.decode(r.body.next_cursor).epoch, olderEpoch + 1, 'position moves in the new epoch');
++});
++
++t('SSE id is the cursor; Last-Event-ID resumes from a cursor, and a bare seq still works', async () => {
++    const epoch = await h.store.epoch();
++    const c1 = await sse(h.base, '/realtime/stream?topics=live.stream.*', { headers: { Authorization: `Bearer ${readerSvc}` } });
++    const s1 = await publish({ visibility: 'public', event_type: 'live.stream.a' });
++    await c1.waitFor(x => x.events().length === 1);
++    const msg = c1.messages.find(m => !m.event);
++    assert.deepStrictEqual(cursor.decode(msg.id), { seq: s1.seq, epoch });
++    c1.close();
++
++    const s2 = await publish({ visibility: 'public', event_type: 'live.stream.b' });
++    const byCursor = await sse(h.base, `/realtime/stream?topics=live.stream.*&last_event_id=${encodeURIComponent(msg.id)}`, { headers: { Authorization: `Bearer ${readerSvc}` } });
++    await byCursor.waitFor(x => x.events().length === 1);
++    assert.deepStrictEqual(byCursor.events().map(e => e.seq), [s2.seq]);
++    assert.deepStrictEqual(byCursor.gaps(), []);
++    byCursor.close();
++
++    // A bare seq is still accepted for this release.
++    const bySeq = await sse(h.base, `/realtime/stream?topics=live.stream.*&last_event_id=${s2.seq}`, { headers: { Authorization: `Bearer ${readerSvc}` } });
++    const s3 = await publish({ visibility: 'public', event_type: 'live.stream.c' });
++    await bySeq.waitFor(x => x.events().length === 1);
++    assert.deepStrictEqual(bySeq.events().map(e => e.seq), [s3.seq]);
++    bySeq.close();
++});
++
++t('SSE reports a gap when the cursor belongs to another epoch', async () => {
++    const c = await sse(h.base, '/realtime/stream?topics=live.stream.*', { headers: { Authorization: `Bearer ${readerSvc}` } });
++    const s = await publish({ visibility: 'public', event_type: 'live.stream.d' });
++    await c.waitFor(x => x.events().length === 1);
++    const stale = c.messages.find(m => !m.event).id;
++    c.close();
++    await h.store.bumpEpoch();
++    const again = await sse(h.base, `/realtime/stream?topics=live.stream.*&last_event_id=${encodeURIComponent(stale)}`, { headers: { Authorization: `Bearer ${readerSvc}` } });
++    await again.waitFor(x => x.gaps().length >= 1);
++    assert.strictEqual(again.gaps()[0].reason, 'epoch');
++    // The retained matching event is replayed after the gap.
++    await again.waitFor(x => x.events().some(e => e.seq === s.seq));
++    again.close();
++});
++
++t('checkpoint round trip: an opaque cursor in, the position and epoch out', async () => {
++    const pub = await publish();
++    const epoch = cursor.decode(pub.cursor).epoch;
++    let r = await request(h.base, 'PUT', '/api/v1/checkpoints', { token: reader, body: { topic: 'live.vod.*', cursor: pub.cursor } });
++    assert.strictEqual(r.status, 200, r.text);
++    assert.ok(validate('events.read-result@1', r.body).valid, JSON.stringify(r.body));
++    assert.strictEqual(r.body.cursor, pub.seq, 'cursor stays the numeric position');
++    assert.strictEqual(r.body.epoch, epoch);
++    assert.strictEqual(r.body.next_cursor, pub.cursor, 'the opaque cursor comes back, ready for after=');
++
++    r = await request(h.base, 'GET', '/api/v1/checkpoints?topic=live.vod.*', { token: reader });
++    assert.strictEqual(r.body.cursor, pub.seq);
++    assert.strictEqual(r.body.epoch, epoch);
++    assert.strictEqual(r.body.next_cursor, pub.cursor);
++    assert.strictEqual(cursor.encode(r.body.cursor, r.body.epoch), pub.cursor);
++
++    // A plain integer still works and takes the current epoch.
++    r = await request(h.base, 'PUT', '/api/v1/checkpoints', { token: reader, body: { topic: 'live.vod.*', cursor: 5 } });
++    assert.strictEqual(r.body.cursor, 5);
++    assert.strictEqual(r.body.epoch, await h.store.epoch());
++    assert.deepStrictEqual(cursor.decode(r.body.next_cursor), { seq: 5, epoch: await h.store.epoch() });
++
++    // A malformed cursor string is refused.
++    r = await request(h.base, 'PUT', '/api/v1/checkpoints', { token: reader, body: { topic: 'live.vod.*', cursor: 'nope' } });
++    assert.strictEqual(r.status, 400);
++    assert.strictEqual(r.body.code, 'events.bad_request');
++});
++
++t('stop', async () => { await h.stop(); });
++
++t.run();
ddiff --git a/test/publish.test.js b/test/publish.test.js
iindex c27cdee..c01c1ff 100644
---- a/test/publish.test.js
++++ b/test/publish.test.js
@@@ -1,6 +1,7 @@
  'use strict';
  const assert = require('assert');
--const { ids } = require('openvibe-contracts');
++const { ids, validate } = require('openvibe-contracts');
++const cursor = require('../server/cursor');
  const { boot, request, serviceToken, appToken, envelope, suite } = require('./helpers');
  
  const t = suite('publish');
@@@ -35,6 +36,8 @@ t('valid envelope -> 201 with a seq; defaults filled', async () => {
      assert.strictEqual(r.body.event_id, env.event_id);
      assert.ok(Number.isInteger(r.body.seq) && r.body.seq > 0);
      assert.strictEqual(r.body.duplicate, false);
++    assert.ok(validate('events.publish-result@1', r.body).valid, JSON.stringify(validate('events.publish-result@1', r.body).errors));
++    assert.deepStrictEqual(cursor.decode(r.body.cursor), { seq: r.body.seq, epoch: await h.store.epoch() });
      const row = await h.store.getEvent(env.event_id);
      assert.strictEqual(row.priority, 'important');
      assert.strictEqual(row.visibility, 'internal');
ddiff --git a/test/pull-health.test.js b/test/pull-health.test.js
iindex eb85700..2d794a1 100644
---- a/test/pull-health.test.js
++++ b/test/pull-health.test.js
@@@ -2,6 +2,8 @@
  const assert = require('assert');
  const crypto = require('crypto');
  const nodeHttp = require('http');
++const { validate } = require('openvibe-contracts');
++const cursor = require('../server/cursor');
  const topics = require('../server/topics');
  const { hasCap, allows } = require('../server/auth');
  const { load } = require('../server/config');
@@@ -49,6 +51,9 @@ t('pull: cursor, topic filter, internal events included for services', async ()
  
      let r = await request(h.base, 'GET', '/api/v1/events?topic=live.vod.*&after_seq=0&limit=1', { token: reader });
      assert.strictEqual(r.status, 200, r.text);
++    assert.ok(validate('events.read-result@1', r.body).valid, JSON.stringify(validate('events.read-result@1', r.body).errors));
++    for (const e of r.body.events) assert.deepStrictEqual(cursor.decode(e.cursor), { seq: e.seq, epoch: await h.store.epoch() });
++    assert.strictEqual(cursor.decode(r.body.next_cursor).seq, r.body.next_after_seq);
      assert.deepStrictEqual(r.body.events.map(e => e.seq), [s1]);
      assert.strictEqual(r.body.events[0].event.visibility, 'internal');
      r = await request(h.base, 'GET', `/api/v1/events?topic=live.vod.*&after_seq=${r.body.next_after_seq}`, { token: reader });
ddiff --git a/test/realtime.test.js b/test/realtime.test.js
iindex f775b93..272f99e 100644
---- a/test/realtime.test.js
++++ b/test/realtime.test.js
@@@ -50,7 +50,8 @@ t('visibility: public to everyone, subject only to that user, internal only to s
      assert.deepStrictEqual(seqs(a), [pub, toAliceAsSubject, byAliceAsActor]);
      assert.deepStrictEqual(seqs(b), [pub], 'bob sees nothing addressed to alice');
      assert.deepStrictEqual(seqs(svc), [pub, internal, toAliceAsSubject, byAliceAsActor]);
--    assert.strictEqual(a.messages.find(m => m.data.includes('hi alice')).id, String(toAliceAsSubject), 'SSE id is the seq');
++    const { encode } = require('../server/cursor');
++    assert.strictEqual(a.messages.find(m => m.data.includes('hi alice')).id, encode(toAliceAsSubject, await h.store.epoch()), 'SSE id is the cursor');
      for (const c of [anon, a, b, svc]) c.close();
  });