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 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
Section titled “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) carries the same seven fields in both SDKs, spelled per language: checkpointId, repoSeq, url, byteLength, sha256, contentType, and expiresAt in TypeScript; checkpoint_id, repo_seq, url, byte_length, sha256, content_type, and expires_at in Python, where the dataclass also carries an extra mapping holding any keys the server sent that the contract does not name. Validate bytes against the descriptor when your downloader does not already do so. In the TypeScript SDK, expiresAt is normalized to a Date; the Python value is left as an epoch-millisecond number (see Read and access). 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 the descriptor’s expiry field as authoritative — expiresAt in TypeScript, expires_at in Python — and re-request access rather than caching a URL.
Resuming an interrupted export
Section titled “Resuming an interrupted export”A broken export download resumes SDK-side, not through the CLI’s --resume, and only on the direct streaming path. This is the TypeScript surface; the Python client exposes its own export helpers described below.
- Pass the
sessionthe broken stream wrote, or rebuild one from the bytes you saved withrestoreRepositoryExportSession(..., { partial: true }).partialis what accepts a file that stops before the trailer, as an interrupted download always does — without it the restore rejects the bytes as truncated. - A session carries its own cursor.
afterDurableIdis the fallback when it has none, and either way the session must carry the opening header. - A prepared export has no mid-point to resume from, because the token is redeemed as one stored object. Passing a cursor alongside a token or
prefer: 'async'is rejected rather than ignored — start a new session and redeem the token again.
Python
Section titled “Python”Install the package and configure auth:
pip install warmhubWarmHubClient.from_env() reads two environment variables:
WH_TOKEN— your WarmHub token (requires at minimumrepo:readandrepo:checkpoint-readfor checkpoint operations)WARMHUB_API_URL— the API base URL (e.g.https://api.warmhub.ai)
Read and access
Section titled “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.
In the Python SDK, RepositoryCheckpointAccess.expires_at is an integer
epoch-millisecond value; it is not converted to a datetime.
Verify offline
Section titled “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 categoryelse: 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.
The Python verifier accepts only archives whose manifest carries
formatVersion: 4. Passing an archive with any other format version raises
RepositoryCheckpointVerificationError with reason == 'manifest_invalid'.
Async client
Section titled “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", )