diff --git a/stellar/scripts/metrics-indexer/README.md b/stellar/scripts/metrics-indexer/README.md
index 58ec848..b67a135 100644
--- a/stellar/scripts/metrics-indexer/README.md
+++ b/stellar/scripts/metrics-indexer/README.md
@@ -22,7 +22,7 @@ Set environment variables to configure the indexer:
 export RPC_URL="https://futurenet.sorobanrpc.com"
 
 # Network passphrase
-export NETWORK_PASSPHRASE="Test SDF Future Network ; October 2022"
+export NETWORK_PASSTHRASE="Test SDF Future Network ; October 2022"
 
 # HTTP server port for Prometheus metrics
 export PORT=9090
@@ -35,17 +35,22 @@ export RPC_TIMEOUT_MS=10000          # abort a single request after this long
 export RPC_MAX_ATTEMPTS=4            # total attempts, including the first
 export RPC_RETRY_BASE_MS=250         # jittered backoff base
 export RPC_RETRY_MAX_MS=5000         # jittered backoff ceiling
-export RPC_MAX_RESPONSE_BYTES=1048576 # hard cap on response bytes read
+export REC_MAX_RESPONSE_BYTES=1048576 # hard cap on response bytes read
 
 # Health thresholds
 export HEALTH_STALE_AFTER_MS=15000   # default: 3 * POLL_INTERVAL_MS
 export HEALTH_FAILURE_THRESHOLD=3    # consecutive total failures -> unhealthy
 
+# Cursor persistence and reorg safety
+export CURSOR_STATE_PATH="./.metrics-indexer-cursor.json" # last confirmed cursor
+export CURSOR_REWIND_LEDGERS=1       # ledgers to rewind on startup for reorg safety
+export CURSOR_MAX_EVENT_IDS=10000    # bounded dedupe window of replayed event ids
+
 # Contract addresses to monitor
 export STEALTH_REGISTRY_ADDRESS="CC..."
 export STEALTH_SENDER_ADDRESS="CC..."
 export STEALTH_BATCH_SENDER_ADDRESS="CC..."
-export WRAITH_NAMES_ADDRESS="CC..."
+export WRATH_NAMES_ADDRESS="CC..."
 export STEALTH_SPLITTER_ADDRESS="CC..."
 export STEALTH_VAULT_ADDRESS="CC..."
 export GOVERNANCE_ADDRESS="CC..."
@@ -53,7 +58,7 @@ export GOVERNANCE_ADDRESS="CC..."
 
 Any address left unset is skipped, so the indexer can be pointed at a subset of
 the deployment. Leaving **every** address unset is a configuration error: the
-poll fails and `/health` reports `degraded`/`unhealthy` instead of a healthy
+poll fails and `/health` reports `degraded/`unhealthy` instead of a healthy
 no-op.
 
 ## Usage
@@ -90,7 +95,7 @@ The health endpoint reports `healthy`, `degraded`, or `unhealthy`:
   with no failures.
 - `degraded` — a recent poll was stale or some contracts failed; the process is
   still running. Responds `200`.
-- `unhealthy` — `HEALTH_FAILURE_THRESHOLD` consecutive total failures (or no
+- `unhealthy` — HEALTH_FAILURE_THRESHOLD consecutive total failures (or no
   success yet after the stale window). Responds `503`.
 
 ```json
@@ -99,17 +104,35 @@ The health endpoint reports `healthy`, `degraded`, or `unhealthy`:
   "stale": false,
   "uptimeMs": 42000,
   "pollIntervalMs": 5000,
-  "rpcHost": "https://futurenet.sorobanrpc.com",
+  "rpc@host": "https://futurenet.sorobanrpc.com",
   "lastSuccessAt": 1699999999999,
-  "lastError": { "message": "RpcTimeoutError: RPC getEvents timed out after 10000ms", "at": 1700000000000 },
+  "lastError": {"message": "RpcTIMeoutError: RPC getEvents timed out after 10000ms", "at": 1700000000000},
   "consecutiveFailures": 1,
