import type {
  ActiveUploadStatus,
  QuarantinedUpload,
  ResumableUploadCheckpoint,
  ResumableUploadCheckpointStore,
  ResumableUploadControlPlane,
  ResumableUploadPort,
  ResumableUploadRequest,
  UploadAbortOutcome,
  UploadFileFingerprint,
  UploadPartCapability,
  UploadPartDescriptor,
  UploadPartExecutor,
  UploadPartReceipt,
  UploadProviderFailure,
  UploadProviderResult,
  UploadSession,
  UploadSessionStatus,
} from "../../../application/ports/browser-transfer/resumable-upload.ts";
import { RESUMABLE_UPLOAD_PROTOCOL } from "../../../application/ports/browser-transfer/resumable-upload.ts";
import type {
  BrowserDataFailureCode,
  BrowserDataObserver,
  BrowserDataOperation,
  BrowserDataRecovery,
  BrowserDataResult,
  TransferProgress,
} from "../../../application/ports/browser-file-storage/shared.ts";
import {
  browserDataFailure,
  browserDataSuccess,
} from "../../browser-file-storage/result.ts";
import {
  isResumableUploadCheckpoint,
  isUploadFileFingerprint,
  isUploadPartReceipt,
  MEDIA_TYPE,
  SAFE_OPAQUE_ID,
  SAFE_REGISTRY_ID,
  SAFE_UPLOAD_KEY,
  samePart,
  SHA256_HEX,
} from "./checkpoint-schema.ts";
import {
  buildUploadPartManifest,
  deriveUploadIdempotencyKey,
  digestRequestBinding,
  digestUploadSessionBinding,
  findManifestPart,
  iterateUploadParts,
  readAndVerifyRangePart,
  snapshotUploadCrypto,
  snapshotUploadSource,
  verifyPartAgainstManifest,
  verifyUploadPartBytes,
  type UploadCrypto,
  type UploadPartManifest,
  type UploadSourceSnapshot,
} from "./upload-byte-source.ts";
import {
  resolveResumableUploadRuntimePolicy,
  type ResumableUploadRuntimePolicy,
} from "./runtime-policy.ts";
import type { UploadCancellationChannel } from "./upload-cancellation-channel.ts";
import type { UploadMutationLock } from "./upload-mutation-lock.ts";

export type ResumableUploadRuntime = ResumableUploadPort &
  Readonly<{
    /**
     * BT-UP-06. Compatibility facade: closes admission and starts the same
     * single-flight drain that `dispose()` awaits.
     */
    close(): void;
    /**
     * BT-UP-06. Awaitable teardown for a future composition owner. It shares
     * one drain promise, aborts the active operation registry and only then
     * closes the checkpoint store and cancellation channel, so success actually
     * means quiescent. No current bootstrap consumer is assumed.
     */
    /**
     * TR-RR-06. Closes admission and returns the *bounded* drain result. A
     * failure means the runtime is still `CLOSING`: physical work the caller
     * must not treat as finished is still in flight.
     */
    dispose(): Promise<BrowserDataResult<void>>;
    lifecycle(): "OPEN" | "CLOSING" | "CLOSED";
  }>;

export type ResumableUploadRuntimeDependencies<Capability> = Readonly<{
  controlPlane: ResumableUploadControlPlane<Capability>;
  partExecutor: UploadPartExecutor<Capability>;
  checkpoints: ResumableUploadCheckpointStore;
  mutationLock: UploadMutationLock;
  /**
   * Runtime-owned ephemeral BroadcastChannel coordination. When omitted,
   * explicit abort is still correct but may wait for another context's lock.
   */
  crossContextCancellation?: UploadCancellationChannel;
  crypto: Crypto;
  policy?: Partial<ResumableUploadRuntimePolicy>;
  now?: () => number;
  random?: () => number;
  sleep?: (delayMs: number, signal: AbortSignal) => Promise<void>;
  observer?: BrowserDataObserver;
}>;

type UploadRequestSnapshot = Readonly<{
  uploadKey: string;
  purpose: string;
  mediaType: string;
  source: UploadSourceSnapshot;
  signal: AbortSignal;
  onProgress?: (progress: TransferProgress) => void;
}>;

type RuntimeDependencies<Capability> = Readonly<{
  controlPlane: ResumableUploadControlPlane<Capability>;
  partExecutor: UploadPartExecutor<Capability>;
  checkpoints: ResumableUploadCheckpointStore;
  mutationLock: UploadMutationLock;
  crossContextCancellation?: UploadCancellationChannel;
  crypto: UploadCrypto;
  policy: ResumableUploadRuntimePolicy;
  now(): number;
  random(): number;
  sleep(delayMs: number, signal: AbortSignal): Promise<void>;
  observer?: BrowserDataObserver;
  /**
   * TR-04. Every raw provider promise, from the moment the collaborator is
   * called until it actually settles. The wrapper that bounds the attempt can
   * settle long before the provider does, so the wrapper registry alone could
   * report an empty set while physical work was still running.
   */
  physicalTasks: Set<Promise<unknown>>;
}>;

type ActiveResolution =
  | Readonly<{
      kind: "ACTIVE";
      checkpoint: ResumableUploadCheckpoint;
    }>
  | Readonly<{
      kind: "COMPLETED";
      upload: QuarantinedUpload;
    }>;

const FAILURE_CODES: ReadonlySet<string> = new Set([
  "ABORTED",
  "BLOCKED",
  "CONFLICT",
  "CORRUPT_DATA",
  "EXPIRED_RESOURCE",
  "INTEGRITY_FAILED",
  "INVALID_INPUT",
  "LIMIT_EXCEEDED",
  "MIGRATION_FAILED",
  "NOT_FOUND",
  "NOT_READABLE",
  "PERMISSION_DENIED",
  "POLICY_REJECTED",
  "QUOTA_EXCEEDED",
  "STALE_RESULT",
  "STORAGE_EVICTED",
  "UNAVAILABLE",
  "UNSUPPORTED",
]);
const RECOVERIES: ReadonlySet<string> = new Set([
  "NONE",
  "RETRY",
  "REOPEN",
  "RESELECT",
  "RELOAD_OTHER_CONTEXTS",
  "READ_ONLY",
  "ONLINE_ONLY",
  "REHYDRATE",
  "EXPORT_REQUIRED",
  "REISSUE_CAPABILITY",
  "RESUME",
  "RESTART",
  "RECONCILE",
]);

