From dfb258521afa772c1f18c23fab81f193f2b2a6f1 Mon Sep 17 00:00:00 2001 From: Gareth Morgan Date: Mon, 2 Nov 2026 10:21:16 -0500 Subject: [PATCH 01/15] Add standalone Manus glove calibration worker and browser example Stage 1 of Manus glove calibration. - src/plugins/manus/calibration: manus_glove_calibrator, a separate executable in the Manus plugin package (not a registered tracking plugin). It owns the SDK connection for one attempt, runs the SDK's calibration steps on operator confirmation, exports each glove's calibration, shuts the SDK down, then sends one terminal result over a line-delimited JSON socket. The wire format and step engine are an SDK-free static library with a fake-SDK test suite. - examples/manus_calibration: a plain HTML page served by a standard library HTTP and server-sent-events server. One browser owns an attempt, confirmations name the prompt they answer, and Host, Origin and a per-tab client id are checked. A supervisor bounds answers only the worker can give, terminates a silent worker with an uncertain outcome, and refuses new attempts while an unreaped worker may hold the gloves. Exported files are verified and published through a hard link and an atomic manifest only after a completed result and clean exit. - CMake: build and install the worker with the manus component; add its tests when the SDK is present. Negative SDK step times (continuous steps with an estimated duration) are operator-paced and carry the estimate as guidance; the worker claims no completion signal for them. A failed SDK release downgrades a completed or cancelled outcome to failed. Known issue: with the real SDK and no working license, the SDK call that starts a step did not return within the supervisor's bound; the cause is not established. An end-to-end calibration, export and tracking load are not yet demonstrated, and the final revision was not retested. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Gareth Morgan --- examples/manus_calibration/README.md | 107 ++ examples/manus_calibration/pyproject.toml | 34 + .../manus_calibration/__init__.py | 4 + .../manus_calibration/__main__.py | 108 ++ .../manus_calibration/controller.py | 235 +++ .../manus_calibration/protocol.py | 523 ++++++ .../manus_calibration/server.py | 409 +++++ .../manus_calibration/static/app.js | 207 +++ .../manus_calibration/static/index.html | 60 + .../manus_calibration/static/style.css | 114 ++ .../manus_calibration/store.py | 318 ++++ .../manus_calibration/wire.py | 82 + .../manus_calibration/worker.py | 804 +++++++++ examples/manus_calibration/tests/conftest.py | 89 + .../manus_calibration/tests/fake_worker.py | 412 +++++ .../tests/test_controller.py | 538 ++++++ .../manus_calibration/tests/test_protocol.py | 439 +++++ .../manus_calibration/tests/test_server.py | 1393 +++++++++++++++ .../manus_calibration/tests/test_store.py | 307 ++++ .../manus_calibration/tests/test_watchdog.py | 395 +++++ examples/manus_calibration/tests/test_wire.py | 236 +++ .../manus_calibration/tests/test_worker.py | 904 ++++++++++ .../manus_calibration/tests/web_helpers.py | 200 +++ src/plugins/manus/CMakeLists.txt | 2 + src/plugins/manus/calibration/CMakeLists.txt | 5 + src/plugins/manus/calibration/README.md | 102 ++ .../manus/calibration/app/CMakeLists.txt | 30 + src/plugins/manus/calibration/app/main.cpp | 282 +++ .../manus/calibration/app/manus_sdk.cpp | 322 ++++ .../manus/calibration/app/manus_sdk.hpp | 40 + .../manus/calibration/core/CMakeLists.txt | 29 + src/plugins/manus/calibration/core/engine.cpp | 533 ++++++ .../manus/calibration/core/export_writer.cpp | 100 ++ .../core/inc/manus_calibration/engine.hpp | 200 +++ .../inc/manus_calibration/export_writer.hpp | 40 + .../core/inc/manus_calibration/protocol.hpp | 210 +++ .../core/inc/manus_calibration/sha256.hpp | 16 + src/plugins/manus/calibration/core/json.cpp | 473 ++++++ src/plugins/manus/calibration/core/json.hpp | 35 + .../manus/calibration/core/protocol.cpp | 526 ++++++ src/plugins/manus/calibration/core/sha256.cpp | 108 ++ tests/cpp/plugins/CMakeLists.txt | 5 + .../plugins/manus_calibration/CMakeLists.txt | 18 + .../plugins/manus_calibration/fake_sdk.hpp | 406 +++++ .../plugins/manus_calibration/test_engine.cpp | 1511 +++++++++++++++++ .../manus_calibration/test_export_writer.cpp | 237 +++ .../manus_calibration/test_protocol.cpp | 636 +++++++ .../plugins/manus_calibration/test_sha256.cpp | 75 + 48 files changed, 13859 insertions(+) create mode 100644 examples/manus_calibration/README.md create mode 100644 examples/manus_calibration/pyproject.toml create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/__init__.py create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/__main__.py create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/controller.py create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/protocol.py create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/server.py create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/app.js create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/index.html create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/style.css create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/store.py create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/wire.py create mode 100644 examples/manus_calibration/python/isaaccapture_examples/manus_calibration/worker.py create mode 100644 examples/manus_calibration/tests/conftest.py create mode 100644 examples/manus_calibration/tests/fake_worker.py create mode 100644 examples/manus_calibration/tests/test_controller.py create mode 100644 examples/manus_calibration/tests/test_protocol.py create mode 100644 examples/manus_calibration/tests/test_server.py create mode 100644 examples/manus_calibration/tests/test_store.py create mode 100644 examples/manus_calibration/tests/test_watchdog.py create mode 100644 examples/manus_calibration/tests/test_wire.py create mode 100644 examples/manus_calibration/tests/test_worker.py create mode 100644 examples/manus_calibration/tests/web_helpers.py create mode 100644 src/plugins/manus/calibration/CMakeLists.txt create mode 100644 src/plugins/manus/calibration/README.md create mode 100644 src/plugins/manus/calibration/app/CMakeLists.txt create mode 100644 src/plugins/manus/calibration/app/main.cpp create mode 100644 src/plugins/manus/calibration/app/manus_sdk.cpp create mode 100644 src/plugins/manus/calibration/app/manus_sdk.hpp create mode 100644 src/plugins/manus/calibration/core/CMakeLists.txt create mode 100644 src/plugins/manus/calibration/core/engine.cpp create mode 100644 src/plugins/manus/calibration/core/export_writer.cpp create mode 100644 src/plugins/manus/calibration/core/inc/manus_calibration/engine.hpp create mode 100644 src/plugins/manus/calibration/core/inc/manus_calibration/export_writer.hpp create mode 100644 src/plugins/manus/calibration/core/inc/manus_calibration/protocol.hpp create mode 100644 src/plugins/manus/calibration/core/inc/manus_calibration/sha256.hpp create mode 100644 src/plugins/manus/calibration/core/json.cpp create mode 100644 src/plugins/manus/calibration/core/json.hpp create mode 100644 src/plugins/manus/calibration/core/protocol.cpp create mode 100644 src/plugins/manus/calibration/core/sha256.cpp create mode 100644 tests/cpp/plugins/manus_calibration/CMakeLists.txt create mode 100644 tests/cpp/plugins/manus_calibration/fake_sdk.hpp create mode 100644 tests/cpp/plugins/manus_calibration/test_engine.cpp create mode 100644 tests/cpp/plugins/manus_calibration/test_export_writer.cpp create mode 100644 tests/cpp/plugins/manus_calibration/test_protocol.cpp create mode 100644 tests/cpp/plugins/manus_calibration/test_sha256.cpp diff --git a/examples/manus_calibration/README.md b/examples/manus_calibration/README.md new file mode 100644 index 0000000000..efd040a2a3 --- /dev/null +++ b/examples/manus_calibration/README.md @@ -0,0 +1,107 @@ + + +# Manus glove calibration + +Calibrate Manus gloves from a browser and save the MCAL files on the machine the gloves are +plugged into. The page guides the operator through the steps the Manus SDK provides; the SDK work +happens in the `manus_glove_calibrator` executable (see +[`src/plugins/manus/calibration/README.md`](../../src/plugins/manus/calibration/README.md)). +No capture session, OpenXR runtime, or CaptureServer is involved. + +This is vendor glove calibration only. Camera calibration, spatial alignment, and flex-sensor +offsets are separate operations. + +## Prerequisites + +- The Manus SDK, with the Manus plugin built against it (`MANUS_SDK_ROOT`, see + `src/plugins/manus/install_manus.sh`), producing `manus_glove_calibrator`. +- A Manus license for the SDK's `Integrated` feature, as Manus requires. Glove discovery and the step + list worked without one in testing, but what the SDK does for a step start, finish and export without a + license has not been established (see Status). +- Gloves and dongle attached to this host, with `src/plugins/manus/70-manus-hid.rules` installed. +- **Every other Manus consumer stopped**: the tracking plugin, capture sessions, other SDK + clients. The dongles are not shared. + +## Run + +The page needs only Python 3.11 or newer and the standard library: + +```bash +cd examples/manus_calibration +PYTHONPATH=python python3 -m isaaccapture_examples.manus_calibration \ + --worker /path/to/manus_glove_calibrator \ + --out-dir ~/manus-calibration +``` + +Open `http://127.0.0.1:8080`. The page has no authentication and can drive hardware, so it binds to +loopback and refuses requests whose Host or Origin is not its own; to use it from another machine, +forward the port (`ssh -L 8080:127.0.0.1:8080 host`). + +| Option | Meaning | +| --- | --- | +| `--worker PATH` | The calibration worker executable. | +| `--out-dir DIR` | Where MCAL files are saved, on this host. A launch setting, never browser input. | +| `--sides {left,right,both}` | Gloves preselected in the page. | +| `--host`, `--port` | Where to serve. | +| `--discovery-timeout SECONDS` | How long the worker waits for the requested gloves. | + +## What the operator sees + +Select the gloves and start. For each SDK step the page shows the SDK's title and instruction. +Every prompt needs its own confirmation, including the first prompt of the second glove, and the +confirmation starts that step in the SDK. + +- A step with a positive SDK time starts a timed sampling window when confirmed, with a progress + bar showing elapsed time (not a quality score). +- A step with no time is confirmed once the operator is already in the pose. +- A step with a negative SDK time is a continuous step whose duration the SDK only estimates. The page + shows the estimate as guidance. The worker has no completion signal for it: confirming starts the SDK + step and the next prompt follows, so a negative time does not prove the SDK finished sampling. + +One browser owns an attempt, and it must have the page's event stream open to start one. Other +browsers see its progress but cannot confirm or cancel. If the owning tab disconnects, the attempt is +cancelled after a 3 second grace period, so a page reload does not cancel it; a reload in time keeps +ownership. + +## Unresponsive worker + +The supervisor bounds the answers only the worker can give: startup through the first prompt, the +acknowledgement of a confirmed step, and the time from the last step to the result. When the worker +stays silent past a bound the attempt ends with an uncertain outcome, nothing is published, and the +worker is terminated. This contains an unresponsive SDK call; it does not show that a license is +missing. A silent worker restarts the bound on any event it sends, so these are inactivity bounds, not +absolute limits. The time the operator takes to confirm a step is never limited. If a worker cannot be +stopped at all, new attempts are refused until the program restarts, because it may still hold the gloves. + +## Status + +Exercised in tests: the worker, the supervisor, storage and the page behave as described against a scripted +stand-in for the worker, and the real SDK discovers both gloves and returns the step lists. + +Not verified: a calibration run from the first step to saved files with real gloves. In testing, the SDK +call that starts a step did not return within 30 seconds, and the cause was not established. So an +exported file, its calibration quality, and loading it through the tracking plugin remain unverified. + +## Saved files + +A set is published only after every requested glove exported a non-empty file, the worker reported +success and exited cleanly, and each file matched the size and SHA-256 the worker reported. Files +land in `--out-dir` as `_.mcal`, and an atomically written manifest under +`--out-dir/manifests/` makes the set visible. A cancelled, failed, or uncertain attempt publishes +nothing, and an existing calibration is never overwritten. Selecting a saved file for tracking is +done with `manus_hand_plugin --left-calibration=FILE --right-calibration=FILE`. + +## Tests + +```bash +cd examples/manus_calibration +python3 -m pip install pytest +PYTHONPATH=python python3 -m pytest +``` + +The tests drive the adapter, controller and HTTP server against a scripted stand-in for the worker, so +they establish the supervision, storage, ownership and web-safety logic, not that the SDK produces a +valid MCAL file. diff --git a/examples/manus_calibration/pyproject.toml b/examples/manus_calibration/pyproject.toml new file mode 100644 index 0000000000..6bb337f907 --- /dev/null +++ b/examples/manus_calibration/pyproject.toml @@ -0,0 +1,34 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +# The dist name mirrors the import path, so an installed example claims no bare +# top-level name in site-packages. +name = "isaaccapture-examples-manus-calibration" +version = "0.0.0" # Internal example - not versioned +description = "Standalone Manus glove calibration with a browser page" +requires-python = ">=3.11,<3.14" +# Standard library only: calibration needs no OpenXR, CUDA, capture session or UI toolkit. +dependencies = [] + +[project.optional-dependencies] +dev = [ + "pytest>=8", +] + +# `only-include` + `sources`, not `packages`, and no __init__.py above +# manus_calibration -- see the packaging rules in examples/README.md. +[tool.hatch.build] +only-include = ["python/isaaccapture_examples/manus_calibration"] + +[tool.hatch.build.targets.wheel] +sources = ["python"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +# Run from a source tree without installing the example. +pythonpath = ["python"] diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/__init__.py b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/__init__.py new file mode 100644 index 0000000000..3507e90c7f --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/__init__.py @@ -0,0 +1,4 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Standalone Manus glove calibration: worker supervision, MCAL storage and a browser page.""" diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/__main__.py b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/__main__.py new file mode 100644 index 0000000000..b87ce998e5 --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/__main__.py @@ -0,0 +1,108 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Command line for the standalone Manus calibration page.""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +import os +import signal +import sys +from pathlib import Path + +from .server import PageServer +from .store import McalStore +from .worker import CalibrationWorker + +logger = logging.getLogger("isaaccapture.examples.manus_calibration") + + +def build_parser() -> argparse.ArgumentParser: + """Arguments for the page; the output directory is a launch setting, never browser input.""" + parser = argparse.ArgumentParser( + prog="python -m isaaccapture_examples.manus_calibration", + description="Calibrate Manus gloves from a browser and save the MCAL files on this host.", + ) + parser.add_argument( + "--worker", + required=True, + type=Path, + help="path to the manus_glove_calibrator executable", + ) + parser.add_argument( + "--out-dir", required=True, type=Path, help="calibration directory on this host" + ) + parser.add_argument( + "--sides", + choices=("left", "right", "both"), + default="both", + help="gloves preselected in the page (default: both)", + ) + parser.add_argument( + "--host", + default="127.0.0.1", + help="interface to serve on; the page has no authentication, so keep loopback and use SSH port forwarding", + ) + parser.add_argument( + "--port", type=int, default=8080, help="port to serve on (default: 8080)" + ) + parser.add_argument( + "--discovery-timeout", + type=float, + default=15.0, + help="seconds the worker waits for the requested gloves (default: 15)", + ) + return parser + + +def _validate(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: + """Fail early, with a usage error, on a worker path that cannot run.""" + if not args.worker.is_file() or not os.access(args.worker, os.X_OK): + parser.error(f"--worker {args.worker} is not an executable file") + if not 0 <= args.port < 65536: + parser.error("--port must be between 0 and 65535") + if args.discovery_timeout <= 0: + parser.error("--discovery-timeout must be positive") + + +async def _serve(args: argparse.Namespace) -> int: + """Serve until interrupted, then cancel any active attempt and wait for the worker to settle.""" + store = McalStore(args.out_dir) + worker = CalibrationWorker( + args.worker, store, discovery_timeout_s=args.discovery_timeout + ) + server = PageServer(worker, store.directory, args.host, args.port, args.sides) + await server.start() + logger.info( + "serving on http://%s:%d, calibration files go to %s", + args.host, + server.port, + store.directory, + ) + + stop = asyncio.Event() + loop = asyncio.get_running_loop() + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, stop.set) + await stop.wait() + logger.info("shutting down; cancelling any active calibration") + await server.aclose() + return 0 + + +def main(argv: list[str] | None = None) -> int: + """Entry point.""" + parser = build_parser() + args = parser.parse_args(argv) + _validate(args, parser) + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s" + ) + return asyncio.run(_serve(args)) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/controller.py b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/controller.py new file mode 100644 index 0000000000..e093ad3fff --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/controller.py @@ -0,0 +1,235 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Presentation state and ownership rules for one calibration at a time, with no UI toolkit. + +Every method runs on the asyncio loop that owns the worker. A UI renders +:meth:`CalibrationController.view`; nothing here (or in the UI) calls the SDK. +""" + +from __future__ import annotations + +import asyncio +import contextlib +from collections.abc import Callable, Sequence +from dataclasses import dataclass, replace +from enum import Enum + +from . import protocol as proto +from .worker import ( + AttemptError, + CalibrationAttempt, + CalibrationOutcome, + CalibrationWorker, + OutcomeKind, +) + + +class Stage(str, Enum): + """Coarse stage the page renders.""" + + IDLE = "idle" + STARTING = "starting" + DISCOVERING = "discovering" + AWAITING_CONFIRMATION = "awaiting_confirmation" + SAMPLING = "sampling" + SAVING = "saving" + STOPPING = "stopping" + FINISHED = "finished" + + +@dataclass(frozen=True) +class View: + """Immutable snapshot of everything the page shows.""" + + stage: Stage = Stage.IDLE + owner: str | None = None + sides: tuple[proto.Side, ...] = () + gloves: tuple[proto.Glove, ...] = () + prompt: proto.Prompt | None = None + confirm_pending: bool = False + elapsed_s: float | None = None + notice: str = "" + outcome: CalibrationOutcome | None = None + + +class NotOwner(Exception): + """The caller is not the client that owns the active attempt.""" + + +class CalibrationController: + """Owns the single attempt, its owner, and the view derived from worker events.""" + + def __init__( + self, + worker: CalibrationWorker, + on_change: Callable[[View], None] | None = None, + ) -> None: + self._worker = worker + self._on_change = on_change + self._view = View() + self._attempt: CalibrationAttempt | None = None + self._watcher: asyncio.Task[None] | None = None + + def view(self) -> View: + return self._view + + @property + def active(self) -> bool: + """True from the start request until the terminal outcome is recorded.""" + return self._attempt is not None + + async def start(self, client_id: str, sides: Sequence[proto.Side]) -> None: + """Begin an attempt owned by ``client_id``; refused while another is active.""" + if self._attempt is not None: + raise AttemptError("a calibration attempt is already active") + try: + attempt = self._worker.new_attempt(self._on_event) + except AttemptError: + # The worker is still winding down a previous attempt. + self._notice("the previous attempt is still shutting down") + raise + self._attempt = attempt + self._set( + View( + stage=Stage.STARTING, + owner=client_id, + sides=tuple(s for s in proto.Side if s in sides), + ) + ) + self._watcher = asyncio.create_task(self._watch(attempt)) + try: + await attempt.start(sides) + except Exception: + # start() settles worker-side failures itself; a rejected request never reached the worker, so + # settle it here and wait for the watcher, leaving the controller free for the next start. + await attempt.cancel() + await self._watcher + raise + + async def confirm(self, client_id: str, token: str) -> None: + """Start the SDK step the browser is looking at; a click on a superseded prompt is refused.""" + attempt = self._require_owner(client_id) + prompt = self._view.prompt + if prompt is None or prompt.token != token: + raise AttemptError("that prompt is no longer current") + if self._view.confirm_pending: + return + self._set(replace(self._view, confirm_pending=True, notice="")) + try: + await attempt.confirm(prompt.token) + except AttemptError as exc: + self._notice(str(exc)) + finally: + if self._attempt is attempt and self._view.confirm_pending: + self._set(replace(self._view, confirm_pending=False)) + + async def cancel(self, client_id: str) -> None: + """Cancel the active attempt; only its owner may.""" + attempt = self._require_owner(client_id) + self._set(replace(self._view, stage=Stage.STOPPING, notice="cancelling")) + await attempt.cancel() + + def owner_attempt(self, client_id: str) -> CalibrationAttempt | None: + """The active attempt ``client_id`` owns, if any; lets a caller bind a later cancel to it.""" + if self._attempt is not None and self._view.owner == client_id: + return self._attempt + return None + + async def client_disconnected( + self, client_id: str, attempt: CalibrationAttempt | None = None + ) -> None: + """Owner loss cancels an active attempt; completed files are unaffected. + + With ``attempt`` given, only that attempt is cancelled, so a late timer cannot hit a newer one. + """ + current = self._attempt + if current is None or self._view.owner != client_id: + return + if attempt is not None and attempt is not current: + return + await current.cancel() + + async def aclose(self) -> None: + """Cancel any active attempt and wait for its outcome, for orderly shutdown.""" + attempt = self._attempt + if attempt is None: + return + await attempt.cancel() + if self._watcher is not None: + with contextlib.suppress(asyncio.CancelledError): + await self._watcher + + def _require_owner(self, client_id: str) -> CalibrationAttempt: + attempt = self._attempt + if attempt is None: + raise AttemptError("no calibration attempt is active") + if self._view.owner != client_id: + raise NotOwner( + "only the browser that started this calibration can control it" + ) + return attempt + + def _on_event(self, event: proto.WorkerEvent) -> None: + """Fold one worker event into the view; the outcome arrives through :meth:`_watch`.""" + view = self._view + if isinstance(event, proto.Status): + self._set( + replace(view, gloves=event.gloves, stage=_STAGE_OF_PHASE[event.phase]) + ) + elif isinstance(event, proto.Prompt): + self._set( + replace( + view, + stage=Stage.AWAITING_CONFIRMATION, + prompt=event, + confirm_pending=False, + elapsed_s=None, + notice="", + ) + ) + elif isinstance(event, proto.StepAccepted): + self._set(replace(view, stage=Stage.SAMPLING)) + elif isinstance(event, proto.StepProgress): + self._set(replace(view, elapsed_s=event.elapsed_s)) + + async def _watch(self, attempt: CalibrationAttempt) -> None: + """Record the terminal outcome and release ownership so a new attempt can start.""" + outcome = await attempt.outcome() + self._attempt = None + self._set( + View( + stage=Stage.FINISHED, + sides=outcome.sides, + outcome=outcome, + notice=_summary(outcome), + ) + ) + + def _notice(self, text: str) -> None: + self._set(replace(self._view, notice=text)) + + def _set(self, view: View) -> None: + self._view = view + if self._on_change is not None: + self._on_change(view) + + +_STAGE_OF_PHASE = { + proto.Phase.DISCOVERING: Stage.DISCOVERING, + proto.Phase.AWAITING_STEP: Stage.AWAITING_CONFIRMATION, + proto.Phase.CAPTURING_STEP: Stage.SAMPLING, + proto.Phase.SAVING: Stage.SAVING, + proto.Phase.STOPPING: Stage.STOPPING, +} + + +def _summary(outcome: CalibrationOutcome) -> str: + """One line for the page; an uncertain outcome is never described as success.""" + if outcome.kind is OutcomeKind.COMPLETED: + return "calibration completed" + if outcome.kind is OutcomeKind.CANCELLED: + return "calibration cancelled; no files were published" + if outcome.kind is OutcomeKind.UNCERTAIN: + return f"outcome uncertain, no files were published: {outcome.message}" + return f"calibration failed: {outcome.message}" diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/protocol.py b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/protocol.py new file mode 100644 index 0000000000..3acd55fd9f --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/protocol.py @@ -0,0 +1,523 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Typed events and commands of the calibration worker's wire protocol. + +The contract is `src/plugins/manus/calibration/README.md`. Parsing is strict: any line that is not +exactly a documented message raises :class:`ProtocolError`, so a misbehaving worker is a visible +failure rather than a skipped line. +""" + +from __future__ import annotations + +import json +import math +import re +from dataclasses import dataclass +from enum import Enum +from typing import Any + +PROTOCOL_VERSION = 1 +MAX_LINE_BYTES = 65536 + +_ATTEMPT_RE = re.compile(r"[A-Za-z0-9_-]{1,64}") +_SHA256_RE = re.compile(r"[0-9a-f]{64}") +_FILE_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}") + + +class ProtocolError(Exception): + """A worker line (or a command we were asked to build) violates the contract.""" + + +class Side(str, Enum): + LEFT = "left" + RIGHT = "right" + + +class Phase(str, Enum): + DISCOVERING = "discovering" + AWAITING_STEP = "awaiting_step" + CAPTURING_STEP = "capturing_step" + SAVING = "saving" + STOPPING = "stopping" + + +class StepMode(str, Enum): + HOLD = "hold" + MOTION = "motion" + + +class Outcome(str, Enum): + COMPLETED = "completed" + CANCELLED = "cancelled" + FAILED = "failed" + + +class RejectReason(str, Enum): + INVALID = "invalid" + UNKNOWN_ATTEMPT = "unknown_attempt" + STALE_TOKEN = "stale_token" + UNEXPECTED = "unexpected" + BUSY = "busy" + + +_ERROR_CODES = frozenset( + { + "sdk_init_failed", + "no_gloves", + "glove_missing", + "glove_replaced", + "sdk_error", + "step_failed", + "export_failed", + "protocol_error", + "internal", + } +) + + +@dataclass(frozen=True) +class Glove: + side: Side + glove_id: int + + +@dataclass(frozen=True) +class ExportedFile: + """One MCAL file as the worker reports it, before the supervisor has verified it.""" + + side: Side + file: str + bytes: int + sha256: str + + +@dataclass(frozen=True) +class WorkerError: + code: str + message: str + side: Side | None = None + step: int | None = None + + +@dataclass(frozen=True) +class Ready: + worker_version: str + sdk_version: str | None + + +@dataclass(frozen=True) +class Status: + attempt: str + phase: Phase + gloves: tuple[Glove, ...] + + +@dataclass(frozen=True) +class Prompt: + attempt: str + token: str + side: Side + glove_id: int + index: int + count: int + title: str + instruction: str + mode: StepMode + duration_s: float + # A continuous SDK step: operator-paced like a hold, with the SDK's estimate of its length as guidance. + continuous: bool = False + estimated_s: float = 0.0 + + +@dataclass(frozen=True) +class StepAccepted: + attempt: str + token: str + + +@dataclass(frozen=True) +class StepProgress: + attempt: str + token: str + elapsed_s: float + + +@dataclass(frozen=True) +class Rejected: + attempt: str | None + command: str + reason: RejectReason + message: str + token: str | None = None + + +@dataclass(frozen=True) +class Exported: + attempt: str + side: Side + glove_id: int + file: str + bytes: int + sha256: str + + +@dataclass(frozen=True) +class Result: + attempt: str | None + outcome: Outcome + files: tuple[ExportedFile, ...] = () + error: WorkerError | None = None + + +WorkerEvent = ( + Ready | Status | Prompt | StepAccepted | StepProgress | Rejected | Exported | Result +) + + +def _fail(context: str, problem: str) -> ProtocolError: + return ProtocolError(f"{context}: {problem}") + + +def _keys( + data: dict[str, Any], required: set[str], optional: set[str], context: str +) -> None: + present = set(data) + missing = required - present + unknown = present - required - optional + if missing: + raise _fail(context, f"missing key(s) {sorted(missing)}") + if unknown: + raise _fail(context, f"unknown key(s) {sorted(unknown)}") + + +def _str( + data: dict[str, Any], key: str, context: str, *, nonempty: bool = False +) -> str: + value = data[key] + if not isinstance(value, str) or (nonempty and not value): + raise _fail( + context, f"{key} must be {'a nonempty ' if nonempty else 'a '}string" + ) + return value + + +def _opt_str(data: dict[str, Any], key: str, context: str) -> str | None: + return None if data[key] is None else _str(data, key, context) + + +def _uint(data: dict[str, Any], key: str, context: str) -> int: + value = data[key] + # bool is an int subclass; a JSON boolean must not pass as a count. + if type(value) is not int or value < 0: + raise _fail(context, f"{key} must be a nonnegative integer") + return value + + +def _seconds(data: dict[str, Any], key: str, context: str) -> float: + value = data[key] + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value < 0 + ): + raise _fail(context, f"{key} must be a finite nonnegative number") + return float(value) + + +def _enum(enum_cls: type[Enum], value: Any, context: str) -> Any: + try: + return enum_cls(value) + except ValueError: + raise _fail( + context, f"{value!r} is not one of {[m.value for m in enum_cls]}" + ) from None + + +def _attempt( + data: dict[str, Any], context: str, *, nullable: bool = False +) -> str | None: + value = data["attempt"] + if value is None and nullable: + return None + if not isinstance(value, str) or not _ATTEMPT_RE.fullmatch(value): + raise _fail(context, "attempt must be 1-64 characters of [A-Za-z0-9_-]") + return value + + +def _file_name(value: Any, context: str) -> str: + # Basename only: the supervisor joins this to its own staging directory. + if not isinstance(value, str) or not _FILE_RE.fullmatch(value) or ".." in value: + raise _fail(context, f"file must be a plain basename, got {value!r}") + return value + + +def _sha(value: Any, context: str) -> str: + if not isinstance(value, str) or not _SHA256_RE.fullmatch(value): + raise _fail(context, "sha256 must be 64 lowercase hex characters") + return value + + +def _gloves(value: Any, context: str) -> tuple[Glove, ...]: + if not isinstance(value, list): + raise _fail(context, "gloves must be an array") + gloves = [] + for item in value: + if not isinstance(item, dict): + raise _fail(context, "each glove must be an object") + _keys(item, {"side", "glove_id"}, set(), f"{context}.gloves[]") + gloves.append( + Glove( + _enum(Side, item["side"], f"{context}.gloves[].side"), + _uint(item, "glove_id", f"{context}.gloves[]"), + ) + ) + sides = [g.side for g in gloves] + if len(sides) != len(set(sides)): + raise _fail(context, "gloves lists a side twice") + return tuple(gloves) + + +def _exported_files(value: Any, context: str) -> tuple[ExportedFile, ...]: + if not isinstance(value, list): + raise _fail(context, "files must be an array") + files = [] + for item in value: + if not isinstance(item, dict): + raise _fail(context, "each file must be an object") + _keys(item, {"side", "file", "bytes", "sha256"}, set(), f"{context}.files[]") + files.append( + ExportedFile( + side=_enum(Side, item["side"], f"{context}.files[].side"), + file=_file_name(item["file"], f"{context}.files[]"), + bytes=_uint(item, "bytes", f"{context}.files[]"), + sha256=_sha(item["sha256"], f"{context}.files[]"), + ) + ) + return tuple(files) + + +def _error(value: Any, context: str) -> WorkerError: + if not isinstance(value, dict): + raise _fail(context, "error must be an object") + _keys(value, {"code", "message"}, {"side", "step"}, f"{context}.error") + code = value["code"] + if code not in _ERROR_CODES: + raise _fail(context, f"unknown error code {code!r}") + side = value.get("side") + step = value.get("step") + if step is not None and (type(step) is not int or step < 0): + raise _fail(context, "error.step must be a nonnegative integer") + return WorkerError( + code=code, + message=_str(value, "message", f"{context}.error"), + side=None if side is None else _enum(Side, side, f"{context}.error.side"), + step=step, + ) + + +def _parse_ready(d: dict[str, Any]) -> Ready: + _keys(d, {"v", "type", "worker_version", "sdk_version"}, set(), "ready") + return Ready( + _str(d, "worker_version", "ready"), _opt_str(d, "sdk_version", "ready") + ) + + +def _parse_status(d: dict[str, Any]) -> Status: + _keys(d, {"v", "type", "attempt", "phase", "gloves"}, set(), "status") + return Status( + _attempt(d, "status"), + _enum(Phase, d["phase"], "status.phase"), + _gloves(d["gloves"], "status"), + ) + + +def _parse_prompt(d: dict[str, Any]) -> Prompt: + keys = { + "v", + "type", + "attempt", + "token", + "side", + "glove_id", + "index", + "count", + "title", + "instruction", + "mode", + "duration_s", + "continuous", + "estimated_s", + } + _keys(d, keys, set(), "prompt") + count = _uint(d, "count", "prompt") + index = _uint(d, "index", "prompt") + if count == 0 or index >= count: + raise _fail("prompt", "index must be below a positive count") + mode = _enum(StepMode, d["mode"], "prompt.mode") + duration = _seconds(d, "duration_s", "prompt") + # Hold steps carry no sampling window; motion steps must have one. + if (mode is StepMode.HOLD) != (duration == 0): + raise _fail("prompt", "duration_s must be 0 for hold and positive for motion") + continuous = d["continuous"] + if type(continuous) is not bool: + raise _fail("prompt", "continuous must be a boolean") + estimated = _seconds(d, "estimated_s", "prompt") + # Only a hold-paced continuous step carries an estimate; a timed motion already has its window. + if continuous and (mode is not StepMode.HOLD or estimated == 0): + raise _fail("prompt", "a continuous step is a hold with a positive estimated_s") + if not continuous and estimated != 0: + raise _fail("prompt", "estimated_s must be 0 unless the step is continuous") + return Prompt( + attempt=_attempt(d, "prompt"), + token=_str(d, "token", "prompt", nonempty=True), + side=_enum(Side, d["side"], "prompt.side"), + glove_id=_uint(d, "glove_id", "prompt"), + index=index, + count=count, + title=_str(d, "title", "prompt"), + instruction=_str(d, "instruction", "prompt", nonempty=True), + mode=mode, + duration_s=duration, + continuous=continuous, + estimated_s=estimated, + ) + + +def _parse_step_accepted(d: dict[str, Any]) -> StepAccepted: + _keys(d, {"v", "type", "attempt", "token"}, set(), "step_accepted") + return StepAccepted( + _attempt(d, "step_accepted"), _str(d, "token", "step_accepted", nonempty=True) + ) + + +def _parse_step_progress(d: dict[str, Any]) -> StepProgress: + _keys(d, {"v", "type", "attempt", "token", "elapsed_s"}, set(), "step_progress") + return StepProgress( + _attempt(d, "step_progress"), + _str(d, "token", "step_progress", nonempty=True), + _seconds(d, "elapsed_s", "step_progress"), + ) + + +def _parse_rejected(d: dict[str, Any]) -> Rejected: + _keys( + d, + {"v", "type", "attempt", "command", "reason", "message"}, + {"token"}, + "rejected", + ) + token = d.get("token") + if token is not None and (not isinstance(token, str) or not token): + raise _fail("rejected", "token must be a nonempty string") + return Rejected( + attempt=_attempt(d, "rejected", nullable=True), + command=_str(d, "command", "rejected", nonempty=True), + reason=_enum(RejectReason, d["reason"], "rejected.reason"), + message=_str(d, "message", "rejected"), + token=token, + ) + + +def _parse_exported(d: dict[str, Any]) -> Exported: + _keys( + d, + {"v", "type", "attempt", "side", "glove_id", "file", "bytes", "sha256"}, + set(), + "exported", + ) + return Exported( + attempt=_attempt(d, "exported"), + side=_enum(Side, d["side"], "exported.side"), + glove_id=_uint(d, "glove_id", "exported"), + file=_file_name(d["file"], "exported"), + bytes=_uint(d, "bytes", "exported"), + sha256=_sha(d["sha256"], "exported"), + ) + + +def _parse_result(d: dict[str, Any]) -> Result: + _keys(d, {"v", "type", "attempt", "outcome"}, {"files", "error"}, "result") + outcome = _enum(Outcome, d["outcome"], "result.outcome") + has_files = "files" in d + has_error = "error" in d + # `files` belongs to completed results only and `error` to failed ones. + if has_files != (outcome is Outcome.COMPLETED): + raise _fail("result", "files must be present exactly for a completed result") + if has_error != (outcome is Outcome.FAILED): + raise _fail("result", "error must be present exactly for a failed result") + return Result( + attempt=_attempt(d, "result", nullable=True), + outcome=outcome, + files=_exported_files(d["files"], "result") if has_files else (), + error=_error(d["error"], "result") if has_error else None, + ) + + +_PARSERS = { + "ready": _parse_ready, + "status": _parse_status, + "prompt": _parse_prompt, + "step_accepted": _parse_step_accepted, + "step_progress": _parse_step_progress, + "rejected": _parse_rejected, + "exported": _parse_exported, + "result": _parse_result, +} + + +def parse_event(line: bytes | str) -> WorkerEvent: + """Parse one worker line into its typed event, or raise :class:`ProtocolError`.""" + if isinstance(line, bytes): + if len(line) > MAX_LINE_BYTES: + raise ProtocolError(f"line exceeds {MAX_LINE_BYTES} bytes") + try: + line = line.decode("utf-8") + except UnicodeDecodeError as exc: + raise ProtocolError(f"line is not UTF-8: {exc}") from None + try: + data = json.loads(line) + except ValueError as exc: + raise ProtocolError(f"line is not JSON: {exc}") from None + if not isinstance(data, dict): + raise ProtocolError("line must be a JSON object") + if type(data.get("v")) is not int or data["v"] != PROTOCOL_VERSION: + raise ProtocolError(f"expected protocol version {PROTOCOL_VERSION}") + parser = _PARSERS.get(data.get("type")) + if parser is None: + raise ProtocolError(f"unknown message type {data.get('type')!r}") + return parser(data) + + +def encode_start(attempt: str, sides: tuple[Side, ...], staging_dir: str) -> bytes: + """Build the `start` command line.""" + if not _ATTEMPT_RE.fullmatch(attempt): + raise ProtocolError("attempt must be 1-64 characters of [A-Za-z0-9_-]") + if not sides or len(set(sides)) != len(sides): + raise ProtocolError("sides must be a nonempty set of unique sides") + # The worker processes left before right regardless of request order. + ordered = [s.value for s in Side if s in sides] + return _line( + { + "v": PROTOCOL_VERSION, + "type": "start", + "attempt": attempt, + "sides": ordered, + "staging_dir": staging_dir, + } + ) + + +def encode_confirm(attempt: str, token: str) -> bytes: + return _line( + {"v": PROTOCOL_VERSION, "type": "confirm", "attempt": attempt, "token": token} + ) + + +def encode_cancel(attempt: str) -> bytes: + return _line({"v": PROTOCOL_VERSION, "type": "cancel", "attempt": attempt}) + + +def _line(message: dict[str, Any]) -> bytes: + return (json.dumps(message, separators=(",", ":")) + "\n").encode("utf-8") diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/server.py b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/server.py new file mode 100644 index 0000000000..63649a88aa --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/server.py @@ -0,0 +1,409 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""A small HTTP server for the calibration page, using only the standard library. + +Static files and a server-sent-event stream per browser tab carry the controller's view out; three +POST endpoints carry the operator's actions in. The page can drive hardware and has no login, so +every request is checked for the page's own origin and every action must carry a per-tab client id. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import logging +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any +from urllib.parse import parse_qs, urlsplit + +from . import protocol as proto +from .controller import CalibrationController, NotOwner, View +from .wire import view_to_dict +from .worker import AttemptError, CalibrationWorker + +logger = logging.getLogger("isaaccapture.examples.manus_calibration.server") + +_STATIC_DIR = Path(__file__).resolve().parent / "static" +_STATIC_TYPES = { + "/": ("index.html", "text/html; charset=utf-8"), + "/app.js": ("app.js", "text/javascript; charset=utf-8"), + "/style.css": ("style.css", "text/css; charset=utf-8"), +} +_CLIENT_ID_RE = re.compile(r"[A-Za-z0-9_-]{8,64}") +_MAX_HEADER_BYTES = 16 * 1024 +_MAX_BODY_BYTES = 4 * 1024 +_REQUEST_TIMEOUT_S = 10.0 +_KEEPALIVE_S = 15.0 +# A reload closes the old stream before the new one opens; this window keeps it from cancelling. +_DISCONNECT_GRACE_S = 3.0 +_LOOPBACK_NAMES = ("127.0.0.1", "localhost", "[::1]") + +_SECURITY_HEADERS = { + "Content-Security-Policy": ( + "default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; " + "base-uri 'none'; form-action 'none'; frame-ancestors 'none'" + ), + "X-Content-Type-Options": "nosniff", + "Referrer-Policy": "no-referrer", + "Cache-Control": "no-store", +} +_REASONS = { + 200: "OK", + 400: "Bad Request", + 403: "Forbidden", + 404: "Not Found", + 405: "Method Not Allowed", + 409: "Conflict", + 413: "Payload Too Large", + 500: "Internal Server Error", +} + + +class _BadRequest(Exception): + """The request cannot be parsed or breaks a limit; carries the HTTP status to answer with.""" + + def __init__(self, status: int, message: str) -> None: + super().__init__(message) + self.status = status + + +@dataclass +class _Request: + method: str + path: str + query: dict[str, list[str]] + headers: dict[str, str] + body: bytes + + +@dataclass +class _Tab: + """One browser tab's server-sent-event state; its id is chosen by the page.""" + + connections: int = 0 + changed: asyncio.Event = field(default_factory=asyncio.Event) + grace: asyncio.TimerHandle | None = None + + +class PageServer: + """Serves the page and runs the controller on the caller's event loop.""" + + def __init__( + self, + worker: CalibrationWorker, + out_dir: Path, + host: str, + port: int, + default_sides: str = "both", + ) -> None: + self._out_dir = str(out_dir) + self._default_sides = default_sides + self._host = host + self._requested_port = port + self._controller = CalibrationController(worker, self._on_change) + self._tabs: dict[str, _Tab] = {} + # Strong references: the loop keeps only weak ones, and a dropped task would skip the cancel. + self._background: set[asyncio.Task[None]] = set() + self._server: asyncio.Server | None = None + self._allowed_hosts: set[str] = set() + self._static = { + path: (name, ctype, (_STATIC_DIR / name).read_bytes()) + for path, (name, ctype) in _STATIC_TYPES.items() + } + + @property + def controller(self) -> CalibrationController: + return self._controller + + @property + def port(self) -> int: + """The bound port; differs from the requested one when that was 0.""" + assert self._server is not None + return self._server.sockets[0].getsockname()[1] + + async def start(self) -> None: + """Begin listening.""" + self._server = await asyncio.start_server( + self._handle, self._host, self._requested_port + ) + names = ( + set(_LOOPBACK_NAMES) + if self._host in ("127.0.0.1", "localhost", "::1") + else {self._host} + ) + self._allowed_hosts = {f"{name}:{self.port}" for name in names} + + async def aclose(self) -> None: + """Cancel any active attempt, wait for the worker to settle, then stop listening.""" + await self._controller.aclose() + if self._server is not None: + self._server.close() + for tab in self._tabs.values(): + tab.changed.set() + await self._server.wait_closed() + + def _on_change(self, _view: View) -> None: + """Wake every open stream; each sends the controller's current view, so bursts coalesce.""" + for tab in self._tabs.values(): + tab.changed.set() + + async def _handle( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + """One connection: parse one request, answer it, close. A stream request stays open.""" + try: + try: + request = await asyncio.wait_for( + self._read_request(reader), _REQUEST_TIMEOUT_S + ) + self._check_origin(request) + await self._route(request, reader, writer) + except _BadRequest as exc: + await self._reply_json( + writer, exc.status, {"ok": False, "error": str(exc)} + ) + except (asyncio.TimeoutError, ConnectionError): + pass + except Exception: + # Detail stays in the log; the browser gets no stack trace. + logger.exception("request failed") + with contextlib.suppress(Exception): + await self._reply_json( + writer, 500, {"ok": False, "error": "internal error"} + ) + finally: + with contextlib.suppress(Exception): + writer.close() + await writer.wait_closed() + + async def _read_request(self, reader: asyncio.StreamReader) -> _Request: + """Parse the request line, headers and a bounded body.""" + try: + block = await reader.readuntil(b"\r\n\r\n") + except asyncio.LimitOverrunError: + raise _BadRequest(413, "request headers too large") from None + except asyncio.IncompleteReadError: + raise ConnectionError from None + if len(block) > _MAX_HEADER_BYTES: + raise _BadRequest(413, "request headers too large") + lines = block.decode("latin-1").split("\r\n") + parts = lines[0].split(" ") + if len(parts) != 3 or not parts[2].startswith("HTTP/1."): + raise _BadRequest(400, "malformed request line") + headers: dict[str, str] = {} + for line in lines[1:]: + if not line: + continue + name, sep, value = line.partition(":") + if not sep: + raise _BadRequest(400, "malformed header") + headers[name.strip().lower()] = value.strip() + length_text = headers.get("content-length", "0") + if not length_text.isdigit() or int(length_text) > _MAX_BODY_BYTES: + raise _BadRequest(413, "request body too large") + try: + body = await reader.readexactly(int(length_text)) + except asyncio.IncompleteReadError: + raise ConnectionError from None + split = urlsplit(parts[1]) + return _Request(parts[0], split.path, parse_qs(split.query), headers, body) + + def _check_origin(self, request: _Request) -> None: + """Refuse a foreign Host (DNS rebinding) or a cross-origin page; a missing Origin is a same-origin GET.""" + host = request.headers.get("host", "") + if host not in self._allowed_hosts: + raise _BadRequest(403, "unexpected Host header") + origin = request.headers.get("origin") + if origin is not None and origin != f"http://{host}": + raise _BadRequest(403, "cross-origin request refused") + + async def _route( + self, + request: _Request, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + ) -> None: + """Dispatch on method and path.""" + if request.method == "GET" and request.path in self._static: + _name, ctype, body = self._static[request.path] + await self._reply(writer, 200, ctype, body) + elif request.method == "GET" and request.path == "/events": + await self._stream(request, reader, writer) + elif request.method == "POST" and request.path in ( + "/api/start", + "/api/confirm", + "/api/cancel", + ): + await self._action(request, writer) + elif request.path in self._static or request.path in ( + "/events", + "/api/start", + "/api/confirm", + "/api/cancel", + ): + raise _BadRequest(405, "method not allowed") + else: + raise _BadRequest(404, "not found") + + async def _action(self, request: _Request, writer: asyncio.StreamWriter) -> None: + """An operator action; the custom header and JSON type make a cross-site form post impossible.""" + client = request.headers.get("x-client-id", "") + if not _CLIENT_ID_RE.fullmatch(client): + raise _BadRequest(400, "missing or invalid X-Client-Id") + if ( + request.headers.get("content-type", "").split(";")[0].strip() + != "application/json" + ): + raise _BadRequest(400, "content type must be application/json") + try: + data: Any = json.loads(request.body or b"{}") + except ValueError: + raise _BadRequest(400, "body is not valid JSON") from None + if not isinstance(data, dict): + raise _BadRequest(400, "body must be a JSON object") + try: + if request.path == "/api/start": + # Without a live stream no disconnect could ever cancel the attempt this tab would own. + tab = self._tabs.get(client) + if tab is None or tab.connections == 0: + raise _BadRequest(409, "the page's event stream is not connected") + await self._controller.start(client, _sides_of(data)) + elif request.path == "/api/confirm": + token = data.get("token") + if not isinstance(token, str) or not token: + raise _BadRequest(400, "token is required") + await self._controller.confirm(client, token) + else: + await self._controller.cancel(client) + except NotOwner as exc: + raise _BadRequest(403, str(exc)) from None + except AttemptError as exc: + raise _BadRequest(409, str(exc)) from None + except proto.ProtocolError as exc: + raise _BadRequest(400, str(exc)) from None + await self._reply_json(writer, 200, {"ok": True}) + + async def _stream( + self, + request: _Request, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + ) -> None: + """Server-sent events for one tab until it disconnects or the server closes.""" + ids = request.query.get("client", []) + if len(ids) != 1 or not _CLIENT_ID_RE.fullmatch(ids[0]): + raise _BadRequest(400, "missing or invalid client id") + client = ids[0] + tab = self._tabs.setdefault(client, _Tab()) + if tab.grace is not None: + tab.grace.cancel() + tab.grace = None + tab.connections += 1 + head = { + "Content-Type": "text/event-stream", + "Connection": "close", + **_SECURITY_HEADERS, + } + writer.write(_head(200, head)) + # EOF on the request side is the only portable sign that the tab went away. + closed = asyncio.ensure_future(reader.read()) + try: + while self._server is not None and self._server.is_serving(): + payload = json.dumps( + view_to_dict( + self._controller.view(), + client, + self._out_dir, + self._default_sides, + ) + ) + writer.write(f"data: {payload}\n\n".encode()) + await writer.drain() + tab.changed.clear() + waiter = asyncio.ensure_future(tab.changed.wait()) + done, _pending = await asyncio.wait( + {waiter, closed}, + timeout=_KEEPALIVE_S, + return_when=asyncio.FIRST_COMPLETED, + ) + waiter.cancel() + if closed in done: + break + if not done: + writer.write(b": keepalive\n\n") + await writer.drain() + except ConnectionError: + pass + finally: + closed.cancel() + tab.connections -= 1 + if tab.connections == 0: + self._schedule_disconnect(client, tab) + + def _schedule_disconnect(self, client: str, tab: _Tab) -> None: + """Treat the tab as gone only if it does not come back within the grace window.""" + loop = asyncio.get_running_loop() + # Bound now to the attempt this tab owns, so a late timer can never cancel a newer attempt. + attempt = self._controller.owner_attempt(client) + + def _gone() -> None: + tab.grace = None + if tab.connections != 0: + return + self._tabs.pop(client, None) + if attempt is None: + return + task = loop.create_task( + self._controller.client_disconnected(client, attempt) + ) + self._background.add(task) + task.add_done_callback(self._background.discard) + + tab.grace = loop.call_later(_DISCONNECT_GRACE_S, _gone) + + async def _reply( + self, writer: asyncio.StreamWriter, status: int, ctype: str, body: bytes + ) -> None: + """A complete response with the security headers.""" + headers = { + "Content-Type": ctype, + "Content-Length": str(len(body)), + "Connection": "close", + **_SECURITY_HEADERS, + } + writer.write(_head(status, headers) + body) + await writer.drain() + + async def _reply_json( + self, writer: asyncio.StreamWriter, status: int, data: dict[str, Any] + ) -> None: + await self._reply(writer, status, "application/json", json.dumps(data).encode()) + + +def _head(status: int, headers: dict[str, str]) -> bytes: + """Status line and headers.""" + lines = [f"HTTP/1.1 {status} {_REASONS.get(status, 'Status')}"] + lines += [f"{name}: {value}" for name, value in headers.items()] + return ("\r\n".join(lines) + "\r\n\r\n").encode("latin-1") + + +def _sides_of(data: dict[str, Any]) -> list[proto.Side]: + """The sides a start request names; anything but a nonempty list of left/right is a bad request.""" + raw = data.get("sides") + if ( + not isinstance(raw, list) + or not raw + or not all(isinstance(item, str) for item in raw) + ): + raise _BadRequest(400, "sides must be a nonempty list") + try: + sides = [proto.Side(item) for item in raw] + except ValueError: + raise _BadRequest(400, "sides must be left or right") from None + if len(set(sides)) != len(sides): + raise _BadRequest(400, "sides must not repeat") + return sides diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/app.js b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/app.js new file mode 100644 index 0000000000..5bc8ec9ed8 --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/app.js @@ -0,0 +1,207 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +"use strict"; + +// Every piece of text from the server is assigned with textContent, never as markup. + +const STAGE_TEXT = { + idle: "Ready.", + starting: "Starting the calibration worker...", + discovering: "Looking for the requested gloves...", + awaiting_confirmation: "Waiting for your confirmation.", + sampling: "The SDK is sampling. Follow the instruction below.", + saving: "Saving the calibration files...", + stopping: "Stopping...", + finished: "Finished.", +}; +const RESULT_HEADING = { + completed: "Calibration saved", + cancelled: "Calibration cancelled", + failed: "Calibration failed", + uncertain: "Outcome uncertain", +}; + +const $ = (id) => document.getElementById(id); + +// One id per browser tab: it names the tab as the owner of an attempt, and survives a reload. +const clientId = (() => { + let id = sessionStorage.getItem("manusCalibrationClient"); + if (!id) { + const bytes = crypto.getRandomValues(new Uint8Array(16)); + id = Array.from(bytes, (b) => b.toString(16).padStart(2, "0")).join(""); + sessionStorage.setItem("manusCalibrationClient", id); + } + return id; +})(); + +let state = null; +let requestInFlight = false; +let cancelInFlight = false; +let localError = ""; + +async function post(path, body) { + const isCancel = path === "/api/cancel"; + requestInFlight = !isCancel || requestInFlight; + cancelInFlight = isCancel || cancelInFlight; + localError = ""; + render(); + try { + const response = await fetch(path, { + method: "POST", + headers: { "Content-Type": "application/json", "X-Client-Id": clientId }, + body: JSON.stringify(body), + }); + const result = await response.json(); + if (!result.ok) { + localError = result.error || "the request was refused"; + } + } catch (error) { + localError = "cannot reach the server"; + } finally { + if (isCancel) { + cancelInFlight = false; + } else { + requestInFlight = false; + } + render(); + } +} + +function selectedSides() { + const value = document.querySelector('input[name="sides"]:checked').value; + return value === "both" ? ["left", "right"] : [value]; +} + +function guidance(prompt) { + if (prompt.mode === "motion") { + return `Confirm, then perform the movement for ${prompt.duration_s.toFixed(1)} seconds.`; + } + if (prompt.continuous) { + return `Get into the pose (or start the movement) for about ${Math.round(prompt.estimated_s)} seconds, then confirm.`; + } + return "Hold the pose first, then confirm to capture it."; +} + +function confirmLabel(prompt) { + if (state.confirm_pending || requestInFlight) { + return "Starting..."; + } + return prompt.mode === "motion" ? `Start motion (${Math.round(prompt.duration_s)} s)` : "Capture pose"; +} + +function renderStatus() { + if (state.stage === "idle") { + $("stage").textContent = "Ready."; + } else if (state.active && !state.owner) { + $("stage").textContent = + "Another browser is running a calibration; this page is read-only until it finishes."; + } else { + $("stage").textContent = STAGE_TEXT[state.stage] || state.stage; + } + $("gloves").textContent = state.gloves.length + ? "Gloves found: " + state.gloves.map((g) => `${g.side} (id ${g.glove_id})`).join(", ") + : ""; + $("notice").textContent = localError || (state.stage === "finished" ? "" : state.notice); +} + +function renderStep() { + const prompt = state.prompt; + $("step").hidden = !prompt; + if (!prompt) { + return; + } + const side = prompt.side.charAt(0).toUpperCase() + prompt.side.slice(1); + $("step-heading").textContent = `${side} glove, step ${prompt.index + 1} of ${prompt.count}`; + $("step-title").textContent = prompt.title; + $("step-instruction").textContent = prompt.instruction; + $("step-guidance").textContent = guidance(prompt); + const sampling = state.stage === "sampling" && prompt.mode === "motion"; + $("progress").hidden = !sampling; + if (sampling && prompt.duration_s > 0 && state.elapsed_s !== null) { + $("progress").value = Math.min(100, (100 * state.elapsed_s) / prompt.duration_s); + } +} + +function renderControls() { + const prompt = state.prompt; + // Other browsers watch; only the owner gets controls. + $("start").disabled = state.active || requestInFlight; + for (const radio of document.querySelectorAll('input[name="sides"]')) { + radio.disabled = state.active; + } + $("actions").hidden = !(state.active && state.owner); + $("confirm").hidden = !prompt; + if (prompt) { + $("confirm").textContent = confirmLabel(prompt); + $("confirm").disabled = !( + state.stage === "awaiting_confirmation" && prompt.token && !state.confirm_pending && !requestInFlight + ); + } + // Cancel stays available while a confirm is in flight: that is when a stuck step needs it. + $("cancel").disabled = state.stage === "stopping" || cancelInFlight; +} + +function renderResult() { + const outcome = state.outcome; + $("result").hidden = !outcome || state.active; + if (!outcome || state.active) { + return; + } + $("result-heading").textContent = RESULT_HEADING[outcome.kind] || outcome.kind; + $("result-message").textContent = outcome.message; + $("result-dir").hidden = !outcome.directory; + $("result-dir").textContent = outcome.directory ? `Saved on the server host in ${outcome.directory}:` : ""; + const list = $("result-files"); + list.replaceChildren( + ...outcome.files.map((file) => { + const item = document.createElement("li"); + item.textContent = `${file.side}: ${file.file} (${file.bytes} bytes)`; + return item; + }), + ); + if (!outcome.directory) { + const none = document.createElement("li"); + none.textContent = "No calibration files were published."; + list.replaceChildren(none); + } + $("result-log").hidden = outcome.log_tail.length === 0; + $("result-log").textContent = outcome.log_tail.join("\n"); +} + +function render() { + if (!state) { + return; + } + $("out-dir").textContent = state.out_dir; + renderStatus(); + renderStep(); + renderControls(); + renderResult(); +} + +$("start").addEventListener("click", () => post("/api/start", { sides: selectedSides() })); +$("confirm").addEventListener("click", () => { + if (state && state.prompt && state.prompt.token) { + post("/api/confirm", { token: state.prompt.token }); + } +}); +$("cancel").addEventListener("click", () => post("/api/cancel", {})); + +const events = new EventSource(`/events?client=${clientId}`); +let defaultApplied = false; +events.onmessage = (event) => { + state = JSON.parse(event.data); + if (!defaultApplied) { + // The launch option preselects the gloves; the operator can still change it. + const radio = document.querySelector(`input[name="sides"][value="${state.default_sides}"]`); + if (radio) { + radio.checked = true; + } + defaultApplied = true; + } + render(); +}; +events.onerror = () => { + $("notice").textContent = "Connection lost; retrying..."; +}; diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/index.html b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/index.html new file mode 100644 index 0000000000..a4c269ae96 --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/index.html @@ -0,0 +1,60 @@ + + + + + + + Manus glove calibration + + + +
+

