Sources: enable stored resolver seeds; read Open Food Facts' official deltas (#153)

ddiff --git a/README.md b/README.md
iindex e1aafe2..af369ca 100644
---- a/README.md
++++ b/README.md
@@@ -106,7 +106,7 @@ succeeded. Health status: `healthy | stale | failing | never_fetched | disabled
  | `sitemap` (`sitemap@1`) | `<urlset>` (+ Google News extension), `<sitemapindex>` | index children are listed, not followed |
  | `jsonld` (`jsonld@2`) | schema.org Product, Offer/AggregateOffer, Review, NewsArticle, BlogPosting (+ Article) | prices, ratings and availability exactly as the page states; missing = null; embedded review text not kept; a page without JSON-LD is a `parse_error`; `list_price` from a ListPrice/StrikethroughPrice/MSRP/SRP `priceSpecification`, as stated |
  | `api` (`api@1`) | JSON or XML official APIs via a registry mapping | `items_path`, `items_shape` (`array` default, `map` for records keyed by id), `fields`, `extra`, `where`/`filter`; epoch fields only where mapped as such; `resolver: "hn-whoishiring"` walks the Hacker News thread's JSON documents instead |
--| `bulk` (`bulk@1`) | one dataset file, streamed (CSV, JSONL or TSV) | `max_bytes` is the size cap, `max_rows` the per-run row limit; the stream stops at the limit, so a multi-GB file is never held in memory |
++| `bulk` (`bulk@1`) | one dataset file, streamed (CSV, JSONL or TSV) | `max_bytes` is the size cap, `max_rows` the per-run row limit; the stream stops at the limit, so a multi-GB file is never held in memory. `resolver: "off-delta"` reads a text index and only the gzipped JSONL delta parts it lists, keeping the last finished part and the current part's row offset in `source_state` (Open Food Facts) |
  | `crawl` (`jsonld@2`, `page@1`) | a whole site from its seeds: its sitemaps and, if asked, its links (see below) | only pages that publish JSON-LD are kept, unless `require_structured: false` keeps a `page` item from the page's own search metadata |
  | `manual` (`manual@1`) | items entered through the API with an evidence URL | no fetcher |
  
ddiff --git a/docs/contracts-proposal/contracts/sources/source.v1.json b/docs/contracts-proposal/contracts/sources/source.v1.json
iindex dec941e..8ca161a 100644
---- a/docs/contracts-proposal/contracts/sources/source.v1.json
++++ b/docs/contracts-proposal/contracts/sources/source.v1.json
@@@ -156,9 +156,10 @@
            },
            "resolver": {
              "enum": [
--              "hn-whoishiring"
++              "hn-whoishiring",
++              "off-delta"
              ],
--            "description": "An api endpoint whose records are walked from several JSON URLs instead of one items_path. hn-whoishiring reads user/whoishiring.json, the latest \"Who is hiring\" story's top-level kids, and each item/<id>.json; it stores one job per comment, a short excerpt with emails and phones removed, and links the comment (Sources, 2026-10-31). The endpoint URL must be a .json URL."
++            "description": "An api endpoint whose records are walked from several JSON URLs instead of one items_path, or a bulk endpoint whose rows come from the gzipped JSONL parts a text index lists. hn-whoishiring reads user/whoishiring.json, the latest \"Who is hiring\" story's top-level kids, and each item/<id>.json; it stores one job per comment, a short excerpt with emails and phones removed, and links the comment (Sources, 2026-10-31). off-delta reads index.txt and then only the delta parts not finished yet, resuming a part at its stored row offset (Open Food Facts; Sources, 2026-10-31). The endpoint URL is a .json URL for hn-whoishiring and the index's .txt URL for off-delta."
            },
            "item_kind": {
              "type": "string",
ddiff --git a/migrations/0009_source_state.sql b/migrations/0009_source_state.sql
nnew file mode 100644
iindex 0000000..4f8dd3c
---- /dev/null
++++ b/migrations/0009_source_state.sql
@@@ -0,0 +1,12 @@
++-- phase: expand
++-- A source's small persistent cursor (server/adapters/off-delta.js): the off-delta bulk resolver keeps the last
++-- completed delta part and, while a part is still being read, its row offset. One row per source and cursor name;
++-- the value is JSON text owned by that resolver. Nothing here belongs to a person.
++
++CREATE TABLE IF NOT EXISTS source_state (
++    source_key  text COLLATE "C" NOT NULL,
++    name        text COLLATE "C" NOT NULL,
++    value       text NOT NULL,
++    updated_at  bigint NOT NULL,
++    PRIMARY KEY (source_key, name)
++);
ddiff --git a/seeds/sources.json b/seeds/sources.json
iindex 877c8db..711a769 100644
---- a/seeds/sources.json
++++ b/seeds/sources.json
@@@ -2311,45 +2311,39 @@
      },
      {
        "key": "open-food-facts",
--      "name": "Open Food Facts product database (daily export)",
++      "name": "Open Food Facts product deltas (official daily delta export)",
        "type": "bulk",
        "category": "food",
        "homepage_url": "https://world.openfoodfacts.org/",
        "endpoints": [
          {
--          "url": "https://static.openfoodfacts.org/data/en.openfoodfacts.org.products.csv.gz",
--          "format": "tsv",
++          "url": "https://static.openfoodfacts.org/data/delta/index.txt",
++          "format": "jsonl",
++          "resolver": "off-delta",
            "item_kind": "food_product",
            "max_bytes": 2147483648,
            "max_rows": 12000,
            "timeout_ms": 120000,
            "fields": {
              "identity": "code",
--            "url": "url",
              "title": "product_name"
            },
            "extra": {
              "brands": "brands",
--            "categories": "categories",
--            "countries": "countries_en",
--            "main_category": "main_category_en",
--            "quantity": "quantity",
              "nutriscore_grade": "nutriscore_grade",
--            "nutriscore_score": "nutriscore_score",
--            "energy_kcal_100g": "energy-kcal_100g",
--            "fat_100g": "fat_100g",
--            "saturated_fat_100g": "saturated-fat_100g",
--            "carbohydrates_100g": "carbohydrates_100g",
--            "sugars_100g": "sugars_100g",
--            "proteins_100g": "proteins_100g",
--            "salt_100g": "salt_100g",
--            "fiber_100g": "fiber_100g",
--            "sodium_100g": "sodium_100g"
++            "energy_kcal_100g": "nutrition.aggregated_set.nutrients.energy-kcal.value",
++            "fat_100g": "nutrition.aggregated_set.nutrients.fat.value",
++            "saturated_fat_100g": "nutrition.aggregated_set.nutrients.saturated-fat.value",
++            "carbohydrates_100g": "nutrition.aggregated_set.nutrients.carbohydrates.value",
++            "sugars_100g": "nutrition.aggregated_set.nutrients.sugars.value",
++            "proteins_100g": "nutrition.aggregated_set.nutrients.proteins.value",
++            "salt_100g": "nutrition.aggregated_set.nutrients.salt.value",
++            "fiber_100g": "nutrition.aggregated_set.nutrients.fiber.value"
            }
          }
        ],
--      "robots_note": "static.openfoodfacts.org/robots.txt checked 2026-10-30 from openvibe-ovh: 'User-agent: *', 'Disallow: /api', 'Disallow: /cgi', 'Disallow: /facets'; /data/ is allowed. The download redirects to openfoodfacts-ds.s3.eu-west-3.amazonaws.com, whose /robots.txt is 404 (no restrictions). The fetcher re-checks robots.txt on both origins, and every redirect hop, on every run.",
--      "terms_note": "Open Food Facts publishes its database under the Open Database License: 'The Open Food Facts database is available under the Open Database License' (https://world.openfoodfacts.org/terms-of-use). Attribution and share-alike apply, and the ODbL notice stays with anything derived. This is the tab-separated CSV export (en.openfoodfacts.org.products.csv.gz, about 1.5 GB compressed): the bulk loader streams it with a 2 GiB cap and stops after 12,000 rows per run, so it never downloads the file whole. Store the code, name, brands, categories, Nutri-Score and the per-100 g nutrient columns; images are CC BY-SA and are not stored. Re-verify before enabling.",
++      "robots_note": "static.openfoodfacts.org/robots.txt checked 2026-10-30 from openvibe-ovh with the ADR 0007 User-Agent: 'User-agent: *', 'Disallow: /api', 'Disallow: /cgi', 'Disallow: /facets'; /data/ and /data/delta/ are allowed. The index and every delta file are served directly by static.openfoodfacts.org (HTTP 200, application/octet-stream, about 22 MB each); the fetcher re-checks robots.txt on every run and on every redirect hop. api.openfoodfacts.org/world.openfoodfacts.org /api are not used.",
++      "terms_note": "Open Food Facts publishes its database under the Open Database License: 'The Open Food Facts database is available under the Open Database License' (https://world.openfoodfacts.org/terms-of-use). Attribution and share-alike apply, and the ODbL notice stays with anything derived. This source reads the official delta index (https://static.openfoodfacts.org/data/delta/index.txt): on the first run it starts at the oldest part the index lists and works forward; every later run fetches only the parts after the last one it finished, and a run stopped at its 12,000-row limit records that part's row offset so the next run continues inside it. The full export is one gzip stream and cannot be resumed without decompressing it from the start, so it is not read: the index had 13 daily parts on 2026-10-30, each about 22 MB compressed. Store the code, name, brands, Nutri-Score and the per-100 g nutrients from nutrition.aggregated_set; images are CC BY-SA and are not stored. Re-verify before enabling.",
        "license_note": "ODbL 1.0 (database); individual contents DbCL; images CC BY-SA. Credit 'Open Food Facts (ODbL)' and link the product page and https://opendatacommons.org/licenses/odbl/1-0/ wherever a product's figures show.",
        "min_interval_ms": 60000,
        "poll_interval_sec": 86400,
ddiff --git a/server/adapters/bulk.js b/server/adapters/bulk.js
iindex 9e6a8a4..46ee56d 100644
---- a/server/adapters/bulk.js
++++ b/server/adapters/bulk.js
@@@ -22,14 +22,17 @@ const { ParseError } = require('./util');
  
  const PARSER_VERSION = 'bulk@1';
  
--function createParser({ endpoint = {}, url, maxItems = 500, summaryMax = 1000 } = {}) {
++function createParser({ endpoint = {}, url, maxItems = 500, summaryMax = 1000, skipRows = 0 } = {}) {
      const format = endpoint.format || 'csv';
      const limit = Math.max(1, Math.min(Number(endpoint.max_rows) || 1000, Number(maxItems) || 500));
      const decoder = new StringDecoder('utf8');
      const items = [];
      const seen = new Set();
++    const skippedAtStart = Math.max(0, Math.floor(Number(skipRows) || 0));
++    let skip = skippedAtStart;
      let skipped = 0;
      let rows = 0;
++    let scanned = 0;
      let stopped = false;
  
      // Line-oriented formats keep the tail of an incomplete line between chunks. CSV needs a small state machine
@@@ -43,9 +46,11 @@ function createParser({ endpoint = {}, url, maxItems = 500, summaryMax = 1000 }
      let first = true;
  
      function accept(rec) {
++        scanned++;
++        if (skip > 0) { skip--; return; }              // a resumed bulk part: rows already kept by an earlier run
          if (stopped) return;
--        if (!rec || typeof rec !== 'object' || Array.isArray(rec)) { skipped++; return; }
          const index = rows++;
++        if (!rec || typeof rec !== 'object' || Array.isArray(rec)) { skipped++; return; }
          const item = mapRecord(rec, rec, undefined, index, endpoint, url, summaryMax);
          if (!item) { skipped++; return; }
          if (seen.has(item.identity)) { skipped++; return; }
@@@ -153,7 +158,7 @@ function createParser({ endpoint = {}, url, maxItems = 500, summaryMax = 1000 }
              if (format === 'csv') { if (tail) csvWrite(tail); csvFinish(); }
              else { if (tail) lineWrite(tail); if (pending) line(pending.endsWith('\r') ? pending.slice(0, -1) : pending); }
          }
--        return { items, skipped, meta: { format, rows, filtered: 0, capped: stopped } };
++        return { items, skipped, meta: { format, rows, scanned, skip_rows: skippedAtStart - skip, filtered: 0, capped: stopped } };
      }
  
      return { write, end, get full() { return stopped; }, get count() { return items.length; } };
ddiff --git a/server/adapters/off-delta.js b/server/adapters/off-delta.js
nnew file mode 100644
iindex 0000000..3aa8da4
---- /dev/null
++++ b/server/adapters/off-delta.js
@@@ -0,0 +1,70 @@
++'use strict';
++/**
++ * A delta index: a text file that lists gzipped JSONL parts (Open Food Facts publishes one such index at
++ * https://static.openfoodfacts.org/data/delta/index.txt). A bulk source with `resolver: "off-delta"` reads the
++ * index, then only the parts it has not finished; the cursor is the last completed part plus, while a part is
++ * still being read, that part's name and row offset. Parts are named `..._<start>_<end>.json.gz`; the end
++ * timestamp gives their order, oldest first, so a run works forward and never re-reads an older part.
++ *
++ * The full Open Food Facts export is one gzip stream: resuming inside it would mean decompressing it from the
++ * start again, so this resolver uses only the official deltas. Every delta is itself gzip, so a run that stops
++ * at the row limit records its row offset in that one part and the next run skips those rows while streaming it.
++ */
++const { ParseError } = require('./util');
++
++const PARSER_VERSION = 'off-delta@1';
++const STATE_NAME = 'delta.cursor';
++const FILE_RE = /^[A-Za-z0-9][A-Za-z0-9._-]*\.jsonl?\.gz$/;
++const RANGE_RE = /_(\d{10})_(\d{10})\.jsonl?\.gz$/;
++
++/** The index's non-empty lines → its parts, oldest first (by the end timestamp in the name). */
++function parseIndex(text, indexUrl) {
++    const files = [];
++    const seen = new Set();
++    for (const raw of String(text == null ? '' : text).split(/\r?\n/)) {
++        const name = raw.trim();
++        if (!name) continue;
++        if (!FILE_RE.test(name) || name.includes('..') || name.includes('/') || name.includes('\\')) {
++            throw new ParseError(`delta index names an unexpected file: ${name.slice(0, 200)}`);
++        }
++        const m = RANGE_RE.exec(name);
++        if (!m) throw new ParseError(`delta file name has no start/end timestamp: ${name.slice(0, 200)}`);
++        const start = Number(m[1]);
++        const end = Number(m[2]);
++        if (!(end > start) || !Number.isSafeInteger(start) || !Number.isSafeInteger(end)) throw new ParseError(`delta file name has a bad time range: ${name.slice(0, 200)}`);
++        if (seen.has(name)) throw new ParseError(`delta index repeats ${name.slice(0, 200)}`);
++        seen.add(name);
++        files.push({ name, start, end, url: new URL(name, indexUrl).toString() });
++    }
++    if (!files.length) throw new ParseError('delta index lists no files');
++    files.sort((a, b) => (a.end - b.end) || (a.name < b.name ? -1 : a.name > b.name ? 1 : 0));
++    return files;
++}
++
++/** The stored cursor JSON → the shape the resolver reads; a corrupt value starts from the oldest part again. */
++function readState(value) {
++    let o = null;
++    try { o = value ? JSON.parse(value) : null; } catch { o = null; }
++    const state = o && typeof o === 'object' ? o : {};
++    return {
++        seen: typeof state.seen === 'string' && state.seen ? state.seen : null,
++        seen_end: Number.isSafeInteger(Number(state.seen_end)) && Number(state.seen_end) > 0 ? Number(state.seen_end) : 0,
++        current: typeof state.current === 'string' && state.current ? state.current : null,
++        offset: Number.isSafeInteger(Number(state.offset)) && Number(state.offset) > 0 ? Number(state.offset) : 0,
++    };
++}
++
++/** The parts this run should read, oldest first: the unfinished one first, then every later unseen one. */
++function nextFiles(files, state) {
++    const seenEnd = state.seen_end || 0;
++    const seenName = state.seen || '';
++    const after = (f) => f.end > seenEnd || (f.end === seenEnd && f.name > seenName);
++    let list = files.filter(after);
++    if (state.current) {
++        const current = files.find((f) => f.name === state.current);
++        if (current && after(current)) list = [current, ...list.filter((f) => f.name !== current.name)];
++    }
++    return list.sort((a, b) => (a.end - b.end) || (a.name < b.name ? -1 : a.name > b.name ? 1 : 0));
++}
++
++module.exports = { parseIndex, readState, nextFiles, PARSER_VERSION, STATE_NAME, FILE_RE, RANGE_RE };
ddiff --git a/server/ingest.js b/server/ingest.js
iindex 1c24697..b308290 100644
---- a/server/ingest.js
++++ b/server/ingest.js
@@@ -28,6 +28,7 @@
  const crypto = require('crypto');
  const { ids } = require('openvibe-contracts');
  const { ADAPTERS, RESOLVERS } = require('./adapters');
++const offDelta = require('./adapters/off-delta');
  const extract = require('./adapters/extract');
  const { createCrawler } = require('./crawl');
  const { sha256, decodeBody, ParseError } = require('./adapters/util');
@@@ -58,6 +59,9 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
          touchEndpoint: db.prepare(`INSERT INTO endpoint_state (source_key, url, last_status, last_fetch_at) VALUES (@k, @u, @status, @at)
              ON CONFLICT(source_key, url) DO UPDATE SET last_status = excluded.last_status, last_fetch_at = excluded.last_fetch_at`),
          requested: db.prepare('UPDATE sources SET last_request_at = ? WHERE key = ?'),
++        stateGet: db.prepare('SELECT value FROM source_state WHERE source_key = ? AND name = ?'),
++        stateSet: db.prepare(`INSERT INTO source_state (source_key, name, value, updated_at) VALUES (@k, @n, @v, @at)
++            ON CONFLICT(source_key, name) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at`),
          lastOk: db.prepare(`SELECT id FROM fetch_runs WHERE source_key = ? AND endpoint_url = ? AND state = 'ok' ORDER BY rid DESC LIMIT 1`),
          finish: db.prepare(`UPDATE sources SET last_run_at = @at, last_state = @state, consecutive_failures = @failures,
              last_success_at = COALESCE(@success_at, last_success_at), next_due_at = @next_due, not_before = @not_before WHERE key = @key`),
@@@ -412,6 +416,165 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
          return { run, failed: false, success: true, recorded: true };
      }
  
++    /**
++     * A bulk source with `resolver: "off-delta"`: read a text index, then only the delta parts not finished yet.
++     * The full export is one gzip stream and cannot be resumed, so the first run starts at the oldest part in the
++     * index. A run reads up to its row limit across parts; a part stopped by the limit keeps its row offset in
++     * source_state, and the next run streams that part again only to skip those rows, then continues.
++     */
++    async function fetchOffDeltaEndpoint(source, endpoint, trigger) {
++        const adapter = ADAPTERS.bulk;
++        const startedAt = now();
++        const run = baseRun(source, endpoint.url, trigger, startedAt);
++        const target = new URL(endpoint.url);
++        const host = target.host;
++
++        const headers = { Accept: 'text/plain' };
++        const sensitive = [];
++        let requestUrl = target;
++        if (source.auth.mode !== 'none') {
++            const secret = config.secrets[source.auth.env];
++            if (!secret) {
++                Object.assign(run, { state: 'disabled', error_code: 'credential_missing', detail: `environment variable ${source.auth.env} is not set`, finished_at: now() });
++                return { run, failed: false };
++            }
++            if (source.auth.mode === 'bearer') { headers.Authorization = `Bearer ${secret}`; sensitive.push('Authorization'); }
++            if (source.auth.mode === 'header') { headers[source.auth.header] = secret; sensitive.push(source.auth.header); }
++            if (source.auth.mode === 'query') { requestUrl = new URL(target); requestUrl.searchParams.set(source.auth.param, secret); }
++        }
++
++        const verdict = await robots.check(target.toString());
++        setCrawlDelay(host, verdict.crawlDelaySec);
++        if (!verdict.allowed) {
++            if (verdict.refusedCode) Object.assign(run, { state: 'http_error', error_code: verdict.refusedCode, detail: `${host} is not fetchable (${verdict.refusedCode})`, finished_at: now() });
++            else Object.assign(run, { state: 'robots_denied', error_code: 'robots', detail: verdict.reason, finished_at: now() });
++            return { run, failed: true };
++        }
++
++        await space(host, source.min_interval_ms);
++        await st.requested.run(now(), source.key);
++        const hash = crypto.createHash('sha256');
++        let bytes = 0;
++        let lastStatus = null;
++        const beforeHop = async (next) => {
++            const v = await robots.check(next.toString());
++            if (!v.allowed) return { refuse: v.reason };
++            if (next.host !== host) {
++                setCrawlDelay(next.host, v.crawlDelaySec);
++                await space(next.host, source.min_interval_ms);
++            }
++            return null;
++        };
++        const read = async (url, cap, accept) => {
++            const v = await robots.check(url);
++            if (!v.allowed) throw Object.assign(new Error(v.reason || 'robots.txt disallows this URL'), { code: 'robots_denied' });
++            const r = await fetcher.fetchUrl(url, {
++                headers: { Accept: accept }, capBytes: cap, timeoutMs: endpoint.timeout_ms || undefined,
++                sensitiveHeaders: sensitive, beforeHop,
++            });
++            lastStatus = r.status;
++            run.http_status = r.status;
++            if (r.status === 429) throw Object.assign(new Error('the delta index answered 429'), { code: 'rate_limited' });
++            if (r.status !== 200 && r.status !== 203) throw Object.assign(new Error(`HTTP ${r.status} for ${url}`), { code: 'http_error' });
++            bytes += r.bytes || 0;
++            hash.update(r.body || Buffer.alloc(0));
++            return r;
++        };
++        const failure = (err) => {
++            const state = err.code === 'robots_denied' ? 'robots_denied' : err.code === 'rate_limited' ? 'rate_limited'
++                : err.code === 'timeout' ? 'timeout' : err.code === 'parse_error' ? 'parse_error' : 'http_error';
++            Object.assign(run, { state, error_code: err.code || 'network', detail: String(err.message).slice(0, 500), finished_at: now() });
++            if (err.code !== 'parse_error') log.error(`[ingest] ${source.key}: off-delta failed: ${err.stack || err}`);
++            return { run, failed: true };
++        };
++
++        let files;
++        try {
++            const index = await read(requestUrl.toString(), Math.min(endpoint.max_bytes || 1024 * 1024, 1024 * 1024), 'text/plain');
++            files = offDelta.parseIndex(decodeBody(index.body, index.headers['content-type'] || 'text/plain'), endpoint.url);
++        } catch (err) { return failure(err); }
++
++        const stateRow = await st.stateGet.get(source.key, offDelta.STATE_NAME);
++        const state = offDelta.readState(stateRow && stateRow.value);
++        if (state.current && !files.some((f) => f.name === state.current)) {
++            state.current = null;
++            state.offset = 0;
++        }
++        const queue = offDelta.nextFiles(files, state);
++        const budget = Math.max(1, Math.min(Number(endpoint.max_rows) || 1000, Number(source.max_items) || 500));
++        let remaining = budget;
++        const collected = new Map();         // identity → item, later parts win
++        let duplicates = 0;
++        let skipped = 0;
++        let cursor = { seen: state.seen, seen_end: state.seen_end, current: state.current, offset: state.offset };
++
++        for (const file of queue) {
++            if (remaining <= 0) break;
++            const offset = cursor.current === file.name ? cursor.offset : 0;
++            const parser = adapter.createParser({
++                endpoint: { ...endpoint, max_rows: Math.min(Number(endpoint.max_rows) || 1000, remaining) },
++                url: file.url, maxItems: remaining, summaryMax: config.items.summaryMax, skipRows: offset,
++            });
++            let res;
++            try {
++                const v = await robots.check(file.url);
++                if (!v.allowed) throw Object.assign(new Error(v.reason || 'robots.txt disallows this delta'), { code: 'robots_denied' });
++                res = await fetcher.streamUrl(file.url, {
++                    headers: { Accept: adapter.accept },
++                    capBytes: endpoint.max_bytes || config.fetch.maxBytes,
++                    timeoutMs: endpoint.timeout_ms || undefined,
++                    beforeHop,
++                }, (chunk) => { hash.update(chunk); return parser.write(chunk); });
++            } catch (err) { return failure(err); }
++            lastStatus = res.status;
++            run.http_status = res.status;
++            if (res.status === 429) {
++                const wait = Math.min(retryAfterMs(res.headers['retry-after'], now()) ?? 10 * 60 * 1000, 24 * 3600 * 1000);
++                Object.assign(run, { state: 'rate_limited', error_code: 'upstream_429', detail: `the site asked us to wait ${Math.round(wait / 1000)} s` });
++                return { run, failed: true, notBefore: now() + wait };
++            }
++            if (res.status !== 200 && res.status !== 203) {
++                Object.assign(run, { state: 'http_error', error_code: `http_${res.status}`, detail: `HTTP ${res.status}` });
++                return { run, failed: true };
++            }
++            bytes += res.bytes || 0;
++            let parsed;
++            try { parsed = parser.end(); } catch (err) { return failure(err); }
++            for (const it of parsed.items) {
++                if (collected.has(it.identity)) duplicates++;
++                collected.delete(it.identity);
++                collected.set(it.identity, it);
++            }
++            skipped += parsed.skipped;
++            remaining -= parsed.items.length;
++            if (res.stopped) {               // the row limit stopped this part: keep its offset for the next run
++                cursor = { ...cursor, current: file.name, offset: offset + parsed.meta.rows };
++                break;
++            }
++            cursor = { seen: file.name, seen_end: file.end, current: null, offset: 0 };
++            if (remaining <= 0) break;
++        }
++
++        const parsedItems = [...collected.values()];
++        run.bytes = bytes;
++        run.finished_at = now();
++        run.raw_body_hash = hash.digest('hex');
++        run.parser_version = offDelta.PARSER_VERSION;
++        await db.tx(async () => {
++            const counts = await items.ingest(source, parsedItems, {
++                runId: run.id, rawBodyHash: run.raw_body_hash, parserVersion: offDelta.PARSER_VERSION, retrievedAt: run.finished_at,
++            });
++            Object.assign(run, {
++                state: 'ok', items_seen: parsedItems.length, items_created: counts.created, items_updated: counts.updated,
++                items_unchanged: counts.unchanged, items_skipped: skipped + duplicates + counts.skipped,
++            });
++            await st.insertRun.run(run);
++            await st.stateSet.run({ k: source.key, n: offDelta.STATE_NAME, v: JSON.stringify(cursor), at: run.finished_at });
++            await st.saveValidators.run({ k: source.key, u: endpoint.url, etag: null, lm: null, status: lastStatus, at: run.finished_at });
++        });
++        return { run, failed: false, success: true, recorded: true };
++    }
++
      const crawler = createCrawler({
          db, config, items, robots, fetcher, spacer, now, log,
          record: { baseRun, insertRun: (r) => st.insertRun.run(r), requested: (t, key) => st.requested.run(t, key) },
@@@ -429,7 +592,9 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
              let out;
              try {
                  out = source.type === 'bulk'
--                    ? await fetchBulkEndpoint(source, endpoint, trigger)
++                    ? endpoint.resolver === 'off-delta'
++                        ? await fetchOffDeltaEndpoint(source, endpoint, trigger)
++                        : await fetchBulkEndpoint(source, endpoint, trigger)
                      : endpoint.resolver
                          ? await fetchResolverEndpoint(source, endpoint, trigger)
                          : await fetchEndpoint(source, endpoint, trigger);
ddiff --git a/server/registry.js b/server/registry.js
iindex 7418701..3495557 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -108,10 +108,16 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
              // it has no field mapping of its own, so it does not go through the api mapping checks.
              if (e.resolver !== undefined) {
                  for (const k of Object.keys(e)) {
--                    if (!['url', 'format', 'resolver', 'item_kind', 'max_bytes', 'timeout_ms'].includes(k)) throw new RegistryError(`endpoints[${i}].${k} is not a resolver field`);
++                    // The normalized row written by the create path carries the api defaults; accepting them back
++                    // (only as their defaults) is what lets PATCH { enabled: true } validate a stored resolver.
++                    if (!['url', 'format', 'resolver', 'item_kind', 'max_bytes', 'timeout_ms', 'items_shape', 'items_path', 'fields', 'extra'].includes(k)) throw new RegistryError(`endpoints[${i}].${k} is not a resolver field`);
                  }
                  if (e.resolver !== 'hn-whoishiring') throw new RegistryError(`endpoints[${i}].resolver must be hn-whoishiring`);
                  if (e.format !== undefined && e.format !== 'json') throw new RegistryError(`endpoints[${i}].format must be json for a resolver`);
++                if (e.items_shape !== undefined && e.items_shape !== 'array') throw new RegistryError(`endpoints[${i}].items_shape must be array for a resolver`);
++                if (e.items_path !== undefined && e.items_path !== '') throw new RegistryError(`endpoints[${i}].items_path must be "" for a resolver`);
++                if (e.fields !== undefined && (!e.fields || typeof e.fields !== 'object' || Array.isArray(e.fields) || Object.keys(e.fields).length)) throw new RegistryError(`endpoints[${i}].fields must be empty for a resolver`);
++                if (e.extra !== undefined && (!e.extra || typeof e.extra !== 'object' || Array.isArray(e.extra) || Object.keys(e.extra).length)) throw new RegistryError(`endpoints[${i}].extra must be empty for a resolver`);
                  if (!/\.json(\?|$)/.test(out.url)) throw new RegistryError(`endpoints[${i}].resolver needs a .json endpoint`);
                  if (e.item_kind !== undefined && !/^[a-z][a-z0-9_]{1,39}$/.test(e.item_kind)) throw new RegistryError(`endpoints[${i}].item_kind is malformed`);
                  if (e.max_bytes !== undefined && (!Number.isInteger(e.max_bytes) || e.max_bytes < 1024 || e.max_bytes > 32 * 1024 * 1024)) throw new RegistryError(`endpoints[${i}].max_bytes must be an integer from 1024 to 33554432`);
@@@ -138,7 +144,13 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
              const err = checkApiMapping(e, { bulk: true });
              if (err) throw new RegistryError(`endpoints[${i}]: ${err}`);
              for (const k of Object.keys(e)) {
--                if (!['url', 'format', 'item_kind', 'fields', 'extra', 'extra_lists', 'max_bytes', 'max_rows', 'timeout_ms'].includes(k)) throw new RegistryError(`endpoints[${i}].${k} is not a bulk field`);
++                if (!['url', 'format', 'item_kind', 'fields', 'extra', 'extra_lists', 'max_bytes', 'max_rows', 'timeout_ms', 'resolver'].includes(k)) throw new RegistryError(`endpoints[${i}].${k} is not a bulk field`);
++            }
++            // An off-delta bulk source reads a text index and then only the delta files it lists (Open Food Facts).
++            if (e.resolver !== undefined) {
++                if (e.resolver !== 'off-delta') throw new RegistryError(`endpoints[${i}].resolver must be off-delta`);
++                if (!/\.txt(\?|$)/.test(out.url)) throw new RegistryError(`endpoints[${i}].resolver needs the index .txt endpoint`);
++                if (e.format !== 'jsonl') throw new RegistryError(`endpoints[${i}].format must be jsonl for an off-delta resolver`);
              }
              Object.assign(out, {
                  format: e.format || 'csv',
@@@ -150,6 +162,7 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
              if (e.extra_lists) out.extra_lists = e.extra_lists;
              if (e.max_bytes !== undefined) out.max_bytes = e.max_bytes;
              if (e.timeout_ms !== undefined) out.timeout_ms = e.timeout_ms;
++            if (e.resolver !== undefined) out.resolver = e.resolver;
          } else {
              // A feed may read structured facts out of its items' prose (adapters/extract.js).
              for (const k of Object.keys(e)) if (k !== 'url' && !(k === 'extract' && (input.type === 'rss' || input.type === 'atom'))) throw new RegistryError(`endpoints[${i}].${k} is only for api sources`);
@@@ -242,6 +255,7 @@ function createRegistry({ db, now = () => Date.now(), maxItemsCap = 12000 }) {
          del: db.prepare('DELETE FROM sources WHERE key = ?'),
          itemCount: db.prepare('SELECT COUNT(*) AS n FROM items WHERE source_key = ?'),
          dropFrontier: db.prepare('DELETE FROM crawl_frontier WHERE source_key = ?'),
++        dropState: db.prepare('DELETE FROM source_state WHERE source_key = ?'),
          resetValidators: db.prepare('DELETE FROM endpoint_state WHERE source_key = ?'),
      };
  
@@@ -333,6 +347,7 @@ function createRegistry({ db, now = () => Date.now(), maxItemsCap = 12000 }) {
          if ((await st.itemCount.get(key)).n) throw new RegistryError('the source has items; disable it instead (provenance must stay resolvable)', 'sources.has_items');
          await st.del.run(key);
          await st.dropFrontier.run(key);
++        await st.dropState.run(key);
          return true;
      }
  
ddiff --git a/test/fixtures/off/delta-1.jsonl b/test/fixtures/off/delta-1.jsonl
nnew file mode 100644
iindex 0000000..d54b6c6
---- /dev/null
++++ b/test/fixtures/off/delta-1.jsonl
@@@ -0,0 +1,2 @@
++{"code":"0000000000011","product_name":"Large Eggs","brands":"Example Brand","nutriscore_grade":"b","nutrition":{"aggregated_set":{"nutrients":{"energy-kcal":{"value":143},"proteins":{"value":12.6},"fat":{"value":9.5},"saturated-fat":{"value":3.1},"carbohydrates":{"value":0.7},"sugars":{"value":0.4},"salt":{"value":0.3},"fiber":{"value":0}}}}}
++{"code":"0000000000012","product_name":"Free Range Large Eggs","brands":"Example Brand","nutriscore_grade":"a","nutrition":{"aggregated_set":{"nutrients":{"energy-kcal":{"value":140},"proteins":{"value":13},"fat":{"value":9.5},"saturated-fat":{"value":3},"carbohydrates":{"value":0.7},"sugars":{"value":0.4},"salt":{"value":0.3},"fiber":{"value":0}}}}}
ddiff --git a/test/fixtures/off/delta-2.jsonl b/test/fixtures/off/delta-2.jsonl
nnew file mode 100644
iindex 0000000..8c27902
---- /dev/null
++++ b/test/fixtures/off/delta-2.jsonl
@@@ -0,0 +1,2 @@
++{"code":"0000000000021","product_name":"Whole Milk","brands":"Dairy Co","nutriscore_grade":"c","nutrition":{"aggregated_set":{"nutrients":{"energy-kcal":{"value":64},"proteins":{"value":3.3},"fat":{"value":3.6},"saturated-fat":{"value":2.3},"carbohydrates":{"value":4.8},"sugars":{"value":4.8},"salt":{"value":0.1},"fiber":{"value":0}}}}}
++{"code":"0000000000022","product_name":"Cheddar Cheese Block","brands":"Dairy Co","nutriscore_grade":"d","nutrition":{"aggregated_set":{"nutrients":{"energy-kcal":{"value":402},"proteins":{"value":25},"fat":{"value":33},"saturated-fat":{"value":21},"carbohydrates":{"value":1.3},"sugars":{"value":0.5},"salt":{"value":1.8},"fiber":{"value":0}}}}}
ddiff --git a/test/fixtures/off/delta-3.jsonl b/test/fixtures/off/delta-3.jsonl
nnew file mode 100644
iindex 0000000..c3e1c9f
---- /dev/null
++++ b/test/fixtures/off/delta-3.jsonl
@@@ -0,0 +1 @@
++{"code":"0000000000031","product_name":"White Bread","brands":"Bakery","nutriscore_grade":"c","nutrition":{"aggregated_set":{"nutrients":{"energy-kcal":{"value":265},"proteins":{"value":9},"fat":{"value":3.2},"saturated-fat":{"value":0.7},"carbohydrates":{"value":49},"sugars":{"value":5},"salt":{"value":1.2},"fiber":{"value":2.7}}}}}
ddiff --git a/test/fixtures/off/delta-4.jsonl b/test/fixtures/off/delta-4.jsonl
nnew file mode 100644
iindex 0000000..5ce141f
---- /dev/null
++++ b/test/fixtures/off/delta-4.jsonl
@@@ -0,0 +1 @@
++{"code":"0000000000041","product_name":"Greek Yogurt","brands":"Dairy Co","nutriscore_grade":"b","nutrition":{"aggregated_set":{"nutrients":{"energy-kcal":{"value":97},"proteins":{"value":9},"fat":{"value":5},"saturated-fat":{"value":3.3},"carbohydrates":{"value":3.6},"sugars":{"value":3.2},"salt":{"value":0.1},"fiber":{"value":0}}}}}
ddiff --git a/test/fixtures/off/delta-index.txt b/test/fixtures/off/delta-index.txt
nnew file mode 100644
iindex 0000000..8e83f4d
---- /dev/null
++++ b/test/fixtures/off/delta-index.txt
@@@ -0,0 +1,3 @@
++openfoodfacts_products_1793200000_1793300000.json.gz
++openfoodfacts_products_1793100000_1793200000.json.gz
++openfoodfacts_products_1793000000_1793100000.json.gz
ddiff --git a/test/off-delta.test.js b/test/off-delta.test.js
nnew file mode 100644
iindex 0000000..8710be9
---- /dev/null
++++ b/test/off-delta.test.js
@@@ -0,0 +1,104 @@
++'use strict';
++/**
++ * Open Food Facts' official delta index (server/adapters/off-delta.js, the ingest off-delta branch):
++ *   - the index's parts are read oldest first;
++ *   - a run reads up to max_rows across parts and stops inside one, keeping its row offset;
++ *   - the next run skips those rows, finishes the part, and continues to the next one;
++ *   - a new part appearing at the top of the index is picked up next;
++ *   - the full export is never requested.
++ * Both the index and the delta files are recorded fixtures (gzipped in the stub); no live call here.
++ */
++const assert = require('assert');
++const zlib = require('zlib');
++const { boot, fixture, site, sourceDef, suite } = require('./helpers');
++const { parseIndex, readState, nextFiles } = require('../server/adapters/off-delta');
++
++const t = suite('off-delta');
++
++const NAMES = [
++    'openfoodfacts_products_1793000000_1793100000.json.gz',
++    'openfoodfacts_products_1793100000_1793200000.json.gz',
++    'openfoodfacts_products_1793200000_1793300000.json.gz',
++];
++const NEWEST = 'openfoodfacts_products_1793300000_1793400000.json.gz';
++const BODY = {
++    [NAMES[0]]: 'off/delta-1.jsonl',
++    [NAMES[1]]: 'off/delta-2.jsonl',
++    [NAMES[2]]: 'off/delta-3.jsonl',
++    [NEWEST]: 'off/delta-4.jsonl',
++};
++const indexBody = (names = NAMES) => names.join('\n') + '\n';
++
++t('the index parses oldest first, refuses a path or a name with no time range, and a cursor resumes a part', () => {
++    const files = parseIndex(fixture('off/delta-index.txt'), 'https://static.example.org/data/delta/index.txt');
++    assert.deepStrictEqual(files.map((f) => f.name), NAMES, 'oldest first, though the recorded index lists newest first');
++    assert.deepStrictEqual([files[0].start, files[0].end, files[0].url], [1793000000, 1793100000, `https://static.example.org/data/delta/${NAMES[0]}`]);
++    assert.throws(() => parseIndex('../evil_1793000000_1793100000.json.gz', 'https://x.example/data/delta/index.txt'), /unexpected file/);
++    assert.throws(() => parseIndex('openfoodfacts_products.json.gz', 'https://x.example/data/delta/index.txt'), /no start\/end/);
++    assert.throws(() => parseIndex('', 'https://x.example/data/delta/index.txt'), /lists no files/);
++    const state = readState(JSON.stringify({ seen: NAMES[0], seen_end: 1793100000, current: NAMES[1], offset: 1 }));
++    assert.deepStrictEqual(nextFiles(files, state).map((f) => f.name), [NAMES[1], NAMES[2]]);
++    assert.deepStrictEqual(readState('not json'), { seen: null, seen_end: 0, current: null, offset: 0 });
++});
++
++t('the first run reads old parts up to the limit and keeps the part and row offset; later runs continue', async () => {
++    const svc = await boot();
++    let index = indexBody([...NAMES].reverse());
++    const web = await site({
++        '/robots.txt': () => ({ body: 'User-agent: *\nAllow: /\n' }),
++        '/data/delta/index.txt': () => ({ headers: { 'Content-Type': 'text/plain' }, body: index }),
++        '*': (req, res, ctx) => {
++            const name = ctx.url.pathname.split('/').pop();
++            const file = BODY[name];
++            if (!file || !ctx.url.pathname.startsWith('/data/delta/')) return { status: 404, body: 'no such delta' };
++            return { headers: { 'Content-Type': 'application/octet-stream' }, body: zlib.gzipSync(fixture(file)) };
++        },
++    });
++    const stateOf = async () => JSON.parse((await svc.db.prepare('SELECT value FROM source_state WHERE source_key = ? AND name = ?').get('off-delta-test', 'delta.cursor')).value);
++    const count = async () => (await svc.db.prepare('SELECT COUNT(*) AS n FROM items WHERE source_key = ?').get('off-delta-test')).n;
++    try {
++        await svc.registry.create(sourceDef({
++            key: 'off-delta-test', type: 'bulk', category: 'food', enabled: true, max_items: 100,
++            endpoints: [{
++                url: `${web.origin}/data/delta/index.txt`, format: 'jsonl', resolver: 'off-delta', item_kind: 'food_product',
++                max_bytes: 1048576, max_rows: 3, timeout_ms: 10000,
++                fields: { identity: 'code', title: 'product_name' },
++                extra: {
++                    brands: 'brands', nutriscore_grade: 'nutriscore_grade',
++                    energy_kcal_100g: 'nutrition.aggregated_set.nutrients.energy-kcal.value',
++                    proteins_100g: 'nutrition.aggregated_set.nutrients.proteins.value',
++                },
++            }],
++        }), 'test');
++
++        const first = await svc.ingest.run('off-delta-test');
++        assert.deepStrictEqual([first.runs[0].state, first.runs[0].items_seen], ['ok', 3]);
++        assert.deepStrictEqual((await stateOf()), { seen: NAMES[0], seen_end: 1793100000, current: NAMES[1], offset: 1 });
++        assert.strictEqual(await count(), 3);
++        assert.strictEqual(web.hits(`/data/delta/${NAMES[0]}`).length, 1);
++        assert.strictEqual(web.hits(`/data/delta/${NAMES[1]}`).length, 1);
++        assert.strictEqual(web.hits(`/data/delta/${NAMES[2]}`).length, 0, 'the limit stopped before the third part');
++        assert.ok(!web.requests.some((r) => /products\.csv|products\.jsonl/.test(r.path)), 'the full export is never requested');
++        const eggs = JSON.parse((await svc.db.prepare('SELECT fields FROM items WHERE source_key = ? AND identity = ?').get('off-delta-test', '0000000000011')).fields);
++        assert.deepStrictEqual([eggs.brands, eggs.nutriscore_grade, eggs.energy_kcal_100g, eggs.proteins_100g], ['Example Brand', 'b', 143, 12.6]);
++
++        const second = await svc.ingest.run('off-delta-test');
++        assert.deepStrictEqual([second.runs[0].state, second.runs[0].items_seen], ['ok', 2]);
++        assert.deepStrictEqual((await stateOf()), { seen: NAMES[2], seen_end: 1793300000, current: null, offset: 0 });
++        assert.strictEqual(await count(), 5);
++        assert.strictEqual(web.hits(`/data/delta/${NAMES[1]}`).length, 2, 'the unfinished part is streamed again to skip its first row');
++        assert.strictEqual(web.hits(`/data/delta/${NAMES[2]}`).length, 1);
++
++        const third = await svc.ingest.run('off-delta-test');
++        assert.deepStrictEqual([third.runs[0].state, third.runs[0].items_seen], ['ok', 0]);
++        assert.strictEqual(web.hits(`/data/delta/${NAMES[2]}`).length, 1, 'nothing re-read when no part is new');
++
++        index = indexBody([NEWEST, ...NAMES]);
++        const fourth = await svc.ingest.run('off-delta-test');
++        assert.deepStrictEqual([fourth.runs[0].state, fourth.runs[0].items_seen], ['ok', 1]);
++        assert.strictEqual(await count(), 6);
++        assert.deepStrictEqual((await stateOf()), { seen: NEWEST, seen_end: 1793400000, current: null, offset: 0 });
++    } finally { await svc.stop(); await web.close(); }
++});
++
++t.run();
ddiff --git a/test/seeds.test.js b/test/seeds.test.js
iindex 68eec29..c1d6f77 100644
---- a/test/seeds.test.js
++++ b/test/seeds.test.js
@@@ -56,4 +56,22 @@ t('every seed creates through the registry and stays disabled', async () => {
      }
  });
  
++t('every stored seed can be enabled by the same PATCH an operator would send', async () => {
++    // The host keeps the normalized row the create path wrote, not the raw seed: a resolver row carries
++    // items_shape/items_path/fields/extra defaults that its own validation must accept when PATCH revalidates it.
++    const fresh = await require('./db').testDb();
++    try {
++        const registry = createRegistry({ db: fresh.db });
++        for (const s of seeds) {
++            await registry.create(s, 'seed');
++            const row = await registry.patch(s.key, { enabled: true }, 'seed-check');
++            assert.strictEqual(Boolean(row.enabled), true, `${s.key} cannot be enabled: the host's PATCH would fail`);
++        }
++        const enabled = await fresh.db.prepare('SELECT COUNT(*) AS n FROM sources WHERE enabled = 1').get();
++        assert.strictEqual(enabled.n, seeds.length, 'every seed must be enableable in a test database');
++    } finally {
++        await fresh.close();
++    }
++});
++
  t.run();