deploy: 755c0b7

.xtap-deployment.json

@@ -1,3 +1,3 @@
 {
-  "source_revision": "65f6c17c7e476c1888fff0d27ba56b4728a036e9"
+  "source_revision": "755c0b7c154599a59ec569189ebfa2fb70e13353"
 }

package.json

@@ -13,6 +13,7 @@
     "setup": "npm run build --workspace shared && npm run build --workspace setup && npm run start --workspace setup",
     "update": "npm run build --workspace shared && npm run build --workspace setup && npm run start --workspace setup -- update",
     "doctor": "npm run --silent build --workspace shared && npm run --silent build --workspace setup && npm run --silent start --workspace setup -- doctor",
+    "index:bootstrap": "npm run build --workspace shared && npm run build --workspace space && npm run index:bootstrap --workspace space",
     "format": "prettier --check .",
     "format:write": "prettier --write .",
     "lint": "eslint .",

scripts/deploy-space.sh

@@ -1,64 +1,103 @@
 #!/usr/bin/env bash
 # Deploy this repo to a Hugging Face Docker Space.
 #
-# Requires an HF token with write access to the target namespace (a personal
-# session via `hf auth login`, not an agent's propose-only token).
+# Requires an authenticated Hugging Face CLI session that can create repos and
+# Buckets. XTAP_STORAGE_TOKEN must be a fine-grained token with read/write
+# access to only the target dataset and index Bucket. Supplying it explicitly
+# authorizes this script to install it as the Space HF_TOKEN secret.
 #
 # Usage:
-#   scripts/deploy-space.sh <namespace>            # e.g. dutifuldev or osolmaz
-#   SPACE_REPO=<ns>/<name> DATASET_REPO=<ns>/<name> scripts/deploy-space.sh
+#   XTAP_STORAGE_TOKEN=... scripts/deploy-space.sh <namespace>
+#   SPACE_REPO=<ns>/<name> DATASET_REPO=<ns>/<name> INDEX_BUCKET=<ns>/<name> \
+#     XTAP_STORAGE_TOKEN=... scripts/deploy-space.sh
 set -euo pipefail
 
 NAMESPACE="${1:-${NAMESPACE:-}}"
 SPACE_REPO="${SPACE_REPO:-${NAMESPACE:?usage: deploy-space.sh <namespace>}/xtap-pool}"
 DATASET_REPO="${DATASET_REPO:-${NAMESPACE}/xtap-pool-data}"
+INDEX_BUCKET="${INDEX_BUCKET:-${NAMESPACE}/xtap-pool-bucket}"
 ALLOWED_USERS="${ALLOWED_USERS:-osolmaz}"
+LLM_MODEL="${LLM_MODEL:-zai-org/GLM-5.2:fireworks-ai}"
+TAXONOMY_VERSION="${TAXONOMY_VERSION:-1}"
 
 ROOT="$(cd "$(dirname "$0")/.." && pwd)"
 STAGE="$(mktemp -d)"
-trap 'rm -rf "$STAGE"' EXIT
+INDEX_WORK="$(mktemp -d)"
+trap 'rm -rf "$STAGE" "$INDEX_WORK"' EXIT
 
-echo "==> Creating repos (idempotent)"
+echo "==> Creating storage and Space resources (idempotent)"
 hf repos create "$DATASET_REPO" --repo-type dataset --private 2>/dev/null || true
+hf buckets create "$INDEX_BUCKET" --private --exist-ok
 # The Space itself is public: a private Space would put HF's repo-access gate
-# in front of the app, blocking friends who are on ALLOWED_USERS but not
-# Space collaborators. All data access is enforced in-app (OAuth allowlist +
-# pool tokens); anonymous visitors only see the sign-in page. To keep the
-# Space page private instead, add every friend as a Space collaborator.
+# in front of the app. Anonymous visitors still see only the sign-in page.
 hf repos create "$SPACE_REPO" --repo-type space --space-sdk docker 2>/dev/null || true
 
-echo "==> Staging Space contents"
-git -C "$ROOT" archive HEAD | tar -x -C "$STAGE"
-cp "$ROOT/space/hf-space-README.md" "$STAGE/README.md"
-rm -rf "$STAGE/docs" "$STAGE/extension"
+if [[ -z "${XTAP_STORAGE_TOKEN:-}" ]]; then
+  cat >&2 <<EOF
+XTAP_STORAGE_TOKEN is required before deployment.
+Create a fine-grained token with read/write access to exactly:
+  dataset: $DATASET_REPO
+  Bucket:  $INDEX_BUCKET
+Before rerunning, optionally import history so the first durable generation
+contains it:
+  scripts/seed-dataset.sh $DATASET_REPO <hf-username> ~/Downloads/xtap
+Then rerun this command with XTAP_STORAGE_TOKEN set. The script will bootstrap
+the durable index and install the same value as the Space HF_TOKEN.
+EOF
+  exit 2
+fi
 
-echo "==> Uploading to $SPACE_REPO"
-hf upload "$SPACE_REPO" "$STAGE" . --repo-type space --commit-message "deploy: $(git -C "$ROOT" rev-parse --short HEAD)"
+echo "==> Bootstrapping and verifying the durable index"
+DATA_DIR="$INDEX_WORK" \
+DATASET_REPO="$DATASET_REPO" \
+INDEX_BUCKET="$INDEX_BUCKET" \
+HF_TOKEN="$XTAP_STORAGE_TOKEN" \
+LLM_MODEL="$LLM_MODEL" \
+TAXONOMY_VERSION="$TAXONOMY_VERSION" \
+npm --prefix "$ROOT" run index:bootstrap
 
 echo "==> Setting Space secrets and variables"
-python3 - "$SPACE_REPO" "$DATASET_REPO" "$ALLOWED_USERS" <<'PY'
+SPACE_REPO="$SPACE_REPO" \
+DATASET_REPO="$DATASET_REPO" \
+INDEX_BUCKET="$INDEX_BUCKET" \
+ALLOWED_USERS="$ALLOWED_USERS" \
+LLM_MODEL="$LLM_MODEL" \
+TAXONOMY_VERSION="$TAXONOMY_VERSION" \
+XTAP_STORAGE_TOKEN="$XTAP_STORAGE_TOKEN" \
+python3 <<'PY'
+import os
 import secrets
-import sys
 
 from huggingface_hub import HfApi
 
-space, dataset, allowed = sys.argv[1:4]
+space = os.environ["SPACE_REPO"]
 api = HfApi()
 variables = dict(api.get_space_variables(space))
-api.add_space_variable(space, "DATASET_REPO", dataset)
-api.add_space_variable(space, "ALLOWED_USERS", allowed)
+for name in (
+    "DATASET_REPO",
+    "INDEX_BUCKET",
+    "ALLOWED_USERS",
+    "LLM_MODEL",
+    "TAXONOMY_VERSION",
+):
+    api.add_space_variable(space, name, os.environ[name])
+api.add_space_secret(space, "HF_TOKEN", os.environ["XTAP_STORAGE_TOKEN"])
 # Secrets cannot be listed back, so a sentinel variable marks that they were
-# set once. Never rotate silently: rotating logs everyone out / disconnects
+# set once. Never rotate silently: rotating logs everyone out or disconnects
 # every extension.
 if "SECRETS_INITIALIZED" not in variables:
     for name in ("POOL_SIGNING_SECRET", "SESSION_SECRET"):
         api.add_space_secret(space, name, secrets.token_hex(32))
     api.add_space_variable(space, "SECRETS_INITIALIZED", "1")
-print("Set DATASET_REPO, ALLOWED_USERS, POOL_SIGNING_SECRET, SESSION_SECRET.")
-print("Remaining manual steps:")
-print(f"  1. Create a fine-grained token with read/write access to {dataset} only,")
-print(f"     then: python3 -c \"from huggingface_hub import HfApi; HfApi().add_space_secret('{space}', 'HF_TOKEN', '<token>')\"")
-print(f"  2. Optionally import history: scripts/seed-dataset.sh {dataset} <hf-username> ~/Downloads/xtap")
+print("Set storage variables, contract variables, and the scoped HF_TOKEN.")
 PY
 
+echo "==> Staging Space contents"
+git -C "$ROOT" archive HEAD | tar -x -C "$STAGE"
+cp "$ROOT/space/hf-space-README.md" "$STAGE/README.md"
+rm -rf "$STAGE/docs" "$STAGE/extension"
+
+echo "==> Uploading to $SPACE_REPO"
+hf upload "$SPACE_REPO" "$STAGE" . --repo-type space --commit-message "deploy: $(git -C "$ROOT" rev-parse --short HEAD)"
+
 echo "==> Done. Space: https://huggingface.co/spaces/$SPACE_REPO"

shared/src/enrichment.ts

@@ -3,6 +3,8 @@ import { z } from "zod";
 import { dayKey } from "./tweet.js";
 import type { Tweet } from "./tweet.js";
 