export function createResumableUploadRuntime<Capability>(
  inputDependencies: ResumableUploadRuntimeDependencies<Capability>,
): ResumableUploadRuntime {
  /** TR-04. Raw provider work, tracked independently of its bounded wrapper. */
  const physicalTasks = new Set<Promise<unknown>>();
  const dependencies = snapshotDependencies(inputDependencies, physicalTasks);
  const lifetime = new AbortController();
  const localUploads = new Map<string, Set<AbortController>>();
  /** BT-UP-06. Terminal settlement of every admitted operation. */
  const activeOperations = new Set<Promise<unknown>>();
  let closed = false;
  let lifecycle: "OPEN" | "CLOSING" | "CLOSED" = "OPEN";
  let drain: Promise<BrowserDataResult<void>> | null = null;
  const cancelLocalUploads = (uploadKey: string): void => {
    if (!SAFE_UPLOAD_KEY.test(uploadKey)) return;
    for (const controller of localUploads.get(uploadKey) ?? []) {
      controller.abort();
    }
  };
  const releaseCrossContextCancellation =
    dependencies.crossContextCancellation?.subscribe(
      cancelLocalUploads,
    );

  const runtime: ResumableUploadRuntime = Object.freeze({
    async upload(
      input: ResumableUploadRequest,
    ): Promise<BrowserDataResult<QuarantinedUpload>> {
      if (closed) {
        return browserDataFailure("UNAVAILABLE", "UPLOAD_SESSION", {
          recovery: "RESUME",
        });
      }
      let request: UploadRequestSnapshot;
      try {
        request = snapshotRequest(input, dependencies.policy);
      } catch {
        return browserDataFailure("INVALID_INPUT", "UPLOAD_SESSION");
      }
      const localUpload = new AbortController();
      const uploadsForKey =
        localUploads.get(request.uploadKey) ?? new Set<AbortController>();
      uploadsForKey.add(localUpload);
      localUploads.set(request.uploadKey, uploadsForKey);
      const lifetimeScope = combineAbortSignals(
        request.signal,
        lifetime.signal,
      );
      const operationScope = combineAbortSignals(
        lifetimeScope.signal,
        localUpload.signal,
      );
      if (operationScope.signal.aborted) {
        operationScope.release();
        lifetimeScope.release();
        uploadsForKey.delete(localUpload);
        if (uploadsForKey.size === 0) {
          localUploads.delete(request.uploadKey);
        }
        return browserDataFailure("ABORTED", "UPLOAD_SESSION");
      }
      const operation = dependencies.mutationLock.run(
        request.uploadKey,
        operationScope.signal,
        async () =>
          await executeUpload(
            dependencies,
            Object.freeze({
              ...request,
              signal: operationScope.signal,
            }),
          ),
      );
      // Tracked until terminal settlement so dispose() can prove quiescence.
      const tracked = Promise.resolve(operation).catch(() => undefined);
      activeOperations.add(tracked);
      void tracked.finally(() => activeOperations.delete(tracked));
      try {
        return await operation;
      } catch (error) {
        return mapLockFailure(error, "UPLOAD_SESSION");
      } finally {
        operationScope.release();
        lifetimeScope.release();
        uploadsForKey.delete(localUpload);
        if (uploadsForKey.size === 0) {
          localUploads.delete(request.uploadKey);
        }
      }
    },

    async abort(
      input: Parameters<ResumableUploadPort["abort"]>[0],
    ): Promise<BrowserDataResult<UploadAbortOutcome>> {
      if (closed) {
        return browserDataFailure("UNAVAILABLE", "UPLOAD_ABORT", {
          recovery: "RESUME",
        });
      }
      if (
        !input ||
        typeof input !== "object" ||
        typeof input.uploadKey !== "string" ||
        !SAFE_UPLOAD_KEY.test(input.uploadKey) ||
        !isAbortSignal(input.signal)
      ) {
        return browserDataFailure("INVALID_INPUT", "UPLOAD_ABORT");
      }
      const uploadKey = input.uploadKey;
      if (!input.signal.aborted) {
        cancelLocalUploads(uploadKey);
        dependencies.crossContextCancellation?.publish(uploadKey);
      }
      const combined = combineAbortSignals(input.signal, lifetime.signal);
      const operation = dependencies.mutationLock.run(
        uploadKey,
        combined.signal,
        async () =>
          await executeAbort(dependencies, uploadKey, combined.signal),
      );
      // TR-RR-06. An abort is admitted physical work like an upload, so it is
      // tracked from admission and `dispose()` cannot step over it.
      const tracked = Promise.resolve(operation).catch(() => undefined);
      activeOperations.add(tracked);
      void tracked.finally(() => activeOperations.delete(tracked));
      try {
        return await operation;
      } catch (error) {
        return mapLockFailure(error, "UPLOAD_ABORT");
      } finally {
        combined.release();
      }
    },

    close() {
      // BT-UP-06. Admission closes synchronously; the drain runs behind the
      // same single-flight promise dispose() returns.
      void startDrain();
    },

    dispose(): Promise<BrowserDataResult<void>> {
      return startDrain();
    },

    lifecycle: () => lifecycle,
  });

  function startDrain(): Promise<BrowserDataResult<void>> {
    drain ??= (async () => {
      closed = true;
      lifecycle = "CLOSING";
      releaseCrossContextCancellation?.();
      dependencies.crossContextCancellation?.close();
      // Abort every admitted operation, then wait for their real settlement.
      lifetime.abort();
      for (const controllers of localUploads.values()) {
        for (const controller of controllers) controller.abort();
      }
      // TR-RR-06. Bounded. A non-cooperative mutation lock or provider must not
      // make teardown unbounded, and an unproved drain is reported as such
      // rather than closed over.
      let timer: ReturnType<typeof setTimeout> | undefined;
      const expired = new Promise<"EXPIRED">((resolve) => {
        timer = setTimeout(
          () => resolve("EXPIRED"),
          dependencies.policy.cleanupDeadlineMs,
        );
      });
      // TR-04. Quiescence means both registries: the bounded wrappers and the
      // raw provider work they may have outlived. A settling wrapper can still
      // register more physical work, so the drain repeats until both are empty
      // or the cleanup deadline expires.
      const quiescent = (async () => {
        while (activeOperations.size > 0 || physicalTasks.size > 0) {
          await Promise.allSettled([...activeOperations, ...physicalTasks]);
        }
        return "DRAINED" as const;
      })();
      const drained = await Promise.race([quiescent, expired]);
      if (timer !== undefined) clearTimeout(timer);
      if (drained === "EXPIRED") {
        // The store stays open: something can still write a checkpoint.
        return browserDataFailure("UNAVAILABLE", "UPLOAD_ABORT", {
          retryable: true,
          recovery: "RESUME",
        });
      }
      // The store closes only after nothing can still write a checkpoint.
      dependencies.checkpoints.close();
      lifecycle = "CLOSED";
      return browserDataSuccess(undefined);
    })();
    return drain;
  }
  return runtime;
}

async function executeUpload<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  request: UploadRequestSnapshot,
): Promise<BrowserDataResult<QuarantinedUpload>> {
  reportProgress(request, "VALIDATING", 0);
  if (
    request.source.byteLength > dependencies.policy.maxFileBytes ||
    Math.ceil(
      request.source.byteLength / dependencies.policy.partSizeBytes,
    ) > dependencies.policy.maxPartCount
  ) {
    return browserDataFailure("LIMIT_EXCEEDED", "UPLOAD_SESSION");
  }
  const manifest = await buildUploadPartManifest({
    source: request.source,
    partSizeBytes: dependencies.policy.partSizeBytes,
    maxPartCount: dependencies.policy.maxPartCount,
    maxSourceChunkBytes: dependencies.policy.maxSourceChunkBytes,
    crypto: dependencies.crypto,
    signal: request.signal,
    onPreparedBytes(bytes) {
      reportProgress(request, "PREPARING", bytes);
    },
  });
  if (!manifest.ok) return manifest;
  const requestBinding = await digestRequestBinding({
    uploadKey: request.uploadKey,
    purpose: request.purpose,
    mediaType: request.mediaType,
    fingerprint: manifest.value.fingerprint,
    crypto: dependencies.crypto,
    signal: request.signal,
  });
  if (!requestBinding.ok) return requestBinding;

  const stored = await dependencies.checkpoints.read(
    request.uploadKey,
    request.signal,
  );
  if (!stored.ok) return remapResult(stored, "UPLOAD_RECONCILE");
  if (
    stored.value &&
    (!sameFingerprint(
      stored.value.fingerprint,
      manifest.value.fingerprint,
    ) ||
      stored.value.requestBindingSha256 !== requestBinding.value)
  ) {
    return browserDataFailure("STALE_RESULT", "UPLOAD_RECONCILE", {
      recovery: "RESELECT",
    });
  }
  if (stored.value?.state === "ABORT_PENDING") {
    return browserDataFailure("CONFLICT", "UPLOAD_RECONCILE", {
      recovery: "RECONCILE",
    });
  }

  const active = await resolveActiveSession(
    dependencies,
    request,
    manifest.value,
    requestBinding.value,
    stored.value,
  );
  if (!active.ok) return active;
  if (active.value.kind === "COMPLETED") {
    return browserDataSuccess(active.value.upload);
  }

  const transferred = await transferMissingParts(
    dependencies,
    request,
    manifest.value,
    active.value.checkpoint,
  );
  if (!transferred.ok) return transferred;

  reportProgress(
    request,
    "VERIFYING",
    manifest.value.fingerprint.byteLength,
  );
  const reconciled = await reconcileActiveCheckpoint(
    dependencies,
    request,
    manifest.value,
    transferred.value,
  );
  if (!reconciled.ok) return reconciled;
  if (reconciled.value.kind === "COMPLETED") {
    return browserDataSuccess(reconciled.value.upload);
  }
  if (reconciled.value.kind !== "ACTIVE") {
    const removed = await dependencies.checkpoints.remove({
      uploadKey: transferred.value.uploadKey,
      expectedRevision: transferred.value.revision,
      signal: request.signal,
    });
    if (!removed.ok) {
      return remapResult(removed, "UPLOAD_RECONCILE");
    }
    return browserDataFailure("EXPIRED_RESOURCE", "UPLOAD_SESSION", {
      recovery: "RESTART",
    });
  }
  if (
    reconciled.value.checkpoint.acceptedParts.length !==
    manifest.value.parts.length
  ) {
    return browserDataFailure("STALE_RESULT", "UPLOAD_RECONCILE", {
      retryable: true,
      recovery: "RECONCILE",
    });
  }

  reportProgress(
    request,
    "FINALIZING",
    manifest.value.fingerprint.byteLength,
  );
  return await completeUpload(
    dependencies,
    request,
    manifest.value,
    reconciled.value.checkpoint,
  );
}

async function resolveActiveSession<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  request: UploadRequestSnapshot,
  manifest: UploadPartManifest,
  requestBindingSha256: string,
  existing: ResumableUploadCheckpoint | null,
): Promise<BrowserDataResult<ActiveResolution>> {
  let checkpoint = existing;
  for (let restartIndex = 0; restartIndex < 2; restartIndex += 1) {
    if (!checkpoint) {
      const created = await createUploadSession(
        dependencies,
        request,
        manifest,
        requestBindingSha256,
      );
      if (!created.ok) return created;
      checkpoint = created.value;
    }
    const reconciled = await reconcileActiveCheckpoint(
      dependencies,
      request,
      manifest,
      checkpoint,
    );
    if (!reconciled.ok) return reconciled;
    if (
      reconciled.value.kind === "ACTIVE" ||
      reconciled.value.kind === "COMPLETED"
    ) {
      return browserDataSuccess(reconciled.value);
    }
    if (restartIndex === 1) {
      const removed = await dependencies.checkpoints.remove({
        uploadKey: checkpoint.uploadKey,
        expectedRevision: checkpoint.revision,
        signal: request.signal,
      });
      if (!removed.ok) {
        return remapResult(removed, "UPLOAD_RECONCILE");
      }
      break;
    }
    const replaced = await createUploadSession(
      dependencies,
      request,
      manifest,
      requestBindingSha256,
      checkpoint,
    );
    if (!replaced.ok) return replaced;
    checkpoint = replaced.value;
  }
  return browserDataFailure("EXPIRED_RESOURCE", "UPLOAD_SESSION", {
    recovery: "RESTART",
  });
}

