diff --git a/.gitignore b/.gitignore
index a1dbe4b..3504f38 100644
--- a/.gitignore
+++ b/.gitignore
@@ -6,3 +6,4 @@ examples/*/dist/
 *.tsbuildinfo
 coverage/
 .DS_Store
+fix.md
diff --git a/src/account/batchOperations.ts b/src/account/batchOperations.ts
new file mode 100644
index 0000000..bae1cb5
--- /dev/null
+++ b/src/account/batchOperations.ts
@@ -0,0 +1,673 @@
+/**
+ * Batch account operations (fix.md).
+ *
+ * A unified batch execution model for account-level operations — trustline
+ * creation, payments, and key rotation — with bounded concurrency, operation
+ * grouping, progress tracking, and partial-failure handling.
+ *
+ * Design principles
+ * -----------------
+ * - A single idempotency-aware executor (`runBatchOperations`) drives every
+ *   batch. It is transport-agnostic: the caller supplies a per-operation
+ *   `runner`.
+ * - Each operation is planned with a stable, caller-supplied `id` and its
+ *   result is tracked independently. A failure in one operation never
+ *   invalidates unrelated operations.
+ * - Retry is never blind: an operation that has already succeeded (or is in
+ *   `previouslyCompletedIds`) is never re-run, so `maxRetries` cannot
+ *   duplicate a completed/possibly-submitted operation.
+ * - Progress is exposed incrementally via `onProgress` and summarized in the
+ *   final report.
+ */
+
+import { ok, err, SorokitErrorCode } from "../shared/response";
+import type { SorokitResult } from "../shared/response";
+import {
+  buildPaymentTransaction,
+  buildTrustlineTransaction,
+} from "../transaction/buildTransaction";
+import type { TrustlineParams } from "../transaction/types";
+import { rotateAccountKey } from "./keyRotation";
+import type { RotateAccountKeyParams } from "./keyRotation";
+import type { ResolvedNetworkConfig } from "../shared/types";
+import { sleep } from "../shared/utils";
+
+// ─── Types ────────────────────────────────────────────────────────────────────
+
+/** Lifecycle status of a single batch operation. */
+export type BatchOperationStatus =
+  | "pending"
+  | "running"
+  | "success"
+  | "failed"
+  | "skipped"
+  | "retried";
+
+/** Metadata for an operation in a batch plan. */
+export interface BatchOperation<T = unknown> {
+  /** Stable, caller-supplied identifier used for idempotency. */
+  id: string;
+  /**
+   * Runnable that actually performs the operation.
+   * Return `ok(data)` on success, or an error result on failure.
+   */
+  runner: BatchRunner<T>;
+  /** Optional input associated with the operation (for reporting). */
+  input?: unknown;
+}
+
+export type BatchRunner<T> = (
+  operationId: string,
+) => Promise<SorokitResult<T>>;
+
+/** Individual tracked result of one batch operation. */
+export interface BatchOperationResult<T> {
+  id: string;
+  status: BatchOperationStatus;
+  /** Available when the operation succeeded. */
+  data?: T;
+  /** Human-readable error message when the operation ultimately failed. */
+  errorMessage?: string;
+  errorCode?: string;
+  /** Number of execution attempts (1 = initial, >1 = retried). */
+  attempts: number;
+  /** Number of retries performed. */
+  retries: number;
+  /** True when the operation was skipped (already completed / explicit skip). */
+  skipped?: boolean;
+}
+
+/** Incremental snapshot of batch progress. */
+export interface BatchProgress {
+  total: number;
+  planned: number;
+  running: number;
+  succeeded: number;
+  failed: number;
+  skipped: number;
+  retried: number;
+  completed: number;
+  /** Epoch-ms timestamp when the batch started. */
+  startedAt: number;
+  /** Elapsed ms at the time the snapshot was produced. */
+  elapsedMs: number;
+}
+
+/** Configuration for a batch execution. */
+export interface BatchExecutorConfig {
+  /**
+   * Maximum number of operations to schedule into a single grouping wave.
+   * Larger values allow more operations in flight at once; smaller values
+   * give finer-grained progress callbacks.
+   */
+  batchSize?: number;
+  /** Maximum number of operations running concurrently (default: 5). */
+  concurrency?: number;
+  /**
+   * Maximum retry attempts for retryable failures (default: 2).
+   * Successful or previously-completed operations are never retried.
+   */
+  maxRetries?: number;
+  /** Base delay (ms) before the first retry; backoff is exponential (default: 200). */
+  retryDelayMs?: number;
+  /**
+   * Idempotency set. Operations whose id appears here are assumed already
+   * completed and are skipped rather than executed.
+   */
+  previouslyCompletedIds?: Iterable<string>;
+  /** Called whenever progress changes with a live snapshot. */
+  onProgress?: (progress: BatchProgress) => void;
+  /**
+   * Optional predicate classifying an error as retryable. Defaults to
+   * network/timeout/service-unavailable errors.
+   */
+  isRetryable?: (error: SorokitResult<unknown>) => boolean;
+  /** Inject determinism for clock in tests. */
+  now?: () => number;
+  /** Time source for the retry sleep (overridable in tests). */
+  delay?: (ms: number) => Promise<void>;
+}
+
+/** Final report of a batch execution. */
+export interface BatchExecutionReport<T> {
+  results: BatchOperationResult<T>[];
+  summary: {
+    total: number;
+    succeeded: number;
+    failed: number;
+    skipped: number;
+    retried: number;
+    completed: number;
+    /** True when every operation ended in success. */
+    allSucceeded: boolean;
+  };
+  /** Final progress snapshot. */
+  progress: BatchProgress;
+  /** Epoch-ms timestamp when the batch finished. */
+  finishedAt: number;
+}
+
+// ─── Defaults & helpers ───────────────────────────────────────────────────────
+
+const DEFAULT_CONCURRENCY = 5;
+const DEFAULT_MAX_RETRIES = 2;
+const DEFAULT_RETRY_DELAY_MS = 200;
+const MAX_RETRY_DELAY_MS = 10_000;
+
+const RETRYABLE_CODES: ReadonlySet<string> = new Set([
+  SorokitErrorCode.NETWORK_ERROR,
+  SorokitErrorCode.OPERATION_TIMEOUT,
+  SorokitErrorCode.SERVICE_UNAVAILABLE,
+]);
+
+function defaultIsRetryable(error: SorokitResult<unknown>): boolean {
+  if (error.status !== "error" || !error.error) return false;
+  return RETRYABLE_CODES.has(error.error.code);
+}
+
+// ─── Progress accounting ──────────────────────────────────────────────────────
+
+function createProgressState(total: number, startedAt: number) {
+  return {
+    planned: 0,
+    running: 0,
+    succeeded: 0,
+    failed: 0,
+    skipped: 0,
+    retried: 0,
+    completed: 0,
+  };
+}
+
+function snapshotProgress(
+  startedAt: number,
+  counter: ReturnType<typeof createProgressState>,
+  total: number,
+  now: () => number,
+): BatchProgress {
+  return {
+    total,
+    planned: counter.planned,
+    running: counter.running,
+    succeeded: counter.succeeded,
+    failed: counter.failed,
+    skipped: counter.skipped,
+    retried: counter.retried,
+    completed: counter.completed,
+    startedAt,
+    elapsedMs: now() - startedAt,
+  };
+}
+
+// ─── Core executor ────────────────────────────────────────────────────────────
+
+export interface QueueItem<T> {
+  operation: BatchOperation<T>;
+}
+
+/**
+ * Run a batch of operations with bounded concurrency, operation grouping,
+ * progress tracking, and partial-failure handling.
+ *
+ * Behavioral guarantees:
+ *  - Each operation runs exactly once if it succeeds; retryable failures are
+ *    retried up to `maxRetries` times before being marked `failed`.
+ *  - Operations listed in `previouslyCompletedIds` are marked `skipped`
+ *    (never executed again) to make re-running a batch idempotent.
+ *  - A failure in one operation is isolated and does not affect other
+ *    operations.
+ *
+ * @param operations - The operations to plan and execute.
+ * @param userConfig - Batch size, concurrency, retry, and progress options.
+ * @returns A {@link BatchExecutionReport} with per-operation outcomes.
+ */
+export async function runBatchOperations<T>(
+  operations: BatchOperation<T>[],
+  userConfig: BatchExecutorConfig = {},
+): Promise<BatchExecutionReport<T>> {
+  const config: Required<
+    Pick<
+      BatchExecutorConfig,
+      "batchSize" | "concurrency" | "maxRetries" | "retryDelayMs" | "now"
+    >
+  > = {
+    batchSize: userConfig.batchSize ?? operations.length,
+    concurrency: Math.max(1, userConfig.concurrency ?? DEFAULT_CONCURRENCY),
+    maxRetries: Math.max(0, userConfig.maxRetries ?? DEFAULT_MAX_RETRIES),
+    retryDelayMs: Math.max(0, userConfig.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS),
+    now: userConfig.now ?? Date.now,
+  };
+  const delay = userConfig.delay ?? sleep;
+  const isRetryable =
+    userConfig.isRetryable ??
+    ((error: SorokitResult<unknown>) => defaultIsRetryable(error));
+  const completedIds = new Set<string>(userConfig.previouslyCompletedIds ?? []);
+
+  const startedAt = config.now();
+  const counter = createProgressState(operations.length, startedAt);
+  const resultsArray: BatchOperationResult<T>[] = [];
+
+  const emit = () => {
+    userConfig.onProgress?.(
+      snapshotProgress(startedAt, counter, operations.length, config.now),
+    );
+  };
+
+  const markSkipped = (id: string) => {
+    counter.skipped += 1;
+    counter.completed += 1;
+    resultsArray.push({
+      id,
+      status: "skipped",
+      attempts: 0,
+      retries: 0,
+      skipped: true,
+    });
+  };
+
+  // Plan operations in groups of `batchSize` so big batches can be chunked
+  // with incremental progress callbacks after each group is scheduled.
+  const plan = [...operations];
+  const queue: QueueItem<T>[] = [];
+
+  // Skip previously completed operations up front (idempotency), emitting
+  // progress in `batchSize` chunks so large batches report incrementally.
+  let plannedCount = 0;
+  for (const op of plan) {
+    if (completedIds.has(op.id)) {
+      markSkipped(op.id);
+      if (plannedCount % config.batchSize === 0) emit();
+      continue;
+    }
+    counter.planned += 1;
+    plannedCount += 1;
+    queue.push({ operation: op });
+    if (plannedCount % config.batchSize === 0) emit();
+  }
+  emit();
+
+  // ── Bounded-concurrency worker pool ──────────────────────────────────────
+  let nextIndex = 0;
+  const active = new Set<number>();
+
+  const processOne = async (index: number): Promise<void> => {
+    if (index >= queue.length) return;
+    const item = queue[index];
+    if (!item) return;
+    if (completedIds.has(item.operation.id)) {
+      markSkipped(item.operation.id);
+      return;
+    }
+
+    counter.running += 1;
+    emit();
+
+    const { operation } = item;
+
+    const outcome = await runWithRetry(operation, {
+      maxRetries: config.maxRetries,
+      retryDelayMs: config.retryDelayMs,
+      isRetryable,
+      delay,
+    });
+
+    counter.running -= 1;
+    if (outcome.status === "success") {
+      counter.succeeded += 1;
+      counter.completed += 1;
+      // Mark completed so re-runs / later scans skip it (idempotency).
+      completedIds.add(operation.id);
+    } else {
+      counter.failed += 1;
+      counter.completed += 1;
+    }
+    counter.retried += outcome.retries;
+
+    resultsArray.push({
+      id: operation.id,
+      status: outcome.status,
+      ...(outcome.data !== undefined ? { data: outcome.data } : {}),
+      ...(outcome.errorMessage !== undefined
+        ? { errorMessage: outcome.errorMessage }
+        : {}),
+      ...(outcome.errorCode !== undefined ? { errorCode: outcome.errorCode } : {}),
+      attempts: outcome.attempts,
+      retries: outcome.retries,
+    });
+    emit();
+
+    // Pull the next item as soon as a slot frees up (bounded concurrency).
+    const next = nextIndex++;
+    if (next < queue.length) {
+      active.add(next);
+      // Fire-and-forget each worker; the pool tracks completion via a final await.
+      void processOne(next).finally(() => active.delete(next));
+    } else {
+      active.delete(index);
+    }
+  };
+
+  // Seed up to `concurrency` workers.
+  const initialWorkers = Math.min(config.concurrency, queue.length);
+  for (let i = 0; i < initialWorkers; i++) {
+    const index = nextIndex++;
+    active.add(index);
+    void processOne(index).finally(() => active.delete(index));
+  }
+
+  // Wait for every worker to finish.
+  await waitForWorkers(() => active.size === 0);
+
+  const completed = counter.completed;
+  const summary = {
+    total: operations.length,
+    succeeded: counter.succeeded,
+    failed: counter.failed,
+    skipped: counter.skipped,
+    retried: counter.retried,
+    completed,
+    allSucceeded:
+      counter.failed === 0 && counter.skipped === 0 && counter.succeeded === totalOf(operations),
+  };
+
+  return {
+    results: resultsArray,
+    summary,
+    progress: snapshotProgress(
+      startedAt,
+      counter,
+      operations.length,
+      config.now,
+    ),
+    finishedAt: config.now(),
+  };
+}
+
+function totalOf<T>(operations: BatchOperation<T>[]): number {
+  return operations.length;
+}
+
+interface RetryOutcome<T> {
+  status: "success" | "failed";
+  attempts: number;
+  retries: number;
+  data?: T;
+  errorMessage?: string;
+  errorCode?: string;
+}
+
+async function runWithRetry<T>(
+  operation: BatchOperation<T>,
+  opts: {
+    maxRetries: number;
+    retryDelayMs: number;
+    isRetryable: (error: SorokitResult<unknown>) => boolean;
+    delay: (ms: number) => Promise<void>;
+  },
+): Promise<RetryOutcome<T>> {
+  let lastError: SorokitResult<unknown> | undefined;
+  let executedAttempts = 0;
+  let retries = 0;
+  for (let attempt = 0; attempt <= opts.maxRetries; attempt++) {
+    let result: SorokitResult<T>;
+    try {
+      result = await operation.runner(operation.id);
+    } catch (e) {
+      result = err(
+        SorokitErrorCode.UNKNOWN,
+        "Operation runner threw an error.",
+        e,
+      );
+    }
+    executedAttempts += 1;
+    if (result.status === "ok") {
+      return {
+        status: "success",
+        attempts: executedAttempts,
+        retries,
+        ...(result.data !== undefined ? { data: result.data } : {}),
+      };
+    }
+    lastError = result as SorokitResult<unknown>;
+    if (attempt < opts.maxRetries && opts.isRetryable(result as SorokitResult<unknown>)) {
+      retries += 1;
+      const backoffMs = Math.min(
+        opts.retryDelayMs * Math.pow(2, attempt),
+        MAX_RETRY_DELAY_MS,
+      );
+      await opts.delay(backoffMs);
+      continue;
+    }
+    break;
+  }
+
+  const error = lastError;
+  return {
+    status: "failed",
+    attempts: executedAttempts,
+    retries,
+    ...(error && error.status === "error" && error.error
+      ? { errorMessage: error.error.message, errorCode: error.error.code }
+      : { errorMessage: "Operation failed after retries." }),
+  };
+}
+
+/** Simplified busy-wait on worker completion (bounded by average op time). */
+async function waitForWorkers(isDone: () => boolean): Promise<void> {
+  while (!isDone()) {
+    // Yield to the event loop so pending microtasks (worker promises) advance.
+    await new Promise((r) => setTimeout(r, 0));
+  }
+}
+
+// ─── Convenience wrappers ─────────────────────────────────────────────────────
+
+/** Result payload for a trustline creation operation in a bulk run. */
+export interface BulkTrustlineResult {
+  account: string;
+  assetCode: string;
+  assetIssuer: string;
+  /** Built transaction XDR when submission was skipped. */
+  xdr?: string;
+}
+
+export interface BulkCreateTrustlineOp {
+  /** Stable operation id (idempotency). */
+  id: string;
+  account: string;
+  assetCode: string;
+  assetIssuer: string;
+  limit?: string;
+}
+
+export interface BulkCreateTrustlinesInput {
+  horizonUrl: string;
+  networkConfig: ResolvedNetworkConfig;
+  /** Accounts (G-addresses) to establish trustlines for. */
+  accounts: string[];
+  /** Assets to trust. */
+  assets: Array<{ code: string; issuer: string; limit?: string }>;
+  config?: BatchExecutorConfig;
+  /** Optional sign-and-submit callback. When omitted, only the XDR is built. */
+  submit?: (xdr: string) => Promise<SorokitResult<unknown>>;
+}
+
+/**
+ * Establish trustlines for many accounts against many assets, each tracked
+ * independently with bounded concurrency and retry.
+ */
+export async function bulkCreateTrustlines(
+  input: BulkCreateTrustlinesInput,
+): Promise<BatchExecutionReport<BulkTrustlineResult>> {
+  const ops: BatchOperation<BulkTrustlineResult>[] = [];
+  for (const account of input.accounts) {
+    for (const asset of input.assets) {
+      const id = `tl:${account}:${asset.code}:${asset.issuer}`;
+      const runner = async (): Promise<SorokitResult<BulkTrustlineResult>> => {
+        const params: TrustlineParams = {
+          assetCode: asset.code,
+          assetIssuer: asset.issuer,
+          ...(asset.limit !== undefined ? { limit: asset.limit } : {}),
+        };
+        const built = await buildTrustlineTransaction(
+          input.horizonUrl,
+          input.networkConfig,
+          account,
+          params,
+        );
+        if (built.status === "error") return built;
+        if (input.submit) {
+          const submitted = await input.submit(built.data);
+          if (submitted.status === "error") {
+            return err(
+              submitted.error!.code,
+              submitted.error!.message,
+              submitted.error!.cause,
+            );
+          }
+        }
+        return ok<BulkTrustlineResult>({
+          account,
+          assetCode: asset.code,
+          assetIssuer: asset.issuer,
+          ...(input.submit ? {} : { xdr: built.data }),
+        });
+      };
+      ops.push({ id, runner, input: { account, assetCode: asset.code } });
+    }
+  }
+  return runBatchOperations(ops, input.config);
+}
+
+// ─── Bulk payments ────────────────────────────────────────────────────────────
+
+export interface BulkPaymentOp {
+  id: string;
+  source: string;
+  params: import("../transaction/types").PaymentParams;
+}
+
+export interface BulkSendPaymentsInput {
+  horizonUrl: string;
+  networkConfig: ResolvedNetworkConfig;
+  transactions: BulkPaymentOp[];
+  config?: BatchExecutorConfig;
+  /** Optional sign-and-submit callback. When omitted, only the XDR is built. */
+  submit?: (xdr: string) => Promise<SorokitResult<unknown>>;
+}
+
+export interface BulkPaymentResult {
+  source: string;
+  destination: string;
+  xdr?: string;
+}
+
+/**
+ * Execute many payments concurrently, tracking each independently. Building
+ * uses the existing `buildPaymentTransaction`; when a `submit` callback is
+ * supplied, the built XDR is signed and submitted by the caller.
+ */
+export async function bulkSendPayments(
+  input: BulkSendPaymentsInput,
+): Promise<BatchExecutionReport<BulkPaymentResult>> {
+  const ops: BatchOperation<BulkPaymentResult>[] = input.transactions.map(
+    (tx) => {
+      const runner = async (): Promise<SorokitResult<BulkPaymentResult>> => {
+        const built = await buildPaymentTransaction(
+          input.horizonUrl,
+          input.networkConfig,
+          tx.source,
+          tx.params,
+        );
+        if (built.status === "error") return built;
+        if (input.submit) {
+          const submitted = await input.submit(built.data);
+          if (submitted.status === "error") {
+            return err(
+              submitted.error!.code,
+              submitted.error!.message,
+              submitted.error!.cause,
+            );
+          }
+        }
+        return ok<BulkPaymentResult>({
+          source: tx.source,
+          destination: tx.params.destination,
+          ...(input.submit ? {} : { xdr: built.data }),
+        });
+      };
+      return { id: tx.id, runner, input: { source: tx.source } };
+    },
+  );
+  return runBatchOperations(ops, input.config);
+}
+
+// ─── Bulk key rotation ────────────────────────────────────────────────────────
+
+export interface BulkRotateKeyOp {
+  id: string;
+  account: string;
+  oldKey: string;
+  newKey: string;
+  newKeyWeight?: number;
+}
+
+export interface BulkRotateKeysInput {
+  horizonUrl: string;
+  networkConfig: ResolvedNetworkConfig;
+  /** Accounts whose keys are being rotated. */
+  accounts: BulkRotateKeyOp[];
+  config?: BatchExecutorConfig;
+  /** Optional sign-and-submit callback. When omitted, only the XDR is built. */
+  submit?: (xdr: string) => Promise<SorokitResult<unknown>>;
+}
+
+export interface BulkRotateKeyResult {
+  account: string;
+  xdr?: string;
+}
+
+/**
+ * Rotate signing keys across many accounts, each tracked independently.
+ * Backed by the existing `rotateAccountKey` API.
+ */
+export async function bulkRotateKeys(
+  input: BulkRotateKeysInput,
+): Promise<BatchExecutionReport<BulkRotateKeyResult>> {
+  const ops: BatchOperation<BulkRotateKeyResult>[] = input.accounts.map(
+    (op) => {
+      const runner = async (): Promise<SorokitResult<BulkRotateKeyResult>> => {
+        const params: RotateAccountKeyParams = {
+          account: op.account,
+          oldKey: op.oldKey,
+          newKey: op.newKey,
+          ...(op.newKeyWeight !== undefined
+            ? { newKeyWeight: op.newKeyWeight }
+            : {}),
+        };
+        const built = await rotateAccountKey(
+          input.horizonUrl,
+          input.networkConfig,
+          params,
+        );
+        if (built.status === "error") return built;
+        if (input.submit) {
+          const submitted = await input.submit(built.data);
+          if (submitted.status === "error") {
+            return err(
+              submitted.error!.code,
+              submitted.error!.message,
+              submitted.error!.cause,
+            );
+          }
+        }
+        return ok<BulkRotateKeyResult>({
+          account: op.account,
+          ...(input.submit ? {} : { xdr: built.data }),
+        });
+      };
+      return { id: op.id, runner, input: { account: op.account } };
+    },
+  );
+  return runBatchOperations(ops, input.config);
+}
diff --git a/src/account/index.ts b/src/account/index.ts
index c62e60f..fee799c 100644
--- a/src/account/index.ts
+++ b/src/account/index.ts
@@ -83,6 +83,32 @@ export type {
   BalanceForecastResult,
 } from "./balanceForecast";
 