Manus glove calibration

+

+ Stop the capture session and every other Manus consumer first: the gloves are not shared. + Calibration files are saved on the server host in . +

+ +
+
+ Gloves + + + +
+ +
+ +
+

+

+ +
+ + + + + + +
+ + + diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/style.css b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/style.css new file mode 100644 index 0000000000..e051cf45bf --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/static/style.css @@ -0,0 +1,114 @@ +/* SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. */ +/* SPDX-License-Identifier: Apache-2.0 */ + +:root { + --bg: #fff; + --fg: #1b1f24; + --muted: #57606a; + --accent: #1a7f37; + --danger: #cf222e; + --border: #d0d7de; + color-scheme: light dark; +} + +@media (prefers-color-scheme: dark) { + :root { + --bg: #0d1117; + --fg: #e6edf3; + --muted: #9198a1; + --accent: #3fb950; + --danger: #f85149; + --border: #30363d; + } +} + +body { + margin: 0; + background: var(--bg); + color: var(--fg); + font: 16px/1.5 system-ui, sans-serif; +} + +main { + max-width: 40rem; + margin: 0 auto; + padding: 1.5rem 1rem 3rem; +} + +h1 { + font-size: 1.5rem; +} + +h2 { + font-size: 1.2rem; + margin-bottom: 0.25rem; +} + +section { + margin: 1.25rem 0; +} + +fieldset { + border: 1px solid var(--border); + border-radius: 6px; + display: inline-block; + margin: 0 0 1rem; +} + +label { + margin-right: 1rem; +} + +button { + font: inherit; + padding: 0.6rem 1.2rem; + border: 0; + border-radius: 6px; + background: var(--accent); + color: #fff; + cursor: pointer; +} + +button.danger { + background: var(--danger); +} + +button:disabled { + opacity: 0.45; + cursor: not-allowed; +} + +.actions { + display: flex; + gap: 0.75rem; +} + +.step-title { + font-weight: 600; +} + +.hint { + color: var(--muted); + font-style: italic; +} + +progress { + width: 100%; + height: 1rem; +} + +#notice { + font-weight: 600; +} + +pre { + overflow-x: auto; + padding: 0.75rem; + border: 1px solid var(--border); + border-radius: 6px; + font-size: 0.8rem; +} + +code { + overflow-wrap: anywhere; +} diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/store.py b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/store.py new file mode 100644 index 0000000000..a6e11715fb --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/store.py @@ -0,0 +1,318 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Transactional storage for exported MCAL files on the server host. + +A calibration set is selectable only through its manifest. Files are verified in a private staging +directory, hard-linked to unique final names (a link never overwrites), and the manifest is written +last through a temporary file and a no-overwrite link. A set that fails at any point leaves no +manifest, so a partial pair can never be selected later. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import re +import secrets +import shutil +import stat +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path + +from .protocol import ExportedFile, Side + +logger = logging.getLogger("isaaccapture.examples.manus_calibration.store") + +MANIFEST_VERSION = 1 +_ATTEMPT_RE = re.compile(r"[A-Za-z0-9_-]{1,64}") +_FILE_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}\.mcal") +_SET_ID_RE = re.compile(r"[0-9]{8}T[0-9]{6}Z-[0-9a-f]{8}") +_READ_CHUNK = 1 << 20 +_SET_ID_ATTEMPTS = 8 + + +class StoreError(Exception): + """A staged export could not be verified or published; nothing was made selectable.""" + + +@dataclass(frozen=True) +class PublishedFile: + side: Side + file: str + bytes: int + sha256: str + + +@dataclass(frozen=True) +class PublishedSet: + set_id: str + attempt: str + created: str + files: tuple[PublishedFile, ...] + manifest: Path + + def path_of(self, side: Side) -> Path: + """Absolute path of this set's file for ``side``, in the manifest's directory tree.""" + name = next(f.file for f in self.files if f.side is side) + return self.manifest.parent.parent / name + + +def _sha256_of(path: Path) -> tuple[int, str]: + """Size and digest of a regular file, refusing symlinks.""" + flags = os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + try: + fd = os.open(path, flags) + except OSError as exc: + raise StoreError(f"cannot open {path.name}: {exc.strerror or exc}") from None + # Checked on the descriptor, before fdopen: a directory opens read-only and fdopen then raises. + try: + if not stat.S_ISREG(os.fstat(fd).st_mode): + raise StoreError(f"{path.name} is not a regular file") + handle = os.fdopen(fd, "rb") + except BaseException: + os.close(fd) + raise + with handle: + digest = hashlib.sha256() + size = 0 + while chunk := handle.read(_READ_CHUNK): + digest.update(chunk) + size += len(chunk) + return size, digest.hexdigest() + + +def _fsync_dir(path: Path) -> None: + fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +class McalStore: + """The configured calibration directory and the only code that publishes into it.""" + + def __init__(self, directory: str | os.PathLike[str]) -> None: + self._directory = Path(directory).expanduser().resolve() + + @property + def directory(self) -> Path: + return self._directory + + @property + def _staging_root(self) -> Path: + return self._directory / ".staging" + + @property + def _manifest_dir(self) -> Path: + return self._directory / "manifests" + + def create_staging(self, attempt: str) -> Path: + """Create the private (0700) staging directory the worker writes into for ``attempt``.""" + if not _ATTEMPT_RE.fullmatch(attempt): + raise StoreError("attempt id must be 1-64 characters of [A-Za-z0-9_-]") + try: + self._directory.mkdir(mode=0o700, parents=True, exist_ok=True) + self._staging_root.mkdir(mode=0o700, exist_ok=True) + staging = self._staging_root / attempt + staging.mkdir(mode=0o700) + except OSError as exc: + raise StoreError( + f"cannot create staging for {attempt}: {exc.strerror or exc}" + ) from None + return staging + + def discard_staging(self, attempt: str) -> None: + """Remove ``attempt``'s staging directory. Safe to call when it is absent.""" + staging = self._staging_root / attempt + # Only a real directory directly under the staging root is ever removed. + if ( + not _ATTEMPT_RE.fullmatch(attempt) + or staging.is_symlink() + or not staging.is_dir() + ): + return + shutil.rmtree(staging, ignore_errors=True) + + def publish( + self, + attempt: str, + requested: Sequence[Side], + exports: Mapping[Side, ExportedFile], + ) -> PublishedSet: + """Verify every requested side's staged file and publish the set, or raise :class:`StoreError`.""" + staging = self._staging_root / attempt + if set(exports) != set(requested): + raise StoreError( + f"exports {sorted(s.value for s in exports)} do not match requested sides" + ) + verified = {side: self._verify(staging, exports[side]) for side in requested} + created: list[Path] = [] + try: + set_id, finals = self._link_set(staging, verified, created) + manifest = self._write_manifest(set_id, attempt, verified, finals) + except BaseException: + # No manifest exists yet, so these links were never selectable; remove them. + for path in created: + path.unlink(missing_ok=True) + raise + return PublishedSet( + set_id=set_id, + attempt=attempt, + created=_created_of(set_id), + files=tuple( + PublishedFile( + side, finals[side], verified[side].bytes, verified[side].sha256 + ) + for side in requested + ), + manifest=manifest, + ) + + def list_published(self) -> list[PublishedSet]: + """Every manifest whose files all still exist with the recorded size and digest.""" + published = [] + if not self._manifest_dir.is_dir(): + return published + for manifest in sorted(self._manifest_dir.glob("*.json")): + if manifest.name.startswith("."): + continue + try: + published.append(self._load(manifest)) + except (StoreError, OSError, ValueError, KeyError, TypeError) as exc: + logger.warning( + "ignoring calibration manifest %s: %s", manifest.name, exc + ) + return published + + def _verify(self, staging: Path, export: ExportedFile) -> ExportedFile: + """Check a staged file against the worker's reported size and digest.""" + if not _FILE_RE.fullmatch(export.file): + raise StoreError(f"unexpected export name {export.file!r}") + path = staging / export.file + size, digest = _sha256_of(path) + if size == 0 or export.bytes == 0: + raise StoreError(f"{export.file} is empty") + if size != export.bytes or digest != export.sha256: + raise StoreError( + f"{export.file} does not match the size/digest the worker reported" + ) + return export + + def _link_set( + self, + staging: Path, + verified: Mapping[Side, ExportedFile], + created: list[Path], + ) -> tuple[str, dict[Side, str]]: + """Hard-link the staged files to a fresh set id; a link never replaces an existing file.""" + for _ in range(_SET_ID_ATTEMPTS): + set_id = ( + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + + "-" + + secrets.token_hex(4) + ) + finals = {side: f"{set_id}_{side.value}.mcal" for side in verified} + try: + for side, export in verified.items(): + os.link(staging / export.file, self._directory / finals[side]) + created.append(self._directory / finals[side]) + except FileExistsError: + # Id collision with an earlier set: undo this round and draw another id. + for path in created: + path.unlink(missing_ok=True) + created.clear() + continue + except OSError as exc: + raise StoreError( + f"cannot publish calibration files: {exc.strerror or exc}" + ) from None + return set_id, finals + raise StoreError("could not allocate a unique calibration set id") + + def _write_manifest( + self, + set_id: str, + attempt: str, + verified: Mapping[Side, ExportedFile], + finals: Mapping[Side, str], + ) -> Path: + """Write the manifest through a temporary file and a no-overwrite link.""" + body = { + "version": MANIFEST_VERSION, + "set_id": set_id, + "attempt": attempt, + "created": _created_of(set_id), + "files": [ + { + "side": side.value, + "file": finals[side], + "bytes": export.bytes, + "sha256": export.sha256, + } + for side, export in verified.items() + ], + } + final = self._manifest_dir / f"{set_id}.json" + temporary = self._manifest_dir / f".{set_id}.tmp" + linked = False + try: + self._manifest_dir.mkdir(mode=0o700, exist_ok=True) + fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(body, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + try: + os.link(temporary, final) + linked = True + finally: + temporary.unlink(missing_ok=True) + _fsync_dir(self._manifest_dir) + _fsync_dir(self._directory) + except OSError as exc: + # Remove only a manifest this call created; an existing one is never touched. + if linked: + final.unlink(missing_ok=True) + raise StoreError( + f"cannot write calibration manifest: {exc.strerror or exc}" + ) from None + return final + + def _load(self, manifest: Path) -> PublishedSet: + """Parse and verify one manifest; raise if anything about the set is not as recorded.""" + data = json.loads(manifest.read_text(encoding="utf-8")) + if data["version"] != MANIFEST_VERSION or not _SET_ID_RE.fullmatch( + data["set_id"] + ): + raise StoreError("unsupported manifest") + if manifest.name != f"{data['set_id']}.json": + raise StoreError("manifest name does not match its set id") + files = [] + for item in data["files"]: + if not _FILE_RE.fullmatch(item["file"]): + raise StoreError(f"unexpected file name {item['file']!r}") + size, digest = _sha256_of(self._directory / item["file"]) + if size != item["bytes"] or digest != item["sha256"] or size == 0: + raise StoreError(f"{item['file']} no longer matches its manifest") + files.append(PublishedFile(Side(item["side"]), item["file"], size, digest)) + if not files or len({f.side for f in files}) != len(files): + raise StoreError("manifest lists no files or a side twice") + return PublishedSet( + data["set_id"], data["attempt"], data["created"], tuple(files), manifest + ) + + +def _created_of(set_id: str) -> str: + """ISO-8601 UTC timestamp encoded in a set id.""" + return ( + datetime.strptime(set_id[:16], "%Y%m%dT%H%M%SZ") + .replace(tzinfo=timezone.utc) + .isoformat() + ) diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/wire.py b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/wire.py new file mode 100644 index 0000000000..48371bf351 --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/wire.py @@ -0,0 +1,82 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""JSON the page renders, built from the controller's view for one browser.""" + +from __future__ import annotations + +from typing import Any + +from .controller import Stage, View + +_LOG_LINES = 8 +_ACTIVE = frozenset( + { + Stage.STARTING, + Stage.DISCOVERING, + Stage.AWAITING_CONFIRMATION, + Stage.SAMPLING, + Stage.SAVING, + Stage.STOPPING, + } +) + + +def view_to_dict( + view: View, client_id: str, out_dir: str, default_sides: str = "both" +) -> dict[str, Any]: + """Snapshot for ``client_id``; only the owning browser is offered the prompt token.""" + owner = view.owner == client_id + active = view.stage in _ACTIVE + prompt = None + if view.prompt is not None and active: + prompt = { + "side": view.prompt.side.value, + "index": view.prompt.index, + "count": view.prompt.count, + "title": view.prompt.title, + "instruction": view.prompt.instruction, + "mode": view.prompt.mode.value, + "duration_s": view.prompt.duration_s, + "continuous": view.prompt.continuous, + "estimated_s": view.prompt.estimated_s, + # The token lets the owner's click name the prompt it saw; others never need it. + "token": view.prompt.token if owner else None, + } + return { + "stage": view.stage.value, + "active": active, + "owner": owner, + "out_dir": out_dir, + "default_sides": default_sides, + "gloves": [{"side": g.side.value, "glove_id": g.glove_id} for g in view.gloves], + "prompt": prompt, + "confirm_pending": view.confirm_pending, + "elapsed_s": view.elapsed_s, + "notice": view.notice, + "outcome": _outcome(view), + } + + +def _outcome(view: View) -> dict[str, Any] | None: + """Terminal result: where files landed, or why none were published.""" + outcome = view.outcome + if outcome is None: + return None + files = [] + directory = None + if outcome.published is not None: + directory = str(outcome.published.manifest.parent.parent) + files = [ + {"side": f.side.value, "file": f.file, "bytes": f.bytes} + for f in outcome.published.files + ] + return { + "kind": outcome.kind.value, + "message": outcome.message, + "directory": directory, + "files": files, + "log_tail": list(outcome.log_tail[-_LOG_LINES:]) + if outcome.kind.value in ("failed", "uncertain") + else [], + } diff --git a/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/worker.py b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/worker.py new file mode 100644 index 0000000000..d84499ec5c --- /dev/null +++ b/examples/manus_calibration/python/isaaccapture_examples/manus_calibration/worker.py @@ -0,0 +1,804 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Supervises one run of the calibration worker executable. + +:class:`CalibrationWorker` holds the launch settings; :meth:`CalibrationWorker.new_attempt` returns +a :class:`CalibrationAttempt`, which runs the worker once and settles on exactly one +:class:`CalibrationOutcome`. Nothing here imports a UI toolkit. + +Outcome rule: the worker's own terminal ``result`` event is the only evidence of what happened. +Without one (crash, kill, EOF, protocol violation, forced termination) the outcome is +``UNCERTAIN``, never ``COMPLETED``. +""" + +from __future__ import annotations + +import asyncio +import collections +import contextlib +import logging +import secrets +import socket +import subprocess +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from enum import Enum +from pathlib import Path +from typing import Any + +from . import protocol as proto +from .store import McalStore, PublishedSet, StoreError + +logger = logging.getLogger("isaaccapture.examples.manus_calibration.worker") + +_LOG_LINE_CHARS = 1000 +_DRAIN_CHUNK = 4096 +_PROGRESS_SLACK_S = 1e-6 +_WATCHDOG_POLL_S = 0.25 +_TASK_SETTLE_S = 2.0 + + +class AttemptError(Exception): + """The attempt API was used in a way the attempt's state does not allow.""" + + +class ConfirmError(AttemptError): + """A confirmation was refused, by this supervisor or by the worker.""" + + +class AttemptEnded(AttemptError): + """The attempt reached its terminal outcome before the awaited event.""" + + +class AttemptState(str, Enum): + NEW = "new" + STARTING = "starting" + RUNNING = "running" + FINISHING = "finishing" + DONE = "done" + + +class OutcomeKind(str, Enum): + COMPLETED = "completed" + CANCELLED = "cancelled" + FAILED = "failed" + UNCERTAIN = "uncertain" + + +@dataclass(frozen=True) +class CalibrationOutcome: + """Terminal result of an attempt. ``published`` is set only for ``COMPLETED``.""" + + kind: OutcomeKind + attempt: str + sides: tuple[proto.Side, ...] + message: str + published: PublishedSet | None = None + error: proto.WorkerError | None = None + exit_code: int | None = None + forced_termination: bool = False + log_tail: tuple[str, ...] = () + + +class CalibrationWorker: + """Launch settings for the worker executable and the single-attempt-at-a-time guard.""" + + def __init__( + self, + executable: str | Path, + store: McalStore, + *, + discovery_timeout_s: float = 15.0, + ready_timeout_s: float = 10.0, + cancel_grace_s: float = 10.0, + exit_grace_s: float = 5.0, + term_grace_s: float = 3.0, + kill_wait_s: float = 2.0, + log_tail_lines: int = 100, + startup_timeout_s: float | None = None, + confirm_timeout_s: float = 30.0, + finish_timeout_s: float = 60.0, + ) -> None: + self.executable = Path(executable) + self.store = store + self.discovery_timeout_s = discovery_timeout_s + self.ready_timeout_s = ready_timeout_s + self.cancel_grace_s = cancel_grace_s + self.exit_grace_s = exit_grace_s + self.term_grace_s = term_grace_s + self.kill_wait_s = kill_wait_s + self.log_tail_lines = log_tail_lines + # Parent-side deadlines for answers only the worker can give. A worker blocked inside the SDK + # cannot enforce its own, and operator think time is deliberately never bounded. + self.startup_timeout_s = ( + discovery_timeout_s + 45.0 + if startup_timeout_s is None + else startup_timeout_s + ) + self.confirm_timeout_s = confirm_timeout_s + self.finish_timeout_s = finish_timeout_s + # Set when a worker could not be reaped: it may still hold the gloves, so nothing new may start. + self.unreleased: str | None = None + self._active: CalibrationAttempt | None = None + + def new_attempt( + self, on_event: Callable[[proto.WorkerEvent], None] | None = None + ) -> CalibrationAttempt: + """Create the attempt for one run; the worker allows only one active attempt at a time.""" + # An unreaped worker may still own the dongles, so admission stays closed until this program restarts. + if self.unreleased is not None: + raise AttemptError(self.unreleased) + if self._active is not None and self._active.state is not AttemptState.DONE: + raise AttemptError("a calibration attempt is already active") + attempt = CalibrationAttempt(self, on_event) + self._active = attempt + return attempt + + +class CalibrationAttempt: + """One worker process, one attempt, one terminal outcome.""" + + def __init__( + self, + owner: CalibrationWorker, + on_event: Callable[[proto.WorkerEvent], None] | None, + ) -> None: + self.attempt_id = f"cal-{secrets.token_hex(6)}" + self._owner = owner + self._on_event = on_event + self._state = AttemptState.NEW + self._sides: tuple[proto.Side, ...] = () + self._proc: asyncio.subprocess.Process | None = None + self._writer: asyncio.StreamWriter | None = None + self._tasks: list[asyncio.Task[Any]] = [] + self._tail: collections.deque[str] = collections.deque( + maxlen=owner.log_tail_lines + ) + self._outcome: asyncio.Future[CalibrationOutcome] = ( + asyncio.get_running_loop().create_future() + ) + + self._ready = asyncio.Event() + self._cancel_requested = asyncio.Event() + self._result_seen = asyncio.Event() + self._start_sent = False + self._user_cancel = False + self._reader_task: asyncio.Task[None] | None = None + self._abort_reason: str | None = None + self._violation: str | None = None + self._deadline_at: float | None = None + self._deadline_span = 0.0 + self._deadline_what = "" + self._unreaped = False + + self._result: proto.Result | None = None + self._exports: dict[proto.Side, proto.Exported] = {} + self._seen_tokens: set[str] = set() + self._prompt: proto.Prompt | None = None + self._pending_confirm: tuple[str, asyncio.Future[None]] | None = None + self._accepted_token: str | None = None + + @property + def state(self) -> AttemptState: + return self._state + + @property + def prompt(self) -> proto.Prompt | None: + """The prompt awaiting confirmation, if any.""" + return self._prompt + + @property + def sides(self) -> tuple[proto.Side, ...]: + return self._sides + + async def start(self, sides: Sequence[proto.Side]) -> None: + """Launch the worker and send ``start``. Worker-side failures settle :meth:`outcome`, not raise.""" + if self._state is not AttemptState.NEW: + raise AttemptError("an attempt can be started only once") + sides = tuple(sides) + # Validates the sides before anything is created or launched. + proto.encode_start(self.attempt_id, sides, "/") + self._sides = tuple(s for s in proto.Side if s in sides) + self._state = AttemptState.STARTING + try: + staging = self._owner.store.create_staging(self.attempt_id) + except StoreError as exc: + self._finish( + CalibrationOutcome( + OutcomeKind.FAILED, self.attempt_id, self._sides, str(exc) + ) + ) + raise + try: + await self._launch() + except OSError as exc: + self._finish_failed( + "spawn_failed", + f"cannot launch {self._owner.executable}: {exc.strerror or exc}", + ) + return + self._tasks.append( + asyncio.create_task(self._supervise(), name=f"{self.attempt_id}-supervise") + ) + await self._wait_ready() + if ( + self._state is AttemptState.DONE + or self._abort_reason + or self._cancel_requested.is_set() + ): + return + await self._send(proto.encode_start(self.attempt_id, self._sides, str(staging))) + self._start_sent = True + self._state = AttemptState.RUNNING + self._expect( + "SDK initialisation, connection and discovery", + self._owner.startup_timeout_s, + ) + + async def confirm(self, token: str) -> None: + """Start the SDK step offered under ``token``; returns when the worker accepts it. + + A repeated confirmation of a token already accepted or in flight never reaches the worker twice. + """ + if self._accepted_token == token: + return + if self._pending_confirm is not None and self._pending_confirm[0] == token: + await asyncio.shield(self._pending_confirm[1]) + return + if self._state is not AttemptState.RUNNING: + raise AttemptEnded("the attempt is not running") + if ( + self._prompt is None + or self._prompt.token != token + or self._pending_confirm is not None + ): + raise ConfirmError("that prompt is not awaiting confirmation") + done: asyncio.Future[None] = asyncio.get_running_loop().create_future() + self._pending_confirm = (token, done) + self._expect("the SDK to start the step", self._owner.confirm_timeout_s) + await self._send(proto.encode_confirm(self.attempt_id, token)) + await asyncio.shield(done) + + async def cancel(self) -> None: + """Ask the worker to cancel. Idempotent; a result already reported is not changed.""" + if self._state is AttemptState.DONE or self._cancel_requested.is_set(): + return + if self._state is AttemptState.NEW: + self._finish( + CalibrationOutcome( + OutcomeKind.CANCELLED, + self.attempt_id, + self._sides, + "cancelled before start", + ) + ) + return + self._user_cancel = True + self._deadline_at = None + self._cancel_requested.set() + if self._start_sent: + await self._send(proto.encode_cancel(self.attempt_id)) + else: + # The worker has not been told to start, so there is nothing to wind down. + self._abort("cancelled before the worker was started") + + async def outcome(self) -> CalibrationOutcome: + """Wait for the terminal outcome.""" + return await asyncio.shield(self._outcome) + + async def _launch(self) -> None: + parent, child = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) + try: + self._proc = await asyncio.create_subprocess_exec( + str(self._owner.executable), + f"--ipc-fd={child.fileno()}", + f"--discovery-timeout={self._owner.discovery_timeout_s}", + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + pass_fds=(child.fileno(),), + ) + except BaseException: + parent.close() + raise + finally: + child.close() + # The line limit leaves room for the newline; longer lines surface as a protocol violation. + reader, self._writer = await asyncio.open_unix_connection( + sock=parent, limit=proto.MAX_LINE_BYTES + 1 + ) + assert self._proc.stdout is not None and self._proc.stderr is not None + self._tasks.append( + asyncio.create_task(self._drain(self._proc.stdout, "stdout")) + ) + self._tasks.append( + asyncio.create_task(self._drain(self._proc.stderr, "stderr")) + ) + self._reader_task = asyncio.create_task( + self._read_events(reader), name=f"{self.attempt_id}-reader" + ) + self._tasks.append(self._reader_task) + + async def _wait_ready(self) -> None: + """Wait for the worker's ``ready`` event; abort the worker if it does not arrive in time.""" + waiters = { + asyncio.ensure_future(self._ready.wait()), + asyncio.ensure_future(self._cancel_requested.wait()), + self._outcome, + } + try: + done, _ = await asyncio.wait( + waiters, + timeout=self._owner.ready_timeout_s, + return_when=asyncio.FIRST_COMPLETED, + ) + finally: + for waiter in waiters: + if waiter is not self._outcome: + waiter.cancel() + if not done: + self._abort( + f"worker did not report ready within {self._owner.ready_timeout_s:g}s" + ) + + async def _send(self, line: bytes) -> None: + writer = self._writer + if writer is None or writer.is_closing(): + return + try: + writer.write(line) + await writer.drain() + except (ConnectionError, OSError): + # A dead worker is reported by the supervisor through its missing result. + logger.debug("worker connection closed while sending") + + async def _drain(self, stream: asyncio.StreamReader, name: str) -> None: + """Keep a bounded tail of the worker's stdout/stderr and never let a full pipe block it.""" + partial = "" + while chunk := await stream.read(_DRAIN_CHUNK): + partial += chunk.decode("utf-8", errors="replace") + *lines, partial = partial.split("\n") + for line in lines: + self._tail.append(f"{name}: {line[:_LOG_LINE_CHARS]}") + partial = partial[-_LOG_LINE_CHARS:] + if partial: + self._tail.append(f"{name}: {partial[:_LOG_LINE_CHARS]}") + + async def _read_events(self, reader: asyncio.StreamReader) -> None: + """Parse worker lines until EOF; any violation aborts the worker.""" + while True: + try: + line = await reader.readline() + except (ValueError, asyncio.LimitOverrunError): + self._violate(f"worker line exceeds {proto.MAX_LINE_BYTES} bytes") + return + except (ConnectionError, OSError): + return + if not line: + return + if not line.endswith(b"\n"): + self._violate("worker line is not newline-terminated") + return + try: + if self._result is not None: + raise proto.ProtocolError("worker sent a message after its result") + event = proto.parse_event(line[:-1]) + self._handle(event) + except proto.ProtocolError as exc: + self._violate(str(exc)) + return + + def _violate(self, message: str) -> None: + if self._violation is None: + self._violation = message + logger.warning("worker protocol violation: %s", message) + self._abort(f"protocol violation: {message}") + + def _expect(self, what: str, span: float) -> None: + """Start the clock on an answer only the worker can give.""" + self._deadline_what = what + self._deadline_span = span + self._deadline_at = asyncio.get_running_loop().time() + span + + def _touch(self) -> None: + """Any worker event proves it is alive, so a running deadline restarts.""" + if self._deadline_at is not None: + self._deadline_at = asyncio.get_running_loop().time() + self._deadline_span + + async def _watchdog(self) -> None: + """Abort when the worker stays silent past an armed deadline.""" + loop = asyncio.get_running_loop() + while True: + await asyncio.sleep(_WATCHDOG_POLL_S) + if self._deadline_at is not None and loop.time() >= self._deadline_at: + self._abort( + f"the worker did not respond within {self._deadline_span:g} s while waiting for " + f"{self._deadline_what}; it may be blocked inside the SDK" + ) + return + + def _abort(self, reason: str) -> None: + """Stop waiting for the worker: the supervisor terminates it without a grace period.""" + if self._abort_reason is None: + self._abort_reason = reason + self._cancel_requested.set() + + def _handle(self, event: proto.WorkerEvent) -> None: + """Check ``event`` against the attempt's state, then record it and notify the listener.""" + if not self._ready.is_set(): + if not isinstance(event, proto.Ready): + raise proto.ProtocolError( + f"expected ready first, got {type(event).__name__}" + ) + self._ready.set() + elif isinstance(event, proto.Ready): + raise proto.ProtocolError("ready sent twice") + elif not self._start_sent: + raise proto.ProtocolError( + f"{type(event).__name__} before the attempt was started" + ) + attempt = getattr(event, "attempt", self.attempt_id) + if attempt not in (self.attempt_id, None): + raise proto.ProtocolError( + f"event for attempt {attempt!r}, expected {self.attempt_id!r}" + ) + self._touch() + if isinstance(event, proto.Prompt): + self._on_prompt(event) + elif isinstance(event, proto.StepAccepted): + self._on_accepted(event) + elif isinstance(event, proto.StepProgress): + self._on_progress(event) + elif isinstance(event, proto.Rejected): + self._on_rejected(event) + elif isinstance(event, proto.Exported): + self._on_exported(event) + elif isinstance(event, proto.Result): + self._deadline_at = None + self._result = event + self._result_seen.set() + self._notify(event) + + def _on_prompt(self, prompt: proto.Prompt) -> None: + if self._prompt is not None and self._accepted_token != self._prompt.token: + raise proto.ProtocolError( + "prompt offered while the previous one is unconfirmed" + ) + if prompt.token in self._seen_tokens: + raise proto.ProtocolError("prompt token reused") + if prompt.side not in self._sides: + raise proto.ProtocolError("prompt for a side that was not requested") + self._seen_tokens.add(prompt.token) + self._prompt = prompt + # The operator's think time is unbounded. + self._deadline_at = None + + def _on_accepted(self, accepted: proto.StepAccepted) -> None: + if self._pending_confirm is None or self._pending_confirm[0] != accepted.token: + raise proto.ProtocolError("step accepted without a matching confirmation") + done = self._pending_confirm[1] + self._pending_confirm = None + self._accepted_token = accepted.token + if not done.done(): + done.set_result(None) + # The next prompt follows at once, or the worker finishes and exports after the last step. + window = self._prompt.duration_s if self._prompt is not None else 0.0 + self._expect( + "the next step or the result", self._owner.finish_timeout_s + window + ) + + def _on_progress(self, progress: proto.StepProgress) -> None: + prompt = self._prompt + if ( + prompt is None + or progress.token != prompt.token + or self._accepted_token != prompt.token + ): + raise proto.ProtocolError("progress for a step that is not running") + if progress.elapsed_s > prompt.duration_s + _PROGRESS_SLACK_S: + raise proto.ProtocolError("progress exceeds the step duration") + + def _on_rejected(self, rejected: proto.Rejected) -> None: + pending = self._pending_confirm + if ( + pending is not None + and rejected.command == "confirm" + and rejected.token in (None, pending[0]) + ): + self._pending_confirm = None + # The prompt stays offered, so we are back to waiting for the operator. + self._deadline_at = None + if not pending[1].done(): + pending[1].set_exception( + ConfirmError(f"{rejected.reason.value}: {rejected.message}") + ) + + def _on_exported(self, exported: proto.Exported) -> None: + if exported.side not in self._sides or exported.side in self._exports: + raise proto.ProtocolError( + f"unexpected export for the {exported.side.value} side" + ) + self._exports[exported.side] = exported + + def _notify(self, event: proto.WorkerEvent) -> None: + if self._on_event is None: + return + try: + self._on_event(event) + except Exception: + # A faulty listener must not stop supervision of the worker. + logger.exception("calibration event listener raised") + + async def _supervise(self) -> None: + """Run :meth:`_supervise_worker`, settling the outcome even if supervision is cancelled.""" + try: + await self._supervise_worker() + except asyncio.CancelledError: + if self._proc is not None and self._proc.returncode is None: + with contextlib.suppress(ProcessLookupError): + self._proc.kill() + self._finish( + self._outcome_of( + OutcomeKind.UNCERTAIN, "supervision was cancelled", None, True + ) + ) + raise + + async def _supervise_worker(self) -> None: + """Wait for the worker to end (or be stopped), reap it, and settle the outcome.""" + reader = self._reader_task + assert reader is not None + cancel = asyncio.create_task(self._cancel_requested.wait()) + result = asyncio.create_task(self._result_seen.wait()) + watchdog = asyncio.create_task(self._watchdog()) + try: + await asyncio.wait( + {reader, cancel, result}, return_when=asyncio.FIRST_COMPLETED + ) + if not reader.done(): + if self._abort_reason: + grace = 0.0 + elif self._result_seen.is_set(): + grace = self._owner.exit_grace_s + else: + grace = self._owner.cancel_grace_s + with contextlib.suppress(TimeoutError): + async with asyncio.timeout(grace): + await asyncio.shield(reader) + finally: + cancel.cancel() + result.cancel() + watchdog.cancel() + self._state = AttemptState.FINISHING + forced = await self._stop_process() + # Exit not established (even SIGKILL did not reap it): the process may still hold the gloves. + self._unreaped = self._proc is not None and self._proc.returncode is None + if not reader.done(): + reader.cancel() + await self._settle_tasks() + await self._finalize(forced) + + async def _settle_tasks(self) -> None: + """Wait a bounded time for the drain and reader tasks; a surviving process can hold their pipes open.""" + others = [t for t in self._tasks if t is not asyncio.current_task()] + if not others: + return + _done, pending = await asyncio.wait(others, timeout=_TASK_SETTLE_S) + for task in pending: + task.cancel() + + async def _stop_process(self) -> bool: + """Reap the worker, escalating to SIGTERM then SIGKILL. Returns whether it had to be forced.""" + proc = self._proc + assert proc is not None + grace = 0.0 if self._abort_reason else self._owner.exit_grace_s + with contextlib.suppress(TimeoutError): + async with asyncio.timeout(grace): + await proc.wait() + if proc.returncode is not None: + return False + with contextlib.suppress(ProcessLookupError): + proc.terminate() + try: + async with asyncio.timeout(self._owner.term_grace_s): + await proc.wait() + return True + except TimeoutError: + pass + with contextlib.suppress(ProcessLookupError): + proc.kill() + with contextlib.suppress(TimeoutError): + async with asyncio.timeout(self._owner.kill_wait_s): + await proc.wait() + return True + + async def _finalize(self, forced: bool) -> None: + """Turn the evidence collected into the one terminal outcome, publishing files only on success.""" + proc = self._proc + code = proc.returncode if proc is not None else None + try: + outcome = await self._decide(code, forced) + except Exception as exc: + logger.exception("calibration finalization failed") + outcome = self._outcome_of( + OutcomeKind.UNCERTAIN, + f"internal error finishing the attempt: {exc}", + code, + forced, + ) + self._finish(outcome) + + async def _decide(self, code: int | None, forced: bool) -> CalibrationOutcome: + if self._unreaped: + pid = self._proc.pid if self._proc is not None else "?" + message = ( + f"the worker (pid {pid}) could not be stopped and may still be using the gloves; " + "restart this program once it has exited" + ) + self._owner.unreleased = message + return self._outcome_of( + OutcomeKind.UNCERTAIN, + message, + code, + forced, + error=proto.WorkerError("worker_unreaped", message), + ) + result = self._result + if self._violation is not None: + # Before `start` the worker has not opened the SDK, so only a started worker leaves it unknown. + kind = OutcomeKind.UNCERTAIN if self._start_sent else OutcomeKind.FAILED + return self._outcome_of( + kind, + f"worker broke the protocol ({self._violation})", + code, + forced, + error=proto.WorkerError("protocol_error", self._violation), + ) + if result is None: + if not self._start_sent: + if self._user_cancel: + return self._outcome_of( + OutcomeKind.CANCELLED, + "cancelled before the worker started", + code, + forced, + ) + return self._outcome_of( + OutcomeKind.FAILED, + self._abort_reason + or f"worker exited before it was started (exit code {code})", + code, + forced, + error=proto.WorkerError( + "worker_exit", + self._abort_reason or "worker exited before start", + ), + ) + reason = ( + self._abort_reason + or f"worker ended without reporting a result (exit code {code})" + ) + return self._outcome_of( + OutcomeKind.UNCERTAIN, + f"{reason}; the worker was stopped and its SDK state is unknown", + code, + forced, + error=proto.WorkerError("worker_exit", f"no result; exit code {code}"), + ) + if forced: + return self._outcome_of( + OutcomeKind.UNCERTAIN, + f"worker reported {result.outcome.value} but had to be terminated afterwards", + code, + forced, + ) + if result.outcome is proto.Outcome.FAILED: + assert result.error is not None + return self._outcome_of( + OutcomeKind.FAILED, + result.error.message, + code, + forced, + error=result.error, + ) + if code != 0: + return self._outcome_of( + OutcomeKind.UNCERTAIN, + f"worker reported {result.outcome.value} but exited with code {code}", + code, + forced, + ) + if result.outcome is proto.Outcome.CANCELLED: + return self._outcome_of(OutcomeKind.CANCELLED, "cancelled", code, forced) + return await self._publish(result, code, forced) + + async def _publish( + self, result: proto.Result, code: int | None, forced: bool + ) -> CalibrationOutcome: + """Verify and publish a completed result's files.""" + reported = {f.side: f for f in result.files} + exported = { + side: proto.ExportedFile(side, e.file, e.bytes, e.sha256) + for side, e in self._exports.items() + } + if reported != exported or set(reported) != set(self._sides): + return self._outcome_of( + OutcomeKind.UNCERTAIN, + "worker's completed result does not match the files it exported", + code, + forced, + error=proto.WorkerError( + "protocol_error", "result files differ from exported files" + ), + ) + try: + published = await asyncio.to_thread( + self._owner.store.publish, self.attempt_id, self._sides, exported + ) + except StoreError as exc: + return self._outcome_of( + OutcomeKind.FAILED, + f"calibration files were not saved: {exc}", + code, + forced, + error=proto.WorkerError("store_error", str(exc)), + ) + return self._outcome_of( + OutcomeKind.COMPLETED, + "calibration saved", + code, + forced, + published=published, + ) + + def _outcome_of( + self, + kind: OutcomeKind, + message: str, + code: int | None, + forced: bool, + *, + error: proto.WorkerError | None = None, + published: PublishedSet | None = None, + ) -> CalibrationOutcome: + return CalibrationOutcome( + kind=kind, + attempt=self.attempt_id, + sides=self._sides, + message=message, + published=published, + error=error, + exit_code=code, + forced_termination=forced, + log_tail=tuple(self._tail), + ) + + def _finish_failed(self, code: str, message: str) -> None: + self._finish( + self._outcome_of( + OutcomeKind.FAILED, + message, + None, + False, + error=proto.WorkerError(code, message), + ) + ) + + def _finish(self, outcome: CalibrationOutcome) -> None: + """Settle the attempt exactly once: release staging, fail waiters, publish the outcome.""" + if self._outcome.done(): + return + # A worker that may still be running can still write here, so its staging is left in place. + if not self._unreaped: + self._owner.store.discard_staging(self.attempt_id) + self._state = AttemptState.DONE + self._prompt = None + if self._pending_confirm is not None: + pending = self._pending_confirm[1] + self._pending_confirm = None + if not pending.done(): + pending.set_exception( + AttemptEnded("the attempt ended before the step was accepted") + ) + self._outcome.set_result(outcome) diff --git a/examples/manus_calibration/tests/conftest.py b/examples/manus_calibration/tests/conftest.py new file mode 100644 index 0000000000..c99930ee3c --- /dev/null +++ b/examples/manus_calibration/tests/conftest.py @@ -0,0 +1,89 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import asyncio +import json +import sys +from pathlib import Path + +import pytest +from isaaccapture_examples.manus_calibration import protocol as proto +from isaaccapture_examples.manus_calibration.store import McalStore +from isaaccapture_examples.manus_calibration.worker import ( + AttemptState, + CalibrationAttempt, + CalibrationOutcome, + CalibrationWorker, +) + +FAKE_WORKER = Path(__file__).with_name("fake_worker.py") +BOTH = (proto.Side.LEFT, proto.Side.RIGHT) + + +@pytest.fixture +def commands_log(tmp_path: Path) -> Path: + return tmp_path / "commands.log" + + +@pytest.fixture +def make_worker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, commands_log: Path): + """Build a CalibrationWorker around the scripted fake, with short supervision timeouts.""" + + def factory( + scenario: dict | None = None, **settings + ) -> tuple[CalibrationWorker, McalStore]: + wrapper = tmp_path / "fake-calibrator" + wrapper.write_text( + f"#!{sys.executable}\nimport runpy\nrunpy.run_path({str(FAKE_WORKER)!r}, run_name='__main__')\n" + ) + wrapper.chmod(0o755) + monkeypatch.setenv("FAKE_WORKER_SCENARIO", json.dumps(scenario or {})) + monkeypatch.setenv("FAKE_WORKER_LOG", str(commands_log)) + store = McalStore(tmp_path / "calibrations") + timeouts = { + "ready_timeout_s": 5, + "cancel_grace_s": 2, + "exit_grace_s": 2, + "term_grace_s": 1, + "kill_wait_s": 2, + } + return CalibrationWorker(wrapper, store, **{**timeouts, **settings}), store + + return factory + + +def commands(path: Path) -> list[str]: + """Command types the fake worker received, in order.""" + return path.read_text().split() if path.exists() else [] + + +async def drive( + attempt: CalibrationAttempt, + sides=BOTH, + *, + until: type | None = None, + timeout: float = 20.0, +) -> CalibrationOutcome: + """Start ``attempt`` and confirm every prompt until it settles.""" + await attempt.start(sides) + confirmed: set[str] = set() + async with asyncio.timeout(timeout): + while attempt.state is not AttemptState.DONE: + prompt = attempt.prompt + if prompt is not None and prompt.token not in confirmed: + confirmed.add(prompt.token) + try: + await attempt.confirm(prompt.token) + except Exception: + pass + await asyncio.sleep(0.01) + return await attempt.outcome() + + +async def wait_for(predicate, timeout: float = 10.0) -> None: + async with asyncio.timeout(timeout): + # The predicate reads plain attributes, so there is no event to await. + while not predicate(): # noqa: ASYNC110 + await asyncio.sleep(0.01) diff --git a/examples/manus_calibration/tests/fake_worker.py b/examples/manus_calibration/tests/fake_worker.py new file mode 100644 index 0000000000..ba3e435bb2 --- /dev/null +++ b/examples/manus_calibration/tests/fake_worker.py @@ -0,0 +1,412 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""A scripted stand-in for ``manus_glove_calibrator`` that speaks the worker contract. + +Behavior comes from the JSON object in ``FAKE_WORKER_SCENARIO``; ``FAKE_WORKER_LOG`` names a file +that receives one line per command the fake reads, so tests can count what reached the worker. +A step may set ``"continuous": true`` with ``"estimated": SECONDS`` (and ``"time": 0``) to model the SDK's +continuous step, which is a hold-paced prompt that carries an estimated duration. +Hang scenarios keep the process alive and silent (``ignore_term`` makes it survive SIGTERM too): +``hang_after_start``, ``hang_after_discovering``, ``hang_on_confirm`` and ``hang_after_last_accept``. +``chatty_discovery`` and ``chatty_finish`` (seconds) keep emitting status events while the worker is busy. +Run as ``python fake_worker.py --ipc-fd=N``. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import select +import signal +import socket +import sys +import time + +SCENARIO = json.loads(os.environ.get("FAKE_WORKER_SCENARIO", "{}")) +LOG_PATH = os.environ.get("FAKE_WORKER_LOG") +STEPS = SCENARIO.get( + "steps", + [ + {"title": "Flat hand", "instruction": "Lay your hand flat.", "time": 0.0}, + {"title": "Wave", "instruction": "Wave your fingers.", "time": 0.4}, + ], +) +GLOVE_IDS = {"left": 11, "right": 22} + + +class Cancelled(Exception): + pass + + +def hang() -> None: + """Stay alive and silent, as a worker blocked inside the SDK would.""" + time.sleep(60) + + +def chatter(channel: "Channel", phase: str, seconds: float, gloves: list) -> None: + """Emit a status event every 0.2 s for ``seconds``, modelling a slow but live worker.""" + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + channel.status(phase, gloves) + channel.sleep(0.2) + + +# Set once the first confirmation has been refused by the `reject_first_confirm` scenario. +_REJECTED_ONCE = False + + +def flood(total: int) -> None: + """Write about ``total`` bytes of short lines plus one overlong line to stdout and stderr.""" + line = "x" * 99 + "\n" + # Looks like an IPC message; it travels on stdout, which is never parsed. + decoy = '{"v": 1, "type": "result", "attempt": null, "outcome": "failed", "error": {"code": "internal", "message": "decoy"}}\n' + for stream in (sys.stdout, sys.stderr): + stream.write(decoy) + for _ in range(total // len(line)): + stream.write(line) + stream.write("y" * 5000 + "\n") + stream.flush() + + +def main() -> int: + fd = next( + int(a.split("=", 1)[1]) for a in sys.argv[1:] if a.startswith("--ipc-fd=") + ) + sock = socket.socket(fileno=fd) + channel = Channel(sock) + print("fake worker log line on stdout", flush=True) + print("fake worker log line on stderr", file=sys.stderr, flush=True) + if SCENARIO.get("noise_bytes"): + flood(SCENARIO["noise_bytes"]) + if SCENARIO.get("exit_before_ready"): + return 3 + if SCENARIO.get("ignore_term"): + signal.signal(signal.SIGTERM, signal.SIG_IGN) + if SCENARIO.get("term_exits_zero"): + # A worker that shuts down "cleanly" when terminated, so its exit code alone looks like success. + signal.signal(signal.SIGTERM, lambda *_: os._exit(0)) + if SCENARIO.get("never_ready"): + time.sleep(60) + return 0 + if SCENARIO.get("garbage_ready"): + channel.raw(b"this is not json\n") + time.sleep(60) + return 0 + channel.send({"type": "ready", "worker_version": "fake-1", "sdk_version": None}) + start = channel.read_command("start") + attempt = start["attempt"] + channel.attempt = attempt + try: + return run_attempt(channel, attempt, start) + except Cancelled: + return cancelled(channel, attempt) + + +def run_attempt(channel: "Channel", attempt: str, start: dict) -> int: + sides = start["sides"] + staging = start["staging_dir"] + if SCENARIO.get("hang_after_start"): + hang() + if SCENARIO.get("malformed_line"): + channel.raw(b"{bad json\n") + time.sleep(60) + if SCENARIO.get("oversized"): + channel.raw(b'{"v":1,"type":"status","pad":"' + b"x" * 70000 + b'"}\n') + time.sleep(60) + if SCENARIO.get("unknown_event"): + channel.send({"type": "mystery", "attempt": attempt}) + time.sleep(60) + if SCENARIO.get("spurious_accept"): + channel.send( + {"type": "step_accepted", "attempt": attempt, "token": "never-offered"} + ) + time.sleep(60) + if SCENARIO.get("unrequested_side_prompt"): + other = "right" if "right" not in sides else "left" + channel.send( + { + "type": "prompt", + "attempt": attempt, + "token": "x-0", + "side": other, + "glove_id": GLOVE_IDS[other], + "index": 0, + "count": 1, + "title": "t", + "instruction": "i", + "mode": "hold", + "duration_s": 0.0, + "continuous": False, + "estimated_s": 0.0, + } + ) + time.sleep(60) + if SCENARIO.get("wrong_attempt"): + channel.send( + { + "type": "status", + "attempt": "other-attempt", + "phase": "discovering", + "gloves": [], + } + ) + time.sleep(60) + channel.status("discovering", []) + if SCENARIO.get("hang_after_discovering"): + hang() + if SCENARIO.get("chatty_discovery"): + chatter(channel, "discovering", SCENARIO["chatty_discovery"], []) + if SCENARIO.get("discover_delay"): + channel.sleep(SCENARIO["discover_delay"]) + gloves = [{"side": s, "glove_id": GLOVE_IDS[s]} for s in sides] + channel.status("discovering", gloves) + exported = [] + for side in sides: + for index, step in enumerate(STEPS): + token = f"{side}-{index}" + channel.send( + { + "type": "prompt", + "attempt": attempt, + "token": token, + "side": side, + "glove_id": GLOVE_IDS[side], + "index": index, + "count": len(STEPS), + "title": step["title"], + "instruction": step["instruction"], + "mode": "motion" if step["time"] > 0 else "hold", + "duration_s": step["time"], + "continuous": bool(step.get("continuous", False)), + "estimated_s": float(step.get("estimated", 0.0)), + } + ) + channel.status("awaiting_step", gloves) + if SCENARIO.get("crash_after_prompt"): + os._exit(9) + confirm_step(channel, attempt, token, step) + last = side == sides[-1] and index == len(STEPS) - 1 + if last and SCENARIO.get("hang_after_last_accept"): + hang() + if last and SCENARIO.get("chatty_finish"): + chatter(channel, "saving", SCENARIO["chatty_finish"], gloves) + if SCENARIO.get("fail_at_step") == [side, index]: + return fail( + channel, + attempt, + "step_failed", + "the SDK refused the step", + side, + index, + ) + if SCENARIO.get("save_delay"): + channel.status("saving", gloves) + channel.sleep(SCENARIO["save_delay"]) + exported.append(export(channel, attempt, staging, side)) + if SCENARIO.get("fail_after_first_export"): + return fail( + channel, attempt, "export_failed", "second glove failed", None, None + ) + channel.status("saving", gloves) + if SCENARIO.get("no_result"): + return 0 + files = [{k: e[k] for k in ("side", "file", "bytes", "sha256")} for e in exported] + if SCENARIO.get("result_mismatch"): + files[0]["bytes"] += 1 + if SCENARIO.get("partial_result"): + files = files[:1] + channel.send( + {"type": "result", "attempt": attempt, "outcome": "completed", "files": files} + ) + if SCENARIO.get("extra_after_result"): + channel.status("saving", gloves) + if SCENARIO.get("hang_after_result"): + time.sleep(60) + return int(SCENARIO.get("exit_code_after_result", 0)) + + +def confirm_step(channel: "Channel", attempt: str, token: str, step: dict) -> None: + """Wait for the confirmation of ``token``, refusing anything else, then run the step.""" + global _REJECTED_ONCE + accepted = False + while not accepted: + command = channel.read_command("confirm", "cancel") + if command["type"] == "cancel": + if SCENARIO.get("ignore_cancel"): + time.sleep(60) + raise Cancelled + if SCENARIO.get("hang_on_confirm"): + hang() + if SCENARIO.get("reject_first_confirm") and not _REJECTED_ONCE: + _REJECTED_ONCE = True + channel.send( + { + "type": "rejected", + "attempt": attempt, + "command": "confirm", + "reason": "unexpected", + "message": "not now", + "token": command["token"], + } + ) + continue + if command["token"] != token: + channel.send( + { + "type": "rejected", + "attempt": attempt, + "command": "confirm", + "reason": "stale_token", + "message": "not the offered prompt", + "token": command["token"], + } + ) + continue + channel.send({"type": "step_accepted", "attempt": attempt, "token": token}) + accepted = True + if step["time"] > 0: + elapsed = 0.0 + channel.status("capturing_step", []) + if SCENARIO.get("progress_overrun"): + channel.send( + { + "type": "step_progress", + "attempt": attempt, + "token": token, + "elapsed_s": step["time"] + 5, + } + ) + time.sleep(60) + while elapsed < step["time"]: + channel.sleep(0.1) + elapsed = min(step["time"], elapsed + 0.1) + channel.send( + { + "type": "step_progress", + "attempt": attempt, + "token": token, + "elapsed_s": elapsed, + } + ) + + +def export(channel: "Channel", attempt: str, staging: str, side: str) -> dict: + empty = SCENARIO.get("empty_export") or SCENARIO.get("empty_side") == side + data = b"" if empty else (b"MCAL-" + side.encode()) * 64 + name = f"{attempt}_{side}.mcal" + path = os.path.join(staging, name) + if SCENARIO.get("symlink_side") == side: + os.symlink(SCENARIO["symlink_target"], path) + elif SCENARIO.get("directory_side") == side: + os.mkdir(path) + elif not SCENARIO.get("missing_file"): + with open(path + ".part", "wb") as handle: + handle.write(data) + os.rename(path + ".part", path) + corrupt = SCENARIO.get("bad_sha") or SCENARIO.get("bad_sha_side") == side + digest = hashlib.sha256(b"different" if corrupt else data).hexdigest() + reported = ( + SCENARIO.get("reported_name", name) + if SCENARIO.get("name_side", side) == side + else name + ) + event = { + "type": "exported", + "attempt": attempt, + "side": side, + "glove_id": GLOVE_IDS[side], + "file": reported, + "bytes": len(data), + "sha256": digest, + } + if not SCENARIO.get("no_export_events"): + channel.send(event) + return event + + +def fail(channel: "Channel", attempt: str, code: str, message: str, side, step) -> int: + error = {"code": code, "message": message} + if side is not None: + error.update(side=side, step=step) + channel.send( + {"type": "result", "attempt": attempt, "outcome": "failed", "error": error} + ) + return 1 + + +def cancelled(channel: "Channel", attempt: str) -> int: + # A worker that needs time to stop its SDK after a cancel. + time.sleep(float(SCENARIO.get("slow_cancel", 0))) + channel.status("stopping", []) + channel.send({"type": "result", "attempt": attempt, "outcome": "cancelled"}) + return 0 + + +class Channel: + def __init__(self, sock: socket.socket) -> None: + self.sock = sock + self.buffer = b"" + self.attempt: str | None = None + + def raw(self, data: bytes) -> None: + self.sock.sendall(data) + + def send(self, message: dict) -> None: + self.raw((json.dumps({"v": 1, **message}) + "\n").encode()) + + def status(self, phase: str, gloves: list) -> None: + self.send( + { + "type": "status", + "attempt": self.attempt, + "phase": phase, + "gloves": gloves, + } + ) + + def poll_command(self, timeout: float) -> dict | None: + """Return the next buffered or arriving command, or None when ``timeout`` passes.""" + while b"\n" not in self.buffer: + ready, _, _ = select.select([self.sock], [], [], timeout) + if not ready: + return None + chunk = self.sock.recv(4096) + if not chunk: + raise Cancelled + self.buffer += chunk + line, self.buffer = self.buffer.split(b"\n", 1) + command = json.loads(line) + if LOG_PATH: + with open(LOG_PATH, "a", encoding="utf-8") as log: + log.write(command["type"] + "\n") + return command + + def read_command(self, *types: str) -> dict: + while True: + command = self.poll_command(30.0) + if command is None: + raise SystemExit("fake worker timed out waiting for a command") + if command["type"] == "cancel" and "cancel" not in types: + if SCENARIO.get("ignore_cancel"): + continue + raise Cancelled + if command["type"] in types: + return command + + def sleep(self, seconds: float) -> None: + """Sleep while still honoring a cancel command.""" + deadline = time.monotonic() + seconds + while (remaining := deadline - time.monotonic()) > 0: + command = self.poll_command(min(0.05, remaining)) + if ( + command is not None + and command["type"] == "cancel" + and not SCENARIO.get("ignore_cancel") + ): + raise Cancelled + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/manus_calibration/tests/test_controller.py b/examples/manus_calibration/tests/test_controller.py new file mode 100644 index 0000000000..985b2b7183 --- /dev/null +++ b/examples/manus_calibration/tests/test_controller.py @@ -0,0 +1,538 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Ownership, stage and confirmation rules of the controller, over a real fake-worker process.""" + +from __future__ import annotations + +import asyncio + +import pytest +from conftest import commands, wait_for +from isaaccapture_examples.manus_calibration import protocol as proto +from isaaccapture_examples.manus_calibration.controller import ( + CalibrationController, + NotOwner, + Stage, +) +from isaaccapture_examples.manus_calibration.worker import AttemptError, OutcomeKind + +HOLD = { + "steps": [ + {"title": "A", "instruction": "a", "time": 0.0}, + {"title": "B", "instruction": "b", "time": 0.0}, + ] +} +ONE_HOLD = {"steps": [{"title": "A", "instruction": "a", "time": 0.0}]} +LEFT = (proto.Side.LEFT,) +BOTH = (proto.Side.LEFT, proto.Side.RIGHT) +OWNER = "owner-client-0001" +OTHER = "other-client-0002" + + +def build(make_worker, scenario=None, **settings): + """A controller over a fake worker, plus the views it announced.""" + worker, store = make_worker({**HOLD, **(scenario or {})}, **settings) + views = [] + return CalibrationController(worker, views.append), store, views + + +async def at_prompt(controller: CalibrationController) -> None: + await wait_for( + lambda: ( + controller.view().stage is Stage.AWAITING_CONFIRMATION + and controller.view().prompt is not None + ) + ) + + +async def finish(controller: CalibrationController, client: str = OWNER): + """Confirm each prompt as ``client`` until the attempt settles; returns the last view.""" + confirmed: set[str] = set() + async with asyncio.timeout(30): + while controller.view().stage is not Stage.FINISHED: + view = controller.view() + prompt = view.prompt + if ( + prompt + and prompt.token not in confirmed + and view.stage is Stage.AWAITING_CONFIRMATION + ): + confirmed.add(prompt.token) + await controller.confirm(client, prompt.token) + await asyncio.sleep(0.01) + return controller.view() + + +def is_subsequence(wanted, seen) -> bool: + iterator = iter(seen) + return all(any(item == other for other in iterator) for item in wanted) + + +def test_a_new_controller_is_idle(make_worker): + controller, _store, views = build(make_worker) + view = controller.view() + assert view.stage is Stage.IDLE + assert view.owner is None + assert view.prompt is None + assert view.outcome is None + assert controller.active is False + assert views == [] + + +def test_start_assigns_the_owner_and_reaches_the_first_prompt(make_worker): + async def scenario(): + controller, _store, _views = build(make_worker) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + view = controller.view() + assert controller.active is True + assert view.owner == OWNER + assert view.sides == LEFT + assert view.prompt.side is proto.Side.LEFT + assert view.prompt.index == 0 + assert [g.side for g in view.gloves] == [proto.Side.LEFT] + await controller.aclose() + + asyncio.run(scenario()) + + +def test_the_sides_are_reported_in_worker_order_whatever_order_was_asked(make_worker): + async def scenario(): + controller, _store, _views = build(make_worker) + await controller.start(OWNER, (proto.Side.RIGHT, proto.Side.LEFT)) + assert controller.view().sides == BOTH + await controller.aclose() + + asyncio.run(scenario()) + + +@pytest.mark.parametrize("second", [OWNER, OTHER]) +def test_a_second_start_is_refused_and_changes_nothing(make_worker, second): + async def scenario(): + controller, _store, _views = build(make_worker) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + before = controller.view() + with pytest.raises(AttemptError, match="already active"): + await controller.start(second, BOTH) + after = controller.view() + assert after.owner == OWNER + assert after.sides == before.sides + assert after.prompt == before.prompt + await controller.aclose() + + asyncio.run(scenario()) + + +def test_only_the_owner_may_confirm_or_cancel(make_worker, commands_log): + async def scenario(): + controller, _store, _views = build(make_worker) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + token = controller.view().prompt.token + with pytest.raises(NotOwner): + await controller.confirm(OTHER, token) + with pytest.raises(NotOwner): + await controller.cancel(OTHER) + view = controller.view() + assert view.stage is Stage.AWAITING_CONFIRMATION + assert view.confirm_pending is False + assert "confirm" not in commands(commands_log) + assert "cancel" not in commands(commands_log) + await controller.aclose() + + asyncio.run(scenario()) + + +def test_confirm_and_cancel_with_nothing_active_are_refused(make_worker): + async def scenario(): + controller, _store, _views = build(make_worker) + with pytest.raises(AttemptError, match="no calibration attempt"): + await controller.confirm(OWNER, "left-0") + with pytest.raises(AttemptError, match="no calibration attempt"): + await controller.cancel(OWNER) + + asyncio.run(scenario()) + + +@pytest.mark.parametrize("token", ["wrong-token", "", "right-0"]) +def test_a_wrong_or_stale_token_is_refused_without_a_step( + make_worker, commands_log, token +): + async def scenario(): + controller, _store, _views = build(make_worker) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + with pytest.raises(AttemptError, match="no longer current"): + await controller.confirm(OWNER, token) + assert controller.view().confirm_pending is False + assert "confirm" not in commands(commands_log) + await controller.aclose() + + asyncio.run(scenario()) + + +def test_a_superseded_prompt_token_is_refused_after_the_next_prompt_appears( + make_worker, commands_log +): + async def scenario(): + controller, _store, _views = build(make_worker) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + first = controller.view().prompt + await controller.confirm(OWNER, first.token) + await wait_for( + lambda: ( + controller.view().prompt is not None + and controller.view().prompt.token != first.token + and controller.view().stage is Stage.AWAITING_CONFIRMATION + ) + ) + sent = commands(commands_log).count("confirm") + with pytest.raises(AttemptError, match="no longer current"): + await controller.confirm(OWNER, first.token) + assert commands(commands_log).count("confirm") == sent + await controller.aclose() + + asyncio.run(scenario()) + + +@pytest.mark.parametrize( + "sides", [LEFT, (proto.Side.RIGHT,), BOTH], ids=["left", "right", "both"] +) +def test_a_full_run_completes_publishes_and_releases_ownership(make_worker, sides): + async def scenario(): + controller, store, _views = build(make_worker) + await controller.start(OWNER, sides) + view = await finish(controller) + assert view.stage is Stage.FINISHED + assert view.owner is None + assert view.prompt is None + assert view.outcome.kind is OutcomeKind.COMPLETED + assert {f.side for f in view.outcome.published.files} == set(sides) + assert len(store.list_published()) == 1 + assert controller.active is False + assert "completed" in view.notice + + asyncio.run(scenario()) + + +def test_a_new_attempt_can_start_after_a_finished_one(make_worker): + async def scenario(): + controller, store, _views = build(make_worker) + await controller.start(OWNER, LEFT) + await finish(controller) + await controller.start(OTHER, LEFT) + assert controller.view().owner == OTHER + await finish(controller, OTHER) + assert len(store.list_published()) == 2 + + asyncio.run(scenario()) + + +def test_stages_advance_in_order_and_motion_reports_progress(make_worker): + async def scenario(): + steps = {"steps": [{"title": "W", "instruction": "w", "time": 0.4}]} + controller, _store, views = build(make_worker, steps) + await controller.start(OWNER, LEFT) + await finish(controller) + stages = [v.stage for v in views] + assert is_subsequence( + [ + Stage.STARTING, + Stage.DISCOVERING, + Stage.AWAITING_CONFIRMATION, + Stage.SAMPLING, + Stage.FINISHED, + ], + stages, + ) + elapsed = [v.elapsed_s for v in views if v.elapsed_s is not None] + assert elapsed + assert max(elapsed) <= 0.4 + 1e-6 + assert elapsed == sorted(elapsed) + + asyncio.run(scenario()) + + +def test_confirm_pending_is_raised_while_the_worker_decides_then_cleared( + make_worker, +): + async def scenario(): + controller, _store, views = build(make_worker) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + await controller.confirm(OWNER, controller.view().prompt.token) + assert any(v.confirm_pending for v in views) + assert controller.view().confirm_pending is False + await controller.aclose() + + asyncio.run(scenario()) + + +def test_a_second_confirm_while_one_is_in_flight_sends_only_one_command( + make_worker, commands_log +): + async def scenario(): + controller, _store, _views = build(make_worker, ONE_HOLD) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + token = controller.view().prompt.token + await asyncio.gather( + controller.confirm(OWNER, token), controller.confirm(OWNER, token) + ) + assert commands(commands_log).count("confirm") == 1 + await finish(controller) + + asyncio.run(scenario()) + + +def test_a_refused_confirmation_becomes_a_notice_and_can_be_retried(make_worker): + async def scenario(): + controller, _store, _views = build( + make_worker, {**ONE_HOLD, "reject_first_confirm": True} + ) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + token = controller.view().prompt.token + await controller.confirm(OWNER, token) + view = controller.view() + assert view.notice + assert view.confirm_pending is False + assert view.stage is Stage.AWAITING_CONFIRMATION + await controller.confirm(OWNER, token) + view = await finish(controller) + assert view.outcome.kind is OutcomeKind.COMPLETED + + asyncio.run(scenario()) + + +def test_cancel_by_the_owner_ends_cancelled_and_publishes_nothing(make_worker): + async def scenario(): + controller, store, views = build(make_worker) + await controller.start(OWNER, BOTH) + await at_prompt(controller) + await controller.cancel(OWNER) + assert any(v.stage is Stage.STOPPING for v in views) + await wait_for(lambda: controller.view().stage is Stage.FINISHED) + view = controller.view() + assert view.outcome.kind is OutcomeKind.CANCELLED + assert view.outcome.published is None + assert store.list_published() == [] + assert controller.active is False + + asyncio.run(scenario()) + + +def test_only_the_owner_disconnecting_cancels_the_attempt(make_worker): + async def scenario(): + controller, store, _views = build(make_worker) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + await controller.client_disconnected(OTHER) + await asyncio.sleep(0.3) + assert controller.view().stage is Stage.AWAITING_CONFIRMATION + assert controller.active is True + await controller.client_disconnected(OWNER) + await wait_for(lambda: controller.view().stage is Stage.FINISHED) + assert controller.view().outcome.kind is OutcomeKind.CANCELLED + assert store.list_published() == [] + + asyncio.run(scenario()) + + +def test_a_disconnect_with_nothing_active_is_ignored(make_worker): + async def scenario(): + controller, _store, views = build(make_worker) + await controller.client_disconnected(OWNER) + assert views == [] + + asyncio.run(scenario()) + + +def test_a_completed_attempts_files_survive_the_owner_disconnecting_later( + make_worker, +): + async def scenario(): + controller, store, _views = build(make_worker) + await controller.start(OWNER, LEFT) + await finish(controller) + await controller.client_disconnected(OWNER) + assert len(store.list_published()) == 1 + assert controller.view().outcome.kind is OutcomeKind.COMPLETED + + asyncio.run(scenario()) + + +def test_aclose_cancels_an_active_attempt_and_waits_for_the_worker(make_worker): + async def scenario(): + controller, store, _views = build(make_worker) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + await controller.aclose() + outcome = controller.view().outcome + assert outcome.kind is OutcomeKind.CANCELLED + assert outcome.exit_code == 0 + assert outcome.forced_termination is False + assert store.list_published() == [] + + asyncio.run(scenario()) + + +def test_aclose_with_nothing_active_is_a_no_op(make_worker): + async def scenario(): + controller, _store, views = build(make_worker) + await controller.aclose() + await controller.aclose() + assert views == [] + + asyncio.run(scenario()) + + +def test_a_failed_run_reports_the_failure_and_publishes_nothing(make_worker): + async def scenario(): + controller, store, _views = build(make_worker, {"fail_at_step": ["left", 0]}) + await controller.start(OWNER, LEFT) + view = await finish(controller) + assert view.outcome.kind is OutcomeKind.FAILED + assert view.outcome.error.code == "step_failed" + assert "failed" in view.notice + assert view.outcome.published is None + assert store.list_published() == [] + + asyncio.run(scenario()) + + +def test_a_worker_that_never_reports_a_result_is_uncertain_not_complete( + make_worker, +): + async def scenario(): + controller, store, _views = build(make_worker, {"no_result": True}) + await controller.start(OWNER, LEFT) + view = await finish(controller) + assert view.outcome.kind is OutcomeKind.UNCERTAIN + assert "uncertain" in view.notice + assert "completed" not in view.notice + assert store.list_published() == [] + + asyncio.run(scenario()) + + +def test_a_worker_that_cannot_start_finishes_failed_and_frees_the_controller( + tmp_path, make_worker +): + async def scenario(): + controller, _store, _views = build(make_worker, {"exit_before_ready": True}) + await controller.start(OWNER, LEFT) + await wait_for(lambda: controller.view().stage is Stage.FINISHED) + assert controller.view().outcome.kind is not OutcomeKind.COMPLETED + assert controller.active is False + + asyncio.run(scenario()) + + +def test_an_invalid_start_leaves_the_controller_usable(make_worker): + async def scenario(): + controller, _store, _views = build(make_worker) + with pytest.raises(proto.ProtocolError): + await controller.start(OWNER, ()) + assert controller.active is False + await controller.start(OWNER, LEFT) + await controller.aclose() + + asyncio.run(scenario()) + + +def test_owner_attempt_names_the_active_attempt_of_its_owner_only(make_worker): + async def scenario(): + controller, _store, _views = build(make_worker, ONE_HOLD) + assert controller.owner_attempt(OWNER) is None + await controller.start(OWNER, LEFT) + await at_prompt(controller) + assert controller.owner_attempt(OWNER) is not None + assert controller.owner_attempt(OTHER) is None + await controller.aclose() + assert controller.owner_attempt(OWNER) is None + + asyncio.run(scenario()) + + +@pytest.mark.parametrize("second_owner", [OWNER, OTHER]) +def test_a_late_disconnect_bound_to_an_earlier_attempt_cannot_cancel_a_later_one( + make_worker, commands_log, second_owner +): + async def scenario(): + controller, store, _views = build(make_worker, ONE_HOLD) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + earlier = controller.owner_attempt(OWNER) + await controller.cancel(OWNER) + await wait_for(lambda: controller.view().stage is Stage.FINISHED) + await controller.start(second_owner, LEFT) + await at_prompt(controller) + later = controller.owner_attempt(second_owner) + assert later is not None + assert later is not earlier + cancels = commands(commands_log).count("cancel") + # The late timer of the earlier attempt fires now. + await controller.client_disconnected(OWNER, attempt=earlier) + await asyncio.sleep(0.3) + assert controller.active is True + assert controller.view().stage is Stage.AWAITING_CONFIRMATION + assert commands(commands_log).count("cancel") == cancels + # Bound to the right attempt, the same call does cancel it. + await controller.client_disconnected(second_owner, attempt=later) + await wait_for(lambda: controller.view().stage is Stage.FINISHED) + assert controller.view().outcome.kind is OutcomeKind.CANCELLED + assert store.list_published() == [] + + asyncio.run(scenario()) + + +def test_a_disconnect_bound_to_the_current_attempt_cancels_it_and_reaps_the_worker( + make_worker, +): + async def scenario(): + controller, _store, _views = build(make_worker, ONE_HOLD) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + await controller.client_disconnected( + OWNER, attempt=controller.owner_attempt(OWNER) + ) + await wait_for(lambda: controller.view().stage is Stage.FINISHED) + outcome = controller.view().outcome + assert outcome.kind is OutcomeKind.CANCELLED + assert outcome.exit_code == 0 + assert outcome.forced_termination is False + + asyncio.run(scenario()) + + +def test_a_bound_disconnect_by_a_non_owner_is_ignored(make_worker): + async def scenario(): + controller, _store, _views = build(make_worker, ONE_HOLD) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + await controller.client_disconnected( + OTHER, attempt=controller.owner_attempt(OWNER) + ) + await asyncio.sleep(0.3) + assert controller.active is True + await controller.aclose() + + asyncio.run(scenario()) + + +def test_a_bound_disconnect_for_a_finished_attempt_is_ignored(make_worker): + async def scenario(): + controller, store, _views = build(make_worker, ONE_HOLD) + await controller.start(OWNER, LEFT) + await at_prompt(controller) + earlier = controller.owner_attempt(OWNER) + await finish(controller) + await controller.client_disconnected(OWNER, attempt=earlier) + assert controller.view().outcome.kind is OutcomeKind.COMPLETED + assert len(store.list_published()) == 1 + + asyncio.run(scenario()) diff --git a/examples/manus_calibration/tests/test_protocol.py b/examples/manus_calibration/tests/test_protocol.py new file mode 100644 index 0000000000..87474eea79 --- /dev/null +++ b/examples/manus_calibration/tests/test_protocol.py @@ -0,0 +1,439 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import json + +import pytest +from isaaccapture_examples.manus_calibration import protocol as proto + +SHA = "a" * 64 +ATTEMPT = "cal-abc" + + +def line(message: dict) -> str: + return json.dumps({"v": 1, **message}) + + +VALID = { + "ready": {"type": "ready", "worker_version": "1", "sdk_version": None}, + "status": { + "type": "status", + "attempt": ATTEMPT, + "phase": "discovering", + "gloves": [{"side": "left", "glove_id": 3}], + }, + "prompt": { + "type": "prompt", + "attempt": ATTEMPT, + "token": "t0", + "side": "left", + "glove_id": 3, + "index": 0, + "count": 2, + "title": "Flat", + "instruction": "Lay flat", + "mode": "hold", + "duration_s": 0, + "continuous": False, + "estimated_s": 0, + }, + "step_accepted": {"type": "step_accepted", "attempt": ATTEMPT, "token": "t0"}, + "step_progress": { + "type": "step_progress", + "attempt": ATTEMPT, + "token": "t0", + "elapsed_s": 0.5, + }, + "rejected": { + "type": "rejected", + "attempt": None, + "command": "confirm", + "reason": "stale_token", + "message": "no", + "token": "t9", + }, + "exported": { + "type": "exported", + "attempt": ATTEMPT, + "side": "right", + "glove_id": 4, + "file": "cal-abc_right.mcal", + "bytes": 10, + "sha256": SHA, + }, + "result": { + "type": "result", + "attempt": ATTEMPT, + "outcome": "completed", + "files": [{"side": "left", "file": "a.mcal", "bytes": 1, "sha256": SHA}], + }, +} + + +@pytest.mark.parametrize("name", sorted(VALID)) +def test_every_documented_event_parses(name): + event = proto.parse_event(line(VALID[name])) + assert type(event).__name__.lower().replace("_", "") == name.replace("_", "") + + +def test_failed_result_carries_an_error_and_cancelled_carries_neither(): + failed = proto.parse_event( + line( + { + "type": "result", + "attempt": None, + "outcome": "failed", + "error": { + "code": "no_gloves", + "message": "none", + "side": "left", + "step": 1, + }, + } + ) + ) + assert failed.error == proto.WorkerError("no_gloves", "none", proto.Side.LEFT, 1) + cancelled = proto.parse_event( + line({"type": "result", "attempt": ATTEMPT, "outcome": "cancelled"}) + ) + assert cancelled.files == () and cancelled.error is None + + +def mutate(name: str, **changes) -> str: + message = dict(VALID[name]) + for key, value in changes.items(): + if value is KeyError: + message.pop(key) + else: + message[key] = value + return line(message) + + +@pytest.mark.parametrize( + ("name", "changes"), + [ + ("prompt", {"extra": 1}), + ("prompt", {"title": KeyError}), + ("prompt", {"index": 2}), + ("prompt", {"count": 0}), + ("prompt", {"index": True}), + ("prompt", {"mode": "hold", "duration_s": 1.0}), + ("prompt", {"mode": "motion", "duration_s": 0}), + ("prompt", {"mode": "dance"}), + ("prompt", {"instruction": ""}), + ("prompt", {"token": ""}), + ("prompt", {"side": "middle"}), + ("prompt", {"attempt": "has space"}), + ("step_progress", {"elapsed_s": -1}), + ("step_progress", {"elapsed_s": True}), + ("step_progress", {"elapsed_s": "1"}), + ("status", {"phase": "napping"}), + ("status", {"gloves": "left"}), + ( + "status", + { + "gloves": [ + {"side": "left", "glove_id": 1}, + {"side": "left", "glove_id": 2}, + ] + }, + ), + ("exported", {"file": "../escape.mcal"}), + ("exported", {"file": "/abs.mcal"}), + ("exported", {"file": "a/b.mcal"}), + ("exported", {"sha256": "ABC"}), + ("exported", {"bytes": -1}), + ("exported", {"attempt": None}), + ("rejected", {"reason": "because"}), + ("result", {"outcome": "completed", "files": KeyError}), + ("result", {"outcome": "cancelled"}), + ("result", {"outcome": "failed"}), + ("result", {"outcome": "failed", "error": {"code": "weird", "message": "x"}}), + ("ready", {"worker_version": 1}), + ("ready", {"v": 2}), + ], +) +def test_malformed_events_are_protocol_errors(name, changes): + with pytest.raises(proto.ProtocolError): + proto.parse_event(mutate(name, **changes)) + + +FILE = {"side": "left", "file": "a.mcal", "bytes": 1, "sha256": SHA} +ERROR = {"code": "internal", "message": "x"} + + +@pytest.mark.parametrize( + ("name", "changes"), + [ + ("status", {"gloves": ["left"]}), + ("status", {"gloves": [{"side": "left", "glove_id": 1, "extra": 0}]}), + ("result", {"outcome": "completed", "files": "left"}), + ("result", {"outcome": "completed", "files": ["left"]}), + ("result", {"outcome": "completed", "files": [{**FILE, "extra": 1}]}), + ("result", {"outcome": "completed", "files": [{**FILE, "side": "middle"}]}), + ("result", {"outcome": "completed", "files": [{**FILE, "sha256": "short"}]}), + ("result", {"outcome": "completed", "error": ERROR}), + ("result", {"files": KeyError, "outcome": "cancelled", "error": ERROR}), + ("result", {"outcome": "cancelled", "files": []}), + ("result", {"outcome": "failed", "error": ERROR, "files": []}), + ("result", {"files": KeyError, "outcome": "failed", "error": "boom"}), + ( + "result", + { + "files": KeyError, + "outcome": "failed", + "error": {"code": "weird", "message": "x"}, + }, + ), + ("result", {"files": KeyError, "outcome": "failed", "error": {"message": "x"}}), + ( + "result", + {"files": KeyError, "outcome": "failed", "error": {**ERROR, "extra": 1}}, + ), + ( + "result", + {"files": KeyError, "outcome": "failed", "error": {**ERROR, "step": -1}}, + ), + ( + "result", + {"files": KeyError, "outcome": "failed", "error": {**ERROR, "step": True}}, + ), + ( + "result", + {"files": KeyError, "outcome": "failed", "error": {**ERROR, "step": 1.5}}, + ), + ( + "result", + { + "files": KeyError, + "outcome": "failed", + "error": {**ERROR, "side": "middle"}, + }, + ), + ("rejected", {"token": ""}), + ("rejected", {"token": 5}), + ("rejected", {"command": ""}), + ], +) +def test_more_malformed_events_are_protocol_errors(name, changes): + with pytest.raises(proto.ProtocolError): + proto.parse_event(mutate(name, **changes)) + + +def test_a_completed_result_may_list_no_files_but_never_omit_the_field(): + event = proto.parse_event( + line( + {"type": "result", "attempt": ATTEMPT, "outcome": "completed", "files": []} + ) + ) + assert event.files == () + with pytest.raises(proto.ProtocolError): + proto.parse_event( + line({"type": "result", "attempt": ATTEMPT, "outcome": "completed"}) + ) + + +def test_every_documented_error_code_parses(): + for code in sorted(proto._ERROR_CODES): + event = proto.parse_event( + line( + { + "type": "result", + "attempt": ATTEMPT, + "outcome": "failed", + "error": {"code": code, "message": "m"}, + } + ) + ) + assert event.error.code == code + + +@pytest.mark.parametrize( + "raw", + [ + "", + "not json", + "[]", + "null", + '{"v":1}', + '{"v":1,"type":"nope"}', + '{"v":"1","type":"ready"}', + '{"v":1,"type":"step_progress","attempt":"a","token":"t","elapsed_s":NaN}', + '{"v":1,"type":"step_progress","attempt":"a","token":"t","elapsed_s":Infinity}', + b"\xff\xfe", + b"x" * (proto.MAX_LINE_BYTES + 1), + ], +) +def test_garbage_lines_are_protocol_errors(raw): + with pytest.raises(proto.ProtocolError): + proto.parse_event(raw) + + +def test_commands_round_trip_and_order_sides_left_first(): + start = json.loads( + proto.encode_start("cal-1", (proto.Side.RIGHT, proto.Side.LEFT), "/stage") + ) + assert start == { + "v": 1, + "type": "start", + "attempt": "cal-1", + "sides": ["left", "right"], + "staging_dir": "/stage", + } + assert json.loads(proto.encode_confirm("cal-1", "tok")) == { + "v": 1, + "type": "confirm", + "attempt": "cal-1", + "token": "tok", + } + assert json.loads(proto.encode_cancel("cal-1")) == { + "v": 1, + "type": "cancel", + "attempt": "cal-1", + } + assert proto.encode_cancel("cal-1").endswith(b"\n") + + +@pytest.mark.parametrize("sides", [(), (proto.Side.LEFT, proto.Side.LEFT)]) +def test_start_rejects_empty_or_duplicate_sides(sides): + with pytest.raises(proto.ProtocolError): + proto.encode_start("cal-1", sides, "/stage") + + +def test_start_rejects_a_bad_attempt_id(): + with pytest.raises(proto.ProtocolError): + proto.encode_start("no spaces", (proto.Side.LEFT,), "/stage") + + +# --- prompt: continuous steps ------------------------------------------------------------- + + +def test_a_plain_hold_prompt_is_not_continuous(): + prompt = proto.parse_event(line(VALID["prompt"])) + assert prompt.continuous is False + assert prompt.estimated_s == 0.0 + assert prompt.mode is proto.StepMode.HOLD + + +def test_a_continuous_hold_prompt_carries_its_estimate(): + prompt = proto.parse_event(mutate("prompt", continuous=True, estimated_s=3.5)) + assert prompt.continuous is True + assert prompt.estimated_s == 3.5 + assert prompt.mode is proto.StepMode.HOLD + assert prompt.duration_s == 0.0 + + +def test_an_integer_estimate_is_accepted_and_read_as_a_float(): + prompt = proto.parse_event(mutate("prompt", continuous=True, estimated_s=3)) + assert prompt.estimated_s == 3.0 + assert isinstance(prompt.estimated_s, float) + + +def test_a_timed_motion_prompt_is_not_continuous_and_has_no_estimate(): + prompt = proto.parse_event( + mutate("prompt", mode="motion", duration_s=0.4, continuous=False, estimated_s=0) + ) + assert prompt.mode is proto.StepMode.MOTION + assert prompt.duration_s == 0.4 + assert prompt.continuous is False + assert prompt.estimated_s == 0.0 + + +def test_the_prompt_dataclass_defaults_to_a_non_continuous_step(): + prompt = proto.Prompt( + attempt=ATTEMPT, + token="t0", + side=proto.Side.LEFT, + glove_id=3, + index=0, + count=1, + title="t", + instruction="i", + mode=proto.StepMode.HOLD, + duration_s=0.0, + ) + assert prompt.continuous is False + assert prompt.estimated_s == 0.0 + with pytest.raises(AttributeError): + prompt.continuous = True # frozen + + +@pytest.mark.parametrize( + "changes", + [ + # continuous needs a positive estimate + {"continuous": True, "estimated_s": 0}, + {"continuous": True, "estimated_s": 0.0}, + # continuous is a hold, never a timed motion + {"continuous": True, "estimated_s": 3, "mode": "motion", "duration_s": 0.4}, + # a non-continuous step has no estimate, whatever its mode + {"continuous": False, "estimated_s": 3}, + {"continuous": False, "estimated_s": 0.001}, + {"continuous": False, "estimated_s": 2, "mode": "motion", "duration_s": 0.4}, + ], + ids=[ + "continuous-zero-int", + "continuous-zero-float", + "continuous-motion", + "plain-estimate", + "plain-tiny-estimate", + "motion-estimate", + ], +) +def test_inconsistent_continuous_and_estimate_pairs_are_protocol_errors(changes): + with pytest.raises(proto.ProtocolError): + proto.parse_event(mutate("prompt", **changes)) + + +@pytest.mark.parametrize("missing", ["continuous", "estimated_s"]) +def test_a_prompt_missing_a_continuous_field_is_a_protocol_error(missing): + with pytest.raises(proto.ProtocolError): + proto.parse_event(mutate("prompt", **{missing: KeyError})) + + +def test_a_prompt_missing_both_continuous_fields_is_a_protocol_error(): + with pytest.raises(proto.ProtocolError): + proto.parse_event(mutate("prompt", continuous=KeyError, estimated_s=KeyError)) + + +@pytest.mark.parametrize("value", [1, 0, "true", "false", None, [], {}, 1.0]) +def test_continuous_must_be_a_strict_boolean(value): + with pytest.raises(proto.ProtocolError): + proto.parse_event(mutate("prompt", continuous=value)) + + +@pytest.mark.parametrize("value", ["3", None, True, False, [], {}, [3]]) +def test_estimated_s_must_be_a_number(value): + with pytest.raises(proto.ProtocolError): + proto.parse_event(mutate("prompt", continuous=True, estimated_s=value)) + + +@pytest.mark.parametrize( + "value", + [-1, -0.001, float("nan"), float("inf"), float("-inf")], + ids=["minus-one", "minus-tiny", "nan", "inf", "minus-inf"], +) +def test_estimated_s_must_be_finite_and_not_negative(value): + for continuous in (True, False): + with pytest.raises(proto.ProtocolError): + proto.parse_event( + mutate("prompt", continuous=continuous, estimated_s=value) + ) + + +def test_an_estimate_beyond_the_float_range_is_a_protocol_error(): + text = mutate("prompt", continuous=True, estimated_s=1).replace( + '"estimated_s": 1', '"estimated_s": 1e400' + ) + assert "1e400" in text + with pytest.raises(proto.ProtocolError): + proto.parse_event(text) + + +def test_an_extra_key_on_a_continuous_prompt_is_still_a_protocol_error(): + with pytest.raises(proto.ProtocolError): + proto.parse_event( + mutate("prompt", continuous=True, estimated_s=3, extra_field=1) + ) diff --git a/examples/manus_calibration/tests/test_server.py b/examples/manus_calibration/tests/test_server.py new file mode 100644 index 0000000000..0cc87c117b --- /dev/null +++ b/examples/manus_calibration/tests/test_server.py @@ -0,0 +1,1393 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""The page server over real sockets and a real fake-worker subprocess.""" + +from __future__ import annotations + +import asyncio +import json + +import pytest +from conftest import commands +from isaaccapture_examples.manus_calibration import server as server_module +from isaaccapture_examples.manus_calibration.worker import OutcomeKind +from web_helpers import ( + CLIENT_A, + CLIENT_B, + Sse, + api, + raw_request, + request, + run_flow, + serving, +) + +HOLD = { + "steps": [ + {"title": "A", "instruction": "a", "time": 0.0}, + {"title": "B", "instruction": "b", "time": 0.0}, + ] +} +# One hold step: the worker then waits for a confirmation, which keeps an attempt open. +WAITING = {"steps": [{"title": "A", "instruction": "a", "time": 0.0}]} +# A long motion step keeps the worker sampling after one confirmation. +SAMPLING = {"steps": [{"title": "W", "instruction": "w", "time": 3.0}]} +HEADERS_ON_EVERY_RESPONSE = { + "content-security-policy": ( + "default-src 'none'; script-src 'self'; style-src 'self'; " + "connect-src 'self'; base-uri 'none'; form-action 'none'; " + "frame-ancestors 'none'" + ), + "x-content-type-options": "nosniff", + "referrer-policy": "no-referrer", + "cache-control": "no-store", +} + + +def test_static_assets_are_served_with_their_content_types(make_worker): + async def scenario(): + async with serving(make_worker) as (server, _store): + expected = { + "/": "text/html; charset=utf-8", + "/app.js": "text/javascript; charset=utf-8", + "/style.css": "text/css; charset=utf-8", + } + for path, content_type in expected.items(): + reply = await request(server.port, "GET", path) + assert reply.status == 200, path + assert reply.headers["content-type"] == content_type + assert int(reply.headers["content-length"]) == len(reply.body) > 0 + page = (await request(server.port, "GET", "/")).body.decode() + assert "Manus glove calibration" in page + assert '