type ReconciliationResolution =
  | ActiveResolution
  | Readonly<{ kind: "TERMINAL" }>;

async function reconcileActiveCheckpoint<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  request: UploadRequestSnapshot,
  manifest: UploadPartManifest,
  checkpoint: ResumableUploadCheckpoint,
): Promise<BrowserDataResult<ReconciliationResolution>> {
  const statusResult = await callWithRetry(
    dependencies,
    "UPLOAD_RECONCILE",
    request.signal,
    async (attemptSignal) =>
      await dependencies.controlPlane.getStatus({
        protocol: RESUMABLE_UPLOAD_PROTOCOL,
        sessionId: checkpoint.sessionId,
        requestBindingSha256: checkpoint.requestBindingSha256,
        fingerprint: checkpoint.fingerprint,
        signal: attemptSignal,
      }),
  );
  if (!statusResult.ok) {
    if (
      statusResult.error.code === "NOT_FOUND" ||
      statusResult.error.code === "EXPIRED_RESOURCE"
    ) {
      return browserDataSuccess(
        Object.freeze({ kind: "TERMINAL" as const }),
      );
    }
    return stripProviderFailure(statusResult);
  }
  const status = validateStatus(
    statusResult.value,
    checkpoint,
    manifest,
    dependencies,
  );
  if (!status.ok) return status;
  if (status.value.state === "QUARANTINED") {
    const removed = await dependencies.checkpoints.remove({
      uploadKey: checkpoint.uploadKey,
      expectedRevision: checkpoint.revision,
      signal: request.signal,
    });
    if (!removed.ok) return remapResult(removed, "UPLOAD_RECONCILE");
    return browserDataSuccess(
      Object.freeze({
        kind: "COMPLETED" as const,
        upload: quarantinedOutcome(
          status.value.resourceId,
          status.value.session,
          true,
        ),
      }),
    );
  }
  if (status.value.state !== "ACTIVE") {
    return browserDataSuccess(Object.freeze({ kind: "TERMINAL" as const }));
  }
  const acceptedParts = snapshotReceipts(status.value.acceptedParts);
  const unchanged =
    acceptedParts.length === checkpoint.acceptedParts.length &&
    acceptedParts.every((part, index) => {
      const current = checkpoint.acceptedParts[index];
      return Boolean(
        current &&
          samePart(part, current) &&
          part.receiptToken === current.receiptToken,
      );
    });
  let nextCheckpoint = checkpoint;
  if (!unchanged) {
    const now = safeNow(dependencies, "UPLOAD_RECONCILE");
    if (!now.ok) return now;
    const candidate = checkpointFrom({
      previous: checkpoint,
      revision: checkpoint.revision + 1,
      acceptedParts,
      updatedAtEpochMs: now.value,
    });
    const persisted = await dependencies.checkpoints.compareAndSwap({
      expectedRevision: checkpoint.revision,
      checkpoint: candidate,
      signal: request.signal,
    });
    if (!persisted.ok) {
      return remapResult(persisted, "UPLOAD_RECONCILE");
    }
    nextCheckpoint = persisted.value;
  }
  return browserDataSuccess(
    Object.freeze({
      kind: "ACTIVE" as const,
      checkpoint: nextCheckpoint,
    }),
  );
}

async function createUploadSession<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  request: UploadRequestSnapshot,
  manifest: UploadPartManifest,
  requestBindingSha256: string,
  previous?: ResumableUploadCheckpoint,
): Promise<BrowserDataResult<ResumableUploadCheckpoint>> {
  const idempotencyKey = await deriveUploadIdempotencyKey({
    label: "CREATE",
    requestBindingSha256,
    ...(previous ? { sessionId: previous.sessionId } : {}),
    crypto: dependencies.crypto,
    signal: request.signal,
  });
  if (!idempotencyKey.ok) return idempotencyKey;
  const created = await callWithRetry(
    dependencies,
    "UPLOAD_SESSION",
    request.signal,
    async (attemptSignal) =>
      await dependencies.controlPlane.createSession({
        protocol: RESUMABLE_UPLOAD_PROTOCOL,
        uploadKey: request.uploadKey,
        purpose: request.purpose,
        mediaType: request.mediaType,
        requestBindingSha256,
        fingerprint: manifest.fingerprint,
        requestedPartSizeBytes: dependencies.policy.partSizeBytes,
        requestedMaxConcurrency: dependencies.policy.maxConcurrency,
        idempotencyKey: idempotencyKey.value,
        signal: attemptSignal,
      }),
  );
  if (!created.ok) return stripProviderFailure(created);
  const session = validateSession(
    created.value,
    requestBindingSha256,
    manifest.fingerprint,
    dependencies,
  );
  if (!session.ok) return session;
  const now = safeNow(dependencies, "UPLOAD_SESSION");
  if (!now.ok) return now;
  const checkpoint = checkpointFrom({
    protocol: RESUMABLE_UPLOAD_PROTOCOL,
    revision: (previous?.revision ?? 0) + 1,
    state: "ACTIVE",
    uploadKey: request.uploadKey,
    requestBindingSha256,
    fingerprint: manifest.fingerprint,
    sessionId: session.value.sessionId,
    sessionExpiresAtEpochMs: session.value.expiresAtEpochMs,
    sessionMaxConcurrency: session.value.maxConcurrency,
    acceptedParts: Object.freeze([]),
    updatedAtEpochMs: now.value,
  });
  const persisted = await dependencies.checkpoints.compareAndSwap({
    expectedRevision: previous?.revision ?? null,
    checkpoint,
    signal: request.signal,
  });
  if (!persisted.ok) {
    // The server owns expiry/garbage collection for a create that succeeded
    // before a local durable checkpoint could commit.
    return remapResult(persisted, "UPLOAD_RECONCILE");
  }
  return persisted;
}

async function transferMissingParts<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  request: UploadRequestSnapshot,
  manifest: UploadPartManifest,
  initialCheckpoint: ResumableUploadCheckpoint,
): Promise<BrowserDataResult<ResumableUploadCheckpoint>> {
  let checkpoint = initialCheckpoint;
  let persistenceTail = Promise.resolve<BrowserDataResult<void>>(
    browserDataSuccess(undefined),
  );
  const accepted = new Map(
    checkpoint.acceptedParts.map((part) => [part.partNumber, part]),
  );
  let transferredBytes = checkpoint.acceptedParts.reduce(
    (total, part) => total + part.byteLength,
    0,
  );
  reportProgress(request, "TRANSFERRING", transferredBytes);

  const persistReceipt = async (
    receipt: UploadPartReceipt,
  ): Promise<BrowserDataResult<void>> => {
    const pending: Promise<BrowserDataResult<void>> =
      persistenceTail.then(async (prior): Promise<BrowserDataResult<void>> => {
        if (!prior.ok) return prior;
        const existing = accepted.get(receipt.partNumber);
        if (existing) {
          return samePart(existing, receipt) &&
            existing.receiptToken === receipt.receiptToken
            ? browserDataSuccess(undefined)
            : browserDataFailure("CONFLICT", "UPLOAD_RECONCILE", {
                recovery: "RECONCILE",
              });
        }
        const now = safeNow(dependencies, "UPLOAD_RECONCILE");
        if (!now.ok) return now;
        const acceptedParts = snapshotReceipts(
          [...accepted.values(), receipt].sort(
            (left, right) => left.partNumber - right.partNumber,
          ),
        );
        const candidate = checkpointFrom({
          previous: checkpoint,
          revision: checkpoint.revision + 1,
          acceptedParts,
          updatedAtEpochMs: now.value,
        });
        const saved = await dependencies.checkpoints.compareAndSwap({
          expectedRevision: checkpoint.revision,
          checkpoint: candidate,
          signal: request.signal,
        });
        if (!saved.ok) return remapResult(saved, "UPLOAD_RECONCILE");
        checkpoint = saved.value;
        accepted.set(receipt.partNumber, receipt);
        transferredBytes += receipt.byteLength;
        reportProgress(request, "TRANSFERRING", transferredBytes);
        return browserDataSuccess(undefined);
      });
    persistenceTail = pending;
    return await pending;
  };

  if (request.source.kind === "RANGE_READER") {
    const missing = manifest.parts.filter(
      (part) => !accepted.has(part.partNumber),
    );
    let nextIndex = 0;
    let firstFailure: BrowserDataResult<never> | null = null;
    const maxByMemory = Math.max(
      1,
      Math.floor(
        dependencies.policy.maxInFlightBytes /
          (dependencies.policy.partSizeBytes *
            dependencies.policy.partBufferCopyFactor),
      ),
    );
    const workerCount = Math.min(
      dependencies.policy.maxConcurrency,
      checkpoint.sessionMaxConcurrency,
      maxByMemory,
      missing.length,
    );
    const workers = Array.from({ length: workerCount }, async () => {
      while (!firstFailure) {
        const index = nextIndex;
        nextIndex += 1;
        const part = missing[index];
        if (!part) return;
        const bytes = await readAndVerifyRangePart({
          source: request.source as Extract<
            UploadSourceSnapshot,
            { kind: "RANGE_READER" }
          >,
          part,
          crypto: dependencies.crypto,
          signal: request.signal,
        });
        if (!bytes.ok) {
          firstFailure = bytes;
          return;
        }
        const uploaded = await uploadPartWithRetry(
          dependencies,
          request,
          manifest,
          checkpoint,
          part,
          bytes.value,
        );
        observeTerminal(
          dependencies.observer,
          "UPLOAD_PART",
          uploaded,
          part.byteLength,
        );
        if (!uploaded.ok) {
          firstFailure = uploaded;
          return;
        }
        const persisted = await persistReceipt(uploaded.value);
        if (!persisted.ok) {
          firstFailure = persisted;
          return;
        }
      }
    });
    await Promise.all(workers);
    if (firstFailure) return firstFailure;
  } else {
    for await (const partResult of iterateUploadParts({
      source: request.source,
      partSizeBytes: manifest.fingerprint.partSizeBytes,
      maxSourceChunkBytes: dependencies.policy.maxSourceChunkBytes,
      signal: request.signal,
      operation: "UPLOAD_PART",
    })) {
      if (!partResult.ok) return partResult;
      const part = findManifestPart(
        manifest,
        partResult.value.partNumber,
      );
      if (
        !part ||
        part.offset !== partResult.value.offset ||
        part.byteLength !== partResult.value.bytes.byteLength
      ) {
        return browserDataFailure(
          "INTEGRITY_FAILED",
          "UPLOAD_PART",
          { recovery: "RESELECT" },
        );
      }
      const verified = await verifyUploadPartBytes({
        bytes: partResult.value.bytes,
        part,
        crypto: dependencies.crypto,
        signal: request.signal,
      });
      if (!verified.ok) return verified;
      if (accepted.has(part.partNumber)) continue;
      const uploaded = await uploadPartWithRetry(
        dependencies,
        request,
        manifest,
        checkpoint,
        part,
        verified.value,
      );
      observeTerminal(
        dependencies.observer,
        "UPLOAD_PART",
        uploaded,
        part.byteLength,
      );
      if (!uploaded.ok) return uploaded;
      const persisted = await persistReceipt(uploaded.value);
      if (!persisted.ok) return persisted;
    }
  }
  const persisted = await persistenceTail;
  return persisted.ok ? browserDataSuccess(checkpoint) : persisted;
}

