diff --git a/.agents/skills/blacknode-workflow/SKILL.md b/.agents/skills/blacknode-workflow/SKILL.md index 22df6a5..0ba9457 100644 --- a/.agents/skills/blacknode-workflow/SKILL.md +++ b/.agents/skills/blacknode-workflow/SKILL.md @@ -43,6 +43,25 @@ return here to integrate it into a validated workflow. 6. Use `PythonFn` only for workflow-local adapters. Use `blacknode-development` for reusable nodes or packages. +## Outcome-First Workflow Design + +- Begin with the requested outcome and build the shortest coherent graph that + produces it. +- A tracked template must perform a useful end-to-end task. Do not create + templates whose main result is proving a node works, showcasing wiring, + smoke testing, or confirming that a dependency is ready. +- Every visible node must materially create, transform, route, persist, + deploy, or operate something needed by the final result. +- Do not add checker, test, report, echo, or confirmation nodes as graph + padding. The node that owns an operation should validate inputs, preflight + dependencies, report progress, and return actionable errors when practical. +- Keep implementation validation in automated tests or untracked local + developer workflows rather than turning it into a product template. +- Preserve required physical-motion safety, authorization, cost consent, and + destructive-action confirmation. Integrate these controls into the owning + action or managed service when possible; keep a separate node when the + safety contract must be explicit and reusable. + ## Available Surfaces Preferred MCP stdio command: @@ -187,7 +206,7 @@ then return a concise graph plan with node ids, node types, key params, edges, entrypoint, and expected result. Build loop: -1. Understand the user goal and choose the smallest runnable graph. +1. Understand the user goal and choose the smallest outcome-producing graph. 2. Inspect list_nodes or get_node_schema before using unfamiliar nodes. 3. Create or load a workflow. 4. Add nodes with stable, descriptive ids. @@ -221,6 +240,8 @@ Use `list_nodes` for the live catalog. Current core groups: ## Graph Reliability Rules +- Make every node contribute directly to the requested result; never add a + node merely to prove, echo, check, or confirm another node. - Treat Blacknode workflows as DAGs. Do not create cycles or back-edges. - Always connect from `outputs` to `inputs`; never invent port names. - Respect types: `Any` accepts everything, exact type matches are valid, and diff --git a/AGENTS.md b/AGENTS.md index 5250d18..d2d0e81 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -43,6 +43,26 @@ under `packages/` are separate Git repositories and carry their own `AGENTS.md`. - Keep physical motion disarmed by default. Retain stale-data, joint-limit, and shutdown safeguards in every transport path. +## Outcome-first workflows and templates + +- Ship tracked templates only when they perform a real user or operator task + and produce the intended artifact, service, deployment, model, dataset, + action, or decision. +- Do not add tracked templates whose primary purpose is proving that a node + works, showcasing wiring, smoke testing, or confirming readiness. Put those + checks in automated tests or untracked local developer workflows. +- Keep graphs as short and direct as the outcome permits. Every visible node + must materially create, transform, route, persist, deploy, or operate + something required by the result. +- Do not pad workflows with separate checker, test, report, echo, or + confirmation nodes when the node that owns the operation can validate its + inputs, preflight dependencies, report progress, and return actionable + errors. +- Required physical-motion safety, authorization, cost consent, and + destructive-action confirmation remain mandatory. Prefer integrating these + controls into the owning action or managed service; use a separate node only + when the safety contract must remain explicit and reusable. + ## Managed Runtime release decision Decide whether a managed-device Runtime release is required before completing diff --git a/docs/agent-guide.md b/docs/agent-guide.md index 664178d..f80e02c 100644 --- a/docs/agent-guide.md +++ b/docs/agent-guide.md @@ -359,6 +359,19 @@ Use these instead: ## Template Workflow +Shared templates are outcome-producing product workflows, not test fixtures or +proof-of-concept graphs. A template should perform a real task and produce the +artifact, service, deployment, model, dataset, action, or decision its name +promises. Keep the graph short and direct, and include only nodes that +materially contribute to that result. + +Do not add checker, test, echo, report, or confirmation nodes merely to prove +that another node works. Put implementation checks in automated tests or local +developer workflows. Operation-owning nodes should handle practical preflight +validation, progress, and actionable failures. Required motion safety, +authorization, cost consent, and destructive-action confirmation still apply +and should be integrated into the owning action when the contract allows it. + To make a new shared template: 1. Build and save the workflow in the editor. diff --git a/docs/packages.md b/docs/packages.md index ac74295..c4b5742 100644 --- a/docs/packages.md +++ b/docs/packages.md @@ -169,7 +169,7 @@ development: | `blacknode-agent` | Persistent task memory and executive planning, mission execution, skill selection, confirmation, and review. | | `blacknode-motion` | Arm and base planning, trajectories, execution, learned policies, arbitration, and motion safety. | | `blacknode-cuda` | CUDA capability, image-processing, tensor-operation, and optional benchmark components backed by internal kernels. | -| `blacknode-dataset` | Default recording, replay, and validation with optional evaluation, export, and repository publishing. | +| `blacknode-dataset` | Default recording, replay, and validation with optional BlacknodeDataset adapters, evaluation, export, and repository publishing. | | `blacknode-drivers` | Selectively enabled concrete physical drivers; the `feetech` component provides inert bus configuration, read-only probing, and torque-safe bus primitives. | | `blacknode-isaac` | Direct closed-loop ACT and compatible PPO evaluation using Isaac Sim articulation state, semantic observations, safety-gated targets, and runtime replay logs. | | `blacknode-perception` | Camera, tracking, VLM, and spatial-perception components, organized as selectable components. | @@ -177,7 +177,7 @@ development: | `blacknode-ros2` | Native DDS graph, topic, service, and diagnostic integration with optional rosbridge and managed processes. | | `blacknode-runtime` | Authenticated remote deployment, target manifests, process supervision, logs, and rollback on Raspberry Pi, Jetson, and Linux targets. | | `blacknode-skills` | Task-level follow, pick-place, delivery, docking, and inspection behavior over stable capabilities. | -| `blacknode-training` | Optional dataset checks, managed jobs, checkpoints, policy previews, and deployable policy artifacts for training workloads. | +| `blacknode-training` | Optional managed policy, reinforcement-learning, and OpenPI π0.5 VLA training with deployable model artifacts. | Keep the layers separate: `blacknode-robot` owns profiles, calibration, connected devices, normalized telemetry, and the generic robot contract; diff --git a/docs/project-artifacts.md b/docs/project-artifacts.md index 4d097a5..30dadc1 100644 --- a/docs/project-artifacts.md +++ b/docs/project-artifacts.md @@ -23,6 +23,7 @@ of its files and native manifest. | Training run | `blacknode-training` | run ID, output path, phase, progress, losses | | Checkpoint | `blacknode-training` | checkpoint path, run ID, step | | Policy | `blacknode-training` | policy path, type, source checkpoint, dimensions | +| VLA model | `blacknode-training` | model ID, architecture, provider, base revision, dataset revision, checkpoint digest | | Replay evaluation | `blacknode-training` | episode, frames, aggregate errors | | Simulation run | `blacknode-isaac` | run ID, log path, phase, inference counters | @@ -56,7 +57,7 @@ Each indexed reference has this provider-neutral shape: ``` `artifact_type` is one of `dataset`, `training_run`, `checkpoint`, `policy`, -`evaluation`, or `simulation_run`. `status` is `available`, `running`, +`model`, `evaluation`, or `simulation_run`. `status` is `available`, `running`, `completed`, or `failed`. The ID is deterministic from provider, artifact type, and locator. Repeated @@ -78,6 +79,7 @@ The v1 importer understands: - `blacknode.training-run` - `blacknode.action-chunking-checkpoint` - `blacknode.policy-artifact` +- `blacknode.vla-model` - `blacknode.policy-replay-metrics` - `blacknode.policy-runtime` from an Isaac node @@ -103,7 +105,7 @@ successful node cook into a failed cook. | Stage | Complete evidence | |---|---| | Collect | A linked dataset reports one or more saved episodes | -| Train | A linked policy artifact exists | +| Train | A linked policy or VLA model artifact exists | | Simulate | A linked simulation run or evaluation is completed | A created empty dataset, running training job, checkpoint, or running diff --git a/docs/vla-training.md b/docs/vla-training.md new file mode 100644 index 0000000..01fdf67 --- /dev/null +++ b/docs/vla-training.md @@ -0,0 +1,51 @@ +# VLA Training + +Blacknode Cloud VLA Training turns a versioned robotics dataset into a persisted +Blacknode model. V0 supports OpenPI π0.5 LoRA training through its native JAX +stack. Warp remains available for simulation and synthetic-data workloads; it +is not part of the supervised π0.5 training path. + +## V0 workflow + +Use the **OpenPI π0.5 Fine-Tune** template. It contains two outcome-producing +nodes: + +1. `LeRobotDataset` resolves a local LeRobot v3 dataset or an immutable Hugging + Face dataset revision as a `blacknode.dataset-source`. +2. `OpenPIFineTune` adapts the source to the OpenPI-pinned LeRobot format, + computes normalization statistics, runs π0.5 JAX LoRA training, and exports + a `blacknode.vla-model`. + +The workflow entrypoint is the trained `model` output. It has no confirmation, +checker, or pass-through output nodes. Required dataset validation and artifact +integrity checks run inside the nodes that own those responsibilities. + +For a remote source, replace `PIN_DATASET_COMMIT` with the dataset repository's +immutable commit SHA before submitting the workflow. The Cloud executor uses an +NVIDIA L40S profile for this V0 workload. + +## BlacknodeDataset boundary + +`BlacknodeDataset` is the model-independent data boundary. Its lazy adapter +exposes episode metadata, timestamps, observations, actions, task language, +robot identity, and camera-frame references without loading a complete dataset +into memory. V0 provides native and LeRobot v3 adapters. Future ROS bag, +simulation, and robot-recorder sources can implement the same adapter contract. + +The OpenPI provider performs model-specific conversion. This keeps OpenPI, +JAX, checkpoint layout, action transforms, and normalization details outside +the dataset core. + +## Model artifact + +A completed run writes a `blacknode.vla-model` manifest alongside: + +- the LoRA checkpoint archive; +- training configuration and pinned OpenPI revision; +- normalization statistics; +- structured metrics and logs; +- dataset URI and immutable revision; +- inference compatibility metadata. + +The artifact remains disarmed: `physical_motion_authorized` is always `false`. +Deployment and real-robot inference are separate, guarded workflows. diff --git a/editor-server/artifact_store.py b/editor-server/artifact_store.py index 99efb1f..066742e 100644 --- a/editor-server/artifact_store.py +++ b/editor-server/artifact_store.py @@ -70,6 +70,8 @@ def _path_locator(value: Any) -> str: clean = _clean_text(value, maximum=2000) if not clean: return "" + if clean.startswith(("blacknode://", "blacknode-cloud://")): + return clean return str(Path(clean).expanduser().resolve()) @@ -192,6 +194,38 @@ def _artifact_candidate( "metrics", ), ) + elif kind == "blacknode.vla-model": + artifact_type = "model" + provider = "blacknode-training" + locator = _path_locator(payload.get("path")) + model_id = _clean_text(payload.get("model_id"), maximum=160) + name = model_id or (Path(locator).name if locator else "VLA model") + status = "completed" + metadata = _safe_metadata( + payload, + ( + "model_id", + "owner", + "provider", + "architecture", + "backend", + "base_model", + "base_model_revision", + "dataset", + "training_method", + "step", + "seed", + "action_horizon", + "action_mode", + "checkpoint", + "checkpoint_sha256", + "normalization", + "metrics", + "inference", + "physical_motion_authorized", + "job_id", + ), + ) elif kind == "blacknode.policy-replay-metrics": artifact_type = "evaluation" provider = "blacknode-training" @@ -432,6 +466,7 @@ def inspect_path( if payload.get("kind") in { "blacknode.episode-dataset", "blacknode.policy-artifact", + "blacknode.vla-model", }: payload["path"] = str(candidate.parent) elif not payload.get("path"): @@ -503,6 +538,6 @@ def _save(self, records: dict[str, dict[str, Any]]) -> None: def _hydrate(record: dict[str, Any]) -> dict[str, Any]: locator = str(record.get("locator") or "") exists = True - if locator and not locator.startswith("blacknode://"): + if locator and not locator.startswith(("blacknode://", "blacknode-cloud://")): exists = Path(locator).exists() return {**record, "exists": exists} diff --git a/editor-server/cloud_client.py b/editor-server/cloud_client.py index a7def3f..28081c6 100644 --- a/editor-server/cloud_client.py +++ b/editor-server/cloud_client.py @@ -8,7 +8,7 @@ import urllib.request from collections.abc import Iterator from dataclasses import dataclass -from typing import Any +from typing import Any, BinaryIO @dataclass(frozen=True) @@ -92,6 +92,48 @@ def chunks() -> Iterator[bytes]: return chunks(), media_type, disposition +def upload( + path: str, + stream: BinaryIO, + *, + size: int, + headers: dict[str, str], + authorization: str, + timeout: float = 86_400.0, +) -> dict[str, Any]: + config = configuration() + if not config.available: + raise CloudClientError(503, "Configure BLACKNODE_CLOUD_URL on the editor server.") + request = urllib.request.Request( + f"{config.base_url}{path}", + data=stream, + headers={ + "Accept": "application/json", + "Authorization": f"Bearer {authorization}", + "Content-Length": str(size), + "Content-Type": "application/gzip", + **headers, + }, + method="PUT", + ) + try: + response = urllib.request.urlopen(request, timeout=timeout) + except urllib.error.HTTPError as exc: + message = _error_message(exc) + exc.close() + raise CloudClientError(exc.code, message) from exc + except (OSError, urllib.error.URLError) as exc: + raise CloudClientError(502, "Blacknode Cloud is unreachable.") from exc + with response: + try: + value = json.loads(response.read().decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise CloudClientError(502, "Blacknode Cloud returned invalid JSON.") from exc + if not isinstance(value, dict): + raise CloudClientError(502, "Blacknode Cloud returned an invalid response.") + return value + + def _open( method: str, path: str, diff --git a/editor-server/device_registry.py b/editor-server/device_registry.py index 9fb5649..2f43d4b 100644 --- a/editor-server/device_registry.py +++ b/editor-server/device_registry.py @@ -632,6 +632,14 @@ def set_deployment_motion_armed( timeout=15.0, ) + def save_deployment_map(self, deployment_id: str) -> dict[str, Any]: + return self._request( + "POST", + f"{self._deployment_endpoint(deployment_id)}/control", + payload={"command": "save-map"}, + timeout=150.0, + ) + def ros2_diagnostics(self) -> dict[str, Any]: return self._request("GET", "/diagnostics/ros2", timeout=90.0) diff --git a/editor-server/server.py b/editor-server/server.py index 936ae1a..44e8372 100644 --- a/editor-server/server.py +++ b/editor-server/server.py @@ -1,6 +1,6 @@ """Blacknode editor backend — FastAPI server the React editor talks to.""" from __future__ import annotations -import asyncio, uuid, os, sys, json, threading, re, queue, io, contextlib, time, subprocess, importlib, signal, shlex, hashlib, math, copy, base64 +import asyncio, uuid, os, sys, json, threading, re, queue, io, contextlib, time, subprocess, importlib, signal, shlex, hashlib, math, copy, base64, tempfile from array import array import urllib.error, urllib.parse, urllib.request from concurrent.futures import ThreadPoolExecutor, wait @@ -503,6 +503,16 @@ class CloudJobReq(BaseModel): max_runtime_seconds: int = Field(default=3600, ge=60, le=86_400) +class CloudVLATrainReq(BaseModel): + dataset_uri: str = Field(min_length=1, max_length=1000) + dataset_revision: str = Field(default="", max_length=200) + steps: int = Field(default=5000, ge=1, le=10_000_000) + batch_size: int = Field(default=8, ge=1, le=1024) + action_horizon: int = Field(default=10, ge=1, le=256) + max_runtime_seconds: int = Field(default=14_400, ge=60, le=86_400) + project_ref: str | None = None + + class CloudLoginReq(BaseModel): email: str = Field(min_length=3, max_length=254) password: str = Field(min_length=1, max_length=200) @@ -4536,6 +4546,132 @@ def create_cloud_job(req: CloudJobReq, request: Request): ) +@app.get("/cloud/datasets") +def list_cloud_datasets(request: Request): + return _cloud_user_call(request, "GET", "/v1/datasets") + + +@app.put("/cloud/datasets") +async def upload_cloud_dataset(request: Request): + name = Path(request.headers.get("X-Dataset-Name", "")).name + if not name.endswith((".tar.gz", ".tgz")): + raise HTTPException(400, "Choose a .tar.gz LeRobot dataset archive.") + descriptor, temporary_name = tempfile.mkstemp(prefix="blacknode-dataset-", suffix=".tar.gz") + os.close(descriptor) + temporary = Path(temporary_name).resolve() + digest = hashlib.sha256() + size = 0 + try: + with temporary.open("wb") as handle: + async for chunk in request.stream(): + size += len(chunk) + if size > 50 * 1024 * 1024 * 1024: + raise HTTPException(413, "Dataset archive exceeds the 50 GiB V0 limit.") + digest.update(chunk) + handle.write(chunk) + if not size: + raise HTTPException(400, "Dataset archive is empty.") + sha256 = digest.hexdigest() + asset_id = f"dataset_{sha256[:32]}" + try: + with temporary.open("rb") as stream: + return cloud_client.upload( + f"/v1/datasets/{asset_id}", + stream, + size=size, + headers={ + "X-Dataset-Name": name, + "X-Dataset-SHA256": sha256, + "X-Dataset-Size": str(size), + }, + authorization=_cloud_session(request).token, + ) + except cloud_client.CloudClientError as exc: + raise HTTPException(exc.status, str(exc)) from exc + finally: + temporary.unlink(missing_ok=True) + + +@app.post("/cloud/vla/jobs") +def create_cloud_vla_job(req: CloudVLATrainReq, request: Request): + uri = req.dataset_uri.strip() + revision = req.dataset_revision.strip() + if uri.startswith("hf://") and not revision: + raise HTTPException(400, "Pin the Hugging Face dataset to an immutable revision.") + if not uri.startswith(("hf://", "blacknode-cloud://datasets/")): + raise HTTPException(400, "Choose a pinned Hugging Face or uploaded Cloud dataset.") + workflow = { + "kind": "blacknode.workflow", + "schema_version": 1, + "name": "OpenPI π0.5 Fine-Tune", + "entrypoint": {"node_id": "train", "port": "model"}, + "metadata": { + "source": "blacknode-editor-vla", + "required_packages": ["blacknode-dataset", "blacknode-training"], + "required_components": [ + "blacknode-dataset/adapters", + "blacknode-training/vla-openpi", + ], + "cloud": {"workload": "vla_train", "gpu_class": "l40s"}, + "safety": {"physical_motion_authorized": False}, + }, + "node_meta": { + "dataset": { + "id": "dataset", + "type": "LeRobotDataset", + "params": {"uri": uri, "revision": revision, "source_uri": ""}, + "inputs": ["trigger", "uri", "revision", "source_uri"], + "outputs": ["dataset", "uri", "revision", "report"], + }, + "train": { + "id": "train", + "type": "OpenPIFineTune", + "params": { + "action": "run", + "dataset": {}, + "run_id": f"pi05-{uuid.uuid4().hex[:12]}", + "output_dir": "", + "steps": req.steps, + "batch_size": req.batch_size, + "action_horizon": req.action_horizon, + "action_mode": "absolute_joint", + "learning_rate": 0.00005, + "save_interval": min(1000, req.steps), + "seed": 42, + "resume": True, + "overwrite": False, + }, + "inputs": [ + "trigger", "action", "dataset", "run_id", "output_dir", "steps", + "batch_size", "action_horizon", "action_mode", "learning_rate", + "save_interval", "seed", "resume", "overwrite", + ], + "outputs": [ + "ok", "running", "phase", "step", "progress", "status", "metrics", + "model", "model_path", "report", + ], + }, + }, + "edges": [ + {"from": "dataset", "from_port": "dataset", "to": "train", "to_port": "dataset"} + ], + } + return _cloud_user_call( + request, + "POST", + "/v1/jobs", + { + "contract_version": "blacknode.cloud.jobs/v1", + "project_ref": req.project_ref, + "workflow": workflow, + "compute": { + "gpu_class": "l40s", + "gpu_count": 1, + "max_runtime_seconds": req.max_runtime_seconds, + }, + "runtime": {"release": "gpu-development"}, + }, + ) @app.get("/cloud/jobs/{job_id}") def get_cloud_job(job_id: str, request: Request): return _cloud_user_call(request, "GET", f"/v1/jobs/{_cloud_job_id(job_id)}") @@ -8274,6 +8410,31 @@ def _workflow_motion_controls(workflow: dict[str, Any]) -> list[dict[str, str]]: return controls +def _workflow_mapping_controls(workflow: dict[str, Any]) -> list[dict[str, Any]]: + """Capture the portable MapEnvironment settings for device-side controls.""" + controls: list[dict[str, Any]] = [] + for node_id, meta in (workflow.get("node_meta") or {}).items(): + if not isinstance(meta, dict) or str(meta.get("type") or "") != "MapEnvironment": + continue + params = meta.get("params") if isinstance(meta.get("params"), dict) else {} + controls.append({ + "kind": "slam_toolbox", + "node_id": str(node_id), + "map_topic": str(params.get("map_topic") or "/map"), + "save_directory": str(params.get("save_directory") or "~/Blacknode/maps"), + "map_name": str(params.get("map_name") or "map_01"), + "save_map_service": str( + params.get("save_map_service") or "/slam_toolbox/save_map" + ), + "serialize_service": str( + params.get("serialize_service") or "/slam_toolbox/serialize_map" + ), + "serialize_pose_graph": bool(params.get("serialize_pose_graph", True)), + "service_timeout": float(params.get("service_timeout") or 30.0), + }) + return controls + + def _disarm_workflow_motion_controls(workflow: dict[str, Any]) -> list[str]: """Force remotely controlled motion gates off in the deployed snapshot.""" node_meta = workflow.get("node_meta") @@ -11335,6 +11496,7 @@ def report(percent: int, message: str) -> None: "required_capabilities": _workflow_required_capabilities(workflow), "telemetry_required": _workflow_requires_deployment_telemetry(workflow), "motion_controls": _workflow_motion_controls(workflow), + "mapping_controls": _workflow_mapping_controls(workflow), "required_packages": _workflow_target_packages(workflow), "package_requirements": _workflow_target_package_specs(workflow), "blacknode_version": str(getattr(bn, "__version__", "")), @@ -11496,6 +11658,73 @@ def control_device_deployment_motion( return result +@app.post("/devices/{device_id}/deployments/{deployment_id}/mapping/save") +def save_device_deployment_map(device_id: str, deployment_id: str): + deployment = _require_targeted_deployment(device_id, deployment_id) + if str(deployment.get("state") or "") != "running": + raise HTTPException(409, "Start mapping before saving the map.") + if int(deployment.get("mapping_control_count") or 0) != 1: + raise HTTPException(409, "This deployment does not contain one MapEnvironment control.") + try: + return _runtime_client_or_404(device_id).save_deployment_map(deployment_id) + except DeviceRegistryError as exc: + detail = str(exc) + if "HTTP 404" in detail or "not found" in detail.casefold(): + raise HTTPException( + 409, + "This Runtime cannot save deployed maps yet. Update blacknode-runtime " + "to 0.4.13 or newer and stage the mapping workflow again.", + ) from exc + raise HTTPException(502, detail) from exc + + +@app.get("/devices/{device_id}/deployments/{deployment_id}/mapping/snapshot") +def get_device_deployment_map_snapshot(device_id: str, deployment_id: str): + deployment = _require_targeted_deployment(device_id, deployment_id) + if str(deployment.get("state") or "") != "running": + raise HTTPException(409, "Start mapping to view the live occupancy map.") + if int(deployment.get("mapping_control_count") or 0) != 1: + raise HTTPException(409, "This deployment does not contain one MapEnvironment control.") + topic = str(deployment.get("mapping_topic") or "/map").strip() + stream_id = ("map-" + re.sub(r"[^a-z0-9-]+", "-", deployment_id.lower()))[:64].rstrip("-") + try: + client = _runtime_client_or_404(device_id) + response = client.ros2_topic_status(stream_id) + current_outputs = ( + response.get("outputs") + if isinstance(response.get("outputs"), dict) + else {} + ) + current_status = ( + current_outputs.get("status") + if isinstance(current_outputs.get("status"), dict) + else {} + ) + if not current_status.get("worker_alive"): + response = client.start_ros2_topic( + stream_id, + { + "topic": topic, + "message_type": "nav_msgs/msg/OccupancyGrid", + "node_name": "blacknode_mapping_view", + "history": 1, + "timeout": 3.0, + "stale_after_seconds": 5.0, + "qos": "transient_local", + }, + ) + except DeviceRegistryError as exc: + raise HTTPException(502, str(exc)) from exc + outputs = response.get("outputs") if isinstance(response.get("outputs"), dict) else {} + return { + "deployment_id": deployment_id, + "topic": topic, + "message": outputs.get("message") if isinstance(outputs.get("message"), dict) else {}, + "status": outputs.get("status") if isinstance(outputs.get("status"), dict) else {}, + "report": str(outputs.get("report") or ""), + } + + @app.get("/devices/{device_id}/ros2-diagnostics") def get_device_ros2_diagnostics(device_id: str): try: @@ -11534,10 +11763,18 @@ def start_device_deployment(device_id: str, deployment_id: str): @app.post("/devices/{device_id}/deployments/{deployment_id}/stop") def stop_device_deployment(device_id: str, deployment_id: str): - _require_targeted_deployment(device_id, deployment_id) + current = _require_targeted_deployment(device_id, deployment_id) try: runtime_client = _runtime_client_or_404(device_id) deployment = runtime_client.stop_deployment(deployment_id) + if int(current.get("mapping_control_count") or 0) == 1: + stream_id = ( + "map-" + re.sub(r"[^a-z0-9-]+", "-", deployment_id.lower()) + )[:64].rstrip("-") + try: + runtime_client.stop_ros2_topic(stream_id) + except DeviceRegistryError: + pass except DeviceRegistryError as exc: raise HTTPException(502, str(exc)) from exc remaining = _target_deployment_records( diff --git a/editor/src/App.tsx b/editor/src/App.tsx index a56a36f..d3d55a0 100644 --- a/editor/src/App.tsx +++ b/editor/src/App.tsx @@ -1639,6 +1639,45 @@ function WorkspaceApp() { } }, [cloudJobPending, serverOk]) + const handleCloudVlaRun = useCallback(async (request: { + dataset_uri: string + dataset_revision: string + steps: number + batch_size: number + action_horizon: number + max_runtime_seconds: number + }) => { + if (cloudJobPending) return + setCloudJobPending(true) + setCloudJobError('') + try { + setCloudJob(null) + const created = await api.createCloudVlaJob({ + ...request, + project_ref: activeProject?.id ?? activeTab?.slug ?? null, + }) + setCloudJob(created) + setCloudPanelView('job') + void api.cloudStatus().then(setCloudAccountStatus).catch(() => undefined) + } catch (cause) { + const message = cause instanceof Error ? cause.message : String(cause) + setCloudJobError(message) + throw cause + } finally { + setCloudJobPending(false) + } + }, [activeProject?.id, activeTab?.slug, cloudJobPending]) + + const handleCloudJobCompleted = useCallback((job: CloudJob) => { + if (!activeProject || job.result === null || job.result === undefined) return + void api.importProjectArtifacts(activeProject.id, { + node_type: job.workload_kind === 'vla_train' ? 'OpenPIFineTune' : '', + value: job.result, + }).then(() => { + useStore.setState(state => ({ projectRevision: state.projectRevision + 1 })) + }).catch(() => undefined) + }, [activeProject]) + const handleResetRun = useCallback(() => { stopCook() setActiveRunMode(null) @@ -2819,6 +2858,8 @@ function WorkspaceApp() { accountStatus={cloudAccountStatus} onAccountStatus={setCloudAccountStatus} onRun={() => void handleCloudRun()} + onRunVla={handleCloudVlaRun} + onJobCompleted={handleCloudJobCompleted} onClose={() => setCloudPanelOpen(false)} /> diff --git a/editor/src/api.ts b/editor/src/api.ts index 2071051..875e087 100644 --- a/editor/src/api.ts +++ b/editor/src/api.ts @@ -809,6 +809,7 @@ export type ProjectArtifactType = | 'training_run' | 'checkpoint' | 'policy' + | 'model' | 'simulation_run' | 'evaluation' @@ -992,6 +993,9 @@ export interface RemoteDeployment { error: string motion_armed?: boolean motion_control_count?: number + mapping_control_count?: number + mapping_topic?: string + last_map_artifact?: Record created_at: string updated_at: string } @@ -1101,6 +1105,8 @@ export interface CloudJob { id: string project_ref: string | null workflow_name: string + workload_kind: 'workflow' | 'vla_train' + compute_provider: string status: CloudJobStatus cleanup_status: string progress: number @@ -1133,6 +1139,39 @@ export interface CloudArtifact { created_at: string } +export interface MappingSnapshot { + deployment_id: string + topic: string + message: { + header?: Record + info?: { + resolution?: number + width?: number + height?: number + origin?: Record + } + data?: number[] + } + status: { + state?: string + source_fresh?: boolean + age_seconds?: number | null + received?: number + error?: string + } + report: string +} + +export interface CloudDataset { + id: string + kind: 'blacknode.cloud-dataset' + name: string + size_bytes: number + sha256: string + media_type: string + locator: string +} + export interface RunRecord extends RunSummary { events: Array & { type: string; ts?: string | number }> workflow?: WorkflowSnapshot @@ -1755,6 +1794,27 @@ export const api = { workflow_name: workflowName, project_ref: projectRef ?? null, }), + listCloudDatasets: () => req('GET', '/cloud/datasets'), + uploadCloudDataset: async (file: File) => { + const response = await fetchBackend('/cloud/datasets', { + method: 'PUT', + headers: { + 'Content-Type': 'application/gzip', + 'X-Dataset-Name': file.name, + }, + body: file, + }) + return responseJson(response, '/cloud/datasets') + }, + createCloudVlaJob: (payload: { + dataset_uri: string + dataset_revision: string + steps: number + batch_size: number + action_horizon: number + max_runtime_seconds: number + project_ref?: string | null + }) => req('POST', '/cloud/vla/jobs', payload), getCloudJob: (jobId: string) => req('GET', `/cloud/jobs/${encodeURIComponent(jobId)}`), cancelCloudJob: (jobId: string) => @@ -2412,6 +2472,26 @@ export const api = { { armed }, 20000, ), + saveRemoteDeploymentMap: (deviceId: string, deploymentId: string) => + req<{ + ok: boolean + id: string + artifact: Record + warning?: string + deployment: RemoteDeployment + }>( + 'POST', + `/devices/${encodeURIComponent(deviceId)}/deployments/${encodeURIComponent(deploymentId)}/mapping/save`, + {}, + 150000, + ), + remoteDeploymentMapSnapshot: (deviceId: string, deploymentId: string) => + req( + 'GET', + `/devices/${encodeURIComponent(deviceId)}/deployments/${encodeURIComponent(deploymentId)}/mapping/snapshot`, + undefined, + 15000, + ), remoteRos2Diagnostics: (deviceId: string) => req( 'GET', diff --git a/editor/src/components/CloudRunPanel.tsx b/editor/src/components/CloudRunPanel.tsx index 95882c2..a1c2b2d 100644 --- a/editor/src/components/CloudRunPanel.tsx +++ b/editor/src/components/CloudRunPanel.tsx @@ -4,6 +4,7 @@ import { api, type CloudArtifact, type CloudCreditEntry, + type CloudDataset, type CloudJob, type CloudJobEvent, type CloudProviderPreference, @@ -21,6 +22,15 @@ interface Props { accountStatus: CloudStatus | null onAccountStatus: (status: CloudStatus) => void onRun: () => void + onRunVla: (request: { + dataset_uri: string + dataset_revision: string + steps: number + batch_size: number + action_horizon: number + max_runtime_seconds: number + }) => Promise + onJobCompleted: (job: CloudJob) => void onClose: () => void } @@ -56,6 +66,8 @@ export default function CloudRunPanel({ accountStatus, onAccountStatus, onRun, + onRunVla, + onJobCompleted, onClose, }: Props) { const [job, setJob] = useState(initialJob) @@ -73,7 +85,19 @@ export default function CloudRunPanel({ const [providerPending, setProviderPending] = useState(false) const [providerError, setProviderError] = useState('') const [providerPreference, setProviderPreference] = useState('auto') + const [datasets, setDatasets] = useState([]) + const [vlaSource, setVlaSource] = useState<'huggingface' | 'cloud'>('huggingface') + const [vlaDatasetUri, setVlaDatasetUri] = useState('hf://lerobot/aloha_sim_insertion_human') + const [vlaDatasetRevision, setVlaDatasetRevision] = useState('') + const [vlaDatasetId, setVlaDatasetId] = useState('') + const [vlaSteps, setVlaSteps] = useState(5000) + const [vlaBatchSize, setVlaBatchSize] = useState(8) + const [vlaActionHorizon, setVlaActionHorizon] = useState(10) + const [vlaRuntime, setVlaRuntime] = useState(14400) + const [vlaPending, setVlaPending] = useState(false) + const [vlaError, setVlaError] = useState('') const nextSeq = useRef(0) + const completedJobId = useRef('') const refreshAccount = useCallback(async () => { const nextStatus = await api.cloudStatus() @@ -101,6 +125,16 @@ export default function CloudRunPanel({ .catch(cause => setAuthError(cause instanceof Error ? cause.message : String(cause))) }, [accountStatus?.authenticated, accountStatus?.account?.id, open]) + useEffect(() => { + if (!open || !accountStatus?.authenticated) return + void api.listCloudDatasets() + .then(items => { + setDatasets(items) + setVlaDatasetId(current => current || items[0]?.id || '') + }) + .catch(cause => setVlaError(cause instanceof Error ? cause.message : String(cause))) + }, [accountStatus?.authenticated, open]) + useEffect(() => { setProviderPreference( accountStatus?.account?.compute_provider_preference @@ -111,7 +145,11 @@ export default function CloudRunPanel({ useEffect(() => { if (job && TERMINAL.has(job.status)) void refreshAccount() - }, [job?.status, refreshAccount]) + if (job?.status === 'COMPLETED' && completedJobId.current !== job.id) { + completedJobId.current = job.id + onJobCompleted(job) + } + }, [job, onJobCompleted, refreshAccount]) useEffect(() => { if (!open || view !== 'job' || !initialJob) return undefined @@ -169,6 +207,19 @@ export default function CloudRunPanel({ return `${x.toFixed(1)},${y.toFixed(1)}` }).join(' ') }, [reward]) + const loss = metrics.filter(metric => metric.name.toLowerCase() === 'loss').slice(-60) + const lossPoints = useMemo(() => { + if (loss.length < 2) return '' + const values = loss.map(item => item.value) + const low = Math.min(...values) + const high = Math.max(...values) + const span = Math.max(0.0001, high - low) + return values.map((value, index) => { + const x = (index / (values.length - 1)) * 300 + const y = 76 - ((value - low) / span) * 68 + return `${x.toFixed(1)},${y.toFixed(1)}` + }).join(' ') + }, [loss]) const logs = events.filter(event => event.type === 'log').slice(-300) const submitAuth = async (event: FormEvent) => { @@ -235,6 +286,48 @@ export default function CloudRunPanel({ } } + const uploadDataset = async (file: File | undefined) => { + if (!file) return + setVlaPending(true) + setVlaError('') + try { + const uploaded = await api.uploadCloudDataset(file) + setDatasets(current => [uploaded, ...current.filter(item => item.id !== uploaded.id)]) + setVlaDatasetId(uploaded.id) + setVlaSource('cloud') + } catch (cause) { + setVlaError(cause instanceof Error ? cause.message : String(cause)) + } finally { + setVlaPending(false) + } + } + + const runVla = async () => { + const datasetUri = vlaSource === 'cloud' + ? (vlaDatasetId ? `blacknode-cloud://datasets/${vlaDatasetId}` : '') + : vlaDatasetUri.trim() + if (!datasetUri) { + setVlaError('Choose a dataset.') + return + } + setVlaPending(true) + setVlaError('') + try { + await onRunVla({ + dataset_uri: datasetUri, + dataset_revision: vlaSource === 'huggingface' ? vlaDatasetRevision.trim() : '', + steps: vlaSteps, + batch_size: vlaBatchSize, + action_horizon: vlaActionHorizon, + max_runtime_seconds: vlaRuntime, + }) + } catch (cause) { + setVlaError(cause instanceof Error ? cause.message : String(cause)) + } finally { + setVlaPending(false) + } + } + if (!open) return null const credits = accountStatus?.credits const account = accountStatus?.account @@ -245,6 +338,9 @@ export default function CloudRunPanel({ : Math.max(credits?.locked ?? 0, credits?.available ?? 0) const activeJob = job && !TERMINAL.has(job.status) const visibleJob = view === 'job' ? job : null + const vlaModel = visibleJob?.result && typeof visibleJob.result === 'object' + ? visibleJob.result as Record + : null const providerOptions = accountStatus?.compute_providers?.options ?? [] const selectedProvider = providerOptions.find(option => option.id === providerPreference) const providerLabel = selectedProvider?.label ?? 'Auto' @@ -261,7 +357,7 @@ export default function CloudRunPanel({ {(pending || (!accountStatus && !error)) &&
Connecting to Blacknode Cloud…
} - {(error || pollError || authError || providerError) &&
{error || pollError || authError || providerError}
} + {(error || pollError || authError || providerError || vlaError) &&
{error || pollError || authError || providerError || vlaError}
} {!pending && accountStatus && !accountStatus.configured && !error && (
Blacknode Cloud is not configured on this editor server.
)} @@ -370,6 +466,55 @@ export default function CloudRunPanel({ +
+
Fine Tune VLAπ0.5 · JAX LoRA
+ + {vlaSource === 'huggingface' ? ( + <> + + + + ) : ( + <> + + + + )} +
+ + + + +
+ GPU is selected by Blacknode Cloud. V0 resolves to one NVIDIA L40S and produces a downloadable Blacknode VLA model. + +
+
Credit history{history.length}
@@ -399,8 +544,13 @@ export default function CloudRunPanel({
Job{visibleJob.id}
GPUNVIDIA L40S
+
Workload{visibleJob.workload_kind === 'vla_train' ? 'VLA training' : 'Workflow'}
+
Provider{visibleJob.compute_provider}
Status● {visibleJob.status}
Runtime{elapsed(visibleJob)}
+ {vlaModel?.kind === 'blacknode.vla-model' && ( +
Inference{(vlaModel.inference as Record | undefined)?.verified ? 'Verified' : 'Unavailable'}
+ )}
@@ -416,6 +566,15 @@ export default function CloudRunPanel({
)} + {lossPoints && ( +
+
Loss{loss[loss.length - 1]?.value.toFixed(4)}
+ + + +
+ )} +
Logs{logs.length} events
{logs.length
diff --git a/editor/src/components/DeploymentsPanel.tsx b/editor/src/components/DeploymentsPanel.tsx
index db13d39..67cb2ce 100644
--- a/editor/src/components/DeploymentsPanel.tsx
+++ b/editor/src/components/DeploymentsPanel.tsx
@@ -1,4 +1,4 @@
-import { useEffect, useMemo, useState, type CSSProperties } from 'react'
+import { useEffect, useMemo, useRef, useState, type CSSProperties } from 'react'
 import {
   api,
   type ComputeDevice,
@@ -12,6 +12,7 @@ import {
   type DeviceActionProgress,
   type HardwareDevice,
   type HardwareDeviceStatus,
+  type MappingSnapshot,
   type RemoteDeployment,
   type RemoteDeploymentState,
 } from '../api'
@@ -794,6 +795,29 @@ export default function DeploymentsPanel({
     }
   }
 
+  const saveRemoteMap = async (deployment: RemoteDeployment) => {
+    if (!selectedDeviceId) return
+    setBusy(true)
+    setError(null)
+    try {
+      const result = await api.saveRemoteDeploymentMap(
+        selectedDeviceId,
+        deployment.id,
+      )
+      await refreshRemote()
+      const mapPath = String(result.artifact?.map_yaml || result.artifact?.directory || '')
+      setRemoteNotice(
+        `Map "${String(result.artifact?.map_name || 'map')}" saved on the device${
+          mapPath ? ` at ${mapPath}` : ''
+        }.${result.warning ? ` ${result.warning}` : ''}`,
+      )
+    } catch (err) {
+      setError(err instanceof Error ? err.message : String(err))
+    } finally {
+      setBusy(false)
+    }
+  }
+
   const runRosDiagnostics = async () => {
     if (!selectedDeviceId) return
     setBusy(true)
@@ -1383,6 +1407,7 @@ export default function DeploymentsPanel({
            openRemoteWorkflow(deployment)}
             onStart={() => startRemote(deployment)}
             onSetMotion={armed => setRemoteMotion(deployment, armed)}
+            onSaveMap={() => saveRemoteMap(deployment)}
             onStop={() => actRemote(() => (
               api.stopRemoteDeployment(selectedDeviceId, deployment.id)
             ))}
@@ -1508,6 +1534,7 @@ function PreflightResult({
 
 function RemoteDeploymentRow({
   deployment,
+  targetDeviceId,
   busy,
   canStage,
   expanded,
@@ -1517,11 +1544,13 @@ function RemoteDeploymentRow({
   onOpenWorkflow,
   onStart,
   onSetMotion,
+  onSaveMap,
   onStop,
   onRollback,
   onDelete,
 }: {
   deployment: RemoteDeployment
+  targetDeviceId: string
   busy: boolean
   canStage: boolean
   expanded: boolean
@@ -1531,11 +1560,13 @@ function RemoteDeploymentRow({
   onOpenWorkflow: () => void
   onStart: () => void
   onSetMotion: (armed: boolean) => void
+  onSaveMap: () => void
   onStop: () => void
   onRollback: () => void
   onDelete: () => void
 }) {
   const isRunning = deployment.state === 'running'
+  const isMapping = Number(deployment.mapping_control_count || 0) === 1
   const canRollback = deployment.revisions.length > 1
   const badges = [
     deployment.project_id
@@ -1577,8 +1608,13 @@ function RemoteDeploymentRow({
           
{log.trim() || 'No remote output captured yet.'}

           

             {isRunning
-              ? 
-              : }
+              ? 
+              : }
+            {isRunning && isMapping && (
+              
+            )}
             
@@ -1614,12 +1650,95 @@ function RemoteDeploymentRow({
               Delete
             
           

+          {isRunning && isMapping && (
+            
+          )}
)}
) } +function LiveOccupancyMap({ + deviceId, + deploymentId, + topic, +}: { + deviceId: string + deploymentId: string + topic: string +}) { + const canvasRef = useRef(null) + const [snapshot, setSnapshot] = useState(null) + const [message, setMessage] = useState('Connecting to the live map…') + + useEffect(() => { + let cancelled = false + const pull = async () => { + try { + const next = await api.remoteDeploymentMapSnapshot(deviceId, deploymentId) + if (cancelled) return + setSnapshot(next) + setMessage(next.report || 'Waiting for occupancy data…') + } catch (err) { + if (!cancelled) setMessage(err instanceof Error ? err.message : String(err)) + } + } + void pull() + const timer = window.setInterval(pull, 2000) + return () => { cancelled = true; window.clearInterval(timer) } + }, [deviceId, deploymentId]) + + useEffect(() => { + const canvas = canvasRef.current + const info = snapshot?.message?.info + const data = snapshot?.message?.data + const width = Math.max(0, Number(info?.width || 0)) + const height = Math.max(0, Number(info?.height || 0)) + if (!canvas || !Array.isArray(data) || !width || !height || data.length < width * height) return + canvas.width = width + canvas.height = height + const context = canvas.getContext('2d') + if (!context) return + const image = context.createImageData(width, height) + for (let y = 0; y < height; y += 1) { + for (let x = 0; x < width; x += 1) { + const sourceIndex = y * width + x + const targetIndex = ((height - y - 1) * width + x) * 4 + const occupancy = Number(data[sourceIndex]) + const color = occupancy < 0 ? [30, 36, 48] : occupancy >= 65 ? [20, 24, 31] : [226, 232, 240] + image.data[targetIndex] = color[0] + image.data[targetIndex + 1] = color[1] + image.data[targetIndex + 2] = color[2] + image.data[targetIndex + 3] = 255 + } + } + context.putImageData(image, 0, 0) + }, [snapshot]) + + const info = snapshot?.message?.info + const fresh = snapshot?.status?.source_fresh !== false && Boolean(snapshot?.message?.data?.length) + return ( +
+
+ Live mapping · {topic} + {fresh ? 'LIVE' : 'WAITING'} +
+ +
+ {message} + {info?.width && info?.height && ( + {info.width} × {info.height} cells · {Number(info.resolution || 0).toFixed(3)} m/cell + )} +
+
+ ) +} + function DeploymentRow({ deployment, busy, expanded, log, onToggle, onStart, onStop, onExport, onDelete }: { deployment: Deployment busy: boolean diff --git a/editor/src/components/ProjectPanel.tsx b/editor/src/components/ProjectPanel.tsx index 55e72be..b481553 100644 --- a/editor/src/components/ProjectPanel.tsx +++ b/editor/src/components/ProjectPanel.tsx @@ -62,6 +62,7 @@ const ARTIFACT_LABELS: Record = { training_run: 'Training run', checkpoint: 'Checkpoint', policy: 'Policy', + model: 'VLA model', simulation_run: 'Simulation run', evaluation: 'Evaluation', } @@ -85,6 +86,11 @@ function artifactDetail(artifact: ProjectArtifact): string { ? `${frames} frame${frames === 1 ? '' : 's'} evaluated` : `${artifact.status} · ${artifact.provider}` } + if (artifact.artifact_type === 'model') { + const architecture = String(metadata.architecture ?? 'VLA') + const backend = String(metadata.backend ?? '') + return `${architecture}${backend ? ` · ${backend}` : ''} · ${artifact.status}` + } return `${artifact.status} · ${artifact.provider}` } @@ -398,10 +404,10 @@ export default function ProjectPanel() { artifact => Number(artifact.metadata.episode_count ?? 0) > 0, ) const trainingArtifacts = selected?.artifacts.filter( - artifact => ['training_run', 'checkpoint', 'policy'].includes(artifact.artifact_type), + artifact => ['training_run', 'checkpoint', 'policy', 'model'].includes(artifact.artifact_type), ) ?? [] const policyArtifacts = trainingArtifacts.filter( - artifact => artifact.artifact_type === 'policy' && artifact.exists, + artifact => ['policy', 'model'].includes(artifact.artifact_type) && artifact.exists, ) const runningTraining = trainingArtifacts.filter( artifact => artifact.status === 'running', @@ -497,7 +503,7 @@ export default function ProjectPanel() { ? 'available' : 'optional', detail: policyArtifacts.length - ? `${policyArtifacts.length} policy artifact${policyArtifacts.length === 1 ? '' : 's'} ready` + ? `${policyArtifacts.length} trained model${policyArtifacts.length === 1 ? '' : 's'} ready` : runningTraining.length ? `${runningTraining.length} training run${runningTraining.length === 1 ? '' : 's'} running` : trainingArtifacts.length diff --git a/editor/src/index.css b/editor/src/index.css index c3f425c..9a61587 100644 --- a/editor/src/index.css +++ b/editor/src/index.css @@ -5429,6 +5429,47 @@ button.bn-device-fact { white-space: normal; } +.bn-live-map { + margin-top: 12px; + padding: 10px; + background: #111720; + border: 1px solid var(--line2); + border-radius: 6px; +} + +.bn-live-map-head, +.bn-live-map-foot { + display: flex; + justify-content: space-between; + gap: 10px; + color: var(--tx3); + font-family: var(--font-mono); + font-size: 11px; +} + +.bn-live-map-head strong { + color: var(--tx1); +} + +.bn-live-map-head span.is-live { + color: var(--ok); +} + +.bn-live-map canvas { + display: block; + width: 100%; + max-height: 420px; + margin: 9px 0; + background: #1e2430; + image-rendering: pixelated; + object-fit: contain; +} + +.bn-live-map-foot { + align-items: flex-start; + flex-wrap: wrap; +} + @container deployment-panel (min-width: 621px) { .bn-deploy-target { padding: 18px 24px 20px; @@ -11795,6 +11836,41 @@ html[data-ui-test="refined"] .bn-package-action-menu summary { overflow-wrap: anywhere; } +.bn-cloud-vla-form { + display: grid; + gap: 12px; +} + +.bn-cloud-vla-form > header { margin-bottom: 0; } +.bn-cloud-vla-form > label, +.bn-cloud-vla-form .bn-cloud-credit-grid > label { + display: grid; + gap: 5px; + color: var(--tx2); + font-size: 11px; +} + +.bn-cloud-vla-form input, +.bn-cloud-vla-form select { + min-width: 0; + width: 100%; + padding: 8px 9px; + color: var(--tx1); + background: var(--panel); + border: 1px solid var(--line2); + border-radius: 6px; +} + +.bn-cloud-vla-form .bn-cloud-credit-grid { + grid-template-columns: repeat(2, minmax(0, 1fr)); + margin-top: 0; +} + +.bn-cloud-vla-form > small { + color: var(--tx3); + line-height: 1.45; +} + .bn-cloud-artifact-list { display: grid; gap: 7px; diff --git a/editor/src/store.ts b/editor/src/store.ts index 2f0522c..7a06242 100644 --- a/editor/src/store.ts +++ b/editor/src/store.ts @@ -108,6 +108,7 @@ const ARTIFACT_KINDS = new Set([ 'blacknode.training-run', 'blacknode.action-chunking-checkpoint', 'blacknode.policy-artifact', + 'blacknode.vla-model', 'blacknode.policy-replay-metrics', 'blacknode.policy-runtime', ]) @@ -142,6 +143,15 @@ const ARTIFACT_CAPTURE_FIELDS = new Set([ 'error', 'name', 'policy_type', + 'model_id', + 'architecture', + 'provider', + 'base_model', + 'base_model_revision', + 'training_method', + 'checkpoint_sha256', + 'normalization', + 'inference', 'backend', 'created_at', 'source_checkpoint', diff --git a/python/blacknode/contracts.py b/python/blacknode/contracts.py index 4b7d0a6..8e93c2d 100644 --- a/python/blacknode/contracts.py +++ b/python/blacknode/contracts.py @@ -50,6 +50,7 @@ "blacknode.episode-dataset": "blacknode-dataset", "blacknode.episode-replay": "blacknode-dataset", "blacknode.dataset-catalog": "blacknode-dataset", + "blacknode.dataset-source": "blacknode-dataset", # Export / publishing "blacknode.hdf5-export": "blacknode-dataset", "blacknode.hub-dataset": "blacknode-dataset", @@ -62,6 +63,7 @@ "blacknode.action-chunking-checkpoint": "blacknode-training", "blacknode.act-policy-model": "blacknode-training", "blacknode.policy-artifact": "blacknode-training", + "blacknode.vla-model": "blacknode-training", "blacknode.policy-prediction": "blacknode-training", "blacknode.policy-preview": "blacknode-training", "blacknode.policy-replay": "blacknode-training", diff --git a/python/blacknode/package_index.py b/python/blacknode/package_index.py index de23f46..dbd32d3 100644 --- a/python/blacknode/package_index.py +++ b/python/blacknode/package_index.py @@ -904,10 +904,17 @@ "HuggingFaceDatasetUpload", "StreamPublisher" ] + }, + "adapters": { + "name": "adapters", + "default": False, + "node_types": [ + "LeRobotDataset" + ] } }, "git_url": "https://github.com/temiroff/blacknode-dataset.git", - "description": "Native episode recording, recovery, validation, LeRobot v3 export, and explicit Hugging Face dataset upload.", + "description": "BlacknodeDataset recording, recovery, validation, lazy LeRobot adaptation, export, and explicit Hugging Face upload.", "node_types": [ "BlacknodeHubExport", "DatasetBrowser", @@ -922,6 +929,7 @@ "HDF5EpisodeExport", "HuggingFaceDatasetUpload", "LeRobotV3Export", + "LeRobotDataset", "StreamPublisher", "TrajectorySmoother" ] @@ -986,10 +994,26 @@ } ] } + }, + "vla-openpi": { + "name": "vla-openpi", + "default": False, + "node_types": [ + "OpenPIFineTune" + ], + "dependencies": { + "requires": [ + { + "package": "blacknode-dataset", + "component": "adapters", + "version": ">=0.3,<1" + } + ] + } } }, "git_url": "https://github.com/temiroff/blacknode-training.git", - "description": "Robot-policy dataset checks, managed training, checkpoints, previews, reinforcement learning, and deployable policy artifacts.", + "description": "Managed robot-policy and OpenPI π0.5 VLA training, checkpoints, previews, and deployable model artifacts.", "node_types": [ "ACTCheckpointInspect", "ACTPolicyExport", @@ -1001,6 +1025,7 @@ "PPOPolicyExport", "PPOPolicyImport", "PPOTraining", + "OpenPIFineTune", "PolicyArtifactLoad", "TrainingDatasetCheck" ] diff --git a/python/blacknode/workflow.py b/python/blacknode/workflow.py index 59aff36..4091b43 100644 --- a/python/blacknode/workflow.py +++ b/python/blacknode/workflow.py @@ -346,6 +346,7 @@ def export_workflow_python(data: Mapping[str, Any], *, style: str = "flat") -> s f"# Entrypoint: {entry_name}.{port}", "# Keep the visual graph as the source of truth, then regenerate this file when the graph changes.", "", + "import atexit", "import os", "import signal", "import sys", @@ -437,6 +438,8 @@ def export_workflow_python(data: Mapping[str, Any], *, style: str = "flat") -> s " if meta.get('type') not in _BLACKNODE_LIVE_RUNTIME_NODE_TYPES:", " continue", " params = meta.get('params') or {}", + " if meta.get('type') == 'MapEnvironment' and str(params.get('action') or 'status').strip().lower() != 'start':", + " continue", " if str(params.get('action') or 'start').strip().lower() != 'stop':", " return True", " return False", @@ -451,6 +454,7 @@ def export_workflow_python(data: Mapping[str, Any], *, style: str = "flat") -> s " 'blacknode.pkg.blacknode_motion.arm.adapters.ros2.joint_motion',", " 'blacknode.pkg.blacknode_robot.robot',", " 'blacknode.pkg.blacknode_ros2.ros2_runtime',", + " 'blacknode.pkg.blacknode_perception.slam.adapters.ros2.mapping',", " 'blacknode.pkg.blacknode_perception.cv2_runtime',", " 'blacknode.pkg.blacknode_cuda.cuda_stream_runtime',", " 'blacknode.pkg.blacknode_cuda.viewer_runtime',", @@ -480,6 +484,7 @@ def export_workflow_python(data: Mapping[str, Any], *, style: str = "flat") -> s " for module_name, status_name in (", " ('blacknode.pkg.blacknode_perception.cv2_runtime', 'runtime_status'),", " ('blacknode.pkg.blacknode_ros2.ros2_runtime', 'runtime_status'),", + " ('blacknode.pkg.blacknode_perception.slam.adapters.ros2.mapping', 'runtime_status'),", " ('blacknode.pkg.blacknode_robot.robot', 'runtime_status'),", " ('blacknode.pkg.blacknode_skills.follow.leader_follower_runtime', 'leader_follower_runtime_status'),", " ):", @@ -512,6 +517,9 @@ def export_workflow_python(data: Mapping[str, Any], *, style: str = "flat") -> s " raise KeyboardInterrupt", "", "", + "atexit.register(_stop_blacknode_runtime_services)", + "", + "", "def _hold_live_runtime_if_needed() -> None:", " if not _workflow_uses_live_runtime():", " return", diff --git a/skills/blacknode-workflow/SKILL.md b/skills/blacknode-workflow/SKILL.md index 22df6a5..0ba9457 100644 --- a/skills/blacknode-workflow/SKILL.md +++ b/skills/blacknode-workflow/SKILL.md @@ -43,6 +43,25 @@ return here to integrate it into a validated workflow. 6. Use `PythonFn` only for workflow-local adapters. Use `blacknode-development` for reusable nodes or packages. +## Outcome-First Workflow Design + +- Begin with the requested outcome and build the shortest coherent graph that + produces it. +- A tracked template must perform a useful end-to-end task. Do not create + templates whose main result is proving a node works, showcasing wiring, + smoke testing, or confirming that a dependency is ready. +- Every visible node must materially create, transform, route, persist, + deploy, or operate something needed by the final result. +- Do not add checker, test, report, echo, or confirmation nodes as graph + padding. The node that owns an operation should validate inputs, preflight + dependencies, report progress, and return actionable errors when practical. +- Keep implementation validation in automated tests or untracked local + developer workflows rather than turning it into a product template. +- Preserve required physical-motion safety, authorization, cost consent, and + destructive-action confirmation. Integrate these controls into the owning + action or managed service when possible; keep a separate node when the + safety contract must be explicit and reusable. + ## Available Surfaces Preferred MCP stdio command: @@ -187,7 +206,7 @@ then return a concise graph plan with node ids, node types, key params, edges, entrypoint, and expected result. Build loop: -1. Understand the user goal and choose the smallest runnable graph. +1. Understand the user goal and choose the smallest outcome-producing graph. 2. Inspect list_nodes or get_node_schema before using unfamiliar nodes. 3. Create or load a workflow. 4. Add nodes with stable, descriptive ids. @@ -221,6 +240,8 @@ Use `list_nodes` for the live catalog. Current core groups: ## Graph Reliability Rules +- Make every node contribute directly to the requested result; never add a + node merely to prove, echo, check, or confirm another node. - Treat Blacknode workflows as DAGs. Do not create cycles or back-edges. - Always connect from `outputs` to `inputs`; never invent port names. - Respect types: `Any` accepts everything, exact type matches are valid, and diff --git a/tests/test_editor_cloud.py b/tests/test_editor_cloud.py index 408389e..0c43b33 100644 --- a/tests/test_editor_cloud.py +++ b/tests/test_editor_cloud.py @@ -299,6 +299,73 @@ def cloud_call(request, method, path, payload=None): self.assertEqual(payload["workflow"]["entrypoint"]["node_id"], "out") self.assertNotIn("image", payload["runtime"]) + def test_create_vla_job_builds_direct_two_node_workflow(self): + calls: list[tuple[str, str, dict]] = [] + + def cloud_call(request, method, path, payload=None): + self.assertIsNotNone(request.cookies.get(server._CLOUD_SESSION_COOKIE)) + calls.append((method, path, payload)) + return { + "id": "job_" + "a" * 32, + "status": "QUEUED", + "workload_kind": "vla_train", + } + + with patch.object(server, "_cloud_user_call", side_effect=cloud_call): + response = self.authenticated_client().post( + "/cloud/vla/jobs", + json={ + "dataset_uri": "hf://owner/dataset", + "dataset_revision": "a" * 40, + "steps": 5000, + "batch_size": 8, + "action_horizon": 10, + "max_runtime_seconds": 14400, + "project_ref": "robot-project", + }, + ) + + self.assertEqual(response.status_code, 200) + method, path, payload = calls[0] + self.assertEqual((method, path), ("POST", "/v1/jobs")) + workflow = payload["workflow"] + self.assertEqual(workflow["entrypoint"], {"node_id": "train", "port": "model"}) + self.assertEqual( + {node["type"] for node in workflow["node_meta"].values()}, + {"LeRobotDataset", "OpenPIFineTune"}, + ) + self.assertEqual(len(workflow["edges"]), 1) + self.assertEqual(workflow["node_meta"]["train"]["params"]["action"], "run") + self.assertNotIn("Output", {node["type"] for node in workflow["node_meta"].values()}) + + def test_dataset_upload_streams_verified_archive_to_cloud(self): + received: dict[str, object] = {} + + def upload(path, stream, **kwargs): + received.update(path=path, content=stream.read(), **kwargs) + return { + "id": "dataset_" + "a" * 32, + "kind": "blacknode.cloud-dataset", + "name": "robot.tar.gz", + "size_bytes": 7, + "sha256": "a" * 64, + "media_type": "application/gzip", + "locator": "/v1/datasets/dataset_" + "a" * 32 + "/download", + } + + with patch.object(cloud_client, "upload", side_effect=upload): + response = self.authenticated_client().put( + "/cloud/datasets", + headers={"X-Dataset-Name": "robot.tar.gz"}, + content=b"archive", + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(received["content"], b"archive") + self.assertEqual(received["size"], 7) + self.assertEqual(received["authorization"], "registered-user-cloud-token-0123456789") + self.assertTrue(str(received["path"]).startswith("/v1/datasets/dataset_")) + def test_invalid_job_id_never_reaches_cloud(self): with patch.object(server, "_cloud_user_call") as cloud_call: response = self.authenticated_client().get("/cloud/jobs/../../secrets") diff --git a/tests/test_editor_devices.py b/tests/test_editor_devices.py index 6079131..d36e4c0 100644 --- a/tests/test_editor_devices.py +++ b/tests/test_editor_devices.py @@ -77,6 +77,7 @@ def __init__( self.runtime_workflows: dict[str, dict] = {} self.runtime_telemetry: dict[str, dict] = {} self.runtime_services: dict[str, dict] = {} + self.runtime_ros2_topics: dict[str, dict] = {} self.runtime_packages = [ {"name": "blacknode-runtime", "version": "0.2.0"}, ] @@ -92,8 +93,10 @@ def __init__( "deployment_ownership_v1", "deployment_workflow_v1", "deployment_motion_control_v1", + "deployment_mapping_control_v1", "ros2_diagnostics_v1", "managed_ros2_services_v1", + "remote_ros2_topic_stream_v1", ] self.runtime_version = runtime_version self.ros2_diagnostics_payload = { @@ -198,6 +201,41 @@ def __call__(self, request, timeout=0): }) if path == "/diagnostics/ros2": return _JsonResponse(self.ros2_diagnostics_payload) + if path.startswith("/ros2/topics/"): + parts = path.strip("/").split("/") + stream_id = parts[2] + action = parts[3] if len(parts) > 3 else "" + if request.method == "GET" and not action: + return _JsonResponse(self.runtime_ros2_topics.get(stream_id, { + "id": stream_id, + "outputs": { + "message": {}, + "status": {"worker_alive": False, "source_fresh": False}, + "report": "topic stream has not been started on this device", + }, + })) + if action == "start": + result = { + "id": stream_id, + "outputs": { + "message": { + "info": {"width": 2, "height": 2, "resolution": 0.05}, + "data": [-1, 0, 75, 100], + }, + "status": { + "worker_alive": True, + "source_fresh": True, + "received": 1, + }, + "report": f"ROS2 streaming {body['topic']}", + }, + } + self.runtime_ros2_topics[stream_id] = result + return _JsonResponse(result) + if action == "stop": + self.runtime_ros2_topics.pop(stream_id, None) + return _JsonResponse({"id": stream_id, "outputs": {"status": {"worker_alive": False}}}) + raise AssertionError(f"Unexpected fake ROS 2 topic action: {action}") if path.startswith("/services/"): parts = path.strip("/").split("/") service_id = parts[1] @@ -299,6 +337,14 @@ def __call__(self, request, timeout=0): "motion_control_count": len( body.get("manifest", {}).get("motion_controls") or [] ), + "mapping_control_count": len( + body.get("manifest", {}).get("mapping_controls") or [] + ), + "mapping_topic": next(( + str(item.get("map_topic") or "/map") + for item in body.get("manifest", {}).get("mapping_controls") or [] + ), ""), + "last_map_artifact": {}, "created_at": "2026-07-23T00:00:00+00:00", "updated_at": "2026-07-23T00:00:01+00:00", } @@ -342,6 +388,20 @@ def __call__(self, request, timeout=0): elif action == "rollback": record.update(state="staged", pid=None) elif action == "control": + if body.get("command") == "save-map": + artifact = { + "kind": "blacknode.map-artifact", + "map_name": "map_01", + "map_yaml": "/home/ubuntu/Blacknode/maps/map_01.yaml", + } + record["last_map_artifact"] = artifact + return _JsonResponse({ + "ok": True, + "id": deployment_id, + "artifact": artifact, + "warning": "", + "deployment": record, + }) armed = body.get("command") == "arm" record["motion_armed"] = armed return _JsonResponse({ @@ -7403,6 +7463,54 @@ def test_remote_deployment_controls_proxy_with_saved_token(self): for _method, _path, authorization, _body in remote_requests )) + def test_running_mapping_deployment_streams_and_saves_occupancy_map(self): + hardware = _HardwareService() + deployment_id = "rosorin-map" + hardware.runtime_deployments[deployment_id] = { + "id": deployment_id, + "name": "ROSOrin Map Environment", + "state": "running", + "staged_revision": "cafebabecafebabe", + "active_revision": "cafebabecafebabe", + "revisions": ["cafebabecafebabe"], + "pid": 4321, + "exit_code": None, + "error": "", + "mapping_control_count": 1, + "mapping_topic": "/map", + "last_map_artifact": {}, + "created_at": "2026-08-10T00:00:00+00:00", + "updated_at": "2026-08-10T00:00:01+00:00", + } + with patch("device_registry.urllib.request.urlopen", side_effect=hardware): + device_id = self.client.post("/devices", json={ + "name": "ROSOrin", + "base_url": "http://192.168.1.87:8765", + "token": hardware.token, + }).json()["device"]["id"] + hardware.runtime_deployments[deployment_id]["target_device_id"] = device_id + snapshot = self.client.get( + f"/devices/{device_id}/deployments/{deployment_id}/mapping/snapshot", + ) + saved = self.client.post( + f"/devices/{device_id}/deployments/{deployment_id}/mapping/save", + ) + stopped = self.client.post( + f"/devices/{device_id}/deployments/{deployment_id}/stop", + ) + + self.assertEqual(snapshot.status_code, 200) + self.assertEqual(snapshot.json()["message"]["info"]["width"], 2) + self.assertEqual(snapshot.json()["message"]["data"], [-1, 0, 75, 100]) + self.assertEqual(saved.status_code, 200) + self.assertEqual(saved.json()["artifact"]["map_name"], "map_01") + self.assertEqual(stopped.json()["state"], "stopped") + self.assertEqual(hardware.runtime_ros2_topics, {}) + self.assertIn( + ("POST", f"/deployments/{deployment_id}/control", f"Bearer {hardware.token}", {"command": "save-map"}), + hardware.requests, + ) + def test_remote_deployments_are_scoped_to_the_selected_robot(self): hardware = _HardwareService() with patch("device_registry.urllib.request.urlopen", side_effect=hardware): diff --git a/tests/test_editor_projects.py b/tests/test_editor_projects.py index ea11aa0..772e04a 100644 --- a/tests/test_editor_projects.py +++ b/tests/test_editor_projects.py @@ -493,6 +493,31 @@ def test_existing_manifest_can_be_added_and_unlinked_without_deletion(self): "Unlinking must not delete the provider-owned artifact reference", ) + def test_vla_model_manifest_is_indexed_as_a_trained_model(self): + model_path = Path(self._tmp.name) / "models" / "pi05-lora" + model_path.mkdir(parents=True) + (model_path / "manifest.json").write_text(json.dumps({ + "kind": "blacknode.vla-model", + "schema_version": 1, + "model_id": "vla-0123456789abcdef", + "provider": "openpi", + "architecture": "pi05", + "backend": "jax", + "training_method": "lora", + "checkpoint": "adapter-checkpoint.tar.gz", + "checkpoint_sha256": "a" * 64, + "physical_motion_authorized": False, + }), encoding="utf-8") + + artifacts = server._artifact_store.inspect_path(model_path) + + self.assertEqual(len(artifacts), 1) + artifact = artifacts[0] + self.assertEqual(artifact["artifact_type"], "model") + self.assertEqual(artifact["provider"], "blacknode-training") + self.assertEqual(artifact["metadata"]["architecture"], "pi05") + self.assertFalse(artifact["metadata"]["physical_motion_authorized"]) + def test_native_dataset_manifest_and_run_outputs_map_to_typed_evidence(self): self._workflow( "learning", diff --git a/tests/test_package_index.py b/tests/test_package_index.py index 1de2c02..d8e63f4 100644 --- a/tests/test_package_index.py +++ b/tests/test_package_index.py @@ -260,6 +260,7 @@ def test_core_index_maps_official_node_types_to_git_packages(): assert payload["nodes"]["DatasetBrowser"]["package"] == "blacknode-dataset" assert payload["nodes"]["HDF5EpisodeExport"]["package"] == "blacknode-dataset" assert payload["nodes"]["StreamPublisher"]["package"] == "blacknode-dataset" + assert payload["nodes"]["LeRobotDataset"]["package"] == "blacknode-dataset" assert payload["nodes"]["ROS2LeaderFollower"]["package"] == "blacknode-skills" assert payload["nodes"]["ROS2PublishJointState"]["package"] == "blacknode-skills" assert payload["nodes"]["ROS2SubscribeJointState"]["package"] == "blacknode-skills" @@ -292,6 +293,16 @@ def test_core_index_maps_official_node_types_to_git_packages(): assert payload["nodes"]["ACTPolicyReplay"]["package"] == "blacknode-training" assert payload["nodes"]["PPOTraining"]["package"] == "blacknode-training" assert payload["nodes"]["PPOPolicyImport"]["package"] == "blacknode-training" + assert payload["nodes"]["OpenPIFineTune"]["package"] == "blacknode-training" + assert payload["packages"]["blacknode-training"]["components"]["vla-openpi"]["dependencies"] == { + "requires": [ + { + "package": "blacknode-dataset", + "component": "adapters", + "version": ">=0.3,<1", + } + ] + } assert payload["packages"]["blacknode-training"]["components"]["reinforcement-learning"]["dependencies"] == { "requires": [ { diff --git a/tests/test_python_roundtrip.py b/tests/test_python_roundtrip.py index 4881d1b..e0b672c 100644 --- a/tests/test_python_roundtrip.py +++ b/tests/test_python_roundtrip.py @@ -159,6 +159,12 @@ def test_live_stream_export_waits_for_ctrl_c_and_stops_runtime(self): script, ) self.assertIn("signal.signal(signal.SIGTERM, _raise_blacknode_stop)", script) + self.assertIn("atexit.register(_stop_blacknode_runtime_services)", script) + self.assertIn("meta.get('type') == 'MapEnvironment'", script) + self.assertIn( + "blacknode.pkg.blacknode_perception.slam.adapters.ros2.mapping", + script, + ) self.assertIn("_hold_live_runtime_if_needed()", script) def test_class_export_is_runnable_and_importable(self):