diff --git a/core/errors/__tests__/explanations.test.ts b/core/errors/__tests__/explanations.test.ts
new file mode 100644
index 0000000..df127b5
--- /dev/null
+++ b/core/errors/__tests__/explanations.test.ts
@@ -0,0 +1,211 @@
+import { describe, expect, it } from "vitest";
+import { ERROR_CODES } from "@/core/errors/error-codes";
+import { UnauthorizedError } from "@/core/auth/middleware";
+import { issueTicket, redeemTicket, transferTicket } from "@/core/invariants/mutations";
+import type { DomainState } from "@/core/invariants/types";
+import { err } from "@/core/result/result";
+import {
+  REJECTION_INVARIANTS,
+  REJECTION_REASONS,
+  REJECTION_SOURCE_CODES,
+  explainRejection,
+  renderRejection,
+  unmappedErrorCodes
+} from "@/core/errors/explanations";
+
+const STATE: DomainState = {
+  events: new Map([
+    [
+      "evt-1",
+      { id: "evt-1", organizerAddress: "G_ORG", totalCapacity: 1, issuedCount: 1, status: "active", feeBasisPoints: 250 }
+    ]
+  ]),
+  tickets: new Map([
+    [
+      "t-1",
+      {
+        id: "t-1",
+        eventId: "evt-1",
+        tierId: "general",
+        ownerAddress: "G_ALICE",
+        priceStroops: 100,
+        status: "issued",
+        issuedAt: 0,
+        validFrom: 100,
+        validUntil: 200,
+        fraudFlag: false,
+        nonce: 0,
+        metadataHash: ""
+      }
+    ]
+  ]),
+  transfers: [],
+  processedNonces: new Set(["used-nonce"]),
+  blacklistedAddresses: new Set(["G_EVE"])
+};
+
+describe("rejection explanations (#190)", () => {
+  it("gives every stable error code a reason code", () => {
+    expect(unmappedErrorCodes()).toEqual([]);
+  });
+
+  it("only maps to reasons that exist and carry recovery hints", () => {
+    const targets = [...Object.values(REJECTION_SOURCE_CODES), ...Object.values(REJECTION_INVARIANTS)];
+    for (const target of targets) {
+      expect(REJECTION_REASONS[target]).toBeDefined();
+    }
+    for (const reason of Object.values(REJECTION_REASONS)) {
+      expect(reason.recovery.length).toBeGreaterThan(0);
+    }
+  });
+
+  it("explains every stable error code as an HTTP-class rejection", () => {
+    for (const definition of Object.values(ERROR_CODES)) {
+      const explanation = explainRejection(definition.code);
+      expect(explanation.status).toBeGreaterThanOrEqual(400);
+      expect(explanation.retryable).toBeTypeOf("boolean");
+    }
+  });
+
+  describe("validation rejection", () => {
+    it("explains an invalid amount from an invariant violation", () => {
+      const roomy: DomainState = {
+        ...STATE,
+        events: new Map([["evt-1", { ...STATE.events.get("evt-1")!, totalCapacity: 5 }]])
+      };
+      const amount = issueTicket(roomy, {
+        id: "t-2",
+        eventId: "evt-1",
+        tierId: "general",
+        ownerAddress: "G_BOB",
+        priceStroops: -1,
+        validFrom: 1,
+        validUntil: 2
+      });
+      if (amount.ok) throw new Error("expected a rejection");
+      const explanation = explainRejection(amount.code);
+      expect(explanation.reasonCode).toBe("invalid_amount");
+      expect(explanation.category).toBe("validation");
+    });
+
+    it("accepts a failed domain result as well as its violation", () => {
+      const early = redeemTicket(STATE, { ticketId: "t-1", claimerAddress: "G_ALICE", timestamp: 5 });
+      expect(explainRejection(early).reasonCode).toBe("outside_validity_window");
+    });
+
+    it("explains a result error code and an idempotency key problem", () => {
+      expect(explainRejection(err("INVALID_ADDRESS")).reasonCode).toBe("invalid_address");
+      expect(explainRejection(err("idempotency_key_invalid")).category).toBe("validation");
+    });
+  });
+
+  describe("policy rejection", () => {
+    it("explains capacity, lock and blacklist rejections", () => {
+      const full = issueTicket(STATE, {
+        id: "t-2",
+        eventId: "evt-1",
+        tierId: "general",
+        ownerAddress: "G_BOB",
+        priceStroops: 1,
+        validFrom: 1,
+        validUntil: 2
+      });
+      if (full.ok) throw new Error("expected a rejection");
+      expect(explainRejection(full.code).reasonCode).toBe("capacity_reached");
+
+      expect(explainRejection({ invariantId: "INV-05" }).reasonCode).toBe("ticket_locked");
+      expect(explainRejection({ invariantId: "INV-11" }).reasonCode).toBe("account_restricted");
+      expect(explainRejection(err("quota_exceeded")).reasonCode).toBe("rate_limited");
+      expect(explainRejection("quota_exceeded").retryable).toBe(true);
+    });
+
+    it("explains a redemption outside the validity window", () => {
+      const early = redeemTicket(STATE, { ticketId: "t-1", claimerAddress: "G_ALICE", timestamp: 5 });
+      if (early.ok) throw new Error("expected a rejection");
+      const explanation = explainRejection(early.code);
+      expect(explanation.reasonCode).toBe("outside_validity_window");
+      expect(explanation.category).toBe("policy");
+    });
+  });
+
+  describe("permission rejection", () => {
+    it("explains a thrown UnauthorizedError without exposing its message", () => {
+      const error = new UnauthorizedError("Permission denied: admin:config requires elevated privileges", "admin:config", "viewer");
+      const explanation = explainRejection(error);
+      expect(explanation.reasonCode).toBe("permission_denied");
+      expect(explanation.category).toBe("permission");
+      expect(JSON.stringify(explanation)).not.toContain("admin:config");
+    });
+
+    it("explains a wrong owner from the domain layer", () => {
+      const stolen = transferTicket(STATE, {
+        id: "x-1",
+        ticketId: "t-1",
+        fromAddress: "G_MALLORY",
+        toAddress: "G_BOB",
+        priceStroops: 10,
+        organizerFeeStroops: 1,
+        protocolFeeStroops: 1,
+        sellerProceedsStroops: 8,
+        nonce: "fresh"
+      });
+      if (stolen.ok) throw new Error("expected a rejection");
+      expect(explainRejection(stolen.code).reasonCode).toBe("not_owner");
+      expect(explainRejection("export_denied").category).toBe("permission");
+    });
+  });
+
+  describe("stale-state rejection", () => {
+    it("explains a replayed nonce and an in-flight duplicate", () => {
+      const replayed = transferTicket(STATE, {
+        id: "x-2",
+        ticketId: "t-1",
+        fromAddress: "G_ALICE",
+        toAddress: "G_BOB",
+        priceStroops: 10,
+        organizerFeeStroops: 1,
+        protocolFeeStroops: 1,
+        sellerProceedsStroops: 8,
+        nonce: "used-nonce"
+      });
+      if (replayed.ok) throw new Error("expected a rejection");
+      expect(explainRejection(replayed.code).reasonCode).toBe("duplicate_request");
+      const inFlight = explainRejection("idempotency_in_flight");
+      expect(inFlight.category).toBe("stale_state");
+      expect(inFlight.retryable).toBe(true);
+    });
+
+    it("explains an already redeemed ticket", () => {
+      expect(explainRejection("ALREADY_REDEEMED").category).toBe("stale_state");
+    });
+  });
+
+  describe("unexpected failure fallback", () => {
+    it("never returns the raw message of an unknown error", () => {
+      const explanation = explainRejection(new Error("ECONNRESET at 10.0.0.7:5432 password=hunter2"), {
+        correlationId: "corr-1"
+      });
+      expect(explanation.reasonCode).toBe("unexpected_failure");
+      expect(explanation.category).toBe("unexpected");
+      expect(explanation.reference).toBe("corr-1");
+      expect(JSON.stringify(explanation)).not.toContain("hunter2");
+    });
+
+    it("falls back for unknown codes, null and prototype keys", () => {
+      for (const input of [undefined, null, 42, "no_such_code", "constructor", err("no_such_code"), {}]) {
+        expect(explainRejection(input).reasonCode).toBe("unexpected_failure");
+      }
+    });
+
+    it("does not treat a success result as a rejection code", () => {
+      expect(explainRejection({ ok: true, code: "FORBIDDEN" }).reasonCode).toBe("unexpected_failure");
+    });
+  });
+
+  it("renders the same text for the same rejection", () => {
+    const text = renderRejection(explainRejection("TICKET_NOT_FOUND", { correlationId: "corr-2" }));
+    expect(text).toContain("Not found: The item you asked for could not be found.");
+    expect(text).toContain("What you can do:");
+    expect(text).toContain("Reference: corr-2");
+  });
+});
diff --git a/core/errors/explanations.ts b/core/errors/explanations.ts
new file mode 100644
index 0000000..41a39b9
--- /dev/null
+++ b/core/errors/explanations.ts
@@ -0,0 +1,402 @@
+/**
+ * User-facing explanations for rejected operations (#190).
+ *
+ * Low-level failures reach the edge of the app in several shapes: a `Result`
+ * error code, a domain `InvariantViolation`, an `UnauthorizedError`, or an
+ * unexpected thrown value. This module maps every one of them onto a single
+ * `RejectionExplanation` so each client renders the same stable reason code,
+ * the same plain-language message and the same recovery hints.
+ *
+ * Rules:
+ *
+ * - **reason codes are stable.** Clients switch on `reasonCode`; a code is
+ *   never renamed or removed, a new one is added instead;
+ * - **explanations are user-safe.** Only the fixed copy below is returned.
+ *   Raw error messages, entity ids, limits and stack traces never leave this
+ *   module, so a low-level message cannot leak through a rejection;
+ * - **unknown input falls back.** Anything that is not recognised becomes
+ *   `unexpected_failure`, so a caller always receives an explanation.
+ */
+
+import { ERROR_CODES } from "@/core/errors/error-codes";
+import { UnauthorizedError } from "@/core/auth/middleware";
+
+export const REJECTION_SCHEMA_VERSION = 1;
+
+export type RejectionCategory =
+  | "validation"
+  | "policy"
+  | "permission"
+  | "stale_state"
+  | "unexpected";
+
+export interface RejectionReason {
+  readonly category: RejectionCategory;
+  readonly title: string;
+  readonly message: string;
+  /** Ordered, actionable steps. Never empty. */
+  readonly recovery: readonly string[];
+  readonly retryable: boolean;
+  /** HTTP status an API response should use. */
+  readonly status: number;
+}
+
+export const REJECTION_REASONS = {
+  // ── Validation ────────────────────────────────────────────────────────────
+  invalid_input: {
+    category: "validation",
+    title: "Check your input",
+    message: "Some of the information you entered is not valid.",
+    recovery: ["Review the highlighted fields.", "Correct them and submit again."],
+    retryable: false,
+    status: 400
+  },
+  invalid_address: {
+    category: "validation",
+    title: "Invalid address",
+    message: "The Stellar address is not valid.",
+    recovery: ["Use a public key that starts with G and is 56 characters long.", "Never paste a secret key."],
+    retryable: false,
+    status: 400
+  },
+  invalid_amount: {
+    category: "validation",
+    title: "Invalid amount",
+    message: "The amount is not valid.",
+    recovery: ["Enter a positive whole or decimal number with at most 7 decimal places."],
+    retryable: false,
+    status: 400
+  },
+  missing_field: {
+    category: "validation",
+    title: "Missing information",
+    message: "A required field was left out.",
+    recovery: ["Fill in every required field.", "Submit again."],
+    retryable: false,
+    status: 400
+  },
+  unsupported_version: {
+    category: "validation",
+    title: "Unsupported version",
+    message: "The requested format version is not supported.",
+    recovery: ["Update the client to the current version.", "Request the current schema version."],
+    retryable: false,
+    status: 400
+  },
+  invalid_request_key: {
+    category: "validation",
+    title: "Invalid request key",
+    message: "The request key is missing or malformed.",
+    recovery: ["Send a unique request key of 8 to 128 letters, digits or . _ : - characters."],
+    retryable: false,
+    status: 400
+  },
+  request_key_reused: {
+    category: "validation",
+    title: "Request key already used",
+    message: "That request key was already used for a different request.",
+    recovery: ["Generate a new request key for each distinct request."],
+    retryable: false,
+    status: 409
+  },
+
+  // ── Policy ────────────────────────────────────────────────────────────────
+  capacity_reached: {
+    category: "policy",
+    title: "Sold out",
+    message: "The event has reached its maximum capacity.",
+    recovery: ["Check whether more tickets are released later.", "Contact the organizer about a waitlist."],
+    retryable: false,
+    status: 409
+  },
+  ticket_locked: {
+    category: "policy",
+    title: "Ticket is locked",
+    message: "This ticket is locked and cannot be used for this action right now.",
+    recovery: ["Wait for the lock to clear.", "Contact support if you believe the lock is a mistake."],
+    retryable: false,
+    status: 409
+  },
+  account_restricted: {
+    category: "policy",
+    title: "Account restricted",
+    message: "An account involved in this action is restricted.",
+    recovery: ["Use a different account.", "Contact support to review the restriction."],
+    retryable: false,
+    status: 403
+  },
+  outside_validity_window: {
+    category: "policy",
+    title: "Outside the valid time window",
+    message: "This action is only allowed within the ticket's validity window.",
+    recovery: ["Check the start and end time on the ticket.", "Try again within that window."],
+    retryable: false,
+    status: 409
+  },
+  event_cancelled: {
+    category: "policy",
+    title: "Event cancelled",
+    message: "This event has been cancelled.",
+    recovery: ["Contact the organizer about refunds."],
+    retryable: false,
+    status: 409
+  },
+  rate_limited: {
+    category: "policy",
+    title: "Too many requests",
+    message: "You have made too many requests in a short time.",
+    recovery: ["Wait a little while.", "Try again at a lower pace."],
+    retryable: true,
+    status: 429
+  },
+
+  // ── Permission ────────────────────────────────────────────────────────────
+  unauthenticated: {
+    category: "permission",
+    title: "Sign in required",
+    message: "You need to sign in to do this.",
+    recovery: ["Sign in again.", "Retry the action."],
+    retryable: false,
+    status: 401
+  },
+  permission_denied: {
+    category: "permission",
+    title: "Not allowed",
+    message: "You do not have permission to do this.",
+    recovery: ["Ask an administrator for access.", "Switch to an account that has access."],
+    retryable: false,
+    status: 403
+  },
+  not_owner: {
+    category: "permission",
+    title: "Not the owner",
+    message: "Only the current owner of this item can do this.",
+    recovery: ["Switch to the wallet that owns the item.", "Check who currently owns it."],
+    retryable: false,
+    status: 403
+  },
+
+  // ── Stale state ───────────────────────────────────────────────────────────
+  not_found: {
+    category: "stale_state",
+    title: "Not found",
+    message: "The item you asked for could not be found.",
+    recovery: ["Refresh the page and check the id.", "It may have been moved or removed."],
+    retryable: false,
+    status: 404
+  },
+  already_redeemed: {
+    category: "stale_state",
+    title: "Already redeemed",
+    message: "This ticket has already been redeemed.",
+    recovery: ["Each ticket can be redeemed once.", "Refresh to see the current ticket status."],
+    retryable: false,
+    status: 409
+  },
+  already_transferred: {
+    category: "stale_state",
+    title: "Already transferred",
+    message: "This ticket has already been transferred.",
+    recovery: ["Refresh to see the current owner."],
+    retryable: false,
+    status: 409
+  },
+  invalid_state_transition: {
+    category: "stale_state",
+    title: "Action no longer possible",
+    message: "The item is in a state that does not allow this action.",
+    recovery: ["Refresh to see the current status.", "Start again from the latest state."],
+    retryable: false,
+    status: 409
+  },
+  duplicate_request: {
+    category: "stale_state",
+    title: "Already processed",
+    message: "This request was already processed.",
+    recovery: ["Refresh to see the result of the earlier request.", "Do not submit it again."],
+    retryable: false,
+    status: 409
+  },
+  request_in_progress: {
+    category: "stale_state",
+    title: "Still in progress",
+    message: "An earlier identical request is still being processed.",
+    recovery: ["Wait a moment.", "Refresh to see the result before trying again."],
+    retryable: true,
+    status: 409
+  },
+  request_expired: {
+    category: "stale_state",
+    title: "Request expired",
+    message: "This request is too old to be replayed.",
+    recovery: ["Start the action again with a new request."],
+    retryable: false,
+    status: 410
+  },
+
+  // ── Unexpected ────────────────────────────────────────────────────────────
+  network_unavailable: {
+    category: "unexpected",
+    title: "Network problem",
+    message: "We could not reach the Stellar network.",
+    recovery: ["Check your connection.", "Try again in a moment."],
+    retryable: true,
+    status: 502
+  },
+  timed_out: {
+    category: "unexpected",
+    title: "Took too long",
+    message: "The operation timed out.",
+    recovery: ["Try again.", "Check the item's status before repeating a payment."],
+    retryable: true,
+    status: 504
+  },
+  unexpected_failure: {
+    category: "unexpected",
+    title: "Something went wrong",
+    message: "An unexpected error occurred.",
+    recovery: ["Try again later.", "If it keeps happening, contact support and quote the reference."],
+    retryable: true,
+    status: 500
+  }
+} as const satisfies Record<string, RejectionReason>;
+
+export type RejectionReasonCode = keyof typeof REJECTION_REASONS;
+
+/** The explanation object every client receives. */
+export interface RejectionExplanation extends RejectionReason {
+  readonly schemaVersion: typeof REJECTION_SCHEMA_VERSION;
+  readonly reasonCode: RejectionReasonCode;
+  /** Quote this to support. Present only when the caller supplied one. */
+  readonly reference?: string;
+}
+
+export interface ExplainOptions {
+  readonly correlationId?: string;
+}
+
+/**
+ * Low-level codes (stable error codes and `Result` codes from idempotency,
+ * quota, export and replay) and the reason each one is explained as.
+ */
+export const REJECTION_SOURCE_CODES: Readonly<Record<string, RejectionReasonCode>> = {
+  INVALID_INPUT: "invalid_input",
+  INVALID_ADDRESS: "invalid_address",
+  INVALID_AMOUNT: "invalid_amount",
+  MISSING_REQUIRED_FIELD: "missing_field",
+  UNAUTHORIZED: "unauthenticated",
+  TOKEN_EXPIRED: "unauthenticated",
+  INVALID_SIGNATURE: "unauthenticated",
+  FORBIDDEN: "permission_denied",
+  NOT_OWNER: "not_owner",
+  NOT_FOUND: "not_found",
+  TICKET_NOT_FOUND: "not_found",
+  EVENT_NOT_FOUND: "not_found",
+  ALREADY_REDEEMED: "already_redeemed",
+  ALREADY_TRANSFERRED: "already_transferred",
+  EVENT_CANCELLED: "event_cancelled",
+  RATE_LIMITED: "rate_limited",
+  INTERNAL_ERROR: "unexpected_failure",
+  NETWORK_ERROR: "network_unavailable",
+  TIMEOUT: "timed_out",
+
+  idempotency_key_missing: "invalid_request_key",
+  idempotency_key_invalid: "invalid_request_key",
+  idempotency_key_conflict: "request_key_reused",
+  idempotency_key_expired: "request_expired",
+  idempotency_in_flight: "request_in_progress",
+  idempotency_not_found: "not_found",
+
+  quota_exceeded: "rate_limited",
+  quota_override_denied: "permission_denied",
+  quota_diagnostics_denied: "permission_denied",
+
+  export_denied: "permission_denied",
+  field_authorization_denied: "permission_denied",
+  schema_unsupported: "unsupported_version",
+
+  replay_unauthorized: "permission_denied",
+  invalid_options: "invalid_input",
+  replay_in_progress: "request_in_progress",
+  version_active: "invalid_state_transition",
+  source_unavailable: "network_unavailable"
+};
+
+/** Domain invariants (`core/invariants`) and the reason each one is explained as. */
+export const REJECTION_INVARIANTS: Readonly<Record<string, RejectionReasonCode>> = {
+  "INV-01": "capacity_reached",
+  // INV-02 also reports a missing ticket; the copy covers both without leaking ids.
+  "INV-02": "not_owner",
+  "INV-03": "invalid_state_transition",
+  "INV-04": "invalid_state_transition",
+  "INV-05": "ticket_locked",
+  "INV-06": "invalid_amount",
+  "INV-07": "invalid_amount",
+  "INV-08": "outside_validity_window",
+  "INV-09": "duplicate_request",
+  "INV-10": "capacity_reached",
+  "INV-11": "account_restricted"
+};
+
+/** Builds the explanation for an already-known reason code. */
+export function explainReason(reasonCode: RejectionReasonCode, options: ExplainOptions = {}): RejectionExplanation {
+  const reason: RejectionReason = REJECTION_REASONS[reasonCode];
+  return {
+    schemaVersion: REJECTION_SCHEMA_VERSION,
+    reasonCode,
+    ...reason,
+    ...(options.correlationId ? { reference: options.correlationId } : {})
+  };
+}
+
+function lookup(table: Readonly<Record<string, RejectionReasonCode>>, key: unknown): RejectionReasonCode | undefined {
+  return typeof key === "string" && Object.prototype.hasOwnProperty.call(table, key) ? table[key] : undefined;
+}
+
+function reasonFor(input: unknown): RejectionReasonCode {
+  if (input instanceof UnauthorizedError) return "permission_denied";
+
+  if (typeof input === "string") {
+    return lookup(REJECTION_SOURCE_CODES, input) ?? "unexpected_failure";
+  }
+
+  if (input && typeof input === "object") {
+    const candidate = input as { ok?: unknown; code?: unknown; invariantId?: unknown };
+    const fromInvariant = lookup(REJECTION_INVARIANTS, candidate.invariantId);
+    if (fromInvariant) return fromInvariant;
+    // A failed `Result`, or any object that carries a stable code.
+    if (candidate.ok !== true) {
+      const fromCode = lookup(REJECTION_SOURCE_CODES, candidate.code);
+      if (fromCode) return fromCode;
+      // Domain mutations fail with the violation itself as the code.
+      if (candidate.ok === false && candidate.code && typeof candidate.code === "object") {
+        return reasonFor(candidate.code);
+      }
+    }
+  }
+
+  return "unexpected_failure";
+}
+
+/**
+ * Maps any rejection onto its explanation. Never throws and never returns the
+ * input's own message: unrecognised input becomes `unexpected_failure`.
+ */
+export function explainRejection(input: unknown, options: ExplainOptions = {}): RejectionExplanation {
+  return explainReason(reasonFor(input), options);
+}
+
+/** Plain-text rendering shared by clients that do not build their own UI. */
+export function renderRejection(explanation: RejectionExplanation): string {
+  const lines = [`${explanation.title}: ${explanation.message}`];
+  if (explanation.recovery.length > 0) {
+    lines.push("What you can do:", ...explanation.recovery.map((step) => `- ${step}`));
+  }
+  if (explanation.reference) lines.push(`Reference: ${explanation.reference}`);
+  return lines.join("\n");
+}
+
+/** Every known stable code must resolve to a reason; used by tests and docs. */
+export function unmappedErrorCodes(): string[] {
+  return Object.keys(ERROR_CODES).filter((code) => !lookup(REJECTION_SOURCE_CODES, code));
+}
diff --git a/core/export/__tests__/trends.test.ts b/core/export/__tests__/trends.test.ts
new file mode 100644
index 0000000..e538cc9
--- /dev/null
+++ b/core/export/__tests__/trends.test.ts
@@ -0,0 +1,231 @@
+import { beforeEach, describe, expect, it } from "vitest";
+import {
+  EXPORT_CURRENT_SCHEMA_VERSION,
+  EXPORT_GENERATOR,
+  createExportIdempotencyStore,
+  createExportQuotaStore
+} from "@/core/export/exporter";
+import {
+  TREND_MAX_WINDOWS,
+  TREND_RECORD_TYPE,
+  aggregateTrends,
+  exportTrends,
+  type TrendEvent,
+  type TrendRow
+} from "@/core/export/trends";
+
+const MAINTAINER = { kind: "maintainer", id: "ada" } as const;
+
+/** Fixture: two days of events across two operations. Totals are checked by hand. */
+const EVENTS: TrendEvent[] = [
+  { at: "2026-10-05T09:00:00.000Z", op: "ticket.redeem", result: "success", amountStroops: 100 },
+  { at: "2026-10-05T09:30:00.000Z", op: "ticket.redeem", result: "success", amountStroops: 250 },
+  { at: "2026-10-05T10:15:00.000Z", op: "ticket.redeem", result: "failure", amountStroops: 999 },
+  { at: "2026-10-05T23:59:59.999Z", op: "ticket.transfer", result: "success", amountStroops: 1_000 },
+  { at: "2026-10-06T00:00:00.000Z", op: "ticket.transfer", result: "failure" },
+  { at: "2026-10-06T12:00:00.000Z", op: "ticket.transfer", result: "success", amountStroops: 500 },
+  { at: "2026-10-12T08:00:00.000Z", op: "ticket.redeem", result: "success", amountStroops: 7 }
+];
+
+const DAY_RANGE = { window: "day", from: "2026-10-05T00:00:00.000Z", to: "2026-10-07T00:00:00.000Z" } as const;
+
+function rows(events: readonly TrendEvent[], query: Parameters<typeof aggregateTrends>[1]): TrendRow[] {
+  const result = aggregateTrends(events, query);
+  if (!result.ok) throw new Error(`expected rows, got ${result.code}`);
+  return result.value;
+}
+
+function withoutEnvelopeFields(record: Record<string, unknown>): Record<string, unknown> {
+  const copy = { ...record };
+  delete copy.recordType;
+  delete copy.scope;
+  return copy;
+}
+
+beforeEach(() => {
+  createExportQuotaStore();
+  createExportIdempotencyStore();
+});
+
+describe("aggregateTrends", () => {
+  it("matches the source fixture per day and operation", () => {
+    expect(rows(EVENTS, DAY_RANGE)).toEqual([
+      {
+        windowStart: "2026-10-05T00:00:00.000Z",
+        windowEnd: "2026-10-06T00:00:00.000Z",
+        op: "ticket.redeem",
+        total: 3,
+        successes: 2,
+        failures: 1,
+        failureRate: 0.3333,
+        volumeStroops: "350"
+      },
+      {
+        windowStart: "2026-10-05T00:00:00.000Z",
+        windowEnd: "2026-10-06T00:00:00.000Z",
+        op: "ticket.transfer",
+        total: 1,
+        successes: 1,
+        failures: 0,
+        failureRate: 0,
+        volumeStroops: "1000"
+      },
+      {
+        windowStart: "2026-10-06T00:00:00.000Z",
+        windowEnd: "2026-10-07T00:00:00.000Z",
+        op: "ticket.transfer",
+        total: 2,
+        successes: 1,
+        failures: 1,
+        failureRate: 0.5,
+        volumeStroops: "500"
+      }
+    ]);
+  });
+
+  it("treats the range as inclusive of from and exclusive of to", () => {
+    const tight = rows(EVENTS, { window: "day", from: "2026-10-05T23:59:59.999Z", to: "2026-10-06T00:00:00.000Z" });
+    expect(tight.map((row) => row.total)).toEqual([1]);
+    expect(tight[0].op).toBe("ticket.transfer");
+  });
+
+  it("buckets hours and Monday-based UTC weeks", () => {
+    const hourly = rows(EVENTS, { window: "hour", from: "2026-10-05T09:00:00.000Z", to: "2026-10-05T11:00:00.000Z" });
+    expect(hourly.map((row) => [row.windowStart, row.total])).toEqual([
+      ["2026-10-05T09:00:00.000Z", 2],
+      ["2026-10-05T10:00:00.000Z", 1]
+    ]);
+
+    const weekly = rows(EVENTS, { window: "week", from: "2026-10-05T00:00:00.000Z", to: "2026-10-19T00:00:00.000Z" });
+    expect(weekly.map((row) => [row.windowStart, row.op, row.total])).toEqual([
+      ["2026-10-05T00:00:00.000Z", "ticket.redeem", 3],
+      ["2026-10-05T00:00:00.000Z", "ticket.transfer", 3],
+      ["2026-10-12T00:00:00.000Z", "ticket.redeem", 1]
+    ]);
+  });
+
+  it("returns no rows for empty data or an empty range", () => {
+    expect(rows([], DAY_RANGE)).toEqual([]);
+    expect(rows(EVENTS, { window: "day", from: "2027-01-01T00:00:00.000Z", to: "2027-01-02T00:00:00.000Z" })).toEqual([]);
+  });
+
+  it("does not depend on event order and does not carry per-account fields", () => {
+    const withAccounts = EVENTS.map((event) => ({ ...event, account: "GSECRETACCOUNT", ipAddress: "10.0.0.1" }));
+    const shuffled = [...withAccounts].reverse();
+    expect(rows(shuffled, DAY_RANGE)).toEqual(rows(EVENTS, DAY_RANGE));
+    expect(JSON.stringify(rows(shuffled, DAY_RANGE))).not.toMatch(/GSECRETACCOUNT|10\.0\.0\.1/);
+  });
+
+  it("keeps stroop totals exact beyond the safe integer range", () => {
+    const big = Number.MAX_SAFE_INTEGER;
+    const result = rows(
+      [
+        { at: "2026-10-05T01:00:00.000Z", op: "pay", result: "success", amountStroops: big },
+        { at: "2026-10-05T02:00:00.000Z", op: "pay", result: "success", amountStroops: big }
+      ],
+      DAY_RANGE
+    );
+    expect(result[0].volumeStroops).toBe((BigInt(big) * 2n).toString());
+  });
+
+  it("rejects bad windows, ranges and events instead of under-counting", () => {
+    const code = (events: readonly TrendEvent[], query: Parameters<typeof aggregateTrends>[1]) => {
+      const result = aggregateTrends(events, query);
+      return result.ok ? "ok" : result.code;
+    };
+    expect(code(EVENTS, { ...DAY_RANGE, window: "minute" as never })).toBe("trend_invalid_window");
+    expect(code(EVENTS, { ...DAY_RANGE, from: "not-a-date" })).toBe("trend_invalid_range");
+    expect(code(EVENTS, { ...DAY_RANGE, to: DAY_RANGE.from })).toBe("trend_invalid_range");
+    expect(code(EVENTS, { window: "hour", from: 0, to: (TREND_MAX_WINDOWS + 1) * 3_600_000 })).toBe(
+      "trend_range_too_large"
+    );
+    expect(code([{ at: "garbage", op: "x", result: "success" }], DAY_RANGE)).toBe("trend_invalid_event");
+    expect(code([{ at: 0, op: "x", result: "success", amountStroops: -1 }], DAY_RANGE)).toBe("trend_invalid_event");
+    expect(code([{ at: 0, op: "x", result: "success", amountStroops: 1.5 }], DAY_RANGE)).toBe("trend_invalid_event");
+  });
+});
+
+describe("exportTrends", () => {
+  it("includes schema version and generation metadata", () => {
+    const result = exportTrends({ ...DAY_RANGE, actor: MAINTAINER, correlationId: "corr-1" }, EVENTS);
+    expect(result.ok).toBe(true);
+    if (!result.ok) return;
+    const envelope = result.value;
+    expect(envelope.schemaVersion).toBe(EXPORT_CURRENT_SCHEMA_VERSION);
+    expect(envelope.generator).toBe(EXPORT_GENERATOR);
+    expect(envelope.scope).toBe("maintainer");
+    expect(envelope.correlationId).toBe("corr-1");
+    expect(Date.parse(envelope.generatedAt)).not.toBeNaN();
+    expect(Date.parse(envelope.expiresAt)).toBeGreaterThan(Date.parse(envelope.generatedAt));
+    expect(envelope.totalRecords).toBe(3);
+    expect(envelope.records.every((record) => record.recordType === TREND_RECORD_TYPE)).toBe(true);
+  });
+
+  it("exports the same figures the aggregation produces", () => {
+    const result = exportTrends({ ...DAY_RANGE, actor: MAINTAINER }, EVENTS);
+    if (!result.ok) throw new Error(result.code);
+    const expected = rows(EVENTS, DAY_RANGE);
+    expect(result.value.records.map(withoutEnvelopeFields)).toEqual(expected);
+  });
+
+  it("returns an empty, still versioned envelope for empty data", () => {
+    const result = exportTrends({ ...DAY_RANGE, actor: MAINTAINER }, []);
+    if (!result.ok) throw new Error(result.code);
+    expect(result.value.recordCount).toBe(0);
+    expect(result.value.totalRecords).toBe(0);
+    expect(result.value.hasMore).toBe(false);
+    expect(result.value.schemaVersion).toBe(EXPORT_CURRENT_SCHEMA_VERSION);
+  });
+
+  it("denies every actor that is not a maintainer", () => {
+    for (const actor of [{ kind: "user", id: "bob" }, { kind: "service", id: "cron" }] as const) {
+      const result = exportTrends({ ...DAY_RANGE, actor }, EVENTS);
+      expect(result.ok).toBe(false);
+      if (!result.ok) expect(result.code).toBe("export_denied");
+    }
+  });
+
+  it("refuses a denied actor before reading the events", () => {
+    const result = exportTrends({ ...DAY_RANGE, window: "minute" as never, actor: { kind: "user" } }, EVENTS);
+    expect(result.ok).toBe(false);
+    if (!result.ok) expect(result.code).toBe("export_denied");
+  });
+
+  it("reports an invalid query to a maintainer", () => {
+    const result = exportTrends({ ...DAY_RANGE, to: DAY_RANGE.from, actor: MAINTAINER }, EVENTS);
+    expect(result.ok).toBe(false);
+    if (!result.ok) expect(result.code).toBe("trend_invalid_range");
+  });
+
+  it("pages a large result set without losing or repeating a row", () => {
+    const large: TrendEvent[] = [];
+    for (let index = 0; index < 5_000; index += 1) {
+      large.push({
+        at: Date.parse("2026-10-05T00:00:00.000Z") + index * 60_000,
+        op: `op.${index % 50}`,
+        result: index % 7 === 0 ? "failure" : "success",
+        amountStroops: index
+      });
+    }
+    const query = { window: "hour", from: "2026-10-05T00:00:00.000Z", to: "2026-10-09T00:00:00.000Z" } as const;
+    const expected = rows(large, query);
+    expect(expected.length).toBeGreaterThan(400);
+
+    const collected: Array<Record<string, unknown>> = [];
+    let cursor: string | undefined;
+    let pages = 0;
+    do {
+      const page = exportTrends({ ...query, actor: MAINTAINER, pageSize: 1_000, cursor }, large);
+      if (!page.ok) throw new Error(page.code);
+      collected.push(...page.value.records);
+      cursor = page.value.nextCursor ?? undefined;
+      pages += 1;
+    } while (cursor && pages < 20);
+
+    expect(collected).toHaveLength(expected.length);
+    expect(collected.map(withoutEnvelopeFields)).toEqual(expected);
+    const volume = collected.reduce((sum, row) => sum + BigInt(row.volumeStroops as string), 0n);
+    const expectedVolume = large.reduce((sum, event) => sum + (event.result === "success" ? BigInt(event.amountStroops!) : 0n), 0n);
+    expect(volume).toBe(expectedVolume);
+  });
+});
diff --git a/core/export/trends.ts b/core/export/trends.ts
new file mode 100644
index 0000000..e68c8fc
--- /dev/null
+++ b/core/export/trends.ts
@@ -0,0 +1,225 @@
+/**
+ * Historical trend export for operational and financial metrics (#191).
+ *
+ * Maintainers need to see how operations, failures and volume move over time.
+ * This module buckets raw operation events into fixed UTC windows and hands the
+ * result to the privacy-safe exporter (`exportRecords`), so a trend export
+ * inherits everything an export already guarantees: a schema version,
+ * generation metadata and expiry, redaction, pagination, the `export.generate`
+ * quota, idempotency and an audit entry for every attempt, denied ones included.
+ *
+ * What a trend row contains (schema `1.0`), one row per window and operation:
+ *
+ * - `windowStart` / `windowEnd`: the UTC window, `[start, end)`;
+ * - `op`: the operation name, e.g. `ticket.redeem`;
+ * - `total`, `successes`, `failures`, `failureRate`;
+ * - `volumeStroops`: the sum of `amountStroops` over *successful* events, as a
+ *   decimal string so large totals never lose precision.
+ *
+ * What it never contains: accounts, addresses, sessions, IPs or any other
+ * per-principal dimension. Events may carry them, but aggregation reads only
+ * the fields above, so they cannot reach an export. Trends are maintainer
+ * scope only: any other actor is refused with `export_denied`.
+ *
+ * Aggregation is pure and uses UTC, so the same events always produce the same
+ * rows regardless of the host's time zone.
+ */
+
+import {
+  EXPORT_CURRENT_SCHEMA_VERSION,
+  authorizeExport,
+  exportRecords,
+  type ExportActor,
+  type ExportEnvelope,
+  type ExportErrorCode,
+  type ExportRecord,
+  type ExportRecordSource
+} from "@/core/export/exporter";
+import { err, ok, type Result } from "@/core/result/result";
+
+export type TrendWindow = "hour" | "day" | "week";
+
+export type TrendErrorCode =
+  | "trend_invalid_window"
+  | "trend_invalid_range"
+  | "trend_range_too_large"
+  | "trend_invalid_event";
+
+export const TREND_RECORD_TYPE = "metric_trend";
+
+/** Upper bound on windows in one export, so a request cannot ask for years of minutes. */
+export const TREND_MAX_WINDOWS = 1_000;
+
+const HOUR_MS = 60 * 60 * 1_000;
+const DAY_MS = 24 * HOUR_MS;
+const WEEK_MS = 7 * DAY_MS;
+/** 1970-01-01 is a Thursday; this is the Monday before it. */
+const WEEK_ORIGIN_MS = -3 * DAY_MS;
+
+const WINDOW_MS: Record<TrendWindow, number> = { hour: HOUR_MS, day: DAY_MS, week: WEEK_MS };
+
+/** One raw operation event. Only the fields below are ever read. */
+export interface TrendEvent {
+  /** ISO-8601 string or epoch milliseconds. */
+  readonly at: string | number;
+  readonly op: string;
+  readonly result: "success" | "failure";
+  /** Integer stroops moved by a successful event. Ignored for failures. */
+  readonly amountStroops?: number;
+}
+
+export interface TrendRow extends ExportRecord {
+  windowStart: string;
+  windowEnd: string;
+  op: string;
+  total: number;
+  successes: number;
+  failures: number;
+  failureRate: number;
+  volumeStroops: string;
+}
+
+export interface TrendQuery {
+  readonly window: TrendWindow;
+  /** Inclusive start of the range. */
+  readonly from: string | number;
+  /** Exclusive end of the range. */
+  readonly to: string | number;
+}
+
+function toMillis(value: string | number): number {
+  return typeof value === "number" ? value : Date.parse(value);
+}
+
+function windowStart(at: number, window: TrendWindow): number {
+  const size = WINDOW_MS[window];
+  const origin = window === "week" ? WEEK_ORIGIN_MS : 0;
+  return origin + Math.floor((at - origin) / size) * size;
+}
+
+/**
+ * Buckets events into windows. Events outside `[from, to)` are ignored; an
+ * event that cannot be read is an error, not a silent skip, so a total never
+ * quietly under-counts.
+ */
+export function aggregateTrends(
+  events: readonly TrendEvent[],
+  query: TrendQuery
+): Result<TrendRow[], TrendErrorCode, { index?: number }> {
+  if (!Object.prototype.hasOwnProperty.call(WINDOW_MS, query.window)) {
+    return err("trend_invalid_window");
+  }
+  const from = toMillis(query.from);
+  const to = toMillis(query.to);
+  if (!Number.isFinite(from) || !Number.isFinite(to) || to <= from) {
+    return err("trend_invalid_range");
+  }
+  const windows = Math.ceil((to - windowStart(from, query.window)) / WINDOW_MS[query.window]);
+  if (windows > TREND_MAX_WINDOWS) return err("trend_range_too_large");
+
+  const buckets = new Map<string, { start: number; op: string; successes: number; failures: number; volume: bigint }>();
+
+  for (let index = 0; index < events.length; index += 1) {
+    const event = events[index];
+    const at = toMillis(event.at);
+    const validAmount =
+      event.amountStroops === undefined ||
+      (Number.isSafeInteger(event.amountStroops) && event.amountStroops >= 0);
+    if (
+      !Number.isFinite(at) ||
+      typeof event.op !== "string" ||
+      event.op.length === 0 ||
+      (event.result !== "success" && event.result !== "failure") ||
+      !validAmount
+    ) {
+      return err("trend_invalid_event", { index });
+    }
+    if (at < from || at >= to) continue;
+
+    const start = windowStart(at, query.window);
+    const key = `${start}\u0000${event.op}`;
+    let bucket = buckets.get(key);
+    if (!bucket) {
+      bucket = { start, op: event.op, successes: 0, failures: 0, volume: 0n };
+      buckets.set(key, bucket);
+    }
+    if (event.result === "success") {
+      bucket.successes += 1;
+      bucket.volume += BigInt(event.amountStroops ?? 0);
+    } else {
+      bucket.failures += 1;
+    }
+  }
+
+  const size = WINDOW_MS[query.window];
+  const rows = [...buckets.values()]
+    .sort((a, b) => a.start - b.start || (a.op < b.op ? -1 : a.op > b.op ? 1 : 0))
+    .map((bucket): TrendRow => {
+      const total = bucket.successes + bucket.failures;
+      return {
+        windowStart: new Date(bucket.start).toISOString(),
+        windowEnd: new Date(bucket.start + size).toISOString(),
+        op: bucket.op,
+        total,
+        successes: bucket.successes,
+        failures: bucket.failures,
+        failureRate: Math.round((bucket.failures / total) * 10_000) / 10_000,
+        volumeStroops: bucket.volume.toString()
+      };
+    });
+
+  return ok(rows);
+}
+
+export interface TrendExportRequest extends TrendQuery {
+  readonly actor: ExportActor;
+  readonly pageSize?: number;
+  readonly cursor?: string;
+  readonly correlationId?: string;
+  readonly idempotencyKey?: string;
+  /** Who the `export.generate` quota is charged to. */
+  readonly principal?: string;
+  readonly ttlMs?: number;
+}
+
+/**
+ * Generates a versioned, maintainer-scope trend export. The envelope is the
+ * standard export envelope, so `schemaVersion`, `generatedAt`, `generator`,
+ * `expiresAt` and the pagination fields are present on every result.
+ */
+export function exportTrends(
+  request: TrendExportRequest,
+  events: readonly TrendEvent[]
+): Result<ExportEnvelope, ExportErrorCode | TrendErrorCode, unknown> {
+  // Aggregate only for an actor who may see the result; a refused actor still
+  // goes through `exportRecords` so the denial is audited like any other.
+  const authorized = authorizeExport(request.actor, "maintainer").ok;
+  let rows: TrendRow[] = [];
+  if (authorized) {
+    const aggregated = aggregateTrends(events, request);
+    if (!aggregated.ok) return aggregated;
+    rows = aggregated.value;
+  }
+
+  const source: ExportRecordSource = {
+    recordType: TREND_RECORD_TYPE,
+    schemaVersion: EXPORT_CURRENT_SCHEMA_VERSION,
+    scope: "maintainer",
+    collect: () => rows
+  };
+
+  return exportRecords(
+    {
+      schemaVersion: EXPORT_CURRENT_SCHEMA_VERSION,
+      scope: "maintainer",
+      actor: request.actor,
+      pageSize: request.pageSize,
+      cursor: request.cursor,
+      correlationId: request.correlationId,
+      idempotencyKey: request.idempotencyKey,
+      principal: request.principal,
+      ttlMs: request.ttlMs
+    },
+    [source]
+  );
+}
diff --git a/core/invariants/__tests__/mutations.concurrency.test.ts b/core/invariants/__tests__/mutations.concurrency.test.ts
new file mode 100644
index 0000000..1d9121d
--- /dev/null
+++ b/core/invariants/__tests__/mutations.concurrency.test.ts
@@ -0,0 +1,347 @@
+import { describe, expect, it } from 'vitest';
+import { createIdempotencyStore, withIdempotency, type IdempotencyStorage } from '@/core/idempotency/idempotency';
+import { createMemoryProjectionStore } from '@/core/replay/store';
+import { err, ok } from '@/core/result/result';
+import { DomainInvariantChecker } from '../checker';
+import { commitMutation, createMemoryStateCell, type StateCell } from '../commit';
+import { issueTicket, redeemTicket, transferTicket, type TransferTicketParams } from '../mutations';
+import type { DomainState, Ticket } from '../types';
+
+/**
+ * Concurrency stress tests for the critical mutation paths (#192).
+ *
+ * Every race here is deterministic. The mutations are synchronous, so the
+ * interleaving comes from `commitMutation` awaiting its cell: every call that
+ * starts in the same tick reads the same snapshot, and the order in which they
+ * then try to commit is the order they were started. There are no timers, no
+ * sleeps and no randomness, so a failure reproduces exactly.
+ */
+
+const VALID_FROM = 1_000;
+const VALID_UNTIL = 2_000;
+const INSIDE_WINDOW = 1_500;
+
+function ticket(id: string, ownerAddress: string, overrides: Partial<Ticket> = {}): Ticket {
+  return {
+    id,
+    eventId: 'evt-1',
+    tierId: 'general',
+    ownerAddress,
+    priceStroops: 100,
+    status: 'issued',
+    issuedAt: 0,
+    validFrom: VALID_FROM,
+    validUntil: VALID_UNTIL,
+    fraudFlag: false,
+    nonce: 0,
+    metadataHash: '',
+    ...overrides
+  };
+}
+
+function stateWith(capacity: number, tickets: Ticket[] = []): DomainState {
+  return {
+    events: new Map([
+      [
+        'evt-1',
+        {
+          id: 'evt-1',
+          organizerAddress: 'G_ORGANIZER',
+          totalCapacity: capacity,
+          issuedCount: tickets.length,
+          status: 'active',
+          feeBasisPoints: 250
+        }
+      ]
+    ]),
+    tickets: new Map(tickets.map((item) => [item.id, item])),
+    transfers: [],
+    processedNonces: new Set<string>(),
+    blacklistedAddresses: new Set<string>()
+  };
+}
+
+async function current(cell: StateCell): Promise<DomainState> {
+  return (await cell.read()).state;
+}
+
+const issue = (id: string, owner = `G_BUYER_${id}`) => ({
+  id,
+  eventId: 'evt-1',
+  tierId: 'general',
+  ownerAddress: owner,
+  priceStroops: 100,
+  validFrom: VALID_FROM,
+  validUntil: VALID_UNTIL
+});
+
+function transferParams(overrides: Partial<TransferTicketParams> & Pick<TransferTicketParams, 'id' | 'toAddress' | 'nonce'>) {
+  return {
+    ticketId: 't-1',
+    fromAddress: 'G_ALICE',
+    priceStroops: 100,
+    organizerFeeStroops: 5,
+    protocolFeeStroops: 5,
+    sellerProceedsStroops: 90,
+    timestamp: INSIDE_WINDOW,
+    ...overrides
+  };
+}
+
+describe('concurrent issuance', () => {
+  it('lets every request succeed while capacity remains, with no duplicate ids', async () => {
+    const cell = createMemoryStateCell(stateWith(10));
+    const outcomes = await Promise.all(
+      ['a', 'b', 'c', 'd'].map((id) => commitMutation(cell, (state) => issueTicket(state, issue(id)), { maxAttempts: 20 }))
+    );
+
+    expect(outcomes.every((outcome) => outcome.ok)).toBe(true);
+    const state = await current(cell);
+    expect([...state.tickets.keys()].sort()).toEqual(['a', 'b', 'c', 'd']);
+    expect(state.events.get('evt-1')!.issuedCount).toBe(4);
+    expect(DomainInvariantChecker.verifyAll(state)).toEqual([]);
+  });
+
+  it('never oversells: exactly capacity requests win and the rest see the capacity invariant', async () => {
+    const capacity = 3;
+    const cell = createMemoryStateCell(stateWith(capacity));
+    const ids = Array.from({ length: 12 }, (_, index) => `t-${index}`);
+
+    const outcomes = await Promise.all(
+      ids.map((id) => commitMutation(cell, (state) => issueTicket(state, issue(id)), { maxAttempts: 20 }))
+    );
+
+    const winners = outcomes.filter((outcome) => outcome.ok);
+    const losers = outcomes.filter((outcome) => !outcome.ok);
+    expect(winners).toHaveLength(capacity);
+    for (const loser of losers) {
+      expect(loser).toMatchObject({ ok: false, reason: 'rejected', code: { invariantId: 'INV-10' } });
+    }
+
+    const state = await current(cell);
+    expect(state.tickets.size).toBe(capacity);
+    expect(state.events.get('evt-1')!.issuedCount).toBe(capacity);
+    expect(DomainInvariantChecker.verifyAll(state)).toEqual([]);
+  });
+
+  it('is repeatable: the same race produces the same winners every run', async () => {
+    const run = async () => {
+      const cell = createMemoryStateCell(stateWith(2));
+      const outcomes = await Promise.all(
+        ['x', 'y', 'z', 'w'].map((id) => commitMutation(cell, (state) => issueTicket(state, issue(id)), { maxAttempts: 20 }))
+      );
+      return outcomes.map((outcome) => outcome.ok);
+    };
+    expect(await run()).toEqual(await run());
+  });
+});
+
+describe('concurrent redemption', () => {
+  it('redeems a ticket exactly once; every other attempt is refused as terminal', async () => {
+    const cell = createMemoryStateCell(stateWith(5, [ticket('t-1', 'G_ALICE')]));
+    const attempts = Array.from({ length: 8 }, () =>
+      commitMutation(
+        cell,
+        (state) => redeemTicket(state, { ticketId: 't-1', claimerAddress: 'G_ALICE', timestamp: INSIDE_WINDOW }),
+        { maxAttempts: 20 }
+      )
+    );
+    const outcomes = await Promise.all(attempts);
+
+    expect(outcomes.filter((outcome) => outcome.ok)).toHaveLength(1);
+    for (const outcome of outcomes.filter((item) => !item.ok)) {
+      expect(outcome).toMatchObject({ reason: 'rejected', code: { invariantId: 'INV-03' } });
+    }
+    expect((await current(cell)).tickets.get('t-1')!.status).toBe('redeemed');
+  });
+
+  it('refuses a redemption that races a transfer away from the claimer', async () => {
+    const cell = createMemoryStateCell(stateWith(5, [ticket('t-1', 'G_ALICE')]));
+    const [transferred, redeemed] = await Promise.all([
+      commitMutation(cell, (state) => transferTicket(state, transferParams({ id: 'x-1', toAddress: 'G_BOB', nonce: 'n-1' }))),
+      commitMutation(cell, (state) => redeemTicket(state, { ticketId: 't-1', claimerAddress: 'G_ALICE', timestamp: INSIDE_WINDOW }))
+    ]);
+
+    expect(transferred.ok).toBe(true);
+    expect(redeemed).toMatchObject({ ok: false, reason: 'rejected' });
+    const final = (await current(cell)).tickets.get('t-1')!;
+    expect(final.ownerAddress).toBe('G_BOB');
+    expect(final.status).toBe('transferred');
+  });
+});
+
+describe('concurrent transfers', () => {
+  it('lets only one of several conflicting transfers of the same ticket land', async () => {
+    const cell = createMemoryStateCell(stateWith(5, [ticket('t-1', 'G_ALICE')]));
+    const recipients = ['G_BOB', 'G_CAROL', 'G_DAVE', 'G_ERIN'];
+    const outcomes = await Promise.all(
+      recipients.map((to, index) =>
+        commitMutation(cell, (state) => transferTicket(state, transferParams({ id: `x-${index}`, toAddress: to, nonce: `n-${index}` })), {
+          maxAttempts: 20
+        })
+      )
+    );
+
+    expect(outcomes.filter((outcome) => outcome.ok)).toHaveLength(1);
+    for (const outcome of outcomes.filter((item) => !item.ok)) {
+      expect(outcome).toMatchObject({ reason: 'rejected', code: { invariantId: 'INV-02' } });
+    }
+    const state = await current(cell);
+    expect(state.transfers).toHaveLength(1);
+    expect(state.tickets.get('t-1')!.ownerAddress).toBe(state.transfers[0].toAddress);
+    expect(state.tickets.get('t-1')!.nonce).toBe(1);
+    expect(state.processedNonces.size).toBe(1);
+  });
+
+  it('rejects a replayed nonce on a second ticket instead of recording two transfers', async () => {
+    const cell = createMemoryStateCell(stateWith(5, [ticket('t-1', 'G_ALICE'), ticket('t-2', 'G_ALICE')]));
+    const outcomes = await Promise.all([
+      commitMutation(cell, (state) => transferTicket(state, transferParams({ id: 'x-1', ticketId: 't-1', toAddress: 'G_BOB', nonce: 'shared' })), {
+        maxAttempts: 5
+      }),
+      commitMutation(cell, (state) => transferTicket(state, transferParams({ id: 'x-2', ticketId: 't-2', toAddress: 'G_BOB', nonce: 'shared' })), {
+        maxAttempts: 5
+      })
+    ]);
+
+    expect(outcomes.map((outcome) => outcome.ok)).toEqual([true, false]);
+    expect(outcomes[1]).toMatchObject({ reason: 'rejected', code: { invariantId: 'INV-09' } });
+    const state = await current(cell);
+    expect(state.transfers.map((transfer) => transfer.id)).toEqual(['x-1']);
+    expect(state.tickets.get('t-2')!.ownerAddress).toBe('G_ALICE');
+  });
+});
+
+describe('lost races and retries', () => {
+  it('reports a conflict, changes nothing, and succeeds on retry', async () => {
+    const cell = createMemoryStateCell(stateWith(5));
+    const [first, second] = await Promise.all([
+      commitMutation(cell, (state) => issueTicket(state, issue('a')), { maxAttempts: 1 }),
+      commitMutation(cell, (state) => issueTicket(state, issue('b')), { maxAttempts: 1 })
+    ]);
+
+    expect(first).toMatchObject({ ok: true, attempts: 1 });
+    expect(second).toEqual({ ok: false, reason: 'conflict', attempts: 1 });
+    expect([...(await current(cell)).tickets.keys()]).toEqual(['a']);
+
+    const retry = await commitMutation(cell, (state) => issueTicket(state, issue('b')), { maxAttempts: 1 });
+    expect(retry).toMatchObject({ ok: true, attempts: 1 });
+    expect([...(await current(cell)).tickets.keys()].sort()).toEqual(['a', 'b']);
+  });
+
+  it('counts the attempts a loser needed before it won', async () => {
+    const cell = createMemoryStateCell(stateWith(5));
+    const outcomes = await Promise.all(
+      ['a', 'b', 'c'].map((id) => commitMutation(cell, (state) => issueTicket(state, issue(id)), { maxAttempts: 5 }))
+    );
+    expect(outcomes.map((outcome) => outcome.attempts)).toEqual([1, 2, 3]);
+  });
+});
+
+describe('duplicate retries with an idempotency key', () => {
+  function memoryStorage(): IdempotencyStorage {
+    const values = new Map<string, string>();
+    return {
+      getItem: (key) => values.get(key) ?? null,
+      setItem: (key, value) => void values.set(key, value),
+      removeItem: (key) => void values.delete(key)
+    };
+  }
+
+  it('records one transfer however many times the request is submitted', async () => {
+    const cell = createMemoryStateCell(stateWith(5, [ticket('t-1', 'G_ALICE')]));
+    const store = createIdempotencyStore({ storage: memoryStorage(), now: () => 0, newId: () => 'corr' });
+    const params = transferParams({ id: 'x-1', toAddress: 'G_BOB', nonce: 'n-1' });
+    const submit = () =>
+      withIdempotency(store, {
+        key: 'transfer:t-1:attempt-1',
+        operation: 'ticket.transfer',
+        request: params,
+        run: async () => {
+          const outcome = await commitMutation(cell, (state) => transferTicket(state, params), { maxAttempts: 20 });
+          return outcome.ok ? ok(params.id) : err('transfer_rejected');
+        }
+      });
+
+    const simultaneous = await Promise.all([submit(), submit(), submit(), submit(), submit()]);
+    const executed = simultaneous.filter((result) => result.ok && !result.value.replayed);
+    const inFlight = simultaneous.filter((result) => !result.ok && result.code === 'idempotency_in_flight');
+    expect(executed).toHaveLength(1);
+    expect(inFlight).toHaveLength(4);
+
+    // A retry after the first attempt settled replays the stored outcome.
+    const retried = await submit();
+    expect(retried).toMatchObject({ ok: true, value: { value: 'x-1', replayed: true } });
+
+    const state = await current(cell);
+    expect(state.transfers).toHaveLength(1);
+    expect(state.tickets.get('t-1')!.nonce).toBe(1);
+  });
+
+  it('refuses a reused key for a different request without running it', async () => {
+    const cell = createMemoryStateCell(stateWith(5, [ticket('t-1', 'G_ALICE')]));
+    const store = createIdempotencyStore({ storage: memoryStorage(), now: () => 0, newId: () => 'corr' });
+    const run = (params: TransferTicketParams) => () =>
+      commitMutation(cell, (state) => transferTicket(state, params)).then((outcome) => (outcome.ok ? ok(params.id) : err('transfer_rejected')));
+
+    const first = transferParams({ id: 'x-1', toAddress: 'G_BOB', nonce: 'n-1' });
+    const other = transferParams({ id: 'x-2', toAddress: 'G_CAROL', nonce: 'n-2' });
+    await withIdempotency(store, { key: 'transfer:reused', operation: 'ticket.transfer', request: first, run: run(first) });
+    const reused = await withIdempotency(store, { key: 'transfer:reused', operation: 'ticket.transfer', request: other, run: run(other) });
+
+    expect(reused).toMatchObject({ ok: false, code: 'idempotency_key_conflict' });
+    expect((await current(cell)).transfers.map((transfer) => transfer.id)).toEqual(['x-1']);
+  });
+});
+
+describe('lock timeout', () => {
+  const TTL = 1_000;
+
+  it('grants a projection lock to exactly one of several simultaneous owners', () => {
+    const store = createMemoryProjectionStore();
+    const granted = ['run-1', 'run-2', 'run-3'].filter((owner) => store.acquireLock('tickets', owner, 0, TTL));
+    expect(granted).toEqual(['run-1']);
+  });
+
+  it('keeps the lock until the timeout, then lets another owner take it over', () => {
+    const store = createMemoryProjectionStore();
+    expect(store.acquireLock('tickets', 'run-1', 0, TTL)).toBe(true);
+    expect(store.acquireLock('tickets', 'run-2', TTL - 1, TTL)).toBe(false);
+    expect(store.acquireLock('tickets', 'run-2', TTL, TTL)).toBe(true);
+    // The first owner has lost it and can no longer refresh or release it.
+    expect(store.acquireLock('tickets', 'run-1', TTL + 1, TTL)).toBe(false);
+    store.releaseLock('tickets', 'run-1');
+    expect(store.acquireLock('tickets', 'run-3', TTL + 2, TTL)).toBe(false);
+  });
+
+  it('lets the live owner refresh its own lock and frees it on release', () => {
+    const store = createMemoryProjectionStore();
+    expect(store.acquireLock('tickets', 'run-1', 0, TTL)).toBe(true);
+    expect(store.acquireLock('tickets', 'run-1', TTL - 1, TTL)).toBe(true);
+    expect(store.acquireLock('tickets', 'run-2', TTL, TTL)).toBe(false);
+    store.releaseLock('tickets', 'run-1');
+    expect(store.acquireLock('tickets', 'run-2', TTL, TTL)).toBe(true);
+  });
+
+  it('expires an abandoned in-flight idempotency claim instead of replaying it', () => {
+    let now = 0;
+    const values = new Map<string, string>();
+    const store = createIdempotencyStore({
+      ttlMs: TTL,
+      now: () => now,
+      newId: () => 'corr',
+      storage: {
+        getItem: (key) => values.get(key) ?? null,
+        setItem: (key, value) => void values.set(key, value),
+        removeItem: (key) => void values.delete(key)
+      }
+    });
+    const request = { key: 'claim:abandoned-1', operation: 'ticket.transfer', request: { id: 1 } };
+
+    expect(store.begin(request).ok).toBe(true);
+    now = TTL - 1;
+    expect(store.begin(request)).toMatchObject({ ok: false, code: 'idempotency_in_flight' });
+    now = TTL;
+    expect(store.begin(request)).toMatchObject({ ok: false, code: 'idempotency_key_expired' });
+  });
+});
diff --git a/core/invariants/commit.ts b/core/invariants/commit.ts
new file mode 100644
index 0000000..4efddda
--- /dev/null
+++ b/core/invariants/commit.ts
@@ -0,0 +1,91 @@
+/**
+ * Optimistic commit for domain mutations (#192).
+ *
+ * `issueTicket`, `transferTicket` and `redeemTicket` are pure: they validate a
+ * snapshot and return the next one. They are only safe under concurrency if
+ * the snapshot they validated is still the current one when the result is
+ * stored. Two requests that both read the same snapshot would otherwise both
+ * pass the capacity, ownership or nonce check and the second write would
+ * overwrite the first.
+ *
+ * `commitMutation` closes that gap with compare-and-set on a version:
+ *
+ * 1. read the state and its version;
+ * 2. run the mutation on that snapshot;
+ * 3. store the result only if the version is unchanged;
+ * 4. otherwise re-read and run the mutation again on the new snapshot.
+ *
+ * A request that lost a race is therefore judged against the winner's result,
+ * so it fails with the real invariant (capacity reached, no longer the owner,
+ * already redeemed, nonce replayed) rather than overwriting it. After
+ * `maxAttempts` lost races it gives up with `reason: "conflict"` and has changed
+ * nothing, so the caller may retry with the same idempotency key.
+ *
+ * The cell is an interface because a real store (database row, contract
+ * ledger entry) is asynchronous; `createMemoryStateCell` is the reference
+ * implementation for tests.
+ */
+
+import type { DomainState } from './types';
+
+export interface VersionedState {
+  readonly version: number;
+  readonly state: DomainState;
+}
+
+export interface StateCell {
+  read(): Promise<VersionedState>;
+  /**
+   * Installs `next` and bumps the version only if the cell is still at
+   * `expectedVersion`. Must be atomic. Returns false when it lost the race.
+   */
+  compareAndSet(expectedVersion: number, next: DomainState): Promise<boolean>;
+}
+
+export const DEFAULT_COMMIT_ATTEMPTS = 10;
+
+export function createMemoryStateCell(initial: DomainState): StateCell {
+  let current: VersionedState = { version: 0, state: initial };
+  return {
+    async read() {
+      return current;
+    },
+    async compareAndSet(expectedVersion, next) {
+      if (current.version !== expectedVersion) return false;
+      current = { version: expectedVersion + 1, state: next };
+      return true;
+    }
+  };
+}
+
+type MutationResult<E> =
+  | { readonly ok: true; readonly value: DomainState }
+  | { readonly ok: false; readonly code: E };
+
+export type CommitOutcome<E> =
+  | { readonly ok: true; readonly value: DomainState; readonly attempts: number }
+  /** The mutation refused the latest snapshot. Nothing was stored. */
+  | { readonly ok: false; readonly reason: 'rejected'; readonly code: E; readonly attempts: number }
+  /** Every attempt lost a race. Nothing was stored; retrying is safe. */
+  | { readonly ok: false; readonly reason: 'conflict'; readonly attempts: number };
+
+export interface CommitOptions {
+  readonly maxAttempts?: number;
+}
+
+export async function commitMutation<E>(
+  cell: StateCell,
+  mutate: (state: DomainState) => MutationResult<E>,
+  options: CommitOptions = {}
+): Promise<CommitOutcome<E>> {
+  const maxAttempts = Math.max(1, options.maxAttempts ?? DEFAULT_COMMIT_ATTEMPTS);
+  for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
+    const { version, state } = await cell.read();
+    const result = mutate(state);
+    if (!result.ok) return { ok: false, reason: 'rejected', code: result.code, attempts: attempt };
+    if (await cell.compareAndSet(version, result.value)) {
+      return { ok: true, value: result.value, attempts: attempt };
+    }
+  }
+  return { ok: false, reason: 'conflict', attempts: maxAttempts };
+}
diff --git a/core/invariants/index.ts b/core/invariants/index.ts
index 3dfe77f..292c2d2 100644
--- a/core/invariants/index.ts
+++ b/core/invariants/index.ts
@@ -1,3 +1,4 @@
 export * from './types';
 export * from './checker';
 export * from './mutations';