async function uploadPartWithRetry<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  request: UploadRequestSnapshot,
  manifest: UploadPartManifest,
  checkpoint: ResumableUploadCheckpoint,
  part: UploadPartDescriptor,
  bytes: Uint8Array,
): Promise<BrowserDataResult<UploadPartReceipt>> {
  const idempotencyKey = await deriveUploadIdempotencyKey({
    label: "PART",
    requestBindingSha256: checkpoint.requestBindingSha256,
    sessionId: checkpoint.sessionId,
    part,
    crypto: dependencies.crypto,
    signal: request.signal,
  });
  if (!idempotencyKey.ok) return idempotencyKey;
  const uploadBinding = await digestUploadSessionBinding({
    requestBindingSha256: checkpoint.requestBindingSha256,
    sessionId: checkpoint.sessionId,
    fingerprint: checkpoint.fingerprint,
    crypto: dependencies.crypto,
    signal: request.signal,
  });
  if (!uploadBinding.ok) return uploadBinding;

  let lastFailure: UploadProviderFailure | null = null;
  for (
    let attempt = 0;
    attempt <= dependencies.policy.maxRetries;
    attempt += 1
  ) {
    if (request.signal.aborted) {
      return browserDataFailure("ABORTED", "UPLOAD_PART");
    }
    const now = safeNow(dependencies, "UPLOAD_PART");
    if (!now.ok) return now;
    if (checkpoint.sessionExpiresAtEpochMs <= now.value) {
      return browserDataFailure("EXPIRED_RESOURCE", "UPLOAD_SESSION", {
        recovery: "RESTART",
      });
    }
    if (attempt > 0 && lastFailure) {
      const delayed = await waitForRetry(
        dependencies,
        lastFailure,
        attempt - 1,
        request.signal,
        "UPLOAD_PART",
      );
      if (!delayed.ok) return delayed;
    }
    const issued = await invokeProviderAttempt(
      dependencies,
      "UPLOAD_PART",
      request.signal,
      async (attemptSignal) =>
        await dependencies.controlPlane.issuePartCapability({
          protocol: RESUMABLE_UPLOAD_PROTOCOL,
          sessionId: checkpoint.sessionId,
          requestBindingSha256: checkpoint.requestBindingSha256,
          uploadBindingSha256: uploadBinding.value,
          fingerprint: checkpoint.fingerprint,
          mediaType: request.mediaType,
          part,
          idempotencyKey: idempotencyKey.value,
          signal: attemptSignal,
        }),
    );
    if (!issued.ok) {
      lastFailure = issued.error;
      if (!canRetry(issued.error, attempt, dependencies.policy)) {
        return stripProviderFailure(issued);
      }
      continue;
    }
    const capability = validatePartCapability(
      issued.value,
      checkpoint,
      uploadBinding.value,
      dependencies,
    );
    if (!capability.ok) {
      if (
        capability.error.code === "EXPIRED_RESOURCE" &&
        attempt < dependencies.policy.maxRetries
      ) {
        lastFailure = Object.freeze({
          ...capability.error,
          operation: "UPLOAD_PART",
          retryable: true,
          recovery: "REISSUE_CAPABILITY",
        });
        continue;
      }
      return capability;
    }
    const uploaded = await invokeProviderAttempt(
      dependencies,
      "UPLOAD_PART",
      request.signal,
      async (attemptSignal) =>
        await dependencies.partExecutor.uploadPart({
          protocol: RESUMABLE_UPLOAD_PROTOCOL,
          capability: capability.value.capability,
          sessionId: checkpoint.sessionId,
          requestBindingSha256: checkpoint.requestBindingSha256,
          uploadBindingSha256: uploadBinding.value,
          fingerprint: manifest.fingerprint,
          mediaType: request.mediaType,
          part,
          bytes,
          idempotencyKey: idempotencyKey.value,
          signal: attemptSignal,
        }),
    );
    if (uploaded.ok) {
      if (
        !isUploadPartReceipt(uploaded.value) ||
        !samePart(uploaded.value, part)
      ) {
        return browserDataFailure(
          "INTEGRITY_FAILED",
          "UPLOAD_PART",
          { recovery: "RECONCILE" },
        );
      }
      return browserDataSuccess(
        Object.freeze({ ...uploaded.value }),
      );
    }
    lastFailure = uploaded.error;
    if (!canRetry(uploaded.error, attempt, dependencies.policy)) {
      return stripProviderFailure(uploaded);
    }
  }
  return lastFailure
    ? stripProviderFailure(
        Object.freeze({ ok: false, error: lastFailure }),
      )
    : browserDataFailure("UNAVAILABLE", "UPLOAD_PART", {
        recovery: "RESUME",
      });
}

async function completeUpload<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  request: UploadRequestSnapshot,
  manifest: UploadPartManifest,
  checkpoint: ResumableUploadCheckpoint,
): Promise<BrowserDataResult<QuarantinedUpload>> {
  const orderedParts = snapshotReceipts(checkpoint.acceptedParts);
  if (
    orderedParts.length !== manifest.parts.length ||
    !orderedParts.every((part, index) => {
      const expected = manifest.parts[index];
      return Boolean(expected && samePart(part, expected));
    })
  ) {
    return browserDataFailure(
      "INTEGRITY_FAILED",
      "UPLOAD_COMPLETE",
      { recovery: "RECONCILE" },
    );
  }
  const idempotencyKey = await deriveUploadIdempotencyKey({
    label: "COMPLETE",
    requestBindingSha256: checkpoint.requestBindingSha256,
    sessionId: checkpoint.sessionId,
    crypto: dependencies.crypto,
    signal: request.signal,
  });
  if (!idempotencyKey.ok) return idempotencyKey;
  const completed = await callWithRetry(
    dependencies,
    "UPLOAD_COMPLETE",
    request.signal,
    async (attemptSignal) =>
      await dependencies.controlPlane.complete({
        protocol: RESUMABLE_UPLOAD_PROTOCOL,
        sessionId: checkpoint.sessionId,
        requestBindingSha256: checkpoint.requestBindingSha256,
        fingerprint: checkpoint.fingerprint,
        orderedParts,
        idempotencyKey: idempotencyKey.value,
        signal: attemptSignal,
      }),
  );
  if (!completed.ok) return stripProviderFailure(completed);
  const value = completed.value;
  if (
    !exactKeys(value, [
      "state",
      "protocol",
      "sessionId",
      "requestBindingSha256",
      "fingerprint",
      "resourceId",
    ]) ||
    value.state !== "QUARANTINED" ||
    value.protocol !== RESUMABLE_UPLOAD_PROTOCOL ||
    value.sessionId !== checkpoint.sessionId ||
    value.requestBindingSha256 !== checkpoint.requestBindingSha256 ||
    !sameFingerprint(value.fingerprint, checkpoint.fingerprint) ||
    !SAFE_OPAQUE_ID.test(value.resourceId)
  ) {
    return browserDataFailure(
      "INTEGRITY_FAILED",
      "UPLOAD_COMPLETE",
      { recovery: "RECONCILE" },
    );
  }
  const removed = await dependencies.checkpoints.remove({
    uploadKey: checkpoint.uploadKey,
    expectedRevision: checkpoint.revision,
    signal: request.signal,
  });
  if (!removed.ok) return remapResult(removed, "UPLOAD_RECONCILE");
  return browserDataSuccess(
    quarantinedOutcome(value.resourceId, sessionFromCheckpoint(checkpoint), false),
  );
}

