Repository Checkpoints

A repository checkpoint packages a repo’s contents as a downloadable point-in-time archive captured at one exact repoSeq. You can verify it offline. This page covers checkpoint lifecycle operations for both the TypeScript and Python SDKs. The TypeScript client ships as @warmhub/sdk-ts — see the SDK overview for installation and auth setup. The Python client ships as the warmhub package — see Python setup below for installation and auth details.

The SDK exposes repository checkpoint lifecycle operations at client.repo.checkpoint. It returns status records and signed-access descriptors; it never downloads checkpoint bytes for you.

import { WarmHubClient } from '@warmhub/sdk-ts'
import { verifyRepositoryCheckpointArchive } from '@warmhub/sdk-ts/checkpoint'
import { createHash } from 'node:crypto'
import { createReadStream } from 'node:fs'
import { writeFile } from 'node:fs/promises'

const client = new WarmHubClient({
  apiUrl: 'https://api.warmhub.ai',
  auth: { getToken: async () => process.env.WH_TOKEN },
})

let checkpoint = await client.repo.checkpoint.generate('acme', 'widgets')
while (checkpoint.state === 'queued' || checkpoint.state === 'running') {
  await new Promise((resolve) => setTimeout(resolve, 1_000))
  checkpoint = await client.repo.checkpoint.status('acme', 'widgets', {
    checkpointId: checkpoint.checkpointId,
  })
}
if (checkpoint.state !== 'complete') {
  // Both 'attempts_exhausted' and 'deadline_exceeded' with nextAction === 'retry'
  // are retryable — call checkpoint.retry() to reset the budget and try again.
  // Other failure codes are terminal.
  if (checkpoint.nextAction === 'retry') {
    checkpoint = await client.repo.checkpoint.retry('acme', 'widgets', checkpoint.checkpointId)
    // retry() transitions the checkpoint back to 'queued'. Resume polling until
    // it reaches 'complete' before accessing the artifact.
    while (checkpoint.state === 'queued' || checkpoint.state === 'running') {
      await new Promise((resolve) => setTimeout(resolve, 1_000))
      checkpoint = await client.repo.checkpoint.status('acme', 'widgets', {
        checkpointId: checkpoint.checkpointId,
      })
    }
    if (checkpoint.state !== 'complete') {
      throw new Error(`Checkpoint failed after retry: ${checkpoint.failureCode}`)
    }
  } else {
    throw new Error(`Checkpoint failed: ${checkpoint.failureCode}`)
  }
}

const access = await client.repo.checkpoint.getAccess('acme', 'widgets', {
  checkpoint: { checkpointId: checkpoint.checkpointId },
  artifact: 'archive',
})

// access.url is short-lived. Fetch it directly, without a WarmHub
// Authorization header, and verify the descriptor before publishing bytes.
const response = await fetch(access.url)
if (!response.ok) throw new Error(`Checkpoint download failed: ${response.status}`)
const bytes = new Uint8Array(await response.arrayBuffer())
const sha256 = createHash('sha256').update(bytes).digest('hex')
if (bytes.byteLength !== access.byteLength || sha256 !== access.sha256) {
  throw new Error('Checkpoint download did not match its access descriptor')
}
await writeFile('checkpoint.zip', bytes, { flag: 'wx' })

const verified = await verifyRepositoryCheckpointArchive(
  createReadStream('checkpoint.zip'),
)

verified is a typed identity, count, and digest summary. The verifier reads only supplied archive bytes; it makes no WarmHub client, transport, or network request.

Lifecycle methods

Method	Result
client.repo.checkpoint.generate(org, repo, { atLeastRepoSeq? })	Requests a checkpoint and returns its current status.
client.repo.checkpoint.status(org, repo, { checkpointId } | { repoSeq })	Returns one checkpoint status.
client.repo.checkpoint.latest(org, repo)	Returns the latest completed checkpoint, or null.
client.repo.checkpoint.retry(org, repo, checkpointId)	Retries a failed checkpoint and returns its status. The checkpoint transitions back to queued; poll status(...) until state === 'complete' before calling getAccess(...).
client.repo.checkpoint.getAccess(org, repo, { checkpoint, artifact })	Returns a short-lived signed descriptor for an archive, manifest, or chunk.