+export * from './commit';
diff --git a/core/migration/__tests__/backfill.test.ts b/core/migration/__tests__/backfill.test.ts
new file mode 100644
index 0000000..a29687a
--- /dev/null
+++ b/core/migration/__tests__/backfill.test.ts
@@ -0,0 +1,333 @@
+import { describe, expect, it } from "vitest";
+import {
+  createMemoryBackfillTarget,
+  createMemoryCheckpointStore,
+  rollbackBackfill,
+  runBackfill,
+  type BackfillDefinition,
+  type BackfillRecord,
+  type BackfillTarget
+} from "../backfill";
+import { err, ok } from "@/core/result/result";
+
+interface Row extends BackfillRecord {
+  readonly id: string;
+  readonly schema: number;
+  readonly label?: string;
+}
+
+const NOW = Date.parse("2026-10-18T00:00:00.000Z");
+const clock = () => NOW;
+
+/** r-01..r-10 on the old schema, plus rows the transform must skip, leave alone, or choke on. */
+function fixture(): Row[] {
+  const rows: Row[] = Array.from({ length: 10 }, (_, index) => ({
+    id: `r-${String(index + 1).padStart(2, "0")}`,
+    schema: 1,
+    label: `  Label ${index + 1} `
+  }));
+  rows.push({ id: "r-11", schema: 1 }); // no label: skipped
+  rows.push({ id: "r-12", schema: 2, label: "done" }); // already migrated: unchanged
+  return rows;
+}
+
+const definition: BackfillDefinition<Row> = {
+  id: "label-normalise",
+  version: "1",
+  transform: (row) => {
+    if (row.label === undefined) return err("missing_label");
+    if (row.id === "r-poison") throw new Error("boom for G-secret-token");
+    return ok({ ...row, schema: 2, label: row.label.trim().toLowerCase() });
+  }
+};
+
+function counting(target: BackfillTarget<Row>) {
+  const writes: string[] = [];
+  let reads = 0;
+  const wrapped: BackfillTarget<Row> = {
+    readBatch: (afterId, limit) => {
+      reads += 1;
+      return target.readBatch(afterId, limit);
+    },
+    write: async (record) => {
+      writes.push(record.id);
+      return target.write(record);
+    }
+  };
+  return { wrapped, writes, reads: () => reads };
+}
+
+const maintainer = { role: "maintainer" } as const;
+
+describe("dry run", () => {
+  it("previews changes without writing or checkpointing", async () => {
+    const target = createMemoryBackfillTarget(fixture());
+    const store = createMemoryCheckpointStore<Row>();
+
+    const result = await runBackfill(definition, target, { ...maintainer, batchSize: 4, sampleSize: 3, checkpoints: store });
+    if (!result.ok) throw new Error(result.code);
+    const report = result.value;
+
+    expect(report).toMatchObject({
+      mode: "dry_run",
+      status: "completed",
+      scanned: 12,
+      updated: 10,
+      unchanged: 1,
+      skippedCount: 1,
+      failedCount: 0,
+      checkpointRevision: null
+    });
+    expect(report.skipped).toEqual([{ id: "r-11", reason: "missing_label" }]);
+    expect(report.diffs).toHaveLength(3);
+    expect(report.diffs[0]).toMatchObject({
+      recordId: "r-01",
+      action: "update",
+      before: { schema: 1, label: "  Label 1 " },
+      after: { schema: 2, label: "label 1" }
+    });
+    expect(target.records()).toEqual(fixture());
+    expect(await store.load(definition.id)).toBeUndefined();
+  });
+
+  it("redacts secret-shaped values in the diff sample", async () => {
+    const secret = "S" + "A".repeat(55);
+    const target = createMemoryBackfillTarget<Row>([{ id: "r-01", schema: 1, label: secret }]);
+    const result = await runBackfill(definition, target, maintainer);
+    if (!result.ok) throw new Error(result.code);
+    expect(JSON.stringify(result.value.diffs)).not.toContain(secret);
+  });
+});
+
+describe("option and permission checks", () => {
+  it("refuses roles without admin:recovery", async () => {
+    for (const role of ["viewer", "service"] as const) {
+      const result = await runBackfill(definition, createMemoryBackfillTarget(fixture()), { role });
+      expect(result).toMatchObject({ ok: false, code: "backfill_unauthorized" });
+    }
+  });
+
+  it("refuses to apply without a checkpoint store and rejects bad numbers", async () => {
+    const target = createMemoryBackfillTarget(fixture());
+    expect(await runBackfill(definition, target, { ...maintainer, apply: true })).toMatchObject({
+      ok: false,
+      code: "backfill_invalid_options"
+    });
+    for (const bad of [{ batchSize: 0 }, { batchSize: 1.5 }, { maxFailures: -1 }, { sampleSize: -1 }]) {
+      expect(await runBackfill(definition, target, { ...maintainer, ...bad })).toMatchObject({
+        ok: false,
+        code: "backfill_invalid_options"
+      });
+    }
+    expect(target.records()).toEqual(fixture());
+  });
+});
+
+describe("apply", () => {
+  it("writes each changed record once, checkpoints per batch, and is idempotent", async () => {
+    const target = createMemoryBackfillTarget(fixture());
+    const spy = counting(target);
+    const store = createMemoryCheckpointStore<Row>();
+
+    const first = await runBackfill(definition, spy.wrapped, { ...maintainer, apply: true, checkpoints: store, batchSize: 5, now: clock });
+    if (!first.ok) throw new Error(first.code);
+    expect(first.value).toMatchObject({ mode: "apply", status: "completed", scanned: 12, updated: 10, unchanged: 1, skippedCount: 1 });
+    expect(first.value.checkpointRevision).toBe(3); // 12 rows in batches of 5, 5, 2
+    expect(spy.writes).toHaveLength(10);
+    expect(target.records().filter((row) => row.schema === 2)).toHaveLength(11);
+    expect(target.records().find((row) => row.id === "r-01")).toEqual({ id: "r-01", schema: 2, label: "label 1" });
+
+    const saved = await store.load(definition.id);
+    expect(saved).toMatchObject({ status: "completed", lastId: "r-12", updated: 10, updatedAt: "2026-10-18T00:00:00.000Z" });
+    expect(saved!.journal).toHaveLength(10);
+
+    // A completed backfill is not silently run again...
+    expect(await runBackfill(definition, spy.wrapped, { ...maintainer, apply: true, checkpoints: store })).toMatchObject({
+      ok: false,
+      code: "backfill_already_completed"
+    });
+    // ...but a deliberate restart finds nothing left to write.
+    const again = await runBackfill(definition, spy.wrapped, { ...maintainer, apply: true, checkpoints: store, restart: true });
+    if (!again.ok) throw new Error(again.code);
+    expect(again.value).toMatchObject({ updated: 0, unchanged: 11, skippedCount: 1 });
+    expect(spy.writes).toHaveLength(10);
+    // The restart keeps the first run's before-images, so rollback still works.
+    expect((await store.load(definition.id))!.journal).toHaveLength(10);
+  });
+
+  it("resumes after the last committed batch when the source drops mid-run", async () => {
+    const target = createMemoryBackfillTarget(fixture());
+    const spy = counting(target);
+    const store = createMemoryCheckpointStore<Row>();
+    let calls = 0;
+    const flaky: BackfillTarget<Row> = {
+      readBatch: (afterId, limit) => {
+        calls += 1;
+        if (calls === 3) return Promise.reject(new Error("connection reset"));
+        return spy.wrapped.readBatch(afterId, limit);
+      },
+      write: spy.wrapped.write
+    };
+
+    const interrupted = await runBackfill(definition, flaky, { ...maintainer, apply: true, checkpoints: store, batchSize: 4 });
+    expect(interrupted).toMatchObject({ ok: false, code: "backfill_source_unavailable" });
+    const midway = await store.load(definition.id);
+    expect(midway).toMatchObject({ status: "halted", lastId: "r-08", scanned: 8 });
+    expect(spy.writes).toEqual(["r-01", "r-02", "r-03", "r-04", "r-05", "r-06", "r-07", "r-08"]);
+
+    const resumed = await runBackfill(definition, flaky, { ...maintainer, apply: true, checkpoints: store, batchSize: 4 });
+    if (!resumed.ok) throw new Error(resumed.code);
+    expect(resumed.value).toMatchObject({ status: "completed", resumedAfter: "r-08", scanned: 12, updated: 10 });
+    // Each record was written exactly once across both runs.
+    expect(spy.writes).toEqual(["r-01", "r-02", "r-03", "r-04", "r-05", "r-06", "r-07", "r-08", "r-09", "r-10"]);
+
+    const complete = createMemoryBackfillTarget(fixture());
+    await runBackfill(definition, complete, { ...maintainer, apply: true, checkpoints: createMemoryCheckpointStore<Row>() });
+    expect(target.records()).toEqual(complete.records());
+  });
+
+  it("refuses to resume a checkpoint from a different transform version", async () => {
+    const target = createMemoryBackfillTarget(fixture());
+    const store = createMemoryCheckpointStore<Row>();
+    let calls = 0;
+    const flaky: BackfillTarget<Row> = {
+      readBatch: (afterId, limit) => (++calls === 2 ? Promise.reject(new Error("down")) : target.readBatch(afterId, limit)),
+      write: target.write
+    };
+    await runBackfill(definition, flaky, { ...maintainer, apply: true, checkpoints: store, batchSize: 4 });
+
+    const changed = { ...definition, version: "2" };
+    expect(await runBackfill(changed, target, { ...maintainer, apply: true, checkpoints: store })).toMatchObject({
+      ok: false,
+      code: "backfill_version_mismatch"
+    });
+  });
+
+  it("stops the second of two simultaneous operators instead of letting both advance", async () => {
+    const target = createMemoryBackfillTarget(fixture());
+    const store = createMemoryCheckpointStore<Row>();
+    const run = () => runBackfill(definition, target, { ...maintainer, apply: true, checkpoints: store });
+
+    const outcomes = await Promise.all([run(), run()]);
+    expect(outcomes.map((outcome) => (outcome.ok ? "ok" : outcome.code))).toEqual(["ok", "backfill_checkpoint_conflict"]);
+    expect(await store.load(definition.id)).toMatchObject({ revision: 1, status: "completed" });
+  });
+});
+
+describe("skipped and failed records", () => {
+  const rows = (): Row[] => [
+    { id: "a-1", schema: 1, label: "One" },
+    { id: "a-2", schema: 1 },
+    { id: "r-poison", schema: 1, label: "Bad" },
+    { id: "z-1", schema: 1, label: "Last" }
+  ];
+
+  it("reports skipped and failed records and keeps going below the failure limit", async () => {
+    const target = createMemoryBackfillTarget(rows());
+    const result = await runBackfill(definition, target, { ...maintainer, apply: true, checkpoints: createMemoryCheckpointStore<Row>() });
+    if (!result.ok) throw new Error(result.code);
+
+    expect(result.value).toMatchObject({ status: "completed", scanned: 4, updated: 2, skippedCount: 1, failedCount: 1 });
+    expect(result.value.skipped).toEqual([{ id: "a-2", reason: "missing_label" }]);
+    expect(result.value.failed).toHaveLength(1);
+    expect(result.value.failed[0].id).toBe("r-poison");
+    expect(target.records().find((row) => row.id === "r-poison")!.schema).toBe(1);
+  });
+
+  it("halts after the batch that exceeds maxFailures and resumes past it", async () => {
+    const target = createMemoryBackfillTarget(rows());
+    const store = createMemoryCheckpointStore<Row>();
+    const options = { ...maintainer, apply: true, checkpoints: store, batchSize: 3, maxFailures: 0 } as const;
+
+    const halted = await runBackfill(definition, target, options);
+    if (!halted.ok) throw new Error(halted.code);
+    expect(halted.value).toMatchObject({ status: "halted", lastId: "r-poison", failedCount: 1 });
+    expect(await store.load(definition.id)).toMatchObject({ status: "halted" });
+
+    const resumed = await runBackfill(definition, target, options);
+    if (!resumed.ok) throw new Error(resumed.code);
+    expect(resumed.value).toMatchObject({ status: "completed", resumedAfter: "r-poison", scanned: 4 });
+  });
+
+  it("counts a failed write as failed and does not journal it", async () => {
+    const inner = createMemoryBackfillTarget(fixture());
+    const target: BackfillTarget<Row> = {
+      readBatch: inner.readBatch,
+      write: (record) => (record.id === "r-03" ? Promise.reject(new Error("disk full")) : inner.write(record))
+    };
+    const store = createMemoryCheckpointStore<Row>();
+    const result = await runBackfill(definition, target, { ...maintainer, apply: true, checkpoints: store });
+    if (!result.ok) throw new Error(result.code);
+
+    expect(result.value).toMatchObject({ updated: 9, failedCount: 1 });
+    expect(result.value.failed).toEqual([{ id: "r-03", error: "disk full" }]);
+    expect((await store.load(definition.id))!.journal.map((entry) => entry.id)).not.toContain("r-03");
+    expect(inner.records().find((row) => row.id === "r-03")!.schema).toBe(1);
+  });
+
+  it("stops when the source returns records out of order", async () => {
+    const store = createMemoryCheckpointStore<Row>();
+    const target: BackfillTarget<Row> = {
+      readBatch: async () => [
+        { id: "b", schema: 1, label: "B" },
+        { id: "a", schema: 1, label: "A" }
+      ],
+      write: async () => undefined
+    };
+    expect(await runBackfill(definition, target, { ...maintainer, apply: true, checkpoints: store })).toMatchObject({
+      ok: false,
+      code: "backfill_source_unavailable"
+    });
+  });
+});
+
+describe("rollback", () => {
+  it("restores every journaled before-image and marks the checkpoint rolled back", async () => {
+    const target = createMemoryBackfillTarget(fixture());
+    const store = createMemoryCheckpointStore<Row>();
+    await runBackfill(definition, target, { ...maintainer, apply: true, checkpoints: store });
+    expect(target.records()).not.toEqual(fixture());
+
+    const result = await rollbackBackfill(definition.id, target, store, { role: "maintainer" });
+    if (!result.ok) throw new Error(result.code);
+    expect(result.value).toEqual({ backfillId: definition.id, restored: 10, failed: [] });
+    expect(target.records()).toEqual(fixture());
+    expect(await store.load(definition.id)).toMatchObject({ status: "rolled_back", journal: [] });
+  });
+
+  it("keeps unrestored entries journaled so rollback can be re-run", async () => {
+    const inner = createMemoryBackfillTarget(fixture());
+    const store = createMemoryCheckpointStore<Row>();
+    await runBackfill(definition, inner, { ...maintainer, apply: true, checkpoints: store });
+
+    let broken = true;
+    const flaky: BackfillTarget<Row> = {
+      readBatch: inner.readBatch,
+      write: (record) => (broken && record.id === "r-05" ? Promise.reject(new Error("locked")) : inner.write(record))
+    };
+    const first = await rollbackBackfill(definition.id, flaky, store, { role: "maintainer" });
+    if (!first.ok) throw new Error(first.code);
+    expect(first.value.restored).toBe(9);
+    expect(first.value.failed).toEqual([{ id: "r-05", error: "locked" }]);
+    expect(await store.load(definition.id)).toMatchObject({ status: "completed" });
+
+    broken = false;
+    const second = await rollbackBackfill(definition.id, flaky, store, { role: "maintainer" });
+    if (!second.ok) throw new Error(second.code);
+    expect(second.value).toMatchObject({ restored: 1, failed: [] });
+    expect(inner.records()).toEqual(fixture());
+  });
+
+  it("needs a checkpoint and the right role", async () => {
+    const target = createMemoryBackfillTarget(fixture());
+    const store = createMemoryCheckpointStore<Row>();
+    expect(await rollbackBackfill("missing", target, store, { role: "maintainer" })).toMatchObject({
+      ok: false,
+      code: "backfill_no_checkpoint"
+    });
+    expect(await rollbackBackfill("missing", target, store, { role: "viewer" })).toMatchObject({
+      ok: false,
+      code: "backfill_unauthorized"
+    });
+  });
+});
diff --git a/core/migration/backfill.ts b/core/migration/backfill.ts
new file mode 100644
index 0000000..8dc5019
--- /dev/null
+++ b/core/migration/backfill.ts
@@ -0,0 +1,450 @@
+/**
+ * Operator-safe backfill (#201).
+ *
+ * A backfill rewrites existing records to a new shape. This module makes that
+ * safe to run by hand against live data:
+ *
+ * - **dry-run first.** The default mode scans everything, writes nothing, and
+ *   returns the counts plus a bounded sample of before/after diffs. Applying
+ *   is an explicit choice (`apply: true`);
+ * - **checkpointed.** Applying saves a checkpoint after every batch. A run that
+ *   stops (crash, halt, Ctrl-C) resumes after the last committed record;
+ * - **idempotent.** A record the transform leaves unchanged is counted as
+ *   `unchanged` and not written, so re-running a backfill, or resuming one
+ *   whose last batch half-landed, never rewrites finished records;
+ * - **reported, not hidden.** Records the transform declines are `skipped` with
+ *   a reason; records that throw, or whose write fails, are `failed`. The run
+ *   halts at the end of the batch in which more than `maxFailures` records
+ *   have failed since it started;
+ * - **single operator.** Checkpoints carry a revision and are saved with
+ *   compare-and-set, so two operators running the same backfill cannot both
+ *   advance it: the loser stops with `backfill_checkpoint_conflict`;
+ * - **reversible.** Every applied update journals the record's before-image in
+ *   the checkpoint, and `rollbackBackfill` writes those images back.
+ *
+ * Records are visited in ascending `id` order through an injected target, and
+ * the clock and stores are injected too, so a test can assert an exact run.
+ * See `docs/BACKFILL.md` for operating procedure and failure modes.
+ */
+
+import { hasPermission, type Role } from "@/core/auth/types";
+import { canonicalize } from "@/core/idempotency/idempotency";
+import { err, ok, type Result } from "@/core/result/result";
+import { redact } from "@/core/telemetry/telemetry";
+import type { MigrationDiff } from "./types";
+
+export type BackfillErrorCode =
+  | "backfill_invalid_options"
+  | "backfill_unauthorized"
+  | "backfill_version_mismatch"
+  | "backfill_already_completed"
+  | "backfill_checkpoint_conflict"
+  | "backfill_source_unavailable"
+  | "backfill_no_checkpoint";
+
+export interface BackfillRecord {
+  readonly id: string;
+  readonly [key: string]: unknown;
+}
+
+export interface BackfillDefinition<T extends BackfillRecord> {
+  /** Stable name; also the checkpoint key. */
+  readonly id: string;
+  /** Bump when the transform changes. A checkpoint from another version is refused. */
+  readonly version: string;
+  /**
+   * Returns the new record, or `err(reason)` to skip a record this backfill
+   * does not apply to. Must be idempotent: transforming its own output
+   * returns the same record. Must keep `id`.
+   */
+  readonly transform: (record: T) => Result<T, string>;
+}
+
+/** Where records live. `readBatch` returns records with `id` > `afterId`, ascending. */
+export interface BackfillTarget<T extends BackfillRecord> {
+  readBatch(afterId: string | null, limit: number): Promise<readonly T[]>;
+  write(record: T): Promise<void>;
+}
+
+export type BackfillStatus = "running" | "halted" | "completed" | "rolled_back";
+
+export interface BackfillSkip {
+  readonly id: string;
+  readonly reason: string;
+}
+
+export interface BackfillFailure {
+  readonly id: string;
+  readonly error: string;
+}
+
+export interface BackfillCheckpoint<T extends BackfillRecord> {
+  readonly backfillId: string;
+  readonly version: string;
+  /** Incremented on every save; the compare-and-set token. */
+  readonly revision: number;
+  readonly status: BackfillStatus;
+  /** Last record whose batch was committed. Resume continues after it. */
+  readonly lastId: string | null;
+  readonly scanned: number;
+  readonly updated: number;
+  readonly unchanged: number;
+  readonly skippedCount: number;
+  readonly failedCount: number;
+  /** First `BACKFILL_MAX_REPORTED` entries; the counts above are exact. */
+  readonly skipped: readonly BackfillSkip[];
+  readonly failed: readonly BackfillFailure[];
+  /** Before-images of every applied update, in write order. */
+  readonly journal: readonly { readonly id: string; readonly before: T }[];
+  readonly updatedAt: string;
+}
+
+export interface CheckpointStore<T extends BackfillRecord> {
+  load(backfillId: string): Promise<BackfillCheckpoint<T> | undefined>;
+  /**
+   * Stores `checkpoint` only if the stored revision equals `expectedRevision`
+   * (`null` meaning none stored). Must be atomic. Returns false on a mismatch.
+   */
+  save(checkpoint: BackfillCheckpoint<T>, expectedRevision: number | null): Promise<boolean>;
+}
+
+export const BACKFILL_DEFAULT_BATCH_SIZE = 100;
+export const BACKFILL_MAX_BATCH_SIZE = 10_000;
+export const BACKFILL_DEFAULT_MAX_FAILURES = 10;
+export const BACKFILL_DEFAULT_SAMPLE_SIZE = 20;
+export const BACKFILL_MAX_REPORTED = 1_000;
+
+export interface BackfillOptions<T extends BackfillRecord> {
+  /** Backfills need `admin:recovery`. */
+  readonly role: Role;
+  /** Writes only when true. The default is a dry run. */
+  readonly apply?: boolean;
+  /** Required to apply. */
+  readonly checkpoints?: CheckpointStore<T>;
+  /** Apply from the start even if a checkpoint exists (or is completed). */
+  readonly restart?: boolean;
+  readonly batchSize?: number;
+  /** The run halts after the batch in which more than this many records have failed. */
+  readonly maxFailures?: number;
+  /** How many before/after diffs the report carries. */
+  readonly sampleSize?: number;
+  readonly now?: () => number;
+}
+
+export interface BackfillReport {
+  readonly backfillId: string;
+  readonly version: string;
+  readonly mode: "dry_run" | "apply";
+  readonly status: "completed" | "halted";
+  /** Id the run resumed after, or `null` when it started from the beginning. */
+  readonly resumedAfter: string | null;
+  readonly lastId: string | null;
+  readonly scanned: number;
+  /** Records changed (dry run: records that would change). */
+  readonly updated: number;
+  readonly unchanged: number;
+  readonly skippedCount: number;
+  readonly failedCount: number;
+  readonly skipped: readonly BackfillSkip[];
+  readonly failed: readonly BackfillFailure[];
+  /** Bounded sample of the changes, redacted for display. */
+  readonly diffs: readonly MigrationDiff[];
+  readonly checkpointRevision: number | null;
+}
+
+function failureMessage(error: unknown): string {
+  return String(redact(error instanceof Error ? error.message : String(error)));
+}
+
+function validate<T extends BackfillRecord>(options: BackfillOptions<T>): string | undefined {
+  const batchSize = options.batchSize ?? BACKFILL_DEFAULT_BATCH_SIZE;
+  if (!Number.isInteger(batchSize) || batchSize < 1 || batchSize > BACKFILL_MAX_BATCH_SIZE) {
+    return `batchSize must be an integer from 1 to ${BACKFILL_MAX_BATCH_SIZE}`;
+  }
+  const maxFailures = options.maxFailures ?? BACKFILL_DEFAULT_MAX_FAILURES;
+  if (!Number.isInteger(maxFailures) || maxFailures < 0) return "maxFailures must be a non-negative integer";
+  const sampleSize = options.sampleSize ?? BACKFILL_DEFAULT_SAMPLE_SIZE;
+  if (!Number.isInteger(sampleSize) || sampleSize < 0) return "sampleSize must be a non-negative integer";
+  if (options.apply && !options.checkpoints) return "applying requires a checkpoint store";
+  return undefined;
+}
+
+function fresh<T extends BackfillRecord>(definition: BackfillDefinition<T>, at: string): BackfillCheckpoint<T> {
+  return {
+    backfillId: definition.id,
+    version: definition.version,
+    revision: 0,
+    status: "running",
+    lastId: null,
+    scanned: 0,
+    updated: 0,
+    unchanged: 0,
+    skippedCount: 0,
+    failedCount: 0,
+    skipped: [],
+    failed: [],
+    journal: [],
+    updatedAt: at
+  };
+}
+
+/**
+ * Runs a backfill. A dry run (the default) never writes and ignores any
+ * checkpoint; it always scans from the start so the diff is complete.
+ */
+export async function runBackfill<T extends BackfillRecord>(
+  definition: BackfillDefinition<T>,
+  target: BackfillTarget<T>,
+  options: BackfillOptions<T>
+): Promise<Result<BackfillReport, BackfillErrorCode, string>> {
+  const invalid = validate(options);
+  if (invalid) return err("backfill_invalid_options", invalid);
+  if (!hasPermission(options.role, "admin:recovery")) return err("backfill_unauthorized");
+
+  const apply = options.apply === true;
+  const store = options.checkpoints;
+  const now = options.now ?? (() => Date.now());
+  const batchSize = options.batchSize ?? BACKFILL_DEFAULT_BATCH_SIZE;
+  const maxFailures = options.maxFailures ?? BACKFILL_DEFAULT_MAX_FAILURES;
+  const sampleSize = options.sampleSize ?? BACKFILL_DEFAULT_SAMPLE_SIZE;
+  const at = () => new Date(now()).toISOString();
+
+  let state = fresh(definition, at());
+  let storedRevision: number | null = null;
+  let resumedAfter: string | null = null;
+
+  if (apply && store) {
+    const existing = await store.load(definition.id);
+    if (existing) {
+      storedRevision = existing.revision;
+      if (!options.restart) {
+        if (existing.version !== definition.version) return err("backfill_version_mismatch");
+        if (existing.status === "completed" || existing.status === "rolled_back") {
+          return err("backfill_already_completed");
+        }
+        state = existing;
+        resumedAfter = existing.lastId;
+      } else {
+        // A restart rescans from the start but keeps the earlier before-images,
+        // so the original data stays recoverable.
+        state = { ...state, journal: existing.journal };
+      }
+    }
+  }
+
+  const diffs: MigrationDiff[] = [];
+  const skipped: BackfillSkip[] = [...state.skipped];
+  const failed: BackfillFailure[] = [...state.failed];
+  const journal = [...state.journal];
+  let { scanned, updated, unchanged, skippedCount, failedCount } = state;
+  /** Failures in this invocation only, so resuming a halted run is not halted again at once. */
+  let runFailures = 0;
+  let lastId = state.lastId;
+  let status: "completed" | "halted" = "completed";
+  let revision = storedRevision;
+
+  const snapshot = (running: BackfillStatus): BackfillCheckpoint<T> => ({
+    backfillId: definition.id,
+    version: definition.version,
+    revision: (revision ?? 0) + 1,
+    status: running,
+    lastId,
+    scanned,
+    updated,
+    unchanged,
+    skippedCount,
+    failedCount,
+    skipped,
+    failed,
+    journal,
+    updatedAt: at()
+  });
+
+  const persist = async (running: BackfillStatus): Promise<boolean> => {
+    if (!apply || !store) return true;
+    const next = snapshot(running);
+    if (!(await store.save(next, revision))) return false;
+    revision = next.revision;
+    return true;
+  };
+
+  const recordFailure = (id: string, error: string) => {
+    failedCount += 1;
+    runFailures += 1;
+    if (failed.length < BACKFILL_MAX_REPORTED) failed.push({ id, error });
+  };
+
+  for (;;) {
+    let batch: readonly T[];
+    try {
+      batch = await target.readBatch(lastId, batchSize);
+    } catch {
+      // Progress so far is already saved; the operator resumes once the source is back.
+      await persist("halted");
+      return err("backfill_source_unavailable");
+    }
+
+    let previous = lastId;
+    for (const record of batch) {
+      if (previous !== null && !(record.id > previous)) {
+        await persist("halted");
+        return err("backfill_source_unavailable", `records out of order at ${record.id}`);
+      }
+      previous = record.id;
+      scanned += 1;
+
+      let outcome: Result<T, string>;
+      try {
+        outcome = definition.transform(record);
+      } catch (error) {
+        recordFailure(record.id, failureMessage(error));
+        continue;
+      }
+      if (!outcome.ok) {
+        skippedCount += 1;
+        if (skipped.length < BACKFILL_MAX_REPORTED) skipped.push({ id: record.id, reason: outcome.code });
+        continue;
+      }
+      if (outcome.value.id !== record.id) {
+        recordFailure(record.id, "transform changed the record id");
+        continue;
+      }
+      if (canonicalize(outcome.value) === canonicalize(record)) {
+        unchanged += 1;
+        continue;
+      }
+
+      if (apply) {
+        try {
+          await target.write(outcome.value);
+        } catch (error) {
+          recordFailure(record.id, failureMessage(error));
+          continue;
+        }
+        journal.push({ id: record.id, before: record });
+      }
+      updated += 1;
+      if (diffs.length < sampleSize) {
+        diffs.push({
+          recordId: record.id,
+          action: "update",
+          before: redact(record) as Record<string, unknown>,
+          after: redact(outcome.value) as Record<string, unknown>
+        });
+      }
+    }
+
+    if (batch.length > 0) lastId = batch[batch.length - 1].id;
+    const halted = runFailures > maxFailures;
+    const done = batch.length < batchSize;
+    status = halted ? "halted" : "completed";
+    if (!(await persist(halted ? "halted" : done ? "completed" : "running"))) {
+      return err("backfill_checkpoint_conflict");
+    }
+    if (halted || done) break;
+  }
+
+  return ok({
+    backfillId: definition.id,
+    version: definition.version,
+    mode: apply ? "apply" : "dry_run",
+    status,
+    resumedAfter,
+    lastId,
+    scanned,
+    updated,
+    unchanged,
+    skippedCount,
+    failedCount,
+    skipped,
+    failed,
+    diffs,
+    checkpointRevision: apply ? revision : null
+  });
+}
+
+export interface RollbackReport {
+  readonly backfillId: string;
+  readonly restored: number;
+  /** Records that could not be restored; they stay journaled for another attempt. */
+  readonly failed: readonly BackfillFailure[];
+}
+
+/**
+ * Writes the journaled before-images back, newest first. Restored entries are
+ * removed from the journal, so a partly failed rollback can simply be re-run.
+ * It overwrites the whole record, including changes made since the backfill.
+ */
+export async function rollbackBackfill<T extends BackfillRecord>(
+  backfillId: string,
+  target: BackfillTarget<T>,
+  checkpoints: CheckpointStore<T>,
+  options: { readonly role: Role; readonly now?: () => number }
+): Promise<Result<RollbackReport, BackfillErrorCode>> {
+  if (!hasPermission(options.role, "admin:recovery")) return err("backfill_unauthorized");
+  const checkpoint = await checkpoints.load(backfillId);
+  if (!checkpoint) return err("backfill_no_checkpoint");
+
+  const remaining = [...checkpoint.journal];
+  const failures: BackfillFailure[] = [];
+  let restored = 0;
+  for (let index = checkpoint.journal.length - 1; index >= 0; index -= 1) {
+    const entry = checkpoint.journal[index];
+    try {
+      await target.write(entry.before);
+      remaining.splice(index, 1);
+      restored += 1;
+    } catch (error) {
+      failures.push({ id: entry.id, error: failureMessage(error) });
+    }
+  }
+
+  const now = options.now ?? (() => Date.now());
+  const saved = await checkpoints.save(
+    {
+      ...checkpoint,
+      revision: checkpoint.revision + 1,
+      status: remaining.length === 0 ? "rolled_back" : checkpoint.status,
+      journal: remaining,
+      updatedAt: new Date(now()).toISOString()
+    },
+    checkpoint.revision
+  );
+  if (!saved) return err("backfill_checkpoint_conflict");
+  return ok({ backfillId, restored, failed: failures });
+}
+
+/** In-memory checkpoint store: the reference implementation for tests. */
+export function createMemoryCheckpointStore<T extends BackfillRecord>(): CheckpointStore<T> {
+  const stored = new Map<string, BackfillCheckpoint<T>>();
+  return {
+    async load(backfillId) {
+      return stored.get(backfillId);
+    },
+    async save(checkpoint, expectedRevision) {
+      const current = stored.get(checkpoint.backfillId);
+      if ((current?.revision ?? null) !== expectedRevision) return false;
+      stored.set(checkpoint.backfillId, checkpoint);
+      return true;
+    }
+  };
+}
+
+/** In-memory target over an id-keyed record set: the reference implementation for tests. */
+export function createMemoryBackfillTarget<T extends BackfillRecord>(
+  initial: readonly T[]
+): BackfillTarget<T> & { readonly records: () => readonly T[] } {
+  const rows = new Map(initial.map((record) => [record.id, record]));
+  const sorted = () => [...rows.values()].sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0));
+  return {
+    async readBatch(afterId, limit) {
+      return sorted()
+        .filter((record) => afterId === null || record.id > afterId)
+        .slice(0, limit);
+    },
+    async write(record) {
+      rows.set(record.id, record);
+    },
+    records: sorted
+  };
+}
diff --git a/core/migration/index.ts b/core/migration/index.ts
index cded058..18cc0bb 100644
--- a/core/migration/index.ts
+++ b/core/migration/index.ts
@@ -1,3 +1,4 @@
 export * from './types';
 export * from './postChecks';
 export * from './runner';