async function executeAbort<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  uploadKey: string,
  signal: AbortSignal,
): Promise<BrowserDataResult<UploadAbortOutcome>> {
  const found = await dependencies.checkpoints.read(uploadKey, signal);
  if (!found.ok) return remapResult(found, "UPLOAD_ABORT");
  if (!found.value) {
    return browserDataSuccess(Object.freeze({ state: "NOT_FOUND" }));
  }
  let checkpoint = found.value;
  if (checkpoint.state === "ACTIVE") {
    const now = safeNow(dependencies, "UPLOAD_ABORT");
    if (!now.ok) return now;
    const pending = checkpointFrom({
      previous: checkpoint,
      revision: checkpoint.revision + 1,
      state: "ABORT_PENDING",
      updatedAtEpochMs: now.value,
    });
    const saved = await dependencies.checkpoints.compareAndSwap({
      expectedRevision: checkpoint.revision,
      checkpoint: pending,
      signal,
    });
    if (!saved.ok) return remapResult(saved, "UPLOAD_ABORT");
    checkpoint = saved.value;
  }
  const idempotencyKey = await deriveUploadIdempotencyKey({
    label: "ABORT",
    requestBindingSha256: checkpoint.requestBindingSha256,
    sessionId: checkpoint.sessionId,
    crypto: dependencies.crypto,
    signal,
  });
  if (!idempotencyKey.ok) return idempotencyKey;
  const aborted = await callWithRetry(
    dependencies,
    "UPLOAD_ABORT",
    signal,
    async (attemptSignal) =>
      await dependencies.controlPlane.abort({
        protocol: RESUMABLE_UPLOAD_PROTOCOL,
        sessionId: checkpoint.sessionId,
        requestBindingSha256: checkpoint.requestBindingSha256,
        idempotencyKey: idempotencyKey.value,
        signal: attemptSignal,
      }),
  );
  if (!aborted.ok) {
    if (
      aborted.error.code === "NOT_FOUND" ||
      aborted.error.code === "EXPIRED_RESOURCE"
    ) {
      const removed = await dependencies.checkpoints.remove({
        uploadKey,
        expectedRevision: checkpoint.revision,
        signal,
      });
      return removed.ok
        ? browserDataSuccess(
            Object.freeze({ state: "ORPHANED" as const }),
          )
        : remapResult(removed, "UPLOAD_ABORT");
    }
    return stripProviderFailure(aborted);
  }
  if (
    !exactKeys(aborted.value, ["state"]) ||
    ![
      "ABORTED",
      "NOT_FOUND",
      "EXPIRED",
      "ALREADY_COMPLETED",
    ].includes(aborted.value.state)
  ) {
    return browserDataFailure("CORRUPT_DATA", "UPLOAD_ABORT", {
      recovery: "RECONCILE",
    });
  }
  const removed = await dependencies.checkpoints.remove({
    uploadKey,
    expectedRevision: checkpoint.revision,
    signal,
  });
  if (!removed.ok) return remapResult(removed, "UPLOAD_ABORT");
  const state: UploadAbortOutcome["state"] =
    aborted.value.state === "ABORTED"
      ? "ABORTED"
      : aborted.value.state === "ALREADY_COMPLETED"
        ? "ALREADY_COMPLETED"
        : "ORPHANED";
  return browserDataSuccess(Object.freeze({ state }));
}

async function callWithRetry<Capability, Value>(
  dependencies: RuntimeDependencies<Capability>,
  operation: BrowserDataOperation,
  signal: AbortSignal,
  action: (
    attemptSignal: AbortSignal,
  ) => Promise<UploadProviderResult<Value>>,
): Promise<UploadProviderResult<Value>> {
  let lastFailure: UploadProviderFailure | null = null;
  for (
    let attempt = 0;
    attempt <= dependencies.policy.maxRetries;
    attempt += 1
  ) {
    if (signal.aborted) {
      return providerFailure("ABORTED", operation, false, "NONE");
    }
    if (attempt > 0 && lastFailure) {
      const delayed = await waitForRetry(
        dependencies,
        lastFailure,
        attempt - 1,
        signal,
        operation,
      );
      if (!delayed.ok) {
        return Object.freeze({ ok: false, error: delayed.error });
      }
    }
    const result = await invokeProviderAttempt(
      dependencies,
      operation,
      signal,
      action,
    );
    if (result.ok) {
      observeTerminal(dependencies.observer, operation, result);
      return result;
    }
    lastFailure = result.error;
    if (!canRetry(result.error, attempt, dependencies.policy)) {
      observeTerminal(dependencies.observer, operation, result);
      return result;
    }
  }
  const exhausted = Object.freeze({
    ok: false,
    error:
      lastFailure ??
      providerFailureValue(
        "UNAVAILABLE",
        operation,
        false,
        "RESUME",
      ),
  }) as UploadProviderResult<Value>;
  observeTerminal(dependencies.observer, operation, exhausted);
  return exhausted;
}

async function invokeProviderAttempt<Capability, Value>(
  dependencies: RuntimeDependencies<Capability>,
  operation: BrowserDataOperation,
  parentSignal: AbortSignal,
  action: (
    attemptSignal: AbortSignal,
  ) => Promise<UploadProviderResult<Value>>,
): Promise<UploadProviderResult<Value>> {
  if (parentSignal.aborted) {
    return providerFailure("ABORTED", operation, false, "NONE");
  }
  const controller = new AbortController();
  let timer: ReturnType<typeof setTimeout> | undefined;
  let releaseAbort = () => {};
  const deadline = new Promise<UploadProviderResult<Value>>((resolve) => {
    const abort = () => {
      controller.abort();
      resolve(providerFailure("ABORTED", operation, false, "NONE"));
    };
    parentSignal.addEventListener("abort", abort, { once: true });
    releaseAbort = () =>
      parentSignal.removeEventListener("abort", abort);
    timer = setTimeout(() => {
      controller.abort();
      resolve(
        providerFailure("UNAVAILABLE", operation, true, "RESUME"),
      );
    }, dependencies.policy.providerAttemptTimeoutMs);
  });
  try {
    // TR-04. The raw promise enters the physical registry the moment the
    // provider is called and stays there until it truly settles. Racing it
    // against a deadline let the bounded wrapper settle first and leave the
    // set empty, so `dispose()` reported a drained runtime while the provider
    // was still running.
    const raw = action(controller.signal);
    const tracked = Promise.resolve(raw).then(
      () => undefined,
      () => undefined,
    );
    dependencies.physicalTasks.add(tracked);
    void tracked.finally(() => dependencies.physicalTasks.delete(tracked));
    return await Promise.race([
      invokeProvider(operation, () => raw),
      deadline,
    ]);
  } finally {
    if (timer) clearTimeout(timer);
    releaseAbort();
  }
}

async function invokeProvider<Value>(
  operation: BrowserDataOperation,
  action: () => Promise<UploadProviderResult<Value>>,
): Promise<UploadProviderResult<Value>> {
  try {
    const result = await action();
    if (!result || typeof result !== "object") {
      return providerFailure("UNAVAILABLE", operation, true, "RESUME");
    }
    if (result.ok === true) return result;
    if (result.ok !== false) {
      return providerFailure("UNAVAILABLE", operation, true, "RESUME");
    }
    const normalized = normalizeProviderFailure(result.error, operation);
    return Object.freeze({ ok: false, error: normalized });
  } catch {
    return providerFailure("UNAVAILABLE", operation, true, "RESUME");
  }
}

async function waitForRetry<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  failure: UploadProviderFailure,
  retryIndex: number,
  signal: AbortSignal,
  operation: BrowserDataOperation,
): Promise<BrowserDataResult<void>> {
  const retryAfter =
    Number.isSafeInteger(failure.retryAfterMs) &&
    (failure.retryAfterMs ?? -1) >= 0 &&
    (failure.retryAfterMs ?? 0) <= dependencies.policy.maxRetryAfterMs
      ? (failure.retryAfterMs ?? 0)
      : null;
  const randomValue = safeRandom(dependencies);
  if (!randomValue.ok) return randomValue;
  const exponential = Math.min(
    dependencies.policy.retryMaxDelayMs,
    dependencies.policy.retryBaseDelayMs * 2 ** retryIndex,
  );
  const jittered = Math.floor(exponential * (0.5 + randomValue.value / 2));
  const delayMs = Math.max(jittered, retryAfter ?? 0);
  try {
    await dependencies.sleep(delayMs, signal);
    return signal.aborted
      ? browserDataFailure("ABORTED", operation)
      : browserDataSuccess(undefined);
  } catch {
    return signal.aborted
      ? browserDataFailure("ABORTED", operation)
      : browserDataFailure("UNAVAILABLE", operation, {
          retryable: true,
          recovery: "RESUME",
        });
  }
}