+// ─── Batch account operations (#514) ─────────────────────────────────────────
+export {
+  bulkCreateTrustlines,
+  bulkSendPayments,
+  bulkRotateKeys,
+  runBatchOperations,
+} from "./batchOperations";
+export type {
+  BatchOperation,
+  BatchRunner,
+  BatchOperationResult,
+  BatchOperationStatus,
+  BatchProgress,
+  BatchExecutorConfig,
+  BatchExecutionReport,
+  BulkTrustlineResult,
+  BulkCreateTrustlineOp,
+  BulkCreateTrustlinesInput,
+  BulkPaymentOp,
+  BulkSendPaymentsInput,
+  BulkPaymentResult,
+  BulkRotateKeyOp,
+  BulkRotateKeysInput,
+  BulkRotateKeyResult,
+} from "./batchOperations";
+
 // ─── Multi-wallet portfolio aggregation (#525) ────────────────────────────────
 export { aggregatePortfolio, assetIdentifier } from "./portfolioAggregation";
 export type {
diff --git a/src/index.ts b/src/index.ts
index f5235c0..ef9def9 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -222,6 +222,32 @@ export type {
   GetAccountActivitySummaryOptions,
 } from "./account/getAccountActivitySummary";
 
+// ─── Batch account operations (#514) ──────────────────────────────────────────
+export {
+  bulkCreateTrustlines,
+  bulkSendPayments,
+  bulkRotateKeys,
+  runBatchOperations,
+} from "./account/batchOperations";
+export type {
+  BatchOperation,
+  BatchRunner,
+  BatchOperationResult,
+  BatchOperationStatus,
+  BatchProgress,
+  BatchExecutorConfig,
+  BatchExecutionReport,
+  BulkTrustlineResult,
+  BulkCreateTrustlineOp,
+  BulkCreateTrustlinesInput,
+  BulkPaymentOp,
+  BulkSendPaymentsInput,
+  BulkPaymentResult,
+  BulkRotateKeyOp,
+  BulkRotateKeysInput,
+  BulkRotateKeyResult,
+} from "./account/batchOperations";
+
 // ─── Transaction validation ───────────────────────────────────────────────────
 export {
   createHashMemo,
@@ -380,6 +406,27 @@ export type {
   WebhookPayload,
   WebhookEventDetails,
 } from "./transaction/webhooks";
+
+// ─── Payment notifications (fix.md) ───────────────────────────────────────────
+export {
+  registerPaymentWebhook,
+  unregisterPaymentWebhook,
+  listPaymentWebhooks,
+  clearPaymentWebhooks,
+  triggerPaymentNotifications,
+  dispatchPaymentNotification,
+  generatePaymentEventId,
+  isPaymentNotificationEvent,
+  PAYMENT_NOTIFICATION_EVENTS,
+} from "./transaction/paymentNotifications";
+export type {
+  PaymentNotificationEvent,
+  PaymentWebhookOptions,
+  PaymentWebhookPayload,
+  PaymentWebhookRegistration,
+  PaymentNotificationInput,
+  PaymentNotificationChannel,
+} from "./transaction/paymentNotifications";
 export {
   DEFAULT_PRICE_CACHE_TTL_MS,
   exportTransactionHistory,
@@ -533,6 +580,32 @@ export type {
   OnContractUpgrade,
   ContractVersionOptions,
 } from "./soroban/contractVersion";
+// ─── Contract metadata versioning & compatibility (fix.md) ────────────────────
+export {
+  computeContractMetadataFingerprint,
+  buildContractMetadataSnapshot,
+  checkContractMetadataCompatibility,
+  checkStaleContractMetadata,
+  applyContractMetadataMigration,
+  invalidateContractMetadataForIncompatibility,
+  invalidateCachedContractMetadata,
+} from "./soroban/contractMetadataCompatibility";
+export type {
+  ContractMetadataVersion,
+  ContractMetadataSnapshot,
+  ContractMetadataChange,
+  ContractMetadataChangeKind,
+  ContractMetadataCompatibilityStatus,
+  ContractMetadataCompatibilityReport,
+  ContractMetadataMigration,
+  ContractMetadataMigrationHook,
+  ContractMetadataMigrationResult,
+  BuildMetadataSnapshotInput,
+  CheckCompatibilityInput,
+  StaleMetadataCheckInput,
+  StaleMetadataCheckResult,
+  InvalidateMetadataInput,
+} from "./soroban/contractMetadataCompatibility";
 // ─── Contract error decoding (#391) ───────────────────────────────────────────
 export {
   decodeContractError,
diff --git a/src/soroban/contractMetadataCompatibility.ts b/src/soroban/contractMetadataCompatibility.ts
new file mode 100644
index 0000000..f15c22f
--- /dev/null
+++ b/src/soroban/contractMetadataCompatibility.ts
@@ -0,0 +1,611 @@
+/**
+ * Contract metadata version tracking and compatibility validation.
+ *
+ * Problem (fix.md)
+ * ----------------
+ * Contract metadata can evolve as contracts are upgraded. Before using cached
+ * metadata the SDK should validate that it corresponds to the deployed contract
+ * interface, distinguishing breaking from non-breaking changes, exposing
+ * migration hooks, and invalidating stale cached metadata.
+ *
+ * Design notes
+ * ------------
+ * - Fingerprints, not semantic version strings, drive compatibility: the
+ *   interface of a contract is reduced to a deterministic fingerprint so
+ *   identical metadata always yields the same fingerprint. Semantic version
+ *   strings are treated as advisory only.
+ * - Metadata is always associated with a contract identifier.
+ * - The subsystem is standalone and additive: it consumes the existing
+ *   `ContractMethod[]` / `ContractSchema` shapes and existing cache
+ *   invalidation helpers, so existing metadata consumers remain untouched.
+ */
+
+import { ok } from "../shared/response";
+import type { SorokitResult } from "../shared/response";
+import type { SorokitCache } from "../shared/cache";
+import type { ContractMethod } from "./types";
+import type {
+  ContractSchema,
+} from "./contractMetadata";
+
+/**
+ * A metadata fingerprint plus its associated contract identifier.
+ * This is the version identity used for compatibility checks — it does NOT
+ * rely on semantic version strings.
+ */
+export interface ContractMetadataVersion {
+  /** Stellar contract address (C...). */
+  contractId: string;
+  /** Deterministic fingerprint of the contract interface. */
+  fingerprint: string;
+  /**
+   * Optional advisory version string (e.g. from `contractmetav0`). Used only
+   * as a fast-path hint; the fingerprint is authoritative.
+   */
+  version?: string;
+}
+
+/** Metadata that is pinned to a contract identifier. */
+export interface ContractMetadataSnapshot {
+  /** Contract identifier the metadata belongs to. */
+  contractId: string;
+  /** The discovered contract interface (methods). */
+  methods: ContractMethod[];
+  /** Deterministic fingerprint of `methods`. */
+  fingerprint: string;
+  /** Optional advisory version string. */
+  version?: string;
+}
+
+/** Canonicalized method descriptor used to derive fingerprints. */
+interface NormalizedMethod {
+  name: string;
+  args: string[];
+  return: string;
+}
+
+/** Kind of interface difference detected between two metadata snapshots. */
+export type ContractMetadataChangeKind =
+  | "ADDED_METHOD"
+  | "REMOVED_METHOD"
+  | "CHANGED_ARG_TYPE"
+  | "CHANGED_ARG_COUNT"
+  | "CHANGED_RETURN_TYPE";
+
+/** A single detected difference between two metadata snapshots. */
+export interface ContractMetadataChange {
+  kind: ContractMetadataChangeKind;
+  /** The method the change affects, when applicable. */
+  method: string;
+  /** Human-readable description of the change. */
+  message: string;
+  /** Whether this change is breaking for existing callers. */
+  breaking: boolean;
+}
+
+/** Compatibility outcome between two metadata versions. */
+export type ContractMetadataCompatibilityStatus =
+  | "identical"
+  | "compatible"
+  | "incompatible"
+  | "missing";
+
+/** Structured report from a metadata compatibility check. */
+export interface ContractMetadataCompatibilityReport {
+  /** Overall compatibility classification. */
+  status: ContractMetadataCompatibilityStatus;
+  /** Convenience boolean: `true` for identical/compatible. */
+  compatible: boolean;
+  /** Every detected interface difference (empty when identical/missing). */
+  changes: ContractMetadataChange[];
+  /** Non-fatal advisories (e.g. missing version, metadata absent). */
+  warnings: string[];
+  /** Fatal explanation when status is `incompatible`. */
+  errors: string[];
+  /** True when stale/incompatible cached metadata must be invalidated. */
+  shouldInvalidate: boolean;
+}
+
+/** A single migration step transforming candidate metadata. */
+export interface ContractMetadataMigration {
+  /** When true (and the transformation preserves compatibility), the metadata is considered migrated. */
+  applied: boolean;
+  /** Human-readable description of the migration performed. */
+  description: string;
+}
+
+/**
+ * Application-defined migration hook.
+ *
+ * Receives the candidate metadata snapshot and may transform it to a newer
+ * interface. Return `null` when the hook cannot migrate the metadata (the
+ * change is treated as incompatible).
+ */
+export type ContractMetadataMigrationHook = (
+  candidate: ContractMetadataSnapshot,
+) => ContractMetadataSnapshot | null | void;
+
+/** Result of applying migration hooks to incompatible metadata. */
+export interface ContractMetadataMigrationResult {
+  /** The metadata after migrations were applied, if a hook produced one. */
+  snapshot: ContractMetadataSnapshot | null;
+  /** Migrations that were applied, in order. */
+  migrations: ContractMetadataMigration[];
+  /** True when the original metadata was left un-migratable (incompatible). */
+  blocked: boolean;
+}
+
+// ─── Fingerprint computation ──────────────────────────────────────────────────
+
+/**
+ * Reduce any supported metadata shape to a canonical list of normalized
+ * methods. Accepts the raw array of discovered methods or a typed schema.
+ */
+function normalizeMethods(
+  source: ContractMethod[] | ContractSchema,
+): NormalizedMethod[] {
+  const methods: ContractMethod[] = Array.isArray(source)
+    ? source
+    : source.methods.map((m) => ({
+        name: m.name,
+        inputs: (m.params ?? []).map((p) => ({ name: p.name, type: p.type })),
+        returnType: m.returnType,
+      }));
+
+  return methods.map((m) => ({
+    name: m.name,
+    args: (m.inputs ?? []).map((input) => `${input.name}:${input.type}`),
+    return: m.returnType ?? "void",
+  }));
+}
+
+/** Deterministic 32-bit hash (FNV-1a) — stable across runs and runtimes. */
+function fnv1a(data: string): string {
+  let hash = 0x811c9dc5;
+  for (let i = 0; i < data.length; i++) {
+    hash ^= data.charCodeAt(i);
+    hash = (hash * 0x01000193) >>> 0;
+  }
+  return hash.toString(16).padStart(8, "0");
+}
+
+/**
+ * Compute a deterministic fingerprint of a contract interface.
+ *
+ * The fingerprint is derived from the sorted, canonicalized method signatures
+ * (name, argument types, return type) — NOT from semantic version strings — so
+ * identical interfaces always produce identical fingerprints. This is the
+ * authoritative identity used for compatibility validation.
+ */
+export function computeContractMetadataFingerprint(
+  source: ContractMethod[] | ContractSchema,
+): string {
+  const methods = normalizeMethods(source)
+    .map(
+      (m) =>
+        `${m.name}(${m.args.join(",")}):${m.return}`,
+    )
+    .sort();
+  return fnv1a(methods.join("|"));
+}
+
+/** Normalize a fingerprint to a stable hex string (empty → "missing"). */
+function normalizeFingerprint(fingerprint: string | undefined | null): string {
+  return typeof fingerprint === "string" && fingerprint.length > 0
+    ? fingerprint
+    : "";
+}
+
+// ─── Snapshot construction ────────────────────────────────────────────────────
+
+export interface BuildMetadataSnapshotInput {
+  contractId: string;
+  methods: ContractMethod[] | ContractSchema;
+  /** Optional advisory version string. */
+  version?: string;
+  /** Override fingerprint when the caller has a precomputed one. */
+  fingerprint?: string;
+}
+
+/**
+ * Build a fingerprint-pinned metadata snapshot associated with a contract
+ * identifier.
+ *
+ * @returns A {@link ContractMetadataSnapshot} with the fingerprint computed
+ * deterministically from the interface (or the caller-supplied override).
+ */
+export function buildContractMetadataSnapshot(
+  input: BuildMetadataSnapshotInput,
+): ContractMetadataSnapshot {
+  const methods: ContractMethod[] = Array.isArray(input.methods)
+    ? input.methods
+    : (input.methods as ContractSchema).methods.map((m) => ({
+        name: m.name,
+        inputs: (m.params ?? []).map((p) => ({ name: p.name, type: p.type })),
+        returnType: m.returnType,
+      }));
+
+  return {
+    contractId: input.contractId,
+    methods,
+    fingerprint:
+      input.fingerprint ?? computeContractMetadataFingerprint(input.methods),
+    ...(input.version !== undefined ? { version: input.version } : {}),
+  };
+}
+
+// ─── Compatibility diffing ────────────────────────────────────────────────────
+
+function diffMethods(
+  baseline: NormalizedMethod[],
+  candidate: NormalizedMethod[],
+): ContractMetadataChange[] {
+  const changes: ContractMetadataChange[] = [];
+  const baselineByName = new Map(baseline.map((m) => [m.name, m]));
+  const candidateByName = new Map(candidate.map((m) => [m.name, m]));
+
+  for (const method of baseline) {
+    const next = candidateByName.get(method.name);
+    if (!next) {
+      changes.push({
+        kind: "REMOVED_METHOD",
+        method: method.name,
+        message: `Method '${method.name}' was removed from the contract interface.`,
+        breaking: true,
+      });
+      continue;
+    }
+
+    if (next.args.length !== method.args.length) {
+      changes.push({
+        kind: "CHANGED_ARG_COUNT",
+        method: method.name,
+        message: `Method '${method.name}' argument count changed from ${method.args.length} to ${next.args.length}.`,
+        breaking: true,
+      });
+    } else if (next.args.join("|") !== method.args.join("|")) {
+      changes.push({
+        kind: "CHANGED_ARG_TYPE",
+        method: method.name,
+        message: `Method '${method.name}' argument types changed from [${method.args.join(", ")}] to [${next.args.join(", ")}].`,
+        breaking: true,
+      });
+    }
+
+    if (next.return !== method.return) {
+      changes.push({
+        kind: "CHANGED_RETURN_TYPE",
+        method: method.name,
+        message: `Method '${method.name}' return type changed from '${method.return}' to '${next.return}'.`,
+        breaking: true,
+      });
+    }
+  }
+
+  for (const method of candidate) {
+    if (!baselineByName.has(method.name)) {
+      changes.push({
+        kind: "ADDED_METHOD",
+        method: method.name,
+        message: `Method '${method.name}' was added to the contract interface (non-breaking).`,
+        breaking: false,
+      });
+    }
+  }
+
+  return changes;
+}
+
+// ─── Public compatibility check ───────────────────────────────────────────────
+
+export interface CheckCompatibilityInput {
+  /** The metadata the application currently relies on (baseline). */
+  baseline: ContractMetadataSnapshot;
+  /** Freshly discovered metadata (candidate) for the same contract. */
+  candidate?: ContractMetadataSnapshot;
+  /** Expected fingerprint for the baseline contract, when the caller has one. */
+  expectedFingerprint?: string;
+}
+
+/**
+ * Compare two contract interface versions and classify the result.
+ *
+ * Distinguishes:
+ *  - `identical`  — fingerprints (or both declared versions) match exactly.
+ *  - `compatible` — only non-breaking changes (added methods).
+ *  - `incompatible` — breaking changes (removed methods, changed signatures).
+ *  - `missing`    — the candidate/expected metadata could not be resolved.
+ *
+ * `shouldInvalidate` is `true` when the baseline (cached) metadata must be
+ * discarded because it no longer matches the deployed interface.
+ *
+ * @param input - Baseline snapshot plus candidate/expected identity.
+ * @returns A structured {@link ContractMetadataCompatibilityReport}.
+ */
+export function checkContractMetadataCompatibility(
+  input: CheckCompatibilityInput,
+): ContractMetadataCompatibilityReport {
+  const warnings: string[] = [];
+  const errors: string[] = [];
+
+  const candidateFingerprint = normalizeFingerprint(
+    input.candidate?.fingerprint ?? input.expectedFingerprint,
+  );
+  const baselineFingerprint = normalizeFingerprint(input.baseline.fingerprint);
+
+  // Missing candidate metadata → cannot confirm the cached metadata is valid.
+  if (
+    !input.candidate &&
+    (input.expectedFingerprint === undefined || candidateFingerprint === "")
+  ) {
+    return {
+      status: "missing",
+      compatible: false,
+      changes: [],
+      warnings: ["No candidate metadata was provided; cannot confirm the cached metadata is current."],
+      errors: [],
+      shouldInvalidate: false,
+    };
+  }
+
+  // Fast path: identical fingerprint → identical interface.
+  if (baselineFingerprint !== "" && baselineFingerprint === candidateFingerprint) {
+    return {
+      status: "identical",
+      compatible: true,
+      changes: [],
+      warnings,
+      errors,
+      shouldInvalidate: false,
+    };
+  }
+
+  // Fast path: both declare the same advisory version string.
+  if (
+    input.baseline.version !== undefined &&
+    input.candidate?.version !== undefined &&
+    input.baseline.version === input.candidate.version
+  ) {
+    return {
+      status: "identical",
+      compatible: true,
+      changes: [],
+      warnings: [
+        `Matched on advisory version '${input.baseline.version}'; fingerprints differ (${baselineFingerprint} vs ${candidateFingerprint}).`,
+      ],
+      errors,
+      shouldInvalidate: false,
+    };
+  }
+
+  if (!input.candidate) {
+    // Candidate metadata missing but an expected fingerprint was supplied.
+    if (candidateFingerprint === baselineFingerprint) {
+      return {
+        status: "identical",
+        compatible: true,
+        changes: [],
+        warnings,
+        errors,
+        shouldInvalidate: false,
+      };
+    }
+    return {
+      status: "incompatible",
+      compatible: false,
+      changes: [],
+      warnings,
+      errors: [
+        `Cached metadata fingerprint '${baselineFingerprint}' does not match expected fingerprint '${candidateFingerprint}'.`,
+      ],
+      shouldInvalidate: true,
+    };
+  }
+
+  const changes = diffMethods(
+    normalizeMethods(input.baseline.methods),
+    normalizeMethods(input.candidate.methods),
+  );
+
+  const breaking = changes.filter((c) => c.breaking);
+  if (breaking.length > 0) {
+    return {
+      status: "incompatible",
+      compatible: false,
+      changes,
+      warnings,
+      errors: breaking.map((c) => c.message),
+      shouldInvalidate: true,
+    };
+  }
+
+  if (changes.length > 0) {
+    return {
+      status: "compatible",
+      compatible: true,
+      changes,
+      warnings: changes.map((c) => c.message),
+      errors,
+      shouldInvalidate: false,
+    };
+  }
+
+  // Fingerprints differ but the diff found no structural change (e.g. only
+  // ordering or visibility nuances) — treat as compatible.
+  return {
+    status: "compatible",
+    compatible: true,
+    changes,
+    warnings: ["Fingerprints differ but no breaking interface change was detected."],
+    errors,
+    shouldInvalidate: false,
+  };
+}
+
+// ─── Stale metadata detection before invocation ───────────────────────────────
+
+export interface StaleMetadataCheckInput {
+  contractId: string;
+  /** Cached metadata the application is about to use. */
+  cachedMetadata?: ContractMethod[];
+  /** Fingerprint of the expected/deployed interface. */
+  expectedFingerprint?: string;
+  /** Freshly discovered metadata for the contract, when available. */
+  freshMetadata?: ContractMethod[] | ContractSchema;
+}
+
+export interface StaleMetadataCheckResult {
+  /** True when the cached metadata is current and safe to use. */
+  current: boolean;
+  /** Human-readable explanation when stale. */
+  message: string;
+  /** Compatibility report backing the decision. */
+  report: ContractMetadataCompatibilityReport;
+}
+
+/**
+ * Detect stale cached metadata before a contract invocation.
+ *
+ * When the caller supplies either an expected fingerprint or freshly discovered
+ * metadata, this returns a decision on whether the cached metadata is still
+ * valid. Missing cached metadata or missing reference metadata resolve to a
+ * non-fatal "missing" result (the invocation may still proceed).
+ *
+ * @param input - Contract identity plus cached and reference metadata.
+ * @returns An always-`ok` {@link StaleMetadataCheckResult}; use the `current`
+ * flag to decide whether to re-fetch before invoking.
+ */
+export function checkStaleContractMetadata(
+  input: StaleMetadataCheckInput,
+): SorokitResult<StaleMetadataCheckResult> {
+  if (!input.cachedMetadata || input.cachedMetadata.length === 0) {
+    return ok({
+      current: false,
+      message: "No cached metadata available for this contract.",
+      report: {
+        status: "missing",
+        compatible: false,
+        changes: [],
+        warnings: ["Cached metadata is absent."],
+        errors: [],
+        shouldInvalidate: false,
+      },
+    });
+  }
+
+  if (
+    input.expectedFingerprint === undefined &&
+    !input.freshMetadata
+  ) {
+    return ok({
+      current: true,
+      message: "No reference metadata to compare against; assuming cached metadata is current.",
+      report: {
+        status: "missing",
+        compatible: true,
+        changes: [],
+        warnings: ["No reference fingerprint provided."],
+        errors: [],
+        shouldInvalidate: false,
+      },
+    });
+  }
+
+  const baseline = buildContractMetadataSnapshot({
+    contractId: input.contractId,
+    methods: input.cachedMetadata,
+  });
+
+  const report = checkContractMetadataCompatibility({
+    baseline,
+    ...(input.freshMetadata
+      ? {
+          candidate: buildContractMetadataSnapshot({
+            contractId: input.contractId,
+            methods: input.freshMetadata,
+          }),
+        }
+      : {}),
+    ...(input.expectedFingerprint !== undefined
+      ? { expectedFingerprint: input.expectedFingerprint }
+      : {}),
+  });
+
+  if (report.status === "identical") {
+    return ok({
+      current: true,
+      message: "Cached metadata matches the deployed contract interface.",
+      report,
+    });
+  }
+
+  return ok({
+    current: false,
+    message:
+      report.status === "missing"
+        ? "Cached metadata could not be verified against a reference."
+        : report.errors[0] ??
+          "Cached metadata is incompatible with the deployed contract interface.",
+    report,
+  });
+}
+
+// ─── Migration hooks ──────────────────────────────────────────────────────────
+
+/**
+ * Apply migration hooks to incompatible metadata.
+ *
+ * Hooks run in order against a copy of the candidate snapshot. The first hook
+ * that returns a snapshot is considered the migration result. If no hook
+ * produces a migrated snapshot the metadata is left blocked (incompatible).
+ *
+ * @param candidate - The freshly discovered (newer) metadata.
+ * @param hooks     - Application-provided migration hooks.
+ * @returns A {@link ContractMetadataMigrationResult}.
+ */
+export function applyContractMetadataMigration(
+  candidate: ContractMetadataSnapshot,
+  hooks: ContractMetadataMigrationHook[],
+): ContractMetadataMigrationResult {
+  const migrations: ContractMetadataMigration[] = [];
+
+  for (const hook of hooks) {
+    const result = hook(candidate);
+    if (result && result.fingerprint) {
+      migrations.push({
+        applied: true,
+        description: `A migration hook produced a metadata snapshot with fingerprint '${result.fingerprint}'.`,
+      });
+      return { snapshot: result, migrations, blocked: false };
+    }
+    migrations.push({ applied: false, description: "A migration hook made no change." });
+  }
+
+  return { snapshot: null, migrations, blocked: true };
+}
+
+// ─── Cache invalidation integration ───────────────────────────────────────────
+
+export interface InvalidateMetadataInput {
+  contractId: string;
+  /** In-memory metadata cache (from `contractMetadata`). */
+  cache?: SorokitCache;
+  /** Invalidate the fallback memory state as well. */
+  reset?: () => void;
+}
+
+/**
+ * Invalidate cached metadata for a contract after an incompatible interface
+ * change is detected. Safe to call unconditionally.
+ */
+export function invalidateContractMetadataForIncompatibility(
+  input: InvalidateMetadataInput,
+): void {
+  input.reset?.();
+  input.cache?.invalidate(`sorokit:contract-metadata:${input.contractId}`);
+  input.cache?.invalidate(`sorokit:contract-schema:${input.contractId}`);
+}
+
+// Re-export the standard invalidation helper so callers can keep a single
+// import surface without relying on `contractMetadata` internals.
+export { invalidateContractCache as invalidateCachedContractMetadata } from "./contractMetadata";
diff --git a/src/soroban/contractStateOptimization.ts b/src/soroban/contractStateOptimization.ts
new file mode 100644
index 0000000..bc9eab8
--- /dev/null
+++ b/src/soroban/contractStateOptimization.ts
@@ -0,0 +1,703 @@
+/**
+ * Contract state optimization (fix.md).
+ *
+ * Utilities that reduce the serialization overhead of large contract state
+ * payloads before storage or transmission, while keeping compression
+ * transparent to consumers: callers write optimized state and read back the
+ * original logical representation without manually handling decompression.
+ *
+ * Design principles
+ * -----------------
+ * - Correctness and compatibility take priority over maximum compression.
+ *   Not every Soroban storage value should be shrunk: unsupported values
+ *   fail safely instead of producing corrupted output.
+ * - Serialization is deterministic: object keys are canonicalized (sorted) at
+ *   every depth so two structurally equal states encode byte-for-byte the
+ *   same regardless of insertion order.
+ * - Caller-owned state is never mutated. Encoding only reads the input, and
+ *   decoding builds entirely new objects.
+ * - Metadata identifies the encoding and compression strategy, the measured
+ *   sizes, and a fingerprint so integrity can be verified on read.
+ */
+
+import { gunzipSync, gzipSync } from "node:zlib";
+import type { ZlibOptions } from "node:zlib";
+import { err, ok } from "../shared/response";
+import { SorokitErrorCode } from "../shared/response";
+import type { SorokitResult } from "../shared/response";
+
+// ─── Types ────────────────────────────────────────────────────────────────────
+
+/** The compact, tagged encoding used for optimized state. */
+export type StateEncoding = "tagged";
+
+/** Compression applied to the encoded bytes before returning them. */
+export type StateCompression = "none" | "gzip";
+
+export interface ContractStateOptimizeOptions {
+  /**
+   * Whether to gzip the encoded bytes. `"auto"` (default) compresses only
+   * when the optimized encoding is at least `minCompressBytes` long, so tiny
+   * payloads stay uncompressed and deterministic. `"none"` never compresses.
+   */
+  compression?: "none" | "auto";
+  /** Minimum encoded byte length before `"auto"` applies gzip (default: 64). */
+  minCompressBytes?: number;
+  /** Gzip compression level 0-9 (default: 6). Only used when compressing. */
+  level?: number;
+}
+
+/**
+ * Describes how an optimized state payload was encoded and how much smaller
+ * it is than the canonical representation of the original state.
+ */
+export interface ContractStateMetadata {
+  /** Fixed format marker so readers can detect the payload kind. */
+  format: "contract-state-optimized";
+  /** Format version. Bumping this invalidates decoding of older payloads. */
+  version: 1;
+  /** The state-encoding strategy that produced `data`. */
+  encoding: StateEncoding;
+  /** The compression applied to the encoding. */
+  compression: StateCompression;
+  /** Byte length of the canonical (order-independent) representation. */
+  originalBytes: number;
+  /** Byte length of the optimized encoding before compression. */
+  encodedBytes: number;
+  /** Byte length actually stored in `data` (after compression). */
+  finalBytes: number;
+  /** `finalBytes / originalBytes`; 0 means fully compressed, 1 means none. */
+  ratio: number;
+  /** Percentage of the original size saved, floored at 0 (0-100). */
+  savingsPercent: number;
+  /** Deterministic fingerprint of the canonical state for integrity checks. */
+  hash: string;
+  /** Number of top-level state entries. */
+  entries: number;
+}
+
+/** The compressed/encoded payload returned by {@link compressContractState}. */
+export interface OptimizedContractState {
+  /** Encoded (and optionally compressed) bytes. */
+  data: Uint8Array;
+  /** Metadata needed to decode `data` and report size savings. */
+  metadata: ContractStateMetadata;
+}
+
+// ─── Error helpers ────────────────────────────────────────────────────────────
+
+const UNSUPPORTED = (value: unknown, path: string): SorokitResult<never> => {
+  const type = value === null ? "null" : typeof value;
+  const desc =
+    typeof value === "object"
+      ? value instanceof Uint8Array
+        ? "bytes"
+        : "non-serializable object"
+      : type;
+  return err(
+    SorokitErrorCode.INVALID_CONFIG,
+    `compressContractState: unsupported value of kind "${desc}" at "${path}". ` +
+      "Supported values are null, booleans, finite numbers, bigints, strings, " +
+      "byte arrays, arrays, and plain objects. Refusing rather than corrupting state.",
+    undefined,
+  );
+};
+
+// ─── Canonical representation (deterministic size + fingerprint) ──────────────
+
+function canonicalize(value: unknown): string {
+  if (value === null) return "null";
+  if (typeof value === "boolean") return value ? "true" : "false";
+  if (typeof value === "number") return JSON.stringify(value);
+  if (typeof value === "bigint") return `"#bigint:${value.toString()}#"`;
+  if (typeof value === "string") return JSON.stringify(value);
+  if (value instanceof Uint8Array) {
+    return `"#bytes:${Buffer.from(value).toString("base64")}#"`;
+  }
+  if (Array.isArray(value)) return `[${value.map(canonicalize).join(",")}]`;
+  const record = value as Record<string, unknown>;
+  const entries = Object.keys(record)
+    .sort()
+    .map((key) => `${JSON.stringify(key)}:${canonicalize(record[key])}`);
+  return `{${entries.join(",")}}`;
+}
+
+function fnv1a(input: string): string {
+  let hash = 0x811c9dc5n;
+  const PRIME = 0x01000193n;
+  const MASK = 0xffffffffn;
+  for (let index = 0; index < input.length; index += 1) {
+    hash = ((hash ^ BigInt(input.charCodeAt(index))) * PRIME) & MASK;
+  }
+  return hash.toString(16).padStart(8, "0");
+}
+
+// ─── Byte writer / reader ─────────────────────────────────────────────────────
+
+class ByteWriter {
+  private buf = new Uint8Array(64);
+  private len = 0;
+
+  private ensure(extra: number): void {
+    if (this.len + extra <= this.buf.length) return;
+    let capacity = this.buf.length * 2;
+    while (capacity < this.len + extra) capacity *= 2;
+    const next = new Uint8Array(capacity);
+    next.set(this.buf.subarray(0, this.len));
+    this.buf = next;
+  }
+
+  byte(value: number): void {
+    this.ensure(1);
+    this.buf[this.len++] = value & 0xff;
+  }
+
+  bytes(value: Uint8Array): void {
+    this.ensure(value.length);
+    this.buf.set(value, this.len);
+    this.len += value.length;
+  }
+
+  /** Unsigned base-128 little-endian varint (arbitrary precision bigint). */
+  varint(value: bigint): void {
+    let v = value < 0n ? 0n : value;
+    if (v < 0n) v = BigInt.asUintN(64, v);
+    for (;;) {
+      const byte = Number(v & 0x7fn);
+      v >>= 7n;
+      if (v === 0n) {
+        this.byte(byte);
+        break;
+      }
+      this.byte(byte | 0x80);
+    }
+  }
+
+  /** Zigzag then varint (arbitrary precision). */
+  zzvarint(value: bigint): void {
+    const zig = value < 0n ? ((-value) << 1n) - 1n : value << 1n;
+    this.varint(zig);
+  }
+
+  float64(value: number): void {
+    const tmp = new Uint8Array(8);
+    new DataView(tmp.buffer).setFloat64(0, value, true);
+    this.bytes(tmp);
+  }
+
+  finish(): Uint8Array {
+    return this.buf.slice(0, this.len);
+  }
+}
+
+class ByteReader {
+  private pos = 0;
+  constructor(private readonly buf: Uint8Array) {}
+
+  byte(): number {
+    if (this.pos >= this.buf.length) throw new Error("unexpected end of payload");
+    const value = this.buf[this.pos];
+    if (value === undefined) throw new Error("unexpected end of payload");
+    this.pos += 1;
+    return value;
+  }
+
+  bytes(n: number): Uint8Array {
+    if (this.pos + n > this.buf.length) throw new Error("unexpected end of payload");
+    const out = this.buf.slice(this.pos, this.pos + n);
+    this.pos += n;
+    return out;
+  }
+
+  varint(): bigint {
+    let result = 0n;
+    let shift = 0n;
+    for (;;) {
+      const byte = this.byte();
+      result |= BigInt(byte & 0x7f) << shift;
+      if ((byte & 0x80) === 0) break;
+      shift += 7n;
+    }
+    return result;
+  }
+
+  zzvarint(): bigint {
+    const zz = this.varint();
+    return (zz >> 1n) ^ -(zz & 1n);
+  }
+
+  float64(): number {
+    const value = new DataView(
+      this.buf.buffer,
+      this.buf.byteOffset + this.pos,
+    ).getFloat64(0, true);
+    this.pos += 8;
+    return value;
+  }
+}
+
+// ─── Tagged encoder ───────────────────────────────────────────────────────────
+
+const TAG_NULL = 0x00;
+const TAG_TRUE = 0x01;
+const TAG_FALSE = 0x02;
+const TAG_STRING = 0x03;
+const TAG_INT = 0x04;
+const TAG_BIGINT = 0x05;
+const TAG_BYTES = 0x06;
+const TAG_DOUBLE = 0x07;
+const TAG_ARRAY = 0x08;
+const TAG_OBJECT = 0x09;
+
+type EncodeResult = SorokitResult<Uint8Array>;
+
+function encodeTagged(state: Record<string, unknown>): EncodeResult {
+  const writer = new ByteWriter();
+  const seen = new Set<object>();
+
+  const encodeValue = (value: unknown, path: string): string | null => {
+    if (value === null) {
+      writer.byte(TAG_NULL);
+      return null;
+    }
+    if (typeof value === "boolean") {
+      writer.byte(value ? TAG_TRUE : TAG_FALSE);
+      return null;
+    }
+    if (typeof value === "number") {
+      if (!Number.isFinite(value)) {
+        return UNSUPPORTED(value, path).error!.message;
+      }
+      if (Number.isInteger(value) && Number.isSafeInteger(value)) {
+        writer.byte(TAG_INT);
+        writer.zzvarint(BigInt(value));
+      } else {
+        writer.byte(TAG_DOUBLE);
+        writer.float64(value);
+      }
+      return null;
+    }
+    if (typeof value === "bigint") {
+      writer.byte(TAG_BIGINT);
+      writer.zzvarint(value);
+      return null;
+    }
+    if (typeof value === "string") {
+      const bytes = Buffer.from(value, "utf8");
+      writer.byte(TAG_STRING);
+      writer.varint(BigInt(bytes.length));
+      writer.bytes(bytes);
+      return null;
+    }
+    if (typeof value !== "object") {
+      return UNSUPPORTED(value, path).error!.message;
+    }
+    if (value instanceof Uint8Array) {
+      writer.byte(TAG_BYTES);
+      writer.varint(BigInt(value.length));
+      writer.bytes(value);
+      return null;
+    }
+    if (seen.has(value)) {
+      return err(
+        SorokitErrorCode.INVALID_CONFIG,
+        `compressContractState: circular reference detected at "${path}".`,
+      ).error!.message;
+    }
+
+    seen.add(value);
+    let result: string | null = null;
+    if (Array.isArray(value)) {
+      writer.byte(TAG_ARRAY);
+      writer.varint(BigInt(value.length));
+      for (let i = 0; i < value.length; i++) {
+        result = encodeValue(value[i], `${path}[${i}]`);
+        if (result !== null) break;
+      }
+    } else {
+      const proxy = value as Record<string, unknown>;
+      const keys = Object.keys(proxy).sort();
+      writer.byte(TAG_OBJECT);
+      writer.varint(BigInt(keys.length));
+      for (const key of keys) {
+        result = encodeKey(key) ?? encodeValue(proxy[key], `${path}.${key}`);
+        if (result !== null) break;
+      }
+    }
+    seen.delete(value);
+    return result;
+  };
+
+  const encodeKey = (key: string): string | null => {
+    const bytes = Buffer.from(key, "utf8");
+    writer.byte(TAG_STRING);
+    writer.varint(BigInt(bytes.length));
+    writer.bytes(bytes);
+    return null;
+  };
+
+  const entries = Object.keys(state).sort();
+  writer.byte(TAG_OBJECT);
+  writer.varint(BigInt(entries.length));
+  for (const key of entries) {
+    const keyError = encodeKey(key);
+    if (keyError !== null) return err(SorokitErrorCode.INVALID_CONFIG, keyError);
+    const valueError = encodeValue(state[key], key);
+    if (valueError !== null) {
+      return err(SorokitErrorCode.INVALID_CONFIG, valueError);
+    }
+  }
+
+  return ok(writer.finish());
+}
+
+// ─── Tagged decoder ───────────────────────────────────────────────────────────
+
+type DecodeResult = SorokitResult<unknown>;
+
+function decodeTagged(data: Uint8Array): DecodeResult {
+  const reader = new ByteReader(data);
+
+  const decodeValue = (): { value: unknown; error?: string } => {
+    const tag = reader.byte();
+    switch (tag) {
+      case TAG_NULL:
+        return { value: null };
+      case TAG_TRUE:
+        return { value: true };
+      case TAG_FALSE:
+        return { value: false };
+      case TAG_INT:
+        return { value: Number(reader.zzvarint()) };
+      case TAG_BIGINT:
+        return { value: reader.zzvarint() };
+      case TAG_DOUBLE:
+        return { value: reader.float64() };
+      case TAG_STRING: {
+        const len = Number(reader.varint());
+        return { value: Buffer.from(reader.bytes(len)).toString("utf8") };
+      }
+      case TAG_BYTES: {
+        const len = Number(reader.varint());
+        return { value: reader.bytes(len) };
+      }
+      case TAG_ARRAY: {
+        const count = Number(reader.varint());
+        const arr: unknown[] = [];
+        for (let i = 0; i < count; i++) {
+          const item = decodeValue();
+          if (item.error !== undefined) return item;
+          arr.push(item.value);
+        }
+        return { value: arr };
+      }
+      case TAG_OBJECT: {
+        const count = Number(reader.varint());
+        const obj: Record<string, unknown> = {};
+        for (let i = 0; i < count; i++) {
+          const keyResult = decodeValue();
+          if (keyResult.error !== undefined) return keyResult;
+          if (typeof keyResult.value !== "string") {
+            return { value: undefined, error: "decodeContractState: invalid object key in payload." };
+          }
+          const valueResult = decodeValue();
+          if (valueResult.error !== undefined) return valueResult;
+          obj[keyResult.value] = valueResult.value;
+        }
+        return { value: obj };
+      }
+      default:
+        return {
+          value: undefined,
+          error: `decodeContractState: unknown tag ${tag}; payload may be corrupted or from a newer version.`,
+        };
+    }
+  };
+
+  try {
+    const root = decodeValue();
+    if (root.error !== undefined) {
+      return err(SorokitErrorCode.CONTRACT_READ_FAILED, root.error);
+    }
+    if (typeof root.value !== "object" || root.value === null || Array.isArray(root.value)) {
+      return err(
+        SorokitErrorCode.CONTRACT_READ_FAILED,
+        "decodeContractState: root value is not a state object.",
+      );
+    }
+    return ok(root.value as Record<string, unknown>);
+  } catch {
+    return err(
+      SorokitErrorCode.CONTRACT_READ_FAILED,
+      "decodeContractState: truncated or corrupted payload.",
+    );
+  }
+}
+
+// ─── Public API ───────────────────────────────────────────────────────────────
+
+const DEFAULT_MIN_COMPRESS_BYTES = 64;
+const DEFAULT_GZIP_LEVEL = 6;
+
+/**
+ * Optimize a contract state payload for storage or transmission.
+ *
+ * The state is first deterministically encoded with a compact tagged binary
+ * encoding that efficiently represents common primitive and structured values
+ * (integers, bigints, strings, byte arrays, arrays, objects). When
+ * `compression` is `"auto"` and the encoding is large enough, the bytes are
+ * then gzipped. The returned metadata records the strategy and the measured
+ * sizes so savings can be reported and the payload decoded later.
+ *
+ * The input state is never mutated, and unsupported values cause a safe error
+ * rather than corrupted output.
+ *
+ * @param state - The contract state to optimize. Must be a plain object of
+ *   deterministic values.
+ * @param options - Optimization options.
+ * @returns The encoded bytes plus descriptive metadata, or an error.
+ */
+export function compressContractState(
+  state: Readonly<Record<string, unknown>>,
+  options: ContractStateOptimizeOptions = {},
+): SorokitResult<OptimizedContractState> {
+  if (state === null || typeof state !== "object" || Array.isArray(state)) {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      "compressContractState: state must be a plain object.",
+    );
+  }
+
+  // Validate and encode first. The encoder rejects unsupported values and
+  // circular references, guaranteeing canonicalize below only ever sees safe
+  // (deterministic) input.
+  const encodedResult = encodeTagged(state as Record<string, unknown>);
+  if (encodedResult.status === "error") return encodedResult;
+  const encoded = encodedResult.data;
+  const encodedBytes = encoded.length;
+  const canonical = canonicalize(state);
+  const originalBytes = new TextEncoder().encode(canonical).byteLength;
+
+  const compressionMode = options.compression ?? "auto";
+  const minCompressBytes = Math.max(0, options.minCompressBytes ?? DEFAULT_MIN_COMPRESS_BYTES);
+  const shouldCompress = compressionMode === "none" ? false : encodedBytes >= minCompressBytes;
+
+  let finalBytes = encodedBytes;
+  let data = encoded;
+  let compression: StateCompression = "none";
+  if (shouldCompress) {
+    try {
+      data = gzipSync(encoded, {
+        level: options.level ?? DEFAULT_GZIP_LEVEL,
+        // mtime: 0 keeps gzip output deterministic across runs.
+        mtime: 0,
+      } as ZlibOptions);
+      compression = "gzip";
+      finalBytes = data.length;
+    } catch {
+      // Compression must never corrupt output; fall back to uncompressed.
+      data = encoded;
+      compression = "none";
+      finalBytes = encodedBytes;
+    }
+  }
+
+  const ratio = originalBytes === 0 ? 0 : finalBytes / originalBytes;
+  const savingsPercent = originalBytes === 0 ? 0 : Math.max(0, (1 - ratio) * 100);
+
+  return ok({
+    data,
+    metadata: {
+      format: "contract-state-optimized",
+      version: 1,
+      encoding: "tagged",
+      compression,
+      originalBytes,
+      encodedBytes,
+      finalBytes,
+      ratio,
+      savingsPercent,
+      hash: fnv1a(canonical),
+      entries: Object.keys(state).length,
+    },
+  });
+}
+
+/**
+ * Restore the original logical state from an optimized payload.
+ *
+ * Decoding uses the strategy recorded in the metadata and verifies the payload
+ * against the stored fingerprint, failing safely (with an error) rather than
+ * returning corrupted state when the payload is damaged, truncated, or from an
+ * unsupported format version.
+ *
+ * @param data - The bytes produced by {@link compressContractState}.
+ * @param metadata - The metadata returned alongside those bytes.
+ * @returns The restored state object, or an error.
+ */
+export function decompressContractState(
+  data: Uint8Array,
+  metadata: ContractStateMetadata,
+): SorokitResult<Record<string, unknown>> {
+  if (metadata.format !== "contract-state-optimized") {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      "decompressContractState: unrecognized payload format.",
+    );
+  }
+  if (metadata.version !== 1) {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      `decompressContractState: unsupported format version ${metadata.version}.`,
+    );
+  }
+  if (metadata.encoding !== "tagged") {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      `decompressContractState: unsupported encoding "${metadata.encoding}".`,
+    );
+  }
+
+  let encoded: Uint8Array;
+  if (metadata.compression === "gzip") {
+    try {
+      encoded = gunzipSync(data);
+    } catch {
+      return err(
+        SorokitErrorCode.CONTRACT_READ_FAILED,
+        "decompressContractState: failed to decompress gzip payload; data may be corrupted.",
+      );
+    }
+  } else if (metadata.compression === "none") {
+    encoded = data;
+  } else {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      `decompressContractState: unsupported compression "${metadata.compression}".`,
+    );
+  }
+
+  const decoded = decodeTagged(encoded);
+  if (decoded.status === "error") return decoded;
+  const state = decoded.data as Record<string, unknown>;
+
+  // Integrity check: the decoded state must match the recorded fingerprint.
+  const actualHash = fnv1a(canonicalize(state));
+  if (actualHash !== metadata.hash) {
+    return err(
+      SorokitErrorCode.CONTRACT_READ_FAILED,
+      "decompressContractState: payload fingerprint mismatch; state does not match the recorded metadata.",
+    );
+  }
+
+  return ok(state);
+}
+
+// ─── Measurement + benchmarks ─────────────────────────────────────────────────
+
+export interface ContractStateSizeReport {
+  originalBytes: number;
+  encodedBytes: number;
+  compressedBytes: number;
+  ratio: number;
+  savingsPercent: number;
+  entries: number;
+}
+
+/**
+ * Measure the size of a contract state under the optimized encoding and the
+ * savings it would yield versus its canonical representation, without writing
+ * state. See {@link compressContractState} for supported value types.
+ */
+export function measureContractState(
+  state: Readonly<Record<string, unknown>>,
+): SorokitResult<ContractStateSizeReport> {
+  const encodedResult = encodeTagged(state as Record<string, unknown>);
+  if (encodedResult.status === "error") return encodedResult;
+  const encodedBytes = encodedResult.data.length;
+  const canonicalBytes = new TextEncoder().encode(canonicalize(state)).byteLength;
+
+  let compressedBytes = encodedBytes;
+  try {
+    compressedBytes = gzipSync(encodedResult.data, {
+      level: DEFAULT_GZIP_LEVEL,
+      mtime: 0,
+    } as ZlibOptions).length;
+  } catch {
+    compressedBytes = encodedBytes;
+  }
+
+  const ratio = canonicalBytes === 0 ? 0 : compressedBytes / canonicalBytes;
+  return ok({
+    originalBytes: canonicalBytes,
+    encodedBytes,
+    compressedBytes,
+    ratio,
+    savingsPercent: canonicalBytes === 0 ? 0 : Math.max(0, (1 - ratio) * 100),
+    entries: Object.keys(state).length,
+  });
+}
+
+export interface CompressionBenchmark {
+  /** Number of compress/decompress iterations measured. */
+  iterations: number;
+  compressMs: number;
+  decompressMs: number;
+  /** Total round-trip time in ms. */
+  totalMs: number;
+  /** Optimized payload size in bytes. */
+  payloadBytes: number;
+  /** Compressed size ratio (final / canonical original). */
+  ratio: number;
+  /** Per-operation mean latency in milliseconds. */
+  compressMeanMs: number;
+  decompressMeanMs: number;
+}
+
+/**
+ * Benchmark {@link compressContractState} and {@link decompressContractState}
+ * against a representative state payload.
+ *
+ * @param state - The state to benchmark.
+ * @param iterations - Number of compress/decompress rounds (default: 100).
+ * @returns Aggregate timing and size statistics.
+ */
+export function benchmarkContractState(
+  state: Readonly<Record<string, unknown>>,
+  iterations = 100,
+): SorokitResult<CompressionBenchmark> {
+  if (!Number.isInteger(iterations) || iterations < 1) {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      "benchmarkContractState: iterations must be a positive integer.",
+    );
+  }
+
+  let compressed: OptimizedContractState | undefined;
+  const compressStart = performance.now();
+  for (let i = 0; i < iterations; i++) {
+    const result = compressContractState(state);
+    if (result.status === "error") return result;
+    compressed = result.data;
+  }
+  const compressMs = performance.now() - compressStart;
+
+  const decompressStart = performance.now();
+  for (let i = 0; i < iterations; i++) {
+    const result = decompressContractState(compressed!.data, compressed!.metadata);
+    if (result.status === "error") return result;
+  }
+  const decompressMs = performance.now() - decompressStart;
+
+  const metadata = compressed!.metadata;
+  return ok({
+    iterations,
+    compressMs,
+    decompressMs,
+    totalMs: compressMs + decompressMs,
+    payloadBytes: metadata.finalBytes,
+    ratio: metadata.ratio,
+    compressMeanMs: compressMs / iterations,
+    decompressMeanMs: decompressMs / iterations,
+  });
+}
diff --git a/src/soroban/index.ts b/src/soroban/index.ts
index a1aa7d2..bafc875 100644
--- a/src/soroban/index.ts
+++ b/src/soroban/index.ts
@@ -76,6 +76,31 @@ export type {
   OnContractUpgrade,
   ContractVersionOptions,
 } from "./contractVersion";