+export * from './backfill';
diff --git a/docs/BACKFILL.md b/docs/BACKFILL.md
new file mode 100644
index 0000000..3a5655a
--- /dev/null
+++ b/docs/BACKFILL.md
@@ -0,0 +1,115 @@
+# Operator-Safe Backfill
+
+`core/migration/backfill.ts` rewrites existing records to a new shape in a way
+an operator can preview, interrupt, resume and undo. `scripts/backfill.mjs` is
+the command line front end for a JSON file of records.
+
+## Running a backfill
+
+A backfill is a definition plus a target:
+
+```js
+// my-backfill.mjs
+export default {
+  id: "label-normalise", // also the checkpoint key
+  version: "1", // bump whenever transform changes
+  transform(record) {
+    if (record.label === undefined) return { ok: false, code: "missing_label" }; // skipped
+    return { ok: true, value: { ...record, schema: 2, label: record.label.trim().toLowerCase() } };
+  }
+};
+```
+
+`transform` must keep `id` and must be idempotent: transforming its own output
+returns the same record.
+
+```sh
+# 1. Preview. Writes nothing. Prints counts and a sample of before/after diffs.
+UTIX_USER_ROLE=maintainer npm run backfill -- --input records.json --definition ./my-backfill.mjs
+
+# 2. Apply. Checkpoints after every batch; resumes automatically if run again.
+UTIX_USER_ROLE=maintainer npm run backfill -- --input records.json --definition ./my-backfill.mjs --apply
+
+# 3. Undo.
+UTIX_USER_ROLE=maintainer npm run backfill -- --input records.json --definition ./my-backfill.mjs --rollback
+```
+
+Other flags: `--checkpoint <file>` (default `<input>.checkpoint.json`),
+`--restart`, `--batch-size N` (default 100), `--max-failures N` (default 10),
+`--sample N` (diffs shown, default 20), `--json`. The command needs the
+`admin:recovery` permission (`maintainer`). It exits 1 on a refusal, a halt, or
+any failed record.
+
+In code, call `runBackfill(definition, target, { role, apply, checkpoints })` with
+a `BackfillTarget` (`readBatch`, `write`) and a `CheckpointStore`
+(`load`, compare-and-set `save`). `createMemoryBackfillTarget` and
+`createMemoryCheckpointStore` are the reference implementations.
+
+## What the report says
+
+| Field | Meaning |
+| --- | --- |
+| `updated` | Records changed (dry run: records that would change) |
+| `unchanged` | Transform returned the record as it was; not written |
+| `skippedCount` / `skipped` | Records the transform declined, with its reason |
+| `failedCount` / `failed` | Transform threw, changed the id, or the write failed |
+| `diffs` | Bounded sample of changes; values are redacted for display |
+| `resumedAfter` | The id a resumed run continued after |
+
+`skipped` and `failed` list the first 1 000 entries; the counts are exact.
+
+## Safety properties
+
+- **Dry run is the default** and always scans from the start, ignoring any
+  checkpoint, so the diff is complete.
+- **Resume.** A checkpoint is saved after every batch. Running the same command
+  again continues after the last committed record. A checkpoint from another
+  `version` is refused (`backfill_version_mismatch`).
+- **Idempotent.** Already-converted records come back `unchanged` and are not
+  rewritten, so re-running is cheap and safe.
+- **One operator at a time.** Saves are compare-and-set on a revision. The
+  second of two simultaneous runs stops with `backfill_checkpoint_conflict`.
+- **Failure budget.** The run halts at the end of the batch in which more than
+  `--max-failures` records failed since it started. Skips do not count.
+
+## Edge cases and failure modes
+
+| Situation | What happens | What to do |
+| --- | --- | --- |
+| Source read fails mid-run | `backfill_source_unavailable`; checkpoint kept as `halted` | Fix the source, run the same command |
+| Source returns ids out of order | `backfill_source_unavailable`; run stops | Fix the source ordering (ascending by `id`) |
+| Crash between a write and its checkpoint | Up to one batch was written but not journaled; on resume those records read as `unchanged` | Restore those from the backup (see below) if you need to undo |
+| Transform bug found after a partial run | Records already written keep the old transform's output | Fix, bump `version`, run with `--restart`; earlier before-images are kept |
+| Run already completed | `backfill_already_completed` | `--restart` to rescan (it will write nothing if nothing changed) |
+| Failed records | Reported, left unchanged, not retried on resume | Fix the cause, then `--restart`; unchanged records are skipped quickly |
+| Two operators | Second gets `backfill_checkpoint_conflict` | Let the first finish, then resume |
+| Wrong role | `backfill_unauthorized` | Run as a maintainer |
+
+## Rollback and recovery
+
+1. **Journal.** Every applied update records the record's before-image in the
+   checkpoint. `--rollback` (or `rollbackBackfill`) writes them back, newest
+   first, removes each restored entry, and marks the checkpoint `rolled_back`.
+   Entries that fail to restore stay journaled, so rollback can be re-run.
+2. **Overwrite warning.** Rollback restores whole records. Changes made to
+   those records after the backfill are lost; check before rolling back.
+3. **Backup.** Before the first write the CLI copies the input to
+   `<input>.bak`. It is the recovery path for the one case the journal cannot
+   cover (a crash mid-batch) and for a corrupt checkpoint.
+4. **Checkpoint file.** It holds full before-images, so protect it like the
+   data. If it is lost, restore from `.bak`.
+5. **Rolled back or completed backfills** are not re-run silently; use
+   `--restart`. A restart keeps the earlier journal.
+
+## Compatibility
+
+Additive only. No existing API, schema or user-facing behavior changes, and the
+CLI never touches data unless `--apply` or `--rollback` is passed. No migration
+is required.
+
+## Tests
+
+`core/migration/__tests__/backfill.test.ts` covers the dry run (no writes, no
+checkpoint, bounded and redacted diffs), apply and idempotent re-run, resume
+after a source failure, version mismatch, concurrent operators, skipped and
+failed records, the failure limit, out-of-order sources and rollback.
diff --git a/docs/CONCURRENCY.md b/docs/CONCURRENCY.md
new file mode 100644
index 0000000..8afd456
--- /dev/null
+++ b/docs/CONCURRENCY.md
@@ -0,0 +1,72 @@
+# Concurrency on Critical Mutation Paths
+
+Ticket issuance, transfer and redemption are pure functions in
+`core/invariants/mutations.ts`: each validates one snapshot and returns the
+next. Concurrency safety is therefore decided where the result is stored, not
+inside the functions.
+
+## Vulnerable paths
+
+| Path | Race if two requests read the same snapshot | Guard |
+| --- | --- | --- |
+| `issueTicket` | Both pass the capacity check, event is oversold (`INV-10`) | `commitMutation` |
+| `transferTicket` | Both pass the ownership check, one transfer is lost or the ticket changes hands twice (`INV-02`) | `commitMutation` |
+| `transferTicket` nonce | Same nonce accepted twice (`INV-09`) | `commitMutation` |
+| `redeemTicket` | Same ticket redeemed twice (`INV-03`) | `commitMutation` |
+| Retried client request | The same request runs twice | idempotency key (`core/idempotency`) |
+| Projection rebuild | Two rebuilds write one projection | per-projection lock with timeout (`core/replay`) |
+
+## Strategy
+
+`core/invariants/commit.ts` implements optimistic concurrency:
+
+1. `StateCell.read()` returns the state and its version.
+2. The mutation runs on that snapshot.
+3. `StateCell.compareAndSet(version, next)` stores the result only if the
+   version is unchanged.
+4. If another request committed first, the loser re-reads and runs the
+   mutation again on the new snapshot.
+
+A loser is judged against the winner's result, so it fails with the real
+invariant (capacity reached, not the owner, already redeemed, nonce replayed)
+instead of overwriting the winner. After `maxAttempts` (default 10) lost races
+it returns `{ ok: false, reason: "conflict" }`. Nothing was stored, so the
+client may retry.
+
+`compareAndSet` must be atomic in the real store (a conditional update on a
+version column, or a ledger entry sequence). `createMemoryStateCell` is the
+reference implementation used by the tests.
+
+## Idempotency and locks
+
+- **Duplicate retries.** Wrap the commit in `withIdempotency`. While the first
+  attempt is running, duplicates get `idempotency_in_flight`; after it settles,
+  they replay the stored outcome. A key reused for a different request gets
+  `idempotency_key_conflict`.
+- **Lock timeout.** `ProjectionStore.acquireLock(name, owner, now, ttlMs)` grants
+  the lock to one owner; the owner may refresh it; once `ttlMs` has passed
+  another owner may take it over, and the old owner can no longer release it.
+  An idempotency claim that outlives its TTL is not replayed: the caller gets
+  `idempotency_key_expired` and must use a new key. In both cases a crashed
+  worker cannot block an operation forever.
+
+## Tests
+
+`core/invariants/__tests__/mutations.concurrency.test.ts` covers, with no
+timers and no randomness:
+
+- concurrent success (many issuances within capacity, no duplicate ids);
+- conflicting requests (oversell, double redeem, double transfer, replayed
+  nonce, redeem racing a transfer);
+- lost races (`conflict` result, retry succeeds, attempt counts);
+- duplicate retries (one transfer however many submissions, replay after
+  settle, key reuse refused);
+- lock timeout (single grant, takeover after the TTL, refresh and release by the
+  owner only, expired idempotency claim).
+
+Run with `npx vitest run core/invariants/__tests__/mutations.concurrency.test.ts`.
+
+## Limits
+
+The in-memory cell serializes within one process. Cross-process safety depends
+on the real store providing an atomic `compareAndSet`.
diff --git a/docs/EXPORTS.md b/docs/EXPORTS.md
index 3fb902e..e513ff6 100644
--- a/docs/EXPORTS.md
+++ b/docs/EXPORTS.md
@@ -98,7 +98,9 @@ materialised; without it the stream falls back to `collect`.
 
 `npm run test` covers large exports (5 000 records, paged without loss), empty
 exports, denied (`export_denied`) and unsupported-version exports, redaction,