function canRetry(
  failure: UploadProviderFailure,
  attempt: number,
  policy: ResumableUploadRuntimePolicy,
): boolean {
  return (
    failure.retryable &&
    attempt < policy.maxRetries &&
    (failure.retryAfterMs === undefined ||
      (Number.isSafeInteger(failure.retryAfterMs) &&
        failure.retryAfterMs >= 0 &&
        failure.retryAfterMs <= policy.maxRetryAfterMs))
  );
}

function validatePartCapability<Capability>(
  value: UploadPartCapability<Capability>,
  checkpoint: ResumableUploadCheckpoint,
  uploadBindingSha256: string,
  dependencies: RuntimeDependencies<Capability>,
): BrowserDataResult<UploadPartCapability<Capability>> {
  if (
    !exactKeys(value, [
      "capability",
      "uploadBindingSha256",
      "expiresAtEpochMs",
    ]) ||
    value.uploadBindingSha256 !== uploadBindingSha256 ||
    !SHA256_HEX.test(value.uploadBindingSha256) ||
    !Number.isSafeInteger(value.expiresAtEpochMs)
  ) {
    return browserDataFailure("CORRUPT_DATA", "UPLOAD_PART", {
      recovery: "REISSUE_CAPABILITY",
    });
  }
  const now = safeNow(dependencies, "UPLOAD_PART");
  if (!now.ok) return now;
  if (
    value.expiresAtEpochMs <=
    now.value + dependencies.policy.capabilityRefreshSkewMs
  ) {
    return browserDataFailure("EXPIRED_RESOURCE", "UPLOAD_PART", {
      retryable: true,
      recovery: "REISSUE_CAPABILITY",
    });
  }
  if (value.expiresAtEpochMs > checkpoint.sessionExpiresAtEpochMs) {
    return browserDataFailure("POLICY_REJECTED", "UPLOAD_PART", {
      recovery: "REISSUE_CAPABILITY",
    });
  }
  return browserDataSuccess(value);
}

function validateStatus<Capability>(
  value: UploadSessionStatus,
  checkpoint: ResumableUploadCheckpoint,
  manifest: UploadPartManifest,
  dependencies: RuntimeDependencies<Capability>,
): BrowserDataResult<UploadSessionStatus> {
  if (!value || typeof value !== "object" || Array.isArray(value)) {
    return browserDataFailure("CORRUPT_DATA", "UPLOAD_RECONCILE", {
      recovery: "RECONCILE",
    });
  }
  if (value.state === "ACTIVE") {
    if (!exactKeys(value, ["state", "session", "acceptedParts"])) {
      return browserDataFailure(
        "CORRUPT_DATA",
        "UPLOAD_RECONCILE",
        { recovery: "RECONCILE" },
      );
    }
    const session = validateSession(
      value.session,
      checkpoint.requestBindingSha256,
      manifest.fingerprint,
      dependencies,
      true,
    );
    if (
      !session.ok ||
      session.value.sessionId !== checkpoint.sessionId ||
      session.value.maxConcurrency !== checkpoint.sessionMaxConcurrency ||
      session.value.expiresAtEpochMs !==
        checkpoint.sessionExpiresAtEpochMs
    ) {
      return browserDataFailure(
        "INTEGRITY_FAILED",
        "UPLOAD_RECONCILE",
        { recovery: "RECONCILE" },
      );
    }
    const receipts = validateServerReceipts(
      value,
      manifest,
      "UPLOAD_RECONCILE",
    );
    if (!receipts.ok) return receipts;
    return browserDataSuccess(
      Object.freeze({
        state: "ACTIVE" as const,
        session: session.value,
        acceptedParts: receipts.value,
      }),
    );
  }
  if (value.state === "QUARANTINED") {
    if (!exactKeys(value, ["state", "session", "resourceId"])) {
      return browserDataFailure(
        "CORRUPT_DATA",
        "UPLOAD_RECONCILE",
        { recovery: "RECONCILE" },
      );
    }
    const session = validateSession(
      value.session,
      checkpoint.requestBindingSha256,
      manifest.fingerprint,
      dependencies,
      false,
    );
    if (
      !session.ok ||
      session.value.sessionId !== checkpoint.sessionId ||
      session.value.maxConcurrency !== checkpoint.sessionMaxConcurrency ||
      session.value.expiresAtEpochMs !==
        checkpoint.sessionExpiresAtEpochMs ||
      !SAFE_OPAQUE_ID.test(value.resourceId)
    ) {
      return browserDataFailure(
        "INTEGRITY_FAILED",
        "UPLOAD_RECONCILE",
        { recovery: "RECONCILE" },
      );
    }
    return browserDataSuccess(
      Object.freeze({
        state: "QUARANTINED" as const,
        session: session.value,
        resourceId: value.resourceId,
      }),
    );
  }
  if (
    !["ABORTED", "EXPIRED", "NOT_FOUND"].includes(value.state) ||
    !exactKeys(value, [
      "state",
      "protocol",
      "sessionId",
      "requestBindingSha256",
    ]) ||
    value.protocol !== RESUMABLE_UPLOAD_PROTOCOL ||
    value.sessionId !== checkpoint.sessionId ||
    value.requestBindingSha256 !== checkpoint.requestBindingSha256
  ) {
    return browserDataFailure("INTEGRITY_FAILED", "UPLOAD_RECONCILE", {
      recovery: "RECONCILE",
    });
  }
  return browserDataSuccess(
    Object.freeze({
      state: value.state,
      protocol: RESUMABLE_UPLOAD_PROTOCOL,
      sessionId: value.sessionId,
      requestBindingSha256: value.requestBindingSha256,
    }),
  );
}

function validateSession<Capability>(
  value: UploadSession,
  requestBindingSha256: string,
  fingerprint: UploadFileFingerprint,
  dependencies: RuntimeDependencies<Capability>,
  requireActiveExpiry = true,
): BrowserDataResult<UploadSession> {
  if (
    !exactKeys(value, [
      "protocol",
      "sessionId",
      "requestBindingSha256",
      "fingerprint",
      "partSizeBytes",
      "partCount",
      "maxConcurrency",
      "expiresAtEpochMs",
    ]) ||
    value.protocol !== RESUMABLE_UPLOAD_PROTOCOL ||
    !SAFE_OPAQUE_ID.test(value.sessionId) ||
    value.requestBindingSha256 !== requestBindingSha256 ||
    !SHA256_HEX.test(value.requestBindingSha256) ||
    !isUploadFileFingerprint(value.fingerprint) ||
    !sameFingerprint(value.fingerprint, fingerprint) ||
    value.partSizeBytes !== fingerprint.partSizeBytes ||
    value.partCount !== fingerprint.partCount ||
    !Number.isSafeInteger(value.maxConcurrency) ||
    value.maxConcurrency < 1 ||
    value.maxConcurrency > dependencies.policy.maxConcurrency ||
    !Number.isSafeInteger(value.expiresAtEpochMs)
  ) {
    return browserDataFailure("INTEGRITY_FAILED", "UPLOAD_SESSION", {
      recovery: "RECONCILE",
    });
  }
  const now = safeNow(dependencies, "UPLOAD_SESSION");
  if (!now.ok) return now;
  if (
    requireActiveExpiry &&
    (value.expiresAtEpochMs <= now.value ||
      value.expiresAtEpochMs - now.value >
        dependencies.policy.maxSessionLifetimeMs)
  ) {
    return browserDataFailure("EXPIRED_RESOURCE", "UPLOAD_SESSION", {
      recovery: "RESTART",
    });
  }
  return browserDataSuccess(
    Object.freeze({
      ...value,
      fingerprint: Object.freeze({ ...value.fingerprint }),
    }),
  );
}

function validateServerReceipts(
  status: ActiveUploadStatus,
  manifest: UploadPartManifest,
  operation: BrowserDataOperation,
): BrowserDataResult<readonly UploadPartReceipt[]> {
  if (
    !Array.isArray(status.acceptedParts) ||
    status.acceptedParts.length > manifest.parts.length
  ) {
    return browserDataFailure("CORRUPT_DATA", operation, {
      recovery: "RECONCILE",
    });
  }
  let previousPart = 0;
  for (const receipt of status.acceptedParts) {
    if (
      !isUploadPartReceipt(receipt) ||
      receipt.partNumber <= previousPart ||
      !verifyPartAgainstManifest(receipt, manifest)
    ) {
      return browserDataFailure("INTEGRITY_FAILED", operation, {
        recovery: "RECONCILE",
      });
    }
    previousPart = receipt.partNumber;
  }
  return browserDataSuccess(snapshotReceipts(status.acceptedParts));
}