Checkpoint status, latest, and access requests require unrestricted repo:read plus repo:checkpoint-read. Generate and retry also require repo:checkpoint-generate; repo:admin remains compatible for those operations but does not substitute for checkpoint-read. Newly issued owner/admin role PATs include the narrow capability; existing owner/admin PATs with repo:admin remain compatible. A missing checkpoint is reported as NOT_FOUND; do not treat that response as evidence about repositories for which the caller has no checkpoint authority.

getAccess accepts checkpoint: 'latest', { checkpointId }, or { repoSeq }. Its artifact may be 'archive', 'manifest', or { chunkPath }. The descriptor (RepositoryCheckpointAccess) contains checkpointId, repoSeq, url, byteLength, sha256, contentType, and expiresAt (converted to a Date); validate bytes against it when your downloader does not already do so. The signed URL is short-lived — one hour by default — and its lifetime is an environment-wide setting a caller cannot override per request, so treat expiresAt as authoritative and re-request access rather than caching a URL.

Python

Setup

Install the package and configure auth:

pip install warmhub

WarmHubClient.from_env() reads two environment variables:

• WH_TOKEN — your WarmHub token (requires at minimum repo:read and repo:checkpoint-read for checkpoint operations)
• WARMHUB_API_URL — the API base URL (e.g. https://api.warmhub.ai)

Generate and access

The Python SDK exposes the same control plane as client.repo.checkpoint. Use the repository handle when all calls target one repository:

import time

from warmhub import WarmHubClient

with WarmHubClient.from_env() as client:
    checkpoints = client.repository("acme/widgets").repo.checkpoint
    checkpoint = checkpoints.generate()
    while checkpoint.state in ("queued", "running"):
        time.sleep(1)
        checkpoint = checkpoints.status(checkpoint_id=checkpoint.checkpoint_id)

    latest = checkpoints.latest()
    if latest is not None:
        access = checkpoints.get_access(
            checkpoint_id=latest.checkpoint_id,
            artifact="archive",
        )

    if checkpoint.state == "failed" and checkpoint.next_action == "retry":
        checkpoint = checkpoints.retry(checkpoint_id=checkpoint.checkpoint_id)

Pass repo_seq= instead of checkpoint_id= to select a retained sequence. get_access(checkpoint="latest", artifact="manifest") selects the newest complete checkpoint; use artifact="chunk", chunk_path="..." for one chunk. Its URL is short-lived and must be fetched without a WarmHub authorization header. Verify the downloaded byte length and SHA-256 before using it.

Verify offline

The caller downloads access.url separately. Pass those local bytes to the verifier along with the identity and integrity facts from the access descriptor:

from warmhub import (
    RepositoryCheckpointVerificationError,
    RepositoryCheckpointVerificationExpected,
    verify_repository_checkpoint_archive,
)

expected = RepositoryCheckpointVerificationExpected(
    checkpoint_id=access.checkpoint_id,
    repo_seq=access.repo_seq,
    byte_length=access.byte_length,
    sha256=access.sha256,
)

try:
    verified = verify_repository_checkpoint_archive("checkpoint.zip", expected=expected)
except RepositoryCheckpointVerificationError as error:
    print(error.reason)  # stable failure category
else:
    print(verified.checkpoint_id, verified.record_count, verified.root_digest)

This is local/offline: it returns a typed identity, count, and digest summary, or raises RepositoryCheckpointVerificationError with a stable reason. It accepts a path, bytes-like object, binary file, or Iterable[bytes]. There is no async variant of the verifier; only the synchronous verify_repository_checkpoint_archive(...) is exported.

Async client

The async client surface covers the same checkpoint control-plane methods:

from warmhub import AsyncWarmHubClient

async with AsyncWarmHubClient.from_env() as client:
    checkpoints = client.repository("acme/widgets").repo.checkpoint
    checkpoint = await checkpoints.generate()
    latest = await checkpoints.latest()
    if latest is not None:
        access = await checkpoints.get_access(
            checkpoint_id=latest.checkpoint_id,
            artifact="archive",
        )

Offline archive verification

verifyRepositoryCheckpointArchive(stream) reads and verifies an archive locally: no WarmHub profile, SDK transport, or network connection is required. It returns the checkpoint identity, sequence, record count, total bytes, and root digest, or throws a RepositoryCheckpointVerificationError with a stable verification reason accessible as err.reason. The Python verify_repository_checkpoint_archive(...) surface has the same offline trust boundary and accepts optional access-descriptor facts through RepositoryCheckpointVerificationExpected; see Verify offline above for the full Python example.

Use this verifier after downloading an archive and before trusting it as an external snapshot.