Skip to content

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.

MethodResult
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.

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 session the broken stream wrote, or rebuild one from the bytes you saved with restoreRepositoryExportSession(..., { partial: true }). partial is 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. afterDurableId is 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.

Install the package and configure auth:

Terminal window
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)

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.

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.

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'.

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",
)