function checkpointFrom(
  input:
    | Readonly<{
        revision: number;
        protocol: typeof RESUMABLE_UPLOAD_PROTOCOL;
        state: "ACTIVE" | "ABORT_PENDING";
        uploadKey: string;
        requestBindingSha256: string;
        fingerprint: UploadFileFingerprint;
        sessionId: string;
        sessionExpiresAtEpochMs: number;
        sessionMaxConcurrency: number;
        acceptedParts: readonly UploadPartReceipt[];
        updatedAtEpochMs: number;
      }>
    | Readonly<{
        previous: ResumableUploadCheckpoint;
        revision: number;
        state?: "ACTIVE" | "ABORT_PENDING";
        acceptedParts?: readonly UploadPartReceipt[];
        updatedAtEpochMs: number;
      }>,
): ResumableUploadCheckpoint {
  const candidate =
    "previous" in input
      ? {
          ...input.previous,
          revision: input.revision,
          state: input.state ?? input.previous.state,
          acceptedParts:
            input.acceptedParts ?? input.previous.acceptedParts,
          updatedAtEpochMs: input.updatedAtEpochMs,
        }
      : input;
  const checkpoint: ResumableUploadCheckpoint = Object.freeze({
    ...candidate,
    schemaVersion: 1,
    fingerprint: Object.freeze({ ...candidate.fingerprint }),
    acceptedParts: snapshotReceipts(candidate.acceptedParts),
  });
  if (!isResumableUploadCheckpoint(checkpoint)) {
    throw new TypeError("Upload checkpoint construction failed.");
  }
  return checkpoint;
}

function sessionFromCheckpoint(
  checkpoint: ResumableUploadCheckpoint,
): UploadSession {
  return Object.freeze({
    protocol: RESUMABLE_UPLOAD_PROTOCOL,
    sessionId: checkpoint.sessionId,
    requestBindingSha256: checkpoint.requestBindingSha256,
    fingerprint: checkpoint.fingerprint,
    partSizeBytes: checkpoint.fingerprint.partSizeBytes,
    partCount: checkpoint.fingerprint.partCount,
    maxConcurrency: checkpoint.sessionMaxConcurrency,
    expiresAtEpochMs: checkpoint.sessionExpiresAtEpochMs,
  });
}

function quarantinedOutcome(
  resourceId: string,
  session: UploadSession,
  replayed: boolean,
): QuarantinedUpload {
  return Object.freeze({
    state: "QUARANTINED",
    resourceId,
    byteLength: session.fingerprint.byteLength,
    replayed,
  });
}

function snapshotRequest(
  input: ResumableUploadRequest,
  policy: ResumableUploadRuntimePolicy,
): UploadRequestSnapshot {
  if (
    !input ||
    typeof input !== "object" ||
    typeof input.uploadKey !== "string" ||
    !SAFE_UPLOAD_KEY.test(input.uploadKey) ||
    typeof input.purpose !== "string" ||
    !SAFE_REGISTRY_ID.test(input.purpose) ||
    typeof input.mediaType !== "string" ||
    !MEDIA_TYPE.test(input.mediaType) ||
    !isAbortSignal(input.signal) ||
    (input.onProgress !== undefined &&
      typeof input.onProgress !== "function")
  ) {
    throw new TypeError("Upload request is invalid.");
  }
  const source = snapshotUploadSource(input.source);
  if (
    source.byteLength > policy.maxFileBytes ||
    Math.ceil(source.byteLength / policy.partSizeBytes) >
      policy.maxPartCount
  ) {
    throw new TypeError("Upload request exceeds policy.");
  }
  return Object.freeze({
    uploadKey: input.uploadKey,
    purpose: input.purpose,
    mediaType: input.mediaType,
    source,
    signal: input.signal,
    ...(input.onProgress ? { onProgress: input.onProgress } : {}),
  });
}

function snapshotDependencies<Capability>(
  input: ResumableUploadRuntimeDependencies<Capability>,
  physicalTasks: Set<Promise<unknown>>,
): RuntimeDependencies<Capability> {
  const policy = resolveResumableUploadRuntimePolicy(input.policy);
  const controlPlane = snapshotControlPlane(input.controlPlane);
  const partExecutor = snapshotPartExecutor(input.partExecutor);
  const checkpoints = snapshotCheckpointStore(input.checkpoints);
  const mutationLock = snapshotMutationLock(input.mutationLock);
  const crossContextCancellation = snapshotCancellationChannel(
    input.crossContextCancellation,
  );
  const crypto = snapshotUploadCrypto(input.crypto);
  const nowSource = input.now ?? Date.now;
  const randomSource = input.random ?? Math.random;
  const sleepSource = input.sleep ?? abortableSleep;
  const observer = snapshotObserver(input.observer);
  if (
    typeof nowSource !== "function" ||
    typeof randomSource !== "function" ||
    typeof sleepSource !== "function"
  ) {
    throw new TypeError("Upload runtime dependency is invalid.");
  }
  return Object.freeze({
    physicalTasks,
    controlPlane,
    partExecutor,
    checkpoints,
    mutationLock,
    ...(crossContextCancellation
      ? { crossContextCancellation }
      : {}),
    crypto,
    policy,
    now: () => nowSource(),
    random: () => randomSource(),
    sleep: async (delayMs, signal) =>
      await sleepSource(delayMs, signal),
    ...(observer ? { observer } : {}),
  });
}

function snapshotCancellationChannel(
  value: UploadCancellationChannel | undefined,
): UploadCancellationChannel | undefined {
  if (!value) return undefined;
  const publish = value.publish;
  const subscribe = value.subscribe;
  const close = value.close;
  if (
    typeof publish !== "function" ||
    typeof subscribe !== "function" ||
    typeof close !== "function"
  ) {
    throw new TypeError(
      "Upload cancellation channel is invalid.",
    );
  }
  return Object.freeze({
    publish(uploadKey: string): boolean {
      try {
        return publish.call(value, uploadKey) === true;
      } catch {
        return false;
      }
    },
    subscribe(
      listener: (uploadKey: string) => void,
    ): () => void {
      const release = subscribe.call(value, listener);
      if (typeof release !== "function") {
        throw new TypeError(
          "Upload cancellation subscription is invalid.",
        );
      }
      let active = true;
      return () => {
        if (!active) return;
        active = false;
        try {
          release();
        } catch {
          // Runtime close remains terminal.
        }
      };
    },
    close(): void {
      try {
        close.call(value);
      } catch {
        // Runtime close remains terminal.
      }
    },
  });
}

function snapshotObserver(
  value: BrowserDataObserver | undefined,
): BrowserDataObserver | undefined {
  if (!value) return undefined;
  const record = value.record;
  if (typeof record !== "function") {
    throw new TypeError("Upload observer is invalid.");
  }
  return Object.freeze({
    record(
      observation: Parameters<BrowserDataObserver["record"]>[0],
    ) {
      record.call(value, Object.freeze({ ...observation }));
    },
  });
}

function snapshotControlPlane<Capability>(
  value: ResumableUploadControlPlane<Capability>,
): ResumableUploadControlPlane<Capability> {
  const createSession = value?.createSession;
  const getStatus = value?.getStatus;
  const issuePartCapability = value?.issuePartCapability;
  const complete = value?.complete;
  const abort = value?.abort;
  if (
    typeof createSession !== "function" ||
    typeof getStatus !== "function" ||
    typeof issuePartCapability !== "function" ||
    typeof complete !== "function" ||
    typeof abort !== "function"
  ) {
    throw new TypeError("Upload control plane is invalid.");
  }
  const snapshot: ResumableUploadControlPlane<Capability> = {
    async createSession(
      input: Parameters<
        ResumableUploadControlPlane<Capability>["createSession"]
      >[0],
    ) {
      return await createSession.call(value, input);
    },
    async getStatus(
      input: Parameters<
        ResumableUploadControlPlane<Capability>["getStatus"]
      >[0],
    ) {
      return await getStatus.call(value, input);
    },
    async issuePartCapability(
      input: Parameters<
        ResumableUploadControlPlane<Capability>["issuePartCapability"]
      >[0],
    ) {
      return await issuePartCapability.call(value, input);
    },
    async complete(
      input: Parameters<
        ResumableUploadControlPlane<Capability>["complete"]
      >[0],
    ) {
      return await complete.call(value, input);
    },
    async abort(
      input: Parameters<
        ResumableUploadControlPlane<Capability>["abort"]
      >[0],
    ) {
      return await abort.call(value, input);
    },
  };
  return Object.freeze(snapshot);
}

function snapshotPartExecutor<Capability>(
  value: UploadPartExecutor<Capability>,
): UploadPartExecutor<Capability> {
  const uploadPart = value?.uploadPart;
  if (typeof uploadPart !== "function") {
    throw new TypeError("Upload part executor is invalid.");
  }
  const snapshot: UploadPartExecutor<Capability> = {
    async uploadPart(
      input: Parameters<UploadPartExecutor<Capability>["uploadPart"]>[0],
    ) {
      return await uploadPart.call(value, input);
    },
  };
  return Object.freeze(snapshot);
}

