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.

Checkpoint generation is retired

Nothing creates new checkpoints. client.repo.checkpoint.generate(...) and .retry(...) are gone from both SDKs. For a fresh snapshot, use client.repo.export(...) in the TypeScript SDK, or open_repository_export(...) in the Python SDK — the export format, delta, resume, and verification rules are documented under repository export. If you already have an export token minted by request_repository_export(...), redeem it with open_repository_export_token(...). Checkpoints already stored stay readable through the methods below until the announced cleanup.

The SDK exposes repository checkpoint read 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 },
})

const checkpoint = await client.repo.checkpoint.latest('acme', 'widgets')
if (checkpoint === null) {
  throw new Error('No stored checkpoint for acme/widgets')
}

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 — checkpoint identity, sequence, record count, total bytes, and root digest. The verifier reads only supplied archive bytes; it makes no WarmHub client, transport, or network request. On failure it throws RepositoryCheckpointVerificationError with a stable verification reason on err.reason. Pass the access descriptor’s facts as RepositoryCheckpointVerificationExpected to have the verifier check identity as well as integrity.

Read methods

Method	Result
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.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. Newly issued owner/admin role PATs include the narrow capability. 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)

Read and access

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

from warmhub import WarmHubClient

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

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. Use this verifier after downloading an archive and before trusting it as an external snapshot.

Async client

The async client surface covers the same checkpoint read methods:

from warmhub import AsyncWarmHubClient

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