-and retention expiry — see `core/export/__tests__/exporter.test.ts`.
+and retention expiry — see `core/export/__tests__/exporter.test.ts`. Trend
+aggregation, date ranges, empty data, permission denial and a 5 000 event
+paged export are covered in `core/export/__tests__/trends.test.ts`.
 
 Each export emits `export.generate` and `export.authorize` telemetry; see
 [docs/TELEMETRY.md](./TELEMETRY.md).
diff --git a/docs/REJECTION_EXPLANATIONS.md b/docs/REJECTION_EXPLANATIONS.md
new file mode 100644
index 0000000..d8f125c
--- /dev/null
+++ b/docs/REJECTION_EXPLANATIONS.md
@@ -0,0 +1,71 @@
+# Rejection Explanations
+
+`core/errors/explanations.ts` turns any rejected operation into one
+user-safe `RejectionExplanation`, so every client shows the same reason, the
+same wording and the same recovery steps.
+
+```ts
+import { explainRejection, renderRejection } from "@/core/errors/explanations";
+
+const result = redeemTicket(state, params);
+if (!result.ok) {
+  const explanation = explainRejection(result, { correlationId });
+  // { reasonCode: "outside_validity_window", category: "policy", title, message,
+  //   recovery: [...], retryable: false, status: 409, reference }
+  show(renderRejection(explanation));
+}
+```
+
+`explainRejection` accepts a failed `Result`, a bare stable code, a domain
+`InvariantViolation`, an `UnauthorizedError`, or any thrown value.
+
+## Reason codes
+
+Reason codes are stable. Clients switch on `reasonCode`; a code is never
+renamed or removed, a new one is added instead. Each belongs to one category:
+
+| Category | Meaning | Reason codes |
+| --- | --- | --- |
+| `validation` | The request itself is wrong | `invalid_input`, `invalid_address`, `invalid_amount`, `missing_field`, `unsupported_version`, `invalid_request_key`, `request_key_reused` |
+| `policy` | A rule refuses a valid request | `capacity_reached`, `ticket_locked`, `account_restricted`, `outside_validity_window`, `event_cancelled`, `rate_limited` |
+| `permission` | The caller may not do this | `unauthenticated`, `permission_denied`, `not_owner` |
+| `stale_state` | The world moved since the caller looked | `not_found`, `already_redeemed`, `already_transferred`, `invalid_state_transition`, `duplicate_request`, `request_in_progress`, `request_expired` |
+| `unexpected` | Not an expected rejection | `network_unavailable`, `timed_out`, `unexpected_failure` |
+
+Every reason carries at least one ordered `recovery` step, a `retryable` flag
+and the HTTP `status` an API response should use.
+
+## Mapping
+
+Low-level inputs map to reasons through two tables, exported so tests and
+docs cannot drift from them:
+
+- `REJECTION_SOURCE_CODES`: stable error codes (`core/errors/error-codes.ts`)
+  and `Result` codes from idempotency, quota, export and replay.
+- `REJECTION_INVARIANTS`: domain invariants `INV-01` to `INV-11`
+  (`core/invariants`). `INV-02` covers both "not the owner" and "ticket not
+  found", so its copy is deliberately neutral.
+
+`unmappedErrorCodes()` lists any stable error code without a reason; the test
+suite requires it to be empty, so adding an error code without mapping it
+fails CI.
+
+## Safety
+
+- Only the fixed copy in `REJECTION_REASONS` is returned. A raw error
+  message, entity id, limit or stack trace never appears in an explanation.
+- Unrecognised input (unknown code, `null`, an `Error`) becomes
+  `unexpected_failure`, so callers always receive an explanation.
+- `reference` echoes the correlation id you pass in so support can find the
+  server-side record; it is omitted when none is given.
+
+## Adding a rejection
+
+1. Add the reason to `REJECTION_REASONS` with a category, copy, at least one
+   recovery step, `retryable` and `status`.
+2. Map the low-level code or invariant in `REJECTION_SOURCE_CODES` or
+   `REJECTION_INVARIANTS`.
+3. Add a case to `core/errors/__tests__/explanations.test.ts`.
+
+No migration is needed: the module is additive and does not change existing
+error codes, `Result` shapes or API responses.
diff --git a/package.json b/package.json
index a16bc8d..0334499 100644
--- a/package.json
+++ b/package.json
@@ -46,7 +46,8 @@
     "bundle:check": "node scripts/bundle-budget.mjs",
     "diagnostics:contributor": "node --loader ./scripts/loader.mjs scripts/contributor-diagnostics.mjs",
     "doctor": "node --loader ./scripts/loader.mjs scripts/contributor-diagnostics.mjs",
