Metadata-Version: 2.4
Name: qualia-sdk
Version: 0.8.0rc6
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Dist: httpx[http2]>=0.25.0,<1
Requires-Dist: pydantic>=2.0.0
Requires-Dist: opentelemetry-api>=1.20.0
Requires-Dist: opentelemetry-instrumentation-httpx>=0.41b0
Summary: Python SDK for the Qualia robotics data and VLA platform
Keywords: qualia,vla,robotics,fine-tuning,machine-learning
Author: Qualia Studios
License-Expression: MIT
Requires-Python: >=3.10
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Documentation, https://docs.qualiastudios.dev
Project-URL: Homepage, https://qualiastudios.dev

# Qualia Python SDK

The official Python SDK for the [Qualia](https://app.qualiastudios.dev) robotics data and VLA platform.

## Draft a Task from person-picked episodes (staged)

Create a Task manually with `client.tasks.create("Pick a box")`, even without
any episodes. To ask for a proposal, first pick **every** reference yourself:
1–3 success takes and optionally 1–2 takes per named failure/recovery mode.
Nothing samples, suggests or drops references. Each episode is a full
`accounts/{account}/datasets/{dataset}/episodes/{episode}` name:

```python
task = client.tasks.create(
    "Pick a box",
    references=[
        {
            "role": "SUCCESS",
            "episode": "accounts/<account>/datasets/<dataset>/episodes/<episode>",
        },
    ],
)
price = client.task_drafts.estimate(task.name)  # no reservation or model call
run = client.task_drafts.create(task.name)  # reserves credits
result = client.task_drafts.wait(run.name)  # or get/list/cancel
if result.state == "SUCCEEDED":
    # Inspect result.proposal and its procedure; edit before saving if needed.
    saved = client.task_drafts.apply(result)
    # Publish only when a person explicitly decides to:
    # client.tasks.publish(saved.name, etag=saved.etag)
```

`max_cost=` is an optional credit cap for estimate/create. Drafting does not
change or publish the Task; `apply(..., edited=...)` saves reviewed text and
graph, and `publish` is separate. This API is staged: the `vlm-task-draft`
template is disabled until rollout authorization; calls to price/start fail
until it is enabled, and live billing/graph-check QA is still pending.

## Installation

```bash
pip install qualia-sdk
```

The native transfer engine behind `client.data.upload` / `download` is
bundled in the wheel — no extra step.

Hugging Face imports need no client extra: the SDK submits a pinned revision
and polls while the dataset streams server-side straight into storage.

Newer servers may add fields to their responses; an older SDK ignores the
fields it does not know. Request arguments stay strict, so a misspelt key
still fails before anything is sent.

Upload errors marked `X-Qualia-Retryable: false` stop after one request. Typed
ingest configuration and protocol failures show the gateway's readable cause in
Python and the CLI. The transfer exception retains `code`, `retryable`, `detail`
and `status_code`; ordinary unavailable and timeout failures still retry.
The advisory failure heartbeat sends the readable cause without Python class
names, internal gateway routes or serialized error objects. It remains one line
of at most 300 characters, with signed URL query strings removed. Reporting is
best-effort and never replaces the caller's original exception.

### 0.8.0rc6: set a handheld dataset up from a script

`client.datasets` gains `setup`, `poses`, `resolve_poses` and `wait_resolve`, and
`update(reach_settings=)`; `client.checks` gains `reach_status` and
`cancel_reach`. Together they do what the dashboard's Setup strip does (pick the
robot and the solver, resolve the poses at the TCP, run and follow the solver),
and the matching CLI is `qualia-cli==0.3.0rc6` (`qualia dataset setup`,
`qualia dataset poses`, `qualia check reach-status`). See
[Set up a handheld (UMI) dataset](#set-up-a-handheld-umi-dataset-qualiav1). They
need no server change.

### Upgrading to 0.8.0rc4

This source candidate adds saved upload destinations shared with the CLI's
local rig configuration. `Qualia(workspace=...)` selects the workspace;
`data.upload(..., merge_target=...)` chooses an existing merged dataset and
`standalone=True` overrides a saved destination for one upload. Saved choices
apply only to the matching API host, account and workspace. The server binds
the intent to the logical source upload across transfer retries; one a merge
revert released before merges were final stays released instead of silently
appending again.

The matching CLI is `qualia-cli==0.3.0rc4`. After these candidates are published,
install this SDK with `pip install 'qualia-sdk==0.8.0rc4'` and use a deployment
that supports the merged-upload lifecycle. The examples below show the scope
and override options. Merge history is in the dataset's Versions tab; a merge is final, and existing
frozen versions keep their source references.

### 0.8.0rc3: links to Tasks

Episodes, uploads and mixture sources can name the Task they are about
(`linked_task`, `accounts/{a}/tasks/{t}`), beside the free-text task label
training reads. Nothing is removed: `episodes.assign_task` takes
`linked_task=` (the label then defaults to the task's name, so `task` is
optional), `episodes.link_task` links or unlinks one episode,
`datasets.update(linked_task=)` links an upload, and
`mixtures.add_source(linked_task=)` follows every upload (and merged dataset)
linked to a task.
A task can still be deleted: its sources then read `state == "TASK_DELETED"`,
resolve to nothing, and are only removed (`mixtures.remove_source`).

An upload's Task is its episodes' too (QUA-3413): an episode with no link of its
own reads the upload's, so `datasets.update(linked_task=)` links every one in
a single request. Episodes that link the old Task themselves keep it unless you
pass `relink_episodes=True`; `episodes.assign_task` still overrides the
upload's Task for the episodes it names.
A merged dataset's Task is its own (QUA-3511): `datasets.update(linked_task=)`
sets it on the merged dataset alone, and its episodes with no link of their own
read it, as an upload's read the upload's (its uploads keep theirs, which
nothing reads any more); `relink_episodes=True` works there as on an upload.
Linking by label links a merged dataset by its own task label, as it links an
upload, and never an upload a merge took.
`episodes.assign_task` through it links each episode in its upload: all or none
per upload.

### Upgrading to 0.8.0rc2

Use `client.datasets.update(dataset, tags=["umi", "real-world", "human", "gello"])`
to change collection classifications through their saved metadata. Nonempty
custom-only input (`tags=["warehouse-a"]`), or unchanged classifications with
edited customs, replaces custom tags while preserving declarations and inference.
When classifications are present, echo them to remove all customs without declaring methods.
`tags=[]` clears customs and explicitly records Unknown in every collection dimension.
The SDK reads the owner and uses its etag unless one is supplied; stale writes fail.
These edits require a server that returns independent collection metadata.
`method` and `methods` remain compatibility views. Merged datasets own their
tags and keep them on append. See the [collection contract](../../docs/operations/qds-service.md#collection-metadata)
and [calibrated TCP export](../../docs/operations/qds-service.md#calibrated-handheld-lerobot-state).

### Upgrading to 0.8.0rc1

This release removes the old version and episode-edit helpers from
`client.data`. Their replacements on `client.versions` and `client.episodes`
speak qualia.v1, so they need your account (see 0.7.0rc7 below), and an
episode is named through the upload it was recorded in: pass `dataset=` with a
bare id, or the full name `accounts/{a}/datasets/{d}/episodes/{e}`.

A qualia.v1 tag now carries `dataset_version_id`: the immutable DatasetVersion
UUID that `client.finetune.create`, `materialize_version`, `health_for_version`
and `get_training_features` take. Tag an upload or a mixture with
`versions.create`. A definition built from filters or an explicit episode
selection is still pinned with `data.pin_version(def_id)`: saved queries have
no qualia.v1 equivalent, so that path stays on `/v1`.

| Removed | Use instead |
|---|---|
| `data.get_version(version_id)` | `versions.get(tag, dataset=)` (also `"stable"` or a digest prefix) |
| `data.list_versions(def_id)` | `versions.list(dataset)` |
| `data.list_version_sources(version_id)` | `versions.list_source_commits(tag, dataset=)` |
| `data.episode_markers(id)` | `episodes.get(id, dataset=).markers` (`time_seconds`, not `at_us`) |
| `data.split_episode(id, cut_at=, start_at_first=, end_at_last=, dry_run=)` | `episodes.split_at_markers(dataset, [id], labels=, start_labels=, end_labels=, validate_only=)` |
| `data.split_episode(id, cuts=[{"at_us": ...}])` | `episodes.split(dataset, {id: [seconds, ...]})` (a placed cut carries no label) |
| `data.split_episodes(ids, ...)` | `episodes.split_at_markers(dataset, ids, ...)`; a parent it cannot cut is absent from `splits` |
| `data.unsplit_episode(id)`, `data.unsplit_episodes(ids)` | `episodes.unsplit(id, dataset=)` per parent; `removed_count`, not `deleted` |
| `data.revise_episode(id, ...)` | `episodes.update(id, dataset=, trim=, subtasks=, tags=, task=, etag=)` |
| `data.trim_episodes(ids, start_at_first=, end_at_last=)` | `episodes.trim_to_markers(dataset, ids, start_labels=, end_labels=)` |
| `data.clear_trims(ids)` | `episodes.clear_trims(dataset, ids)` |
| `data.name_subtasks(ids, at=, boundary=)` | `episodes.name_subtasks(dataset, ids, labels=, boundary=)` |

What changed in the shapes:

- Times are seconds from the start of the recording (`time_seconds`,
  `start_seconds`, `end_seconds`), not microseconds or `trim_start_s`.
- Batch edits return `EpisodeBatch`: `episodes` holds each edited
  `qualia.v1.Episode` as it now stands, and `skipped` holds
  `SkippedEpisode(episode=<name>, reason=...)`.
- An edit is checked against the `etag` the helper read, not an
  `expected_parent_rev`; a stale one raises `QualiaAPIError` with `code`
  `"aborted"`. qualia.v1 edits carry no revision note, `tier` or
  `review_state` (rulings are `episodes.keep`/`reject`).
- A split preview (`validate_only=True`) names its children with throwaway
  ids; a split reports `dropped_lead_frame_count`/`dropped_tail_frame_count`.

`client.events.stream` now follows one job and needs `job_id=` (a
project-restricted key also passes `project_id=`). The server no longer has an
account-wide stream, so the client-side job filter is gone. The stream carries
log, telemetry and capacity-wait lines, not status: `pipeline.pipeline_status`
events no longer arrive on it, so a loop that broke on one never ends. Use
`client.finetune.watch(job_id)`, which reads the status with `finetune.get` and
reopens the stream until the job ends.

Removed with them: the models `EpisodeMarker(s)`, `EpisodeSplit`,
`EpisodeSplitBatch`, `SplitChild`, `EpisodeUnsplit`, `EpisodeRevisionCommit`,
`EpisodeTrim`, `EpisodeTrimBatch`, `EpisodeSubtask(s)`, `EpisodeSubtaskBatch`,
`SkippedEpisode`, `DatasetVersionPage` and `DatasetVersionSourcePage` (the
qualia.v1 ones live in `qualia.v1`). Definitions (`create_definition` and
friends) and `list_episode_revisions` are unchanged.

### Upgrading to 0.7.0rc7

New resources speak the qualia.v1 API, the one the dashboard reads:
`client.datasets`, `client.episodes`, `client.versions` and `client.mixtures`
(see below). They need your account: pass `account=`, set `QUALIA_ACCOUNT`, or
run `qualia auth login --account <uuid>`.

`client.data.list()` is unchanged: it still reads the old `/v1/datasets/list`
route and returns `qualia.Dataset`. It is now the legacy call;
`client.datasets.list(filter="kind != MIXTURE")` replaces it and returns
`qualia.v1.Dataset` (an upload merged into another dataset is left out; the
merged dataset, kind `MERGED`, stands in for it). The fields map like this:

| `client.data.list` (`qualia.Dataset`) | `client.datasets.list` (`qualia.v1.Dataset`) |
|---|---|
| `artifact_id` | the last segment of `name` (`accounts/{a}/datasets/{id}`) |
| `path` | `repository` |
| `subtype` (`"mcap_ros2"`) | `source_format` (`"MCAP_ROS2"`) |
| `qds.ingest_verdict` (`None` = published) | `verdict` (`"VERDICT_UNSPECIFIED"` = published) |
| `qds.verdict_reason` | `verdict_reason` |
| `qds.episode_count`, `qds.total_frames` | `counts.episodes`, `counts.frames` |
| `published_at` | `create_time` |
| `metadata`, `jobs`, `qds.size_bytes`, `qds.file_count`, `qds.quality_avg` | gone |

`search=` becomes `display_name:"<text>"` in the filter. The server cannot
filter on the format, so to keep one `subtype`, check `source_format` on each
item.

### Upgrading to 0.6.8

Hugging Face imports now run server-side. `DataResource.preflight_huggingface`
and the `max_files`, `max_bytes`, `work_dir`, and `quiet` arguments to
`import_huggingface` were client-side download controls and have been removed;
the `huggingface` installation extra was retired with them. The service now
applies its own bounded admission and buffering limits. Install plain
`qualia-sdk>=0.6.8`, pass an exact repository, 40-hex revision, dataset name,
and `source_format`, and use `idempotency_key`, `poll_interval`, or
`wait_timeout` when recovery or polling control is needed. The terminal result
contains `dataset_id` and `dataset_version_id`; it no longer exposes the former
client-published `git_oid` or `snapshot_digest` fields.

CLI users should upgrade to `qualia-cli>=0.1.3`, which requires
`qualia-sdk>=0.6.8` and uses this server-side import contract.

### Git checkouts no longer fall back to Git LFS

The transfer filter that `qualia git install` wires into a cloned dataset
fetches a file's bytes only by the `x-qds-oid` content hash the filter itself
writes into the pointer. Qualia's git remote no longer serves the Git LFS batch
API, so there is no fallback: checking out a stock git-lfs pointer (only
`version`, `oid sha256:` and `size`) fails with
`pointer carries no x-qds-oid content hash`, and `git lfs pull` against the
remote fails too. To fix a file, put its original bytes back in the working
tree, run `qualia git install`, then `git add --renormalize <path>`, commit and
push; the new pointer carries `x-qds-oid`.

## Quick Start

```python
from qualia import Qualia

# Initialize the client
client = Qualia(api_key="your-api-key")

# Or use the QUALIA_API_KEY environment variable
client = Qualia()

# List available VLA models
models = client.models.list()
for model in models:
    print(f"{model.id}: {model.name}")
    print(f"  Camera slots: {model.camera_slots}")

# Create a project
project = client.projects.create(name="My Robot Project")
print(f"Created project: {project.project_id}")

# Get the exact cameras for an immutable dataset version
version_id = "01234567-89ab-4cde-8fab-0123456789ab"
features = client.data.get_training_features(version_id)
print([camera.key for camera in features.camera_features])

# Start a finetune job
job = client.finetune.create(
    project_id=project.project_id,
    model_id="lerobot/pi05_base",
    vla_type="pi05",
    dataset_version_id=version_id,
    hours=2.0,
    camera_mappings={"cam_1": "observation.images.top"},
)
print(f"Started job: {job.job_id}")

# Check job status
status = client.finetune.get(job.job_id)
print(f"Status: {status.status}")
print(f"Current op: {status.current_op}")
for op in status.ops:  # one per pipeline op, in run order
    print(op.name, op.status)

# Cancel a job if needed
result = client.finetune.cancel(job.job_id)
```

## Resources

### Credits

```python
# Get your credit balance
balance = client.credits.get()
print(f"Available credits: {balance.balance}")
```

### Data

Publish and download robotics datasets.

```python
# Stage and atomically publish a local dataset directory.
# source_format is one of: "lerobot_v2_1", "lerobot_v3", "mcap_ros2"; when
# omitted it is read from the tree (meta/info.json, or any .mcap).
result = client.data.upload("/data/gongsta", name="gongsta")
print(result["dataset_id"], result["episode_count"], result["total_frames"])
# result["ingest_verdict"] is "published", "degraded" or "underivable" (bytes
# stored, no episode derived); pass raise_on_underivable=True to raise instead.
# The report card lists outlier episodes, such as a take that barely moves,
# separately from flagged ones, and an outlier never flags an episode.
# An MCAP upload also returns result["channel_summary"]: the topics that fed the
# state and action columns, with "action: none (no action column)" when there is
# none. A rig that needs to declare them writes <session>/meta/ingest.toml; see
# the data.upload docstring and qualia.ingest_config_schema().
# While it transfers, stderr shows "X / Y sent": source bytes already in storage,
# whether or not the server has verified them yet (verification follows the
# transfer). on_progress events carry that figure as "bytes" and the verified
# part as "bytes_verified".

# Upload protocol v5 starts at qualia-sdk 0.6.9. The 0.6.8 wheel speaks v4;
# current deployments refuse it at the initial `begin` request, before any data
# transfer, with this same actionable floor:
#   pip install 'qualia-sdk>=0.6.9'
# V5 stages bytes directly to object storage, then submits the complete tree to
# one durable publication operation. Canonical bytes deduplicate across later
# attempts; interrupted transfers can also reuse locally remembered staging.

# LeRobot 2.1 remains a distinct source format; it is not aliased to v3.
# LeRobot trees are checked before any byte moves. Structure the server always
# refuses (unreadable meta/info.json, a codebase_version that does not match
# source_format, no parquet under data/, a v2.1 tree without episodes.jsonl or
# tasks.jsonl) raises UploadBlockedError, and allow_underivable=True cannot
# override it. A missing video only warns, naming the episode and camera:
# that episode will be flagged, and you should probably exclude it.
# These checks match the server's own reading: a camera is any feature whose
# dtype is "video" in any letter case, and a .parquet shard counts in any case.
client.data.upload(
    "/data/lerobot-2.1", name="legacy-recording", source_format="lerobot_v2_1"
)

# Upload a recorded robot MCAP session (one .mcap per episode directory;
# episode.json is optional metadata). Damaged or empty bags upload with a
# warning; a session no bag can derive raises UploadBlockedError before any
# byte moves (allow_underivable=True uploads it anyway).
# A recording cut short (killed recorder) uploads its readable part, with a
# source_truncated warning, when that part carries state the server can
# derive; the pre-flight blocks it as source_underivable only when it can
# prove there is none. The result's `validation` report card marks each
# episode `ok` or `flagged` (older cards say `accepted`, which reads as ok).
client.data.upload(
    "/data/session-2026-07-20", name="dk1-kitchen", source_format="mcap_ros2"
)

# Qualia() also reads a saved `qualia dataset destination set` from the shared
# local CLI config when its API host, account, and workspace match. Select a
# workspace explicitly with the constructor or QUALIA_WORKSPACE; this is sent
# as the gateway's X-Qualia-Workspace header.
client = Qualia(
    api_key="your-api-key", account="<account-uuid>", workspace="<project-uuid>"
)

# Pin one upload to an existing merged dataset, or override the saved default.
result = client.data.upload(
    "/data/week-2",
    name="week-2",
    source_format="lerobot_v3",
    merge_target="accounts/<account-uuid>/datasets/<merged-uuid>",
)
print(result["merge_append"])  # appended or action_needed

# A failed append leaves the published upload receipt and dataset available in
# result; review it and append explicitly with client.datasets.merge(...).
# An append compares no label and no Task: the upload keeps its own operator,
# task label and Task, and the merged dataset its own. Retries see the same
# durable intent. `standalone=True` opts out for one upload.
client.data.upload(
    "/data/calibration",
    name="calibration",
    source_format="lerobot_v3",
    standalone=True,
)

# A handheld (UMI) recording converts to gripper poses as it lands: `rig=`
# names a gripper (its name, id or rig key) whose rig and TCP calibration the
# server writes into the upload's meta/ingest.toml. An upload into a merged
# dataset whose uploads agree on one rig takes it without `rig=`. The local
# file is never written; result["ingest_config"] says what was composed.
result = client.data.upload(
    "/data/umi-session",
    name="umi-session",
    source_format="mcap_ros2",
    rig="umi-handheld",
)

# Import one exact Hugging Face dataset commit. The SDK submits the pinned
# coordinates and polls while Hub bytes stream server-side into storage.
revision = "17c29c487d886878ddcb3c5649101b2860fbc047"
imported = client.data.import_huggingface(
    "qualia-robotics/openarm-cube-in-box-v1",
    revision,
    "openarm-cube-in-box",
    source_format="lerobot_v3",
)
print(imported["dataset_id"], imported["dataset_version_id"])

# Download a dataset (returns the written file paths). An interrupted download
# keeps its progress in .gongsta-copy.qualia-download-partial; run it again to resume.
# These are the files as recorded: trims and splits are not applied, and when the
# dataset has any it warns (qualia.UncuratedDownloadWarning) with the curated path.
paths = client.data.download("gongsta", dest="/data/gongsta-copy")
# The curated episodes: a tagged version, as a verified LeRobot tree.
paths = client.data.download(
    "gongsta", dest="/data/gongsta-1.0", version=f"{dataset_id}:1.0"
)

# List your datasets (legacy route; client.datasets.list is the qualia.v1 replacement)
page = client.data.list()
for d in page.items:
    # verdict: published | degraded | underivable; None = recorded before verdicts (published)
    print(
        d.artifact_id,
        d.path,
        d.qds and d.qds.ingest_verdict,
        d.qds and d.qds.verdict_reason,
    )

# Watch the account's in-flight uploads from any machine
for op in client.data.uploads():
    print(op.name, op.phase, op.progress_line or op.message)

# Get the immutable, exact camera features for mapping
features = client.data.get_training_features("01234567-89ab-4cde-8fab-0123456789ab")

# Explore one bounded page across the whole account lake. The page, totals, and
# bounded self-excluding facets share one database-statement MVCC snapshot.
from qualia import LakeExploreQuery, LakeNumericPredicate

lake = client.data.explore(
    LakeExploreQuery(
        tasks=["fold shirt"],
        tiers=["retrainable"],
        numeric_predicates=[LakeNumericPredicate(field="jerk_mean", op="lt", value=0.4)],
    )
)
for episode in lake.episodes:
    print(
        episode.dataset_name,
        episode.episode_index,
        episode.metrics,
        episode.channel_predicate_evidence,
    )

# Run native health analysis over one immutable DatasetVersion. The complete
# membership identity is re-proven; exact trajectory bytes are read for the
# declared deterministic sample (at most 120 members). This is not a byte scrub
# of unsampled artifacts or video, and model-based evidence is marked unavailable.
report = client.data.health_for_version("01234567-89ab-4cde-8fab-0123456789ab")
print(
    report.health_score,
    report.sampling.sampled_member_count,
    report.member_count,
    report.evidence_completeness.native_trajectory_metrics.status,
)
for flagged in report.flagged_episodes:
    print(flagged.episode_revision_id, flagged.reasons)

# Explain bad-data recommendations without mutating the source revision.
evidence = client.data.suggest_curation(["01234567-89ab-4cde-8fab-0123456789ab"])
for suggestion in evidence.suggestions:
    print(suggestion.episode_index, suggestion.risk_score, suggestion.reasons)

# Atomically publish the complete recommendation set as a new immutable
# DatasetVersion. Source bytes/history remain available. Preserve this UUID
# when retrying so a lost response recovers the same publication.
from uuid import uuid4

receipt = client.data.apply_curation(
    evidence,
    idempotency_key=uuid4(),
    definition_id="fedcba98-7654-4321-8fed-cba987654321",
    note="Reviewed automated bad-data evidence",
)
print(receipt.version_id, receipt.git_commit_id)
```

#### Conflicting generated paths

Uploads refuse `meta/ingest_report.json` before hashing, packing or contacting
the gateway: ingest generates this path for LeRobot 2.1, LeRobot v3 and MCAP.
MCAP also reserves `meta/info.json` for derived episode metadata; LeRobot uses
that file as source input. Case aliases and file/directory collisions are refused
too. Other metadata remains valid, subject to the existing path contract and
trusted `meta/qds` namespace rule.

The error names the source and generated paths. Keep the original tree and rename
the conflicting file in a separate copy before uploading it. The SDK does not
skip, delete or overwrite the conflicting file. Episode-dependent generated paths
are checked against the actual worker output; a collision there is terminal and
names the path even when client preflight was bypassed.

#### Resuming an interrupted upload

Run the same upload command again with the same dataset name. On deployments
supporting staging reuse, the SDK remembers opaque staging handles before each
object PUT. A retry asks the server to check the stored object's tenant,
ObjectID claim, encoded length and envelope SHA-256, then copy it inside object
storage into a fresh staging handle. Those bytes do not cross your uplink again.
The copy remains unverified staging: normal worker verification and atomic
publication still run. This does not skip verification (QUA-1704).

Hints are scoped to the gateway and authenticated repository and live under
`$XDG_CACHE_HOME/qualia/upload-staging-v1` (default `~/.cache/qualia/...`). They
contain no API keys, presigned URLs or dataset payloads. The cache is capped at
65,536 handles / 32 MiB and expires hints after seven days. Override its location
with `QUALIA_UPLOAD_STAGING_CACHE_DIR`; set that variable to an empty string to
disable it. A cache inside the upload source is automatically disabled.

This resumes complete objects, not partial PUTs, and requires the same local
cache. Missing/deleted staging, unavailable caches and older deployments fall
back to uploading the affected objects. Uploads made before this SDK recorded
hints cannot recover their unpromoted staging this way.

#### Cancelled uploads

An upload in progress can be cancelled from the dashboard (or by any client
acting on the account). The server cannot stop your process, so it tells the
SDK: the next transfer heartbeat answers `cancelled`, and `data.upload` stops
and raises `UploadCancelledError` (its `repo_id` names the repository). Nothing
is published. Bytes already sent are kept for the usual retention, so running
the same upload again can resume it (see above) rather than start over.

```python
from qualia import UploadCancelledError

try:
    client.data.upload("/data/session", name="dk1-kitchen")
except UploadCancelledError as e:
    print(f"cancelled by the user: {e.repo_id}")
```

The SDK names its transfer session to the server in the
`x-qualia-upload-session` header on the publication request, so a cancel that
arrives between heartbeats still refuses the publication. An SDK from before
upload cancellation (QUA-3009) never reads the heartbeat's answer and keeps
sending bytes after a cancel; the server refuses its publication instead
(QUA-3025), and it fails with a 409 saying the upload was cancelled by the user.

### Finetune

```python
# Create a VLA finetune job (default job_type="vla")
job = client.finetune.create(
    project_id="...",
    model_id="lerobot/pi05_base",  # Server-pinned built-in alias
    vla_type="pi05",  # pi05 or act
    dataset_version_id="01234567-89ab-4cde-8fab-0123456789ab",
    hours=2.0,  # Training duration (max 168)
    camera_mappings={  # Map model slots to dataset keys
        "cam_1": "observation.images.top",
    },
    # Optional parameters:
    instance_type="gpu_1x_a100",  # From client.instances.list()
    region="us-east-1",
    batch_size=32,
    name="My training run",
)

# ACT trains from scratch, so it takes no model_id
act_job = client.finetune.create(
    project_id="...",
    vla_type="act",
    dataset_version_id="01234567-89ab-4cde-8fab-0123456789ab",
    hours=2.0,
    camera_mappings={"cam_1": "observation.images.top"},
)

# Get job status
status = client.finetune.get(job.job_id)

# A pipeline with map nodes runs each map's body once per item; status.maps
# gives each map's op_run_id and item counts (skipped = the credit budget ran
# out before the item started). Page its items, or walk them all:
for m in status.maps:
    print(m.node_id, f"{m.counts.completed}/{m.counts.total}", m.counts.failed)
    page = client.finetune.map_items(job.job_id, m.op_run_id, status="failed")
    for item in client.finetune.iter_map_items(job.job_id, m.op_run_id):
        print(item.ordinal, item.attempt, item.status, item.error)

# Follow a job's log, telemetry and capacity-wait lines until it ends. The
# stream has no replay and the server ends it now and then: watch reads the
# status with get() and reopens it until the job completes, fails or is
# cancelled. Read get() afterwards for the full history.
for event in client.finetune.watch(job.job_id):
    if message := event.data.get("message"):
        print(message)

# Cancel a job
result = client.finetune.cancel(job.job_id)
```

Finetuning supports ACT and Pi0.5. SmolVLA and SARM finetuning, the `reward`
and `vla_w_reward` job types and RA-BC (`use_rabc`, `sarm_*`) were retired:
`create()` raises `ValueError` for them before any request is sent. Jobs and
models trained with them stay readable, so `job.metadata.vla_type` and
`job_type` are plain strings that may still hold `"smolvla"`, `"sarm"`,
`"reward"` or `"vla_w_reward"`.

A failed job says who caused it: `status.failure_cause` is `user` when something you set or
authored made it fail (`status.failure_subject` names it, e.g. `dataset_version:<id>` or
`parameter:base_model`; `status.failure_reason` says why), and `platform` otherwise.
`status.credits` is the settlement (reserved, charged, refunded, reason). A `user` failure is
charged its usage, up to the reservation; a `platform` failure is refunded in full. See the
[finetune reference](../../ui/docs/src/content/docs/sdk/resources/finetune.mdx#when-a-job-fails).

#### Advanced: Custom Hyperparameters

You can customize model hyperparameters for fine-grained control over training.
The SDK validates hyperparameters before submitting the job, so invalid
configurations are caught early.

```python
# 1. Get default hyperparameters for your model
params = client.finetune.get_hyperparams_defaults(
    vla_type="pi05",
    model_id="lerobot/pi05_base",
)

# 2. Customize the parameters as needed
params["training"]["learning_rate"] = 1e-5
params["training"]["num_epochs"] = 50

# 3. (Optional) Validate before creating the job
validation = client.finetune.validate_hyperparams(
    vla_type="pi05",
    hyperparams=params,
)
if not validation.valid:
    for issue in validation.issues:
        print(f"  {issue.field}: {issue.message}")

# 4. Create the job with custom hyperparameters
# Note: create() internally calls validate_hyperparams() when vla_hyper_spec
# is provided. If validation fails, a ValueError is raised and no job is created.
job = client.finetune.create(
    project_id=project.project_id,
    model_id="lerobot/pi05_base",
    vla_type="pi05",
    dataset_version_id="01234567-89ab-4cde-8fab-0123456789ab",
    hours=2.0,
    camera_mappings={"cam_1": "observation.images.side"},
    vla_hyper_spec=params,
)
```

### Instances

```python
# List available GPU instances
instances = client.instances.list()
for inst in instances:
    print(f"{inst.id}: {inst.gpu_description} - {inst.credits_per_hour} credits/hr")
    print(f"  Specs: {inst.specs.gpu_count}x GPU, {inst.specs.memory_gib}GB RAM")
    print(f"  Regions: {[r.name for r in inst.regions]}")
```

### Models

```python
# List available VLA model types
models = client.models.list()
for model in models:
    print(f"{model.id}: {model.name}")
    print(f"  Base model: {model.base_model_id}")
    print(f"  Camera slots: {model.camera_slots}")
```

### Projects

```python
# Create a project
project = client.projects.create(
    name="My Project",
    description="Optional description",
)

# List all projects
projects = client.projects.list()
for p in projects:
    print(f"{p.name}: {len(p.jobs)} jobs")

# Delete a project (fails if it has active jobs)
client.projects.delete(project.project_id)
```

### Tasks

A task says what a robot should do and how success shows, as five pieces of
prose: `description`, `scene`, `steps` (one per line), `success` and `failure`.
It has one editable draft; publishing writes an immutable, numbered revision,
so later edits never change what was published. Tasks live under your account
(`accounts/{account}/tasks/{task}`): pass `account=` or set `QUALIA_ACCOUNT`,
or use full resource names.

```python
from qualia import Qualia, TaskSpec

client = Qualia(account="<account-uuid>")

task = client.tasks.create(
    "Restock shelf",
    draft=TaskSpec(description="Move every box from the tray to the shelf"),
)

# Save single fields; the rest of the stored draft stays. "" clears a field.
task = client.tasks.update(
    task.id,
    steps="Pick a box from the tray\nPlace it upright on the shelf",
    success="The tray is empty and every box stands upright",
    etag=task.etag,  # refuses a stale write
)

# Or replace the whole draft
task = client.tasks.update(task.id, draft=task.draft, display_name="Restock")

# Publishing needs a description and a success
revision = client.tasks.publish(task.id, etag=task.etag, note="first cut")
client.tasks.revisions(task.id)  # newest first
client.tasks.revision(task.id, 1)

client.tasks.delete(task.id, force=True)  # account owners; force drops revisions
```

Lists page with `limit`/`cursor` like the other resources. A revision's
`content_hash` is the sha256 of its spec: one hash, one content.

A task may also carry an optional **procedure** beside the prose: the task as a
state machine of steps (one start, actions, success and failure outcomes) and
transitions, a DAG that checks report against. A step's one-line description is
its condition (what an action gets done, what an outcome looks like); there is
no separate `conditions` list. It saves on its own like a
prose field, and publishing checks it is well formed when there is one:

```python
from qualia import Procedure, Step, StepKind, Transition

procedure = Procedure(
    steps=[
        # the start needs no description: the scene says how an attempt begins
        Step(id="start", kind=StepKind.START),
        # exactly 6; min_count alone is "at least", max_count alone "up to"
        Step(
            id="place",
            description="Place a box upright on the shelf",
            min_count=6,
            max_count=6,
        ),
        Step(id="shelved", kind=StepKind.SUCCESS, description="All boxes shelved"),
        Step(id="fallen", kind=StepKind.FAILURE, description="A box fell over"),
    ],
    transitions=[
        Transition(source="start", target="place"),
        Transition(source="place", target="shelved"),
        Transition(source="place", target="fallen"),
    ],
)
task = client.tasks.update(task.id, procedure=procedure, etag=task.etag)
task.draft.procedure.edges()  # [("start", "place"), ("place", "shelved"), ("place", "fallen")]
client.tasks.update(task.id, procedure=Procedure())  # an empty one clears it
```

### Datasets, episodes, versions and mixtures (qualia.v1)

These four resources speak the qualia.v1 API, the one the dashboard uses. A
dataset is an upload (one recording session), a mixture (kept up to date from
other datasets) or a merge of uploads (several uploads reviewed, versioned and
trained as one). Episodes, versions (tags) and a mixture's sources nest under a
dataset. Everything lives under your account: pass `account=` or set
`QUALIA_ACCOUNT`, or use full resource names. One episode can be read through
several datasets, so a bare episode id also needs `dataset=`; its full name
(`accounts/{a}/datasets/{d}/episodes/{e}`) carries the scope.

The returned models (`qualia.v1.Dataset`, `qualia.v1.Episode`, ...) and the
REST routes are generated from `proto/qualia/v1` into `qualia/v1/_gen.py` when
the wheel is built (in a q-core checkout, run `./proto/gen.sh sdk` first).
Attributes are the proto field names; enums are value names such as `"KEPT"`, listed in `qualia.v1.ENUMS`.

```python
client = Qualia(account="<account-uuid>")

# Datasets: uploads and mixtures, with any AIP-160 filter
uploads = client.datasets.list(filter='kind = UPLOAD AND labels.task = "fold"')
mix = client.datasets.create("Folding, all rigs")
client.datasets.update(mix.name, description="every folding session")

# Merge a week of one bench's sessions into one dataset of its own: reviewed,
# tagged, mixed and trained as one, every take keeping its rulings, issues,
# notes and trim. The uploads must share cameras, features, channel names, fps
# and where their poses are (else 400 FAILED_PRECONDITION naming what differs,
# with nothing written); where they differ in a robot, method, reach settings or
# channel key, validate_only=True lists it and resolutions= pick one value for
# all; where they differ in a label or Task, the merged dataset takes their
# commonest value unless resolutions= pick another, as its own, and each upload
# keeps its own. Collection tags are seeded once on C, a method resolution
# changes only C, and appending preserves its existing tags. A merge is
# final: it cannot be undone,
# and the uploads become the merged dataset's history. They leave `list()`;
# their reads (`get()`, which names the merged dataset, versions, episodes,
# issues) keep answering, and every write goes through the merged dataset.
week = client.datasets.merge([u.name for u in uploads.items], display_name="Fold bench")
week.source_datasets  # the uploads, in the order merged
client.datasets.merge([next_upload.name], into=week.name)  # add one later

# Mixtures read sources: a dataset by name, or every upload and merged dataset
# with one label (a merged dataset by its own labels and method tags)
client.mixtures.add_source(mix.name, dataset=uploads.items[0].name, follow="head")
client.mixtures.add_source(mix.name, label=("task", "fold"))  # follows stable

# Tasks: link what was recorded against one. The label stays the instruction
# training reads; a link changes no revision or version.
fold = client.tasks.list().items[0]
client.datasets.update(uploads.items[0].name, linked_task=fold.name)
client.episodes.assign_task(
    uploads.items[0].name, filter="state = KEPT", linked_task=fold.name
)
client.mixtures.add_source(
    mix.name, linked_task=fold.name
)  # every upload or merged dataset linked to it
linked = client.episodes.list("-", filter=f'linked_task = "{fold.name}"')
# Publishing a task links the unlinked labels that are its name; near misses
# (case, spacing, punctuation) are previewed, then linked by pick
preview = client.episodes.link_task_labels(fold.name, validate_only=True)
client.episodes.link_task_labels(fold.name, [m.label for m in preview.matches])

# Episodes: list, edit, rule. "-" lists across the whole account.
page = client.episodes.list(mix.name, filter="state = UNREVIEWED")
episode = client.episodes.get(page.items[0].name, view="analysis")
client.episodes.update(episode.name, tags=["clean"], etag=episode.etag)
# Many episodes in one call: all of them change or none do, one revision each.
# Up to 500 episodes of one upload; trims and task links are one per call
# (update(), link_task()). A refusal names its edit ("edits[3] (...): ...").
from uuid import uuid4

from qualia.resources.episodes import EpisodeEdit

takes = client.episodes.list(uploads.items[0].name, filter="state = UNREVIEWED").items
client.episodes.update_many(
    uploads.items[0].name,
    [EpisodeEdit(episode=e.name, tags=["clean"], etag=e.etag) for e in takes],
    request_id=uuid4(),  # a retry with the same id answers the first reply
)
client.episodes.keep(mix.name, [e.name for e in page.items])
client.episodes.reject(mix.name, filter='open_issue_codes:"jerky"', reason="jerky")
client.episodes.split_at_markers(
    uploads.items[0].name, filter="child_count = 0", labels=["success"]
)
# What the studio plays: short-lived camera URIs, and the joints sampled evenly
videos = client.episodes.sign_videos(episode.name)  # .unpublished for a split child
series = client.episodes.sample_series(episode.name, max_points=400)  # .unpublished too

# Marker edits, one revision per episode; skipped ones say why
client.episodes.trim_to_markers(uploads.items[0].name, ids, start_labels=["success"])
client.episodes.name_subtasks(uploads.items[0].name, ids, labels=["grasp", "place"])

# Versions: tag what a dataset yields now; cite it as <slug>:<tag>
diff = client.versions.compare(mix.name)
tag = client.versions.create(
    mix.name, diff.suggested_version, note=diff.draft_note, digest=diff.target_digest
)
print(tag.reference)  # "folding-all-rigs:1.0"
print(tag.dataset_version_id)  # the UUID training takes
commits = client.versions.list_source_commits(tag.name)  # the repositories it pins
```

Lists page with `limit`/`cursor`; `page_info.total_count` counts every match
where the API reports it. A stale `etag` raises `QualiaAPIError` with status
409 and `code` `"aborted"`.

Training takes a tag's `dataset_version_id`.

A saved dataset version can have no kept episodes (`episode_count = 0`). It
remains readable, and the source episodes stay stored. Training and export
refuse that version.

Deleting an upload whose episodes are Task references removes those
references (the Tasks stay), so it is refused (400, `failed_precondition`)
until you say so:

```python
upload = client.datasets.get(uploads.items[0].name)
for held in upload.task_references:  # one per Task, with its episode count
    print(held.display_name, held.episode_count)
print(upload.task_referenced_episode_count)  # distinct episodes in all
client.datasets.delete(upload.name, force=True)
```

### Set up a handheld (UMI) dataset (qualia.v1)

`client.datasets` does what the dashboard's Setup strip does: pick the robot a
handheld upload plays on, pick the solver and placement, and move the poses to
the robot's TCP. Nothing here needs the dashboard.

```python
# Where are the poses? at is "root" (the gripper root's <side>.ee.* poses),
# "tcp" (each hand's TCP too) or "recorded" (a raw recording, which a resolve
# converts). None means the dataset holds no handheld poses.
poses = client.datasets.poses(upload.name)
poses.at, poses.rig  # "root", "umi-handheld"
poses.capture_gripper  # the workspace's gripper for that rig: its calibration is what a resolve takes
poses.current  # at the TCP: resolved with the capture gripper's calibration as it is now?

# Pick the robot and the solver. Only what you give is written; the rest of the
# reach settings are kept, and repeating the call writes nothing.
setup = client.datasets.setup(
    upload.name,
    robot=robot.name,  # an id or accounts/{a}/robots/{r}
    solver="pyroki",  # "pyroki" | "dls" | "" (follow the robot's model)
    base=[0.0, 0.0, 0.0],  # metres in the robot's base frame, within 5 m
    heading_deg=0.0,
)
setup.changed  # ["target_robot", "reach_settings"]
setup.resolve  # the resolve it started, or None
setup.resolve_skipped  # why none started: "robot-without-tcp", "up-to-date", ...
```

Picking a robot starts a resolve at the TCP by itself, exactly when the dataset
is an upload with handheld poses, **every arm of the robot has a TCP**, the
workspace has a gripper for the upload's rig with a tracker-to-TCP calibration,
and the poses are not at the TCP yet (or are there with another calibration).
The resolve uses that capture gripper's calibration, never the robot's own
gripper's. A robot with no TCP, a mixture and a merged dataset leave the poses
alone, and `setup` never undoes poses. `resolve=False` never resolves;
`resolve=True` resolves with the capture gripper whatever the robot.

A resolve re-derives the whole upload, so it takes as long as ingesting it did:

```python
setup = client.datasets.setup(
    upload.name,
    robot=robot.name,
    wait=True,
    on_progress=lambda r: print(r.state, r.progress),
)
# or start it, and wait later
started = client.datasets.resolve_poses(upload.name, "tcp", wait=False)
client.datasets.wait_resolve(started.operation)  # polls Datasets.GetReingest
client.datasets.resolve_poses(upload.name, "root")  # back to the gripper root
```

`wait` defaults to two hours of polling for `resolve_poses`; when it runs out
(`TimeoutError`) the work keeps running server-side, so wait for it with
`wait_resolve` instead of resolving again. A resolve that fails raises
`QualiaError` with the server's reason; a ruling or trim made while it runs is
one cause, and resolving again is the way on.

Run solver is never pressed by itself. Press it with `run_solver=True` on
`setup` (it first waits for a resolve it started), or `checks.run(upload.name,
"reach")`, then follow it:

```python
client.datasets.setup(upload.name, run_solver=True)
# settings, current_key, last_run, stale
status = client.checks.reach_status(upload.name)
status.last_run.state, status.last_run.done, status.last_run.total
client.checks.cancel_reach(upload.name)  # takes already solved keep their results
```

`datasets.update(upload.name, reach_settings={"solver": "dls", "base_m":
[0.1, 0, 0], "home_heading_rad": 0.0})` writes the settings directly; it
replaces all three, so what you leave out follows the robot's model. Saving
them starts no solve.

### Checks (qualia.v1)

`client.checks` runs a check over a dataset, follows its run and reads what it
found. The VLM check (`vlm`) has a model read each episode against the
published Task it links: confident failures become issues, uncertain readings
notes. Ask what a run would cost first; asking reserves nothing, and a run is
never charged more than it reserved.

```python
estimate = client.checks.estimate(dataset.name, max_episodes=5, max_cost=20)
# Match the trial's selection and ceiling when starting it.
started = client.checks.run(
    dataset.name, model=estimate.models[0], max_episodes=5, max_cost=20
)
run = client.checks.wait(started.check_runs[-1])  # QUEUED → RUNNING → SUCCEEDED | FAILED
client.checks.list_runs("vlm", dataset=dataset.name, state="FAILED")
for issue in client.checks.findings(dataset.name).items:
    f = (
        issue.finding
    )  # verdict, confidence, explanation, moments (recording seconds), model
    print(f.verdict, f.confidence, issue.text)
```

`model` is for the VLM check only, and must be one the deployment offers
(`estimate(...).models`, the default first); another is refused with 400.
`max_episodes` samples linked episodes across the scope, and `max_cost` caps
credits reserved and spent. Both are VLM-only positive integers. Omit them
to check every eligible episode without a ceiling.

The checks catalog (QUA-3524) holds versioned checks such as `task.graph`.
`check_ids`, `categories` and `all` select a union of them, run as one run
with one model call per episode for every selected VLM check:

```python
for check in client.checks.list(dataset.name).items:  # runnable, last_result
    print(check.check_id, check.runnable, check.unavailable_reason)
answer = client.checks.estimate(dataset.name, categories=["task"], max_episodes=50)
started = client.checks.run(
    dataset.name, categories=["task"], max_episodes=50, max_cost=answer.estimate.credits
)
client.checks.wait(started.check_run)
client.checks.list_findings(dataset.name, "task.graph", verdict="FAIL")
```

A catalog run needs `max_cost` covering the bound `estimate` answers; a
retry with the same `request_id` returns the first run.

The reach check (`"reach"`, Run solver) is pressed with `checks.run(dataset.name,
"reach")`; `checks.reach_status` and `checks.cancel_reach` follow and stop it (see
[Set up a handheld (UMI) dataset](#set-up-a-handheld-umi-dataset-qualiav1)).

## Configuration

### Environment Variables

- `QUALIA_API_KEY`: Your API key (used if not passed to constructor)
- `QUALIA_BASE_URL`: Override the API base URL (default: `https://api.qualiastudios.dev`)
- `QUALIA_ACCOUNT`: Account UUID for account-scoped resources (tasks, datasets,
  episodes, versions and mixtures)

### Custom HTTP Client

```python
import httpx

# Use a custom httpx client for advanced configuration
custom_client = httpx.Client(
    timeout=60.0,
    limits=httpx.Limits(max_connections=10),
)

client = Qualia(api_key="...", httpx_client=custom_client)
```

### Context Manager

```python
# Automatically close the client when done
with Qualia(api_key="...") as client:
    models = client.models.list()
```

## Error Handling

```python
from qualia import (
    Qualia,
    QualiaError,
    QualiaAPIError,
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
)

try:
    client = Qualia(api_key="invalid-key")
    client.models.list()
except AuthenticationError as e:
    print(f"Auth failed: {e}")
except NotFoundError as e:
    print(f"Not found: {e}")
except ValidationError as e:
    print(f"Validation error: {e}")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}s")
except QualiaAPIError as e:
    print(f"API error [{e.status_code}]: {e.message}")
except QualiaError as e:
    print(f"SDK error: {e}")
```

On the account-scoped routes (`/v1/accounts/...`, e.g. tasks) `QualiaAPIError.code`
carries the API's code name: `ABORTED` (409) is a stale etag, so re-read and
retry; `FAILED_PRECONDITION` (400) is a business-rule refusal such as
publishing a draft without a description or success.

For structured finetune HTTP 400 refusals, the SDK exposes `detail.message`
directly as `QualiaAPIError.message` when the JSON `detail` is an object with
a string `message`.

## Requirements

- Python 3.10+
- httpx
- pydantic

## License

MIT

Saved uploader destinations require a known matching account, normalized API host
and workspace. Selecting a different API key without an explicit/confirmed
account does not inherit the saved destination or its workspace. Automatic
upload completion uses the server's upload intent; an append reverted before
merges were final stays terminal (a merge is final now, so none is reverted).

