From 731779f914b64c095e21588c62a4c8ad9fba8cbb Mon Sep 17 00:00:00 2001 From: RomirJ Date: Tue, 22 Sep 2026 12:30:27 -0700 Subject: [PATCH 1/2] fix(eval): keep local LIBERO rollout on CUDA --- src/tether/eval/local_runner.py | 1 + tests/test_eval_checkpoints.py | 1 + 2 files changed, 2 insertions(+) diff --git a/src/tether/eval/local_runner.py b/src/tether/eval/local_runner.py index 48cfd3f..198b6f4 100644 --- a/src/tether/eval/local_runner.py +++ b/src/tether/eval/local_runner.py @@ -139,6 +139,7 @@ def run_local_libero( save_video_dir=config.output_dir + "/videos" if config.video else "", label=checkpoint.identity, use_native=True, + verification_device="cuda", evidence_dir=config.output_dir + "/evidence" if config.capture_evidence else "", evidence_limits=CaptureLimits( max_bytes=config.evidence_max_bytes, diff --git a/tests/test_eval_checkpoints.py b/tests/test_eval_checkpoints.py index 4efd02d..d1af323 100644 --- a/tests/test_eval_checkpoints.py +++ b/tests/test_eval_checkpoints.py @@ -59,6 +59,7 @@ def rollout(**kwargs): report = run_local_libero(config, checkpoint, loader=loader, rollout=rollout) assert captured["task_indices"] == [2] assert captured["seed"] == 41 + assert captured["verification_device"] == "cuda" assert [episode.success for episode in report.results[0].episodes] == [True, False] assert report.results[0].episodes[1].n_steps == 220 From ac42346d0893ebaff4a69cd9c4a3b825cb9b62bb Mon Sep 17 00:00:00 2001 From: rylinjames Date: Mon, 5 Oct 2026 02:47:23 -0700 Subject: [PATCH 2/2] Admit verified local SmolVLA datasets through normal preflight --- README.md | 56 ++ src/tether/finetune/cli.py | 20 +- src/tether/finetune/config.py | 14 +- src/tether/finetune/local_dataset.py | 836 ++++++++++++++++++ src/tether/finetune/preflight/dataset_size.py | 14 +- src/tether/finetune/preflight/runner.py | 39 +- src/tether/finetune/preflight/schema.py | 51 +- src/tether/finetune/run.py | 33 +- tests/test_finetune_local_dataset.py | 769 ++++++++++++++++ 9 files changed, 1802 insertions(+), 30 deletions(-) create mode 100644 src/tether/finetune/local_dataset.py create mode 100644 tests/test_finetune_local_dataset.py diff --git a/README.md b/README.md index 5095a0b..ddf88be 100644 --- a/README.md +++ b/README.md @@ -476,6 +476,62 @@ thresholds: max_missed_control_budget: 0 ``` +## Local Studio dataset admission (SmolVLA) + +A local dataset must be a completed Studio corrected export created with the +`verified-moments-v1` statistics policy and `smolvla` normalization profile. +Keep the entire export directory intact, including its sibling receipt and +provenance. `--dataset-root` points to `EXPORT/dataset` itself. + +```bash +tether finetune \ + --base /path/to/your/smolvla-checkpoint \ + --dataset studio-local/my-corrected-dataset \ + --dataset-root /path/to/EXPORT/dataset \ + --dataset-manifest-sha256 \ + --output ./local-admission-check \ + --dry-run +``` + +The digest is `export-receipt.json`'s canonical `sha256` field (SHA-256 of the +sorted, compact JSON receipt excluding that field), not the hash of the pretty +printed file or `artifact_manifest_sha256`. Tether verifies that digest, every +inventoried file's size/hash, the complete local inventory, retained source and +split bindings, LeRobot 0.5.1 writer/source pins, and feature/statistics metadata. +Only development rows exported to the `train` split are admitted. Missing files, +symlinks, incomplete markers, unsupported profiles, and integrity failures block +before training. Tether admission, preflight and command verification read only +local dataset metadata and never substitute Hub metadata. The downstream +LeRobot trainer has its own download fallback if files later disappear; this +lane requires the verified export to remain complete and immutable. + +Preflight reads the actual base config and retains SmolVLA's declared state and +action padding limits. A local base config makes this check offline; remote +bases may require fetching their config. ACT and other policies, local resume, +`--skip-preflight`, dataset revisions and conflicting Python extra arguments are +unsupported for this lane. Output must be outside the export directory. + +`--dataset` remains the LeRobot repository label for local input. No dataset +publication is performed. The existing Hub dataset workflow remains available +when the two local options are omitted. Python callers can use +`FinetuneConfig(dataset_root=..., dataset_manifest_sha256=..., ...)` and the +stdlib-only shared `tether.finetune.local_dataset.verify_local_export` function; +refusals raise `LocalDatasetError` with a stable `code`. + +The export processor check uses dataset statistics and records no ImageNet +substitution. Training keeps LeRobot's existing `use_imagenet_stats=True` +default (or an explicit `dataset.use_imagenet_stats` Python extra argument). +Preflight records both facts. SmolVLA uses `VISUAL=IDENTITY`, so these image +statistics are not consumed by that normalization mode. Tether never silently +changes the training setting. + +Admission checks retained evidence and local bytes; it does not rerun the +Parquet loader, instantiate a policy, or demonstrate training quality. Keep the +export immutable during preflight and training. The command builder verifies +again before emitting `dataset.root`, but path rechecks do not eliminate +filesystem changes between verification and use. Studio's job-owned copy is a +separate snapshot boundary. + ## Evidence knobs, not extra products Advanced teams can enable more runtime evidence with `tether serve` flags: diff --git a/src/tether/finetune/cli.py b/src/tether/finetune/cli.py index 6e3b4f0..e67a96c 100644 --- a/src/tether/finetune/cli.py +++ b/src/tether/finetune/cli.py @@ -51,6 +51,14 @@ def finetune_command( "", "--dataset-revision", help="Exact Hugging Face revision for the training dataset.", ), + dataset_root: str = typer.Option( + "", "--dataset-root", + help="Actual EXPORT/dataset directory from a qualified Studio SmolVLA export.", + ), + dataset_manifest_sha256: str = typer.Option( + "", "--dataset-manifest-sha256", + help="Canonical sibling export-receipt.json SHA-256 (excludes its sha256 field).", + ), output: str = typer.Option( ..., "--output", @@ -103,8 +111,8 @@ def finetune_command( skip_preflight: bool = typer.Option( False, "--skip-preflight", - help="Skip preflight validation. Escape hatch for local-dataset " - "or gated-repo flows where preflight can't resolve schema. " + help="Skip preflight validation for remote/gated dataset flows. " + "Never permitted with --dataset-root. " "Only set if you know what you're doing.", ), resume: bool = typer.Option( @@ -140,6 +148,8 @@ def finetune_command( base_revision=base_revision or None, dataset=dataset, dataset_revision=dataset_revision or None, + dataset_root=Path(dataset_root) if dataset_root else None, + dataset_manifest_sha256=dataset_manifest_sha256 or None, output=Path(output), num_steps=num_steps, batch_size=batch_size, @@ -157,18 +167,18 @@ def finetune_command( resume=resume, ) - console.print(f"[bold]tether finetune[/bold] — v0.3 MVP (SmolVLA LoRA)") + console.print("[bold]tether finetune[/bold] — v0.3 MVP (SmolVLA LoRA)") console.print(f" base: {cfg.base}") console.print(f" dataset: {cfg.dataset}") console.print(f" output: {cfg.output}") console.print(f" steps: {cfg.num_steps} batch={cfg.batch_size} " f"lr={cfg.learning_rate} lora_r={cfg.lora_rank}") - console.print(f" backend: lerobot (openpi-JAX + hf_transformers in v0.5+)") + console.print(" backend: lerobot (openpi-JAX + hf_transformers in v0.5+)") console.print() result = run_finetune(cfg) - console.print(f"\n[bold]Result[/bold]") + console.print("\n[bold]Result[/bold]") console.print(f" status: {result.status}") if result.error: console.print(f" [red]error:[/red] {result.error}") diff --git a/src/tether/finetune/config.py b/src/tether/finetune/config.py index fa2d72d..c88f8ba 100644 --- a/src/tether/finetune/config.py +++ b/src/tether/finetune/config.py @@ -24,7 +24,7 @@ class FinetuneConfig: """HF model id of the base checkpoint (e.g. lerobot/smolvla_base).""" dataset: str - """HF dataset id to fine-tune on (e.g. lerobot/libero).""" + """HF dataset id, or an explicit LeRobot repository label for dataset_root.""" output: Path """Output directory. Will contain model.onnx + VERIFICATION.md after @@ -92,8 +92,8 @@ class FinetuneConfig: Useful for validating config before committing to a multi-hour run.""" skip_preflight: bool = False - """If True, skip preflight validation. Escape hatch for local-dataset - or gated-repo flows where preflight can't resolve the schema. + """If True, skip preflight validation. Escape hatch for remote gated-repo flows where preflight cannot resolve + the schema. Never permitted with dataset_root. Only set if you know what you're doing.""" phase: str = "train" @@ -178,8 +178,16 @@ class FinetuneConfig: contract yet and fail before starting work. """ + dataset_root: Path | None = None + """Actual EXPORT/dataset directory of a qualified local Studio export.""" + + dataset_manifest_sha256: str | None = None + """Canonical sibling export-receipt.json digest, excluding its sha256 field.""" + def __post_init__(self) -> None: self.output = Path(self.output) + if self.dataset_root is not None: + self.dataset_root = Path(self.dataset_root) if not (0.0 <= self.mix_ratio <= 1.0): raise ValueError( f"mix_ratio must be in [0, 1], got {self.mix_ratio}" diff --git a/src/tether/finetune/local_dataset.py b/src/tether/finetune/local_dataset.py new file mode 100644 index 0000000..1cb452f --- /dev/null +++ b/src/tether/finetune/local_dataset.py @@ -0,0 +1,836 @@ +"""Offline admission of one Studio verified-moments-v1 SmolVLA export. + +The caller supplies the export receipt's canonical SHA-256, not an arbitrary +receipt path. This verifies retained evidence and bytes, without importing +LeRobot, reading model weights, or claiming to re-run the exporter/loader. +Exports must remain immutable after verification, including during training. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +from pathlib import Path, PurePosixPath +import re +import stat +from typing import Any, NoReturn + + +ADAPTER_SOURCE = "11c0a0cc0d5f4cd81de9904ec0d08f0305ba9ecec6e5952c25fe435c857afdf3" +WRITER_SOURCE = "1396b9fab7aecddd10006c33c47a487ffdcb54b4" +WRITER_WHEEL = "bbd11021023fde0947b6d1ff1c52fe91c86a28ab09a96359892f3ef7e8866862" +WRITER_FILES = { + "lerobot/datasets/lerobot_dataset.py": "03207172bbc82d541f93168017b11e2b3807cf719126bf47d04873257a4cbbd8", + "lerobot/datasets/utils.py": "1dd05df5f1d12721f227fb3601d99c22cd845cb59627b68004d3184c9e8e1486", + "lerobot/datasets/compute_stats.py": "7b0621bbdd36987939742a1803b3491b794db7b0e6393677e9b1a3e1d86aeafc", + "lerobot/datasets/dataset_writer.py": "f377ab064840d45f2dab98da27f5c71b5dfb64f70789fec5c82935eecc925207", + "lerobot/datasets/dataset_reader.py": "c5893ecb6406a6852f01dcc4d6662f6ca2320cd931bc9e4d1cd7dc83dfaa9d6c", + "lerobot/datasets/dataset_metadata.py": "df392d4c0b584177735550251c45877ba7b644377003e3fa08b71536c9969b59", + "lerobot/datasets/io_utils.py": "1b557bab9f1f1445602dfd5dc5bca69a3fe2c34c400ae36d3c69b5c7d4d1120a", + "lerobot/datasets/feature_utils.py": "1477145c9f27051511e661f5189f838d8fdfacdb0c7c8424a64d4c181027ce7a", + "lerobot/datasets/image_writer.py": "99ce7371c50b7fb0931b49228fa5e3ae30302e5559efd7f64dead6334aa9e088", +} +NORMALIZATION_FILES = { + "lerobot/processor/normalize_processor.py": "faedf40f64e1fcc6aeb287ff757abe6e85983560d82fcde060252d8f75e277d6", + "lerobot/policies/act/configuration_act.py": "11aa9037f981a957d63632561c795fac243e7898ee9302f7afac2f5d45920f7b", + "lerobot/policies/smolvla/configuration_smolvla.py": "6c1f371ccb7ed073774c459387f85626edcf568edbbaa64b4cea1bf70a80feca", + "lerobot/configs/default.py": "e077db856b4ecad9476bdf12fb2e72a0294ae3a302413afa21774a8788ab0d20", +} +IDENTITY_FIELDS = ["task", "seed", "episode", "initial_state", "embodiment"] +PARTITIONS = ("development", "holdout", "test") +MOMENTS = {"count", "min", "max", "mean", "std"} + + +class LocalDatasetError(ValueError): + """A stable refusal code for local admission callers, including Studio.""" + + def __init__(self, code: str, message: str): + self.code = code + super().__init__(message) + + +def _fail(code: str, message: str) -> NoReturn: + raise LocalDatasetError(code, message) + + +def _require(condition: bool, code: str, message: str) -> None: + if not condition: + _fail(code, message) + + +def _hash(value: Any) -> str: + return hashlib.sha256( + json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode() + ).hexdigest() + + +def _digest(value: Any) -> bool: + return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value) is not None + + +def _integer(value: Any, minimum: int = 0) -> bool: + return type(value) is int and value >= minimum + + +def _object(value: Any, label: str) -> dict: + _require(isinstance(value, dict), "invalid-metadata", f"{label} must be an object.") + return value + + +def _pairs(pairs: list[tuple[str, Any]]) -> dict: + result = {} + for key, value in pairs: + _require(key not in result, "invalid-json", f"Duplicate JSON key: {key}") + result[key] = value + return result + + +def _loads(text: str) -> Any: + return json.loads( + text, + object_pairs_hook=_pairs, + parse_constant=lambda v: _fail("invalid-json", f"Nonfinite JSON: {v}"), + ) + + +def _json(path: Path) -> Any: + return _loads(path.read_text(encoding="utf-8")) + + +def _file_hash(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def _safe_relative(value: Any) -> str: + _require( + isinstance(value, str) and bool(value) and "\\" not in value, + "unsafe-path", + "Artifacts require relative POSIX paths.", + ) + path = PurePosixPath(value) + _require( + not path.is_absolute() and ".." not in path.parts and str(path) == value, + "unsafe-path", + f"Unsafe artifact path: {value!r}", + ) + return value + + +def _root(path: Path) -> Path: + path = Path(path).expanduser().absolute() + for part in [*reversed(path.parents), path]: + _require(not part.is_symlink(), "unsafe-path", f"Symlink path component: {part}") + path = path.resolve(strict=True) + _require( + path.name == "dataset" and path.is_dir(), + "dataset-root", + "dataset_root must be the actual EXPORT/dataset directory.", + ) + return path + + +def _inventory(export: Path, receipt: dict) -> dict[str, dict]: + artifacts = receipt.get("artifacts") + _require( + isinstance(artifacts, list) and bool(artifacts), + "artifact-inventory", + "A complete artifact inventory is required.", + ) + indexed = {} + for item in artifacts: + item = _object(item, "artifact") + relative = _safe_relative(item.get("path")) + _require( + relative != "export-receipt.json" and relative not in indexed, + "artifact-inventory", + "Artifact paths must be unique and exclude the receipt.", + ) + _require( + set(item) == {"path", "size_bytes", "sha256"} + and _integer(item.get("size_bytes"), 1) + and _digest(item.get("sha256")), + "artifact-inventory", + f"Invalid artifact metadata: {relative}", + ) + indexed[relative] = item + _require( + receipt.get("artifact_manifest_sha256") == _hash(artifacts), + "artifact-manifest", + "Artifact manifest digest differs from the receipt.", + ) + actual = set() + # os.walk does not follow directory symlinks; explicitly reject every entry. + for directory, dirs, files in os.walk(export, followlinks=False): + for name in dirs + files: + path = Path(directory) / name + relative = path.relative_to(export).as_posix() + _require( + not any( + marker in name.lower() for marker in ("incomplete", "pending", "quarantine") + ), + "incomplete-export", + f"Unfinished export marker: {relative}", + ) + mode = path.lstat().st_mode + _require( + stat.S_ISREG(mode) or stat.S_ISDIR(mode), + "unsafe-path", + f"Only regular files/directories are allowed: {relative}", + ) + if stat.S_ISREG(mode) and relative != "export-receipt.json": + actual.add(relative) + _require( + actual == set(indexed), + "artifact-inventory", + "Actual export files differ from the complete receipt inventory.", + ) + for relative, item in indexed.items(): + path = export / relative + _require( + path.stat().st_size == item["size_bytes"] and _file_hash(path) == item["sha256"], + "artifact-integrity", + f"Artifact size or SHA-256 mismatch: {relative}", + ) + required = { + "dataset/meta/info.json", + "dataset/meta/stats.json", + "dataset/meta/tasks.parquet", + "provenance/source-identity.json", + "provenance/split.json", + "provenance/partition-rows.json", + "provenance/frames.jsonl", + } + _require( + required <= actual + and any(re.fullmatch(r"dataset/data/chunk-\d{3,}/file-\d{3,}\.parquet", p) for p in actual) + and any( + re.fullmatch(r"dataset/meta/episodes/chunk-\d{3,}/file-\d{3,}\.parquet", p) + for p in actual + ), + "missing-artifact", + "Required LeRobot v3 metadata, data, and provenance must be local.", + ) + return indexed + + +def _shape(value: Any, shape: list[int], label: str) -> list[float]: + if not shape: + _require( + type(value) in (int, float) and math.isfinite(value), + "statistics-shape", + f"{label} must contain finite numbers.", + ) + return [value] + _require( + isinstance(value, list) and len(value) == shape[0], + "statistics-shape", + f"Invalid {label} shape; expected {shape}.", + ) + return [number for child in value for number in _shape(child, shape[1:], label)] + + +def _features_and_stats(info: dict, stats: dict, receipt: dict) -> str: + _require( + info.get("codebase_version") == "v3.0", "dataset-version", "Only LeRobot v3.0 is admitted." + ) + frames, episodes = info.get("total_frames"), info.get("total_episodes") + _require( + _integer(frames, 1) + and _integer(episodes, 1) + and _integer(info.get("total_tasks"), 1) + and info["total_tasks"] <= episodes <= frames, + "dataset-counts", + "Invalid local dataset counts.", + ) + _require( + info.get("splits") == {"train": f"0:{episodes}"} + and info.get("data_path") == "data/chunk-{chunk_index:03d}/file-{file_index:03d}.parquet" + and info.get("video_path") is None, + "dataset-layout", + "Only the exported development/train still-image layout is admitted.", + ) + contract = _object(receipt.get("contract"), "contract") + camera = contract.get("camera_key") + _require( + isinstance(camera, str) + and re.fullmatch(r"observation\.images\.[A-Za-z0-9_]+", camera) is not None, + "feature-contract", + "An explicit single camera key is required.", + ) + _require( + _integer(contract.get("fps"), 1) + and contract["fps"] <= 1000 + and info.get("fps") == contract["fps"] + and isinstance(contract.get("state_key"), str) + and bool(contract["state_key"]), + "feature-contract", + "State key and retained sample rate must agree.", + ) + features = _object(info.get("features"), "features") + expected = { + "action", + "observation.state", + camera, + "timestamp", + "frame_index", + "episode_index", + "index", + "task_index", + } + _require( + set(features) == expected and set(stats) == expected, + "feature-contract", + "The complete exported feature/statistics sets must agree.", + ) + for key, names_key in (("action", "action_names"), ("observation.state", "state_names")): + feature = _object(features[key], key) + names = contract.get(names_key) + _require( + isinstance(names, list) + and bool(names) + and all(isinstance(n, str) and n.strip() for n in names) + and len(set(names)) == len(names) + and feature.get("names") == names + and feature.get("shape") == [len(names)] + and feature.get("dtype") == "float32", + "feature-contract", + f"{key} shape/names must match the retained contract.", + ) + visual = _object(features[camera], camera) + shape = visual.get("shape") + _require( + isinstance(shape, list) + and len(shape) == 3 + and all(_integer(n, 1) for n in shape) + and shape[2] == 3 + and visual.get("dtype") == "image" + and visual.get("names") == ["height", "width", "channels"], + "feature-contract", + "One retained HWC RGB still-image feature is required.", + ) + for key in expected - {"action", "observation.state", camera}: + feature = _object(features[key], key) + _require( + feature.get("shape") == [1] + and feature.get("dtype") == ("float32" if key == "timestamp" else "int64"), + "feature-contract", + f"Invalid index/timestamp feature: {key}", + ) + for key, feature in features.items(): + values = _object(stats[key], f"statistics/{key}") + _require( + set(values) == MOMENTS, + "statistics-policy", + f"{key} requires moments-only statistics; quantiles are unsupported.", + ) + stat_shape = [3, 1, 1] if key == camera else feature["shape"] + flattened = { + name: _shape(values[name], stat_shape, f"{key}/{name}") + for name in ("min", "max", "mean", "std") + } + count = values["count"] + _require( + isinstance(count, list) + and len(count) == 1 + and _integer(count[0], 1) + and (count[0] <= frames if key == camera else count[0] == frames), + "statistics-count", + f"Invalid retained {key} statistics count.", + ) + _require( + all( + lo <= mean <= hi and std >= 0 + for lo, mean, hi, std in zip( + flattened["min"], flattened["mean"], flattened["max"], flattened["std"] + ) + ), + "statistics-values", + f"Invalid {key} moment bounds.", + ) + return camera + + +def _verify_frame_identity(example: dict, case: dict) -> None: + """Check retained row identities using the exporter's explicit source labels. + + source_config is not copied into the export: those fallback values are bound + by the selected receipt's case, split, and original source-manifest digest. + Do not pretend to independently recover absent source configuration. + """ + source = _object(example.get("source"), "frame source") + origins = _object(case.get("identity_sources"), "case identity sources") + candidates = { + "task": [ + (example.get("task"), "example.task"), + (example.get("label"), "example.label"), + (source.get("task"), "source.task"), + ], + "embodiment": [ + (example.get("embodiment"), "example.embodiment"), + (source.get("embodiment"), "source.embodiment"), + ], + } + config_origins = { + "task": {"source_config.task", "source_config.instruction"}, + "embodiment": {"source_config.embodiment", "source_config.policy.embodiment"}, + "initial_state": {"source_config.initial_state"}, + } + for key in IDENTITY_FIELDS: + if key in candidates: + found = next( + ( + (v.strip(), origin) + for v, origin in candidates[key] + if isinstance(v, str) and v.strip() + ), + None, + ) + else: + value = source.get(key) + found = (value, "source." + key) if value is not None else None + if found is not None: + valid = found == (case.get(key), origins.get(key)) + else: + valid = origins.get(key) in config_origins.get(key, set()) + _require( + valid, + "split-leakage", + "Frame identity differs from its retained development case/source.", + ) + + +def _provenance(export: Path, receipt: dict, info: dict, adapter: dict) -> None: + identity = _object(_json(export / "provenance/source-identity.json"), "source identity") + split = _object(_json(export / "provenance/split.json"), "split") + _require( + all( + _digest(receipt.get(k)) + for k in ("source_dataset_sha256", "source_manifest_sha256", "split_sha256") + ), + "source-binding", + "Retained source/split digests are required.", + ) + _require( + identity.get("id") == receipt.get("source_dataset_id") + and isinstance(identity.get("id"), str) + and bool(identity["id"]) + and identity.get("sha256") == receipt["source_dataset_sha256"] + and identity.get("source_type") == "materialized-dataset-edit" + and identity.get("format") == "tether-recording-dataset-v1", + "source-binding", + "Retained corrected source identity differs from the receipt.", + ) + _require( + split.get("schema") == 2 + and split.get("identity_fields") == IDENTITY_FIELDS + and split.get("dataset_id") == identity["id"] + and split.get("dataset_sha256") == identity["sha256"] + and split.get("sha256") == receipt["split_sha256"] + and _hash({k: v for k, v in split.items() if k not in {"created_at", "sha256"}}) + == split["sha256"], + "split-binding", + "Retained split does not match source identity and canonical digest.", + ) + locations = {} + for partition in PARTITIONS: + cases = split.get(partition) + _require( + isinstance(cases, list) and (bool(cases) or partition == "test"), + "split-binding", + "Development and holdout require explicit cases.", + ) + for case in cases: + case = _object(case, "split case") + _require( + set(case) == set(IDENTITY_FIELDS) + and all( + isinstance(case[k], str) and case[k].strip() for k in ("task", "embodiment") + ) + and all(_integer(case[k]) for k in ("seed", "episode")) + and (case["initial_state"] is None or _integer(case["initial_state"])), + "split-binding", + "Invalid retained split identity.", + ) + key = _hash(case) + _require(key not in locations, "split-leakage", "Split case appears more than once.") + locations[key] = partition + rows = _json(export / "provenance/partition-rows.json") + _require( + isinstance(rows, list) and bool(rows), + "source-binding", + "Retained partition rows are required.", + ) + counts = dict.fromkeys(PARTITIONS, 0) + for index, row in enumerate(rows): + row = _object(row, "partition row") + _require( + row.get("source_index") == index + and type(row.get("source_index")) is int + and row.get("partition") in PARTITIONS + and _digest(row.get("example_sha256")), + "source-binding", + "Invalid partition row identity.", + ) + counts[row["partition"]] += 1 + _require( + counts == receipt.get("partition_row_counts") + and counts["development"] == info["total_frames"] + and adapter.get("training_partition_sha256") == _hash(split["development"]) + and adapter.get("training_rows_sha256") + == _hash([r for r in rows if r["partition"] == "development"]) + and adapter.get("source_dataset_sha256") == identity["sha256"] + and adapter.get("split_sha256") == split["sha256"], + "source-binding", + "Statistics/source/development partition binding differs.", + ) + episodes = receipt.get("episodes") + _require( + isinstance(episodes, list) and len(episodes) == info["total_episodes"], + "source-binding", + "Retained episode count differs from local metadata.", + ) + ordered = [] + for episode_index, episode in enumerate(episodes): + episode = _object(episode, "episode") + case = _object(episode.get("case"), "episode case") + canonical_case = {key: case.get(key) for key in IDENTITY_FIELDS} + _require( + episode.get("partition") == "development" + and locations.get(_hash(canonical_case)) == "development" + and canonical_case["embodiment"] == info.get("robot_type") + and _integer(canonical_case["initial_state"]), + "split-leakage", + "Every training episode must belong to development.", + ) + indexes = episode.get("indexes") + _require(isinstance(indexes, list) and bool(indexes), "source-binding", "Empty episode.") + for frame_index, index in enumerate(indexes): + _require( + _integer(index) and index < len(rows) and rows[index]["partition"] == "development", + "split-leakage", + "Training episode contains non-development rows.", + ) + ordered.append((index, episode_index, frame_index)) + _require( + len({row[0] for row in ordered}) == len(ordered) + and {row[0] for row in ordered} + == {r["source_index"] for r in rows if r["partition"] == "development"}, + "source-binding", + "Training episodes must cover development rows exactly once.", + ) + with (export / "provenance/frames.jsonl").open(encoding="utf-8") as stream: + frame_count = 0 + for line in stream: + _require( + bool(line.strip()) and frame_count < len(ordered), + "source-binding", + "Unexpected provenance frame.", + ) + frame = _object(_loads(line), "frame") + source_index, episode_index, frame_index = ordered[frame_count] + example = _object(frame.get("example"), "frame example") + _require( + frame.get("source_index") == source_index + and frame.get("episode_index") == episode_index + and frame.get("frame_index") == frame_index + and _hash(example) == rows[source_index]["example_sha256"], + "source-binding", + "Provenance frame differs from retained development row.", + ) + _verify_frame_identity(example, episodes[episode_index]["case"]) + frame_count += 1 + _require( + frame_count == info["total_frames"], + "source-binding", + "Missing retained development frames.", + ) + + +def _verify(dataset_root: Path, expected_manifest_sha256: str, expected_profile: str) -> dict: + _require( + expected_profile == "smolvla", + "unsupported-profile", + "Only SmolVLA local admission is qualified.", + ) + _require( + _digest(expected_manifest_sha256), + "manifest-digest", + "Supply the canonical export receipt SHA-256.", + ) + root = _root(dataset_root) + export = root.parent + receipt_path = export / "export-receipt.json" + _require( + not receipt_path.is_symlink() and receipt_path.is_file(), + "missing-receipt", + "The sibling EXPORT/export-receipt.json is required.", + ) + receipt = _object(_json(receipt_path), "export receipt") + _require( + receipt.get("sha256") == expected_manifest_sha256 + and _hash({k: v for k, v in receipt.items() if k != "sha256"}) == expected_manifest_sha256, + "manifest-digest", + "Export receipt does not match the selected canonical SHA-256.", + ) + _require( + type(receipt.get("schema")) is int + and receipt["schema"] == 1 + and receipt.get("kind") == "studio-corrected-lerobot-offline-export" + and receipt.get("status") == "complete", + "receipt-schema", + "A complete supported Studio export receipt is required.", + ) + _require( + receipt.get("source_partition") == "development" + and receipt.get("lerobot_split") == "train", + "split-leakage", + "Only development exported to train is admitted.", + ) + _require( + receipt.get("statistics_policy") == "verified-moments-v1", + "statistics-policy", + "The verified-moments-v1 export statistics policy is required.", + ) + inventory = _inventory(export, receipt) + writer = _object(receipt.get("writer"), "writer") + _require( + writer.get("version") == "0.5.1" + and writer.get("source_commit") == WRITER_SOURCE + and writer.get("wheel_sha256") == WRITER_WHEEL + and writer.get("source_files") == WRITER_FILES + and _object(writer.get("runtime_versions"), "writer runtime").get("lerobot") == "0.5.1", + "writer-pin", + "Export writer/source pins must match qualified LeRobot 0.5.1.", + ) + info = _object(_json(root / "meta/info.json"), "dataset info") + stats = _object(_json(root / "meta/stats.json"), "dataset statistics") + camera = _features_and_stats(info, stats, receipt) + normalization = _object(receipt.get("normalization_processor"), "normalization processor") + _require( + normalization.get("profile") == expected_profile, + "unsupported-profile", + "Receipt profile must be SmolVLA.", + ) + _require( + normalization.get("status") == "passed" + and normalization.get("resolved_feature_modes") + == {"action": "MEAN_STD", "observation.state": "MEAN_STD", camera: "IDENTITY"} + and normalization.get("source_files") == NORMALIZATION_FILES + and normalization.get("epsilon") == 1e-8 + and normalization.get("numeric_frames") == info["total_frames"] + and normalization.get("image_statistics_used") is False + and normalization.get("imagenet_override_applied_in_this_check") is False + and normalization.get("trainer_use_imagenet_stats_default") is True, + "normalization-contract", + "Pinned SmolVLA normalization evidence must pass and retain VISUAL=IDENTITY.", + ) + loader = _object(receipt.get("loader_roundtrip"), "loader roundtrip") + _require( + loader.get("status") == "passed" + and loader.get("frames_read") == info["total_frames"] + and loader.get("episodes_read") == info["total_episodes"], + "loader-contract", + "Passed loader evidence must agree with local counts.", + ) + adapter = _object(receipt.get("statistics_adapter"), "statistics adapter") + _require( + adapter.get("policy") == "verified-moments-v1" + and adapter.get("adapter_source_sha256") == ADAPTER_SOURCE + and adapter.get("statistics_sha256") == _hash(stats), + "statistics-binding", + "Actual local statistics differ from the verified adapter evidence.", + ) + replacements = adapter.get("replacement_statistics") + originals = adapter.get("original_statistics") + targets = {"dataset/meta/stats.json"} | { + p for p in inventory if p.startswith("dataset/meta/episodes/") + } + _require( + isinstance(replacements, list) + and isinstance(originals, list) + and len(replacements) == len(targets) + and len(originals) == len(targets), + "statistics-binding", + "Complete original and replacement statistics bindings are required.", + ) + _require( + {r.get("path") for r in replacements} == targets + and {r.get("output_path") for r in originals} == targets, + "statistics-binding", + "Statistics paths differ from the local inventory.", + ) + for item in replacements: + _require( + item.get("sha256") == inventory[item["path"]]["sha256"], + "statistics-binding", + "Replacement statistics digest differs.", + ) + for item in originals: + retained = "provenance/upstream-statistics/" + item["output_path"].removeprefix("dataset/") + _require( + item.get("retained_path") == retained + and retained in inventory + and item.get("sha256") == inventory[retained]["sha256"], + "statistics-binding", + "Retained upstream statistics digest differs.", + ) + _provenance(export, receipt, info, adapter) + image_indexes, offset = [], 0 + for episode in receipt["episodes"]: + length = len(episode["indexes"]) + image_indexes.extend(sorted({offset, offset + length - 1})) + offset += length + _require( + normalization.get("image_frame_indexes") == image_indexes + and normalization.get("maximum_image_batch") == 1, + "normalization-contract", + "Representative image check evidence differs from exported episodes.", + ) + return { + "receipt": receipt, + "info": info, + "dataset_root": str(root), + "manifest_sha256": expected_manifest_sha256, + } + + +def verify_local_export( + dataset_root: Path, expected_manifest_sha256: str, *, expected_profile: str = "smolvla" +) -> dict: + """Verify local bytes and retained evidence; never fall back to the Hub. + + Raises LocalDatasetError with a stable code on missing, malformed, unsafe, + unsupported or inconsistent input. No external packages/network are used. + """ + try: + return _verify(dataset_root, expected_manifest_sha256, expected_profile) + except LocalDatasetError: + raise + except ( + OSError, + ValueError, + TypeError, + KeyError, + AttributeError, + OverflowError, + RecursionError, + ) as exc: + raise LocalDatasetError( + "invalid-local-export", f"Cannot verify local export: {exc}" + ) from exc + + +def validate_local_config(cfg: Any) -> None: + """Reject local escape hatches before any preflight or command is built.""" + root, digest = cfg.dataset_root, cfg.dataset_manifest_sha256 + if root is None and digest is None: + return + _require( + root is not None and digest is not None and _digest(digest), + "local-config", + "dataset_root and dataset_manifest_sha256 must be supplied together.", + ) + _require( + not cfg.skip_preflight and not cfg.resume, + "local-config", + "Local dataset admission does not support skip_preflight or resume.", + ) + _require( + cfg.phase == "train" + and cfg.backend == "lerobot" + and cfg.policy == "auto" + and cfg.mode == "lora" + and bool(cfg.base) + and bool(cfg.dataset) + and not cfg.base_dataset, + "unsupported-profile", + "Local datasets require pretrained SmolVLA LoRA (policy=auto, phase=train).", + ) + export = Path(root).expanduser().resolve().parent + _require( + not Path(cfg.output).expanduser().resolve().is_relative_to(export), + "local-config", + "Training output must be outside the immutable export directory.", + ) + _require( + not cfg.dataset_revision, + "local-config", + "Local datasets cannot select a Hub dataset revision.", + ) + for key in cfg.extra_lerobot_args: + _require( + isinstance(key, str) and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_.]*", key) is not None, + "local-config", + "Local extra argument names must be canonical dotted keys.", + ) + normalized = key + _require( + normalized + not in { + "dataset", + "dataset.root", + "dataset.repo_id", + "dataset.revision", + "config_path", + "output_dir", + "resume", + "rename_map", + "peft", + "peft.method_type", + } + and ( + not normalized.startswith("dataset.") or normalized == "dataset.use_imagenet_stats" + ) + and not normalized.startswith( + ( + "policy.input_features.", + "policy.output_features.", + "rename_map.", + "peft.method_type.", + ) + ) + and "normaliz" not in normalized.lower() + and "processor" not in normalized.lower() + and normalized + not in { + "policy", + "policy.type", + "policy.path", + "policy.pretrained_path", + "policy.pretrained_model_path", + "policy.input_features", + "policy.output_features", + "policy.max_action_dim", + "policy.max_state_dim", + "dataset.episodes", + "dataset.streaming", + "dataset.video_backend", + }, + "local-config", + f"extra_lerobot_args cannot override local admission: {key}", + ) + effective_imagenet_stats(cfg) + + +def effective_imagenet_stats(cfg: Any) -> bool: + """Keep LeRobot's default; record any explicit existing escape-hatch value.""" + value = cfg.extra_lerobot_args.get("dataset.use_imagenet_stats", True) + if value is True or (isinstance(value, str) and value in ("true", "True")): + return True + if value is False or (isinstance(value, str) and value in ("false", "False")): + return False + _fail("local-config", "dataset.use_imagenet_stats must be an explicit boolean.") + + +__all__ = ["LocalDatasetError", "verify_local_export"] diff --git a/src/tether/finetune/preflight/dataset_size.py b/src/tether/finetune/preflight/dataset_size.py index 1432dc1..3d0dd1c 100644 --- a/src/tether/finetune/preflight/dataset_size.py +++ b/src/tether/finetune/preflight/dataset_size.py @@ -78,7 +78,14 @@ def _fetch_dataset_info(dataset_repo_id: str) -> dict[str, Any] | None: def check_dataset_size(cfg: FinetuneConfig) -> PreflightCheck: """Warn if episode count is below the stability floor for the base model.""" - policy_type = _infer_policy_type(cfg.base) + if cfg.dataset_root is not None: + from tether.finetune.local_dataset import validate_local_config, verify_local_export + validate_local_config(cfg) + info = verify_local_export(cfg.dataset_root, cfg.dataset_manifest_sha256)["info"] + policy_type = "smolvla" # Mandatory schema check validates the actual base config. + else: + info = None + policy_type = _infer_policy_type(cfg.base) if policy_type is None: return PreflightCheck( name="dataset_size", @@ -94,7 +101,8 @@ def check_dataset_size(cfg: FinetuneConfig) -> PreflightCheck: summary=f"no floor defined for policy_type={policy_type}", ) - info = _fetch_dataset_info(cfg.dataset) + if cfg.dataset_root is None: + info = _fetch_dataset_info(cfg.dataset) if info is None: return PreflightCheck( name="dataset_size", @@ -108,7 +116,7 @@ def check_dataset_size(cfg: FinetuneConfig) -> PreflightCheck: return PreflightCheck( name="dataset_size", severity="warn", - summary=f"dataset info has no episode count", + summary="dataset info has no episode count", ) if num_episodes < floor: diff --git a/src/tether/finetune/preflight/runner.py b/src/tether/finetune/preflight/runner.py index 17acf3e..b3cc518 100644 --- a/src/tether/finetune/preflight/runner.py +++ b/src/tether/finetune/preflight/runner.py @@ -14,7 +14,7 @@ from tether.finetune.config import FinetuneConfig from tether.finetune.preflight.dataset_size import check_dataset_size -from tether.finetune.preflight.result import PreflightReport +from tether.finetune.preflight.result import PreflightCheck, PreflightReport from tether.finetune.preflight.schema import check_schema logger = logging.getLogger(__name__) @@ -35,22 +35,47 @@ def run_preflight(cfg: FinetuneConfig) -> PreflightReport: * norm_stats (base-checkpoint stats reuse vs recompute) """ report = PreflightReport() + from tether.finetune.local_dataset import ( + effective_imagenet_stats, validate_local_config, verify_local_export, + ) + # Local integrity is mandatory. Even an unexpected verifier exception blocks + # before any remote metadata lookup or later checks. + if cfg.dataset_root is not None or cfg.dataset_manifest_sha256 is not None: + try: + validate_local_config(cfg) + verified = verify_local_export(cfg.dataset_root, cfg.dataset_manifest_sha256) + report.add(PreflightCheck( + "local_dataset", "ok", "Local export bytes and retained admission evidence verified.", + {"dataset_root": verified["dataset_root"], + "manifest_sha256": verified["manifest_sha256"], + "statistics_policy": "verified-moments-v1", + "normalization_profile": "smolvla", + "visual_normalization": "IDENTITY", + "export_processor_used_image_statistics": False, + "trainer_use_imagenet_stats_default": True, + "effective_use_imagenet_stats": effective_imagenet_stats(cfg), + "boundary": "Receipt/byte verification; no trainer or model was run. Keep the export immutable."}, + )) + except Exception as exc: + report.add(PreflightCheck("local_dataset", "fail", str(exc), + {"code": getattr(exc, "code", "local-verification-error")})) + return report for check_fn in (check_schema, check_dataset_size): try: result = check_fn(cfg) report.add(result) except Exception as exc: - # A check crashing should not take down the whole run. Log - # it as a warning and keep going. + # Remote checks retain warning behavior. Local admission is + # mandatory, including unexpected schema/count check failures. + severity = "fail" if cfg.dataset_root is not None else "warn" logger.warning( - "[preflight] %s crashed: %s — treating as warn", - check_fn.__name__, exc, + "[preflight] %s crashed: %s — treating as %s", + check_fn.__name__, exc, severity, ) - from tether.finetune.preflight.result import PreflightCheck report.add(PreflightCheck( name=check_fn.__name__.replace("check_", ""), - severity="warn", + severity=severity, summary=f"check crashed: {type(exc).__name__}: {exc}", )) diff --git a/src/tether/finetune/preflight/schema.py b/src/tether/finetune/preflight/schema.py index 48a43c5..594fc41 100644 --- a/src/tether/finetune/preflight/schema.py +++ b/src/tether/finetune/preflight/schema.py @@ -14,6 +14,7 @@ from __future__ import annotations import logging +from pathlib import Path from typing import Any from tether.finetune.config import FinetuneConfig @@ -52,14 +53,20 @@ def _fetch_dataset_features(dataset_repo_id: str) -> dict[str, Any] | None: return info.get("features") or info.get("feature") or {} -def _fetch_base_config(base_id: str) -> dict[str, Any] | None: +def _fetch_base_config(base_id: str, revision: str | None = None) -> dict[str, Any] | None: """Return the base checkpoint's config.json as a dict.""" + local = Path(base_id).expanduser() + if local.is_dir(): + import json + with (local / "config.json").open() as stream: + return json.load(stream) try: from huggingface_hub import hf_hub_download except ImportError: return None try: - cfg_path = hf_hub_download(repo_id=base_id, filename="config.json") + kwargs = {"revision": revision} if revision else {} + cfg_path = hf_hub_download(repo_id=base_id, filename="config.json", **kwargs) except Exception as e: logger.debug("[preflight] couldn't fetch config.json from %s: %s", base_id, e) @@ -115,8 +122,41 @@ def check_schema(cfg: FinetuneConfig) -> PreflightCheck: failure — customers running against local datasets or gated models still need a path through. """ - features = _fetch_dataset_features(cfg.dataset) - base_config = _fetch_base_config(cfg.base) + if cfg.dataset_root is not None: + from tether.finetune.local_dataset import validate_local_config, verify_local_export + validate_local_config(cfg) + features = verify_local_export(cfg.dataset_root, cfg.dataset_manifest_sha256)["info"]["features"] + else: + features = _fetch_dataset_features(cfg.dataset) + base_config = (_fetch_base_config(cfg.base, cfg.base_revision) if cfg.base_revision + else _fetch_base_config(cfg.base)) + local_details = {} + if cfg.dataset_root is not None: + expected_modes = {"VISUAL": "IDENTITY", "STATE": "MEAN_STD", "ACTION": "MEAN_STD"} + if not isinstance(base_config, dict) or base_config.get("type") != "smolvla": + return PreflightCheck("schema", "fail", "Local admission requires a resolvable SmolVLA base config.") + # An omitted field uses only the exact LeRobot 0.5.1 class defaults + # qualified in the verified receipt. Explicit null/contradictory modes + # are not treated as omission. + resolved_modes = base_config.get("normalization_mapping", expected_modes) + if resolved_modes != expected_modes: + return PreflightCheck("schema", "fail", "Base normalization must match the qualified SmolVLA profile.") + local_details = { + "resolved_normalization_mapping": resolved_modes, + "normalization_mapping_source": ( + "base config.json" if "normalization_mapping" in base_config + else "pinned LeRobot 0.5.1 SmolVLAConfig defaults" + ), + } + max_state = base_config.get("max_state_dim") + max_action = base_config.get("max_action_dim") + if (type(max_state) is not int or max_state <= 0 or type(max_action) is not int + or max_action <= 0): + return PreflightCheck("schema", "fail", "Base config must declare its real SmolVLA padding dimensions.") + if features["observation.state"]["shape"][0] > max_state: + return PreflightCheck("schema", "fail", "Dataset state exceeds the base max_state_dim.") + if features["action"]["shape"][0] > max_action: + return PreflightCheck("schema", "fail", "Dataset action exceeds the base max_action_dim.") if features is None: return PreflightCheck( @@ -164,6 +204,7 @@ def check_schema(cfg: FinetuneConfig) -> PreflightCheck: "base_action_dim": base_dim, "max_action_dim": max_action_dim, "uses_action_padding": True, + **local_details, }, ) return PreflightCheck( @@ -190,7 +231,7 @@ def check_schema(cfg: FinetuneConfig) -> PreflightCheck: name="schema", severity="ok", summary=f"action dim matches: {ds_dim}-D on both sides", - detail={"action_dim": ds_dim}, + detail={"action_dim": ds_dim, **local_details}, ) diff --git a/src/tether/finetune/run.py b/src/tether/finetune/run.py index 5a297f6..0f35f36 100644 --- a/src/tether/finetune/run.py +++ b/src/tether/finetune/run.py @@ -19,7 +19,6 @@ import time from dataclasses import replace from pathlib import Path -from typing import Any from tether.finetune.config import FinetuneConfig, FinetuneResult @@ -34,6 +33,11 @@ def _validate_config(cfg: FinetuneConfig) -> list[str]: before we spin up any training process. """ errs: list[str] = [] + from tether.finetune.local_dataset import LocalDatasetError, validate_local_config + try: + validate_local_config(cfg) + except LocalDatasetError as exc: + errs.append(f"[{exc.code}] {exc}") is_distill = getattr(cfg, "phase", "train") == "distill" is_from_scratch = ( getattr(cfg, "policy", "auto") != "auto" @@ -139,6 +143,19 @@ def _build_lerobot_command(cfg: FinetuneConfig) -> list[str]: doesn't expose a top-level precision flag; it's baked into the policy config. v0.5 will add per-policy precision overrides. """ + from tether.finetune.local_dataset import validate_local_config, verify_local_export + validate_local_config(cfg) + local_export = None + if cfg.dataset_root is not None: + # Recheck immediately before emission. Callers must still keep exports + # immutable: a path-based trainer is not a filesystem snapshot boundary. + local_export = verify_local_export(cfg.dataset_root, cfg.dataset_manifest_sha256) + from tether.finetune.preflight.schema import check_schema + schema = check_schema(cfg) + if schema.severity != "ok": + from tether.finetune.local_dataset import LocalDatasetError + raise LocalDatasetError("base-compatibility", schema.summary) + explicit_policy = getattr(cfg, "policy", "auto") is_from_scratch = (explicit_policy != "auto" and cfg.mode == "full") @@ -176,7 +193,7 @@ def _build_lerobot_command(cfg: FinetuneConfig) -> list[str]: cmd = [ "lerobot-train", f"--policy.repo_id={repo_id}", - f"--policy.push_to_hub=false", + "--policy.push_to_hub=false", f"--dataset.repo_id={cfg.dataset}", f"--output_dir={lerobot_output}", f"--steps={cfg.num_steps}", @@ -194,6 +211,8 @@ def _build_lerobot_command(cfg: FinetuneConfig) -> list[str]: # derives camera, state, and action features from the selected dataset. cmd.append("--policy.input_features=null") cmd.append("--policy.output_features=null") + if local_export is not None: + cmd.append(f"--dataset.root={local_export['dataset_root']}") if cfg.dataset_revision: cmd.append(f"--dataset.revision={cfg.dataset_revision}") if is_from_scratch and cfg.chunk_size: @@ -205,7 +224,7 @@ def _build_lerobot_command(cfg: FinetuneConfig) -> list[str]: cmd.append(f"--policy.n_action_steps={cfg.chunk_size}") if cfg.mode == "lora": cmd.extend([ - f"--peft.method_type=lora", + "--peft.method_type=lora", f"--peft.r={cfg.lora_rank}", ]) if cfg.resume: @@ -228,7 +247,7 @@ def _run_lerobot_training( logger.info("[finetune] exec: %s", " ".join(cmd)) log_path.parent.mkdir(parents=True, exist_ok=True) with log_path.open("w") as log: - log.write(f"# tether finetune — lerobot-train invocation\n") + log.write("# tether finetune — lerobot-train invocation\n") log.write(f"# cmd: {' '.join(cmd)}\n\n") log.flush() proc = subprocess.Popen( @@ -397,9 +416,6 @@ def run_finetune(cfg: FinetuneConfig, *, hooks=None) -> FinetuneResult: created. The distill CLI attaches `libero_drop_gate` here before calling run_finetune. """ - cfg.output.mkdir(parents=True, exist_ok=True) - training_log = cfg.output / "training_log.jsonl" - errs = _validate_config(cfg) if errs: return FinetuneResult( @@ -408,6 +424,9 @@ def run_finetune(cfg: FinetuneConfig, *, hooks=None) -> FinetuneResult: error="config validation failed:\n " + "\n ".join(errs), ) + cfg.output.mkdir(parents=True, exist_ok=True) + training_log = cfg.output / "training_log.jsonl" + # Pre-flight validation (v0.5) — catches top customer pains before # any GPU time. Dry-run + skip flags supported. if not cfg.skip_preflight: diff --git a/tests/test_finetune_local_dataset.py b/tests/test_finetune_local_dataset.py new file mode 100644 index 0000000..5afbb96 --- /dev/null +++ b/tests/test_finetune_local_dataset.py @@ -0,0 +1,769 @@ +"""Synthetic admission tests, not a LeRobot writer/loader/training qualification. + +Parquet slots contain labeled opaque bytes: the admission layer verifies their +pinned integrity and retained evidence. It deliberately does not decode them. +Base-config fixtures declare their dimensions explicitly; no model is fetched. +""" + +from __future__ import annotations + +import copy +import hashlib +import json +from pathlib import Path +from unittest.mock import patch + +import pytest + +from tether.finetune.config import FinetuneConfig +from tether.finetune.local_dataset import ( + LocalDatasetError, + NORMALIZATION_FILES, + WRITER_FILES, + WRITER_SOURCE, + WRITER_WHEEL, + verify_local_export, +) +from tether.finetune.preflight import run_preflight +from tether.finetune.preflight.dataset_size import check_dataset_size +from tether.finetune.preflight.schema import check_schema +from tether.finetune.run import _build_lerobot_command, _validate_config, run_finetune + + +def canonical(value): + return hashlib.sha256( + json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + ).hexdigest() + + +def write_json(path, value): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, sort_keys=True, indent=2) + "\n") + + +def seal(export, receipt, *, inventory=True): + if inventory: + receipt["artifacts"] = [ + { + "path": p.relative_to(export).as_posix(), + "size_bytes": p.stat().st_size, + "sha256": hashlib.sha256(p.read_bytes()).hexdigest(), + } + for p in sorted(export.rglob("*")) + if p.is_file() and p.name != "export-receipt.json" + ] + receipt["artifact_manifest_sha256"] = canonical(receipt["artifacts"]) + receipt["sha256"] = canonical({k: v for k, v in receipt.items() if k != "sha256"}) + write_json(export / "export-receipt.json", receipt) + return receipt["sha256"] + + +@pytest.fixture +def local_export(tmp_path): + export = tmp_path / "export" + root = export / "dataset" + camera = "observation.images.wrist" + features = { + "action": {"dtype": "float32", "shape": [2], "names": ["a", "b"]}, + "observation.state": {"dtype": "float32", "shape": [2], "names": ["x", "y"]}, + camera: {"dtype": "image", "shape": [6, 8, 3], "names": ["height", "width", "channels"]}, + **{ + key: { + "dtype": "float32" if key == "timestamp" else "int64", + "shape": [1], + "names": None, + } + for key in ("timestamp", "index", "episode_index", "frame_index", "task_index") + }, + } + info = { + "codebase_version": "v3.0", + "robot_type": "synthetic-arm", + "total_frames": 2, + "total_episodes": 1, + "total_tasks": 1, + "fps": 10, + "splits": {"train": "0:1"}, + "data_path": "data/chunk-{chunk_index:03d}/file-{file_index:03d}.parquet", + "video_path": None, + "features": features, + } + stats = {} + for key, feature in features.items(): + values = [[[0.0]], [[0.0]], [[0.0]]] if key == camera else [0.0] * feature["shape"][0] + stats[key] = { + "count": [2], + **{name: copy.deepcopy(values) for name in ("min", "max", "mean", "std")}, + } + write_json(root / "meta/info.json", info) + write_json(root / "meta/stats.json", stats) + opaque_paths = [ + "dataset/data/chunk-000/file-000.parquet", + "dataset/meta/episodes/chunk-000/file-000.parquet", + "dataset/meta/tasks.parquet", + "provenance/upstream-statistics/meta/episodes/chunk-000/file-000.parquet", + "provenance/upstream-statistics/meta/stats.json", + ] + for relative in opaque_paths: + path = export / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"SYNTHETIC opaque artifact slot; no loader was run.\n") + identity = { + "id": "synthetic-corrected-source", + "sha256": "a" * 64, + "source_type": "materialized-dataset-edit", + "format": "tether-recording-dataset-v1", + } + case = { + "task": "synthetic task", + "seed": 3, + "episode": 0, + "initial_state": 1, + "embodiment": "synthetic-arm", + } + split = { + "schema": 2, + "identity_fields": ["task", "seed", "episode", "initial_state", "embodiment"], + "dataset_id": identity["id"], + "dataset_sha256": identity["sha256"], + "development": [case], + "holdout": [{**case, "episode": 1}], + "test": [], + } + split["sha256"] = canonical(split) + write_json(export / "provenance/source-identity.json", identity) + write_json(export / "provenance/split.json", split) + frames = [ + { + "source_index": index, + "episode_index": 0, + "frame_index": index, + "example": { + "task": case["task"], + "embodiment": case["embodiment"], + "source": {"episode": 0, "seed": 3, "initial_state": 1}, + }, + } + for index in range(2) + ] + rows = [ + { + "source_index": i, + "partition": "development", + "example_sha256": canonical(frame["example"]), + } + for i, frame in enumerate(frames) + ] + rows.append({"source_index": 2, "partition": "holdout", "example_sha256": "d" * 64}) + write_json(export / "provenance/partition-rows.json", rows) + (export / "provenance/frames.jsonl").write_text( + "".join(json.dumps(row) + "\n" for row in frames) + ) + + def file_hash(path): + return hashlib.sha256((export / path).read_bytes()).hexdigest() + + replacement_paths = [ + "dataset/meta/stats.json", + "dataset/meta/episodes/chunk-000/file-000.parquet", + ] + adapter = { + "policy": "verified-moments-v1", + "adapter_source_sha256": "11c0a0cc0d5f4cd81de9904ec0d08f0305ba9ecec6e5952c25fe435c857afdf3", + "statistics_sha256": canonical(stats), + "source_dataset_sha256": identity["sha256"], + "split_sha256": split["sha256"], + "training_partition_sha256": canonical(split["development"]), + "training_rows_sha256": canonical(rows[:2]), + "replacement_statistics": [{"path": p, "sha256": file_hash(p)} for p in replacement_paths], + "original_statistics": [ + { + "output_path": p, + "retained_path": "provenance/upstream-statistics/" + p.removeprefix("dataset/"), + "sha256": file_hash("provenance/upstream-statistics/" + p.removeprefix("dataset/")), + } + for p in replacement_paths + ], + } + receipt = { + "schema": 1, + "kind": "studio-corrected-lerobot-offline-export", + "status": "complete", + "source_dataset_id": identity["id"], + "source_dataset_sha256": identity["sha256"], + "source_manifest_sha256": "b" * 64, + "split_sha256": split["sha256"], + "source_partition": "development", + "lerobot_split": "train", + "partition_row_counts": {"development": 2, "holdout": 1, "test": 0}, + "episodes": [ + { + "case": { + **case, + "identity_sources": { + "task": "example.task", + "embodiment": "example.embodiment", + "seed": "source.seed", + "episode": "source.episode", + "initial_state": "source.initial_state", + }, + }, + "partition": "development", + "indexes": [0, 1], + } + ], + "contract": { + "state_names": ["x", "y"], + "action_names": ["a", "b"], + "state_key": "state", + "camera_key": camera, + "fps": 10, + }, + "writer": { + "version": "0.5.1", + "source_commit": WRITER_SOURCE, + "wheel_sha256": WRITER_WHEEL, + "source_files": dict(WRITER_FILES), + "runtime_versions": {"lerobot": "0.5.1"}, + }, + "statistics_policy": "verified-moments-v1", + "statistics_adapter": adapter, + "normalization_processor": { + "status": "passed", + "profile": "smolvla", + "epsilon": 1e-8, + "source_files": dict(NORMALIZATION_FILES), + "numeric_frames": 2, + "image_frame_indexes": [0, 1], + "maximum_image_batch": 1, + "resolved_feature_modes": { + "action": "MEAN_STD", + "observation.state": "MEAN_STD", + camera: "IDENTITY", + }, + "image_statistics_used": False, + "imagenet_override_applied_in_this_check": False, + "trainer_use_imagenet_stats_default": True, + }, + "loader_roundtrip": {"status": "passed", "frames_read": 2, "episodes_read": 1}, + } + digest = seal(export, receipt) + return root, receipt, digest + + +@pytest.fixture +def local_cfg(local_export, tmp_path): + root, _, digest = local_export + base = tmp_path / "synthetic-base" + write_json( + base / "config.json", + { + "type": "smolvla", + "max_action_dim": 4, + "max_state_dim": 4, + "output_features": {"action": {"shape": [3]}}, + "normalization_mapping": { + "VISUAL": "IDENTITY", + "STATE": "MEAN_STD", + "ACTION": "MEAN_STD", + }, + }, + ) + return FinetuneConfig( + base=str(base), + dataset="studio-local/synthetic", + output=tmp_path / "run", + dataset_root=root, + dataset_manifest_sha256=digest, + dry_run=True, + ) + + +def test_valid_synthetic_admission(local_export): + root, receipt, digest = local_export + result = verify_local_export(root, digest) + assert result == { + "receipt": receipt, + "info": json.loads((root / "meta/info.json").read_text()), + "dataset_root": str(root), + "manifest_sha256": digest, + } + + +@pytest.mark.parametrize( + "field,value,code", + [ + ("schema", 2, "receipt-schema"), + ("status", "incomplete", "receipt-schema"), + ("kind", "other", "receipt-schema"), + ("source_partition", "holdout", "split-leakage"), + ("lerobot_split", "test", "split-leakage"), + ("statistics_policy", "upstream", "statistics-policy"), + ("artifact_manifest_sha256", "0" * 64, "artifact-manifest"), + ], +) +def test_receipt_semantics(local_export, field, value, code): + root, receipt, _ = local_export + receipt[field] = value + if field == "artifact_manifest_sha256": + receipt["sha256"] = canonical({k: v for k, v in receipt.items() if k != "sha256"}) + write_json(root.parent / "export-receipt.json", receipt) + digest = receipt["sha256"] + else: + digest = seal(root.parent, receipt) + with pytest.raises(LocalDatasetError) as error: + verify_local_export(root, digest) + assert error.value.code == code + + +@pytest.mark.parametrize("digest", [None, "", "0" * 64, "A" * 64, "1234"]) +def test_missing_or_mismatched_digest(local_export, digest): + with pytest.raises(LocalDatasetError, match="SHA-256"): + verify_local_export(local_export[0], digest) + + +@pytest.mark.parametrize( + "relative", + [ + "incomplete.json", + "export-receipt.pending.json", + "dataset/meta/info.incomplete.json", + "quarantine.json", + ], +) +def test_marker_refused_even_if_inventoried(local_export, relative): + root, receipt, _ = local_export + (root.parent / relative).write_text("unfinished") + digest = seal(root.parent, receipt) + with pytest.raises(LocalDatasetError) as error: + verify_local_export(root, digest) + assert error.value.code == "incomplete-export" + + +@pytest.mark.parametrize( + "mutation", + [ + "changed", + "deleted", + "extra", + "duplicate", + "traversal", + "absolute", + "symlink_file", + "symlink_dir", + "symlink_root", + "symlink_receipt", + ], +) +def test_inventory_and_paths(local_export, tmp_path, mutation): + root, receipt, digest = local_export + data = root / "data/chunk-000/file-000.parquet" + if mutation == "changed": + data.write_bytes(b"corrupt") + elif mutation == "deleted": + data.unlink() + elif mutation == "extra": + (root / "unlisted.txt").write_text("extra") + elif mutation in ("duplicate", "traversal", "absolute"): + if mutation == "duplicate": + receipt["artifacts"].append(copy.deepcopy(receipt["artifacts"][0])) + else: + receipt["artifacts"][0]["path"] = ( + "../outside" if mutation == "traversal" else "/tmp/outside" + ) + digest = seal(root.parent, receipt, inventory=False) + elif mutation == "symlink_file": + outside = tmp_path / "outside" + outside.write_bytes(data.read_bytes()) + data.unlink() + data.symlink_to(outside) + elif mutation == "symlink_dir": + (root / "extra-dir").symlink_to(tmp_path, target_is_directory=True) + elif mutation == "symlink_root": + link = tmp_path / "linked-export" + link.symlink_to(root.parent, target_is_directory=True) + root = link / "dataset" + elif mutation == "symlink_receipt": + path = root.parent / "export-receipt.json" + outside = tmp_path / "outside-receipt" + path.rename(outside) + path.symlink_to(outside) + with pytest.raises(LocalDatasetError): + verify_local_export(root, digest) + + +@pytest.mark.parametrize( + "part,field,value", + [ + ("writer", "version", "0.5.2"), + ("writer", "source_commit", "0" * 40), + ("writer", "source_files", {}), + ("writer", "wheel_sha256", "0" * 64), + ("normalization_processor", "profile", "act"), + ("normalization_processor", "status", "failed"), + ("normalization_processor", "image_statistics_used", True), + ("normalization_processor", "resolved_feature_modes", {}), + ("normalization_processor", "source_files", {}), + ("loader_roundtrip", "frames_read", 999), + ("loader_roundtrip", "status", "failed"), + ("statistics_adapter", "training_rows_sha256", "0" * 64), + ("statistics_adapter", "training_partition_sha256", "0" * 64), + ("statistics_adapter", "source_dataset_sha256", "0" * 64), + ("statistics_adapter", "split_sha256", "0" * 64), + ("statistics_adapter", "statistics_sha256", "0" * 64), + ], +) +def test_bound_evidence(local_export, part, field, value): + root, receipt, _ = local_export + receipt[part][field] = value + digest = seal(root.parent, receipt) + with pytest.raises(LocalDatasetError): + verify_local_export(root, digest) + + +@pytest.mark.parametrize( + "mutation", + [ + "shape", + "quantile", + "nan", + "count", + "negative_std", + "mean_bounds", + "feature_names", + "v2", + "split", + "frame_binding", + "source_identity", + "partition_rows", + ], +) +def test_metadata_consistency_after_rehash(local_export, mutation): + root, receipt, _ = local_export + info = json.loads((root / "meta/info.json").read_text()) + stats = json.loads((root / "meta/stats.json").read_text()) + if mutation == "shape": + stats["action"]["mean"] = [0] + elif mutation == "quantile": + stats["action"]["q01"] = [0, 0] + elif mutation == "nan": + (root / "meta/stats.json").write_text('{"action": NaN}') + elif mutation == "count": + stats["action"]["count"] = [1] + elif mutation == "negative_std": + stats["action"]["std"] = [-1, 0] + elif mutation == "mean_bounds": + stats["action"]["mean"] = [9, 0] + elif mutation == "feature_names": + info["features"]["action"]["names"] = ["wrong", "names"] + elif mutation == "v2": + info["codebase_version"] = "v2.1" + elif mutation == "split": + info["splits"] = {"train": "0:1", "test": "0:1"} + elif mutation == "frame_binding": + path = root.parent / "provenance/frames.jsonl" + path.write_text(path.read_text().replace('"source_index": 0', '"source_index": 2')) + elif mutation == "source_identity": + path = root.parent / "provenance/source-identity.json" + value = json.loads(path.read_text()) + value["sha256"] = "0" * 64 + write_json(path, value) + elif mutation == "partition_rows": + path = root.parent / "provenance/partition-rows.json" + value = json.loads(path.read_text()) + value[0]["partition"] = "holdout" + write_json(path, value) + write_json(root / "meta/info.json", info) + if mutation != "nan": + write_json(root / "meta/stats.json", stats) + receipt["statistics_adapter"]["statistics_sha256"] = canonical(stats) + for replacement in receipt["statistics_adapter"]["replacement_statistics"]: + replacement["sha256"] = hashlib.sha256( + (root.parent / replacement["path"]).read_bytes() + ).hexdigest() + digest = seal(root.parent, receipt) + with pytest.raises(LocalDatasetError): + verify_local_export(root, digest) + + +def test_preflight_is_local_and_padding_is_real(local_cfg): + with ( + patch( + "tether.finetune.preflight.schema._fetch_dataset_features", + side_effect=AssertionError("Hub forbidden"), + ), + patch( + "tether.finetune.preflight.dataset_size._fetch_dataset_info", + side_effect=AssertionError("Hub forbidden"), + ), + ): + report = run_preflight(local_cfg) + assert not report.has_failures + assert check_schema(local_cfg).detail["uses_action_padding"] is True + assert check_dataset_size(local_cfg).detail["dataset_episodes"] == 1 + command = _build_lerobot_command(local_cfg) + assert f"--dataset.root={local_cfg.dataset_root}" in command + assert "--dataset.repo_id=studio-local/synthetic" in command + assert "--policy.type=smolvla" not in command + assert not any("use_imagenet_stats" in arg for arg in command) + detail = report.checks[0].detail + assert detail["effective_use_imagenet_stats"] is True + assert detail["visual_normalization"] == "IDENTITY" + + +def test_explicit_imagenet_setting_preserved(local_cfg): + local_cfg.extra_lerobot_args = {"dataset.use_imagenet_stats": "false"} + assert run_preflight(local_cfg).checks[0].detail["effective_use_imagenet_stats"] is False + assert "--dataset.use_imagenet_stats=false" in _build_lerobot_command(local_cfg) + + +@pytest.mark.parametrize( + "field,value", + [ + ("skip_preflight", True), + ("resume", True), + ("dataset_root", None), + ("dataset_manifest_sha256", None), + ("dataset_revision", "revision"), + ("policy", "act"), + ("mode", "full"), + ("phase", "distill"), + ("base_dataset", "other/dataset"), + ], +) +def test_local_escape_hatches_block_before_training(local_cfg, field, value): + setattr(local_cfg, field, value) + with patch("tether.finetune.run._run_lerobot_training") as train: + assert _validate_config(local_cfg) + assert run_finetune(local_cfg).status == "aborted" + assert run_preflight(local_cfg).has_failures + with pytest.raises(LocalDatasetError): + _build_lerobot_command(local_cfg) + train.assert_not_called() + + +@pytest.mark.parametrize( + "key", + [ + "dataset.root", + "dataset.repo_id", + "dataset.revision", + "dataset", + "policy.normalization_mapping", + "policy.normalization_mapping.VISUAL", + "policy.pretrained_path", + "config_path", + "output_dir", + "policy.input_features.observation.state", + "policy.output_features.action", + "rename_map", + "peft", + "peft.method_type", + "peft.method_type.other", + "rename_map.action", + "resume", + "dataset.episodes", + "policy.preprocessor_overrides", + "dataset.streaming", + "dataset.image_transforms", + "policy.max_action_dim", + "policy.max_state_dim", + "--dataset.root", + "dataset.root=evil", + ], +) +def test_extra_args_cannot_bypass(local_cfg, key): + local_cfg.extra_lerobot_args = {key: "conflict"} + assert _validate_config(local_cfg) + with pytest.raises(LocalDatasetError): + _build_lerobot_command(local_cfg) + + +@pytest.mark.parametrize( + "change", + [ + {"type": "act"}, + {"max_action_dim": 1}, + {"max_state_dim": 1}, + {"normalization_mapping": {}}, + {"max_action_dim": None}, + ], +) +def test_actual_base_contract_blocks(local_cfg, change): + path = Path(local_cfg.base) / "config.json" + config = json.loads(path.read_text()) + config.update(change) + write_json(path, config) + assert run_preflight(local_cfg).has_failures + with pytest.raises(LocalDatasetError): + _build_lerobot_command(local_cfg) + + +def test_missing_base_metadata_fails_local(local_cfg): + (Path(local_cfg.base) / "config.json").unlink() + assert run_preflight(local_cfg).has_failures + + +@pytest.mark.parametrize( + "error", [LocalDatasetError("integrity", "corrupt"), RuntimeError("unexpected")] +) +def test_verifier_failures_never_become_warning(local_cfg, error): + with ( + patch("tether.finetune.local_dataset.verify_local_export", side_effect=error), + patch("tether.finetune.run._run_lerobot_training") as train, + ): + report = run_preflight(local_cfg) + assert report.has_failures and report.checks[0].severity == "fail" + assert run_finetune(local_cfg).status == "aborted" + train.assert_not_called() + + +def test_schema_crash_fails_local(local_cfg): + with patch( + "tether.finetune.preflight.runner.check_schema", + autospec=True, + side_effect=RuntimeError("unexpected"), + ): + assert run_preflight(local_cfg).has_failures + + +def test_dry_run_never_trains(local_cfg): + with patch("tether.finetune.run._run_lerobot_training") as train: + result = run_finetune(local_cfg) + assert result.status == "ok" + train.assert_not_called() + assert ( + "effective_use_imagenet_stats: True" + in (local_cfg.output / "preflight_report.txt").read_text() + ) + + +def test_cli_flags_and_local_dry_run(local_cfg): + import typer + from typer.testing import CliRunner + from tether.finetune.cli import finetune_command + + app = typer.Typer() + app.command()(finetune_command) + runner = CliRunner() + with patch("tether.finetune.run._run_lerobot_training") as train: + result = runner.invoke( + app, + [ + "--base", + local_cfg.base, + "--dataset", + local_cfg.dataset, + "--dataset-root", + str(local_cfg.dataset_root), + "--dataset-manifest-sha256", + local_cfg.dataset_manifest_sha256, + "--output", + str(local_cfg.output), + "--dry-run", + ], + ) + assert result.exit_code == 0, result.output + assert "status: ok" in result.output + train.assert_not_called() + + +def test_output_cannot_mutate_export(local_cfg): + local_cfg.output = local_cfg.dataset_root.parent / "new-training-output" + assert run_finetune(local_cfg).status == "aborted" + assert not local_cfg.output.exists() + + +def test_missing_receipt_and_wrong_root(local_export): + root, _, digest = local_export + with pytest.raises(LocalDatasetError): + verify_local_export(root.parent, digest) + (root.parent / "export-receipt.json").unlink() + with pytest.raises(LocalDatasetError) as error: + verify_local_export(root, digest) + assert error.value.code == "missing-receipt" + + +def test_duplicate_json_fields_refused(local_export): + root, _, digest = local_export + path = root.parent / "export-receipt.json" + path.write_text(path.read_text().replace('"schema": 1', '"schema": 1, "schema": 1')) + with pytest.raises(LocalDatasetError) as error: + verify_local_export(root, digest) + assert error.value.code == "invalid-json" + + +def test_unsupported_requested_profile(local_export): + root, _, digest = local_export + with pytest.raises(LocalDatasetError) as error: + verify_local_export(root, digest, expected_profile="act") + assert error.value.code == "unsupported-profile" + + +def test_receipt_changed_after_preflight_blocks_command(local_cfg): + assert not run_preflight(local_cfg).has_failures + (local_cfg.dataset_root / "data/chunk-000/file-000.parquet").write_bytes(b"changed later") + with pytest.raises(LocalDatasetError) as error: + _build_lerobot_command(local_cfg) + assert error.value.code == "artifact-integrity" + + +def test_omitted_base_modes_use_only_pinned_defaults(local_cfg): + path = Path(local_cfg.base) / "config.json" + config = json.loads(path.read_text()) + del config["normalization_mapping"] + write_json(path, config) + result = check_schema(local_cfg) + assert result.severity == "ok" + assert ( + result.detail["normalization_mapping_source"] + == "pinned LeRobot 0.5.1 SmolVLAConfig defaults" + ) + config["normalization_mapping"] = None + write_json(path, config) + assert check_schema(local_cfg).severity == "fail" + + +def test_declared_source_config_fallback_remains_receipt_bound(local_export): + root, receipt, _ = local_export + case = receipt["episodes"][0]["case"] + case["identity_sources"].update( + { + "task": "source_config.task", + "embodiment": "source_config.embodiment", + "initial_state": "source_config.initial_state", + } + ) + path = root.parent / "provenance/frames.jsonl" + frames = [json.loads(line) for line in path.read_text().splitlines()] + rows_path = root.parent / "provenance/partition-rows.json" + rows = json.loads(rows_path.read_text()) + for index, frame in enumerate(frames): + del frame["example"]["task"] + del frame["example"]["embodiment"] + del frame["example"]["source"]["initial_state"] + rows[index]["example_sha256"] = canonical(frame["example"]) + path.write_text("".join(json.dumps(frame) + "\n" for frame in frames)) + write_json(rows_path, rows) + receipt["statistics_adapter"]["training_rows_sha256"] = canonical(rows[:2]) + digest = seal(root.parent, receipt) + assert verify_local_export(root, digest)["manifest_sha256"] == digest + case["identity_sources"]["task"] = "derived-task" + digest = seal(root.parent, receipt) + with pytest.raises(LocalDatasetError): + verify_local_export(root, digest) + + +def test_output_extra_cannot_target_immutable_export(local_cfg): + local_cfg.extra_lerobot_args = {"output_dir": str(local_cfg.dataset_root.parent / "training")} + assert run_finetune(local_cfg).status == "aborted" + with pytest.raises(LocalDatasetError): + _build_lerobot_command(local_cfg) + assert not (local_cfg.dataset_root.parent / "training").exists() + + +def test_existing_positional_config_arguments_stay_compatible(tmp_path): + cfg = FinetuneConfig("base", "owner/dataset", tmp_path, "base-rev", "data-rev", 123) + assert cfg.num_steps == 123 + assert cfg.dataset_root is None + assert cfg.dataset_manifest_sha256 is None