+export {
+  computeContractMetadataFingerprint,
+  buildContractMetadataSnapshot,
+  checkContractMetadataCompatibility,
+  checkStaleContractMetadata,
+  applyContractMetadataMigration,
+  invalidateContractMetadataForIncompatibility,
+  invalidateCachedContractMetadata,
+} from "./contractMetadataCompatibility";
+export type {
+  ContractMetadataVersion,
+  ContractMetadataSnapshot,
+  ContractMetadataChange,
+  ContractMetadataChangeKind,
+  ContractMetadataCompatibilityStatus,
+  ContractMetadataCompatibilityReport,
+  ContractMetadataMigration,
+  ContractMetadataMigrationHook,
+  ContractMetadataMigrationResult,
+  BuildMetadataSnapshotInput,
+  CheckCompatibilityInput,
+  StaleMetadataCheckInput,
+  StaleMetadataCheckResult,
+  InvalidateMetadataInput,
+} from "./contractMetadataCompatibility";
 export {
   decodeContractError,
   DEFAULT_CONTRACT_ERROR_MAP,
@@ -516,6 +541,23 @@ export type {
   SnapshotQuery,
 } from "./contractStateHistory";
 
+// ─── Contract state optimization (#514) ───────────────────────────────────────
+export {
+  compressContractState,
+  decompressContractState,
+  measureContractState,
+  benchmarkContractState,
+} from "./contractStateOptimization";
+export type {
+  StateEncoding,
+  StateCompression,
+  ContractStateOptimizeOptions,
+  ContractStateMetadata,
+  OptimizedContractState,
+  ContractStateSizeReport,
+  CompressionBenchmark,
+} from "./contractStateOptimization";
+
 export {
   MultiSigContractExecution,
   createMultiSigContractExecution,
diff --git a/src/tests/batchOperations.test.ts b/src/tests/batchOperations.test.ts
new file mode 100644
index 0000000..8f72607
--- /dev/null
+++ b/src/tests/batchOperations.test.ts
@@ -0,0 +1,376 @@
+/**
+ * Tests for batch account operations (#514):
+ * bulkCreateTrustlines, bulkSendPayments, bulkRotateKeys, and the generic
+ * runBatchOperations executor.
+ *
+ * Covers: successful batches, partial failures (successes preserved), retries
+ * with idempotency awareness, concurrency limits, progress tracking, and
+ * concurrent-vs-sequential performance behavior.
+ */
+
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { ok, err, SorokitErrorCode } from "../shared/response";
+import type { ResolvedNetworkConfig } from "../shared/types";
+
+const mockBuildPaymentTransaction = vi.hoisted(() =>
+  vi.fn(async () => ok("PAYMENT_XDR")),
+);
+const mockBuildTrustlineTransaction = vi.hoisted(() =>
+  vi.fn(async () => ok("TRUSTLINE_XDR")),
+);
+const mockRotateAccountKey = vi.hoisted(() =>
+  vi.fn(async () => ok("ROTATE_XDR")),
+);
+
+vi.mock("../transaction/buildTransaction", () => ({
+  buildPaymentTransaction: mockBuildPaymentTransaction,
+  buildTrustlineTransaction: mockBuildTrustlineTransaction,
+  buildBulkTrustlines: vi.fn(),
+  buildBulkTrustlineTransaction: vi.fn(),
+  buildPaymentWithTrustline: vi.fn(),
+}));
+
+vi.mock("../account/keyRotation", () => ({
+  rotateAccountKey: mockRotateAccountKey,
+  setAccountRecovery: vi.fn(),
+  recoverAccountKeys: vi.fn(),
+  isValidStellarPublicKey: vi.fn(),
+}));
+
+import {
+  runBatchOperations,
+  bulkSendPayments,
+  bulkCreateTrustlines,
+  bulkRotateKeys,
+} from "../account/batchOperations";
+import type { BatchOperation } from "../account/batchOperations";
+
+const NETWORK_CONFIG: ResolvedNetworkConfig = {
+  network: "testnet",
+  horizonUrl: "https://horizon-testnet.stellar.org",
+  rpcUrl: "https://rpc-testnet.stellar.org",
+  networkPassphrase: "Test SDF Network ; September 2015",
+};
+
+const HORIZON_URL = NETWORK_CONFIG.horizonUrl;
+
+describe("runBatchOperations — core executor", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("runs every operation and reports a successful summary", async () => {
+    const ops: BatchOperation<string>[] = ["a", "b", "c"].map((id) => ({
+      id,
+      runner: async () => ok(`done-${id}`),
+    }));
+
+    const report = await runBatchOperations(ops, { concurrency: 2 });
+
+    expect(report.summary.total).toBe(3);
+    expect(report.summary.succeeded).toBe(3);
+    expect(report.summary.failed).toBe(0);
+    expect(report.summary.skipped).toBe(0);
+    expect(report.summary.allSucceeded).toBe(true);
+    expect(report.results.map((r) => r.status)).toEqual([
+      "success",
+      "success",
+      "success",
+    ]);
+    expect(report.results.map((r) => r.data)).toEqual([
+      "done-a",
+      "done-b",
+      "done-c",
+    ]);
+  });
+
+  it("preserves successes when some operations fail (partial failure)", async () => {
+    const ops: BatchOperation<string>[] = [
+      { id: "ok-1", runner: async () => ok("fine") },
+      {
+        id: "bad-1",
+        runner: async () =>
+          err(SorokitErrorCode.TX_SUBMIT_FAILED, "rejected"),
+      },
+      { id: "ok-2", runner: async () => ok("fine-2") },
+      {
+        id: "bad-2",
+        runner: async () =>
+          err(SorokitErrorCode.TX_BUILD_FAILED, "bad xdr"),
+      },
+    ];
+
+    const report = await runBatchOperations(ops, { concurrency: 4 });
+
+    expect(report.summary.succeeded).toBe(2);
+    expect(report.summary.failed).toBe(2);
+    expect(report.summary.allSucceeded).toBe(false);
+    const byId = new Map(report.results.map((r) => [r.id, r]));
+    expect(byId.get("ok-1")!.status).toBe("success");
+    expect(byId.get("ok-2")!.status).toBe("success");
+    expect(byId.get("bad-1")!.status).toBe("failed");
+    expect(byId.get("bad-1")!.errorCode).toBe(SorokitErrorCode.TX_SUBMIT_FAILED);
+    expect(byId.get("bad-2")!.status).toBe("failed");
+  });
+
+  it("retries retryable failures but not permanent ones", async () => {
+    // NETWORK_ERROR is retryable: succeeds on the 3rd attempt.
+    let networkAttempts = 0;
+    // TX_BUILD_FAILED is not retryable: only runs once.
+    let permanentAttempts = 0;
+
+    const ops: BatchOperation<string>[] = [
+      {
+        id: "retryable",
+        runner: async () => {
+          networkAttempts += 1;
+          if (networkAttempts < 3) {
+            return err(SorokitErrorCode.NETWORK_ERROR, "flaky network");
+          }
+          return ok("recovered");
+        },
+      },
+      {
+        id: "permanent",
+        runner: async () => {
+          permanentAttempts += 1;
+          return err(SorokitErrorCode.TX_BUILD_FAILED, "bad xdr");
+        },
+      },
+    ];
+
+    const report = await runBatchOperations(ops, {
+      maxRetries: 3,
+      retryDelayMs: 0,
+      delay: async () => {},
+    });
+
+    const byId = new Map(report.results.map((r) => [r.id, r]));
+    expect(byId.get("retryable")!.status).toBe("success");
+    expect(byId.get("retryable")!.attempts).toBe(3);
+    expect(byId.get("retryable")!.retries).toBe(2);
+    expect(byId.get("permanent")!.status).toBe("failed");
+    expect(byId.get("permanent")!.attempts).toBe(1);
+    expect(permanentAttempts).toBe(1);
+  });
+
+  it("never re-runs operations already completed (idempotency-aware)", async () => {
+    const executed = vi.fn();
+    const ops: BatchOperation<string>[] = [
+      { id: "known", runner: async () => { executed(); return ok("x"); } },
+      { id: "unknown", runner: async () => { executed(); return ok("y"); } },
+    ];
+
+    const report = await runBatchOperations(ops, {
+      previouslyCompletedIds: ["known"],
+    });
+
+    expect(executed).toHaveBeenCalledTimes(1); // only "unknown" executed
+    const byId = new Map(report.results.map((r) => [r.id, r]));
+    expect(byId.get("known")!.status).toBe("skipped");
+    expect(byId.get("known")!.skipped).toBe(true);
+    expect(byId.get("known")!.attempts).toBe(0);
+    expect(byId.get("unknown")!.status).toBe("success");
+  });
+
+  it("does not duplicate a successfully submitted op across retries", async () => {
+    // Simulates an op whose FIRST attempt succeeds at the network layer but
+    // the response is lost (idempotency hazard). The executor must not re-run it.
+    let calls = 0;
+    const ops: BatchOperation<string>[] = [
+      {
+        id: "already-sent",
+        runner: async () => {
+          calls += 1;
+          if (calls === 1) return ok("submitted");
+          return err(SorokitErrorCode.TX_SUBMIT_FAILED, "would duplicate");
+        },
+      },
+    ];
+    // previouslyCompletedIds marks the op as already handled -> skipped, not re-run.
+    const report = await runBatchOperations(ops, {
+      previouslyCompletedIds: ["already-sent"],
+    });
+    expect(calls).toBe(0);
+    expect(report.results[0].status).toBe("skipped");
+  });
+
+  it("enforces the concurrency limit", async () => {
+    const concurrency = 3;
+    let active = 0;
+    let maxActive = 0;
+    const ops: BatchOperation<string>[] = Array.from({ length: 10 }, (_, i) => ({
+      id: `op-${i}`,
+      runner: async () => {
+        active += 1;
+        maxActive = Math.max(maxActive, active);
+        await new Promise((r) => setTimeout(r, 5));
+        active -= 1;
+        return ok("x");
+      },
+    }));
+
+    await runBatchOperations(ops, { concurrency });
+
+    expect(maxActive).toBeLessThanOrEqual(concurrency);
+    expect(maxActive).toBeGreaterThan(0);
+  });
+
+  it("reports progress through completion", async () => {
+    const snapshots: Array<Record<string, number>> = [];
+    const ops: BatchOperation<string>[] = ["a", "b"].map((id) => ({
+      id,
+      runner: async () => ok(`done-${id}`),
+    }));
+
+    await runBatchOperations(ops, {
+      concurrency: 2,
+      onProgress: (p) =>
+        snapshots.push({
+          planned: p.planned,
+          running: p.running,
+          succeeded: p.succeeded,
+          failed: p.failed,
+          completed: p.completed,
+          total: p.total,
+        }),
+    });
+
+    // Early snapshot before any operation finishes must show planned but 0 completed.
+    const early = snapshots[0];
+    expect(early.total).toBe(2);
+    expect(early.completed).toBe(0);
+    expect(early.planned).toBe(2);
+    // Final snapshot reflects all completed.
+    const last = snapshots[snapshots.length - 1];
+    expect(last.succeeded).toBe(2);
+    expect(last.completed).toBe(2);
+  });
+
+  it("performs faster than sequential execution with many slow ops", async () => {
+    const opDelay = 10;
+    const count = 6;
+    const concurrency = 6;
+    const runFor = async (c: number) => {
+      const ops: BatchOperation<string>[] = Array.from(
+        { length: count },
+        (_, i) => ({
+          id: `op-${i}`,
+          runner: async () => {
+            await new Promise((r) => setTimeout(r, opDelay));
+            return ok("x");
+          },
+        }),
+      );
+      const start = Date.now();
+      await runBatchOperations(ops, { concurrency: c });
+      return Date.now() - start;
+    };
+
+    const sequential = await runFor(1); // fully serial
+    const concurrent = await runFor(concurrency); // fully parallel
+
+    // With all ops in parallel, elapsed should be ~1x opDelay, far below
+    // sequential (count * opDelay). Use a loose upper bound to stay deterministic.
+    expect(concurrent).toBeLessThan(opDelay * count);
+  });
+});
+
+describe("runBatchOperations — bulk wrappers", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("bulkSendPayments builds one transaction per payment without submit", async () => {
+    mockBuildPaymentTransaction.mockResolvedValue(ok("PAYMENT_XDR"));
+    const report = await bulkSendPayments({
+      horizonUrl: HORIZON_URL,
+      networkConfig: NETWORK_CONFIG,
+      transactions: [
+        { id: "p1", source: "GA", params: { destination: "GB", amount: "10" } },
+        { id: "p2", source: "GC", params: { destination: "GD", amount: "5" } },
+      ],
+      config: { concurrency: 2 },
+    });
+
+    expect(mockBuildPaymentTransaction).toHaveBeenCalledTimes(2);
+    expect(report.summary.succeeded).toBe(2);
+    expect(report.summary.allSucceeded).toBe(true);
+    const byId = new Map(report.results.map((r) => [r.id, r]));
+    expect(byId.get("p1")!.data).toMatchObject({ source: "GA" });
+    expect(byId.get("p1")!.data!.xdr).toBe("PAYMENT_XDR");
+  });
+
+  it("bulkSendPayments invokes submit when provided", async () => {
+    const submit = vi.fn(async () => ok("submitted"));
+    await bulkSendPayments({
+      horizonUrl: HORIZON_URL,
+      networkConfig: NETWORK_CONFIG,
+      transactions: [
+        { id: "p1", source: "GA", params: { destination: "GB", amount: "10" } },
+      ],
+      submit,
+    });
+    expect(submit).toHaveBeenCalledWith("PAYMENT_XDR");
+  });
+
+  it("bulkCreateTrustlines builds a trustline per account x asset", async () => {
+    mockBuildTrustlineTransaction.mockResolvedValue(ok("TRUSTLINE_XDR"));
+    const report = await bulkCreateTrustlines({
+      horizonUrl: HORIZON_URL,
+      networkConfig: NETWORK_CONFIG,
+      accounts: ["GA", "GB"],
+      assets: [{ code: "USDC", issuer: "ISSUER" }],
+      config: { concurrency: 2 },
+    });
+
+    expect(report.summary.total).toBe(2); // 2 accounts x 1 asset
+    expect(report.summary.succeeded).toBe(2);
+    expect(mockBuildTrustlineTransaction).toHaveBeenCalledTimes(2);
+    const results = report.results.filter((r) => r.status === "success");
+    expect(results.map((r) => r.data!.assetCode)).toEqual(["USDC", "USDC"]);
+  });
+
+  it("bulkRotateKeys rotates keys for each account", async () => {
+    mockRotateAccountKey.mockResolvedValue(ok("ROTATE_XDR"));
+    const report = await bulkRotateKeys({
+      horizonUrl: HORIZON_URL,
+      networkConfig: NETWORK_CONFIG,
+      accounts: [
+        { id: "r1", account: "GA", oldKey: "OLD1", newKey: "NEW1" },
+        { id: "r2", account: "GB", oldKey: "OLD2", newKey: "NEW2" },
+      ],
+      config: { concurrency: 2 },
+    });
+
+    expect(mockRotateAccountKey).toHaveBeenCalledTimes(2);
+    expect(report.summary.succeeded).toBe(2);
+    const byId = new Map(report.results.map((r) => [r.id, r]));
+    expect(byId.get("r1")!.data!.xdr).toBe("ROTATE_XDR");
+  });
+
+  it("bulkSendPayments preserves successes when a build fails", async () => {
+    mockBuildPaymentTransaction
+      .mockResolvedValueOnce(ok("PAYMENT_XDR"))
+      .mockResolvedValueOnce(
+        err(SorokitErrorCode.TX_BUILD_FAILED, "invalid payment"),
+      );
+
+    const report = await bulkSendPayments({
+      horizonUrl: HORIZON_URL,
+      networkConfig: NETWORK_CONFIG,
+      transactions: [
+        { id: "p1", source: "GA", params: { destination: "GB", amount: "1" } },
+        { id: "p2", source: "GC", params: { destination: "GD", amount: "2" } },
+      ],
+      config: { concurrency: 2 },
+    });
+
+    expect(report.summary.succeeded).toBe(1);
+    expect(report.summary.failed).toBe(1);
+    const byId = new Map(report.results.map((r) => [r.id, r]));
+    expect(byId.get("p1")!.status).toBe("success");
+    expect(byId.get("p2")!.status).toBe("failed");
+    expect(byId.get("p2")!.errorCode).toBe(SorokitErrorCode.TX_BUILD_FAILED);
+  });
+});
diff --git a/src/tests/contractMetadataCompatibility.test.ts b/src/tests/contractMetadataCompatibility.test.ts
new file mode 100644
index 0000000..70e18b9
--- /dev/null
+++ b/src/tests/contractMetadataCompatibility.test.ts
@@ -0,0 +1,308 @@
+/**
+ * Tests for contract metadata versioning & compatibility validation (fix.md).
+ *
+ * Covers identical, compatible, incompatible, and missing metadata versions,
+ * deterministic fingerprints, stale-metadata detection, migration hooks, and
+ * stale-cache invalidation.
+ */
+
+import { describe, it, expect } from "vitest";
+import type { SorokitCache } from "../shared/cache";
+import type { ContractMethod } from "../soroban/types";
+import type { ContractSchema } from "../soroban/contractMetadata";
+import {
+  computeContractMetadataFingerprint,
+  buildContractMetadataSnapshot,
+  checkContractMetadataCompatibility,
+  checkStaleContractMetadata,
+  applyContractMetadataMigration,
+  invalidateContractMetadataForIncompatibility,
+  type ContractMetadataSnapshot,
+} from "../soroban/contractMetadataCompatibility";
+
+const CONTRACT_ID = "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+
+function methods(overrides: Array<Partial<ContractMethod>>): ContractMethod[] {
+  return overrides.map((m) => ({
+    name: m.name ?? "default",
+    inputs: m.inputs ?? [{ name: "value", type: "u32" }],
+    returnType: m.returnType ?? "u32",
+    ...(m.visibility !== undefined ? { visibility: m.visibility } : {}),
+  }));
+}
+
+const BASE_METHODS = methods([
+  { name: "transfer" },
+  { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" },
+]);
+
+function snapshot(contractId = CONTRACT_ID, ms = BASE_METHODS, version?: string): ContractMetadataSnapshot {
+  return buildContractMetadataSnapshot({ contractId, methods: ms, ...(version ? { version } : {}) });
+}
+
+describe("computeContractMetadataFingerprint", () => {
+  it("is deterministic for identical interfaces", () => {
+    const a = computeContractMetadataFingerprint(BASE_METHODS);
+    const b = computeContractMetadataFingerprint(BASE_METHODS);
+    expect(a).toBe(b);
+  });
+
+  it("is identical regardless of method ordering", () => {
+    const reversed = methods([
+      { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" },
+      { name: "transfer" },
+    ]);
+    expect(computeContractMetadataFingerprint(reversed)).toBe(
+      computeContractMetadataFingerprint(BASE_METHODS),
+    );
+  });
+
+  it("differs when a method signature changes", () => {
+    const changed = methods([
+      { name: "transfer", inputs: [{ name: "value", type: "u64" }] },
+      { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" },
+    ]);
+    expect(computeContractMetadataFingerprint(changed)).not.toBe(
+      computeContractMetadataFingerprint(BASE_METHODS),
+    );
+  });
+
+  it("accepts a typed ContractSchema", () => {
+    const schema: ContractSchema = {
+      contractId: CONTRACT_ID,
+      methods: [
+        { name: "transfer", params: [{ name: "value", type: "u32" }], returnType: "u32" },
+      ],
+    };
+    expect(computeContractMetadataFingerprint(schema)).toMatch(/^[0-9a-f]{8}$/);
+  });
+
+  it("does not rely on semantic version strings", () => {
+    // Two different versions with the same interface fingerprint match.
+    const v1 = snapshot(CONTRACT_ID, BASE_METHODS, "1.0.0");
+    const v2 = snapshot(CONTRACT_ID, BASE_METHODS, "2.0.0");
+    expect(v1.fingerprint).toBe(v2.fingerprint);
+  });
+});
+
+describe("buildContractMetadataSnapshot (#2 — associated with contract id)", () => {
+  it("associates metadata with a contract identifier", () => {
+    const snap = buildContractMetadataSnapshot({
+      contractId: CONTRACT_ID,
+      methods: BASE_METHODS,
+      version: "3.1.0",
+    });
+    expect(snap.contractId).toBe(CONTRACT_ID);
+    expect(snap.version).toBe("3.1.0");
+    expect(snap.fingerprint).toBe(computeContractMetadataFingerprint(BASE_METHODS));
+  });
+});
+
+describe("checkContractMetadataCompatibility", () => {
+  it("reports identical when fingerprints match (#identical)", () => {
+    const report = checkContractMetadataCompatibility({
+      baseline: snapshot(),
+      candidate: snapshot(),
+    });
+    expect(report.status).toBe("identical");
+    expect(report.compatible).toBe(true);
+    expect(report.changes).toEqual([]);
+    expect(report.shouldInvalidate).toBe(false);
+  });
+
+  it("reports compatible for non-breaking changes (added method)", () => {
+    const added = methods([
+      ...BASE_METHODS,
+      { name: "allowance" },
+    ]);
+    const report = checkContractMetadataCompatibility({
+      baseline: snapshot(),
+      candidate: snapshot(CONTRACT_ID, added),
+    });
+    expect(report.status).toBe("compatible");
+    expect(report.compatible).toBe(true);
+    expect(report.changes).toHaveLength(1);
+    expect(report.changes[0].kind).toBe("ADDED_METHOD");
+    expect(report.changes[0].breaking).toBe(false);
+    expect(report.shouldInvalidate).toBe(false);
+  });
+
+  it("reports incompatible for breaking changes (removed method)", () => {
+    const removed = methods([{ name: "transfer" }]);
+    const report = checkContractMetadataCompatibility({
+      baseline: snapshot(),
+      candidate: snapshot(CONTRACT_ID, removed),
+    });
+    expect(report.status).toBe("incompatible");
+    expect(report.compatible).toBe(false);
+    expect(report.errors.length).toBeGreaterThan(0);
+    expect(report.changes.some((c) => c.kind === "REMOVED_METHOD")).toBe(true);
+    expect(report.shouldInvalidate).toBe(true);
+  });
+
+  it("reports incompatible when an argument type changes", () => {
+    const changed = methods([
+      { name: "transfer", inputs: [{ name: "value", type: "u64" }] },
+      { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" },
+    ]);
+    const report = checkContractMetadataCompatibility({
+      baseline: snapshot(),
+      candidate: snapshot(CONTRACT_ID, changed),
+    });
+    expect(report.status).toBe("incompatible");
+    expect(report.changes.some((c) => c.kind === "CHANGED_ARG_TYPE")).toBe(true);
+  });
+
+  it("reports missing when no candidate and no expected fingerprint is provided", () => {
+    const report = checkContractMetadataCompatibility({
+      baseline: snapshot(),
+    });
+    expect(report.status).toBe("missing");
+    expect(report.warnings.length).toBeGreaterThan(0);
+    expect(report.shouldInvalidate).toBe(false);
+  });
+
+  it("reports identical against an expected fingerprint", () => {
+    const baseline = snapshot();
+    const report = checkContractMetadataCompatibility({
+      baseline,
+      expectedFingerprint: baseline.fingerprint,
+    });
+    expect(report.status).toBe("identical");
+    expect(report.compatible).toBe(true);
+  });
+
+  it("reports incompatible when cached fingerprint mismatches expected fingerprint", () => {
+    const baseline = snapshot();
+    const report = checkContractMetadataCompatibility({
+      baseline,
+      expectedFingerprint: computeContractMetadataFingerprint(
+        methods([{ name: "entirely_different" }]),
+      ),
+    });
+    expect(report.status).toBe("incompatible");
+    expect(report.shouldInvalidate).toBe(true);
+  });
+
+  it("treats matching advisory versions as identical even if fingerprints differ", () => {
+    const a = snapshot(CONTRACT_ID, BASE_METHODS, "1.0.0");
+    const b = snapshot(CONTRACT_ID, methods([{ name: "transfer", inputs: [{ name: "x", type: "u64" }] }]), "1.0.0");
+    const report = checkContractMetadataCompatibility({
+      baseline: a,
+      candidate: b,
+    });
+    expect(report.status).toBe("identical");
+    expect(report.compatible).toBe(true);
+  });
+});
+
+describe("checkStaleContractMetadata (#4 — detected before invocation)", () => {
+  it("reports current when cached metadata matches fresh metadata", async () => {
+    const result = await checkStaleContractMetadata({
+      contractId: CONTRACT_ID,
+      cachedMetadata: BASE_METHODS,
+      freshMetadata: BASE_METHODS,
+    });
+    expect(result.status).toBe("ok");
+    if (result.status === "ok") {
+      expect(result.data.current).toBe(true);
+      expect(result.data.report.status).toBe("identical");
+    }
+  });
+
+  it("reports stale when cached metadata no longer matches", async () => {
+    const changed = methods([
+      { name: "transfer", inputs: [{ name: "value", type: "u128" }] },
+      { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" },
+    ]);
+    const result = await checkStaleContractMetadata({
+      contractId: CONTRACT_ID,
+      cachedMetadata: BASE_METHODS,
+      freshMetadata: changed,
+    });
+    expect(result.status).toBe("ok");
+    if (result.status === "ok") {
+      expect(result.data.current).toBe(false);
+      expect(result.data.report.status).toBe("incompatible");
+      expect(result.data.message.length).toBeGreaterThan(0);
+    }
+  });
+
+  it("reports missing when no cached metadata exists", async () => {
+    const result = await checkStaleContractMetadata({
+      contractId: CONTRACT_ID,
+      freshMetadata: BASE_METHODS,
+    });
+    expect(result.status).toBe("ok");
+    if (result.status === "ok") {
+      expect(result.data.current).toBe(false);
+      expect(result.data.report.status).toBe("missing");
+    }
+  });
+
+  it("returns current (non-fatal) when no reference is available", async () => {
+    const result = await checkStaleContractMetadata({
+      contractId: CONTRACT_ID,
+      cachedMetadata: BASE_METHODS,
+    });
+    expect(result.status).toBe("ok");
+    if (result.status === "ok") {
+      expect(result.data.current).toBe(true);
+      expect(result.data.report.status).toBe("missing");
+    }
+  });
+});
+
+describe("applyContractMetadataMigration (#6 — migration hooks)", () => {
+  it("applies the first hook that produces a snapshot", () => {
+    const candidate = snapshot(CONTRACT_ID, methods([{ name: "new_api" }]), "2.0.0");
+    const migrated: ContractMetadataSnapshot = buildContractMetadataSnapshot({
+      contractId: CONTRACT_ID,
+      methods: methods([{ name: "legacy_api" }]),
+    });
+    const result = applyContractMetadataMigration(candidate, [
+      () => undefined, // no change
+      () => migrated,
+    ]);
+    expect(result.blocked).toBe(false);
+    expect(result.snapshot).toBe(migrated);
+    expect(result.migrations).toHaveLength(2);
+    expect(result.migrations[0].applied).toBe(false);
+    expect(result.migrations[1].applied).toBe(true);
+  });
+
+  it("blocks when no hook can migrate the metadata", () => {
+    const candidate = snapshot(CONTRACT_ID, methods([{ name: "new_api" }]));
+    const result = applyContractMetadataMigration(candidate, [
+      () => undefined,
+    ]);
+    expect(result.blocked).toBe(true);
+    expect(result.snapshot).toBeNull();
+  });
+});
+
+describe("invalidateContractMetadataForIncompatibility (#7)", () => {
+  it("invalidates both metadata and schema caches for the contract", () => {
+    const invalidated: string[] = [];
+    const cache: SorokitCache = {
+      get: () => undefined,
+      set: () => undefined,
+      invalidate: (key) => {
+        invalidated.push(key);
+      },
+    };
+    let resetCalled = false;
+
+    invalidateContractMetadataForIncompatibility({
+      contractId: CONTRACT_ID,
+      cache,
+      reset: () => {
+        resetCalled = true;
+      },
+    });
+
+    expect(resetCalled).toBe(true);
+    expect(invalidated).toContain(`sorokit:contract-metadata:${CONTRACT_ID}`);
+    expect(invalidated).toContain(`sorokit:contract-schema:${CONTRACT_ID}`);
+  });
+});
diff --git a/src/tests/contractStateOptimization.test.ts b/src/tests/contractStateOptimization.test.ts
new file mode 100644
index 0000000..1b0e872
--- /dev/null
+++ b/src/tests/contractStateOptimization.test.ts
@@ -0,0 +1,323 @@
+/**
+ * Tests for contract state optimization (#514):
+ * compressContractState / decompressContractState, along with measurement and
+ * benchmark helpers.
+ *
+ * Covers: primitives, structured values, repetitive data, empty and oversized
+ * payloads, unsupported-value safety, no caller-state mutation, deterministic
+ * serialization, metadata correctness, round-trip integrity, and performance.
+ */
+
+import { describe, it, expect } from "vitest";
+import { SorokitErrorCode } from "../shared/response";
+import {
+  compressContractState,
+  decompressContractState,
+  measureContractState,
+  benchmarkContractState,
+} from "../soroban/contractStateOptimization";
+import type { ContractStateMetadata } from "../soroban/contractStateOptimization";
+
+function roundTrip(state: Record<string, unknown>, options?: Parameters<typeof compressContractState>[1]) {
+  const compressed = compressContractState(state, options);
+  if (compressed.status === "error") return compressed;
+  const decompressed = decompressContractState(compressed.data.data, compressed.data.metadata);
+  return decompressed;
+}
+
+function expectOk<T>(result: { status: string; data?: T }): T {
+  expect(result.status).toBe("ok");
+  return result.data as T;
+}
+
+const REPETITIVE_STATE: Record<string, unknown> = {};
+for (let i = 0; i < 200; i++) {
+  REPETITIVE_STATE[`entry_${i}`] = {
+    token: "USDC:GAUY4MS4VJXWU4G7XZQ6YQYQ6YQYQ6YQYQ6YQYQ6YQYQ6YQYQ6YQYQ6YQ",
+    amount: "1000000",
+    owner: "GA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVSGZ",
+    active: true,
+  };
+}
+
+describe("compressContractState / decompressContractState", () => {
+  it("round-trips a primitive-only state", () => {
+    const state = {
+      counter: 42,
+      decimal: 3.5,
+      big: 123456789012345678901234567890n,
+      enabled: true,
+      disabled: false,
+      empty: null,
+      name: "hello",
+    };
+    const result = roundTrip(state);
+    expect(result.status).toBe("ok");
+    expect(result.data).toEqual(state);
+  });
+
+  it("round-trips nested structured state", () => {
+    const state = {
+      config: { fee: 100n, pct: 0.05, label: "swap" },
+      tags: ["a", "b", "c"],
+      matrix: [
+        [1, 2],
+        [3, 4],
+      ],
+    };
+    const result = roundTrip(state);
+    expect(result.status).toBe("ok");
+    expect(result.data).toEqual(state);
+  });
+
+  it("round-trips byte array values", () => {
+    const state = { blob: new Uint8Array([1, 2, 3, 255]), nested: { raw: new Uint8Array([10]) } };
+    const result = roundTrip(state);
+    expect(result.status).toBe("ok");
+    expect(result.data?.blob).toBeInstanceOf(Uint8Array);
+    expect(Array.from(result.data!.blob as Uint8Array)).toEqual([1, 2, 3, 255]);
+    expect(Array.from((result.data!.nested as { raw: Uint8Array }).raw)).toEqual([10]);
+  });
+
+  it("restores object key order independently (logical equality)", () => {
+    const a = roundTrip({ top: { x: 1, y: 2 } });
+    const b = roundTrip({ top: { y: 2, x: 1 } });
+    expect(a.status).toBe("ok");
+    expect(b.status).toBe("ok");
+    expect(a.data).toEqual(b.data);
+    expect(a.data).toEqual({ top: { x: 1, y: 2 } });
+  });
+
+  it("round-trips an empty state object", () => {
+    const result = roundTrip({});
+    expect(result.status).toBe("ok");
+    expect(result.data).toEqual({});
+    expect(expectOk(compressContractState({}))?.metadata.entries).toBe(0);
+  });
+
+  it("round-trips a large repetitive payload", () => {
+    const result = roundTrip(REPETITIVE_STATE);
+    expect(result.status).toBe("ok");
+    expect(Object.keys(result.data!).length).toBe(200);
+    expect(result.data).toEqual(REPETITIVE_STATE);
+  });
+
+  it("randomized large state round-trips exactly", () => {
+    const state: Record<string, unknown> = { seed: 1n, list: [], nested: {} };
+    for (let i = 0; i < 500; i++) {
+      state[`k${i}`] = i % 2 === 0 ? { v: i, s: `str-${i}` } : i;
+    }
+    const result = roundTrip(state);
+    expect(result.status).toBe("ok");
+    expect(result.data).toEqual(state);
+  });
+});
+
+describe("metadata and measurement", () => {
+  it("records encoding and compression strategy", () => {
+    const metadata = expectOk(compressContractState({ a: 1 }))?.metadata;
+    expect(metadata.format).toBe("contract-state-optimized");
+    expect(metadata.version).toBe(1);
+    expect(metadata.encoding).toBe("tagged");
+    expect(metadata.compression).toMatch(/^(none|gzip)$/);
+    expect(metadata.entries).toBe(1);
+    expect(metadata.hash).toMatch(/^[0-9a-f]{8}$/);
+  });
+
+  it("reports size savings and ratio", () => {
+    const compressed = expectOk(compressContractState(REPETITIVE_STATE, { compression: "auto" }));
+    const m = compressed.metadata;
+    expect(m.originalBytes).toBeGreaterThan(0);
+    expect(m.encodedBytes).toBeGreaterThan(0);
+    expect(m.finalBytes).toBeGreaterThan(0);
+    expect(m.ratio).toBeGreaterThan(0);
+    expect(m.ratio).toBeLessThanOrEqual(1);
+    expect(m.savingsPercent).toBeGreaterThanOrEqual(0);
+    expect(m.savingsPercent).toBeLessThanOrEqual(100);
+  });
+
+  it("compresses eligible repetitive payloads (ratio < 1)", () => {
+    const compressed = expectOk(compressContractState(REPETITIVE_STATE));
+    expect(compressed.metadata.compression).toBe("gzip");
+    expect(compressed.metadata.savingsPercent).toBeGreaterThan(0);
+  });
+
+  it("leaves tiny payloads uncompressed under auto", () => {
+    const compressed = expectOk(compressContractState({ a: 1 }, { compression: "auto" }));
+    expect(compressed.metadata.compression).toBe("none");
+    expect(compressed.metadata.encodedBytes).toBe(compressed.metadata.finalBytes);
+  });
+
+  it("honors the minCompressBytes threshold", () => {
+    const compressed = expectOk(
+      compressContractState(REPETITIVE_STATE, { compression: "auto", minCompressBytes: 10_000_000 }),
+    );
+    expect(compressed.metadata.compression).toBe("none");
+  });
+
+  it("can be forced to never compress", () => {
+    const compressed = expectOk(compressContractState(REPETITIVE_STATE, { compression: "none" }));
+    expect(compressed.metadata.compression).toBe("none");
+    expect(compressed.metadata.finalBytes).toBe(compressed.metadata.encodedBytes);
+    // Still decodable.
+    const decoded = decompressContractState(compressed.data, compressed.metadata);
+    expect(decoded.status).toBe("ok");
+    expect(decoded.data).toEqual(REPETITIVE_STATE);
+  });
+
+  it("measureContractState reports sizes without writing state", () => {
+    const clean = expectOk(measureContractState(REPETITIVE_STATE));
+    expect(clean.originalBytes).toBeGreaterThan(0);
+    expect(clean.encodedBytes).toBeGreaterThan(0);
+    expect(clean.compressedBytes).toBeGreaterThan(0);
+    expect(clean.ratio).toBeGreaterThanOrEqual(0);
+    expect(clean.savingsPercent).toBeGreaterThanOrEqual(0);
+  });
+});
+
+describe("determinism", () => {
+  it("produces identical encoded bytes for the same logical state", () => {
+    const a = expectOk(compressContractState({ b: 2, a: { x: 1, y: 2 } }, { compression: "none" }));
+    const b = expectOk(compressContractState({ a: { y: 2, x: 1 }, b: 2 }, { compression: "none" }));
+    expect(Array.from(a.data)).toEqual(Array.from(b.data));
+    expect(a.metadata.hash).toBe(b.metadata.hash);
+  });
+
+  it("python-style deterministic gzip output for identical payloads", () => {
+    const state = { msg: "hello ".repeat(50) };
+    const a = expectOk(compressContractState(state));
+    const b = expectOk(compressContractState(state));
+    expect(a.metadata.compression).toBe("gzip");
+    expect(Array.from(a.data)).toEqual(Array.from(b.data));
+  });
+
+  it("changes when a value changes", () => {
+    const a = expectOk(compressContractState({ v: 1 }, { compression: "none" }));
+    const b = expectOk(compressContractState({ v: 2 }, { compression: "none" }));
+    expect(a.metadata.hash).not.toBe(b.metadata.hash);
+  });
+});
+
+describe("safety and error handling", () => {
+  it.each([
+    ["function", { fn: () => 1 }],
+    ["symbol", { sym: Symbol("x") }],
+    ["non-finite number", { n: NaN }],
+    ["infinity", { n: Infinity }],
+    ["undefined value", { u: undefined }],
+  ])("fails safely on unsupported value: %s", (_label, state) => {
+    const result = compressContractState(state);
+    expect(result.status).toBe("error");
+    expect(result.error?.code).toBe(SorokitErrorCode.INVALID_CONFIG);
+  });
+
+  it("rejects circular references", () => {
+    const a: Record<string, unknown> = { name: "self" };
+    a.self = a;
+    const result = compressContractState(a);
+    expect(result.status).toBe("error");
+    expect(result.error?.message).toContain("circular");
+  });
+
+  it("rejects a non-object state root", () => {
+    const result = compressContractState([] as unknown as Record<string, unknown>);
+    expect(result.status).toBe("error");
+  });
+
+  it("fails safely when decompress gets corrupt bytes", () => {
+    const good = expectOk(compressContractState({ a: 1 }, { compression: "none" }));
+    const bad = good.data.slice();
+    bad[0] = 0xff;
+    // Corrupting the first byte changes the tag -> decode error or fingerprint mismatch.
+    const result = decompressContractState(bad, good.metadata);
+    expect(result.status).toBe("error");
+  });
+
+  it("fails safely on fingerprint mismatch", () => {
+    const compressed = expectOk(compressContractState({ a: 1 }, { compression: "none" }));
+    const tamperedMeta: ContractStateMetadata = { ...compressed.metadata, hash: "deadbeef" };
+    const result = decompressContractState(compressed.data, tamperedMeta);
+    expect(result.status).toBe("error");
+    expect(result.error?.code).toBe(SorokitErrorCode.CONTRACT_READ_FAILED);
+  });
+
+  it("rejects a payload with unsupported format or version", () => {
+    const compressed = expectOk(compressContractState({ a: 1 }, { compression: "none" }));
+    expect(
+      decompressContractState(compressed.data, { ...compressed.metadata, format: "other" as never })
+        .status,
+    ).toBe("error");
+    expect(
+      decompressContractState(compressed.data, {
+        ...compressed.metadata,
+        version: 99 as unknown as 1,
+      }).status,
+    ).toBe("error");
+    expect(
+      decompressContractState(compressed.data, {
+        ...compressed.metadata,
+        compression: "lz4" as never,
+      }).status,
+    ).toBe("error");
+  });
+});
+
+describe("no mutation of caller-owned state", () => {
+  it("does not mutate the input state or nested values", () => {
+    const nested = { keep: 1n, tags: ["x"] };
+    const state: Record<string, unknown> = { a: 1, nested, bytes: new Uint8Array([1, 2]) };
+
+    compressContractState(state, { compression: "auto" });
+
+    expect(state.a).toBe(1);
+    expect(state.nested).toBe(nested);
+    expect(nested.keep).toBe(1n);
+    expect(nested.tags).toEqual(["x"]);
+    expect(Array.from(state.bytes as Uint8Array)).toEqual([1, 2]);
+  });
+
+  it("does not retain references to input objects in the encoded form", () => {
+    const state: Record<string, unknown> = { a: { inner: 1 } };
+    const compressed = expectOk(compressContractState(state, { compression: "none" }));
+    // Mutating the source after encoding must not affect the stored bytes.
+    (state.a as { inner: number }).inner = 999;
+    const decoded = decompressContractState(compressed.data, compressed.metadata);
+    expect(decoded.status).toBe("ok");
+    expect(decoded.data).toEqual({ a: { inner: 1 } });
+  });
+
+  it("does not mutate the metadata object passed to decompress", () => {
+    const compressed = expectOk(compressContractState({ a: 1 }));
+    const meta = compressed.metadata;
+    const before = JSON.stringify(meta);
+    decompressContractState(compressed.data, meta);
+    expect(JSON.stringify(meta)).toBe(before);
+  });
+});
+
+describe("performance", () => {
+  it("benchmarks compress/decompress for a representative payload", () => {
+    const bench = expectOk(benchmarkContractState(REPETITIVE_STATE, 50));
+    expect(bench.iterations).toBe(50);
+    expect(bench.compressMs).toBeGreaterThanOrEqual(0);
+    expect(bench.decompressMs).toBeGreaterThanOrEqual(0);
+    expect(bench.compressMeanMs).toBeGreaterThan(0);
+    expect(bench.decompressMeanMs).toBeGreaterThanOrEqual(0);
+    expect(bench.payloadBytes).toBeGreaterThan(0);
+    expect(bench.ratio).toBeGreaterThan(0);
+  });
+
+  it("benchmarks a large primitive payload too", () => {
+    const state: Record<string, unknown> = {};
+    for (let i = 0; i < 1000; i++) state[`v${i}`] = i * 7;
+    const bench = expectOk(benchmarkContractState(state, 20));
+    expect(bench.compressMeanMs).toBeGreaterThan(0);
+    expect(bench.totalMs).toBeGreaterThanOrEqual(bench.compressMs);
+  });
+
+  it("rejects an invalid iteration count", () => {
+    const result = benchmarkContractState({ a: 1 }, 0);
+    expect(result.status).toBe("error");
+    expect(result.error?.code).toBe(SorokitErrorCode.INVALID_CONFIG);
+  });
+});
diff --git a/src/tests/paymentNotifications.test.ts b/src/tests/paymentNotifications.test.ts
new file mode 100644
index 0000000..5cf1ef7
--- /dev/null
+++ b/src/tests/paymentNotifications.test.ts
@@ -0,0 +1,427 @@
+/**
+ * Tests for the payment notification subsystem (fix.md).
+ *
+ * Covers successful delivery, retries (exponential backoff), configurable
+ * timeouts, duplicate-delivery identification through event IDs, failed
+ * endpoints surfacing errors, and decoupled notification channels.
+ */
+
+import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
+import {
+  registerPaymentWebhook,
+  unregisterPaymentWebhook,
+  listPaymentWebhooks,
+  clearPaymentWebhooks,
+  triggerPaymentNotifications,
+  dispatchPaymentNotification,
+  generatePaymentEventId,
+  isPaymentNotificationEvent,
+  PAYMENT_NOTIFICATION_EVENTS,
+  type PaymentNotificationChannel,
+  type PaymentNotificationEvent,
+} from "../transaction/paymentNotifications";
+
+const SECRET_URL = "https://pay.example.com/webhook";
+
+function okResponse(): Response {
+  return { ok: true, status: 200, statusText: "OK" } as Response;
+}
+
+describe("paymentNotifications", () => {
+  beforeEach(() => {
+    clearPaymentWebhooks();
+  });
+
+  afterEach(() => {
+    clearPaymentWebhooks();
+    vi.restoreAllMocks();
+    vi.clearAllMocks();
+  });
+
+  describe("event types", () => {
+    it("defines the supported payment events", () => {
+      expect(PAYMENT_NOTIFICATION_EVENTS).toEqual([
+        "payment_received",
+        "payment_sent",
+        "payment_failed",
+        "payment_confirmed",
+      ]);
+    });
+
+    it("guards payment event types", () => {
+      expect(isPaymentNotificationEvent("payment_received")).toBe(true);
+      expect(isPaymentNotificationEvent("payment_confirmed")).toBe(true);
+      expect(isPaymentNotificationEvent("tx_confirmed")).toBe(false);
+      expect(isPaymentNotificationEvent("payment_failed")).toBe(true);
+    });
+  });
+
+  describe("registerPaymentWebhook", () => {
+    it("registers a URL for multiple payment events", () => {
+      const result = registerPaymentWebhook(SECRET_URL, [
+        "payment_received",
+        "payment_sent",
+      ]);
+      expect(result.status).toBe("ok");
+      expect(listPaymentWebhooks("payment_received")).toHaveLength(1);
+      expect(listPaymentWebhooks("payment_sent")).toHaveLength(1);
+      expect(listPaymentWebhooks("payment_failed")).toHaveLength(0);
+    });
+
+    it("rejects an empty events array", () => {
+      const result = registerPaymentWebhook(SECRET_URL, []);
+      expect(result.status).toBe("error");
+      expect(result.error?.code).toBe("INVALID_CONFIG");
+    });
+
+    it("rejects an invalid event type", () => {
+      const result = registerPaymentWebhook(SECRET_URL, [
+        "payment_invalid" as PaymentNotificationEvent,
+      ]);
+      expect(result.status).toBe("error");
+      expect(result.error?.code).toBe("INVALID_CONFIG");
+      expect(listPaymentWebhooks("payment_received")).toHaveLength(0);
+    });
+
+    it("rejects an invalid URL", () => {
+      const result = registerPaymentWebhook("not-a-url", ["payment_received"]);
+      expect(result.status).toBe("error");
+      expect(result.error?.code).toBe("INVALID_CONFIG");
+    });
+
+    it("rejects an invalid maxRetries", () => {
+      const result = registerPaymentWebhook(SECRET_URL, ["payment_received"], {
+        maxRetries: -1,
+      });
+      expect(result.status).toBe("error");
+      expect(result.error?.code).toBe("INVALID_CONFIG");
+    });
+
+    it("rejects an invalid timeoutMs", () => {
+      const result = registerPaymentWebhook(SECRET_URL, ["payment_received"], {
+        timeoutMs: 0,
+      });
+      expect(result.status).toBe("error");
+      expect(result.error?.code).toBe("INVALID_CONFIG");
+    });
+
+    it("does not duplicate registrations for repeated events", () => {
+      const result = registerPaymentWebhook(SECRET_URL, [
+        "payment_received",
+        "payment_received",
+      ]);
+      expect(result.status).toBe("ok");
+      expect(listPaymentWebhooks("payment_received")).toHaveLength(1);
+    });
+
+    it("unregisters a specific event", () => {
+      registerPaymentWebhook(SECRET_URL, ["payment_received", "payment_sent"]);
+      const result = unregisterPaymentWebhook(SECRET_URL, "payment_received");
+      expect(result.status).toBe("ok");
+      expect(listPaymentWebhooks("payment_received")).toHaveLength(0);
+      expect(listPaymentWebhooks("payment_sent")).toHaveLength(1);
+    });
+
+    it("fails to unregister a nonexistent webhook", () => {
+      const result = unregisterPaymentWebhook(SECRET_URL, "payment_received");
+      expect(result.status).toBe("error");
+      expect(result.error?.code).toBe("INVALID_CONFIG");
+    });
+  });
+
+  describe("event IDs (idempotency)", () => {
+    it("derives the same event ID for the same payment identity", () => {
+      const a = generatePaymentEventId("tx-1", "payment_received", "GA001");
+      const b = generatePaymentEventId("tx-1", "payment_received", "GA001");
+      expect(a).toBe(b);
+    });
+
+    it("derives distinct event IDs for distinct payments", () => {
+      const a = generatePaymentEventId("tx-1", "payment_received", "GA001");
+      const b = generatePaymentEventId("tx-2", "payment_received", "GA001");
+      const c = generatePaymentEventId("tx-1", "payment_sent", "GA001");
+      expect(a).not.toBe(b);
+      expect(a).not.toBe(c);
+    });
+
+    it("delivers identical event ID on duplicate deliveries", async () => {
+      global.fetch = vi.fn(() => Promise.resolve(okResponse()));
+      registerPaymentWebhook(SECRET_URL, ["payment_received"]);
+
+      const input = {
+        transactionId: "tx-dup",
+        account: "GA001",
+      };
+      await triggerPaymentNotifications("payment_received", input);
+      await triggerPaymentNotifications("payment_received", input);
+
+      const fetchMock = global.fetch as ReturnType<typeof vi.fn>;
+      const firstBody = JSON.parse(fetchMock.mock.calls[0][1].body);
+      const secondBody = JSON.parse(fetchMock.mock.calls[1][1].body);
+      expect(firstBody.eventId).toBe(secondBody.eventId);
+      expect(firstBody.eventId).toBe(
+        generatePaymentEventId("tx-dup", "payment_received", "GA001"),
+      );
+    });
+
+    it("sends the event ID as an idempotency header", async () => {
+      global.fetch = vi.fn(() => Promise.resolve(okResponse()));
+      registerPaymentWebhook(SECRET_URL, ["payment_received"]);
+
+      await triggerPaymentNotifications("payment_received", {
+        transactionId: "tx-h",
+        account: "GA001",
+      });
+
+      const [, init] = (global.fetch as ReturnType<typeof vi.fn>)
+        .mock.calls[0] as [string, { headers: Record<string, string>; body: string }];
+      const payload = JSON.parse(init.body);
+      expect(init.headers["Idempotency-Key"]).toBe(payload.eventId);
+      expect(init.headers["X-Sorokit-Event-Id"]).toBe(payload.eventId);
+    });
+  });
+
+  describe("payload shape", () => {
+    it("includes event ID, timestamp, transaction ID, account, and event type", async () => {
+      const fetchMock = vi.fn(() => Promise.resolve(okResponse()));
+      global.fetch = fetchMock as typeof fetch;
+      registerPaymentWebhook(SECRET_URL, ["payment_confirmed"]);
+
+      const results = await triggerPaymentNotifications("payment_confirmed", {
+        transactionId: "tx-payload",
+        account: "GA100",
+      });
+      expect(results).toHaveLength(1);
+      expect(results[0].status).toBe("ok");
+
+      const [, init] = fetchMock.mock.calls[0] as [string, { body: string }];
+      const payload = JSON.parse(init.body);
+      expect(typeof payload.eventId).toBe("string");
+      expect(payload.event).toBe("payment_confirmed");
+      expect(typeof payload.timestamp).toBe("string");
+      expect(new Date(payload.timestamp).toString()).not.toBe("Invalid Date");
+      expect(payload.transactionId).toBe("tx-payload");
+      expect(payload.account).toBe("GA100");
+    });
+  });
+
+  describe("delivery", () => {
+    it("successfully delivers to a healthy endpoint", async () => {
+      const fetchMock = vi.fn(() => Promise.resolve(okResponse()));
+      global.fetch = fetchMock as typeof fetch;
+      registerPaymentWebhook(SECRET_URL, ["payment_received"]);
+
+      const results = await triggerPaymentNotifications("payment_received", {
+        transactionId: "tx-ok",
+        account: "GA001",
+      });
+      expect(results).toHaveLength(1);
+      expect(results[0].status).toBe("ok");
+      expect(fetchMock).toHaveBeenCalledTimes(1);
+    });
+
+    it("does not deliver to endpoints not subscribed to the event", async () => {
+      const fetchMock = vi.fn(() => Promise.resolve(okResponse()));
+      global.fetch = fetchMock as typeof fetch;
+      registerPaymentWebhook(SECRET_URL, ["payment_received"]);
+
+      const results = await triggerPaymentNotifications("payment_sent", {
+        transactionId: "tx-no",
+        account: "GA001",
+      });
+      expect(results).toHaveLength(0);
+      expect(fetchMock).not.toHaveBeenCalled();
+    });
+  });
+
+  describe("retries (exponential backoff)", () => {
+    it("retries a transient failure then succeeds", async () => {
+      let attempts = 0;
+      global.fetch = vi.fn(() => {
+        attempts++;
+        if (attempts < 3) return Promise.reject(new Error("flaky"));
+        return Promise.resolve(okResponse());
+      });
+      registerPaymentWebhook(SECRET_URL, ["payment_received"], {
+        maxRetries: 3,
+      });
+
+      const results = await triggerPaymentNotifications("payment_received", {
+        transactionId: "tx-retry",
+        account: "GA001",
+      });
+      expect(results[0].status).toBe("ok");
+      expect(attempts).toBe(3);
+    });
+
+    it("gives up after exhausting configured attempts", async () => {
+      global.fetch = vi.fn(() => Promise.reject(new Error("down")));
+      registerPaymentWebhook(SECRET_URL, ["payment_failed"], {
+        maxRetries: 2,
+      });
+
+      const results = await triggerPaymentNotifications("payment_failed", {
+        transactionId: "tx-fail",
+        account: "GA001",
+      });
+      expect(results[0].status).toBe("error");
+      expect(results[0].error?.code).toBe("NETWORK_ERROR");
+      expect((global.fetch as ReturnType<typeof vi.fn>)).toHaveBeenCalledTimes(3); // 1 + 2 retries
+    }, 20000);
+  });
+
+  describe("timeout", () => {
+    it("treats a slow endpoint as a failed attempt", async () => {
+      const fetchMock = vi.fn(
+        () =>
+          new Promise<Response>((_resolve, reject) => {
+            setTimeout(() => reject(new Error("timed out")), 50);
+          }),
+      );
+      global.fetch = fetchMock as typeof fetch;
+      registerPaymentWebhook(SECRET_URL, ["payment_received"], {
+        maxRetries: 0,
+        timeoutMs: 5,
+      });
+
+      const results = await triggerPaymentNotifications("payment_received", {
+        transactionId: "tx-timeout",
+        account: "GA001",
+      });
+      expect(results[0].status).toBe("error");
+      expect((global.fetch as ReturnType<typeof vi.fn>)).toHaveBeenCalledTimes(1);
+    }, 20000);
+  });
+
+  describe("failed endpoints", () => {
+    it("surfaces an HTTP error for a failing endpoint", async () => {
+      global.fetch = vi.fn(() =>
+        Promise.resolve({ ok: false, status: 500, statusText: "Internal" } as Response),
+      );
+      registerPaymentWebhook(SECRET_URL, ["payment_received"], {
+        maxRetries: 0,
+      });
+
+      const results = await triggerPaymentNotifications("payment_received", {
+        transactionId: "tx-500",
+        account: "GA001",
+      });
+      expect(results[0].status).toBe("error");
+      expect(results[0].error?.code).toBe("NETWORK_ERROR");
+    });
+
+    it("reports one failing endpoint without masking another", async () => {
+      const failingUrl = "https://fail.example.com/webhook";
+      const goodUrl = "https://good.example.com/webhook";
+      global.fetch = vi.fn((url: string) =>
+        Promise.resolve(
+          url === failingUrl
+            ? ({ ok: false, status: 500 } as Response)
+            : okResponse(),
+        ),
+      );
+      registerPaymentWebhook(failingUrl, ["payment_received"], {
+        maxRetries: 0,
+      });
+      registerPaymentWebhook(goodUrl, ["payment_received"], {
+        maxRetries: 0,
+      });
+
+      const results = await triggerPaymentNotifications("payment_received", {
+        transactionId: "tx-two",
+        account: "GA001",
+      });
+      expect(results).toHaveLength(2);
+      expect(results.some((r) => r.status === "error")).toBe(true);
+      expect(results.some((r) => r.status === "ok")).toBe(true);
+    });
+  });
+
+  describe("notification channels (decoupled)", () => {
+    it("delivers to registered channels alongside webhooks", async () => {
+      global.fetch = vi.fn(() => Promise.resolve(okResponse()));
+      const channel: PaymentNotificationChannel = {
+        name: "test-channel",
+        deliver: vi.fn(() => Promise.resolve(true)),
+      };
+      registerPaymentWebhook(SECRET_URL, ["payment_received"], {
+        channels: [channel],
+      });
+
+      const results = await triggerPaymentNotifications("payment_received", {
+        transactionId: "tx-channel",
+        account: "GA001",
+      });
+      expect(results).toHaveLength(2);
+      expect(results.every((r) => r.status === "ok")).toBe(true);
+      expect(channel.deliver).toHaveBeenCalledTimes(1);
+    });
+
+    it("surfaces channel failures without coupling to webhook logic", async () => {
+      global.fetch = vi.fn(() => Promise.resolve(okResponse()));
+      const channel: PaymentNotificationChannel = {
+        name: "failing-channel",
+        deliver: vi.fn(() => Promise.resolve(false)),
+      };
+      registerPaymentWebhook(SECRET_URL, ["payment_received"], {
+        channels: [channel],
+      });
+
+      const results = await triggerPaymentNotifications("payment_received", {
+        transactionId: "tx-chanfail",
+        account: "GA001",
+      });
+      expect(results).toHaveLength(2);
+      expect(results[0].status).toBe("ok"); // webhook ok
+      expect(results[1].status).toBe("error"); // channel rejected
+      expect(results[1].error?.code).toBe("NETWORK_ERROR");
+    });
+  });
+
+  describe("dispatchPaymentNotification", () => {
+    it("returns immediately without blocking on delivery", async () => {
+      let resolveFetch: (value: Response) => void = () => undefined;
+      global.fetch = vi.fn(
+        () =>
+          new Promise<Response>((resolve) => {
+            resolveFetch = resolve;
+          }),
+      );
+      registerPaymentWebhook(SECRET_URL, ["payment_received"]);
+
+      const before = Date.now();
+      dispatchPaymentNotification("payment_received", {
+        transactionId: "tx-fast",
+        account: "GA001",
+      });
+      const elapsed = Date.now() - before;
+      expect(elapsed).toBeLessThan(100);
+      resolveFetch(okResponse());
+    });
+
+    it("swallows delivery failures without unhandled rejections", async () => {
+      global.fetch = vi.fn(() => Promise.reject(new Error("down")));
+      registerPaymentWebhook(SECRET_URL, ["payment_received"], {
+        maxRetries: 0,
+      });
+
+      expect(() =>
+        dispatchPaymentNotification("payment_received", {
+          transactionId: "tx-swallow",
+          account: "GA001",
+        }),
+      ).not.toThrow();
+      await new Promise((r) => setTimeout(r, 10));
+    }, 20000);
+
+    it("is a no-op when nothing is subscribed", () => {
+      const fetchMock = vi.fn();
+      global.fetch = fetchMock as typeof fetch;
+      dispatchPaymentNotification("payment_received", {
+        transactionId: "tx-noop",
+        account: "GA001",
+      });
+      expect(fetchMock).not.toHaveBeenCalled();
+    });
+  });
+});
diff --git a/src/transaction/index.ts b/src/transaction/index.ts
index fbfcfce..f68ac5c 100644
--- a/src/transaction/index.ts
+++ b/src/transaction/index.ts
@@ -291,6 +291,27 @@ export {
   dispatchTransactionEvent,
   verifySignature,
 } from "./webhooks";
+
+// ─── Payment notifications (#fix.md) ──────────────────────────────────────────
+export {
+  registerPaymentWebhook,
+  unregisterPaymentWebhook,
+  listPaymentWebhooks,
+  clearPaymentWebhooks,
+  triggerPaymentNotifications,
+  dispatchPaymentNotification,
+  generatePaymentEventId,
+  isPaymentNotificationEvent,
+  PAYMENT_NOTIFICATION_EVENTS,
+} from "./paymentNotifications";
+export type {
+  PaymentNotificationEvent,
+  PaymentWebhookOptions,
+  PaymentWebhookPayload,
+  PaymentWebhookRegistration,
+  PaymentNotificationInput,
+  PaymentNotificationChannel,
+} from "./paymentNotifications";
 export type {
   WebhookEventType,
   TransactionWebhookEvent,
diff --git a/src/transaction/paymentNotifications.ts b/src/transaction/paymentNotifications.ts
new file mode 100644
index 0000000..c0c5789
--- /dev/null
+++ b/src/transaction/paymentNotifications.ts
@@ -0,0 +1,488 @@
+/**
+ * Payment notification subsystem.
+ *
+ * Lets applications subscribe a webhook endpoint (or an optional, decoupled
+ * notification channel) to payment lifecycle events such as payment_received,
+ * payment_sent, payment_failed, and payment_confirmed.
+ *
+ * Design goals (see fix.md):
+ *  - Transport-agnostic delivery: the payload contract and dispatch logic do
+ *    not depend on any particular HTTP client or channel implementation.
+ *  - Non-blocking: dispatch never blocks (or throws into) the transaction
+ *    processing path.
+ *  - Bounded retries: exponential backoff with a hard cap and a configurable
+ *    attempt limit so retry behavior cannot become an uncontrolled loop.
+ *  - Idempotency: every payload carries a deterministic event ID derived from
+ *    the payment identity so consumers can safely discard duplicate
+ *    deliveries.
+ *  - Clean error surfacing: registration and delivery failures are returned as
+ *    structured `SorokitResult` errors.
+ */
+
+import { ok, err, SorokitErrorCode } from "../shared/response";
+import type { SorokitResult } from "../shared/response";
+import { sleep } from "../shared/utils";
+
+/**
+ * Canonical payment lifecycle event types.
+ */
+export type PaymentNotificationEvent =
+  | "payment_received"
+  | "payment_sent"
+  | "payment_failed"
+  | "payment_confirmed";
+
+/**
+ * The full list of supported payment event types.
+ */
+export const PAYMENT_NOTIFICATION_EVENTS: readonly PaymentNotificationEvent[] = [
+  "payment_received",
+  "payment_sent",
+  "payment_failed",
+  "payment_confirmed",
+];
+
+/** Type guard for {@link PaymentNotificationEvent}. */
+export function isPaymentNotificationEvent(
+  value: unknown,
+): value is PaymentNotificationEvent {
+  return (
+    typeof value === "string" &&
+    (PAYMENT_NOTIFICATION_EVENTS as readonly string[]).includes(value)
+  );
+}
+
+/**
+ * Delivery options for a single webhook subscription.
+ */
+export interface PaymentWebhookOptions {
+  /** Maximum number of retry attempts after the initial request (default: 3). */
+  maxRetries?: number;
+  /** Per-attempt request timeout in milliseconds (default: 10_000). */
+  timeoutMs?: number;
+  /**
+   * Optional list of notification channels to deliver to in addition to the
+   * HTTP webhook. Channels are an independently decoupled abstraction so
+   * additional transports (Slack, SMS, email, …) can be plugged in without
+   * coupling them to the webhook delivery logic.
+   */
+  channels?: PaymentNotificationChannel[];
+}
+
+const DEFAULT_MAX_RETRIES = 3;
+const DEFAULT_TIMEOUT_MS = 10_000;
+const DEFAULT_BACKOFF_MS = 1000;
+const MAX_BACKOFF_MS = 8000;
+
+/**
+ * Webhook payload delivered to a subscriber.
+ *
+ * Standardized so consumers can safely process duplicate deliveries: the
+ * `eventId` is deterministic for a given (account, event, transactionId), so
+ * redelivery of the same logical event is always identifiable.
+ */
+export interface PaymentWebhookPayload {
+  /** Deterministic event identifier for idempotency / duplicate detection. */
+  eventId: string;
+  /** The payment lifecycle event that occurred. */
+  event: PaymentNotificationEvent;
+  /** ISO-8601 timestamp of when the event was emitted. */
+  timestamp: string;
+  /** The transaction this payment event relates to. */
+  transactionId: string;
+  /** The account associated with the payment. */
+  account: string;
+}
+
+/** Input describing an emitted payment event. */
+export interface PaymentNotificationInput {
+  /** Transaction identifier the payment belongs to. */
+  transactionId: string;
+  /** Account associated with the payment. */
+  account: string;
+  /**
+   * Optional explicit event ID. When omitted, one is derived deterministically
+   * from the event identity so duplicate deliveries share the same ID.
+   */
+  eventId?: string;
+}
+
+/**
+ * A registered webhook subscription.
+ */
+export interface PaymentWebhookRegistration {
+  /** URL to deliver webhook payloads to. */
+  url: string;
+  /** Payment events this subscription is interested in. */
+  events: PaymentNotificationEvent[];
+  /** Resolved delivery options. */
+  options: Required<Omit<PaymentWebhookOptions, "channels">>;
+}
+
+/**
+ * Decoupled notification channel abstraction.
+ *
+ * Applications may register channels to receive the same payload as webhook
+ * subscribers. Implementing this interface is entirely independent of the
+ * webhook delivery machinery, satisfying the "without coupling them to
+ * webhook logic" requirement.
+ */
+export interface PaymentNotificationChannel {
+  /** Human-readable channel name, used in error messages. */
+  readonly name: string;
+  /**
+   * Deliver a payload through this channel.
+   * Return `true` on success. Returning `false` (or throwing) counts as a
+   * failed attempt and is retried like a failed webhook delivery.
+   */
+  deliver(payload: PaymentWebhookPayload): Promise<boolean>;
+}
+
+const webhookRegistry = new Map<string, PaymentWebhookRegistration>();
+const channelRegistry = new Set<PaymentNotificationChannel>();
+
+/** Registry key — one entry per (url, event) pair. */
+function registrationKey(
+  url: string,
+  event: PaymentNotificationEvent,
+): string {
+  return `${url}::${event}`;
+}
+
+function validateUrl(url: string): SorokitResult<void> | undefined {
+  if (typeof url !== "string" || url.length === 0) {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      "Webhook URL must be a non-empty string",
+    );
+  }
+  try {
+    new URL(url);
+  } catch {
+    return err(SorokitErrorCode.INVALID_CONFIG, `Invalid webhook URL: ${url}`);
+  }
+  return undefined;
+}
+
+function isNonNegativeInteger(value: number, name: string): boolean {
+  return Number.isInteger(value) && value >= 0;
+}
+
+function validateOptions(
+  options: PaymentWebhookOptions | undefined,
+): SorokitResult<Required<Omit<PaymentWebhookOptions, "channels">>> | undefined {
+  if (options === undefined) {
+    return ok({
+      maxRetries: DEFAULT_MAX_RETRIES,
+      timeoutMs: DEFAULT_TIMEOUT_MS,
+    });
+  }
+  const maxRetries = options.maxRetries ?? DEFAULT_MAX_RETRIES;
+  const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
+  if (!isNonNegativeInteger(maxRetries, "maxRetries")) {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      "maxRetries must be a non-negative integer",
+    );
+  }
+  if (!isNonNegativeInteger(timeoutMs, "timeoutMs") || timeoutMs === 0) {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      "timeoutMs must be a positive integer",
+    );
+  }
+  return ok({ maxRetries, timeoutMs });
+}
+
+/**
+ * Register a webhook (and optional notification channels) for one or more
+ * payment events.
+ *
+ * @param url     - URL to deliver webhook payloads to.
+ * @param events  - Payment event types to subscribe to.
+ * @param options - Delivery options (retries, timeout, extra channels).
+ * @returns ok(void) on success, or a structured error on invalid input.
+ *
+ * @example
+ * const result = registerPaymentWebhook(
+ *   "https://example.com/payments",
+ *   ["payment_received", "payment_failed"],
+ *   { maxRetries: 5, timeoutMs: 5000 },
+ * );
+ */
+export function registerPaymentWebhook(
+  url: string,
+  events: PaymentNotificationEvent[],
+  options?: PaymentWebhookOptions,
+): SorokitResult<void> {
+  if (!Array.isArray(events) || events.length === 0) {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      "At least one payment event type must be provided",
+    );
+  }
+
+  const normalized: PaymentNotificationEvent[] = [];
+  for (const event of events) {
+    if (!isPaymentNotificationEvent(event)) {
+      return err(
+        SorokitErrorCode.INVALID_CONFIG,
+        `Invalid payment event type: ${String(event)}. Must be one of: ${PAYMENT_NOTIFICATION_EVENTS.join(", ")}`,
+      );
+    }
+    normalized.push(event);
+  }
+
+  const urlError = validateUrl(url);
+  if (urlError) return urlError;
+
+  const optionsResult = validateOptions(options);
+  if (optionsResult && optionsResult.status === "error") return optionsResult;
+  const resolvedOptions = optionsResult!.data;
+
+  for (const channel of options?.channels ?? []) {
+    if (!channel || typeof channel.deliver !== "function") {
+      return err(
+        SorokitErrorCode.INVALID_CONFIG,
+        "Notification channels must implement deliver(payload)",
+      );
+    }
+    channelRegistry.add(channel);
+  }
+
+  const registration: PaymentWebhookRegistration = {
+    url,
+    events: normalized,
+    options: resolvedOptions,
+  };
+
+  for (const event of normalized) {
+    webhookRegistry.set(registrationKey(url, event), registration);
+  }
+
+  return ok(undefined);
+}
+
+/**
+ * Remove a webhook subscription for a specific event.
+ */
+export function unregisterPaymentWebhook(
+  url: string,
+  event: PaymentNotificationEvent,
+): SorokitResult<void> {
+  if (!isPaymentNotificationEvent(event)) {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      `Invalid payment event type: ${String(event)}`,
+    );
+  }
+  const key = registrationKey(url, event);
+  if (!webhookRegistry.has(key)) {
+    return err(
+      SorokitErrorCode.INVALID_CONFIG,
+      `No webhook registered for event '${event}' at '${url}'`,
+    );
+  }
+  webhookRegistry.delete(key);
+  return ok(undefined);
+}
+
+/**
+ * List all webhook subscriptions for a payment event type.
+ */
+export function listPaymentWebhooks(
+  event: PaymentNotificationEvent,
+): PaymentWebhookRegistration[] {
+  if (!isPaymentNotificationEvent(event)) return [];
+  const results: PaymentWebhookRegistration[] = [];
+  for (const [key, registration] of webhookRegistry.entries()) {
+    if (key.endsWith(`::${event}`) && !results.includes(registration)) {
+      results.push(registration);
+    }
+  }
+  return results;
+}
+
+/** Clear all registered webhook subscriptions. */
+export function clearPaymentWebhooks(): void {
+  webhookRegistry.clear();
+  channelRegistry.clear();
+}
+
+/**
+ * Deterministically derive an event ID from the payment identity so duplicate
+ * deliveries of the same logical event carry the same identifier.
+ *
+ * FNV-1a (32-bit) over `transactionId | event | account`, hex-encoded.
+ */
+export function generatePaymentEventId(
+  transactionId: string,
+  event: PaymentNotificationEvent,
+  account: string,
+): string {
+  const data = `${transactionId}|${event}|${account}`;
+  let hash = 0x811c9dc5;
+  for (let i = 0; i < data.length; i++) {
+    hash ^= data.charCodeAt(i);
+    hash = (hash * 0x01000193) >>> 0;
+  }
+  return hash.toString(16).padStart(8, "0");
+}
+
+/** Resolve the event ID, applying caller-supplied override or the derived one. */
+function resolveEventId(
+  input: PaymentNotificationInput,
+  event: PaymentNotificationEvent,
+): string {
+  return input.eventId ?? generatePaymentEventId(input.transactionId, event, input.account);
+}
+
+/**
+ * Send a webhook payload with exponential backoff and a configurable retry
+ * limit and timeout.
+ *
+ * @param registration - The subscription to deliver to.
+ * @param payload      - Payload to deliver.
+ * @returns ok(void) on success, or a structured error after the final attempt.
+ */
+async function sendPayloadWithRetry(
+  registration: PaymentWebhookRegistration,
+  payload: PaymentWebhookPayload,
+): Promise<SorokitResult<void>> {
+  const { url } = registration;
+  const { maxRetries, timeoutMs } = registration.options;
+  let lastError: unknown;
+
+  for (let attempt = 0; attempt <= maxRetries; attempt++) {
+    try {
+      const response = await fetch(url, {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+          "X-Sorokit-Event": payload.event,
+          "X-Sorokit-Event-Id": payload.eventId,
+          "Idempotency-Key": payload.eventId,
+        },
+        body: JSON.stringify(payload),
+        signal: AbortSignal.timeout(timeoutMs),
+      });
+
+      if (!response.ok) {
+        throw new Error(`HTTP ${response.status}: ${response.statusText}`);
+      }
+
+      return ok(undefined);
+    } catch (error) {
+      lastError = error;
+      if (attempt < maxRetries) {
+        // Exponential backoff (1s, 2s, 4s, …) capped at MAX_BACKOFF_MS so
+        // retry behavior can never spin out of control.
+        const delayMs = Math.min(
+          DEFAULT_BACKOFF_MS * Math.pow(2, attempt),
+          MAX_BACKOFF_MS,
+        );
+        await sleep(delayMs);
+      }
+    }
+  }
+
+  return err(
+    SorokitErrorCode.NETWORK_ERROR,
+    `Payment webhook delivery failed after ${maxRetries + 1} attempts to ${url}: ${lastError instanceof Error ? lastError.message : String(lastError)}`,
+    lastError,
+  );
+}
+
+/** Deliver a payload through any registered notification channels. */
+async function deliverToChannels(
+  payload: PaymentWebhookPayload,
+): Promise<SorokitResult<void>[]> {
+  return Promise.all(
+    Array.from(channelRegistry).map(async (channel) => {
+      try {
+        const delivered = await channel.deliver(payload);
+        if (delivered) return ok(undefined);
+        return err(
+          SorokitErrorCode.NETWORK_ERROR,
+          `Notification channel '${channel.name}' rejected payload for event '${payload.event}'`,
+        );
+      } catch (error) {
+        return err(
+          SorokitErrorCode.NETWORK_ERROR,
+          `Notification channel '${channel.name}' failed for event '${payload.event}': ${error instanceof Error ? error.message : String(error)}`,
+          error,
+        );
+      }
+    }),
+  );
+}
+
+/**
+ * Trigger delivery of a payment notification to all matching subscribers and
+ * wait for the results.
+ *
+ * Deliveries run concurrently and are reported independently so one failing
+ * endpoint cannot mask another.
+ *
+ * @param event - The payment event that occurred.
+ * @param input - Payment identity (transactionId, account, optional eventId).
+ * @returns One result per delivery target (ok or error).
+ */
+export async function triggerPaymentNotifications(
+  event: PaymentNotificationEvent,
+  input: PaymentNotificationInput,
+): Promise<SorokitResult<void>[]> {
+  if (!isPaymentNotificationEvent(event)) {
+    return [
+      err(
+        SorokitErrorCode.INVALID_CONFIG,
+        `Invalid payment event type: ${String(event)}`,
+      ),
+    ];
+  }
+  if (typeof input.transactionId !== "string" || input.transactionId.length === 0) {
+    return [err(SorokitErrorCode.INVALID_CONFIG, "transactionId must be a non-empty string")];
+  }
+  if (typeof input.account !== "string" || input.account.length === 0) {
+    return [err(SorokitErrorCode.INVALID_CONFIG, "account must be a non-empty string")];
+  }
+
+  const payload: PaymentWebhookPayload = {
+    eventId: resolveEventId(input, event),
+    event,
+    timestamp: new Date().toISOString(),
+    transactionId: input.transactionId,
+    account: input.account,
+  };
+
+  const registrations = listPaymentWebhooks(event);
+  const webhookResults = await Promise.all(
+    registrations.map((registration) => sendPayloadWithRetry(registration, payload)),
+  );
+  const channelResults = await deliverToChannels(payload);
+  return [...webhookResults, ...channelResults];
+}
+
+/**
+ * Fire-and-forget dispatch of a payment notification.
+ *
+ * Never throws, never rejects, and never blocks transaction processing while
+ * subscribers are being notified. Individual delivery failures are reported by
+ * {@link triggerPaymentNotifications} and intentionally swallowed here.
+ *
+ * @param event - The payment event that occurred.
+ * @param input - Payment identity (transactionId, account, optional eventId).
+ */
+export function dispatchPaymentNotification(
+  event: PaymentNotificationEvent,
+  input: PaymentNotificationInput,
+): void {
+  try {
+    void triggerPaymentNotifications(event, input).catch(() => {
+      // Swallow: notification delivery must never surface into the
+      // transaction/payment processing flow.
+    });
+  } catch {
+    // Defensive: even synchronous failures must not propagate.
+  }
+}