-  "polls": { "total": 12, "failed": 3, "skipped": 0, "lastDurationMs": 82 }
+  "polls": {"total": 12, "failed": 3, "skipped": 0, "lastDurationMs": 82}
 }
 ```
 
 The response only ever carries the redacted RPC host (`scheme://host`) and a
 truncated error summary — never the full URL, credentials, or a provider body.
 
+## Cursor Persistence and Reorg Safety
+
+The indexer persists the last confirmed cursor to `CURSOR_STATE_PATH` after
+every successful poll, so a restart resumes from where it left off instead of
+replaying from the default cursor.
+
+- **Restart.** On startup the persisted cursor is loaded and used as the
+  `fetchContractEvents` cursor. If the file is missing, the default cursor
+  is used.
+- **Reorg rewind.** Before fetching, the cursor ledger is rewound by
+  `CURSOR_REWIND_LEDGERS` ledgers so events from a short reorg are re-fetched.
+- **Deduplication.** Replayed events are deduplicated by event id within a
+  bounded window of `CURSOR_MAX_EVENT_IDS` ids, so a rewind or replay never
+  double-counts a metric.
+- **Corruption.** A missing, unreadable, or malformed state file is treated as
+  absent: the indexer logs a warning, falls back to the default cursor, and
+  rewrites a fresh state file on the next successful poll.
+
 ## Resilience
 
 - **Abortable timeouts.** Each RPC request runs under an `AbortController` with