-    "verify:issues-264-275": "node --loader ./scripts/loader.mjs scripts/verify-issues-264-275.mjs"
+    "verify:issues-264-275": "node --loader ./scripts/loader.mjs scripts/verify-issues-264-275.mjs",
+    "backfill": "node --loader ./scripts/loader.mjs scripts/backfill.mjs"
   },
   "dependencies": {
     "@stellar/stellar-sdk": "^13.3.0",
diff --git a/scripts/backfill.mjs b/scripts/backfill.mjs
new file mode 100644
index 0000000..2ed5e3b
--- /dev/null
+++ b/scripts/backfill.mjs
@@ -0,0 +1,151 @@
+#!/usr/bin/env node
+/**
+ * Operator-safe backfill CLI (#201).
+ *
+ * Runs a backfill definition (`core/migration/backfill.ts`) over a JSON file of
+ * records. The default is a dry run: nothing is written and the diff is
+ * printed. `--apply` writes, saving a checkpoint after every batch so an
+ * interrupted run resumes where it stopped.
+ *
+ * Usage:
+ *   npm run backfill -- --input records.json --definition ./my-backfill.mjs
+ *   npm run backfill -- --input records.json --definition ./my-backfill.mjs --apply --checkpoint backfill.checkpoint.json
+ *   npm run backfill -- --input records.json --definition ./my-backfill.mjs --rollback --checkpoint backfill.checkpoint.json
+ *
+ * The definition module default-exports `{ id, version, transform }`.
+ * `transform(record)` returns `{ ok: true, value }` or `{ ok: false, code }`.
+ *
+ * Flags: --apply, --restart, --rollback, --batch-size N, --max-failures N,
+ * --sample N, --json. The role comes from UTIX_USER_ROLE and needs
+ * `admin:recovery` (maintainer). Exits 1 on an error, a halt, or failed records.
+ *
+ * Safety: before the first write `<input>.bak` is created from the untouched
+ * input. Data is flushed to disk before the checkpoint that covers it, so a
+ * crash can leave data ahead of the checkpoint (re-applied as "unchanged"), never
+ * behind it.
+ */
+
+import { copyFile, readFile, rename, writeFile } from "node:fs/promises";
+import { existsSync } from "node:fs";
+import path from "node:path";
+import { pathToFileURL } from "node:url";
+
+const { runBackfill, rollbackBackfill } = await import("../core/migration/backfill.ts");
+const { getAuthContext } = await import("../core/auth/middleware.ts");
+
+function parseArgs(argv) {
+  const args = { flags: new Set(), values: {} };
+  for (let index = 0; index < argv.length; index += 1) {
+    const token = argv[index];
+    if (!token.startsWith("--")) throw new Error(`unexpected argument: ${token}`);
+    const name = token.slice(2);
+    if (["apply", "restart", "rollback", "json"].includes(name)) args.flags.add(name);
+    else if (["input", "definition", "checkpoint", "batch-size", "max-failures", "sample"].includes(name)) {
+      const value = argv[(index += 1)];
+      if (value === undefined) throw new Error(`--${name} needs a value`);
+      args.values[name] = value;
+    } else throw new Error(`unknown flag: --${name}`);
+  }
+  return args;
+}
+
+function number(value) {
+  return value === undefined ? undefined : Number(value);
+}
+
+async function writeJsonAtomic(file, value) {
+  const temp = `${file}.tmp`;
+  await writeFile(temp, `${JSON.stringify(value, null, 2)}\n`);
+  await rename(temp, file);
+}
+
+async function main() {
+  const { flags, values } = parseArgs(process.argv.slice(2));
+  if (!values.input || !values.definition) {
+    throw new Error("usage: backfill --input <records.json> --definition <module.mjs> [--apply] [--checkpoint <file>]");
+  }
+  const inputPath = path.resolve(values.input);
+  const apply = flags.has("apply");
+  const rollback = flags.has("rollback");
+  const checkpointPath = path.resolve(values.checkpoint ?? `${inputPath}.checkpoint.json`);
+
+  const definition = (await import(pathToFileURL(path.resolve(values.definition)).href)).default;
+  const records = JSON.parse(await readFile(inputPath, "utf8"));
+  if (!Array.isArray(records)) throw new Error("input must be a JSON array of records");
+  const byId = new Map(records.map((record) => [record.id, record]));
+  if (byId.size !== records.length) throw new Error("input contains duplicate record ids");
+
+  let dirty = false;
+  const target = {
+    async readBatch(afterId, limit) {
+      return [...byId.values()]
+        .filter((record) => afterId === null || record.id > afterId)
+        .sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0))
+        .slice(0, limit);
+    },
+    async write(record) {
+      if (!dirty && !existsSync(`${inputPath}.bak`)) await copyFile(inputPath, `${inputPath}.bak`);
+      byId.set(record.id, record);
+      dirty = true;
+    }
+  };
+
+  const checkpoints = {
+    async load() {
+      return existsSync(checkpointPath) ? JSON.parse(await readFile(checkpointPath, "utf8")) : undefined;
+    },
+    async save(checkpoint, expectedRevision) {
+      const current = existsSync(checkpointPath) ? JSON.parse(await readFile(checkpointPath, "utf8")) : undefined;
+      if ((current?.revision ?? null) !== expectedRevision) return false;
+      // Data first, then the checkpoint that covers it.
+      if (dirty) {
+        await writeJsonAtomic(inputPath, [...byId.values()].sort((a, b) => (a.id < b.id ? -1 : 1)));
+        dirty = false;
+      }
+      await writeJsonAtomic(checkpointPath, checkpoint);
+      return true;
+    }
+  };
+
+  const { role } = getAuthContext();
+
+  if (rollback) {
+    const result = await rollbackBackfill(definition.id, target, checkpoints, { role });
+    if (!result.ok) throw new Error(`rollback refused: ${result.code}`);
+    console.log(JSON.stringify(result.value, null, 2));
+    process.exitCode = result.value.failed.length > 0 ? 1 : 0;
+    return;
+  }
+
+  const result = await runBackfill(definition, target, {
+    role,
+    apply,
+    restart: flags.has("restart"),
+    checkpoints,
+    batchSize: number(values["batch-size"]),
+    maxFailures: number(values["max-failures"]),
+    sampleSize: number(values.sample)
+  });
+  if (!result.ok) throw new Error(`backfill refused: ${result.code}${result.detail ? ` (${result.detail})` : ""}`);
+
+  const report = result.value;
+  if (flags.has("json")) {
+    console.log(JSON.stringify(report, null, 2));
+  } else {
+    console.log(`${report.mode === "apply" ? "APPLIED" : "DRY RUN (nothing written; pass --apply)"}: ${report.backfillId}@${report.version}`);
+    if (report.resumedAfter) console.log(`resumed after ${report.resumedAfter}`);
+    console.log(
+      `status=${report.status} scanned=${report.scanned} updated=${report.updated} unchanged=${report.unchanged} ` +
+        `skipped=${report.skippedCount} failed=${report.failedCount}`
+    );
+    for (const diff of report.diffs) console.log(`  ${diff.recordId}: ${JSON.stringify(diff.before)} -> ${JSON.stringify(diff.after)}`);
+    for (const skip of report.skipped) console.log(`  skipped ${skip.id}: ${skip.reason}`);
+    for (const failure of report.failed) console.log(`  FAILED ${failure.id}: ${failure.error}`);
+  }
+  process.exitCode = report.status === "halted" || report.failedCount > 0 ? 1 : 0;
+}
+
+main().catch((error) => {
+  console.error(error instanceof Error ? error.message : error);
+  process.exitCode = 1;
+});