function snapshotCheckpointStore(
  value: ResumableUploadCheckpointStore,
): ResumableUploadCheckpointStore {
  const read = value?.read;
  const compareAndSwap = value?.compareAndSwap;
  const remove = value?.remove;
  const close = value?.close;
  if (
    typeof read !== "function" ||
    typeof compareAndSwap !== "function" ||
    typeof remove !== "function" ||
    typeof close !== "function"
  ) {
    throw new TypeError("Upload checkpoint store is invalid.");
  }
  const snapshot: ResumableUploadCheckpointStore = {
    async read(
      uploadKey: string,
      signal?: AbortSignal,
    ) {
      return await read.call(value, uploadKey, signal);
    },
    async compareAndSwap(
      input: Parameters<
        ResumableUploadCheckpointStore["compareAndSwap"]
      >[0],
    ) {
      return await compareAndSwap.call(value, input);
    },
    async remove(
      input: Parameters<ResumableUploadCheckpointStore["remove"]>[0],
    ) {
      return await remove.call(value, input);
    },
    close() {
      close.call(value);
    },
  };
  return Object.freeze(snapshot);
}

function snapshotMutationLock(value: UploadMutationLock): UploadMutationLock {
  const run = value?.run;
  if (typeof run !== "function") {
    throw new TypeError("Upload mutation lock is invalid.");
  }
  return Object.freeze({
    async run<Value>(
      uploadKey: string,
      signal: AbortSignal,
      task: () => Promise<Value>,
    ): Promise<Value> {
      return await (run.call(
        value,
        uploadKey,
        signal,
        task,
      ) as Promise<Value>);
    },
  });
}

function normalizeProviderFailure(
  input: UploadProviderFailure,
  operation: BrowserDataOperation,
): UploadProviderFailure {
  if (
    !input ||
    typeof input !== "object" ||
    !FAILURE_CODES.has(input.code) ||
    typeof input.retryable !== "boolean" ||
    !RECOVERIES.has(input.recovery) ||
    (input.retryAfterMs !== undefined &&
      (!Number.isSafeInteger(input.retryAfterMs) ||
        input.retryAfterMs < 0))
  ) {
    return providerFailureValue(
      "UNAVAILABLE",
      operation,
      true,
      "RESUME",
    );
  }
  return Object.freeze({
    code: input.code,
    operation,
    retryable: input.retryable,
    recovery: input.recovery,
    ...(input.retryAfterMs === undefined
      ? {}
      : { retryAfterMs: input.retryAfterMs }),
  });
}

function providerFailure(
  code: BrowserDataFailureCode,
  operation: BrowserDataOperation,
  retryable: boolean,
  recovery: BrowserDataRecovery,
): UploadProviderResult<never> {
  return Object.freeze({
    ok: false,
    error: providerFailureValue(
      code,
      operation,
      retryable,
      recovery,
    ),
  });
}

function providerFailureValue(
  code: BrowserDataFailureCode,
  operation: BrowserDataOperation,
  retryable: boolean,
  recovery: BrowserDataRecovery,
): UploadProviderFailure {
  return Object.freeze({ code, operation, retryable, recovery });
}

function stripProviderFailure<Value>(
  result: UploadProviderResult<Value>,
): BrowserDataResult<Value> {
  if (result.ok) return browserDataSuccess(result.value);
  return browserDataFailure(result.error.code, result.error.operation, {
    retryable: result.error.retryable,
    recovery: result.error.recovery,
  });
}

function remapResult<Value>(
  result: BrowserDataResult<Value>,
  operation: BrowserDataOperation,
): BrowserDataResult<Value> {
  return result.ok
    ? result
    : browserDataFailure(result.error.code, operation, {
        retryable: result.error.retryable,
        recovery: result.error.recovery,
      });
}

function safeNow<Capability>(
  dependencies: RuntimeDependencies<Capability>,
  operation: BrowserDataOperation,
): BrowserDataResult<number> {
  try {
    const value = dependencies.now();
    return Number.isSafeInteger(value) && value >= 0
      ? browserDataSuccess(value)
      : browserDataFailure("UNAVAILABLE", operation, {
          recovery: "RESUME",
        });
  } catch {
    return browserDataFailure("UNAVAILABLE", operation, {
      recovery: "RESUME",
    });
  }
}

function safeRandom<Capability>(
  dependencies: RuntimeDependencies<Capability>,
): BrowserDataResult<number> {
  try {
    const value = dependencies.random();
    return Number.isFinite(value) && value >= 0 && value <= 1
      ? browserDataSuccess(value)
      : browserDataFailure("UNAVAILABLE", "UPLOAD_RECONCILE", {
          recovery: "RESUME",
        });
  } catch {
    return browserDataFailure("UNAVAILABLE", "UPLOAD_RECONCILE", {
      recovery: "RESUME",
    });
  }
}

function sameFingerprint(
  left: UploadFileFingerprint,
  right: UploadFileFingerprint,
): boolean {
  return (
    left.algorithm === right.algorithm &&
    left.digestHex === right.digestHex &&
    left.byteLength === right.byteLength &&
    left.partSizeBytes === right.partSizeBytes &&
    left.partCount === right.partCount
  );
}

function snapshotReceipts(
  parts: readonly UploadPartReceipt[],
): readonly UploadPartReceipt[] {
  return Object.freeze(
    parts.map((part) => Object.freeze({ ...part })),
  );
}

function exactKeys(
  value: unknown,
  keys: readonly string[],
): value is Record<string, unknown> {
  if (!value || typeof value !== "object" || Array.isArray(value)) {
    return false;
  }
  const actual = Object.keys(value).sort();
  const expected = [...keys].sort();
  return (
    actual.length === expected.length &&
    actual.every((key, index) => key === expected[index])
  );
}

function reportProgress(
  request: UploadRequestSnapshot,
  phase: TransferProgress["phase"],
  transferredBytes: number,
): void {
  try {
    request.onProgress?.(
      Object.freeze({
        phase,
        transferredBytes,
        totalBytes: request.source.byteLength,
      }),
    );
  } catch {
    // Progress observation is best-effort and data-free.
  }
}

function observeTerminal<Value>(
  observer: BrowserDataObserver | undefined,
  operation: BrowserDataOperation,
  result: BrowserDataResult<Value> | UploadProviderResult<Value>,
  byteLength?: number,
): void {
  try {
    observer?.record(
      Object.freeze({
        operation,
        outcome: result.ok ? "SUCCEEDED" : "FAILED",
        ...(result.ok ? {} : { failureCode: result.error.code }),
        ...(byteLength === undefined
          ? {}
          : { byteBucket: transferByteBucket(byteLength) }),
      }),
    );
  } catch {
    // Upload correctness is independent from best-effort observation.
  }
}

function transferByteBucket(
  byteLength: number,
): NonNullable<
  Parameters<BrowserDataObserver["record"]>[0]["byteBucket"]
> {
  if (byteLength === 0) return "ZERO";
  if (byteLength < 1024 * 1024) return "LT1MIB";
  if (byteLength < 10 * 1024 * 1024) return "1_TO_9MIB";
  if (byteLength < 100 * 1024 * 1024) return "10_TO_99MIB";
  return "GTE100MIB";
}

function mapLockFailure(
  error: unknown,
  operation: BrowserDataOperation,
): BrowserDataResult<never> {
  if (error instanceof DOMException && error.name === "AbortError") {
    return browserDataFailure("ABORTED", operation);
  }
  if (
    error instanceof DOMException &&
    error.name === "InvalidStateError"
  ) {
    return browserDataFailure("BLOCKED", operation, {
      retryable: true,
      recovery: "RESUME",
    });
  }
  return browserDataFailure("UNAVAILABLE", operation, {
    retryable: true,
    recovery: "RESUME",
  });
}

function combineAbortSignals(
  first: AbortSignal,
  second: AbortSignal,
): Readonly<{ signal: AbortSignal; release(): void }> {
  const controller = new AbortController();
  const abort = () => controller.abort();
  if (first.aborted || second.aborted) {
    controller.abort();
  } else {
    first.addEventListener("abort", abort, { once: true });
    second.addEventListener("abort", abort, { once: true });
  }
  return Object.freeze({
    signal: controller.signal,
    release() {
      first.removeEventListener("abort", abort);
      second.removeEventListener("abort", abort);
    },
  });
}

function isAbortSignal(value: unknown): value is AbortSignal {
  return Boolean(
    value &&
      typeof value === "object" &&
      typeof (value as AbortSignal).aborted === "boolean" &&
      typeof (value as AbortSignal).addEventListener === "function" &&
      typeof (value as AbortSignal).removeEventListener === "function",
  );
}

async function abortableSleep(
  delayMs: number,
  signal: AbortSignal,
): Promise<void> {
  if (signal.aborted) {
    throw new DOMException("The operation was aborted.", "AbortError");
  }
  await new Promise<void>((resolve, reject) => {
    const timer = setTimeout(finish, delayMs);
    function finish() {
      signal.removeEventListener("abort", abort);
      resolve();
    }
    function abort() {
      clearTimeout(timer);
      signal.removeEventListener("abort", abort);
      reject(new DOMException("The operation was aborted.", "AbortError"));
    }
    signal.addEventListener("abort", abort, { once: true });
  });
}