diff --git a/README.md b/README.md index c63fcef..f2ebd34 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,11 @@ Workflows stay readable as systems grow: each node has typed ports, each connection is validated, and hardware-specific implementations remain behind stable robot capabilities. +Robot Hardware can fan timestamped capability telemetry to optional transports. +Its MQTT adapter publishes versioned robot-state streams for dashboards, fleet +systems, and third-party consumers while local hardware control remains +authenticated and independently supervised. + ## A Robot Workflow, Layer by Layer A Blacknode robot workflow moves from task intent to physical hardware through @@ -148,14 +153,66 @@ disarmed; previous deployments stay stopped until explicitly started again. Each lifecycle action shows stage progress. Each robot also has its own **Pause robot** and **Resume robot** controls. Robot Pause still sends the direct hardware stop when its deployment runtime is unavailable and reports -that runtime address as a warning. The robot detail reports physical torque -only when the hardware provider exposes it; otherwise it shows **Not reported** -instead of inferring torque state from the motion-armed flag. +that runtime address as a warning. The robot detail reports physical torque as +**On**, **Off**, or **Unknown**. Unknown includes the provider's readback reason +when available and is treated as possibly on; select the **Physical torque** +fact card to open or close that explanation. Blacknode never infers physical +torque from the separate motion-armed flag. The last verified Robot Hardware +version is retained in the paired-device record, so an active workflow or +temporary version-report outage does not erase it from the robot card. For robots on SSH-managed computers, **Restart robot service** asks for the SSH password, resolves the exact hardware systemd unit from that robot's saved hardware port, and restarts only that unit. Restart is blocked while a deployment is active or the robot is armed. Blacknode then verifies that the same authenticated robot returned and keeps motion disarmed. +**Check device** checks the Runtime and every attached Robot Hardware service +together. SSH-managed computers also expose one **Runtime + Robot Hardware** +panel. +Its read-only version check compares installed and latest upstream versions +and commits for both repositories, includes the version reported by each live +service, and states whether each service is current or has an update. **Update +Runtime + Robot Hardware** stops active deployments, stops and disarms attached +robots, verifies trusted clean Git checkouts, fast-forwards them, reinstalls the +packages in their existing environments, and restarts only the matched +services. The report also offers **Update Runtime** and **Update Robot +Hardware** independently, so a valid Runtime update remains available when a +Robot Hardware service needs attention. Runtime has its own installation row. +Robot Hardware has one shared installation row because every robot service on +that compute device uses the same checkout and environment; a compact status +line identifies the robot services that were verified. **Restart Robot +Hardware** remains a per-robot action. +Unknown versions and service-resolution errors expose **Repair Runtime** or +**Repair Robot Hardware**. When an authenticated Robot Hardware process was +started manually from the trusted checkout, Repair Robot Hardware verifies and stops only +that listener, installs the saved persistent systemd services, and continues +the update. If an installed unit has drifted to a different HTTP port, Repair +Hardware matches its saved configuration by stable robot identity, reconciles +the service manifest to the editor's paired port, and reinstalls the unit. +An unidentified port owner is never stopped automatically. Local source +changes remain blocked. When every selected commit is current, the +matching control becomes **Reinstall +Runtime + Robot Hardware** and repairs the current package environments before +restarting and verifying the services. Local source changes block either +operation and are never overwritten. +Updating does not switch off physical actuator power; a torque warning remains +visible when the updated hardware service reads an enabled servo torque +register. When no deployment is running, the robot card offers an explicitly +confirmed **Release torque** action. It writes only torque-disable values and +verifies the reported physical state afterward; support the robot first because +its joints may move or drop. +Each robot card also exposes **Monitor live**. Select the robot to view its +current joint positions, derived joint speeds, torque state, stream freshness, +and recent per-joint traces. The monitor reads Robot Hardware while the robot +is idle and automatically switches to the running deployment when that process +owns the serial bus. MQTT is optional: it remains available for external +dashboards and fleet consumers, while the editor monitor uses the authenticated +device connection directly. Deployed monitoring requires Runtime 0.3.9 or +newer and `blacknode-drivers` 0.2.1 or newer; restart a running deployment after +updating so it receives the new telemetry channel. +Devices initially paired manually can use **Enable SSH controls** later. +Blacknode confirms the SSH host key and enables management only when the +installed systemd runtime matches the pairing's exact port and runtime device +identity. The SSH password remains request-only. The installer detects each computer's current address; it does not hardcode a LAN IP. Use a router DHCP reservation or a resolvable hostname for robots that @@ -203,6 +260,10 @@ cannot be deployed accidentally. Remote deployments appear below the preflight report. Expand one to view its device log and use **Run**, **Stop**, **Stage update**, or **Rollback**. +The robot card also distinguishes a running deployment from a stopped, staged, +exited, or failed deployment that remains stored on its Runtime. A stored +deployment exposes **Restart deployment** and **Deployment details**, so a +Runtime or Hardware reinstall does not require sending the same workflow again. **Stage update** uploads the currently validated graph as another revision of that deployment. **Rollback** selects the previous revision without starting it; use **Run** after checking its state. A running deployment must be stopped diff --git a/docs/project-lifecycle.md b/docs/project-lifecycle.md index 1c1d71a..1d71cc2 100644 --- a/docs/project-lifecycle.md +++ b/docs/project-lifecycle.md @@ -37,7 +37,12 @@ setup request and is not saved. For manual setup, install `blacknode-runtime` on the device, run `./service.sh pairing`, and enter the printed runtime URL and token under -**Add device โ†’ Pair manually**. +**Add device โ†’ Pair manually**. You can add SSH management later from that +device's details with **Enable SSH controls**. Blacknode confirms the SSH host +key, verifies that the installed systemd runtime has the same port and runtime +device identity as the existing pairing, and then enables device lifecycle and +robot-service restart actions. The SSH password is used for that request and is +not saved. Open the new square device card and choose **Add robot** for each hardware service connected to that computer. Enter the robot URL and token printed by @@ -87,6 +92,9 @@ The Project shows the running deployment and its owning device. Use **Deployments** to inspect logs, stop, update, or roll back a revision. Stops remain explicit because stopping a hardware workflow may release actuator torque. +The robot card reports stopped and staged deployments that remain stored on the +Runtime and can restart the latest stored deployment after the robot passes the +same connected, disarmed, calibration, and ownership checks. Update the graph, check setup again, and send a new revision to iterate. Project artifacts retain evidence from datasets, training runs, policies, evaluations, diff --git a/editor-server/device_installer.py b/editor-server/device_installer.py index 4a6f0b0..6a78f8e 100644 --- a/editor-server/device_installer.py +++ b/editor-server/device_installer.py @@ -19,6 +19,7 @@ class DeviceInstallError(RuntimeError): _INSTANCE_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,31}$") _INSPECTION_MARKER = "__BLACKNODE_RUNTIME_INSPECTION__=" +_UPDATE_REPORT_MARKER = "__BLACKNODE_UPDATE_REPORT__=" _INSPECTION_SCRIPT = r"""python3 - <<'PY' import glob import json @@ -1038,6 +1039,954 @@ def remote_output(line: str) -> None: connection.close() +def update_managed_services( + *, + host: str, + port: int, + username: str, + password: str, + host_fingerprint: str, + instance_id: str, + runtime_port: int, + hardware_ports: list[int], + hardware_device_ids: dict[int, str] | None = None, + include_runtime: bool = True, + progress: Callable[[dict[str, Any]], None] | None = None, +) -> dict[str, Any]: + """Fast-forward and restart selected managed Runtime and Hardware services.""" + + def report(percent: int, message: str) -> None: + if progress is not None: + progress({ + "progress": max(0, min(100, int(percent))), + "message": str(message), + }) + + selected_instance = _clean_instance_id(instance_id or "default") + selected_runtime_port = int(runtime_port) + if selected_runtime_port < 1 or selected_runtime_port > 65535: + raise DeviceInstallError("The managed runtime port is invalid.") + selected_hardware_ports = sorted({int(value) for value in hardware_ports}) + selected_hardware_targets = [ + { + "port": value, + "device_id": str((hardware_device_ids or {}).get(value) or ""), + } + for value in selected_hardware_ports + ] + if any(value < 1 or value > 65535 for value in selected_hardware_ports): + raise DeviceInstallError("A robot hardware port is invalid.") + if not include_runtime and not selected_hardware_ports: + raise DeviceInstallError("Choose Runtime, Hardware, or both to update.") + + report(3, "Connecting to the verified device") + connection = _connect( + host, + port, + username, + password, + expected_fingerprint=host_fingerprint, + timeout=15.0, + ) + remote_script_path = f"/tmp/blacknode-update-{secrets.token_hex(8)}.sh" + script = r"""#!/usr/bin/env bash +set -euo pipefail +progress() { + echo "__BLACKNODE_UPDATE_PROGRESS__=$1|$2" +} +sudo() { + command sudo -S -p '' "$@" +} +export -f sudo + +include_runtime="$1" +instance="$2" +runtime_port="$3" +hardware_targets_payload="$4" +hardware_ports=() +hardware_device_ids=() +while IFS=$'\t' read -r target_port target_device_id; do + [[ -n "$target_port" ]] || continue + hardware_ports+=("$target_port") + hardware_device_ids+=("$target_device_id") +done < <( + python3 - "$hardware_targets_payload" <<'PY' +import base64 +import json +import sys + +targets = json.loads(base64.urlsafe_b64decode(sys.argv[1]).decode("utf-8")) +for target in targets: + print(f"{int(target['port'])}\t{str(target.get('device_id') or '')}") +PY +) +if [[ "$instance" == "default" ]]; then + runtime_dir="$HOME/blacknode-runtime" + runtime_service="blacknode-runtime.service" +else + [[ "$instance" =~ ^[a-z0-9][a-z0-9-]{0,31}$ ]] || { + echo "Invalid Blacknode runtime instance." + exit 2 + } + runtime_dir="$HOME/blacknode-runtimes/$instance" + runtime_service="blacknode-runtime-$instance.service" +fi + +package_version() { + python3 - "$1" <<'PY' +import sys +import tomllib +from pathlib import Path + +path = Path(sys.argv[1]) / "pyproject.toml" +try: + payload = tomllib.loads(path.read_text(encoding="utf-8")) + print(str((payload.get("project") or {}).get("version") or "unknown")) +except Exception: + print("unknown") +PY +} + +validate_checkout() { + local directory="$1" + local repository="$2" + [[ -d "$directory/.git" && -f "$directory/pyproject.toml" ]] || { + echo "$repository is not an updateable Git checkout: $directory" + exit 3 + } + local origin + origin="$(git -C "$directory" remote get-url origin 2>/dev/null || true)" + python3 - "$origin" "$repository" <<'PY' +import re +import sys + +origin = sys.argv[1].strip().lower().removesuffix(".git").rstrip("/") +repository = sys.argv[2].strip().lower() +if not re.search(rf"(?:github\.com[:/])temiroff/{re.escape(repository)}$", origin): + raise SystemExit( + f"Refusing to update {repository}: its origin is not the trusted " + f"temiroff/{repository} repository." + ) +PY + if [[ -n "$(git -C "$directory" status --porcelain --untracked-files=normal)" ]]; then + echo "$repository has local changes. Commit, stash, or remove them before remote update." + exit 4 + fi + git -C "$directory" rev-parse --verify '@{upstream}' >/dev/null 2>&1 || { + echo "$repository does not have an upstream branch configured." + exit 5 + } +} + +resolve_hardware_unit() { + local hardware_port="$1" + local matches=() + local active_matches=() + local enabled_matches=() + local unit exec_start unit_definition normalized unit_path + mapfile -t candidate_units < <( + { + systemctl list-unit-files 'blacknode-hardware*.service' --no-legend 2>/dev/null + systemctl list-units --all --type=service 'blacknode-hardware*.service' \ + --no-legend 2>/dev/null + find /etc/systemd/system /run/systemd/system -maxdepth 1 \ + -name 'blacknode-hardware*.service' -printf '%f\n' 2>/dev/null + } | awk '{print $1}' | sort -u + ) + for unit in "${candidate_units[@]}"; do + [[ "$unit" =~ ^blacknode-hardware([-.@][A-Za-z0-9_.@-]+)?\.service$ ]] || continue + exec_start="$(systemctl show "$unit" --property=ExecStart --value 2>/dev/null || true)" + unit_definition="" + for unit_path in "/etc/systemd/system/$unit" "/run/systemd/system/$unit"; do + if [[ -f "$unit_path" ]]; then + unit_definition+="$(command cat "$unit_path")" + unit_definition+=$'\n' + fi + done + normalized="${exec_start//\"/}"$'\n'"${unit_definition//\"/}" + if printf '%s\n' "$normalized" \ + | grep -oE -- '--port(=|[[:space:]])[0-9]+' \ + | grep -oE '[0-9]+' \ + | grep -Fxq -- "$hardware_port"; then + matches+=("$unit") + systemctl is-active --quiet "$unit" 2>/dev/null \ + && active_matches+=("$unit") + systemctl is-enabled --quiet "$unit" 2>/dev/null \ + && enabled_matches+=("$unit") + fi + done + if [[ "${#matches[@]}" -eq 0 ]]; then + echo "No installed or active Blacknode Hardware systemd service matches port $hardware_port. Discovered units: ${candidate_units[*]:-none}." >&2 + return 60 + fi + if [[ "${#matches[@]}" -eq 1 ]]; then + printf '%s\n' "${matches[0]}" + return 0 + fi + if [[ "${#active_matches[@]}" -eq 1 ]]; then + printf '%s\n' "${active_matches[0]}" + return 0 + fi + if [[ "${#active_matches[@]}" -gt 1 ]]; then + echo "Multiple active Blacknode Hardware services claim port $hardware_port: ${active_matches[*]}." >&2 + return 61 + fi + if [[ "${#enabled_matches[@]}" -eq 1 ]]; then + printf '%s\n' "${enabled_matches[0]}" + return 0 + fi + echo "Could not choose the persistent Hardware service for port $hardware_port. Candidates: ${matches[*]}. Enabled: ${enabled_matches[*]:-none}." >&2 + return 61 +} + +stop_verified_manual_hardware() { + local hardware_repo="$1" + local hardware_port="$2" + local listener_pids=() + local pid cmdline + if command -v ss >/dev/null 2>&1; then + mapfile -t listener_pids < <( + ss -H -ltnp "sport = :$hardware_port" 2>/dev/null \ + | grep -oE 'pid=[0-9]+' | cut -d= -f2 | sort -u + ) + elif command -v lsof >/dev/null 2>&1; then + mapfile -t listener_pids < <( + lsof -nP -t -iTCP:"$hardware_port" -sTCP:LISTEN 2>/dev/null | sort -u + ) + fi + if [[ "${#listener_pids[@]}" -eq 0 ]]; then + if (echo >/dev/tcp/127.0.0.1/"$hardware_port") >/dev/null 2>&1; then + echo "Port $hardware_port is owned by a process that SSH cannot identify. Stop the manually running Hardware process and retry Repair Hardware." >&2 + return 1 + fi + return 0 + fi + for pid in "${listener_pids[@]}"; do + [[ "$pid" =~ ^[0-9]+$ && -r "/proc/$pid/cmdline" ]] || { + echo "Could not verify the process listening on Hardware port $hardware_port." >&2 + return 1 + } + cmdline="$(tr '\0' ' ' < "/proc/$pid/cmdline")" + if [[ "$cmdline" != *"$hardware_repo/scripts/hardware_service.py"* ]] \ + || [[ ! "$cmdline" =~ --port(=|[[:space:]])${hardware_port}([^0-9]|$) ]]; then + echo "Refusing to stop unverified process $pid on Hardware port $hardware_port: $cmdline" >&2 + return 1 + fi + done + for pid in "${listener_pids[@]}"; do + echo "Stopping verified manually started Hardware process $pid on port $hardware_port." + kill -TERM "$pid" + done + for _attempt in {1..40}; do + listening=false + for pid in "${listener_pids[@]}"; do + kill -0 "$pid" >/dev/null 2>&1 && listening=true + done + [[ "$listening" == false ]] && return 0 + sleep 0.25 + done + echo "The manually running Hardware process on port $hardware_port did not stop. Stop it from its terminal and retry Repair Hardware." >&2 + return 1 +} + +install_persistent_hardware_services() { + local hardware_repo="$HOME/blacknode-hardware" + [[ -x "$hardware_repo/install-service.sh" ]] || { + echo "Repair Hardware requires $hardware_repo/install-service.sh." >&2 + return 1 + } + validate_checkout "$hardware_repo" "blacknode-hardware" + if [[ -f "$hardware_repo/.blacknode-hardware/devices.json" ]]; then + python3 - "$hardware_repo/.blacknode-hardware/devices.json" \ + "$hardware_targets_payload" <<'PY' +import base64 +import json +import os +import sys +from pathlib import Path + +path = Path(sys.argv[1]) +targets = json.loads(base64.urlsafe_b64decode(sys.argv[2]).decode("utf-8")) +payload = json.loads(path.read_text(encoding="utf-8")) +devices = payload.get("devices") or [] +by_id = { + str(item.get("device_id") or ""): item + for item in devices + if isinstance(item, dict) and item.get("device_id") +} +for target in targets: + device_id = str(target.get("device_id") or "") + if not device_id: + raise SystemExit( + f"Cannot reconcile Hardware port {target['port']}: " + "the paired robot has no stable device identity." + ) + device = by_id.get(device_id) + if device is None: + raise SystemExit( + f"Cannot reconcile Hardware port {target['port']}: device identity " + f"{device_id!r} is not present in {path}." + ) + device["service_port"] = int(target["port"]) +ports = [int(item["service_port"]) for item in devices] +if len(ports) != len(set(ports)): + raise SystemExit( + "Cannot reconcile Hardware services because two configured robots " + "would use the same HTTP port." + ) +temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") +temporary.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") +os.replace(temporary, path) +PY + fi + for hardware_port in "${hardware_ports[@]}"; do + stop_verified_manual_hardware "$hardware_repo" "$hardware_port" + done + progress 11 "Installing missing persistent Hardware services" + if [[ -f "$hardware_repo/.blacknode-hardware/devices.json" ]]; then + ( + cd "$hardware_repo" + ./install-service.sh --all + ) + elif [[ "${#hardware_ports[@]}" -eq 1 ]]; then + ( + cd "$hardware_repo" + BLACKNODE_HARDWARE_PORT="${hardware_ports[0]}" ./install-service.sh + ) + else + echo "Multiple Hardware ports require a saved multi-robot configuration. Run ./configure.sh --all before Repair Hardware." >&2 + return 1 + fi +} + +progress 8 "Resolving selected managed services" +if [[ "$include_runtime" == "1" ]]; then + [[ -d "$runtime_dir" ]] || { + echo "The managed runtime directory is missing: $runtime_dir" + exit 3 + } +fi +repair_hardware=false +for hardware_port in "${hardware_ports[@]}"; do + if unit="$(resolve_hardware_unit "$hardware_port")"; then + : + else + resolve_status=$? + if [[ "$resolve_status" -eq 60 ]]; then + repair_hardware=true + else + exit 6 + fi + fi +done +if [[ "$repair_hardware" == true ]]; then + install_persistent_hardware_services || exit 6 +fi +hardware_units=() +hardware_dirs=() +for hardware_port in "${hardware_ports[@]}"; do + [[ "$hardware_port" =~ ^[0-9]{1,5}$ ]] || exit 2 + if ! unit="$(resolve_hardware_unit "$hardware_port")"; then + echo "Repair Hardware did not create exactly one persistent service for port $hardware_port." >&2 + exit 6 + fi + directory="$(systemctl show "$unit" --property=WorkingDirectory --value 2>/dev/null || true)" + if [[ -z "$directory" ]]; then + for unit_path in "/etc/systemd/system/$unit" "/run/systemd/system/$unit"; do + if [[ -f "$unit_path" ]]; then + directory="$( + sed -nE 's/^WorkingDirectory=(.*)$/\1/p' "$unit_path" | tail -n 1 + )" + [[ -n "$directory" ]] && break + fi + done + fi + [[ -n "$directory" ]] || { + echo "$unit does not report its repository WorkingDirectory." >&2 + exit 6 + } + hardware_units+=("$unit") + hardware_dirs+=("$directory") +done + +progress 15 "Checking trusted clean Git checkouts" +if [[ "$include_runtime" == "1" ]]; then + validate_checkout "$runtime_dir" "blacknode-runtime" +fi +unique_hardware_dirs=() +for directory in "${hardware_dirs[@]}"; do + found=false + for existing in "${unique_hardware_dirs[@]}"; do + [[ "$existing" == "$directory" ]] && found=true + done + if [[ "$found" == false ]]; then + validate_checkout "$directory" "blacknode-hardware" + unique_hardware_dirs+=("$directory") + fi +done + +runtime_before_version="" +runtime_before_commit="" +if [[ "$include_runtime" == "1" ]]; then + runtime_before_version="$(package_version "$runtime_dir")" + runtime_before_commit="$(git -C "$runtime_dir" rev-parse --short=12 HEAD)" +fi +hardware_before_versions=() +hardware_before_commits=() +for directory in "${hardware_dirs[@]}"; do + hardware_before_versions+=("$(package_version "$directory")") + hardware_before_commits+=("$(git -C "$directory" rev-parse --short=12 HEAD)") +done + +if [[ "$include_runtime" == "1" ]]; then + progress 25 "Downloading Runtime updates" + git -C "$runtime_dir" fetch --prune +fi +for directory in "${unique_hardware_dirs[@]}"; do + progress 35 "Downloading robot hardware updates" + git -C "$directory" fetch --prune +done + +services=() +if [[ "$include_runtime" == "1" ]]; then + services+=("$runtime_service") +fi +services+=("${hardware_units[@]}") +restore_services() { + local unit + for unit in "${services[@]}"; do + sudo systemctl start "$unit" >/dev/null 2>&1 || true + done +} +trap restore_services EXIT + +progress 45 "Stopping managed services" +for unit in "${hardware_units[@]}"; do + sudo systemctl stop "$unit" +done +if [[ "$include_runtime" == "1" ]]; then + sudo systemctl stop "$runtime_service" +fi + +if [[ "$include_runtime" == "1" ]]; then + progress 55 "Fast-forwarding Blacknode Runtime" + git -C "$runtime_dir" merge --ff-only '@{upstream}' + "$runtime_dir/.venv/bin/python" -m pip install -e "$runtime_dir" +fi + +for directory in "${unique_hardware_dirs[@]}"; do + progress 68 "Fast-forwarding Blacknode Hardware" + git -C "$directory" merge --ff-only '@{upstream}' + "$directory/.venv/bin/python" -m pip install -e "$directory" +done + +progress 80 "Starting the selected updated services" +if [[ "$include_runtime" == "1" ]]; then + sudo systemctl start "$runtime_service" +fi +for unit in "${hardware_units[@]}"; do + sudo systemctl start "$unit" +done + +for unit in "${services[@]}"; do + state="" + for _attempt in {1..60}; do + state="$(systemctl is-active "$unit" 2>/dev/null || true)" + [[ "$state" == "active" ]] && break + sleep 0.25 + done + [[ "$state" == "active" ]] || { + echo "$unit did not return to the active state." + exit 7 + } +done +trap - EXIT + +progress 90 "Collecting installed versions" +report_file="$(mktemp)" +trap 'rm -f -- "$report_file"' EXIT +if [[ "$include_runtime" == "1" ]]; then + runtime_after_version="$(package_version "$runtime_dir")" + runtime_after_commit="$(git -C "$runtime_dir" rev-parse --short=12 HEAD)" + printf 'runtime\t%s\t%s\t%s\t%s\t%s\t%s\n' \ + "$runtime_service" "$runtime_port" "$runtime_before_version" "$runtime_after_version" \ + "$runtime_before_commit" "$runtime_after_commit" >> "$report_file" +fi +for index in "${!hardware_units[@]}"; do + directory="${hardware_dirs[$index]}" + printf 'hardware\t%s\t%s\t%s\t%s\t%s\t%s\n' \ + "${hardware_units[$index]}" "${hardware_ports[$index]}" \ + "${hardware_before_versions[$index]}" "$(package_version "$directory")" \ + "${hardware_before_commits[$index]}" "$(git -C "$directory" rev-parse --short=12 HEAD)" \ + >> "$report_file" +done +python3 - "$report_file" <<'PY' +import json +import sys +from pathlib import Path + +components = [] +for line in Path(sys.argv[1]).read_text(encoding="utf-8").splitlines(): + kind, service, port, before_version, after_version, before_commit, after_commit = line.split("\t") + components.append({ + "kind": kind, + "service_name": service, + "port": int(port), + "before": {"version": before_version, "commit": before_commit}, + "after": {"version": after_version, "commit": after_commit}, + "changed": before_commit != after_commit, + "state": "active", + }) +print("__BLACKNODE_UPDATE_REPORT__=" + json.dumps({ + "ok": True, + "components": components, +}, separators=(",", ":"))) +PY +progress 96 "Managed services updated" +""" + try: + sftp = connection.client.open_sftp() + try: + with sftp.file(remote_script_path, "w") as remote_script: + remote_script.write(script) + sftp.chmod(remote_script_path, 0o700) + finally: + sftp.close() + + def remote_output(line: str) -> None: + match = re.match( + r"^__BLACKNODE_UPDATE_PROGRESS__=(\d{1,3})\|(.*)$", + line.strip(), + ) + if match: + report(int(match.group(1)), match.group(2).strip()) + + targets_payload = base64.urlsafe_b64encode(json.dumps( + selected_hardware_targets, + separators=(",", ":"), + ).encode("utf-8")).decode("ascii") + arguments = " ".join([ + "1" if include_runtime else "0", + selected_instance, + str(selected_runtime_port), + targets_payload, + ]) + output = _run( + connection, + f"bash {remote_script_path} {arguments}", + stdin_text=_sudo_input(password), + timeout=900.0, + on_output=remote_output, + ) + payload_line = next( + ( + line[len(_UPDATE_REPORT_MARKER):] + for line in reversed(output.splitlines()) + if line.startswith(_UPDATE_REPORT_MARKER) + ), + "", + ) + if not payload_line: + raise DeviceInstallError("The device did not return an update report.") + try: + result = json.loads(payload_line) + except json.JSONDecodeError as exc: + raise DeviceInstallError("The device returned an invalid update report.") from exc + if not isinstance(result, dict) or not isinstance(result.get("components"), list): + raise DeviceInstallError("The device returned an incomplete update report.") + result["host_fingerprint"] = connection.fingerprint + return result + finally: + try: + _run(connection, f"rm -f -- {remote_script_path}", timeout=10.0) + except DeviceInstallError: + pass + connection.close() + + +def inspect_managed_service_updates( + *, + host: str, + port: int, + username: str, + password: str, + host_fingerprint: str, + instance_id: str, + runtime_port: int, + hardware_ports: list[int], + hardware_device_ids: dict[int, str] | None = None, + progress: Callable[[dict[str, Any]], None] | None = None, +) -> dict[str, Any]: + """Compare installed managed-service commits with their trusted upstreams.""" + + def report(percent: int, message: str) -> None: + if progress is not None: + progress({ + "progress": max(0, min(100, int(percent))), + "message": str(message), + }) + + selected_instance = _clean_instance_id(instance_id or "default") + selected_runtime_port = int(runtime_port) + if selected_runtime_port < 1 or selected_runtime_port > 65535: + raise DeviceInstallError("The managed runtime port is invalid.") + selected_hardware_ports = sorted({int(value) for value in hardware_ports}) + if any(value < 1 or value > 65535 for value in selected_hardware_ports): + raise DeviceInstallError("A robot hardware port is invalid.") + + report(5, "Connecting to the verified device") + connection = _connect( + host, + port, + username, + password, + expected_fingerprint=host_fingerprint, + timeout=15.0, + ) + arguments = base64.urlsafe_b64encode(json.dumps({ + "instance_id": selected_instance, + "runtime_port": selected_runtime_port, + "hardware_targets": [ + { + "port": value, + "device_id": str((hardware_device_ids or {}).get(value) or ""), + } + for value in selected_hardware_ports + ], + }, separators=(",", ":")).encode("utf-8")).decode("ascii") + script = rf"""python3 - {arguments} <<'PY' +import base64 +import json +import re +import subprocess +import sys +import tomllib +import urllib.request +from pathlib import Path + +MARKER = "__BLACKNODE_UPDATE_CHECK__=" +request = json.loads(base64.urlsafe_b64decode(sys.argv[1]).decode("utf-8")) +home = Path.home() +instance = request["instance_id"] +if instance == "default": + runtime_dir = home / "blacknode-runtime" + runtime_service = "blacknode-runtime.service" +else: + runtime_dir = home / "blacknode-runtimes" / instance + runtime_service = f"blacknode-runtime-{{instance}}.service" + +def command(args, timeout=30): + try: + return subprocess.run( + args, + capture_output=True, + text=True, + check=False, + timeout=timeout, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + return subprocess.CompletedProcess(args, 124, "", str(exc)) + +def project_version(text): + try: + payload = tomllib.loads(text) + return str((payload.get("project") or {{}}).get("version") or "unknown") + except Exception: + return "unknown" + +def package_version(directory): + try: + return project_version( + (Path(directory) / "pyproject.toml").read_text(encoding="utf-8") + ) + except Exception: + return "unknown" + +def resolve_hardware(port, expected_device_id): + installed = command([ + "systemctl", "list-unit-files", "blacknode-hardware*.service", "--no-legend" + ]) + active = command([ + "systemctl", "list-units", "--all", "--type=service", + "blacknode-hardware*.service", "--no-legend", + ]) + candidate_units = {{ + line.split()[0] + for line in (installed.stdout + "\n" + active.stdout).splitlines() + if line.split() + }} + candidate_units.update( + path.name + for root in (Path("/etc/systemd/system"), Path("/run/systemd/system")) + for path in root.glob("blacknode-hardware*.service") + ) + matches = [] + states = {{}} + definitions = {{}} + declared_ports = {{}} + device_ids = {{}} + for unit in sorted(candidate_units): + if not re.fullmatch(r"blacknode-hardware(?:[-.@][A-Za-z0-9_.@-]+)?\.service", unit): + continue + shown = command(["systemctl", "show", unit, "--property=ExecStart", "--value"]) + definition = "" + for root in (Path("/etc/systemd/system"), Path("/run/systemd/system")): + unit_path = root / unit + if unit_path.is_file(): + try: + definition += unit_path.read_text(encoding="utf-8") + "\n" + except OSError: + pass + definitions[unit] = definition + normalized = (shown.stdout + "\n" + definition).replace('"', "") + declared_ports[unit] = sorted({{ + int(value) + for value in re.findall(r"--port(?:=|\s+)([0-9]+)", normalized) + }}) + config_match = re.search(r"--config(?:=|\s+)([^\s;]+)", normalized) + configured_device_id = "" + if config_match: + try: + configured_device_id = str( + json.loads( + Path(config_match.group(1)).read_text(encoding="utf-8") + ).get("device_id") or "" + ) + except Exception: + pass + device_ids[unit] = configured_device_id + states[unit] = {{ + "active": command(["systemctl", "is-active", unit]).stdout.strip() or "unknown", + "enabled": command(["systemctl", "is-enabled", unit]).stdout.strip() or "unknown", + }} + if int(port) in declared_ports[unit]: + matches.append(unit) + resolution_error = "" + if not matches: + identity_matches = [ + unit + for unit in sorted(candidate_units) + if expected_device_id and device_ids.get(unit) == expected_device_id + ] + if identity_matches: + matches = identity_matches + configured = sorted({{ + configured_port + for unit in identity_matches + for configured_port in declared_ports.get(unit, []) + }}) + resolution_error = ( + f"The persistent service for robot {{expected_device_id!r}} " + f"declares port(s) {{configured or ['unknown']}}, while the editor " + f"is paired to port {{port}}. Repair Hardware will reconcile the " + "saved service port using this stable robot identity." + ) + else: + details = ", ".join( + f"{{unit}} (ports={{declared_ports.get(unit) or ['unknown']}}, " + f"device_id={{device_ids.get(unit) or 'unknown'}})" + for unit in sorted(candidate_units) + ) or "none" + raise RuntimeError( + f"No persistent Hardware service matches port {{port}} or robot " + f"identity {{expected_device_id or 'not reported'!r}}. " + f"Discovered units: {{details}}." + ) + active_matches = [ + unit for unit in matches if states[unit]["active"] == "active" + ] + enabled_matches = [ + unit for unit in matches if states[unit]["enabled"] == "enabled" + ] + if len(matches) == 1: + unit = matches[0] + elif len(active_matches) == 1: + unit = active_matches[0] + elif len(active_matches) > 1: + raise RuntimeError( + f"Multiple active Blacknode Hardware services claim port {{port}}: " + + ", ".join(active_matches) + ) + elif len(enabled_matches) == 1: + unit = enabled_matches[0] + else: + details = ", ".join( + f"{{candidate}} (active={{states[candidate]['active']}}, " + f"enabled={{states[candidate]['enabled']}})" + for candidate in matches + ) + raise RuntimeError( + f"Could not choose the persistent Hardware service for port {{port}}. " + f"Candidates: {{details}}." + ) + directory = command([ + "systemctl", "show", unit, "--property=WorkingDirectory", "--value" + ]).stdout.strip() + if not directory: + working_directory = re.findall( + r"^WorkingDirectory=(.+)$", + definitions.get(unit, ""), + flags=re.MULTILINE, + ) + if working_directory: + directory = working_directory[-1].strip().strip('"') + if not directory: + raise RuntimeError(f"{{unit}} does not report its repository WorkingDirectory.") + return unit, Path(directory), resolution_error + +def inspect(kind, repository, service, port, directory): + component = {{ + "kind": kind, + "service_name": service, + "port": int(port), + "installed": {{"version": package_version(directory), "commit": ""}}, + "latest": {{"version": "unknown", "commit": ""}}, + "update_available": False, + "can_update": False, + "dirty": False, + "state": command(["systemctl", "is-active", service]).stdout.strip() or "unknown", + "error": "", + }} + try: + if not (directory / ".git").is_dir(): + raise RuntimeError(f"{{repository}} is not an updateable Git checkout.") + origin = command(["git", "-C", str(directory), "remote", "get-url", "origin"]) + origin_url = origin.stdout.strip() + normalized = origin_url.lower().removesuffix(".git").rstrip("/") + if not re.search( + rf"(?:github\.com[:/])temiroff/{{re.escape(repository)}}$", + normalized, + ): + raise RuntimeError( + f"origin is not the trusted temiroff/{{repository}} repository" + ) + current = command(["git", "-C", str(directory), "rev-parse", "HEAD"]) + if current.returncode: + raise RuntimeError(current.stderr.strip() or "could not read installed commit") + component["installed"]["commit"] = current.stdout.strip()[:12] + dirty = command([ + "git", "-C", str(directory), "status", "--porcelain", "--untracked-files=normal" + ]) + component["dirty"] = bool(dirty.stdout.strip()) + branch = command([ + "git", "-C", str(directory), "rev-parse", "--abbrev-ref", "HEAD", + ]).stdout.strip() + upstream = command([ + "git", "-C", str(directory), "config", "--get", + f"branch.{{branch}}.merge", + ]) + upstream_ref = upstream.stdout.strip() + if not upstream_ref.startswith("refs/heads/"): + raise RuntimeError("the installed branch has no upstream configured") + latest = command(["git", "ls-remote", origin_url, upstream_ref], timeout=45) + if latest.returncode or not latest.stdout.strip(): + raise RuntimeError( + latest.stderr.strip() or "could not read the latest upstream commit" + ) + latest_commit = latest.stdout.split()[0] + component["latest"]["commit"] = latest_commit[:12] + component["update_available"] = current.stdout.strip() != latest_commit + if not component["update_available"]: + component["latest"]["version"] = component["installed"]["version"] + else: + local_latest = command([ + "git", "-C", str(directory), "show", + f"{{latest_commit}}:pyproject.toml", + ]) + if not local_latest.returncode: + component["latest"]["version"] = project_version(local_latest.stdout) + if component["latest"]["version"] == "unknown": + raw_url = ( + f"https://raw.githubusercontent.com/temiroff/{{repository}}/" + f"{{latest_commit}}/pyproject.toml" + ) + try: + with urllib.request.urlopen(raw_url, timeout=15) as response: + component["latest"]["version"] = project_version( + response.read().decode("utf-8") + ) + except Exception: + pass + component["can_update"] = not component["dirty"] + if component["dirty"]: + component["error"] = ( + "Local source changes must be committed, stashed, or removed before update." + ) + except Exception as exc: + component["error"] = str(exc) + return component + +components = [ + inspect( + "runtime", + "blacknode-runtime", + runtime_service, + request["runtime_port"], + runtime_dir, + ) +] +for target in request["hardware_targets"]: + hardware_port = int(target["port"]) + try: + service, directory, resolution_error = resolve_hardware( + hardware_port, + str(target.get("device_id") or ""), + ) + component = inspect( + "hardware", + "blacknode-hardware", + service, + hardware_port, + directory, + ) + if resolution_error: + component["error"] = resolution_error + component["can_update"] = False + components.append(component) + except Exception as exc: + components.append({{ + "kind": "hardware", + "service_name": "unresolved", + "port": int(hardware_port), + "installed": {{"version": "unknown", "commit": ""}}, + "latest": {{"version": "unknown", "commit": ""}}, + "update_available": False, + "can_update": False, + "dirty": False, + "state": "unknown", + "error": str(exc), + }}) +print(MARKER + json.dumps({{ + "ok": all(not item["error"] for item in components), + "components": components, +}}, separators=(",", ":"))) +PY""" + try: + report(25, "Comparing installed Runtime and Hardware commits") + output = _run(connection, script, timeout=120.0) + payload_line = next( + ( + line[len("__BLACKNODE_UPDATE_CHECK__="):] + for line in reversed(output.splitlines()) + if line.startswith("__BLACKNODE_UPDATE_CHECK__=") + ), + "", + ) + if not payload_line: + raise DeviceInstallError("The device did not return a software version report.") + try: + result = json.loads(payload_line) + except json.JSONDecodeError as exc: + raise DeviceInstallError("The device returned an invalid software version report.") from exc + if not isinstance(result, dict) or not isinstance(result.get("components"), list): + raise DeviceInstallError("The device returned an incomplete software version report.") + result["host_fingerprint"] = connection.fingerprint + report(90, "Installed and latest versions compared") + return result + finally: + connection.close() + + def control_runtime( *, host: str, @@ -1129,27 +2078,56 @@ def restart_hardware_service( exit 2 } mapfile -t candidate_units < <( - systemctl list-unit-files 'blacknode-hardware*.service' --no-legend 2>/dev/null \ - | awk '{print $1}' + { + systemctl list-unit-files 'blacknode-hardware*.service' --no-legend 2>/dev/null + systemctl list-units --all --type=service 'blacknode-hardware*.service' \ + --no-legend 2>/dev/null + find /etc/systemd/system /run/systemd/system -maxdepth 1 \ + -name 'blacknode-hardware*.service' -printf '%f\n' 2>/dev/null + } | awk '{print $1}' | sort -u ) matches=() +active_matches=() +enabled_matches=() for unit in "${candidate_units[@]}"; do - [[ "$unit" =~ ^blacknode-hardware([-.][A-Za-z0-9_.@-]+)?\.service$ ]] || continue + [[ "$unit" =~ ^blacknode-hardware([-.@][A-Za-z0-9_.@-]+)?\.service$ ]] || continue exec_start="$(systemctl show "$unit" --property=ExecStart --value 2>/dev/null || true)" - normalized="${exec_start//\"/}" - if [[ "$normalized" =~ --port(=|[[:space:]])${hardware_port}([^0-9]|$) ]]; then + unit_definition="" + for unit_path in "/etc/systemd/system/$unit" "/run/systemd/system/$unit"; do + if [[ -f "$unit_path" ]]; then + unit_definition+="$(command cat "$unit_path")" + unit_definition+=$'\n' + fi + done + normalized="${exec_start//\"/}"$'\n'"${unit_definition//\"/}" + if printf '%s\n' "$normalized" \ + | grep -oE -- '--port(=|[[:space:]])[0-9]+' \ + | grep -oE '[0-9]+' \ + | grep -Fxq -- "$hardware_port"; then matches+=("$unit") + systemctl is-active --quiet "$unit" 2>/dev/null \ + && active_matches+=("$unit") + systemctl is-enabled --quiet "$unit" 2>/dev/null \ + && enabled_matches+=("$unit") fi done if [[ "${#matches[@]}" -eq 0 ]]; then - echo "No Blacknode hardware service owns port $hardware_port." + echo "No Blacknode hardware service owns port $hardware_port. Discovered units: ${candidate_units[*]:-none}." exit 3 fi -if [[ "${#matches[@]}" -ne 1 ]]; then - echo "Multiple Blacknode hardware services claim port $hardware_port." +if [[ "${#matches[@]}" -eq 1 ]]; then + service_name="${matches[0]}" +elif [[ "${#active_matches[@]}" -eq 1 ]]; then + service_name="${active_matches[0]}" +elif [[ "${#active_matches[@]}" -gt 1 ]]; then + echo "Multiple active Blacknode hardware services claim port $hardware_port: ${active_matches[*]}." + exit 4 +elif [[ "${#enabled_matches[@]}" -eq 1 ]]; then + service_name="${enabled_matches[0]}" +else + echo "Could not choose the hardware service for port $hardware_port. Candidates: ${matches[*]}. Enabled: ${enabled_matches[*]:-none}." exit 4 fi -service_name="${matches[0]}" sudo -S -p '' systemctl restart "$service_name" state="" for _attempt in {1..40}; do diff --git a/editor-server/device_registry.py b/editor-server/device_registry.py index e28ed48..cc197b2 100644 --- a/editor-server/device_registry.py +++ b/editor-server/device_registry.py @@ -272,6 +272,16 @@ def deployment_logs(self, deployment_id: str, *, limit: int = 20000) -> dict[str f"{self._deployment_endpoint(deployment_id)}/logs?limit={safe_limit}", ) + def deployment_telemetry( + self, + deployment_id: str, + *, + stream: str = "robot-state", + ) -> dict[str, Any]: + endpoint = self._deployment_endpoint(deployment_id) + query = urllib.parse.urlencode({"stream": stream}) + return self._request("GET", f"{endpoint}/telemetry?{query}") + def delete_deployment(self, deployment_id: str) -> dict[str, Any]: return self._request("DELETE", self._deployment_endpoint(deployment_id)) @@ -497,6 +507,78 @@ def set_host_paused(self, host_id: str, paused: bool) -> dict[str, Any]: ] return public + def set_host_remote_device_id( + self, + host_id: str, + remote_device_id: str, + ) -> dict[str, Any]: + clean_remote_id = str(remote_device_id or "").strip() + if not clean_remote_id: + raise DeviceRegistryError("The runtime manifest has no stable device identity.") + with self._lock: + hosts, records = self._load_payload() + hosts, _changed = self._materialize_hosts(hosts, records) + host = hosts.get(host_id) + if host is None: + raise KeyError(host_id) + current_remote_id = str(host.get("remote_device_id") or "").strip() + if current_remote_id and current_remote_id != clean_remote_id: + raise DeviceRegistryError( + "The paired runtime identity changed. Pair the runtime again " + "before enabling SSH management." + ) + host["remote_device_id"] = clean_remote_id + host["updated_at"] = _iso_now() + self._save_payload(hosts, records) + public = self._public_host(host) + public["robots"] = [ + self._public(record) + for record in records.values() + if str(record.get("host_id") or "") == host_id + ] + return public + + def set_host_management( + self, + host_id: str, + managed_runtime: dict[str, Any], + ) -> dict[str, Any]: + allowed_keys = { + "ssh_host", + "ssh_port", + "ssh_username", + "host_fingerprint", + "instance_id", + "runtime_port", + "service_name", + "runtime_dir", + } + management = { + key: value + for key, value in managed_runtime.items() + if key in allowed_keys + } + if set(management) != allowed_keys: + raise DeviceRegistryError( + "The verified SSH runtime identity is incomplete." + ) + with self._lock: + hosts, records = self._load_payload() + hosts, _changed = self._materialize_hosts(hosts, records) + host = hosts.get(host_id) + if host is None: + raise KeyError(host_id) + host["managed_runtime"] = management + host["updated_at"] = _iso_now() + self._save_payload(hosts, records) + public = self._public_host(host) + public["robots"] = [ + self._public(record) + for record in records.values() + if str(record.get("host_id") or "") == host_id + ] + return public + def rename_host(self, host_id: str, name: str) -> dict[str, Any]: clean_name = str(name or "").strip() if not clean_name: @@ -665,6 +747,10 @@ def pair( "runtime_token_fingerprint": token_fingerprint(clean_runtime_token), "runtime_token_explicit": runtime_token_explicit, "remote_device_id": remote_device_id, + "software_version": ( + str(status.get("software_version") or "").strip() + or str((existing or {}).get("software_version") or "").strip() + ), "paused": False, "created_at": created_at, "updated_at": now, @@ -689,6 +775,25 @@ def pair( self._save_payload(hosts, records) return self._public(record) + def remember_device_software_version( + self, + device_id: str, + software_version: str, + ) -> None: + clean_version = str(software_version or "").strip() + if not clean_version or clean_version.casefold() == "unknown": + return + with self._lock: + hosts, records = self._load_payload() + record = records.get(device_id) + if record is None: + raise KeyError(device_id) + if str(record.get("software_version") or "") == clean_version: + return + record["software_version"] = clean_version + record["updated_at"] = _iso_now() + self._save_payload(hosts, records) + def delete(self, device_id: str) -> bool: with self._lock: hosts, records = self._load_payload() diff --git a/editor-server/server.py b/editor-server/server.py index ec8578f..9433fe8 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 uuid, os, sys, json, threading, re, queue, io, contextlib, time, subprocess, importlib, signal, shlex, hashlib +import asyncio, uuid, os, sys, json, threading, re, queue, io, contextlib, time, subprocess, importlib, signal, shlex, hashlib import urllib.error, urllib.parse, urllib.request from datetime import datetime from pathlib import Path @@ -56,11 +56,13 @@ from device_installer import ( control_runtime, DeviceInstallError, + inspect_managed_service_updates, inspect_runtime, install_runtime, probe_device, restart_hardware_service, uninstall_runtime, + update_managed_services, ) from device_registry import ( DeviceRegistry, @@ -436,6 +438,9 @@ class InstallDeviceHostReq(InspectDeviceHostReq): action: str = "install" instance_id: str = "" +class ConfigureDeviceHostManagementReq(InspectDeviceHostReq): + pass + class UninstallDeviceHostReq(BaseModel): password: str @@ -443,6 +448,10 @@ class RuntimeLifecycleReq(BaseModel): action: str password: str +class UpdateManagedDeviceReq(BaseModel): + password: str + scope: str = "all" + class RobotLifecycleReq(BaseModel): action: str password: str = "" @@ -3086,6 +3095,137 @@ def get_device_host_runtime_status(host_id: str): return _device_host_runtime_status(host_id) +@app.post("/device-hosts/{host_id}/management") +def configure_device_host_management( + host_id: str, + req: ConfigureDeviceHostManagementReq, +): + try: + host = _device_registry.get_host_public(host_id) + except DeviceRegistryError as exc: + raise HTTPException(500, str(exc)) from exc + if host is None: + raise HTTPException(404, "Device not found") + if isinstance(host.get("managed_runtime"), dict): + raise HTTPException(409, "SSH management is already configured for this device.") + + try: + runtime_port = urllib.parse.urlsplit( + str(host.get("runtime_url") or "") + ).port + except ValueError as exc: + raise HTTPException(409, "The paired runtime URL has an invalid port.") from exc + if not runtime_port: + raise HTTPException( + 409, + "The paired runtime URL must include its service port before SSH " + "management can be enabled.", + ) + + try: + inspection = inspect_runtime( + host=req.host, + port=req.port, + username=req.username, + password=req.password, + host_fingerprint=req.host_fingerprint, + ) + except DeviceInstallError as exc: + raise HTTPException(400, str(exc)) from exc + + expected_device_id = str(host.get("remote_device_id") or "").strip() + if not expected_device_id: + try: + manifest = _device_registry.host_client(host_id).manifest() + if ( + manifest.get("service") != "blacknode-runtime" + or manifest.get("protocol_version") != 1 + ): + raise DeviceRegistryError( + "Runtime service identity or protocol is incompatible." + ) + expected_device_id = str(manifest.get("device_id") or "").strip() + _device_registry.set_host_remote_device_id( + host_id, + expected_device_id, + ) + except KeyError as exc: + raise HTTPException(404, "Device not found") from exc + except DeviceRegistryError as exc: + raise HTTPException( + 409, + "Blacknode could not recover a stable identity from this paired " + f"runtime. Pair the runtime again first. {exc}", + ) from exc + matches = [ + instance + for instance in inspection.get("instances", []) + if ( + isinstance(instance, dict) + and int(instance.get("port") or 0) == runtime_port + and bool(instance.get("service_installed")) + and str(instance.get("device_id") or "") == expected_device_id + ) + ] + if not matches: + raise HTTPException( + 409, + f"SSH connected, but no installed Blacknode runtime service on port " + f"{runtime_port} matches this paired device ({expected_device_id or 'unknown ID'}).", + ) + if len(matches) != 1: + raise HTTPException( + 409, + f"Multiple installed runtime services match port {runtime_port}; " + "resolve the duplicate services before enabling SSH management.", + ) + instance = matches[0] + management = { + "ssh_host": str(req.host or "").strip(), + "ssh_port": int(req.port), + "ssh_username": str(req.username or "").strip(), + "host_fingerprint": str(inspection.get("host_fingerprint") or ""), + "instance_id": str(instance.get("instance_id") or ""), + "runtime_port": runtime_port, + "service_name": str(instance.get("service_name") or ""), + "runtime_dir": str(instance.get("runtime_dir") or ""), + } + if not all( + management.get(key) + for key in ( + "ssh_host", + "ssh_username", + "host_fingerprint", + "instance_id", + "service_name", + "runtime_dir", + ) + ): + raise HTTPException( + 409, + "The matching runtime service does not expose a complete management identity.", + ) + try: + device = _device_registry.set_host_management(host_id, management) + except KeyError as exc: + raise HTTPException(404, "Device not found") from exc + except DeviceRegistryError as exc: + raise HTTPException(409, str(exc)) from exc + return { + "ok": True, + "device": device, + "instance": { + key: value + for key, value in instance.items() + if not str(key).startswith("_") + }, + "summary": ( + f"SSH management enabled for {management['service_name']} on " + f"runtime port {runtime_port}." + ), + } + + def _raise_rpc_error(result: dict[str, Any], *, action: str) -> None: error = result.get("error") if isinstance(result, dict) else None if not error: @@ -3284,6 +3424,454 @@ def stream_device_host_lifecycle(host_id: str, req: RuntimeLifecycleReq): ) +def _update_device_host_payload( + host_id: str, + req: UpdateManagedDeviceReq, + progress: Callable[[dict[str, Any]], None] | None = None, +) -> dict[str, Any]: + def report(percent: int, message: str) -> None: + if progress is not None: + progress({ + "progress": max(0, min(100, int(percent))), + "message": str(message), + }) + + if not str(req.password or ""): + raise HTTPException(400, "Enter the SSH password to update this device.") + scope = str(req.scope or "all").strip().lower() + if scope not in {"all", "runtime", "hardware"}: + raise HTTPException(400, "Update scope must be all, runtime, or hardware.") + host = _device_registry.get_host_public(host_id) + if host is None: + raise HTTPException(404, "Device not found") + managed = host.get("managed_runtime") + if not isinstance(managed, dict): + raise HTTPException( + 409, + "Enable SSH controls for this device before updating its services.", + ) + + robots = list(host.get("robots") or []) + hardware_ports: list[int] = [] + hardware_device_ids: dict[int, str] = {} + for robot in robots: + try: + hardware_port = urllib.parse.urlsplit( + str(robot.get("base_url") or "") + ).port + except ValueError as exc: + raise HTTPException( + 409, + f"{robot.get('name') or 'A robot'} has an invalid hardware URL.", + ) from exc + if not hardware_port: + raise HTTPException( + 409, + f"{robot.get('name') or 'A robot'} has no hardware service port.", + ) + hardware_ports.append(hardware_port) + hardware_device_ids[hardware_port] = str( + robot.get("remote_device_id") or "" + ) + include_runtime = scope in {"all", "runtime"} + selected_hardware_ports = ( + hardware_ports if scope in {"all", "hardware"} else [] + ) + if scope == "hardware" and not selected_hardware_ports: + raise HTTPException(409, "This device has no attached Hardware services to update.") + + stopped_deployments: list[str] = [] + controlled_robots: list[str] = [] + warnings: list[str] = [] + runtime_api_unavailable = False + report(5, "Stopping running deployments") + if not host.get("paused"): + last_runtime_error = "" + for attempt in range(3): + try: + runtime_client = _device_registry.host_client(host_id) + for deployment in runtime_client.list_deployments().get("deployments") or []: + if ( + isinstance(deployment, dict) + and str(deployment.get("state") or "") == "running" + and str(deployment.get("id") or "") + ): + deployment_id = str(deployment["id"]) + runtime_client.stop_deployment(deployment_id) + stopped_deployments.append(deployment_id) + last_runtime_error = "" + break + except (DeviceRegistryError, KeyError) as exc: + last_runtime_error = str(exc) + if attempt < 2: + time.sleep(0.35) + if last_runtime_error: + runtime_api_unavailable = True + warnings.append( + "The Runtime API could not stop deployments directly. Blacknode " + "used the verified SSH service identity to stop the Runtime and its " + "deployment process group before updating. " + f"API error: {last_runtime_error}" + ) + + report(12, "Stopping and disarming attached robots") + for robot in robots: + robot_id = str(robot.get("id") or "") + if not robot_id: + continue + try: + result = _paired_device_client(robot_id).rpc({ + "jsonrpc": "2.0", + "id": f"device-update-stop-{robot_id}", + "method": "stop", + "params": {}, + }) + _raise_rpc_error(result, action="prepare update for") + _device_registry.set_device_paused(robot_id, True) + controlled_robots.append(robot_id) + except (DeviceRegistryError, HTTPException, KeyError) as exc: + detail = exc.detail if isinstance(exc, HTTPException) else str(exc) + raise DeviceRegistryError( + f"Robot {robot.get('name') or robot_id} could not be stopped " + f"before update: {detail}" + ) from exc + + runtime_paused_for_fallback = False + try: + if runtime_api_unavailable and not host.get("paused"): + report(14, "Stopping the unreachable Runtime safely over SSH") + control_runtime( + host=str(managed.get("ssh_host") or ""), + port=int(managed.get("ssh_port") or 22), + username=str(managed.get("ssh_username") or ""), + password=req.password, + host_fingerprint=str(managed.get("host_fingerprint") or ""), + instance_id=str(managed.get("instance_id") or "default"), + action="pause", + ) + runtime_paused_for_fallback = True + update = update_managed_services( + host=str(managed.get("ssh_host") or ""), + port=int(managed.get("ssh_port") or 22), + username=str(managed.get("ssh_username") or ""), + password=req.password, + host_fingerprint=str(managed.get("host_fingerprint") or ""), + instance_id=str(managed.get("instance_id") or "default"), + runtime_port=int(managed.get("runtime_port") or 0), + hardware_ports=selected_hardware_ports, + hardware_device_ids={ + hardware_port: hardware_device_ids.get(hardware_port, "") + for hardware_port in selected_hardware_ports + }, + include_runtime=include_runtime, + progress=lambda value: report( + 15 + int(int(value.get("progress") or 0) * 0.75), + str(value.get("message") or "Updating managed services"), + ), + ) + except Exception: + if runtime_paused_for_fallback: + try: + control_runtime( + host=str(managed.get("ssh_host") or ""), + port=int(managed.get("ssh_port") or 22), + username=str(managed.get("ssh_username") or ""), + password=req.password, + host_fingerprint=str(managed.get("host_fingerprint") or ""), + instance_id=str(managed.get("instance_id") or "default"), + action="resume", + ) + except DeviceInstallError: + pass + for robot_id in controlled_robots: + try: + result = _paired_device_client(robot_id).rpc({ + "jsonrpc": "2.0", + "id": f"device-update-recover-{robot_id}", + "method": "resume", + "params": {}, + }) + _raise_rpc_error(result, action="recover after update failure for") + _device_registry.set_device_paused(robot_id, False) + except (DeviceRegistryError, HTTPException, KeyError): + pass + raise + + report(92, "Verifying reported runtime and hardware versions") + runtime_manifest: dict[str, Any] | None = None + runtime_error = "" + for _attempt in range(40): + try: + candidate = _device_registry.host_client(host_id).manifest() + if ( + candidate.get("service") == "blacknode-runtime" + and candidate.get("protocol_version") == 1 + ): + runtime_manifest = candidate + break + except (DeviceRegistryError, KeyError) as exc: + runtime_error = str(exc) + time.sleep(0.25) + if runtime_manifest is None: + raise DeviceRegistryError( + f"The runtime service updated but did not pass verification: {runtime_error}" + ) + + verified_robots: list[dict[str, Any]] = [] + for robot in robots: + robot_id = str(robot.get("id") or "") + expected_id = str(robot.get("remote_device_id") or "") + verified_status: dict[str, Any] | None = None + last_error = "" + for _attempt in range(40): + try: + candidate = _paired_device_client(robot_id).status() + actual_id = str(candidate.get("device_id") or "") + if expected_id and actual_id != expected_id: + raise DeviceRegistryError( + f"returned robot '{actual_id}', expected '{expected_id}'" + ) + verified_status = candidate + break + except (DeviceRegistryError, HTTPException) as exc: + last_error = exc.detail if isinstance(exc, HTTPException) else str(exc) + time.sleep(0.25) + if verified_status is None: + raise DeviceRegistryError( + f"{robot.get('name') or robot_id} did not pass verification: {last_error}" + ) + if verified_status.get("armed"): + raise DeviceRegistryError( + f"{robot.get('name') or robot_id} returned armed after its update." + ) + if verified_status.get("torque_enabled") is True: + warnings.append( + f"{robot.get('name') or robot_id} reports physical servo torque enabled. " + "The software update did not turn off actuator power." + ) + verified_status["paused"] = False + _device_registry.set_device_paused(robot_id, False) + verified_robots.append({ + "id": robot_id, + "name": str(robot.get("name") or robot_id), + "port": urllib.parse.urlsplit(str(robot.get("base_url") or "")).port, + "software_version": str( + verified_status.get("software_version") or "not reported" + ), + "status": verified_status, + }) + + for component in update.get("components") or []: + if not isinstance(component, dict): + continue + if component.get("kind") == "runtime": + component["reported_version"] = str( + runtime_manifest.get("runtime_version") or "not reported" + ) + continue + component_port = int(component.get("port") or 0) + robot_report = next( + ( + item for item in verified_robots + if int(item.get("port") or 0) == component_port + ), + None, + ) + if robot_report: + component["reported_version"] = robot_report["software_version"] + + device = _device_registry.set_host_paused(host_id, False) + changed = sum( + 1 + for component in update.get("components") or [] + if isinstance(component, dict) and component.get("changed") + ) + service_count = len(update.get("components") or []) + current_reinstalled = service_count - changed + if changed: + summary = ( + f"Updated {changed} managed service" + f"{'s' if changed != 1 else ''}" + + ( + f" and reinstalled {current_reinstalled} current service" + f"{'s' if current_reinstalled != 1 else ''}" + if current_reinstalled + else "" + ) + + ". " + ) + else: + summary = ( + f"Reinstalled {service_count} current managed service" + f"{'s' if service_count != 1 else ''}. " + ) + summary += ( + "The runtime and robot monitoring are online, and Blacknode motion remains disarmed." + ) + if warnings: + summary += ( + f" Completed with {len(warnings)} warning" + f"{'s' if len(warnings) != 1 else ''}; check the report." + ) + report(100, summary) + return { + "ok": True, + "scope": scope, + "device": device, + "update": update, + "runtime": runtime_manifest, + "robots": verified_robots, + "stopped_deployments": stopped_deployments, + "controlled_robots": controlled_robots, + "warnings": warnings, + "summary": summary, + } + + +@app.post("/device-hosts/{host_id}/update-stream") +def stream_device_host_update(host_id: str, req: UpdateManagedDeviceReq): + return _lifecycle_stream( + lambda progress: _update_device_host_payload(host_id, req, progress) + ) + + +def _check_device_host_updates_payload( + host_id: str, + req: UpdateManagedDeviceReq, + progress: Callable[[dict[str, Any]], None] | None = None, +) -> dict[str, Any]: + def report(percent: int, message: str) -> None: + if progress is not None: + progress({ + "progress": max(0, min(100, int(percent))), + "message": str(message), + }) + + if not str(req.password or ""): + raise HTTPException(400, "Enter the SSH password to check software versions.") + host = _device_registry.get_host_public(host_id) + if host is None: + raise HTTPException(404, "Device not found") + managed = host.get("managed_runtime") + if not isinstance(managed, dict): + raise HTTPException( + 409, + "Enable SSH controls for this device before checking upstream versions.", + ) + + robots = list(host.get("robots") or []) + hardware_ports: list[int] = [] + hardware_device_ids: dict[int, str] = {} + for robot in robots: + try: + hardware_port = urllib.parse.urlsplit( + str(robot.get("base_url") or "") + ).port + except ValueError as exc: + raise HTTPException( + 409, + f"{robot.get('name') or 'A robot'} has an invalid hardware URL.", + ) from exc + if not hardware_port: + raise HTTPException( + 409, + f"{robot.get('name') or 'A robot'} has no hardware service port.", + ) + hardware_ports.append(hardware_port) + hardware_device_ids[hardware_port] = str( + robot.get("remote_device_id") or "" + ) + + report(8, "Checking installed Runtime and Hardware software") + checked = inspect_managed_service_updates( + host=str(managed.get("ssh_host") or ""), + port=int(managed.get("ssh_port") or 22), + username=str(managed.get("ssh_username") or ""), + password=req.password, + host_fingerprint=str(managed.get("host_fingerprint") or ""), + instance_id=str(managed.get("instance_id") or "default"), + runtime_port=int(managed.get("runtime_port") or 0), + hardware_ports=hardware_ports, + hardware_device_ids=hardware_device_ids, + progress=lambda value: report( + 10 + int(int(value.get("progress") or 0) * 0.75), + str(value.get("message") or "Comparing software versions"), + ), + ) + + warnings: list[str] = [] + runtime_version = "not reported" + try: + manifest = _device_registry.host_client(host_id).manifest() + runtime_version = str(manifest.get("runtime_version") or "not reported") + except (DeviceRegistryError, KeyError) as exc: + warnings.append(f"Runtime live-version check failed: {exc}") + + hardware_versions: dict[int, str] = {} + for robot in robots: + robot_id = str(robot.get("id") or "") + try: + robot_port = urllib.parse.urlsplit(str(robot.get("base_url") or "")).port + status = _paired_device_client(robot_id).status() + if robot_port: + hardware_versions[robot_port] = str( + status.get("software_version") or "not reported" + ) + except (DeviceRegistryError, HTTPException, ValueError) as exc: + detail = exc.detail if isinstance(exc, HTTPException) else str(exc) + warnings.append( + f"{robot.get('name') or robot_id} live-version check failed: {detail}" + ) + + for component in checked.get("components") or []: + if not isinstance(component, dict): + continue + if component.get("kind") == "runtime": + component["reported_version"] = runtime_version + else: + component["reported_version"] = hardware_versions.get( + int(component.get("port") or 0), + "not reported", + ) + + components = [ + item for item in checked.get("components") or [] + if isinstance(item, dict) + ] + available = sum(1 for item in components if item.get("update_available")) + blockers = sum(1 for item in components if item.get("error")) + if blockers: + summary = ( + f"Checked {len(components)} services; {blockers} need attention before " + "Runtime + Hardware can be updated." + ) + elif available: + summary = ( + f"{available} of {len(components)} services have updates available." + ) + else: + summary = f"Runtime + Hardware are current across {len(components)} services." + if warnings: + summary += ( + f" {len(warnings)} live service check" + f"{'s' if len(warnings) != 1 else ''} could not be completed." + ) + report(100, summary) + return { + "ok": blockers == 0, + "check": checked, + "warnings": warnings, + "summary": summary, + } + + +@app.post("/device-hosts/{host_id}/update-check-stream") +def stream_device_host_update_check(host_id: str, req: UpdateManagedDeviceReq): + return _lifecycle_stream( + lambda progress: _check_device_host_updates_payload(host_id, req, progress) + ) + + def _control_robot_lifecycle_payload( device_id: str, req: RobotLifecycleReq, @@ -3626,6 +4214,16 @@ def get_device_status(device_id: str): raise HTTPException(502, str(exc)) from exc +@app.get("/devices/{device_id}/monitor") +def get_device_monitor(device_id: str): + try: + return _device_monitor_snapshot(device_id) + except KeyError as exc: + raise HTTPException(404, "Device not found") from exc + except DeviceRegistryError as exc: + raise HTTPException(502, str(exc)) from exc + + @app.post("/devices/{device_id}/lifecycle") def control_robot_lifecycle(device_id: str, req: RobotLifecycleReq): try: @@ -4857,25 +5455,39 @@ def _set_device_deployment_lease(device_id: str, *, leased: bool) -> None: def _deployment_aware_device_status(device_id: str) -> dict[str, Any]: - """Recover an orphaned hardware lease, or explain the active owner.""" + """Report running deployments separately from physical motion ownership.""" client = _paired_device_client(device_id) status = client.status() saved_device = _device_registry.get_public(device_id) - if saved_device and saved_device.get("paused"): - result = dict(status) - result["paused"] = True - result["notice"] = ( - "Robot is paused and disarmed. Resume it before starting a workflow." - ) - return result + reported_version = str(status.get("software_version") or "").strip() + saved_version = str((saved_device or {}).get("software_version") or "").strip() + if reported_version and reported_version.casefold() != "unknown": + try: + _device_registry.remember_device_software_version( + device_id, + reported_version, + ) + except KeyError: + pass + elif saved_version: + status = { + **status, + "software_version": saved_version, + "software_version_cached": True, + } + paused = bool(saved_device and saved_device.get("paused")) + if paused: + status = {**status, "paused": True} error = str(status.get("error") or "") folded_error = error.casefold() - if "leased" not in folded_error or "deployment" not in folded_error: - return status + hardware_is_leased = ( + "leased" in folded_error + and "deployment" in folded_error + ) try: payload = _device_registry.runtime_client(device_id).list_deployments() - except (DeviceRegistryError, KeyError): + except (DeviceRegistryError, KeyError, AttributeError, TypeError): return status deployments = [ item @@ -4893,32 +5505,274 @@ def _deployment_aware_device_status(device_id: str) -> dict[str, Any]: if active: owner = active[0] result = dict(status) - result["deployment_lease"] = { + deployment = { "id": str(owner.get("id") or ""), "name": str(owner.get("name") or owner.get("id") or "Deployment"), "state": str(owner.get("state") or "running"), } - result.pop("error", None) + if hardware_is_leased: + result["deployment_lease"] = deployment + result.pop("error", None) + result["notice"] = ( + f"Running deployment '{owner.get('name') or owner.get('id')}' " + "controls this robot. Device checks are paused here to prevent " + "another process from opening the same hardware connection." + ) + else: + result["running_deployment"] = deployment + result["notice"] = ( + f"Deployment '{owner.get('name') or owner.get('id')}' is running, " + "but the hardware service does not report that it owns motion control." + ) + return result + + if hardware_is_leased: + try: + _set_device_deployment_lease(device_id, leased=False) + status = client.status() + if paused: + status = {**status, "paused": True} + except HTTPException: + pass + + stored = sorted( + ( + item + for item in deployments + if ( + str(item.get("state") or "") != "running" + and str(item.get("target_device_id") or "") in {"", device_id} + and str(item.get("id") or "") + ) + ), + key=lambda item: ( + str(item.get("updated_at") or ""), + str(item.get("created_at") or ""), + ), + reverse=True, + ) + if stored: + deployment = stored[0] + result = dict(status) + result["stored_deployment"] = { + "id": str(deployment.get("id") or ""), + "name": str( + deployment.get("name") + or deployment.get("id") + or "Deployment" + ), + "state": str(deployment.get("state") or "stopped"), + } result["notice"] = ( - f"Running deployment '{owner.get('name') or owner.get('id')}' " - "controls this robot. Device checks are paused here to prevent " - "another process from opening the same hardware connection." + f"Deployment '{deployment.get('name') or deployment.get('id')}' " + f"is stored on the Runtime in the " + f"{deployment.get('state') or 'stopped'} state" + + ( + ". Resume this robot before restarting it." + if paused + else " and can be restarted." + ) + ) + return result + + if paused: + result = dict(status) + result["notice"] = ( + "Robot is paused and disarmed. Resume it before starting a workflow." ) return result + return status + + +def _device_monitor_snapshot(device_id: str) -> dict[str, Any]: + """Return one normalized robot-state sample from the current bus owner.""" + device = _device_registry.get_public(device_id) + if device is None: + raise KeyError(device_id) + status = _deployment_aware_device_status(device_id) + deployment = status.get("deployment_lease") or status.get("running_deployment") + now = datetime.now().astimezone().isoformat(timespec="milliseconds") + if isinstance(deployment, dict) and deployment.get("id"): + deployment_id = str(deployment["id"]) + try: + telemetry = _device_registry.runtime_client(device_id).deployment_telemetry( + deployment_id, + ) + except (DeviceRegistryError, AttributeError, TypeError) as exc: + detail = str(exc) + endpoint_missing = "HTTP 404" in detail or "not found" in detail.casefold() + return { + "type": "robot_telemetry", + "robot_id": device_id, + "robot_name": str(device.get("name") or device_id), + "source": "deployment", + "source_label": str(deployment.get("name") or deployment_id), + "deployment": deployment, + "available": False, + "stale": True, + "received_at": now, + "message": ( + ( + "This device Runtime does not support deployed monitoring yet. " + "Update Runtime to 0.3.9 or newer; Robot Hardware does not " + "need to be reinstalled. Then stage the workflow again to sync " + "blacknode-drivers 0.2.1 or newer and restart the deployment." + ) + if endpoint_missing + else ( + "The running deployment has not exposed live telemetry yet. " + "Update Runtime and the robot driver package, then restart the " + f"deployment. Details: {detail}" + ) + ), + } + payload = telemetry.get("payload") + available = bool(telemetry.get("available") and isinstance(payload, dict)) + return { + "type": "robot_telemetry", + "robot_id": device_id, + "robot_name": str(device.get("name") or device_id), + "source": "deployment", + "source_label": str(deployment.get("name") or deployment_id), + "deployment": deployment, + "available": available, + "stale": bool(telemetry.get("stale", not available)), + "sequence": int(telemetry.get("sequence") or 0), + "sent_at": str(telemetry.get("sent_at") or ""), + "received_at": str(telemetry.get("received_at") or now), + "age_seconds": telemetry.get("age_seconds"), + "payload": payload if available else None, + "message": str( + telemetry.get("message") + or ( + "Receiving state from the running deployment." + if available + else "Waiting for the running deployment to publish robot state." + ) + ), + } + + positions = { + str(name): float(value) + for name, value in dict(status.get("positions") or {}).items() + if isinstance(value, (int, float)) + } + reported_joint_names = [ + str(name) + for name in (status.get("joint_names") or []) + if str(name) in positions + ] + joint_names = reported_joint_names + [ + name for name in positions if name not in reported_joint_names + ] + available = bool(status.get("connected") and positions) + return { + "type": "robot_telemetry", + "robot_id": device_id, + "robot_name": str(device.get("name") or device_id), + "source": "hardware", + "source_label": "Robot Hardware", + "available": available, + "stale": not available, + "sequence": int(time.time() * 1000), + "sent_at": now, + "received_at": now, + "payload": { + "connected": bool(status.get("connected")), + "armed": bool(status.get("armed")), + "torque_enabled": status.get("torque_enabled"), + "position_unit": "degree", + "velocity_unit": "degree/s", + "joints": [ + { + "name": name, + "position": positions[name], + "velocity": 0.0, + } + for name in joint_names + if name in positions + ], + "error": str(status.get("error") or ""), + }, + "message": ( + "Receiving state from Robot Hardware." + if available + else str( + status.get("error") + or "Robot Hardware is connected but has not reported joint positions." + ) + ), + } + +@app.websocket("/api/devices/{device_id}/monitor/ws") +@app.websocket("/devices/{device_id}/monitor/ws") +async def device_monitor_socket(websocket: WebSocket, device_id: str): + await websocket.accept() try: - _set_device_deployment_lease(device_id, leased=False) - except HTTPException: - return status - return client.status() + while True: + try: + snapshot = await asyncio.to_thread(_device_monitor_snapshot, device_id) + except KeyError: + await websocket.send_json({ + "type": "robot_telemetry", + "robot_id": device_id, + "available": False, + "stale": True, + "message": "Robot is no longer paired with this editor.", + }) + await websocket.close(code=1008) + return + except Exception as exc: # keep a transient device outage visible + snapshot = { + "type": "robot_telemetry", + "robot_id": device_id, + "available": False, + "stale": True, + "received_at": datetime.now().astimezone().isoformat( + timespec="milliseconds" + ), + "message": f"Monitoring temporarily unavailable: {exc}", + } + await websocket.send_json(snapshot) + await asyncio.sleep(0.1) + except (WebSocketDisconnect, RuntimeError): + return @app.get("/devices/{device_id}/deployments") def list_device_deployments(device_id: str): try: - return _runtime_client_or_404(device_id).list_deployments() + payload = _runtime_client_or_404(device_id).list_deployments() except DeviceRegistryError as exc: raise HTTPException(502, str(exc)) from exc + deployments = [ + deployment + for deployment in (payload.get("deployments") or []) + if ( + isinstance(deployment, dict) + and str(deployment.get("target_device_id") or "").strip() + in {"", device_id} + ) + ] + return {**payload, "deployments": deployments} + + +def _require_targeted_deployment( + device_id: str, + deployment_id: str, +) -> dict[str, Any]: + try: + deployment = _runtime_client_or_404(device_id).get_deployment(deployment_id) + except DeviceRegistryError as exc: + raise HTTPException(502, str(exc)) from exc + target_device_id = str(deployment.get("target_device_id") or "").strip() + if target_device_id and target_device_id != device_id: + raise HTTPException( + 404, + f"Deployment '{deployment_id}' does not belong to this robot.", + ) + return deployment def _remote_deployment_owner( @@ -5093,14 +5947,12 @@ def stage_device_deployment(device_id: str, req: RemoteDeployReq): @app.get("/devices/{device_id}/deployments/{deployment_id}") def get_device_deployment(device_id: str, deployment_id: str): - try: - return _runtime_client_or_404(device_id).get_deployment(deployment_id) - except DeviceRegistryError as exc: - raise HTTPException(502, str(exc)) from exc + return _require_targeted_deployment(device_id, deployment_id) @app.post("/devices/{device_id}/deployments/{deployment_id}/start") def start_device_deployment(device_id: str, deployment_id: str): + _require_targeted_deployment(device_id, deployment_id) _require_device_safe_to_start(device_id) _set_device_deployment_lease(device_id, leased=True) try: @@ -5114,6 +5966,7 @@ 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) try: deployment = _runtime_client_or_404(device_id).stop_deployment(deployment_id) except DeviceRegistryError as exc: @@ -5128,6 +5981,7 @@ def rollback_device_deployment( deployment_id: str, req: RemoteRollbackReq, ): + _require_targeted_deployment(device_id, deployment_id) if req.start: _require_device_safe_to_start(device_id) _set_device_deployment_lease(device_id, leased=True) @@ -5153,6 +6007,7 @@ def get_device_deployment_logs( deployment_id: str, limit: int = 20000, ): + _require_targeted_deployment(device_id, deployment_id) try: return _runtime_client_or_404(device_id).deployment_logs( deployment_id, @@ -5164,6 +6019,7 @@ def get_device_deployment_logs( @app.delete("/devices/{device_id}/deployments/{deployment_id}") def delete_device_deployment(device_id: str, deployment_id: str): + _require_targeted_deployment(device_id, deployment_id) try: return _runtime_client_or_404(device_id).delete_deployment(deployment_id) except DeviceRegistryError as exc: @@ -5187,6 +6043,51 @@ def call_device_rpc(device_id: str, req: DeviceRpcReq): raise HTTPException(502, str(exc)) from exc +@app.post("/devices/{device_id}/release-torque") +def release_device_torque(device_id: str): + status = _deployment_aware_device_status(device_id) + if status.get("deployment_lease") or status.get("running_deployment"): + raise HTTPException( + 409, + "Stop the robot deployment before releasing physical torque.", + ) + if status.get("paused"): + raise HTTPException( + 409, + "Resume the robot hardware monitor before releasing physical torque.", + ) + if status.get("torque_enabled") is False: + return {"ok": True, "status": status, "already_released": True} + result = _paired_device_client(device_id).rpc({ + "jsonrpc": "2.0", + "id": f"release-torque-{device_id}", + "method": "disable_torque", + "params": {}, + }) + _raise_rpc_error(result, action="release physical torque for") + verified = _paired_device_client(device_id).status() + if verified.get("torque_enabled") is True: + raise HTTPException( + 409, + "The Hardware service accepted the request but still reports physical torque on.", + ) + verification_warning = "" + if verified.get("torque_enabled") is None: + verification_warning = str( + verified.get("torque_report_error") + or ( + "Robot Hardware sent the torque-off command successfully, but could " + "not read every servo torque-enable register to verify the result." + ) + ) + return { + "ok": True, + "status": verified, + "already_released": False, + "verification_warning": verification_warning, + } + + @app.delete("/devices/{device_id}") def delete_device(device_id: str): try: diff --git a/editor/src/api.ts b/editor/src/api.ts index ff8e222..8c89344 100644 --- a/editor/src/api.ts +++ b/editor/src/api.ts @@ -101,6 +101,7 @@ export interface HardwareDevice { token_fingerprint: string runtime_token_fingerprint?: string runtime_token_configured?: boolean + software_version?: string paused?: boolean created_at: string updated_at: string @@ -216,9 +217,29 @@ export interface DeviceRuntimeStatus { export interface HardwareDeviceStatus { device_id: string + software_version?: string + software_version_cached?: boolean + service_features?: string[] connected: boolean armed: boolean torque_enabled?: boolean + torque_report_error?: string + telemetry?: { + enabled: boolean + streams: string[] + sinks: Array<{ + name: string + configured?: boolean + connected?: boolean + broker?: string + tls?: boolean + topic_prefix?: string + qos?: number + published?: number + last_published_at?: number | null + error?: string + }> + } calibrated?: boolean leased_to_deployment?: boolean paused?: boolean @@ -227,6 +248,16 @@ export interface HardwareDeviceStatus { name: string state: string } + running_deployment?: { + id: string + name: string + state: string + } + stored_deployment?: { + id: string + name: string + state: string + } capabilities: string[] joint_names?: string[] positions?: Record @@ -244,6 +275,58 @@ export interface HardwareDeviceStatus { } } +export interface RobotTelemetryJoint { + name: string + position: number + velocity: number +} + +export interface RobotTelemetrySample { + type: 'robot_telemetry' + robot_id: string + robot_name?: string + source?: 'hardware' | 'deployment' + source_label?: string + deployment?: { + id: string + name: string + state: string + } + available: boolean + stale: boolean + sequence?: number + sent_at?: string + received_at?: string + age_seconds?: number + payload?: { + connected: boolean + armed?: boolean + torque_enabled?: boolean | null + position_unit: string + velocity_unit: string + joints: RobotTelemetryJoint[] + error?: string + battery?: { + level?: number + voltage?: number + charging?: boolean + } + camera_streams?: Array<{ + id: string + label?: string + url?: string + }> + } | null + message?: string +} + +export function deviceMonitorSocketUrl(id: string): string { + const path = `${BASE}/devices/${encodeURIComponent(id)}/monitor/ws` + const url = new URL(path, window.location.href) + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:' + return url.toString() +} + export interface DeviceInstallProgress { progress: number message: string @@ -261,6 +344,64 @@ export interface ComputeDeviceLifecycleResult { summary: string } +export interface ManagedServiceUpdateResult { + ok: boolean + scope: 'all' | 'runtime' | 'hardware' + device: ComputeDevice + update: { + ok: boolean + host_fingerprint: string + components: Array<{ + kind: 'runtime' | 'hardware' + service_name: string + port: number + before: { version: string; commit: string } + after: { version: string; commit: string } + reported_version?: string + changed: boolean + state: string + }> + } + runtime: { + runtime_version?: string + [key: string]: unknown + } + robots: Array<{ + id: string + name: string + port: number + software_version: string + status: HardwareDeviceStatus + }> + stopped_deployments: string[] + controlled_robots: string[] + warnings: string[] + summary: string +} + +export interface ManagedServiceUpdateCheckResult { + ok: boolean + check: { + ok: boolean + host_fingerprint: string + components: Array<{ + kind: 'runtime' | 'hardware' + service_name: string + port: number + installed: { version: string; commit: string } + latest: { version: string; commit: string } + reported_version?: string + update_available: boolean + can_update: boolean + dirty: boolean + state: string + error: string + }> + } + warnings: string[] + summary: string +} + export interface RobotLifecycleResult { ok: boolean action: 'pause' | 'resume' | 'restart' @@ -995,6 +1136,31 @@ export const api = { }, 60000, ), + configureComputeDeviceSsh: ( + id: string, + host: string, + port: number, + username: string, + password: string, + hostFingerprint: string, + ) => + req<{ + ok: boolean + device: ComputeDevice + instance: SshRuntimeInstance + summary: string + }>( + 'POST', + `/device-hosts/${encodeURIComponent(id)}/management`, + { + host, + port, + username, + password, + host_fingerprint: hostFingerprint, + }, + 60000, + ), installComputeDevice: ( name: string, host: string, @@ -1066,6 +1232,27 @@ export const api = { { action, password }, onProgress, ), + updateComputeDevice: ( + id: string, + password: string, + scope: 'all' | 'runtime' | 'hardware', + onProgress: (progress: DeviceActionProgress) => void, + ) => + streamDeviceAction( + `/device-hosts/${encodeURIComponent(id)}/update-stream`, + { password, scope }, + onProgress, + ), + checkComputeDeviceUpdates: ( + id: string, + password: string, + onProgress: (progress: DeviceActionProgress) => void, + ) => + streamDeviceAction( + `/device-hosts/${encodeURIComponent(id)}/update-check-stream`, + { password }, + onProgress, + ), listDevices: () => req<{ devices: HardwareDevice[] }>('GET', '/devices'), pairDevice: (name: string, baseUrl: string, token: string, runtimeToken = '') => req<{ @@ -1196,6 +1383,18 @@ export const api = { { jsonrpc: '2.0', id: requestId, method, params }, 10000, ), + releaseDeviceTorque: (id: string) => + req<{ + ok: boolean + status: HardwareDeviceStatus + already_released: boolean + verification_warning?: string + }>( + 'POST', + `/devices/${encodeURIComponent(id)}/release-torque`, + {}, + 15000, + ), deleteDevice: (id: string) => req<{ ok: boolean; id: string }>('DELETE', `/devices/${encodeURIComponent(id)}`), getOnboarding: () => req<{ package_welcome_seen: boolean }>('GET', '/settings/onboarding'), diff --git a/editor/src/components/DeploymentsPanel.tsx b/editor/src/components/DeploymentsPanel.tsx index 1fca97b..9431c09 100644 --- a/editor/src/components/DeploymentsPanel.tsx +++ b/editor/src/components/DeploymentsPanel.tsx @@ -1,6 +1,7 @@ import { useEffect, useMemo, useState, type CSSProperties } from 'react' import { api, + type ComputeDevice, type DeviceCalibrationCandidate, type DeviceRobotProfile, type Deployment, @@ -52,6 +53,16 @@ function deviceForHardwareIdentity( return matches.length === 1 ? matches[0] : null } +function deploymentsForRobot( + deployments: RemoteDeployment[], + deviceId: string, +): RemoteDeployment[] { + return deployments.filter(deployment => ( + !deployment.target_device_id + || deployment.target_device_id === deviceId + )) +} + const STATE_COLOR: Record = { running: 'var(--ok)', stopped: 'var(--tx3)', @@ -84,15 +95,22 @@ const REMOTE_STATE_LABEL: Record = { interface DeploymentsPanelProps { onOpenTemplates: (query: string) => void + targetDeviceId?: string + onBackToDevices?: () => void } -export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelProps) { +export default function DeploymentsPanel({ + onOpenTemplates, + targetDeviceId = '', + onBackToDevices, +}: DeploymentsPanelProps) { const [deployments, setDeployments] = useState([]) const [error, setError] = useState(null) const [busy, setBusy] = useState(false) const [openId, setOpenId] = useState(null) const [logs, setLogs] = useState>({}) const [devices, setDevices] = useState([]) + const [computeDevices, setComputeDevices] = useState([]) const [selectedDeviceId, setSelectedDeviceId] = useState('') const [preflight, setPreflight] = useState(null) const [remoteDeployments, setRemoteDeployments] = useState([]) @@ -106,6 +124,12 @@ export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelPr const activeTabId = useStore(s => s.activeTabId) const activeTab = tabs.find(tab => tab.id === activeTabId) const activeDeploymentName = deploymentNameFromTab(activeTab?.name) + const selectedRobot = devices.find(device => device.id === selectedDeviceId) ?? null + const selectedComputeDevice = computeDevices.find(device => ( + device.id === selectedRobot?.host_id + || device.robots.some(robot => robot.id === selectedDeviceId) + )) ?? null + const isRobotContext = Boolean(targetDeviceId) const activeProject = useStore(s => s.activeProject) const deploymentProject = ( activeProject @@ -321,7 +345,12 @@ export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelPr const pull = async () => { try { const result = await api.listRemoteDeployments(selectedDeviceId) - if (!cancelled) setRemoteDeployments(result.deployments) + if (!cancelled) { + setRemoteDeployments(deploymentsForRobot( + result.deployments, + selectedDeviceId, + )) + } } catch { if (!cancelled) setRemoteDeployments([]) } @@ -332,17 +361,21 @@ export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelPr }, [selectedDeviceId]) useEffect(() => { - api.listDevices() - .then(result => { - setDevices(result.devices) + Promise.all([api.listDevices(), api.listComputeDevices()]) + .then(([robotResult, computeResult]) => { + setDevices(robotResult.devices) + setComputeDevices(computeResult.devices) setSelectedDeviceId(current => ( - current && result.devices.some(device => device.id === current) + targetDeviceId + && robotResult.devices.some(device => device.id === targetDeviceId) + ? targetDeviceId + : current && robotResult.devices.some(device => device.id === current) ? current - : result.devices[0]?.id ?? '' + : robotResult.devices[0]?.id ?? '' )) }) .catch(err => setError(err instanceof Error ? err.message : String(err))) - }, []) + }, [targetDeviceId]) useEffect(() => { if (!calibrationMatchedDevice) return @@ -470,7 +503,7 @@ export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelPr const refreshRemote = async () => { if (!selectedDeviceId) return const result = await api.listRemoteDeployments(selectedDeviceId) - setRemoteDeployments(result.deployments) + setRemoteDeployments(deploymentsForRobot(result.deployments, selectedDeviceId)) } const actRemote = async (fn: () => Promise) => { @@ -541,8 +574,8 @@ export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelPr await refreshRemote() setRemoteNotice( start - ? `"${name}" was sent to the device and started.` - : `"${name}" was sent to the device and is ready to start.`, + ? `"${name}" was sent to ${selectedComputeDevice?.name || 'the compute device'} and started for ${selectedRobot?.name || 'the selected robot'}.` + : `"${name}" was sent to ${selectedComputeDevice?.name || 'the compute device'} for ${selectedRobot?.name || 'the selected robot'} and is ready to start.`, ) } catch (err) { setError(err instanceof Error ? err.message : String(err)) @@ -559,56 +592,133 @@ export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelPr setRemoteOpenId(current => current === deployment.id ? null : current) } - const running = deployments.filter(d => d.state === 'running').length const remoteRunning = remoteDeployments.filter(d => d.state === 'running').length return (
-
-
-
Deployments
-
- {deployments.length} total ยท {running} running -
+ {isRobotContext ? ( +
+ {onBackToDevices && ( + + )} +
-
- - - + ) : ( +
+
+
Robot deployment
+
Choose a robot for the current workflow
+
+
+ + + +
-
+ )} {error &&
{error}
}
-
Deploy one robot
-
- Choose the computer physically connected to this robot. Use a separate - deployment workflow for each additional robot. +
+ {selectedRobot ? `Deploy to ${selectedRobot.name}` : 'Choose a robot'} +
+
+ Review where the current workflow will run, then deploy it.
- {devices.length > 0 ? ( - - ) : ( -
Pair a Raspberry Pi in the Devices tab first.
- )} + +
+
+ Workflow + {activeDeploymentName} + Current editor tab +
+ +
+ Robot + {selectedRobot?.name || 'No robot selected'} + + {selectedComputeDevice + ? `Connected through ${selectedComputeDevice.name}` + : 'Attach a robot in Devices first'} + +
+
+ + {devices.length > 0 && !isRobotContext ? ( + + ) : devices.length === 0 ? ( +
+ Open Devices, add a compute device, then attach the physical robot to it. +
+ ) : null}
Deployment ownership:{' '} {deploymentProject @@ -627,9 +737,10 @@ export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelPr
0 ? ' is-complete' : ' is-needed'}`}>
1
-
Choose one robot profile
+
Confirm the workflowโ€™s robot profile
- The profile describes this robotโ€™s model, joints, servo IDs, and driver. + This configures how the workflow controls the selected physical robot. + The physical robot remains registered under its compute device.
{robotNodes.length === 0 ? ( @@ -945,10 +1056,10 @@ export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelPr {preflight?.ready && ( <> )} @@ -971,8 +1082,15 @@ export default function DeploymentsPanel({ onOpenTemplates }: DeploymentsPanelPr
- Remote deployments - {remoteDeployments.length} total ยท {remoteRunning} running + + {selectedRobot + ? `Deployments for ${selectedRobot.name}` + : 'Robot deployments'} + + + {selectedComputeDevice ? `Stored on ${selectedComputeDevice.name} ยท ` : ''} + {remoteDeployments.length} total ยท {remoteRunning} running +
+ @@ -1035,7 +1517,11 @@ export default function DevicesPanel() { Install or pair Blacknode Runtime first. Then attach one or more robots and deploy a workflow to the robot you choose.

-
@@ -1049,6 +1535,8 @@ export default function DevicesPanel() { selected={false} onSelect={() => { setSelectedDeviceId(device.id) + setUpdateCheckReport(null) + setUpdateReport(null) setShowDeviceForm(false) setShowRobotForm(false) }} @@ -1060,77 +1548,287 @@ export default function DevicesPanel() { {selectedDevice && (
+
+ +
-
- - Compute device +
{selectedDevice.name} {selectedDevice.runtime_url}
-
+
- + {!selectedDevice.managed_runtime && ( + + )} {selectedDevice.managed_runtime && ( <> + )} -
+ {showSshManagement && !selectedDevice.managed_runtime && ( +
void configureSshManagement(event, selectedDevice)} + > +
+ +
+ Enable SSH lifecycle controls + + Securely connect this paired computer so Blacknode can pause, + resume, and maintain its runtime. + +
+ + + Password not saved + +
+ +
+ + 1 + Connection + +