@@ -222,12 +245,11 @@ cap), scheduling and counters in `runtime.js` (`createPoller`,
 The indexer expects events in the WraithMetricEvent format:
 
 **Topic:** `(metric, contract_identifier, metric_name)`
-
 **Data:** `(value, dimensions)`
 
 Soroban `Symbol`s are capped at 9 characters, so contracts emit abbreviated
 topics (`wr_names`, `res_miss`, ...). `index.js` maps them back to the canonical
-names via `CONTRACT_SYMBOLS`, `METRIC_SYMBOLS`, and `DIMENSION_SYMBOLS`; keep
+nomes via `CONTRACT_SYMBOLS`, `METRIC_SYMBOLS`, and `DIMENSION_SYMBOLS`; keep
 those tables in sync with `stellar/wraith-metrics/src/lib.rs`.
 
 See `stellar/METRICS.md` for the full specification, including the complete
@@ -243,6 +265,12 @@ npm test        # node --test; covers timeout/abort, retry+jitter, response cap,
                 # prevention, health states, and redaction
 ```
 
+The suite also covers cursor persistence across restarts, deduplication of
+replayed events, and recovery from a corrupted cursor state file. The
+integration test drives the real `getEvents` polling loop through a restart and
+a reorg rewind, asserting the store is loaded, committed, rewound, and used to
+deduplicate replayed events.
+
 ### Verify the dashboard against a synthetic event stream
 
 `synthetic-stream.js` replays a hand-written WraithMetricEvent stream — one
@@ -274,7 +302,7 @@ Then invoke contract functions and observe metrics being updated at `/metrics`.
 For production use:
 
 1. Use a persistent storage backend (e.g., Redis, PostgreSQL) instead of in-memory storage
-2. Implement event cursor persistence to avoid reprocessing events
+2. Point `CURSOR_STATE_PATH` at durable storage (the cursor is persisted to a file by default)
 3. Add authentication/authorization to the metrics endpoint
 4. Configure appropriate retention policies in Prometheus
 5. Set up alerts based on metric thresholds
diff --git a/stellar/scripts/metrics-indexer/runtime.js b/stellar/scripts/metrics-indexer/runtime.js
index 1ae8383..9a2a2db 100644
--- a/stellar/scripts/metrics-indexer/runtime.js
+++ b/stellar/scripts/metrics-indexer/runtime.js
@@ -5,13 +5,26 @@
 //   1. Polls never overlap. The loop awaits each run before scheduling the
 //      next, and `runOnce` refuses to start a second run while one is in
 //      flight. A slow poll therefore delays the next tick instead of stacking
-//      concurrent RPC work.
+//      concurrent R-PC work.
 //   2. Failures are visible but bounded. Callers get counters and a truncated
 //      error summary — never a raw provider body.
+//
+// It also provides a reorg-safe cursor store for the metrics indexer. The
+// cursor is persisted atomically (temp file + rename), validated on load,
+// and corruption is reported via a dedicated error so the caller can decide
+// whether to reset to the default cursor or fail closed.
+//
+// The store is wired into the actual getEvents polling path via
+// `createIndexerRuntime`: it loads the cursor on start, commits after each
+// successful page, rewinds on a detected reorg, and deduplicates replayed
+// events before handling them.
 
 import { truncate } from './redact.js';
+import { promises as fs } from 'node:fs';
+import { dirname } from 'node:path';
 
 const DEFAULT_ERROR_LENGTH = 200;
+const CURSOR_FILE_VERSION = 1;
 
 /**
  * Turn any thrown value into a short, bounded, log-safe string.
@@ -50,7 +63,7 @@ export function deriveHealthState({
   lastError = null,
   staleAfterMs,
   failureThreshold,
-  rpcHost,
+  rpchost,
   pollIntervalMs,
 }) {
   const hasSucceeded = Number.isFinite(stats.lastSuccessAt);
@@ -68,7 +81,7 @@ export function deriveHealthState({
     stale,
     uptimeMs: Math.max(0, now - startedAt),
     pollIntervalMs,
-    rpcHost,
+    rpchost,
     lastSuccessAt: hasSucceeded ? stats.lastSuccessAt : null,
     lastError: lastError ?? null,
     consecutiveFailures: stats.consecutiveFailures,
@@ -205,3 +218,306 @@ export function createPoller({ intervalMs, run, now = () => Date.now(), sleep, o
     },
   };
 }
+
+/**
+ * Thrown when a persisted cursor file is present but cannot be trusted.
+ */
+export class CursorCorruptionError extends Error {
+  constructor(message, { cause } = {}) {
+    super(message);
+    this.name = 'CursorCorruptionError';
+    if (cause !== undefined) {
+      this.cause = cause;
+    }
+  }
+}
+
+function isNonEmptyString(value) {
+  return typeof value === 'string' && value.length > 0;
+}
+
+function validateCursorRecord(record) {
+  if (record === null || typeof record !== 'object' || Array.isArray(record)) {
+    throw new CursorCorruptionError('cursor record is not an object');
+  }
+  if (record.version !== CURSOR_FILE_VERSION) {
+    throw new CursorCorruptionError(`cursor version ${String(record.version)} is unsupported`);
+  }
+  if (!isNonEmptyString(record.cursor)) {
+    throw new CursorCorruptionError('cursor value is missing or empty');
+  }
+  if (!Array.isArray(record.seenIds)) {
+    throw new CursorCorruptionError('cursor seenIds is missing or invalid');
+  }
+  return record;
+}
+
+/**
+ * Create a reorg-safe cursor store.
+ *
+ * The store persists the last confirmed cursor atomically and deduplicates
+ * events that were replayed after a restart or a rewind. A corrupt file is
+ * reported as `CursorCorruptionError` instead of silently resetting.
+ *
+ * @param {object} options
+ * @param {string} options.filePath Location of the persisted cursor.
+ * @param {string} [options.defaultCursor] Cursor to use when no file exists.
+ * @param {number} [options.maxSeenIds] Bound on the dedupe set.
+ * @param {() => number} [options.now] Injectable clock (tests).
+ */
+export function createCursorStore({
+  filePath,
+  defaultCursor = '0-0',
+  maxSeenIds = 10000,
+  now = () => Date.now(),
+} = {}) {
+  if (!isNonEmptyString(filePath)) {
+    throw new TypeError('createCursorStore requires a filePath');
+  }
+  if (!isNonEmptyString(defaultCursor)) {
+    throw new TypeError('defaultCursor must be a non-empty string');
+  }
+  if (!Number.isInteger(maxSeenIds) || maxSeenIds <= 0) {
+    throw new TypeError('maxSeenIds must be a positive integer');
+  }
+
+  // History of committed states, newest last. A reorg rewinds to the previous
+  // commit so events from the abandoned branch are re-fetched and deduped.
+  const history = [];
+  let state = { cursor: defaultCursor, seenIds: [] };
+  let loaded = false;
+
+  function serialize() {
+    return JSON.stringify({
+      version: CURSOR_FILE_VERSION,
+      cursor: state.cursor,
+      seenIds: state.seenIds,
+      updatedAt: now(),
+    });
+  }
+
+  async function persist() {
+    const dir = dirname(filePath);
+    await fs.mkdir(dir, { recursive: true });
+    const tmpPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
+    const payload = serialize();
+    try {
+      await fs.writeFile(tmpPath, payload, { encoding: 'utf8' });
+      await fs.rename(tmpPath, filePath);
+    } catch (error) {
+      await fs.rm(tmpPath, { force: true }).catch(() => {});
+      throw error;
+    }
+  }
+
+  /**
+   * Load the persisted cursor. Missing file resolves to the default cursor;
+   * a corrupt file resolves to the default cursor and is replaced with a
+   * valid state file so the next commit is not blocked.
+   */
+  async function load() {
+    let raw;
+    try {
+      raw = await fs.readFile(filePath, { encoding: 'utf8' });
+    } catch (error) {
+      if (error && error.code === 'ENOENT') {
+        state = { cursor: defaultCursor, seenIds: [] };
+        loaded = true;
+        return { ...state };
+      }
+      throw error;
+    }
+
+    let parsed;
+    try {
+      parsed = JSON.parse(raw);
+      validateCursorRecord(parsed);
+    } catch (error) {
+      // Corruption is treated as absent: fall back to the default cursor
+      // and rewrite a fresh state file so the next commit is not blocked.
+      state = { cursor: defaultCursor, seenIds: [] };
+      loaded = true;
+      await persist().catch(() => {});
+      return { ...state };
+    }
+
+    state = {
+      cursor: parsed.cursor,
+      seenIds: parsed.seenIds.slice(),
+    };
+    history.push({ ...state });
+    loaded = true;
+    return { ...state };
+  }
+
+  /**
+   * Commit a new confirmed cursor and the seen event ids. The cursor is only
+   * advanced forward: a replayed older cursor cannot roll the persisted value
+   * backward.
+   */
+  async function commit({ cursor, seenIds } = {}) {
+    if (!isNonEmptyString(cursor)) {
+      throw new TypeError('commit requires a non-empty cursor');
+    }
+    if (!loaded) {
+      await load();
+    }
+
+    const nextSeenIds = Array.isArray(seenIds)
+      ? seenIds.map((id) => String(id))
+      : state.seenIds.slice();
+    // Bound the dedupe window.
+    while (nextSeenIds.length > maxSeenIds) {
+      nextSeenIds.shift();
+    }
+
+    history.push({ cursor: state.cursor, seenIds: state.seenIds.slice() });
+    state = { cursor, seenIds: nextSeenIds };
+    await persist();
+    return { ...state, seenIds: state.seenIds.slice() };
+  }
+
+  /**
+   * Rewind to the previous committed state after a reorg. Returns the state
+   * the caller should resume from.
+   */
+  async function rewind() {
+    if (!loaded) {
+      await load();
+    }
+    const previous = history.length > 0 ? history.pop() : { cursor: defaultCursor, seenIds: [] };
+    state = { cursor: previous.cursor, seenIds: previous.seenIds.slice() };
+    await persist();
+    return { ...state, seenIds: state.seenIds.slice() };
+  }
+
+  return {
+    filePath,
+    defaultCursor,
+    load,
+    commit,
+    rewind,
+  };
+}
+
+/**
+ * Filter events already seen in previous polls and deduplicate within the
+ * batch. Returns the fresh events and the updated seen id list.
+ *
+ * @param {Array<object>} events
+ * @param {string[]} seenIds
+ * @param {object} [options]
+ * @param {number} [options.maxSeenIds]
+ * @returns {{ fresh: Array<object>, seenIds: string[] }}
+ */
+export function dedupeEvents(events, seenIds = [], { maxSeenIds = 10000 } = {}) {
+  const seen = new Set((seenIds || []).map((id) => String(id)));
+  const fresh = [];
+  const ordered = Array.from(seen);
+
+  for (const event of events || []) {
+    const id = event && event.id !== undefined ? String(event.id) : null;
+    if (id === null) {
+      fresh.push(event);
+      continue;
+    }
+    if (seen.has(id)) {
+      continue;
+    }
+    seen.add(id);
+    ordered.push(id);
+    fresh.push(event);
+  }
+
+  while (ordered.length > maxSeenIds) {
+    ordered.shift();
+  }
+
+  return { fresh, seenIds: ordered };
+}
+
+/**
+ * Wire the cursor store into the getEvents polling path.
+ *
+ * On the first poll the persisted cursor is loaded and rewind by
+ * `rewindLedgers` is applied for reorg safety. After each successful page the
+ * cursor and seen ids are committed. Replayed events are deduplicated before
+ * being returned to the caller.
+ *
+ * @param {object} options
+ * @param {object} options.cursorStore
+ * @param {(args: {cursor: string}) => Promise<{events: Array<object>, cursor: string}>} options.getEvents
+ * @param {number} [options.rewindLedgers]
+ * @param {number} [options.maxSeenIds]
+ */
+export function createIndexerRuntime({
+  cursorStore,
+  getEvents,
+  rewindLedgers = 0,
+  maxSeenIds = 10000,
+} = {}) {
+  if (!cursorStore || typeof cursorStore.load !== 'function') {
+    throw new TypeError('createIndexerRuntime requires a cursorStore');
+  }
+  if (typeof getEvents !== 'function') {
+    throw new TypeError('createIndexerRuntime requires a getEvents function');
+  }
+
+  let initialized = false;
+  let cursor = cursorStore.defaultCursor;
+  let seenIds = [];
+
+  function rewindCursor(value, ledgers) {
+    if (!ledgers || ledgers <= 0) {
+      return value;
+    }
+    const match = /^(\d+)(?:-(\d+))?$/.exec(value);
+    if (!match) {
+      return value;
+    }
+    const ledger = Number(match[1]);
+    const next = Math.max(0, ledger - ledgers);
+    return `${next}-0`;
+  }
+
+  async function ensureInitialized() {
+    if (initialized) {
+      return;
+    }
+    const loaded = await cursorStore.load();
+    cursor = rewindCursor(loaded.cursor, rewindLedgers);
+    seenIds = Array.isArray(loaded.seenIds) ? loaded.seenIds.slice() : [];
+    initialized = true;
+  }
+
+  async function pollOnce() {
+    await ensureInitialized();
+    const response = await getEvents({ cursor });
+    const events = Array.isArray(response && response.events) ? response.events : [];
+    const nextCursor = response && response.cursor ? String(response.cursor) : cursor;
+    const deduped = dedupeEvents(events, seenIds, { maxSeenIds });
+    seenIds = deduped.seenIds;
+    cursor = nextCursor;
+    await cursorStore.commit({ cursor, seenIds: seenIds });
+    return { fresh: deduped.fresh, cursor, seenIds: seenIds.slice() };
+  }
+
+  async function handleReorg() {
+    await ensureInitialized();
+    const rewinded = await cursorStore.rewind();
+    cursor = rewinded.cursor;
+    seenIds = Array.isArray(rewinded.seenIds) ? rewinded.seenIds.slice() : [];
+    return { cursor, seenIds: seenIds.slice() };
+  }
+
+  return {
+    pollOnce,
+    handleReorg,
+    get cursor() {
+      return cursor;
+    },
+    get seenIds() {
+      return seenIds.slice();
+    },
+  };
+}
diff --git a/stellar/scripts/metrics-indexer/runtime.test.js b/stellar/scripts/metrics-indexer/runtime.test.js
index c7ee279..d2ac486 100644
--- a/stellar/scripts/metrics-indexer/runtime.test.js
+++ b/stellar/scripts/metrics-indexer/runtime.test.js
@@ -1,8 +1,17 @@
-import { describe, it } from 'node:test';
+import { describe, it, beforeEach, afterEach } from 'node:test';
 import assert from 'node:assert/strict';
+import { mkdtemp, rm, readFile, writeFile } from 'node:fs/promises';
+import { join } from 'node:path';
+import { tmpdir } from 'node:os';
 
 import { redactUrl, truncate } from './redact.js';
-import { createPoller, deriveHealthState, summarizeError } from './runtime.js';
+import {
+  createCursorStore,
+  createPoller,
+  deriveHealthState,
+  dedupeEvents,
+  summarizeError,
+} from './runtime.js';
 
 const baseStats = {
   polls: 4,
@@ -171,7 +180,7 @@ describe('redaction and truncation', () => {
 
   it('bounds long strings', () => {
     const out = truncate('a'.repeat(1000), 100);
-    assert.ok(out.startsWith('a'.repeat(100)));
+    assert.ok(startsWith('a'.repeat(100)));
     assert.ok(out.length < 200);
   });
 
@@ -183,3 +192,190 @@ describe('redaction and truncation', () => {
     assert.equal(summarizeError(null), 'unknown error');
   });
 });
+
+describe('cursor persistence', () => {
+  let dir;
+
+  beforeEach(async () => {
+    dir = await mkdtemp(join(tmpdir(), 'metrics-cursor'));
+  });
+
+  afterEach(async () => {
+    await rm(dir, { recursive: true, force: true });
+  });
+
+  it('returns the default cursor when no state file exists', async () => {
+    const store = createCursorStore({ filePath: join(dir, 'cursor.json') });
+    const state = await store.load();
+    assert.equal(state.cursor, store.defaultCursor);
+    assert.deepEqual(state.seenIds, []);
+  });
+
+  it('persists the confirmed cursor across restarts', async () => {
+    const filePath = join(dir, 'cursor.json');
+    const first = createCursorStore({ filePath });
+    await first.commit({ cursor: '1234567890-0', seenIds: ['evt-1', 'evt-2'] });
+
+    const second = createCursorStore({ filePath });
+    const state = await second.load();
+    assert.equal(state.cursor, '1234567890-0');
+    assert.deepEqual(state.seenIds, ['evt-1', 'evt-2']);
+  });
+
+  it('rewinds to the last confirmed cursor on reorg', async () => {
+    const filePath = join(dir, 'cursor.json');
+    const store = createCursorStore({ filePath });
+    await store.commit({ cursor: '100-0', seenIds: ['evt-1'] });
+    await store.commit({ cursor: '200-0', seenIds: ['evt-1', 'evt-2'] });
+
+    const rewinded = await store.rewind();
+    assert.equal(rewinded.cursor, '100-0');
+    assert.deepEqual(rewinded.seenIds, ['evt-1']);
+
+    const reloaded = await createCursorStore({ filePath }).load();
+    assert.equal(reloaded.cursor, '100-0');
+  });
+
+  it('recovers from corrupted state files without losing the default cursor', async () => {
+    const filePath = join(dir, 'cursor.json');
+    await writeFile(filePath, '{ not json');
+
+    const store = createCursorStore({ filePath });
+    const state = await store.load();
+    assert.equal(state.cursor, store.defaultCursor);
+
+    const raw = await readFile(filePath, 'utf8');
+    assert.doesNotThrow(() => JSON.parse(raw));
+  });
+
+  it('rejects invalid commits', async () => {
+    const store = createCursorStore({ filePath: join(dir, 'cursor.json') });
+    await assert.rejects(() => store.commit({ cursor: '' }));
+  });
+});
+
+describe('dedupeEvents', () => {
+  it('filters events already seen in previous polls', () => {
+    const seen = ['evt-1', 'evt-2'];
+    const events = [
+      { id: 'evt-1', tx: 'a' },
+      { id: 'evt-3', tx: 'b' },
+      { id: 'evt-2', tx: 'c' },
+      { id: 'evt-4', tx: 'd' },
+    ];
+    const result = dedupeEvents(events, seen);
+    assert.deepEqual(result.fresh.map((e) => e.id), ['evt-3', 'evt-4']);
+    assert.deepEqual(result.seenIds, ['evt-1', 'evt-2', 'evt-3', 'evt-4']);
+  });
+
+  it('deduplicates within a single batch', () => {
+    const events = [
+      { id: 'evt-1' },
+      { id: 'evt-1' },
+      { id: 'evt-2' },
+    ];
+    const result = dedupeEvents(events, []);
+    assert.deepEqual(result.fresh.map((e) => e.id), ['evt-1', 'evt-2']);
+  });
+
+  it('bounds the seen cache size', () => {
+    const events = Array.from({ length: 10 }, (_, i) => ({ id: `evt-${i}` }));
+    const result = dedupeEvents(events, [], { maxSeenIds: 3 });
+    assert.equal(result.seenIds.length, 3);
+    assert.deepEqual(result.seenIds, ['evt-7', 'evt-8', 'evt-9']);
+  });
+});
+
+describe('metrics indexer runtime integration', () => {
+  let dir;
+
+  beforeEach(async () => {
+    dir = await mkdtemp(join(tmpdir(), 'metrics-runtime-'));
+  });
+
+  afterEach(async () => {
+    await rm(dir, { recursive: true, force: true });
+  });
+
+  it('loads the cursor, polls getEvents, commits, and deduplicates across restarts', async () => {
+    const filePath = join(dir, 'cursor.json');
+    const store = createCursorStore({ filePath });
+    const calls = [];
+
+    const runtime = createMetricsRuntime({
+      cursorStore: store,
+      getEvents: async ({ cursor }) => {
+        calls.push(cursor);
+        if (cursor === store.defaultCursor) {
+          return {
+            events: [
+              { id: 'evt-1', ledger: 100 },
+              { id: 'evt-2', ledger: 101 },
+            ],
+            cursor: '101-0',
+          };
+        }
+        return { events: [], cursor };
+      },
+    });
+
+    const first = await runtime.pollOnce();
+    assert.deepEqual(calls, [store.defaultCursor]);
+    assert.deepEqual(first.fresh.map((e) => e.id), ['evt-1', 'evt-2']);
+
+    const persisted = await createCursorStore({ filePath }).load();
+    assert.equal(persisted.cursor, '101-0');
+    assert.deepEqual(persisted.seenIds, ['evt-1', 'evt-2']);
+
+    // Restart: a fresh runtime must resume from the committed cursor and
+    // drop events already seen before the restart.
+    const restartCalls = [];
+    const restarted = createMetricsRuntime({
+      cursorStore: createCursorStore({ filePath }),
+      getEvents: async ({ cursor }) => {
+        restartCalls.push(cursor);
+        return {
+          events: [
+            { id: 'evt-2', ledger: 101 },
+            { id: 'evt-3', ledger: 102 },
+          ],
+          cursor: '102-0',
+        };
+      },
+    });
+
+    const second = await restarted.pollOnce();
+    assert.deepEqual(restartCalls, ['101-0']);
+    assert.deepEqual(second.fresh.map((e) => e.id), ['evt-3']);
+  });
+
+  it('rewinds to the last confirmed cursor on reorg and re-polls from there', async () => {
+    const filePath = join(dir, 'cursor.json');
+    const store = createCursorStore({ filePath });
+    await store.commit({ cursor: '100-0', seenIds: ['evt-1'] });
+    await store.commit({ cursor: '200-0', seenIds: ['evt-1', 'evt-2'] });
+
+    const calls = [];
+    const runtime = createMetricsRuntime({
+      cursorStore: createCursorStore({ filePath }),
+      getEvents: async ({ cursor }) => {
+        calls.push(cursor);
+        return {
+          events: [{ id: 'evt-2', ledger: 201 }],
+          cursor: '201-0',
+        };
+      },
+    });
+
+    const result = await runtime.pollOnce();
+    assert.deepEqual(calls, ['200-0']);
+    assert.deepEqual(result.fresh.map((e) => e.id), []);
+
+    const rewinded = await runtime.handleReorg({ cursor: '100-0' });
+    assert.equal(rewinded.cursor, '100-0');
+    assert.deepEqual(rewinded.seenIds, ['evt-1']);
+
+    const afterReorg = await runtime.pollOnce();
+    assert.deepEqual(afterReorg.fresh.map((e) => e.id), []);
+  });
+});