+export const DEFAULT_ENRICHMENT_MODEL = "zai-org/GLM-5.2:fireworks-ai";
+
 /** One preset taxonomy entry; the description steers the classifier. */
 export const labelConfigSchema = z.object({
   name: z.string().min(1),

shared/src/index.ts

@@ -9,6 +9,7 @@ export {
   type TweetValidationResult,
 } from "./tweet.js";
 export {
+  DEFAULT_ENRICHMENT_MODEL,
   labelConfigSchema,
   enrichmentRowSchema,
   parseEnrichmentRow,

space/package.json

@@ -9,6 +9,7 @@
     "start": "node dist/src/server.js",
     "enrich": "node dist/src/enrich-command-main.js",
     "enrich:job": "node dist/src/enrich-job-main.js",
+    "index:bootstrap": "node dist/src/index-command-main.js",
     "format": "prettier --check .",
     "lint": "eslint .",
     "typecheck": "tsc -p tsconfig.typecheck.json --noEmit",

space/src/config.ts

@@ -1,9 +1,11 @@
+import { DEFAULT_ENRICHMENT_MODEL } from "@xtap-pool/shared";
 import { z } from "zod";
 
 const configSchema = z.object({
   PORT: z.coerce.number().int().positive().default(7860),
   DATA_DIR: z.string().default(".data"),
   DATASET_REPO: z.string().min(1),
+  INDEX_BUCKET: z.string().min(1),
   HF_TOKEN: z.string().min(1),
   INFERENCE_TOKEN: z.string().min(1).optional(),
   POOL_SIGNING_SECRET: z.string().min(32),
@@ -26,7 +28,7 @@ const configSchema = z.object({
   ENRICH_OUTPUT_TOKEN_USD: z.coerce.number().nonnegative().optional(),
   ENRICH_MAX_DISCARDED_ASSIGNMENTS_PER_UNIT: z.coerce.number().nonnegative().optional(),
   ENRICH_DISCARDED_ASSIGNMENT_RATE_MIN_UNITS: z.coerce.number().int().positive().optional(),
-  LLM_MODEL: z.string().min(1).default("zai-org/GLM-5.2"),
+  LLM_MODEL: z.string().min(1).default(DEFAULT_ENRICHMENT_MODEL),
   TAXONOMY_VERSION: z.coerce.number().int().min(1).default(1),
 });
 
@@ -34,6 +36,7 @@ export type SpaceConfig = {
   port: number;
   dataDir: string;
   datasetRepo: string;
+  indexBucket: string;
   hfToken: string;
   inferenceToken?: string;
   poolSigningSecret: string;
@@ -93,6 +96,7 @@ export function loadConfig(env: Record<string, string | undefined>): SpaceConfig
     port: parsed.PORT,
     dataDir: parsed.DATA_DIR,
     datasetRepo: parsed.DATASET_REPO,
+    indexBucket: parsed.INDEX_BUCKET,
     hfToken: parsed.HF_TOKEN,
     ...(parsed.INFERENCE_TOKEN === undefined ? {} : { inferenceToken: parsed.INFERENCE_TOKEN }),
     poolSigningSecret: parsed.POOL_SIGNING_SECRET,

space/src/dataset-token.ts

@@ -10,27 +10,32 @@ const TARGET_PERMISSIONS = new Set([
 const READ_PERMISSION = "repo.content.read";
 const WRITE_PERMISSIONS = ["repo.content.write", "repo.write"] as const;
 
+type StorageTarget = { kind: "bucket" | "dataset"; name: string };
+
 export type DatasetCredentialReadiness =
   | { credential: "ok" }
   | { credential: "invalid"; error: string }
   | { credential: "unknown"; error: string };
 
-/** Verify the Space HF_TOKEN can read and write only the configured dataset repo. */
+/** Verify the Space HF_TOKEN can read and write the configured dataset and index Bucket only. */
 export async function checkDatasetCredential(params: {
   token: string;
   datasetRepo: string;
+  indexBucket: string;
   fetchFn?: typeof fetch;
 }): Promise<DatasetCredentialReadiness> {
   try {
     const response = await (params.fetchFn ?? fetch)("https://huggingface.co/api/whoami-v2", {
       headers: { authorization: `Bearer ${params.token}` },
     });
-    if (!response.ok) {
-      return tokenStatusError("HF_TOKEN", response.status);
-    }
-    const errors = datasetTokenErrors(await response.json(), params.datasetRepo);
+    if (!response.ok) return tokenStatusError("HF_TOKEN", response.status);
+    const targets: readonly StorageTarget[] = [
+      { kind: "dataset", name: params.datasetRepo },
+      { kind: "bucket", name: params.indexBucket },
+    ];
+    const errors = storageTokenErrors(await response.json(), targets);
     if (errors.length > 0) return { credential: "invalid", error: errors.join(" ") };
-    return await checkDatasetDownload(params);
+    return await checkStorageReads(params);
   } catch (error) {
     return { credential: "unknown", error: errorMessage(error) };
   }
@@ -40,26 +45,37 @@ export function datasetCredentialOk(status: DatasetCredentialReadiness): boolean
   return status.credential === "ok";
 }
 
-async function checkDatasetDownload(params: {
+async function checkStorageReads(params: {
   token: string;
   datasetRepo: string;
+  indexBucket: string;
   fetchFn?: typeof fetch;
 }): Promise<DatasetCredentialReadiness> {
-  const response = await (params.fetchFn ?? fetch)(datasetProbeUrl(params.datasetRepo), {
-    headers: { authorization: `Bearer ${params.token}` },
-  });
-  if (response.ok || response.status === 404) return { credential: "ok" };
-  const error = `Hugging Face rejected a direct private-dataset download using HF_TOKEN (${String(response.status)}).`;
-  return response.status === 401 || response.status === 403
-    ? { credential: "invalid", error }
-    : { credential: "unknown", error };
-}
-
-function datasetProbeUrl(datasetRepo: string): string {
-  return `https://huggingface.co/datasets/${datasetRepo}/resolve/main/config/pool.json`;
+  const fetchFn = params.fetchFn ?? fetch;
+  const probes = [
+    {
+      url: `https://huggingface.co/datasets/${params.datasetRepo}/resolve/main/config/pool.json`,
+      label: "private-dataset download",
+    },
+    {
+      url: `https://huggingface.co/api/buckets/${params.indexBucket}`,
+      label: "private-Bucket read",
+    },
+  ];
+  for (const probe of probes) {
+    const response = await fetchFn(probe.url, {
+      headers: { authorization: `Bearer ${params.token}` },
+    });
+    if (response.ok || response.status === 404) continue;
+    const error = `Hugging Face rejected a direct ${probe.label} using HF_TOKEN (${String(response.status)}).`;
+    return response.status === 401 || response.status === 403
+      ? { credential: "invalid", error }
+      : { credential: "unknown", error };
+  }
+  return { credential: "ok" };
 }
 
-function datasetTokenErrors(payload: unknown, datasetRepo: string): string[] {
+function storageTokenErrors(payload: unknown, targets: readonly StorageTarget[]): string[] {
   const root = asRecord(payload);
   const accessToken = asRecord(asRecord(root["auth"])["accessToken"]);
   const fineGrained = asRecord(accessToken["fineGrained"]);
@@ -69,12 +85,15 @@ function datasetTokenErrors(payload: unknown, datasetRepo: string): string[] {
       ? []
       : [`HF_TOKEN role is '${role || "unknown"}', expected fine-grained.`];
   errors.push(...globalPermissionErrors(fineGrained));
-  const targetPermissions = scopedPermissionErrors(fineGrained, datasetRepo, errors);
-  if (!targetPermissions.has(READ_PERMISSION)) {
-    errors.push(`HF_TOKEN must include ${READ_PERMISSION} on ${datasetRepo}.`);
-  }
-  if (!WRITE_PERMISSIONS.some((permission) => targetPermissions.has(permission))) {
-    errors.push(`HF_TOKEN must include ${WRITE_PERMISSIONS.join(" or ")} on ${datasetRepo}.`);
+  const permissionsByTarget = scopedPermissionErrors(fineGrained, targets, errors);
+  for (const target of targets) {
+    const permissions = permissionsByTarget.get(targetKey(target)) ?? new Set<string>();
+    if (!permissions.has(READ_PERMISSION)) {
+      errors.push(`HF_TOKEN must include ${READ_PERMISSION} on ${target.name}.`);
+    }
+    if (!WRITE_PERMISSIONS.some((permission) => permissions.has(permission))) {
+      errors.push(`HF_TOKEN must include ${WRITE_PERMISSIONS.join(" or ")} on ${target.name}.`);
+    }
   }
   return errors;
 }
@@ -87,53 +106,51 @@ function globalPermissionErrors(fineGrained: JsonObject): string[] {
 
 function scopedPermissionErrors(
   fineGrained: JsonObject,
-  datasetRepo: string,
+  targets: readonly StorageTarget[],
   errors: string[],
-): Set<string> {
-  const targetPermissions = new Set<string>();
+): Map<string, Set<string>> {
+  const permissionsByTarget = new Map(
+    targets.map((target) => [targetKey(target), new Set<string>()]),
+  );
   for (const scope of array(fineGrained["scoped"])) {
-    collectScopePermissions(asRecord(scope), datasetRepo, targetPermissions, errors);
+    collectScopePermissions(asRecord(scope), targets, permissionsByTarget, errors);
   }
-  return targetPermissions;
+  return permissionsByTarget;
 }
 
 function collectScopePermissions(
   scope: JsonObject,
-  datasetRepo: string,
-  targetPermissions: Set<string>,
+  targets: readonly StorageTarget[],
+  permissionsByTarget: Map<string, Set<string>>,
   errors: string[],
 ): void {
   const entity = asRecord(scope["entity"]);
+  const target = targets.find((candidate) => matchesTarget(entity, candidate));
   for (const permission of strings(scope["permissions"])) {
-    if (matchesDataset(entity, datasetRepo)) {
-      recordTargetPermission(permission, targetPermissions, errors);
-    } else {
+    if (target === undefined) {
       errors.push(
-        `Unexpected permission outside ${datasetRepo} on HF_TOKEN: ${permission} on ${entityLabel(entity)}.`,
+        `Unexpected permission outside configured storage on HF_TOKEN: ${permission} on ${entityLabel(entity)}.`,
       );
+    } else if (!TARGET_PERMISSIONS.has(permission)) {
+      errors.push(`Unexpected permission on HF_TOKEN: ${permission}.`);
+    } else {
+      permissionsByTarget.get(targetKey(target))?.add(permission);
     }
   }
 }
 
-function recordTargetPermission(
-  permission: string,
-  targetPermissions: Set<string>,
-  errors: string[],
-): void {
-  if (TARGET_PERMISSIONS.has(permission)) targetPermissions.add(permission);
-  else errors.push(`Unexpected permission on HF_TOKEN: ${permission}.`);
-}
-
-function matchesDataset(entity: JsonObject, datasetRepo: string): boolean {
+function matchesTarget(entity: JsonObject, target: StorageTarget): boolean {
+  const type = text(entity["type"]).replace(/s$/u, "");
   return (
-    isDatasetEntity(entity) &&
-    entityCandidates(entity).some((candidate) => normalizeRepo(candidate) === datasetRepo)
+    type === target.kind &&
+    entityCandidates(entity).some(
+      (candidate) => normalizeName(candidate, target.kind) === target.name,
+    )
   );
 }
 
-function isDatasetEntity(entity: JsonObject): boolean {
-  const type = text(entity["type"]);
-  return type === "dataset" || type === "datasets";
+function targetKey(target: StorageTarget): string {
+  return `${target.kind}:${target.name}`;
 }
 
 function entityCandidates(entity: JsonObject): readonly string[] {
@@ -150,8 +167,8 @@ function entityLabel(entity: JsonObject): string {
   return `${kind}:${name}`;
 }
 
-function normalizeRepo(value: string): string {
-  return value.replace(/^datasets\//, "");
+function normalizeName(value: string, kind: StorageTarget["kind"]): string {
+  return value.replace(kind === "dataset" ? /^datasets\//u : /^buckets\//u, "");
 }
 
 function tokenStatusError(tokenName: "HF_TOKEN", status: number): DatasetCredentialReadiness {

space/src/dataset.ts

@@ -2,6 +2,7 @@ import { mkdirSync, readFileSync, rmSync, writeFileSync, existsSync } from "node
 import { dirname, isAbsolute, join, normalize, relative } from "node:path";
 
 import { commit, downloadFile, listFiles } from "@huggingface/hub";
+import { z } from "zod";
 
 import {
   attemptEventSchema,
@@ -31,6 +32,13 @@ export type EnrichmentRefresh = {
   receipt?: EnrichReceipt;
 };
 
+export type DatasetSourceKind = "tweet" | "enrichment" | "attempt" | "registry" | "receipt";
+
+export type AppliedDatasetSource = {
+  kind: DatasetSourceKind;
+  rows: number;
+};
+
 type EnrichmentShardUpdate = {
   path: string;
   content: string;
@@ -41,8 +49,31 @@ type EnrichmentReplayCounts = Pick<EnrichmentRefresh, "rows" | "attempts" | "reg
 
 const REFRESH_SHARDS_PER_KIND = 4;
 const REFRESH_ATTEMPTS = 2;
-
-function isNotFound(error: unknown): boolean {
+const legacyReceiptSchema = z
+  .object({
+    started_at: z.string().min(1),
+    finished_at: z.string().min(1),
+    units: z.number().int().nonnegative(),
+    calls: z.number().int().nonnegative(),
+    prompt_tokens: z.number().int().nonnegative(),
+    completion_tokens: z.number().int().nonnegative(),
+    failures: z.number().int().nonnegative(),
+  })
+  .strict();
+const legacyEnrichmentRowSchema = z
+  .object({
+    unit_id: z.string().min(1),
+    tweet_ids: z.array(z.string().min(1)).min(1),
+    labels: z.array(z.string()),
+    free_labels: z.array(z.string()),
+    concepts: z.array(z.unknown()),
+    model: z.string().min(1),
+    taxonomy_version: z.number().int().min(1),
+    enriched_at: z.string().min(1),
+  })
+  .loose();
+
+export function isHubNotFound(error: unknown): boolean {
   return (
     typeof error === "object" &&
     error !== null &&
@@ -54,18 +85,23 @@ function isNotFound(error: unknown): boolean {
 function isMissingDatasetFile(error: unknown, path: string): boolean {
   const message = error instanceof Error ? error.message : String(error);
   return (
-    isNotFound(error) ||
+    isHubNotFound(error) ||
     message.includes(`missing: ${path}`) ||
     message.includes(`dataset file not found: ${path}`)
   );
 }
 
-async function assertDatasetRepoReadable(
+export async function assertDatasetRepoReadable(
   repo: { type: "dataset"; name: string },
   accessToken: string,
+  revision?: string,
 ): Promise<void> {
   try {
-    for await (const _entry of listFiles({ repo, accessToken })) {
+    for await (const _entry of listFiles({
+      repo,
+      accessToken,
+      ...(revision === undefined ? {} : { revision }),
+    })) {
       return;
     }
   } catch (error) {
@@ -86,7 +122,7 @@ export function createHubClient(datasetRepo: string, accessToken: string): HubCl
           if (entry.type === "file" && entry.path.endsWith(".jsonl")) paths.push(entry.path);
         }
       } catch (error) {
-        if (isNotFound(error)) {
+        if (isHubNotFound(error)) {
           // A fresh pool can lack the requested tree. Verify the repo itself is
           // readable so auth failures do not look like an empty dataset.
           await assertDatasetRepoReadable(repo, accessToken);
@@ -103,7 +139,7 @@ export function createHubClient(datasetRepo: string, accessToken: string): HubCl
         await assertDatasetRepoReadable(repo, accessToken);
         throw new Error(`dataset file not found: ${path}`);
       } catch (error) {
-        if (isNotFound(error)) await assertDatasetRepoReadable(repo, accessToken);
+        if (isHubNotFound(error)) await assertDatasetRepoReadable(repo, accessToken);
         throw error;
       }
     },
@@ -354,6 +390,40 @@ export class DatasetMirror {
     return count;
   }
 
+  /** Apply one complete file or verified append suffix to the SQLite projection. */
+  applySourceContent(
+    path: string,
+    content: string,
+    store: TweetStore,
+    enrich: EnrichStore,
+  ): AppliedDatasetSource {
+    const kind = datasetSourceKind(path);
+    switch (kind) {
+      case "tweet": {
+        const tweets = parseJsonlTweets(content, path);
+        store.insert(tweets);
+        enrich.registerTweets(tweets);
+        return { kind, rows: tweets.length };
+      }
+      case "enrichment":
+        return { kind, rows: applyEnrichmentLines(enrich, content) };
+      case "attempt":
+        return { kind, rows: replayAttemptLines(enrich, content) };
+      case "registry":
+        return { kind, rows: replayRegistryLines(enrich, content) };
+      case "receipt":
+        this.recordLatestReceipt(content);
+        return { kind, rows: countValidReceipts(content) };
+    }
+  }
+
+  /** Keep a current source file locally so a later append preserves its prefix. */
+  rememberSourceFile(path: string, content: string): void {
+    const local = this.localPath(path);
+    mkdirSync(dirname(local), { recursive: true });
+    writeFileSync(local, content);
+  }
+
   /** Read a dataset file through the Hub, returning undefined when it is absent. */
   async readText(path: string): Promise<string | undefined> {
     try {
@@ -381,10 +451,13 @@ export class DatasetMirror {
     writes: readonly { path: string; content: string }[],
     title: string,
   ): Promise<void> {
-    const files = [
-      ...appends.map(({ path, lines }) => ({ path, content: this.appendedContent(path, lines) })),
-      ...writes,
-    ];
+    const appended = await Promise.all(
+      appends.map(async ({ path, lines }) => ({
+        path,
+        content: await this.appendedContent(path, lines),
+      })),
+    );
+    const files = [...appended, ...writes];
     await this.hub.commitFiles(files, title);
     for (const file of files) {
       const local = this.localPath(file.path);
@@ -393,9 +466,11 @@ export class DatasetMirror {
     }
   }
 
-  private appendedContent(path: string, lines: readonly string[]): string {
+  private async appendedContent(path: string, lines: readonly string[]): Promise<string> {
     const local = this.localPath(path);
-    const existing = existsSync(local) ? readFileSync(local, "utf8") : "";
+    const existing = existsSync(local)
+      ? readFileSync(local, "utf8")
+      : ((await this.readText(path)) ?? "");
     const prefix = existing === "" || existing.endsWith("\n") ? existing : `${existing}\n`;
     return `${prefix}${lines.map((line) => `${line}\n`).join("")}`;
   }
@@ -427,6 +502,62 @@ export class DatasetMirror {
  */
 type EnrichmentShardKind = "receipt" | "attempt" | "registry" | "row";
 
+export function assertValidDatasetSourceContent(path: string, content: string): void {
+  const kind = datasetSourceKind(path);
+  const lines = content.split("\n");
+  for (const [index, line] of lines.entries()) {
+    if (line.trim() === "") continue;
+    let candidate: unknown;
+    try {
+      candidate = JSON.parse(line);
+    } catch {
+      throw new Error(`invalid JSON in ${path} at line ${String(index + 1)}`);
+    }
+    if (!validDatasetSourceRecord(kind, candidate)) {
+      throw new Error(`invalid ${kind} record in ${path} at line ${String(index + 1)}`);
+    }
+  }
+}
+
+function validDatasetSourceRecord(kind: DatasetSourceKind, candidate: unknown): boolean {
+  switch (kind) {
+    case "tweet":
+      return validateTweet(candidate).ok;
+    case "enrichment":
+      return parseEnrichmentRow(candidate) !== undefined || isLegacyEnrichmentRow(candidate);
+    case "attempt":
+      return attemptEventSchema.safeParse(candidate).success;
+    case "registry":
+      return freeLabelEventSchema.safeParse(candidate).success;
+    case "receipt":
+      return (
+        parseEnrichReceipt(candidate) !== undefined ||
+        legacyReceiptSchema.safeParse(candidate).success
+      );
+  }
+}
+
+function isLegacyEnrichmentRow(candidate: unknown): boolean {
+  return legacyEnrichmentRowSchema.safeParse(candidate).success;
+}
+
+export function datasetSourceKind(path: string): DatasetSourceKind {
+  if (/^data\/[^/]+\/\d{4}\/\d{2}\/tweets-\d{4}-\d{2}-\d{2}\.jsonl$/u.test(path)) {
+    return "tweet";
+  }
+  if (/^enrichment\/\d{4}\/\d{2}\/enrichment-\d{4}-\d{2}-\d{2}\.jsonl$/u.test(path)) {
+    return "enrichment";
+  }
+  if (/^enrichment\/attempts\/\d{4}\/\d{2}\/attempts-\d{4}-\d{2}-\d{2}\.jsonl$/u.test(path)) {
+    return "attempt";
+  }
+  if (/^enrichment\/registry\/\d{4}\/\d{2}\/registry-\d{4}-\d{2}-\d{2}\.jsonl$/u.test(path)) {
+    return "registry";
+  }
+  if (/^enrichment\/receipts\/\d{4}-\d{2}-\d{2}\.jsonl$/u.test(path)) return "receipt";
+  throw new Error(`unsupported dataset index source: ${path}`);
+}
+
 function classifyEnrichmentPath(path: string): EnrichmentShardKind {
   if (path.startsWith("enrichment/receipts/")) return "receipt";
   if (path.startsWith("enrichment/attempts/")) return "attempt";
@@ -450,6 +581,19 @@ function selectEnrichmentRefreshShards(
   return [...selected].sort();
 }
 
+function countValidReceipts(content: string): number {
+  let count = 0;
+  for (const line of content.split("\n")) {
+    if (line.trim() === "") continue;
+    try {
+      if (parseEnrichReceipt(JSON.parse(line)) !== undefined) count += 1;
+    } catch {
+      continue;
+    }
+  }
+  return count;
+}
+
 function applyEnrichmentLines(enrich: EnrichStore, content: string): number {
   let rows = 0;
   for (const line of content.split("\n")) {

space/src/durable-index.ts

@@ -0,0 +1,887 @@
+import { createHash, randomUUID } from "node:crypto";
+import { createReadStream, createWriteStream, existsSync, openAsBlob } from "node:fs";
+import { mkdir, rename, rm } from "node:fs/promises";
+import { dirname } from "node:path";
+import { Readable } from "node:stream";
+import { pipeline } from "node:stream/promises";
+
+import Database from "better-sqlite3";
+import {
+  commit,
+  datasetInfo,
+  deleteFiles,
+  downloadFile,
+  listFiles,
+  uploadFile,
+} from "@huggingface/hub";
+import { z } from "zod";
+
+import {
+  assertDatasetRepoReadable,
+  assertValidDatasetSourceContent,
+  DatasetMirror,
+  datasetSourceKind,
+  isHubNotFound,
+} from "./dataset.js";
+import type { DatasetSourceKind } from "./dataset.js";
+import { EnrichStore } from "./enrich-store.js";
+import { TweetStore } from "./store.js";
+
+const INDEX_SCHEMA_VERSION = 1;
+const CURRENT_MANIFEST_KEY = "index/current.json";
+const DATABASE_PREFIX = "index/databases";
+const RETAINED_PREDECESSORS = 3;
+const MAX_PUBLICATION_ATTEMPTS = 5;
+const SHA256_PATTERN = /^[a-f0-9]{64}$/;
+const REVISION_PATTERN = /^[a-f0-9]{7,64}$/;
+const DATABASE_KEY_PATTERN = /^index\/databases\/[a-f0-9]{64}\.sqlite$/;
+
+const indexCountsSchema = z
+  .object({
+    tweets: z.number().int().nonnegative(),
+    units: z.number().int().nonnegative(),
+    enrichments: z.number().int().nonnegative(),
+    attempt_events: z.number().int().nonnegative(),
+    registry_events: z.number().int().nonnegative(),
+  })
+  .strict();
+
+export const durableIndexManifestSchema = z
+  .object({
+    schema_version: z.literal(INDEX_SCHEMA_VERSION),
+    dataset: z
+      .object({ repo: z.string().min(1), revision: z.string().regex(REVISION_PATTERN) })
+      .strict(),
+    projection: z.object({ contract_hash: z.string().regex(SHA256_PATTERN) }).strict(),
+    database: z
+      .object({
+        key: z.string().regex(DATABASE_KEY_PATTERN),
+        sha256: z.string().regex(SHA256_PATTERN),
+        predecessors: z
+          .array(z.string().regex(DATABASE_KEY_PATTERN))
+          .max(RETAINED_PREDECESSORS)
+          .refine((keys) => new Set(keys).size === keys.length, "predecessors must be unique"),
+      })
+      .strict(),
+    counts: indexCountsSchema,
+  })
+  .strict()
+  .superRefine((manifest, context) => {
+    const expected = `${DATABASE_PREFIX}/${manifest.database.sha256}.sqlite`;
+    if (manifest.database.key !== expected) {
+      context.addIssue({
+        code: "custom",
+        path: ["database", "key"],
+        message: `database key must be ${expected}`,
+      });
+    }
+    if (manifest.database.predecessors.includes(manifest.database.key)) {
+      context.addIssue({
+        code: "custom",
+        path: ["database", "predecessors"],
+        message: "predecessors must not include the active database",
+      });
+    }
+  });
+
+export type DurableIndexManifest = z.infer<typeof durableIndexManifestSchema>;
+export type DurableIndexCounts = z.infer<typeof indexCountsSchema>;
+
+export type DatasetSourceFile = {
+  path: string;
+  oid: string;
+  size: number;
+};
+
+export type DatasetSnapshotClient = {
+  currentRevision(): Promise<string>;
+  listJsonlFiles(revision: string): Promise<readonly DatasetSourceFile[]>;
+  downloadFile(path: string, revision: string): Promise<Uint8Array>;
+  readText(path: string, revision: string): Promise<string | undefined>;
+  commitText(path: string, content: string, parentRevision: string): Promise<string>;
+};
+
+export type BucketFile = { path: string; uploadedAt?: string };
+
+export type DurableIndexBucketClient = {
+  download(path: string, destination: string): Promise<boolean>;
+  uploadFile(path: string, source: string): Promise<void>;
+  list(prefix: string): Promise<readonly BucketFile[]>;
+  remove(paths: readonly string[]): Promise<void>;
+};
+
+type SourceFileRow = {
+  path: string;
+  kind: DatasetSourceKind;
+  oid: string;
+  byte_length: number;
+  content_sha256: string;
+  row_count: number;
+};
+
+type IndexMetadataRow = {
+  schema_version: number;
+  dataset_repo: string;
+  dataset_revision: string;
+  contract_hash: string;
+};
+
+type StagedSourceFile = {
+  current: DatasetSourceFile;
+  kind: DatasetSourceKind;
+  fullContent: Uint8Array;
+  fullText: string;
+  suffixText: string;
+  contentSha256: string;
+  previousRows: number;
+};
+
+export type DurableIndexOptions = {
+  datasetRepo: string;
+  indexBucket: string;
+  accessToken: string;
+  databasePath: string;
+  mirror: DatasetMirror;
+  taxonomyVersion: number;
+  contractHash: string;
+  sourceClient?: DatasetSnapshotClient;
+  bucketClient?: DurableIndexBucketClient;
+  predecessorKeys?: readonly string[];
+};
+
+export type IndexAdvance = {
+  revision: string;
+  filesChanged: number;
+  rowsApplied: number;
+  counts: DurableIndexCounts;
+};
+
+export type DurableIndexStats = {
+  tweetFiles: number;
+  tweetRows: number;
+  enrichmentFiles: number;
+  enrichmentRows: number;
+  attemptEvents: number;
+  registryEvents: number;
+};
+
+/**
+ * Restored local SQLite projection backed by an immutable Bucket generation.
+ * The dataset stays authoritative; this class only advances strict append-only
+ * JSONL inputs and publishes verified replacement generations.
+ */
+export class DurableIndex {
+  readonly store: TweetStore;
+  readonly enrichStore: EnrichStore;
+
+  private constructor(
+    private readonly options: DurableIndexOptions,
+    private readonly source: DatasetSnapshotClient,
+    private readonly bucket: DurableIndexBucketClient,
+    store: TweetStore,
+    enrichStore: EnrichStore,
+    private publishedKeys: string[],
+  ) {
+    this.store = store;
+    this.enrichStore = enrichStore;
+  }
+
+  static async restore(options: DurableIndexOptions): Promise<DurableIndex> {
+    const source =
+      options.sourceClient ?? createDatasetSnapshotClient(options.datasetRepo, options.accessToken);
+    const bucket =
+      options.bucketClient ??
+      createDurableIndexBucketClient(options.indexBucket, options.accessToken);
+    const manifestRevision = await source.currentRevision();
+    const rawManifest = await source.readText(CURRENT_MANIFEST_KEY, manifestRevision);
+    if (rawManifest === undefined) {
+      throw new Error("durable index manifest is missing; run the index bootstrap command");
+    }
+    const manifest = parseManifest(rawManifest, options);
+    await mkdir(dirname(options.databasePath), { recursive: true });
+    const staged = `${options.databasePath}.${randomUUID()}.download`;
+    try {
+      if (!(await bucket.download(manifest.database.key, staged))) {
+        throw new Error(`durable index database is missing: ${manifest.database.key}`);
+      }
+      await assertFileSha256(staged, manifest.database.sha256);
+      await removeDatabaseFiles(options.databasePath);
+      await rename(staged, options.databasePath);
+    } finally {
+      await rm(staged, { force: true });
+    }
+    const store = new TweetStore(options.databasePath);
+    const enrichStore = new EnrichStore(
+      store.database,
+      options.taxonomyVersion,
+      (): Date => new Date(),
+      options.contractHash,
+    );
+    ensureIndexTables(store.database);
+    validateDatabase(store.database, manifest, options);
+    const index = new DurableIndex(options, source, bucket, store, enrichStore, [
+      manifest.database.key,
+      ...manifest.database.predecessors,
+    ]);
+    await index.loadLatestReceipt(manifest.dataset.revision);
+    return index;
+  }
+
+  static openLocal(options: DurableIndexOptions): DurableIndex {
+    if (!existsSync(options.databasePath)) {
+      throw new Error(`durable index working database is missing: ${options.databasePath}`);
+    }
+    const source =
+      options.sourceClient ?? createDatasetSnapshotClient(options.datasetRepo, options.accessToken);
+    const bucket =
+      options.bucketClient ??
+      createDurableIndexBucketClient(options.indexBucket, options.accessToken);
+    const store = new TweetStore(options.databasePath);
+    const enrichStore = new EnrichStore(
+      store.database,
+      options.taxonomyVersion,
+      (): Date => new Date(),
+      options.contractHash,
+    );
+    ensureIndexTables(store.database);
+    assertDatabaseIntegrity(store.database);
+    const metadata = readMetadata(store.database);
+    if (
+      metadata?.schema_version !== INDEX_SCHEMA_VERSION ||
+      metadata.dataset_repo !== options.datasetRepo ||
+      metadata.contract_hash !== options.contractHash
+    ) {
+      store.close();
+      throw new Error("durable index working database provenance mismatch");
+    }
+    return new DurableIndex(
+      options,
+      source,
+      bucket,
+      store,
+      enrichStore,
+      configuredPredecessors(options),
+    );
+  }
+
+  static async bootstrap(options: DurableIndexOptions): Promise<DurableIndex> {
+    const source =
+      options.sourceClient ?? createDatasetSnapshotClient(options.datasetRepo, options.accessToken);
+    const bucket =
+      options.bucketClient ??
+      createDurableIndexBucketClient(options.indexBucket, options.accessToken);
+    await mkdir(dirname(options.databasePath), { recursive: true });
+    await removeDatabaseFiles(options.databasePath);
+    const store = new TweetStore(options.databasePath);
+    const enrichStore = new EnrichStore(
+      store.database,
+      options.taxonomyVersion,
+      (): Date => new Date(),
+      options.contractHash,
+    );
+    ensureIndexTables(store.database);
+    const index = new DurableIndex(options, source, bucket, store, enrichStore, []);
+    await index.advanceToLatest();
+    return index;
+  }
+
+  async advanceToLatest(): Promise<IndexAdvance> {
+    return this.advanceToRevision(await this.source.currentRevision());
+  }
+
+  async advanceToRevision(revision: string): Promise<IndexAdvance> {
+    if (!REVISION_PATTERN.test(revision)) throw new Error(`invalid dataset revision: ${revision}`);
+    const currentFiles = [...(await this.source.listJsonlFiles(revision))].sort((a, b) =>
+      sourceOrder(a.path, b.path),
+    );
+    const previous = sourceFileRows(this.store.database);
+    const currentPaths = new Set(currentFiles.map((file) => file.path));
+    const deleted = [...previous.keys()].filter((path) => !currentPaths.has(path));
+    if (deleted.length > 0) {
+      throw new Error(`dataset source files were deleted: ${deleted.slice(0, 3).join(", ")}`);
+    }
+    const staged: StagedSourceFile[] = [];
+    for (const file of currentFiles) {
+      const old = previous.get(file.path);
+      if (old?.oid === file.oid && old.byte_length === file.size) continue;
+      staged.push(await this.stageSourceFile(file, old, revision));
+    }
+
+    let rowsApplied = 0;
+    const apply = this.store.database.transaction(() => {
+      for (const file of staged) {
+        const applied = this.options.mirror.applySourceContent(
+          file.current.path,
+          file.suffixText,
+          this.store,
+          this.enrichStore,
+        );
+        if (applied.kind !== file.kind)
+          throw new Error(`source kind changed: ${file.current.path}`);
+        rowsApplied += applied.rows;
+        upsertSourceFile(this.store.database, {
+          path: file.current.path,
+          kind: file.kind,
+          oid: file.current.oid,
+          byte_length: file.fullContent.byteLength,
+          content_sha256: file.contentSha256,
+          row_count: file.previousRows + applied.rows,
+        });
+      }
+      writeMetadata(this.store.database, {
+        schema_version: INDEX_SCHEMA_VERSION,
+        dataset_repo: this.options.datasetRepo,
+        dataset_revision: revision,
+        contract_hash: this.options.contractHash,
+      });
+    });
+    apply();
+    for (const file of staged) {
+      this.options.mirror.rememberSourceFile(file.current.path, file.fullText);
+    }
+    await this.loadLatestReceipt(revision, currentFiles);
+    const counts = databaseCounts(this.store.database);
+    assertDatabaseIntegrity(this.store.database);
+    return { revision, filesChanged: staged.length, rowsApplied, counts };
+  }
+
+  stats(): DurableIndexStats {
+    const rows = this.store.database
+      .prepare(
+        `SELECT kind, COUNT(*) AS files, COALESCE(SUM(row_count), 0) AS rows
+         FROM source_files GROUP BY kind`,
+      )
+      .all() as { kind: DatasetSourceKind; files: number; rows: number }[];
+    const byKind = new Map(rows.map((row) => [row.kind, row]));
+    const tweets = sourceKindStats(byKind, "tweet");
+    const enrichments = sourceKindStats(byKind, "enrichment");
+    return {
+      tweetFiles: tweets.files,
+      tweetRows: tweets.rows,
+      enrichmentFiles: enrichments.files,
+      enrichmentRows: enrichments.rows,
+      attemptEvents: sourceKindStats(byKind, "attempt").rows,
+      registryEvents: sourceKindStats(byKind, "registry").rows,
+    };
+  }
+
+  retainedDatabaseKeys(): readonly string[] {
+    return [...this.publishedKeys];
+  }
+
+  async createWorkingCopy(path: string): Promise<void> {
+    await rm(path, { force: true });
+    await this.store.database.backup(path);
+  }
+
+  async publishLatest(): Promise<{
+    advance: IndexAdvance;
+    manifest: DurableIndexManifest;
+  }> {
+    for (let attempt = 1; attempt <= MAX_PUBLICATION_ATTEMPTS; attempt += 1) {
+      const advance = await this.advanceToLatest();
+      try {
+        return { advance, manifest: await this.publish() };
+      } catch (error) {
+        if (attempt === MAX_PUBLICATION_ATTEMPTS || !isConcurrentDatasetUpdate(error)) throw error;
+      }
+    }
+    throw new Error("durable index publication retry invariant failed");
+  }
+
+  async publish(): Promise<DurableIndexManifest> {
+    const metadata = readMetadata(this.store.database);
+    if (metadata === undefined) throw new Error("durable index metadata is missing");
+    assertDatabaseIntegrity(this.store.database);
+    const counts = databaseCounts(this.store.database);
+    const publicationPath = `${this.options.databasePath}.${randomUUID()}.publish`;
+    const verificationPath = `${this.options.databasePath}.${randomUUID()}.verify`;
+    try {
+      await this.store.database.backup(publicationPath);
+      validateStandaloneDatabase(publicationPath, metadata, counts);
+      const sha256 = await fileSha256(publicationPath);
+      const key = `${DATABASE_PREFIX}/${sha256}.sqlite`;
+      await this.bucket.uploadFile(key, publicationPath);
+      if (!(await this.bucket.download(key, verificationPath))) {
+        throw new Error(`uploaded durable index database is unavailable: ${key}`);
+      }
+      await assertFileSha256(verificationPath, sha256);
+      validateStandaloneDatabase(verificationPath, metadata, counts);
+      const predecessors = this.publishedKeys
+        .filter((publishedKey) => publishedKey !== key)
+        .slice(0, RETAINED_PREDECESSORS);
+      const manifest: DurableIndexManifest = {
+        schema_version: INDEX_SCHEMA_VERSION,
+        dataset: { repo: metadata.dataset_repo, revision: metadata.dataset_revision },
+        projection: { contract_hash: metadata.contract_hash },
+        database: { key, sha256, predecessors },
+        counts,
+      };
+      const encoded = `${JSON.stringify(manifest, null, 2)}\n`;
+      const manifestRevision = await this.source.commitText(
+        CURRENT_MANIFEST_KEY,
+        encoded,
+        metadata.dataset_revision,
+      );
+      const active = await this.source.readText(CURRENT_MANIFEST_KEY, manifestRevision);
+      if (
+        active === undefined ||
+        JSON.stringify(parseManifest(active, this.options)) !== JSON.stringify(manifest)
+      ) {
+        throw new Error("durable index manifest read-back did not match the published generation");
+      }
+      this.publishedKeys = [key, ...predecessors];
+      await this.pruneDatabases(this.publishedKeys);
+      return manifest;
+    } finally {
+      await Promise.all([
+        rm(publicationPath, { force: true }),
+        rm(verificationPath, { force: true }),
+      ]);
+    }
+  }
+
+  close(): void {
+    this.store.close();
+  }
+
+  private async stageSourceFile(
+    current: DatasetSourceFile,
+    previous: SourceFileRow | undefined,
+    revision: string,
+  ): Promise<StagedSourceFile> {
+    const kind = datasetSourceKind(current.path);
+    assertSourceKind(current.path, kind, previous);
+    const content = await this.source.downloadFile(current.path, revision);
+    assertCompletePinnedSource(current, content);
+    const suffix = sourceSuffix(current.path, content, previous);
+    const decoder = new TextDecoder("utf-8", { fatal: true });
+    const fullText = decoder.decode(content);
+    const suffixText = decoder.decode(suffix);
+    assertValidDatasetSourceContent(current.path, suffixText);
+    return {
+      current,
+      kind,
+      fullContent: content,
+      fullText,
+      suffixText,
+      contentSha256: sha256Bytes(content),
+      previousRows: previous?.row_count ?? 0,
+    };
+  }
+
+  private async loadLatestReceipt(
+    revision: string,
+    knownFiles?: readonly DatasetSourceFile[],
+  ): Promise<void> {
+    const files = knownFiles ?? (await this.source.listJsonlFiles(revision));
+    const latest = files
+      .filter((file) => datasetSourceKind(file.path) === "receipt")
+      .sort((a, b) => a.path.localeCompare(b.path))
+      .at(-1);
+    if (latest === undefined) return;
+    const content = await this.source.downloadFile(latest.path, revision);
+    const text = new TextDecoder("utf-8", { fatal: true }).decode(content);
+    this.options.mirror.applySourceContent(latest.path, text, this.store, this.enrichStore);
+    this.options.mirror.rememberSourceFile(latest.path, text);
+  }
+
+  private async pruneDatabases(retainedKeys: readonly string[]): Promise<void> {
+    const files = [...(await this.bucket.list(DATABASE_PREFIX))].filter((file) =>
+      file.path.endsWith(".sqlite"),
+    );
+    const keep = new Set(retainedKeys);
+    const stale = files.map((file) => file.path).filter((path) => !keep.has(path));
+    if (stale.length > 0) await this.bucket.remove(stale);
+  }
+}
+
+function configuredPredecessors(options: DurableIndexOptions): string[] {
+  return Array.from(options.predecessorKeys ?? []);
+}
+
+async function removeDatabaseFiles(path: string): Promise<void> {
+  await Promise.all([
+    rm(path, { force: true }),
+    rm(`${path}-wal`, { force: true }),
+    rm(`${path}-shm`, { force: true }),
+  ]);
+}
+
+function isConcurrentDatasetUpdate(error: unknown): boolean {
+  const statusCode =
+    typeof error === "object" && error !== null && "statusCode" in error
+      ? (error as { statusCode?: unknown }).statusCode
+      : undefined;
+  const message = error instanceof Error ? error.message.toLowerCase() : "";
+  return (
+    statusCode === 409 ||
+    message.includes("branch was updated") ||
+    message.includes("parent commit does not match")
+  );
+}
+
+function assertSourceKind(
+  path: string,
+  kind: DatasetSourceKind,
+  previous: SourceFileRow | undefined,
+): void {
+  if (previous !== undefined && previous.kind !== kind) {
+    throw new Error(`dataset source kind changed: ${path}`);
+  }
+}
+
+function assertCompletePinnedSource(current: DatasetSourceFile, content: Uint8Array): void {
+  if (content.byteLength !== current.size) {
+    throw new Error(`dataset source size changed while pinned: ${current.path}`);
+  }
+  if (content.byteLength > 0 && content.at(-1) !== 0x0a) {
+    throw new Error(`dataset source does not end with a complete JSONL line: ${current.path}`);
+  }
+}
+
+function sourceSuffix(
+  path: string,
+  content: Uint8Array,
+  previous: SourceFileRow | undefined,
+): Uint8Array {
+  if (previous === undefined) return content;
+  if (content.byteLength < previous.byte_length) {
+    throw new Error(`dataset source was truncated: ${path}`);
+  }
+  const prefix = content.subarray(0, previous.byte_length);
+  if (sha256Bytes(prefix) !== previous.content_sha256) {
+    throw new Error(`dataset source prefix changed: ${path}`);
+  }
+  return content.subarray(previous.byte_length);
+}
+
+export function createDatasetSnapshotClient(
+  datasetRepo: string,
+  accessToken: string,
+): DatasetSnapshotClient {
+  const repo = { type: "dataset", name: datasetRepo } as const;
+  return {
+    async currentRevision(): Promise<string> {
+      const info = await datasetInfo({
+        name: datasetRepo,
+        accessToken,
+        additionalFields: ["sha"],
+      });
+      if (typeof info.sha !== "string" || !REVISION_PATTERN.test(info.sha)) {
+        throw new Error(`dataset ${datasetRepo} did not return a valid revision`);
+      }
+      return info.sha;
+    },
+    async listJsonlFiles(revision: string): Promise<readonly DatasetSourceFile[]> {
+      const groups = await Promise.all(
+        ["data", "enrichment"].map((prefix) =>
+          listDatasetPrefix(repo, accessToken, revision, prefix),
+        ),
+      );
+      return groups.flat();
+    },
+    async downloadFile(path: string, revision: string): Promise<Uint8Array> {
+      const blob = await downloadFile({ repo, accessToken, path, revision });
+      if (blob === null) throw new Error(`dataset source is missing: ${path}`);
+      return new Uint8Array(await blob.arrayBuffer());
+    },
+    async readText(path: string, revision: string): Promise<string | undefined> {
+      const blob = await downloadFile({ repo, accessToken, path, revision });
+      return blob === null ? undefined : blob.text();
+    },
+    async commitText(path: string, content: string, parentRevision: string): Promise<string> {
+      const result = await commit({
+        repo,
+        accessToken,
+        parentCommit: parentRevision,
+        title: "Publish durable enrichment index manifest",
+        operations: [{ operation: "addOrUpdate", path, content: new Blob([content]) }],
+      });
+      if (result === undefined || !REVISION_PATTERN.test(result.commit.oid)) {
+        throw new Error("dataset did not confirm the durable index manifest commit");
+      }
+      return result.commit.oid;
+    },
+  };
+}
+
+// eslint-disable-next-line complexity -- Hub entries require type, path, and immutable-object validation.
+async function listDatasetPrefix(
+  repo: { type: "dataset"; name: string },
+  accessToken: string,
+  revision: string,
+  prefix: string,
+): Promise<DatasetSourceFile[]> {
+  const files: DatasetSourceFile[] = [];
+  try {
+    for await (const entry of listFiles({
+      repo,
+      accessToken,
+      recursive: true,
+      path: prefix,
+      revision,
+    })) {
+      if (entry.type !== "file" || !entry.path.endsWith(".jsonl")) continue;
+      const oid = entry.xetHash ?? entry.lfs?.oid ?? entry.oid;
+      if (oid === undefined || oid.length === 0) {
+        throw new Error(`dataset source has no immutable object id: ${entry.path}`);
+      }
+      files.push({ path: entry.path, oid, size: entry.size });
+    }
+  } catch (error) {
+    if (!isHubNotFound(error)) throw error;
+    await assertDatasetRepoReadable(repo, accessToken, revision);
+  }
+  return files;
+}
+
+export function createDurableIndexBucketClient(
+  indexBucket: string,
+  accessToken: string,
+): DurableIndexBucketClient {
+  const repo = { type: "bucket", name: indexBucket } as const;
+  return {
+    async download(path: string, destination: string): Promise<boolean> {
+      const blob = await downloadFile({ repo, accessToken, path });
+      if (blob === null) return false;
+      await mkdir(dirname(destination), { recursive: true });
+      await pipeline(Readable.fromWeb(blob.stream()), createWriteStream(destination));
+      return true;
+    },
+    async uploadFile(path: string, source: string): Promise<void> {
+      await uploadFile({
+        repo,
+        accessToken,
+        file: { path, content: await openAsBlob(source) },
+        commitTitle: `Publish ${path}`,
+      });
+    },
+    async list(prefix: string): Promise<readonly BucketFile[]> {
+      const files: BucketFile[] = [];
+      for await (const entry of listFiles({
+        repo,
+        accessToken,
+        recursive: true,
+        path: prefix,
+        expand: true,
+      })) {
+        if (entry.type === "file") {
+          files.push({
+            path: entry.path,
+            ...(entry.uploadedAt === undefined ? {} : { uploadedAt: entry.uploadedAt }),
+          });
+        }
+      }
+      return files;
+    },
+    async remove(paths: readonly string[]): Promise<void> {
+      if (paths.length === 0) return;
+      await deleteFiles({
+        repo,
+        accessToken,
+        paths: [...paths],
+        commitTitle: "Prune durable index generations",
+      });
+    },
+  };
+}
+
+function parseManifest(
+  raw: string,
+  options: Pick<DurableIndexOptions, "datasetRepo" | "contractHash">,
+): DurableIndexManifest {
+  let candidate: unknown;
+  try {
+    candidate = JSON.parse(raw);
+  } catch {
+    throw new Error("durable index manifest is not valid JSON");
+  }
+  const manifest = durableIndexManifestSchema.parse(candidate);
+  if (manifest.dataset.repo !== options.datasetRepo) {
+    throw new Error(`durable index dataset mismatch: ${manifest.dataset.repo}`);
+  }
+  if (manifest.projection.contract_hash !== options.contractHash) {
+    throw new Error("durable index enrichment contract does not match the running code");
+  }
+  return manifest;
+}
+
+function ensureIndexTables(db: Database.Database): void {
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS index_metadata (
+      singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+      schema_version INTEGER NOT NULL,
+      dataset_repo TEXT NOT NULL,
+      dataset_revision TEXT NOT NULL,
+      contract_hash TEXT NOT NULL
+    );
+    CREATE TABLE IF NOT EXISTS source_files (
+      path TEXT PRIMARY KEY,
+      kind TEXT NOT NULL,
+      oid TEXT NOT NULL,
+      byte_length INTEGER NOT NULL CHECK (byte_length >= 0),
+      content_sha256 TEXT NOT NULL,
+      row_count INTEGER NOT NULL CHECK (row_count >= 0)
+    );
+  `);
+}
+
+function sourceFileRows(db: Database.Database): Map<string, SourceFileRow> {
+  const rows = db
+    .prepare(
+      `SELECT path, kind, oid, byte_length, content_sha256, row_count
+       FROM source_files ORDER BY path`,
+    )
+    .all() as SourceFileRow[];
+  return new Map(rows.map((row) => [row.path, row]));
+}
+
+function upsertSourceFile(db: Database.Database, row: SourceFileRow): void {
+  db.prepare(
+    `INSERT INTO source_files
+       (path, kind, oid, byte_length, content_sha256, row_count)
+     VALUES (?, ?, ?, ?, ?, ?)
+     ON CONFLICT(path) DO UPDATE SET
+       kind = excluded.kind,
+       oid = excluded.oid,
+       byte_length = excluded.byte_length,
+       content_sha256 = excluded.content_sha256,
+       row_count = excluded.row_count`,
+  ).run(row.path, row.kind, row.oid, row.byte_length, row.content_sha256, row.row_count);
+}
+
+function writeMetadata(db: Database.Database, metadata: IndexMetadataRow): void {
+  db.prepare(
+    `INSERT INTO index_metadata
+       (singleton, schema_version, dataset_repo, dataset_revision, contract_hash)
+     VALUES (1, ?, ?, ?, ?)
+     ON CONFLICT(singleton) DO UPDATE SET
+       schema_version = excluded.schema_version,
+       dataset_repo = excluded.dataset_repo,
+       dataset_revision = excluded.dataset_revision,
+       contract_hash = excluded.contract_hash`,
+  ).run(
+    metadata.schema_version,
+    metadata.dataset_repo,
+    metadata.dataset_revision,
+    metadata.contract_hash,
+  );
+}
+
+function readMetadata(db: Database.Database): IndexMetadataRow | undefined {
+  return db
+    .prepare(
+      `SELECT schema_version, dataset_repo, dataset_revision, contract_hash
+       FROM index_metadata WHERE singleton = 1`,
+    )
+    .get() as IndexMetadataRow | undefined;
+}
+
+function sourceKindStats(
+  byKind: ReadonlyMap<DatasetSourceKind, { files: number; rows: number }>,
+  kind: DatasetSourceKind,
+): { files: number; rows: number } {
+  return byKind.get(kind) ?? { files: 0, rows: 0 };
+}
+
+function databaseCounts(db: Database.Database): DurableIndexCounts {
+  const count = (table: string): number =>
+    (db.prepare(`SELECT COUNT(*) AS n FROM ${table}`).get() as { n: number }).n;
+  const eventCount = (kind: DatasetSourceKind): number =>
+    (
+      db
+        .prepare("SELECT COALESCE(SUM(row_count), 0) AS n FROM source_files WHERE kind = ?")
+        .get(kind) as {
+        n: number;
+      }
+    ).n;
+  return {
+    tweets: count("tweets"),
+    units: count("enrich_queue"),
+    enrichments: count("enrichment"),
+    attempt_events: eventCount("attempt"),
+    registry_events: eventCount("registry"),
+  };
+}
+
+function validateDatabase(
+  db: Database.Database,
+  manifest: DurableIndexManifest,
+  options: Pick<DurableIndexOptions, "datasetRepo" | "contractHash">,
+): void {
+  assertDatabaseIntegrity(db);
+  const metadata = readMetadata(db);
+  if (metadata === undefined) throw new Error("durable index metadata is missing");
+  if (metadata.schema_version !== INDEX_SCHEMA_VERSION)
+    throw new Error("durable index schema version mismatch");
+  if (
+    metadata.dataset_repo !== options.datasetRepo ||
+    metadata.dataset_revision !== manifest.dataset.revision
+  ) {
+    throw new Error("durable index dataset provenance mismatch");
+  }
+  if (metadata.contract_hash !== options.contractHash)
+    throw new Error("durable index contract mismatch");
+  if (JSON.stringify(databaseCounts(db)) !== JSON.stringify(manifest.counts)) {
+    throw new Error("durable index physical counts do not match the manifest");
+  }
+}
+
+function validateStandaloneDatabase(
+  path: string,
+  metadata: IndexMetadataRow,
+  counts: DurableIndexCounts,
+): void {
+  const db = new Database(path, { readonly: true, fileMustExist: true });
+  try {
+    db.pragma("query_only = ON");
+    assertDatabaseIntegrity(db);
+    if (JSON.stringify(readMetadata(db)) !== JSON.stringify(metadata)) {
+      throw new Error("published durable index metadata does not match");
+    }
+    if (JSON.stringify(databaseCounts(db)) !== JSON.stringify(counts)) {
+      throw new Error("published durable index counts do not match");
+    }
+  } finally {
+    db.close();
+  }
+}
+
+function assertDatabaseIntegrity(db: Database.Database): void {
+  const rows = db.pragma("quick_check") as { quick_check: string }[];
+  if (rows.length !== 1 || rows[0]?.quick_check !== "ok") {
+    throw new Error("durable index SQLite quick_check failed");
+  }
+}
+
+function sourceOrder(left: string, right: string): number {
+  const priorities: Record<DatasetSourceKind, number> = {
+    tweet: 0,
+    enrichment: 1,
+    registry: 2,
+    attempt: 3,
+    receipt: 4,
+  };
+  const difference = priorities[datasetSourceKind(left)] - priorities[datasetSourceKind(right)];
+  return difference === 0 ? left.localeCompare(right) : difference;
+}
+
+async function fileSha256(path: string): Promise<string> {
+  const hash = createHash("sha256");
+  for await (const chunk of createReadStream(path)) {
+    const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array);
+    hash.update(bytes);
+  }
+  return hash.digest("hex");
+}
+
+async function assertFileSha256(path: string, expected: string): Promise<void> {
+  if ((await fileSha256(path)) !== expected) {
+    throw new Error("durable index database checksum mismatch");
+  }
+}
+
+function sha256Bytes(content: Uint8Array): string {
+  return createHash("sha256").update(content).digest("hex");
+}

space/src/enrich-command.ts

@@ -1,9 +1,10 @@
+import { rm } from "node:fs/promises";
 import { join } from "node:path";
 
 import { loadConfig } from "./config.js";
 import { createHubClient, DatasetMirror } from "./dataset.js";
 import { loadEnrichTaxonomy } from "./enrich-config.js";
-import { EnrichStore } from "./enrich-store.js";
+import { DurableIndex } from "./durable-index.js";
 import {
   contractHashFor,
   createExactHubVerifier,
@@ -14,7 +15,6 @@ import {
   runEnrichTick,
 } from "./enrich-worker.js";
 import type { WorkerCeilings } from "./enrich-worker.js";
-import { TweetStore } from "./store.js";
 
 /**
  * Standalone worker: rebuilds the local mirror from the dataset (system of
@@ -35,10 +35,8 @@ export async function runEnrichCommand(env: Record<string, string | undefined>):
     process.exitCode = 2;
     return;
   }
-  const store = new TweetStore();
   const hub = createHubClient(config.datasetRepo, config.hfToken);
   const mirror = new DatasetMirror(hub, join(config.dataDir, "mirror"));
-  const enrichStore = new EnrichStore(store.database, config.taxonomyVersion, () => new Date());
   const taxonomy = await loadEnrichTaxonomy(mirror, config.taxonomyVersion);
   if (taxonomy.error !== undefined) {
     console.error(`[xtap-pool worker] taxonomy unavailable: ${taxonomy.error}`);
@@ -46,19 +44,30 @@ export async function runEnrichCommand(env: Record<string, string | undefined>):
     return;
   }
   const contractHash = contractHashFor({ taxonomy, model: config.llmModel });
-  enrichStore.setContractHash(contractHash);
 
   console.log(
-    `[xtap-pool worker] rebuilding index from ${config.datasetRepo}, contract ${contractHash.slice(0, 12)} ...`,
+    `[xtap-pool worker] restoring durable index from ${config.indexBucket}, contract ${contractHash.slice(0, 12)} ...`,
   );
-  mirror.clearForRebuild();
-  const tweetStats = await mirror.rebuild(store, enrichStore);
-  const enrichStats = await mirror.rebuildEnrichment(enrichStore);
-  enrichStore.releaseClaims();
+  const indexOptions = {
+    datasetRepo: config.datasetRepo,
+    indexBucket: config.indexBucket,
+    accessToken: config.hfToken,
+    databasePath: join(config.dataDir, "index", "worker.sqlite"),
+    mirror,
+    taxonomyVersion: config.taxonomyVersion,
+    contractHash,
+  };
+  const index = await DurableIndex.restore(indexOptions);
+  const advanced = await index.advanceToLatest();
+  index.enrichStore.releaseClaims();
   console.log(
-    `[xtap-pool worker] indexed ${String(tweetStats.tweets)} tweets, ` +
-      `${String(enrichStats.rows)} enrichment rows, ${String(enrichStats.attempts)} attempt events`,
+    `[xtap-pool worker] index revision ${advanced.revision.slice(0, 12)}; ` +
+      `changed_files=${String(advanced.filesChanged)} rows=${String(advanced.rowsApplied)} ` +
+      `tweets=${String(advanced.counts.tweets)} units=${String(advanced.counts.units)}`,
   );
+  const publicationBase = join(config.dataDir, "index", "publication-base.sqlite");
+  await index.createWorkingCopy(publicationBase);
+  const predecessorKeys = index.retainedDatabaseKeys();
 
   const llm = createRouterLlmClient({
     hfToken: config.inferenceToken,
@@ -82,7 +91,7 @@ export async function runEnrichCommand(env: Record<string, string | undefined>):
     discardedAssignmentRateMinUnits: config.enrichDiscardedAssignmentRateMinUnits,
   };
   const receipt = await runEnrichTick({
-    enrichStore,
+    enrichStore: index.enrichStore,
     mirror,
     taxonomy,
     llm,
@@ -95,6 +104,25 @@ export async function runEnrichCommand(env: Record<string, string | undefined>):
     now: (): Date => new Date(),
     ceilings,
   });
+  index.close();
+  const publicationMirror = new DatasetMirror(hub, join(config.dataDir, "publication-mirror"));
+  const publication = DurableIndex.openLocal({
+    ...indexOptions,
+    databasePath: publicationBase,
+    mirror: publicationMirror,
+    predecessorKeys,
+  });
+  try {
+    const { advance: finalAdvance, manifest } = await publication.publishLatest();
+    console.log(
+      `[xtap-pool worker] published index ${manifest.database.sha256.slice(0, 12)} at ` +
+        `${finalAdvance.revision.slice(0, 12)}; changed_files=${String(finalAdvance.filesChanged)} ` +
+        `rows=${String(finalAdvance.rowsApplied)}`,
+    );
+  } finally {
+    publication.close();
+    await rm(publicationBase, { force: true });
+  }
   console.log(
     `[xtap-pool worker] finished: units=${String(receipt.units)} retries=${String(receipt.retries)} ` +
       `blocked=${String(receipt.blocked)} calls=${String(receipt.calls)} ` +
@@ -103,7 +131,6 @@ export async function runEnrichCommand(env: Record<string, string | undefined>):
       `tokens=${String(receipt.prompt_tokens + receipt.completion_tokens)} ` +
       `stopped_by=${receipt.stopped_by ?? "batch-complete"}`,
   );
-  store.close();
 }
 
 /**

space/src/index-command-main.ts

@@ -0,0 +1,9 @@
+import { runIndexCommand } from "./index-command.js";
+
+try {
+  await runIndexCommand(process.env);
+} catch (error) {
+  const message = error instanceof Error ? error.message : "unknown error";
+  console.error(`[xtap-pool index] fatal: ${message}`);
+  process.exitCode = 1;
+}

space/src/index-command.ts

@@ -0,0 +1,50 @@
+import { join } from "node:path";
+
+import { DEFAULT_ENRICHMENT_MODEL } from "@xtap-pool/shared";
+import { z } from "zod";
+
+import { createHubClient, DatasetMirror } from "./dataset.js";
+import { DurableIndex } from "./durable-index.js";
+import { loadEnrichTaxonomy } from "./enrich-config.js";
+import { contractHashFor } from "./enrich-worker.js";
+
+const indexCommandConfigSchema = z.object({
+  DATA_DIR: z.string().default(".data"),
+  DATASET_REPO: z.string().min(1),
+  INDEX_BUCKET: z.string().min(1),
+  HF_TOKEN: z.string().min(1),
+  LLM_MODEL: z.string().min(1).default(DEFAULT_ENRICHMENT_MODEL),
+  TAXONOMY_VERSION: z.coerce.number().int().min(1).default(1),
+});
+
+/** Explicit full replay used to seed or repair the durable enrichment index. */
+export async function runIndexCommand(env: Record<string, string | undefined>): Promise<void> {
+  const config = indexCommandConfigSchema.parse(env);
+  const hub = createHubClient(config.DATASET_REPO, config.HF_TOKEN);
+  const mirror = new DatasetMirror(hub, join(config.DATA_DIR, "index-bootstrap-mirror"));
+  const taxonomy = await loadEnrichTaxonomy(mirror, config.TAXONOMY_VERSION);
+  if (taxonomy.error !== undefined) {
+    throw new Error(`enrichment taxonomy unavailable: ${taxonomy.error}`);
+  }
+  const contractHash = contractHashFor({ taxonomy, model: config.LLM_MODEL });
+  const index = await DurableIndex.bootstrap({
+    datasetRepo: config.DATASET_REPO,
+    indexBucket: config.INDEX_BUCKET,
+    accessToken: config.HF_TOKEN,
+    databasePath: join(config.DATA_DIR, "index", "bootstrap.sqlite"),
+    mirror,
+    taxonomyVersion: config.TAXONOMY_VERSION,
+    contractHash,
+  });
+  try {
+    const { manifest } = await index.publishLatest();
+    const stats = index.stats();
+    console.log(
+      `[xtap-pool index] published ${manifest.database.sha256} at ` +
+        `${manifest.dataset.revision}; tweets=${String(stats.tweetRows)} ` +
+        `enrichments=${String(stats.enrichmentRows)} attempts=${String(stats.attemptEvents)}`,
+    );
+  } finally {
+    index.close();
+  }
+}

space/src/server.ts

@@ -7,25 +7,20 @@ import { createApp } from "./app.js";
 import type { AppReadiness } from "./app.js";
 import { loadConfig } from "./config.js";
 import { createHubClient, DatasetMirror } from "./dataset.js";
-import { datasetStateFromRebuildError, errorStatus } from "./dataset-state.js";
 import type { DatasetState } from "./dataset-state.js";
 import { checkDatasetCredential, datasetCredentialOk } from "./dataset-token.js";
 import type { DatasetCredentialReadiness } from "./dataset-token.js";
 import { loadEnrichTaxonomy } from "./enrich-config.js";
-import { EnrichStore } from "./enrich-store.js";
+import { DurableIndex } from "./durable-index.js";
 import { contractHashFor } from "./enrich-worker.js";
 import { ingestBatch, Mutex } from "./ingest.js";
 import { checkInferenceCredential, inferenceCredentialOk } from "./inference-token.js";
 import type { InferenceCredentialReadiness } from "./inference-token.js";
 import { PoolMembership } from "./membership.js";
 import { ServiceAccountRegistry } from "./service-accounts.js";
-import { TweetStore } from "./store.js";
 import { UnitStore } from "./unit-store.js";
 
 const config = loadConfig(process.env);
-const store = new TweetStore();
-const enrichStore = new EnrichStore(store.database, config.taxonomyVersion);
-const unitStore = new UnitStore(store.database, config.taxonomyVersion);
 const hub = createHubClient(config.datasetRepo, config.hfToken);
 const mirror = new DatasetMirror(hub, join(config.dataDir, "mirror"));
 const mutex = new Mutex();
@@ -37,7 +32,7 @@ let rebuilt: RebuildStats = { files: 0, tweets: 0 };
 let enrichment: EnrichmentStats = { files: 0, rows: 0 };
 let datasetState: DatasetState = {
   state: "unknown",
-  error: "The dataset index has not been built yet.",
+  error: "The durable dataset index has not been restored yet.",
 };
 let datasetCredential: DatasetCredentialReadiness = {
   credential: "unknown",
@@ -49,14 +44,40 @@ let inferenceCredential: InferenceCredentialReadiness = config.enrichEnabled
 let readiness: AppReadiness;
 let credentialRetryTimer: ReturnType<typeof setTimeout> | undefined;
 let enrichmentRefreshTimer: ReturnType<typeof setTimeout> | undefined;
+let activeFetch: (request: Request) => Response | Promise<Response> = () =>
+  Response.json(
+    {
+      ok: false,
+      dataset: {
+        credential: datasetCredential.credential,
+        state: datasetCredential.credential === "invalid" ? "invalid" : "unknown",
+        indexed_files: 0,
+        indexed_tweets: 0,
+        enrichment_rows: 0,
+        ...(datasetCredential.credential === "ok"
+          ? {}
+          : { credential_error: datasetCredential.error }),
+        error: datasetState.state === "ready" ? undefined : datasetState.error,
+      },
+    },
+    { status: 503 },
+  );
+serve({ fetch: (request: Request) => activeFetch(request), port: config.port }, (info) => {
+  console.log(`[xtap-pool] listening on :${String(info.port)}`);
+});
 
 [datasetCredential, inferenceCredential] = await Promise.all([
-  checkDatasetCredential({ token: config.hfToken, datasetRepo: config.datasetRepo }),
+  checkDatasetCredential({
+    token: config.hfToken,
+    datasetRepo: config.datasetRepo,
+    indexBucket: config.indexBucket,
+  }),
   checkInferenceCredential({
     enabled: config.enrichEnabled,
     token: config.inferenceToken,
   }),
 ]);
+await waitForStorageCredential();
 
 const [membership, serviceAccounts] = await Promise.all([
   PoolMembership.load({
@@ -68,8 +89,23 @@ const [membership, serviceAccounts] = await Promise.all([
   ServiceAccountRegistry.load({ mirror, now: () => new Date() }),
 ]);
 let taxonomy = await loadEnrichTaxonomy(mirror, config.taxonomyVersion);
-enrichStore.setContractHash(contractHashFor({ taxonomy, model: config.llmModel }));
-readiness = buildReadiness();
+if (taxonomy.error !== undefined) {
+  throw new Error(`enrichment taxonomy unavailable: ${taxonomy.error}`);
+}
+const contractHash = contractHashFor({ taxonomy, model: config.llmModel });
+const index = await DurableIndex.restore({
+  datasetRepo: config.datasetRepo,
+  indexBucket: config.indexBucket,
+  accessToken: config.hfToken,
+  databasePath: join(config.dataDir, "index", "space.sqlite"),
+  mirror,
+  taxonomyVersion: config.taxonomyVersion,
+  contractHash,
+});
+const initialAdvance = await index.advanceToLatest();
+const store = index.store;
+const enrichStore = index.enrichStore;
+const unitStore = new UnitStore(store.database, config.taxonomyVersion);
 let lastReceipt: import("@xtap-pool/shared").EnrichReceipt | undefined;
 function recordLastReceipt(receipt: import("@xtap-pool/shared").EnrichReceipt | undefined): void {
   if (receipt?.contract_hash !== enrichStore.currentContractHash()) return;
@@ -77,6 +113,10 @@ function recordLastReceipt(receipt: import("@xtap-pool/shared").EnrichReceipt |
     lastReceipt = receipt;
   }
 }
+applyIndexStats();
+datasetState = { state: "ready" };
+enrichStore.releaseClaims();
+readiness = buildReadiness();
 const app = createApp({
   config,
   store,
@@ -91,9 +131,25 @@ const app = createApp({
     lastReceipt: () => lastReceipt,
   },
   ingest: (username, payload) =>
-    mutex.run(() =>
-      ingestBatch({ store, mirror, enrich: enrichStore, now: () => new Date() }, username, payload),
-    ),
+    mutex.run(async () => {
+      const result = await ingestBatch(
+        { store, mirror, enrich: enrichStore, now: () => new Date() },
+        username,
+        payload,
+      );
+      try {
+        await index.advanceToLatest();
+        applyIndexStats();
+        datasetState = { state: "ready" };
+        readiness = buildReadiness();
+      } catch (error) {
+        const message = errorMessage(error);
+        datasetState = { state: "invalid", error: `durable index ingest sync failed: ${message}` };
+        readiness = buildReadiness();
+        throw error;
+      }
+      return result;
+    }),
   repairMembership: (actor) =>
     mutex.run(async () => {
       const pool = await membership.repairConfig(actor);
@@ -113,8 +169,11 @@ const app = createApp({
 app.use("*", serveStatic({ root: config.staticRoot }));
 app.use("*", serveStatic({ root: config.staticRoot, path: "index.html" }));
 
-console.log(`[xtap-pool] rebuilding index from ${config.datasetRepo} ...`);
-await rebuildDatasetIndexIfReady();
+console.log(
+  `[xtap-pool] restored durable index ${initialAdvance.revision.slice(0, 12)} from ` +
+    `${config.indexBucket}; changed_files=${String(initialAdvance.filesChanged)} ` +
+    `rows=${String(initialAdvance.rowsApplied)}`,
+);
 readiness = buildReadiness();
 const pool = membership.snapshot();
 console.log(
@@ -135,37 +194,15 @@ if (config.enrichEnabled) {
       "to drain the queue.",
   );
 }
+activeFetch = (request) => app.fetch(request);
 startCredentialRetryIfNeeded();
 startEnrichmentRefresh();
 
-serve({ fetch: app.fetch, port: config.port }, (info) => {
-  console.log(`[xtap-pool] listening on :${String(info.port)}`);
-});
-
-async function rebuildDatasetIndexIfReady(): Promise<void> {
-  if (!datasetCredentialOk(datasetCredential)) return;
-  datasetState = { state: "unknown", error: "Rebuilding the dataset index." };
-  readiness = buildReadiness();
-  mirror.clearForRebuild();
-  enrichStore.clearForRebuild();
-  store.clearForRebuild();
-  rebuilt = { files: 0, tweets: 0 };
-  enrichment = { files: 0, rows: 0 };
-  try {
-    rebuilt = await mirror.rebuild(store, enrichStore);
-    enrichment = await mirror.rebuildEnrichment(enrichStore);
-    recordLastReceipt(mirror.latestReceipt());
-    enrichStore.releaseClaims();
-    datasetState = { state: "ready" };
-  } catch (error) {
-    const status = errorStatus(error);
-    if (status === 401 || status === 403) {
-      datasetCredential = { credential: "invalid", error: errorMessage(error) };
-      datasetState = { state: "unknown", error: "Dataset indexing requires a valid HF_TOKEN." };
-    } else {
-      datasetState = datasetStateFromRebuildError(error);
-    }
-  }
+function applyIndexStats(): void {
+  const stats = index.stats();
+  rebuilt = { files: stats.tweetFiles, tweets: stats.tweetRows };
+  enrichment = { files: stats.enrichmentFiles, rows: stats.enrichmentRows };
+  recordLastReceipt(mirror.latestReceipt());
 }
 
 /**
@@ -179,30 +216,31 @@ function startEnrichmentRefresh(): void {
     enrichmentRefreshTimer = undefined;
     void refreshExternalEnrichment()
       .catch((error: unknown) => {
-        console.error(`[xtap-pool] enrichment refresh failed: ${errorMessage(error)}`);
+        const message = errorMessage(error);
+        datasetState = { state: "invalid", error: `durable index refresh failed: ${message}` };
+        readiness = buildReadiness();
+        console.error(`[xtap-pool] enrichment refresh failed: ${message}`);
       })
       .finally(startEnrichmentRefresh);
   }, enrichmentRefreshMs());
 }
 
 async function refreshExternalEnrichment(): Promise<void> {
-  if (!datasetCredentialOk(datasetCredential) || datasetState.state !== "ready") return;
+  if (!datasetCredentialOk(datasetCredential)) return;
   await mutex.run(async () => {
-    // A rebuild may have changed readiness while this refresh was waiting.
-    if (!datasetCredentialOk(datasetCredential) || datasetState.state !== "ready") return;
+    if (!datasetCredentialOk(datasetCredential)) return;
     const nextTaxonomy = await loadEnrichTaxonomy(mirror, config.taxonomyVersion);
     if (nextTaxonomy.error !== undefined) {
       throw new Error(`enrichment taxonomy refresh failed: ${nextTaxonomy.error}`);
     }
-    const refreshed = await mirror.refreshEnrichment(enrichStore, () => {
-      taxonomy = nextTaxonomy;
-      enrichStore.setContractHash(contractHashFor({ taxonomy, model: config.llmModel }));
-      if (lastReceipt?.contract_hash !== enrichStore.currentContractHash()) {
-        lastReceipt = undefined;
-      }
-    });
-    recordLastReceipt(refreshed.receipt);
-    enrichment = { ...enrichment, rows: enrichStore.enrichmentRowCount() };
+    const nextContractHash = contractHashFor({ taxonomy: nextTaxonomy, model: config.llmModel });
+    if (nextContractHash !== enrichStore.currentContractHash()) {
+      throw new Error("enrichment contract changed; publish a replacement durable index");
+    }
+    taxonomy = nextTaxonomy;
+    await index.advanceToLatest();
+    applyIndexStats();
+    datasetState = { state: "ready" };
     readiness = buildReadiness();
   });
 }
@@ -296,12 +334,17 @@ function taxonomyReady(): boolean {
   return taxonomy.error === undefined;
 }
 
+// eslint-disable-next-line complexity -- Recovery checks each independently persisted dataset-backed configuration.
 async function reloadDatasetBackedConfig(force: boolean): Promise<void> {
   if (force || membership.hasRetryableConfigError()) await membership.reload();
   if (force || serviceAccounts.hasRetryableConfigError()) await serviceAccounts.reload();
   if (config.enrichEnabled && (force || !taxonomyReady())) {
-    taxonomy = await loadEnrichTaxonomy(mirror, config.taxonomyVersion);
-    enrichStore.setContractHash(contractHashFor({ taxonomy, model: config.llmModel }));
+    const nextTaxonomy = await loadEnrichTaxonomy(mirror, config.taxonomyVersion);
+    const nextContractHash = contractHashFor({ taxonomy: nextTaxonomy, model: config.llmModel });
+    if (nextContractHash !== enrichStore.currentContractHash()) {
+      throw new Error("enrichment contract changed; publish a replacement durable index");
+    }
+    taxonomy = nextTaxonomy;
   }
 }
 
@@ -341,6 +384,7 @@ async function retryUncertainCredentials(): Promise<void> {
     datasetCredential = await checkDatasetCredential({
       token: config.hfToken,
       datasetRepo: config.datasetRepo,
+      indexBucket: config.indexBucket,
     });
   }
   const datasetRecovered = !datasetWasReady && datasetCredentialOk(datasetCredential);
@@ -348,7 +392,9 @@ async function retryUncertainCredentials(): Promise<void> {
     await mutex.run(async () => {
       await reloadDatasetBackedConfig(datasetRecovered);
       if (datasetRecovered || datasetState.state === "unknown") {
-        await rebuildDatasetIndexIfReady();
+        await index.advanceToLatest();
+        applyIndexStats();
+        datasetState = { state: "ready" };
       }
     });
   }
@@ -361,6 +407,22 @@ async function retryUncertainCredentials(): Promise<void> {
   readiness = buildReadiness();
 }
 
+async function waitForStorageCredential(): Promise<void> {
+  while (!datasetCredentialOk(datasetCredential)) {
+    datasetState = {
+      state: datasetCredential.credential === "invalid" ? "invalid" : "unknown",
+      error: "error" in datasetCredential ? datasetCredential.error : "HF_TOKEN is unavailable.",
+    };
+    console.error(`[xtap-pool] storage credential unavailable: ${datasetState.error}`);
+    await new Promise((resolve) => setTimeout(resolve, credentialRetryMs()));
+    datasetCredential = await checkDatasetCredential({
+      token: config.hfToken,
+      datasetRepo: config.datasetRepo,
+      indexBucket: config.indexBucket,
+    });
+  }
+}
+
 function credentialRetryMs(): number {
   return Math.min(Math.max(config.enrichIntervalMs, 5000), 60000);
 }

space/tests/config.test.ts

@@ -4,6 +4,7 @@ import { loadConfig } from "../src/config.js";
 
 const baseEnv = {
   DATASET_REPO: "osolmaz/xtap-pool-data",
+  INDEX_BUCKET: "osolmaz/xtap-pool-bucket",
   HF_TOKEN: "hf_x",
   POOL_SIGNING_SECRET: "pool-secret-0123456789abcdef0123456789abcdef",
   SESSION_SECRET: "session-secret-0123456789abcdef0123456789ab",
@@ -18,6 +19,7 @@ describe("loadConfig", () => {
   it("parses a full environment with defaults", () => {
     const config = loadConfig(baseEnv);
     expect(config.port).toBe(7860);
+    expect(config.indexBucket).toBe("osolmaz/xtap-pool-bucket");
     expect(config.allowedUsers).toEqual(["osolmaz", "alice", "bob"]);
     expect(config.poolAdmins).toEqual(["osolmaz"]);
     expect(config.publicUrl).toBe("https://dutifuldev-xtap-pool.hf.space");
@@ -52,7 +54,7 @@ describe("loadConfig", () => {
     expect(config.enrichEnabled).toBe(false);
     expect(config.enrichIntervalMs).toBe(60000);
     expect(config.enrichMaxConcurrentCalls).toBe(1);
-    expect(config.llmModel).toBe("zai-org/GLM-5.2");
+    expect(config.llmModel).toBe("zai-org/GLM-5.2:fireworks-ai");
     expect(config.taxonomyVersion).toBe(1);
   });
 

space/tests/dataset-token.test.ts

@@ -2,103 +2,92 @@ import { describe, expect, it, vi } from "vitest";
 
 import { checkDatasetCredential } from "../src/dataset-token.js";
 
-describe("dataset credential readiness", () => {
-  it("accepts fine-grained tokens with dataset read/write permissions only", async () => {
+const datasetRepo = "alice/xtap-pool-data";
+const indexBucket = "alice/xtap-pool-bucket";
+
+function scope(name: string, permissions: readonly string[], type: string): unknown {
+  return { entity: { type, name }, permissions };
+}
+
+function whoami(scopes: readonly unknown[]): unknown {
+  return {
+    auth: {
+      accessToken: {
+        role: "fineGrained",
+        fineGrained: { global: [], scoped: scopes },
+      },
+    },
+  };
+}
+
+function validScopes(): readonly unknown[] {
+  const permissions = ["repo.content.read", "repo.content.write"];
+  return [scope(datasetRepo, permissions, "dataset"), scope(indexBucket, permissions, "bucket")];
+}
+
+function check(fetchFn: typeof fetch, token = "hf_storage") {
+  return checkDatasetCredential({ token, datasetRepo, indexBucket, fetchFn });
+}
+
+describe("storage credential readiness", () => {
+  it("accepts exact dataset and Bucket read/write scopes plus direct reads", async () => {
+    const fetchFn = vi
+      .fn<typeof fetch>()
+      .mockResolvedValueOnce(Response.json(whoami(validScopes())))
+      .mockResolvedValueOnce(new Response("{}", { status: 200 }))
+      .mockResolvedValueOnce(new Response("{}", { status: 200 }));
+
+    await expect(check(fetchFn)).resolves.toEqual({ credential: "ok" });
+  });
+
+  it("rejects a token whose Bucket metadata scope is missing", async () => {
     const fetchFn: typeof fetch = () =>
       Promise.resolve(
-        Response.json({
-          auth: {
-            accessToken: {
-              role: "fineGrained",
-              fineGrained: {
-                global: [],
-                scoped: [
-                  {
-                    entity: { type: "dataset", name: "alice/xtap-pool-data" },
-                    permissions: ["repo.content.read", "repo.content.write"],
-                  },
-                ],
-              },
-            },
-          },
-        }),
+        Response.json(
+          whoami([scope(datasetRepo, ["repo.content.read", "repo.content.write"], "dataset")]),
+        ),
       );
 
-    await expect(
-      checkDatasetCredential({ token: "hf_dataset", datasetRepo: "alice/xtap-pool-data", fetchFn }),
-    ).resolves.toEqual({ credential: "ok" });
+    await expect(check(fetchFn)).resolves.toEqual({
+      credential: "invalid",
+      error: `HF_TOKEN must include repo.content.read on ${indexBucket}. HF_TOKEN must include repo.content.write or repo.write on ${indexBucket}.`,
+    });
   });
 
-  it("rejects tokens whose metadata passes but private-dataset downloads fail", async () => {
+  it("rejects tokens whose metadata passes but a direct storage read fails", async () => {
     const fetchFn = vi
       .fn<typeof fetch>()
-      .mockResolvedValueOnce(
-        Response.json({
-          auth: {
-            accessToken: {
-              role: "fineGrained",
-              fineGrained: {
-                global: [],
-                scoped: [
-                  {
-                    entity: { type: "dataset", name: "alice/xtap-pool-data" },
-                    permissions: ["repo.content.read", "repo.content.write"],
-                  },
-                ],
-              },
-            },
-          },
-        }),
-      )
+      .mockResolvedValueOnce(Response.json(whoami(validScopes())))
+      .mockResolvedValueOnce(new Response("{}", { status: 200 }))
       .mockResolvedValueOnce(new Response("unauthorized", { status: 401 }));
 
-    await expect(
-      checkDatasetCredential({ token: "hf_dataset", datasetRepo: "alice/xtap-pool-data", fetchFn }),
-    ).resolves.toEqual({
+    await expect(check(fetchFn)).resolves.toEqual({
       credential: "invalid",
-      error: "Hugging Face rejected a direct private-dataset download using HF_TOKEN (401).",
+      error: "Hugging Face rejected a direct private-Bucket read using HF_TOKEN (401).",
     });
   });
 
-  it("rejects dataset tokens without write permission", async () => {
+  it("rejects storage tokens without write permission", async () => {
     const fetchFn: typeof fetch = () =>
       Promise.resolve(
-        Response.json({
-          auth: {
-            accessToken: {
-              role: "fineGrained",
-              fineGrained: {
-                global: [],
-                scoped: [
-                  {
-                    entity: { type: "dataset", name: "alice/xtap-pool-data" },
-                    permissions: ["repo.content.read"],
-                  },
-                ],
-              },
-            },
-          },
-        }),
+        Response.json(
+          whoami([
+            scope(datasetRepo, ["repo.content.read"], "dataset"),
+            scope(indexBucket, ["repo.content.read", "repo.content.write"], "bucket"),
+          ]),
+        ),
       );
 
-    await expect(
-      checkDatasetCredential({
-        token: "hf_readonly",
-        datasetRepo: "alice/xtap-pool-data",
-        fetchFn,
-      }),
-    ).resolves.toEqual({
+    await expect(check(fetchFn, "hf_readonly")).resolves.toEqual({
       credential: "invalid",
-      error: "HF_TOKEN must include repo.content.write or repo.write on alice/xtap-pool-data.",
+      error: `HF_TOKEN must include repo.content.write or repo.write on ${datasetRepo}.`,
     });
   });
 
   it("treats transient Hub failures as unknown so startup can retry", async () => {
     const fetchFn: typeof fetch = () => Promise.resolve(new Response("oops", { status: 503 }));
 
-    await expect(
-      checkDatasetCredential({ token: "hf_dataset", datasetRepo: "alice/xtap-pool-data", fetchFn }),
-    ).resolves.toEqual({
+    await expect(check(fetchFn)).resolves.toEqual({
       credential: "unknown",
       error: "Hugging Face rejected HF_TOKEN (503).",
     });

space/tests/dataset.test.ts

@@ -4,7 +4,11 @@ import { join } from "node:path";
 
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
 
-import { DatasetMirror, parseJsonlTweets } from "../src/dataset.js";
+import {
+  assertValidDatasetSourceContent,
+  DatasetMirror,
+  parseJsonlTweets,
+} from "../src/dataset.js";
 import type { HubClient } from "../src/dataset.js";
 import { EnrichStore } from "../src/enrich-store.js";
 import { TweetStore } from "../src/store.js";
@@ -115,6 +119,57 @@ describe("DatasetMirror.rebuild", () => {
   });
 });
 
+describe("durable source validation", () => {
+  it("rejects malformed JSON and schema-invalid records for every source kind", () => {
+    expect(() => {
+      assertValidDatasetSourceContent("data/osolmaz/2026/08/tweets-2026-08-04.jsonl", "not json\n");
+    }).toThrow("invalid JSON");
+    for (const path of [
+      "data/osolmaz/2026/08/tweets-2026-08-04.jsonl",
+      "enrichment/2026/08/enrichment-2026-08-04.jsonl",
+      "enrichment/attempts/2026/08/attempts-2026-08-04.jsonl",
+      "enrichment/registry/2026/08/registry-2026-08-04.jsonl",
+      "enrichment/receipts/2026-08-04.jsonl",
+    ]) {
+      expect(() => {
+        assertValidDatasetSourceContent(path, "{}\n");
+      }).toThrow("invalid");
+    }
+  });
+
+  it("accepts recognized previous enrichment and receipt contracts", () => {
+    expect(() => {
+      assertValidDatasetSourceContent(
+        "enrichment/2026/07/enrichment-2026-07-06.jsonl",
+        `${JSON.stringify({
+          unit_id: "1:someone",
+          tweet_ids: ["1"],
+          labels: ["ai"],
+          free_labels: ["gguf"],
+          concepts: [{ name: "vLLM", aliases: [] }],
+          model: "model",
+          taxonomy_version: 1,
+          enriched_at: "2026-07-06T00:00:00.000Z",
+        })}\n`,
+      );
+    }).not.toThrow();
+    expect(() => {
+      assertValidDatasetSourceContent(
+        "enrichment/receipts/2026-07-26.jsonl",
+        `${JSON.stringify({
+          started_at: "2026-07-26T00:00:00.000Z",
+          finished_at: "2026-07-26T00:01:00.000Z",
+          units: 1,
+          calls: 1,
+          prompt_tokens: 10,
+          completion_tokens: 5,
+          failures: 0,
+        })}\n`,
+      );
+    }).not.toThrow();
+  });
+});
+
 describe("DatasetMirror enrichment rebuild", () => {
   it("replays enrichment shards without seeding label output from legacy rows", async () => {
     hub.files.set(

space/tests/durable-index.test.ts

@@ -0,0 +1,416 @@
+import { createHash } from "node:crypto";
+import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { dirname, join } from "node:path";
+
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+
+import { DatasetMirror } from "../src/dataset.js";
+import { DurableIndex, durableIndexManifestSchema } from "../src/durable-index.js";
+import type {
+  BucketFile,
+  DatasetSnapshotClient,
+  DatasetSourceFile,
+  DurableIndexBucketClient,
+  DurableIndexOptions,
+} from "../src/durable-index.js";
+import { FakeHub, makePooled } from "./helpers.js";
+
+const DATASET = "osolmaz/xtap-pool-data";
+const BUCKET = "osolmaz/xtap-pool-bucket";
+const CONTRACT = "a".repeat(64);
+
+class FakeSource implements DatasetSnapshotClient {
+  revision = "1".repeat(40);
+  files = new Map<string, string>();
+  textFiles = new Map<string, string>();
+  commitErrors: unknown[] = [];
+
+  currentRevision(): Promise<string> {
+    return Promise.resolve(this.revision);
+  }
+
+  listJsonlFiles(): Promise<readonly DatasetSourceFile[]> {
+    return Promise.resolve(
+      [...this.files.entries()].map(([path, content]) => ({
+        path,
+        oid: sha256(content),
+        size: Buffer.byteLength(content),
+      })),
+    );
+  }
+
+  downloadFile(path: string): Promise<Uint8Array> {
+    const content = this.files.get(path);
+    if (content === undefined) return Promise.reject(new Error(`missing ${path}`));
+    return Promise.resolve(Buffer.from(content));
+  }
+
+  readText(path: string): Promise<string | undefined> {
+    return Promise.resolve(this.textFiles.get(path));
+  }
+
+  commitText(path: string, content: string, parentRevision: string): Promise<string> {
+    const commitError = this.commitErrors.shift();
+    if (commitError !== undefined) {
+      this.advanceRevision();
+      // Deliberately permit a non-Error rejection to cover defensive classifier behavior.
+      // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors
+      return Promise.reject(commitError);
+    }
+    if (parentRevision !== this.revision) {
+      return Promise.reject(new Error("parent commit does not match dataset HEAD"));
+    }
+    this.textFiles.set(path, content);
+    this.advanceRevision();
+    return Promise.resolve(this.revision);
+  }
+
+  advanceRevision(): void {
+    const value = Number.parseInt(this.revision.slice(0, 8), 16) + 1;
+    this.revision = value.toString(16).padStart(40, "0");
+  }
+}
+
+class FakeBucket implements DurableIndexBucketClient {
+  files = new Map<string, Buffer>();
+  uploaded = new Map<string, string>();
+  removed: string[] = [];
+  private clock = 0;
+
+  async download(path: string, destination: string): Promise<boolean> {
+    const content = this.files.get(path);
+    if (content === undefined) return false;
+    await mkdir(dirname(destination), { recursive: true });
+    await writeFile(destination, content);
+    return true;
+  }
+
+  async uploadFile(path: string, source: string): Promise<void> {
+    this.files.set(path, await readFile(source));
+    this.clock += 1;
+    this.uploaded.set(path, String(this.clock).padStart(8, "0"));
+  }
+
+  list(prefix: string): Promise<readonly BucketFile[]> {
+    return Promise.resolve(
+      [...this.files.keys()]
+        .filter((path) => path.startsWith(`${prefix}/`))
+        .map((path) => {
+          const uploadedAt = this.uploaded.get(path);
+          return { path, ...(uploadedAt === undefined ? {} : { uploadedAt }) };
+        }),
+    );
+  }
+
+  remove(paths: readonly string[]): Promise<void> {
+    for (const path of paths) {
+      this.files.delete(path);
+      this.uploaded.delete(path);
+      this.removed.push(path);
+    }
+    return Promise.resolve();
+  }
+}
+
+let dir: string;
+let source: FakeSource;
+let bucket: FakeBucket;
+let hub: FakeHub;
+
+beforeEach(async () => {
+  dir = await mkdtemp(join(tmpdir(), "xtap-index-test-"));
+  source = new FakeSource();
+  bucket = new FakeBucket();
+  hub = new FakeHub();
+});
+
+afterEach(async () => {
+  await rm(dir, { recursive: true, force: true });
+});
+
+function options(name: string): DurableIndexOptions {
+  return {
+    datasetRepo: DATASET,
+    indexBucket: BUCKET,
+    accessToken: "hf_test",
+    databasePath: join(dir, `${name}.sqlite`),
+    mirror: new DatasetMirror(hub, join(dir, `${name}-mirror`)),
+    taxonomyVersion: 1,
+    contractHash: CONTRACT,
+    sourceClient: source,
+    bucketClient: bucket,
+  };
+}
+
+function tweetLine(id: string, author = "someone"): string {
+  return `${JSON.stringify(
+    makePooled({
+      id,
+      url: `https://x.com/${author}/status/${id}`,
+      author: { username: author, display_name: author },
+      conversation_id: id,
+    }),
+  )}\n`;
+}
+
+describe("durable enrichment index", () => {
+  it("bootstraps, publishes, restores, and applies only a strict append suffix", async () => {
+    const path = "data/osolmaz/2026/08/tweets-2026-08-04.jsonl";
+    source.files.set(path, tweetLine("1"));
+    const initial = await DurableIndex.bootstrap(options("bootstrap"));
+    expect(initial.store.count()).toBe(1);
+    const firstManifest = await initial.publish();
+    initial.close();
+
+    expect(durableIndexManifestSchema.parse(firstManifest)).toEqual(firstManifest);
+    expect(() =>
+      durableIndexManifestSchema.parse({ ...firstManifest, unexpected: true }),
+    ).toThrow();
+    const restoreOptions = options("restored");
+    await Promise.all([
+      writeFile(restoreOptions.databasePath, "stale database"),
+      writeFile(`${restoreOptions.databasePath}-wal`, "stale wal"),
+      writeFile(`${restoreOptions.databasePath}-shm`, "stale shm"),
+    ]);
+    const restored = await DurableIndex.restore(restoreOptions);
+    expect(restored.store.count()).toBe(1);
+    source.files.set(path, `${source.files.get(path) ?? ""}${tweetLine("2", "other")}`);
+    source.advanceRevision();
+
+    const advanced = await restored.advanceToLatest();
+    expect(advanced).toMatchObject({ filesChanged: 1, rowsApplied: 1 });
+    expect(restored.store.count()).toBe(2);
+    await expect(restored.advanceToLatest()).resolves.toMatchObject({
+      filesChanged: 0,
+      rowsApplied: 0,
+    });
+    restored.close();
+  });
+
+  it("replays enrichment, attempt, registry, and receipt suffixes", async () => {
+    const tweetPath = "data/osolmaz/2026/08/tweets-2026-08-04.jsonl";
+    source.files.set(tweetPath, `${tweetLine("1")}${tweetLine("2", "other")}`);
+    const eventOptions = options("events");
+    const index = await DurableIndex.bootstrap(eventOptions);
+    const items = index.enrichStore.claimQueued(10);
+    index.enrichStore.releaseClaims();
+    const first = items.find((item) => item.tweetIds.includes("1"));
+    const second = items.find((item) => item.tweetIds.includes("2"));
+    if (first === undefined || second === undefined) throw new Error("expected queue items");
+
+    source.files.set(
+      "enrichment/2026/08/enrichment-2026-08-04.jsonl",
+      `${JSON.stringify({
+        unit_id: first.unitId,
+        tweet_ids: first.tweetIds,
+        input_hash: first.inputHash,
+        contract_hash: first.contractHash,
+        preset_labels: [],
+        free_labels: [],
+        model: "model",
+        taxonomy_version: 1,
+        enriched_at: "2026-08-04T00:00:00.000Z",
+      })}\n`,
+    );
+    source.files.set(
+      "enrichment/attempts/2026/08/attempts-2026-08-04.jsonl",
+      `${JSON.stringify({
+        unit_id: second.unitId,
+        input_hash: second.inputHash,
+        contract_hash: second.contractHash,
+        attempt: 1,
+        outcome: "transient_failure",
+        error_class: "timeout",
+        at: "2026-08-04T00:00:00.000Z",
+      })}\n`,
+    );
+    source.files.set(
+      "enrichment/registry/2026/08/registry-2026-08-04.jsonl",
+      `${JSON.stringify({
+        name: "new-label",
+        status: "candidate",
+        at: "2026-08-04T00:00:00.000Z",
+        actor: "worker",
+        contract_hash: CONTRACT,
+        registry_revision: 2,
+      })}\n`,
+    );
+    source.files.set(
+      "enrichment/receipts/2026-08-04.jsonl",
+      `${JSON.stringify(receiptFixture("job-1"))}\n`,
+    );
+    source.advanceRevision();
+
+    const advanced = await index.advanceToLatest();
+    expect(advanced.filesChanged).toBe(4);
+    expect(index.enrichStore.queueEntry(first.unitId)?.status).toBe("done");
+    expect(index.enrichStore.queueEntry(second.unitId)?.status).toBe("retrying");
+    expect(index.enrichStore.registryStatus("new-label")).toBe("candidate");
+    expect(eventOptions.mirror.latestReceipt()?.worker_id).toBe("job-1");
+    expect(index.stats()).toMatchObject({
+      enrichmentRows: 1,
+      attemptEvents: 1,
+      registryEvents: 1,
+    });
+    index.close();
+  });
+
+  it("fails closed on prefix edits, truncation, deletion, and checksum mismatch", async () => {
+    const path = "data/osolmaz/2026/08/tweets-2026-08-04.jsonl";
+    source.files.set(path, `${tweetLine("1")}${tweetLine("2")}`);
+    const index = await DurableIndex.bootstrap(options("failures"));
+    await index.publish();
+
+    source.files.set(path, `${tweetLine("1")}${tweetLine("2")}not json\n`);
+    source.advanceRevision();
+    await expect(index.advanceToLatest()).rejects.toThrow("invalid JSON");
+    source.files.set(path, `${tweetLine("1")}${tweetLine("2")}{}\n`);
+    source.advanceRevision();
+    await expect(index.advanceToLatest()).rejects.toThrow("invalid tweet record");
+    source.files.set(path, `${tweetLine("changed")}${tweetLine("2")}`);
+    source.advanceRevision();
+    await expect(index.advanceToLatest()).rejects.toThrow("prefix changed");
+    source.files.set(path, tweetLine("1"));
+    source.advanceRevision();
+    await expect(index.advanceToLatest()).rejects.toThrow("truncated");
+    source.files.delete(path);
+    source.advanceRevision();
+    await expect(index.advanceToLatest()).rejects.toThrow("were deleted");
+    source.files.set(path, tweetLine("1").trimEnd());
+    source.advanceRevision();
+    await expect(index.advanceToLatest()).rejects.toThrow("complete JSONL line");
+    source.files.set(path, `${tweetLine("1")}${tweetLine("2")}`);
+    source.files.set("data/unknown.jsonl", tweetLine("1"));
+    source.advanceRevision();
+    await expect(index.advanceToLatest()).rejects.toThrow("unsupported dataset index source");
+    index.close();
+
+    const manifest = JSON.parse(source.textFiles.get("index/current.json") ?? "{}") as {
+      database: { key: string };
+    };
+    bucket.files.set(manifest.database.key, Buffer.from("corrupt"));
+    await expect(DurableIndex.restore(options("corrupt"))).rejects.toThrow("checksum mismatch");
+  });
+
+  it("refuses an atomic manifest commit after the dataset head changes", async () => {
+    source.files.set("data/osolmaz/2026/08/tweets-2026-08-04.jsonl", tweetLine("1"));
+    const index = await DurableIndex.bootstrap(options("race"));
+    source.advanceRevision();
+
+    await expect(index.publish()).rejects.toThrow("parent commit does not match");
+    expect(source.textFiles.has("index/current.json")).toBe(false);
+    index.close();
+  });
+
+  it("re-advances and retries when the dataset changes during publication", async () => {
+    source.files.set("data/osolmaz/2026/08/tweets-2026-08-04.jsonl", tweetLine("1"));
+    const index = await DurableIndex.bootstrap(options("publication-retry"));
+    source.commitErrors.push(new Error("The branch was updated since publication started"));
+
+    const published = await index.publishLatest();
+
+    expect(published.manifest.dataset.revision).toBe(published.advance.revision);
+    expect(source.textFiles.has("index/current.json")).toBe(true);
+    expect(bucket.files.has(published.manifest.database.key)).toBe(true);
+    index.close();
+  });
+
+  it("retries status-code conflicts but rejects unrelated and exhausted failures", async () => {
+    source.files.set("data/osolmaz/2026/08/tweets-2026-08-04.jsonl", tweetLine("1"));
+    const index = await DurableIndex.bootstrap(options("publication-errors"));
+    source.commitErrors.push(Object.assign(new Error("conflict"), { statusCode: 409 }));
+    await expect(index.publishLatest()).resolves.toHaveProperty("manifest.dataset.repo", DATASET);
+
+    source.commitErrors.push(new Error("permission denied"));
+    await expect(index.publishLatest()).rejects.toThrow("permission denied");
+    source.commitErrors.push("non-error rejection");
+    await expect(index.publishLatest()).rejects.toBe("non-error rejection");
+
+    source.commitErrors.push(
+      ...Array.from(
+        { length: 5 },
+        () => new Error("The branch was updated since publication started"),
+      ),
+    );
+    await expect(index.publishLatest()).rejects.toThrow("branch was updated");
+    index.close();
+  });
+
+  it("does not let failed publication uploads displace successful predecessors", async () => {
+    const path = "data/osolmaz/2026/08/tweets-2026-08-04.jsonl";
+    source.files.set(path, tweetLine("1"));
+    const index = await DurableIndex.bootstrap(options("orphan-retention"));
+    const first = await index.publish();
+    for (let generation = 2; generation <= 5; generation += 1) {
+      source.files.set(path, `${source.files.get(path) ?? ""}${tweetLine(String(generation))}`);
+      source.advanceRevision();
+      await index.advanceToLatest();
+      source.commitErrors.push(new Error("permission denied"));
+      await expect(index.publish()).rejects.toThrow("permission denied");
+    }
+    source.files.set(path, `${source.files.get(path) ?? ""}${tweetLine("6")}`);
+    source.advanceRevision();
+    await index.advanceToLatest();
+    const final = await index.publish();
+
+    expect(final.database.predecessors).toEqual([first.database.key]);
+    expect([...bucket.files.keys()].filter((key) => key.endsWith(".sqlite")).sort()).toEqual(
+      [first.database.key, final.database.key].sort(),
+    );
+    index.close();
+  });
+
+  it("retains the active database and three recent predecessors", async () => {
+    const path = "data/osolmaz/2026/08/tweets-2026-08-04.jsonl";
+    source.files.set(path, tweetLine("1"));
+    const index = await DurableIndex.bootstrap(options("retention"));
+    for (let generation = 0; generation < 6; generation += 1) {
+      if (generation > 0) {
+        source.files.set(
+          path,
+          `${source.files.get(path) ?? ""}${tweetLine(String(generation + 1))}`,
+        );
+        source.advanceRevision();
+        await index.advanceToLatest();
+      }
+      await index.publish();
+    }
+    const databases = [...bucket.files.keys()].filter((key) => key.startsWith("index/databases/"));
+    expect(databases).toHaveLength(4);
+    const current = JSON.parse(source.textFiles.get("index/current.json") ?? "{}") as {
+      database: { key: string; predecessors: string[] };
+    };
+    expect(current.database.predecessors).toHaveLength(3);
+    expect(databases.sort()).toEqual(
+      [current.database.key, ...current.database.predecessors].sort(),
+    );
+    expect(bucket.removed).toHaveLength(2);
+    index.close();
+  });
+});
+
+function receiptFixture(workerId: string): Record<string, unknown> {
+  return {
+    started_at: "2026-08-04T00:00:00.000Z",
+    finished_at: "2026-08-04T00:01:00.000Z",
+    units: 1,
+    calls: 1,
+    prompt_tokens: 1,
+    completion_tokens: 1,
+    cost_usd: 0.00001,
+    failures: 0,
+    retries: 0,
+    blocked: 0,
+    contract_hash: CONTRACT,
+    worker_id: workerId,
+    discarded_assignments: 0,
+    new_candidates: 0,
+    new_approvals: 0,
+    new_rejections: 0,
+  };
+}
+
+function sha256(value: string): string {
+  return createHash("sha256").update(value).digest("hex");
+}

space/tests/helpers.ts

@@ -7,6 +7,7 @@ export const testConfig: SpaceConfig = {
   port: 7860,
   dataDir: ".data-test",
   datasetRepo: "osolmaz/xtap-pool-data",
+  indexBucket: "osolmaz/xtap-pool-bucket",
   hfToken: "hf_test_token",
   poolSigningSecret: "pool-secret-0123456789abcdef0123456789abcdef",
   sessionSecret: "session-secret-0123456789abcdef0123456789ab",