From 8942a19eb36753b5681361fede974eb45d984819 Mon Sep 17 00:00:00 2001 From: niksacdev Date: Tue, 15 Sep 2026 12:22:47 +0530 Subject: [PATCH 1/2] Add repeatable ABC and pi0.5 bimanual evaluations --- .github/workflows/ci.yml | 3 +- CHANGELOG.md | 4 + CLAUDE.md | 10 +- README.md | 13 +- docs/README.md | 1 + .../ADR-038-abc-bimanual-episodes.md | 138 ++++++ docs/architecture/README.md | 3 + docs/product/abc-pi05-comparison.md | 242 +++++++++++ docs/product/reusable-evaluation-core.md | 11 +- docs/ux/2026-09-15-abc-bimanual-review.md | 50 +++ docs/ux/abc-bimanual-user-journey.md | 65 +++ examples/abc-pi05/.gitignore | 3 + examples/abc-pi05/README.md | 151 +++++++ examples/abc-pi05/dataset.py | 401 ++++++++++++++++++ examples/abc-pi05/requirements.txt | 3 + examples/abc-pi05/train.py | 257 +++++++++++ frontend/benchmarks.html | 2 +- frontend/bimanual.css | 44 ++ frontend/bimanual.html | 59 +++ frontend/bimanual.js | 211 +++++++++ frontend/history.html | 2 +- src/rove/benchmarks/api.py | 2 + src/rove/benchmarks/report.py | 157 ++++++- src/rove/bimanual/__init__.py | 1 + src/rove/bimanual/abc_process.py | 135 ++++++ src/rove/bimanual/adapter.py | 190 +++++++++ src/rove/bimanual/api.py | 244 +++++++++++ src/rove/bimanual/cli.py | 191 +++++++++ src/rove/bimanual/config.py | 247 +++++++++++ src/rove/bimanual/episode.py | 206 +++++++++ src/rove/bimanual/pi05_process.py | 50 +++ src/rove/bimanual/pi05_weights.py | 80 ++++ src/rove/bimanual/runtime_probe.py | 66 +++ src/rove/evaluation/context.py | 143 +++++++ src/rove/evaluation/registry.py | 25 +- src/rove/evaluation/trials.py | 3 + src/rove/evaluation/worker.py | 150 ++++--- src/rove/ui/cli.py | 3 +- tests/bimanual_ui.cjs | 124 ++++++ tests/test_abc_pi05_data.py | 362 ++++++++++++++++ tests/test_bimanual_adapter.py | 266 ++++++++++++ tests/test_bimanual_api.py | 247 +++++++++++ tests/test_bimanual_cli.py | 86 ++++ tests/test_bimanual_config.py | 83 ++++ tests/test_bimanual_episode.py | 138 ++++++ tests/test_bimanual_process_cleanup.py | 125 ++++++ tests/test_episode_report.py | 183 ++++++++ tests/test_evaluation_context.py | 242 +++++++++++ 48 files changed, 5323 insertions(+), 99 deletions(-) create mode 100644 docs/architecture/ADR-038-abc-bimanual-episodes.md create mode 100644 docs/product/abc-pi05-comparison.md create mode 100644 docs/ux/2026-09-15-abc-bimanual-review.md create mode 100644 docs/ux/abc-bimanual-user-journey.md create mode 100644 examples/abc-pi05/.gitignore create mode 100644 examples/abc-pi05/README.md create mode 100644 examples/abc-pi05/dataset.py create mode 100644 examples/abc-pi05/requirements.txt create mode 100644 examples/abc-pi05/train.py create mode 100644 frontend/bimanual.css create mode 100644 frontend/bimanual.html create mode 100644 frontend/bimanual.js create mode 100644 src/rove/bimanual/__init__.py create mode 100644 src/rove/bimanual/abc_process.py create mode 100644 src/rove/bimanual/adapter.py create mode 100644 src/rove/bimanual/api.py create mode 100644 src/rove/bimanual/cli.py create mode 100644 src/rove/bimanual/config.py create mode 100644 src/rove/bimanual/episode.py create mode 100644 src/rove/bimanual/pi05_process.py create mode 100644 src/rove/bimanual/pi05_weights.py create mode 100644 src/rove/bimanual/runtime_probe.py create mode 100644 src/rove/evaluation/context.py create mode 100644 tests/bimanual_ui.cjs create mode 100644 tests/test_abc_pi05_data.py create mode 100644 tests/test_bimanual_adapter.py create mode 100644 tests/test_bimanual_api.py create mode 100644 tests/test_bimanual_cli.py create mode 100644 tests/test_bimanual_config.py create mode 100644 tests/test_bimanual_episode.py create mode 100644 tests/test_bimanual_process_cleanup.py create mode 100644 tests/test_episode_report.py create mode 100644 tests/test_evaluation_context.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8ebcf5b..f986c89 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,7 +32,7 @@ jobs: sudo apt-get install -y ffmpeg - name: Static checks run: | - uvx ruff==0.16.7 check src tests scripts + uvx ruff==0.16.7 check src tests scripts examples/abc-pi05 uv run --frozen python -m compileall -q src uv run --frozen bandit -q -r src -c pyproject.toml node --check frontend/strategy-table.js @@ -41,6 +41,7 @@ jobs: node --check frontend/tooltips.js node --check frontend/sample-cases.js node --check frontend/benchmarks.js + node --check frontend/bimanual.js node --check frontend/history.js node --check frontend/history-traces.js node --check frontend/datasets.js diff --git a/CHANGELOG.md b/CHANGELOG.md index 08b2de6..0c72156 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ Format: each entry includes the date, a short title, and a description of what c ## Unreleased +### Compare ABC-VLA and adapted pi0.5 on bimanual episodes + +Run compatible policies through repeated ABC simulator episodes using existing Trials, Campaigns, evidence, traces and reports. A guided comparison and matching CLI preserve scene resets, checkpoint identities, task outcomes and policy-call measurements; a separate ABC/XDOF conversion and pi0.5 training recipe prepares adapted checkpoints. Real data conversion and CPU simulation are validated; GPU training and learned-policy comparison remain pending. + ### Explore ABC and XDOF in Jupyter An executable notebook follows a prepared robot demonstration through camera/state inspection, optional ABC-VLA inference, complete simulator episodes and a small fine-tuning exercise. It uses a separate ABC environment and records actual evidence without adding training to the ROVE application. *(PR #33)* diff --git a/CLAUDE.md b/CLAUDE.md index 9bbd344..5f221ac 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -16,7 +16,7 @@ in [concepts](docs/product/concepts.md) and ROVE evaluates robotics agent pipelines by running real inference through VLM+VLA+LLM+Agent combinations. It orchestrates a standardized 4-step pipeline (perceive → plan → act → verify) across multiple model combinations in parallel, producing ranked comparisons of success rate, latency, and cost for each agent configuration. -ROVE is inference-only. It does not train, fine-tune, or modify models. It runs them in realistic robotics pipelines and measures the outcomes. +ROVE's evaluation engine runs models and measures outcomes. Explicit standalone preparation examples may fine-tune a model before evaluation; training never starts implicitly inside a trial or campaign. - **PyPI package**: `rove-eval` - **Python import**: `import rove` @@ -97,8 +97,10 @@ lifecycle reusable. See [reusable core](docs/product/reusable-evaluation-core.md `rove.evaluations` integrations accept structured inputs, run separate candidate and grader calls, and share SQLite history, evidence, deadlines and reporting. Generic case authoring is currently manifest/CLI based; do not claim the robotics -case builder or strategy catalog already supports arbitrary domains. ABC rollout -hosting and Azure/Fabric integrations are outside this pivot. +case builder or strategy catalog already supports arbitrary domains. The built-in +`abc-bimanual` executor now hosts ABC simulation episodes through this lifecycle; +see [ADR-038](docs/architecture/ADR-038-abc-bimanual-episodes.md). Azure/Fabric +deployment remains future work. ## Changelog @@ -149,7 +151,7 @@ When creating a PR, always update `CHANGELOG.md`: ## What NOT to Build -- No model training or fine-tuning (ROVE is inference-only — it consumes models, not produces them) +- No training inside evaluation workers. Standalone preparation recipes may produce checkpoints which Trials and Campaigns then consume explicitly. - No real robot control (simulation only — output is data, not robot motion) - No model serving infrastructure (ROVE calls models, does not host them) - Keep robotics behavior compatible while reusing Trials/Campaigns for other evaluation domains. Domain-specific inputs and grading belong behind execution adapters; do not build a second campaign engine. diff --git a/README.md b/README.md index 5ee0818..015cdfa 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,15 @@ For a separate hands-on learning exercise, open the run a released VLA in ABC's simulator, and optionally explore fine-tuning. Its GPU steps use an independent environment and are not ROVE application features. +The experimental [ABC-VLA versus pi0.5 comparison](docs/product/abc-pi05-comparison.md) +connects complete bimanual simulator episodes to ROVE's Trials and Campaigns. +Register compatible checkpoints with `rove bimanual setup`, preview fixed scene +cases and repeated policy seeds, then run comparisons with preserved measurements +and evidence. The pi0.5 checkpoint must be adapted to ABC's robot. Local checks +cover the integration contracts, a real ABC CPU simulation with rendering, and +LeRobot dataset conversion/readback. GPU inference, fine-tuning and model +performance remain unverified on the development host. + ![ROVE dashboard](docs/images/rove-welcome.png) ## Try it locally @@ -123,7 +132,9 @@ ROVE has a validated timing, inertia and actuator-mapping contract. A VLM's success judgment is not measured task success. Stage attribution is a heuristic debugging hypothesis, not a demonstrated causal root cause. Perception and planning diagnostics do not control VLA actions in the parallel pipeline. -There is no real closed-loop simulator integration yet; `mock-sim` is a stub. +The experimental ABC integration adds a separate closed-loop episode path; +its [validation scope](docs/product/abc-pi05-comparison.md) remains explicit. +The original image-based pipeline's `mock-sim` remains a stub. ## Connect your models diff --git a/docs/README.md b/docs/README.md index c625c51..6f46087 100644 --- a/docs/README.md +++ b/docs/README.md @@ -6,6 +6,7 @@ | Read | Purpose | | --- | --- | +| [ABC-VLA versus pi0.5](product/abc-pi05-comparison.md) | Experimental bimanual episode comparison, repeated scene cases, measured outcomes and checkpoint adaptation | | [ABC + XDOF Jupyter tutorial](product/abc-jupyter-tutorial.md) | Standalone demonstration-to-policy notebook with optional GPU evaluation and fine-tuning | | [Product specification](PRODUCT_SPEC.md) | Core vision, founding personas, campaign model and delivery status | | [Concepts](product/concepts.md) | Cases, trials, campaigns, strategies, datasets and provenance | diff --git a/docs/architecture/ADR-038-abc-bimanual-episodes.md b/docs/architecture/ADR-038-abc-bimanual-episodes.md new file mode 100644 index 0000000..06f7051 --- /dev/null +++ b/docs/architecture/ADR-038-abc-bimanual-episodes.md @@ -0,0 +1,138 @@ +# ADR-038: Run ABC bimanual episodes through the shared evaluation lifecycle + +## Status + +Accepted. Experimental integration; GPU policy inference, simulator execution and +pi0.5 adaptation require validation on a suitable host before reporting model +performance. Local tests validate contracts, failure handling and durable records. + +## Context + +The user wants to compare ABC-VLA and pi0.5 on the same bimanual robot task and +measure whether adapting a checkpoint improves the results. The image-based +ROVE pipeline produces a single action chunk. It cannot establish completion of +a task requiring repeated observations and actions. + +ABC already supplies the simulator and task evaluator. The reusable evaluation +core from [ADR-036](ADR-036-reusable-evaluation-core.md) already supplies isolated +trial workers, timeouts, campaign attempts, SQLite records, evidence and reports. +The standalone notebook in [ADR-037](ADR-037-standalone-abc-notebook.md) remains a +separate learning exercise; it does not provide the application lifecycle. + +## Decision + +Add the built-in `abc-bimanual` evaluation adapter. It uses the existing +`EvaluationAdapter.predict` and `grade` boundary and the existing campaign runner. +The candidate phase executes a complete episode; the grader interprets recorded +simulator evidence under the frozen success criterion. + +ABC's environment and native VLA run in their pinned runtime. A pi0.5 strategy +uses a separate LeRobot runtime and compatible checkpoint. Interprocess messages +carry current observations and native action chunks. They do not turn ROVE's +application environment into a combined ML dependency installation. + +```mermaid +flowchart TB + UI[ROVE browser and CLI] --> P[Host-registered comparison profile] + P --> C[Existing trial and campaign services] + C --> W[Isolated ROVE evaluation worker] + subgraph PG[One owned process group per episode] + W --> A[ABC environment and optional ABC-VLA runtime] + A <--> L[Optional pi0.5 LeRobot runtime] + end + W --> S[Existing SQLite trial and campaign journals] + W --> E[Existing content-addressed evidence assets] + S --> R[Reports, traces and history] + E --> R + W -. Deadline or cancellation terminates descendants .-> PG +``` + +## Preserve case and attempt semantics + +A case fixes the task and scene reset. Campaign attempt seeds control policy +randomness separately. Each policy receives its own post-action observations, +starting from the same requested scene conditions. A fresh environment per trial +prevents prior robot state from leaking between policies or repeated episodes. + +The environment revision, reset identity, task prompt, action interpretation, +budget and success mode are part of the comparison conditions. Checkpoint and +native processor contents identify the strategy. A different checkpoint can be +compared while keeping those conditions fixed; a changed evaluator or task is +reported as changed conditions. + +ABC's `ever` and `final` success signals are separate measurements. The configured +signal supplies the binary outcome. Inference, simulated duration and complete +worker duration remain separate measures. No collision, intervention or physical +safety measure is inferred merely because the simulator executed. + +## Optional evaluation context + +Existing installed adapters keep their current `predict` and `grade` signatures. +An adapter may implement `bind_context(context)` to receive: + +- A copied declaration of the frozen environment. +- A durable, clocked progress emitter. +- Bounded JSON, binary and local-file artifact recording through managed assets. +- The identities of currently recorded, integrity-checked evidence. + +Core evidence IDs are reserved. An assessment may reference only evidence that +actually exists and remains available. Private case references are passed to the +grader separately, not included in the context environment or candidate input. +This contract separates responsibilities; trusted Python adapters still execute +on the host and are not a security sandbox. + +The resolver fingerprints optional binding source as well as candidate/grader +source. Runtime fingerprints retain the surrounding ROVE implementation identity. +The worker records exception type and failed stage without persisting provider +exception strings that may contain credentials or customer inputs. + +## Configuration and process ownership + +Operator setup registers local paths and fingerprints outside the browser. Browser +requests select known strategies, scene seeds, policy seeds and bounded budgets. +Previews show readiness and planned trial count before execution. No arbitrary +Python module or executable path is loaded from a browser-supplied manifest. + +Descendants inherit the ROVE worker's process group. They must not detach into a +new session, because the existing timeout/cancellation cleanup kills the owned +group. Evidence and progress are written before continuing so interruption does +not remove already-recorded observations. Unfinished trials retain unknown +outcomes and a distinct execution status. + +## Alternatives and consequences + +**Extend the image-only VLA adapter to emit a longer chunk.** This would still +omit observations after actions and would not establish completion. It would also +misrepresent a single model prediction as an entire episode. + +**Add a second ABC-specific campaign runner.** This would duplicate identity, +resume, cancellation, persistence and statistical reporting. Reusing the generic +execution boundary keeps Trials and Campaigns consistent. + +**Install both ML stacks in ROVE.** This creates dependency conflicts and couples +the UI/server environment to CUDA and model-specific versions. Separate runtimes +add process overhead but make ownership and validation explicit. + +Fresh model processes are slower than a persistent serving pool. This is an +acceptable initial tradeoff for reproducibility and bounded resource use. +Checkpoint loading is therefore visible in worker duration and excluded from +reported policy-call latency. Throughput optimization requires its own design +and must preserve episode reset and mutable policy-state isolation. + +## Validation and delivery + +Local tests cover environment binding, reference separation, managed artifact +limits, evidence integrity, inherited binding provenance, error/cancellation +records, report measurement coverage and existing generic executor compatibility. +The real ABC CPU environment completed a 30-control-step smoke test with rendered +observations. Conversion and LeRobot readback also passed for two genuine ABC +training episodes containing 1,870 frames. Neither check ran a learned policy. +Live GPU validation must demonstrate both policies controlling the same supported +ABC robot, compatible native preprocessing, matched resets, finite actions, +complete episode evidence and correct success grading. + +GPU inference and training remain unrun. No published-model comparison is +established until those episodes actually run. +Fine-tuning remains an explicit preparation activity, not a hidden action triggered +by evaluation. See the [product workflow](../product/abc-pi05-comparison.md) for +personas, outcome meanings and the customer's checkpoint-improvement journey. diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 48b3c20..ad0b31c 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -5,6 +5,9 @@ [ADR-037: Standalone ABC notebook](ADR-037-standalone-abc-notebook.md) keeps the XDOF/ABC learning exercise and its optional training environment separate from ROVE. +[ADR-038: ABC bimanual episodes](ADR-038-abc-bimanual-episodes.md) adds a closed-loop +executor using the same Trials, Campaigns, evidence and reporting lifecycle. + Use [current implementation](current-implementation.md) to explain the shipped code, the [target architecture](target-architecture.md) and [system diagram](system-diagram.md) to explain the accepted direction, and [the product specification](../PRODUCT_SPEC.md) diff --git a/docs/product/abc-pi05-comparison.md b/docs/product/abc-pi05-comparison.md new file mode 100644 index 0000000..880d147 --- /dev/null +++ b/docs/product/abc-pi05-comparison.md @@ -0,0 +1,242 @@ +# Compare ABC-VLA and pi0.5 on a bimanual task + +ROVE compares compatible robot policies on the same simulated job, records the +complete episodes, and preserves the evidence needed to compare a later checkpoint. +The first integration uses ABC's **put plastic bottles in the bin** task and +compares ABC-VLA with a pi0.5 checkpoint adapted to ABC's robot. + +**Status:** Experimental implementation. Local contract, persistence and reporting +tests exercise the integration boundary. Real GPU inference, closed-loop simulator +execution and pi0.5 fine-tuning have not been validated on the development host. +No model ranking or customer-performance result is claimed by the included tests. + +## The customer decision + +The useful question is: **Which policy meets this robot's task requirements, and +did adaptation improve it?** The first example establishes the evaluation path +using an existing simulator task. A customer's different workcell, robot or job +requires a compatible environment, inputs, actions and task evaluator. + +| Persona | Job to do | ROVE provides | +| --- | --- | --- | +| Robotics researcher | Compare ABC-VLA and pi0.5 under matched conditions | Fixed scene cases, repeated policy seeds and separately recorded episodes | +| ML engineer | Decide whether an adapted checkpoint improves task performance | Success curves, inference measurements, recorded outcomes and checkpoint identity | +| Platform / AI infrastructure team | Run evaluations automatically and retain reports | Local profile setup, CLI/API execution, SQLite history and portable reports | + +ABC supplies the robot environment, rendering, physics and task-success signal. +LeRobot supplies the pi0.5 policy runtime and its native processors. ROVE owns +configuration, trial scheduling, durable recording, comparison and reporting. +The existing image-based robotics pipeline remains available. + +```mermaid +flowchart LR + C[Cases: task and fixed scene resets] --> R[ROVE comparison] + S[Strategies: frozen compatible checkpoints] --> R + R --> A[ABC-VLA] + R --> P[pi0.5 through LeRobot] + A --> E[Separate ABC episode for each policy] + P --> E + E --> O[Environment goal state and recorded trajectory] + O --> G[Configured episode assessment] + G --> H[Trials, campaign report and checkpoint history] +``` + +## Cases, strategies and repeats + +A case identifies the ABC task and a fixed scene reset seed. The policy gets the +current three camera observations, measured robot state and task instruction. +It does not get the demonstration's future actions or an evaluation reference. + +A strategy selects a compatible policy checkpoint and runtime. Both policies must +use ABC's two-arm state/action conventions and 14 action dimensions. Merely +changing a LIBERO checkpoint's configured dimensions does not adapt that model. +The camera mapping, joint order, gripper conventions and native normalization +must agree with the checkpoint's training. + +The campaign's attempt seeds control policy randomness. They do not silently +change the scene. For example, **3 scene cases × 2 strategies × 3 policy seeds = +18 planned trials**. The two policies therefore receive matched starting scenes; +their later observations differ because each policy executes its own actions. + +Each trial starts a new process and environment. The loop requests an action +chunk, executes the configured prefix, observes the changed scene and requests +the next chunk. The default budget is 3,540 executed steps, with up to 15 actions +executed between predictions. Policy loading and simulator initialization are +part of the trial deadline, but have separate meaning from inference latency. + +```mermaid +sequenceDiagram + participant R as ROVE trial worker + participant E as ABC environment + participant P as Selected policy + participant S as SQLite and managed assets + R->>E: Reset the configured scene + E-->>R: Initial observations and reset evidence + R->>S: Persist episode start + loop Until episode budget or environment termination + R->>P: Current cameras, state and task + P-->>R: Native action chunk + R->>E: Execute configured action prefix + E-->>R: Updated observations and goal state + R->>S: Persist progress and measurements + end + R->>S: Preserve summary and trajectory + R->>R: Grade recorded environment evidence + R->>S: Complete trial with pass, fail or unknown +``` + +## What success means + +The profile explicitly chooses the simulator's success interpretation: + +- **Ever:** the environment reports the task goal at some point in the episode. +- **Final:** the environment reports the task goal at the episode's end. + +The initial default is `ever`, matching the upstream ABC success interpretation. +Both measurements remain visible so temporarily reaching the goal can be +distinguished from leaving the scene in the desired state. The configured choice +is frozen with the campaign; changing it creates different assessment conditions. + +The grader uses recorded simulator state. A policy's text claim or a successfully +produced action chunk does not determine success. Missing, malformed or changed +evidence leaves the result unknown. Errors, timeouts and cancellations remain +execution failures with unresolved outcomes; they are not silently removed from +the planned denominator. + +## What the report shows + +| Measure | Meaning | +| --- | --- | +| Success rate | The fraction meeting the frozen goal criterion; unresolved trials remain explicit | +| pass@k | Estimated probability of at least one success in k attempts of the same strategy on a case | +| pass^k | Estimated probability that all k attempts succeed | +| Goal reached / goal at end | Distinct simulator measurements retained even when only one supplies the grade | +| Executed steps and simulated duration | How much of the episode was executed; simulated duration is not physical cycle time | +| Policy predictions | Number of action chunks requested during the episode | +| Inference latency and time | Measured policy-call timing, separate from model loading and simulator setup | +| Execution and assessment time | The complete worker duration, including initialization and grading | + +The episode-performance table appears near the top of the campaign report. +Every value shows how many planned episodes supplied that measurement. Units and +evidence quality are not pooled. An average of episode p95 inference latencies is +labeled as such; it is not presented as a pooled p95 across all predictions. + +Small seeded examples demonstrate the workflow rather than establish a reliable +model ranking. Repeated deterministic outputs do not establish independent +success probabilities. Matched scene seeds help paired comparisons, but GPU +numerics, policy randomness and the simulator version still matter. + +Policy-call timing includes the selected runtime's adapter overhead. In particular, +pi0.5 communicates through a separate process; its observation transfer and reply +are part of that elapsed time. It is not a kernel-only accelerator benchmark. +Reports also check actual initial-state fingerprints: differing states produce +a prominent unmatched-comparison warning, while missing fingerprints leave +matching conditions unconfirmed. + +## Reproducibility and operating boundary + +Create a profile, replace its placeholder paths with installed runtimes and +compatible local checkpoints, then register it: + +```bash +rove bimanual init --output abc-profile.json +# Edit abc-profile.json before continuing. +rove bimanual setup --config abc-profile.json --store .rove/benchmarks +rove bimanual doctor --store .rove/benchmarks +``` + +`setup` requires every model in the profile to be available. To establish the +ABC-VLA baseline before adapting pi0.5, remove the `pi05` strategy from that +working profile, register ABC-VLA, then add and register pi0.5 after adaptation. +This is explicit setup; evaluation does not produce a missing adapted checkpoint. + +Preview the complete two-policy comparison before launching: + +```bash +rove bimanual plan --strategy abc-vla --strategy pi05 \ + --scene 0 --scene 1 --scene 2 --seed 11 --seed 29 --seed 47 +``` + +This plan contains 18 trials. A single scene and policy seed can first be used for +a standalone comparison, producing two saved trials without creating a campaign: + +```bash +rove bimanual trial --strategy abc-vla --strategy pi05 --scene 0 --seed 11 +``` + +Run the repeated campaign and write an HTML report: + +```bash +rove bimanual run --name "Bottles policy comparison" \ + --strategy abc-vla --strategy pi05 \ + --scene 0 --scene 1 --scene 2 --seed 11 --seed 29 --seed 47 \ + --output bottles-comparison.html +``` + +These commands default to `.rove/benchmarks`; use the same `--store` for setup and +execution when choosing another location. Campaigns and trials remain inspectable +through the existing browser history. Standard `rove benchmark report` and trial +inspection commands can read the same persisted records. + +The campaign command retains its report before returning a nonzero exit status +for interrupted, missing or unassessed executions. An assessed episode that did +not meet its success criterion is a valid evaluation and does not make the +command fail. This distinction lets automated workflows detect broken evaluation +jobs without confusing them with a model's measured failures. + +Setup registers an operator-owned profile with installed interpreter paths, +checkpoint contents, runtime fingerprints and pinned ABC source. The browser +selects registered strategies and bounded episode settings; it cannot supply an +arbitrary executable path. Launch does not implicitly download a model or install +another runtime. + +ABC and pi0.5 use separate Python environments. One ROVE worker owns the whole +episode process group, so timeout or cancellation terminates its descendants. +Progress and already-saved assets survive interruption. This local process +boundary controls lifecycle and dependency conflicts; it is not a sandbox for +untrusted model code. + +The worker's optional evaluation context supplies declared environment settings, +durable progress and bounded managed evidence. It does not add private references +to candidate inputs. Each artifact has a content hash; grading may cite only +recorded, available evidence. Large media must be split into bounded assets. + +Existing ABC/XDOF observation imports remain useful for inspecting demonstrations. +An imported video frame is not automatically an executable simulator case. This +comparison uses ABC's supported task/reset environment explicitly. Training data +and held-out evaluation cases need separate provenance; simulator seeds alone do +not prove that a pretrained model has never seen the underlying task. + +## Improve a checkpoint + +Run the released ABC-VLA baseline and a compatible pi0.5 candidate, inspect failed +episodes and timing, then adapt a new checkpoint outside the evaluation worker. +Register it as a new strategy and rerun the frozen cases, policy seeds, action +budget and success criterion. Preserve the training data split and processor +configuration beside that checkpoint. + +The local [dataset conversion](../../examples/abc-pi05/dataset.py) and +[pi0.5 adaptation recipe](../../examples/abc-pi05/train.py) prepare training data +and an explicit training plan outside the ROVE evaluation process. Their resulting +checkpoint must still pass runtime and embodiment checks before evaluation. + +A fine-tuning smoke test verifies that the training path produces a usable +checkpoint; it does not establish useful specialization. A performance claim +requires held-out episode results. Azure deployment, Fabric integration and +physical robot control are separate follow-up work. + +## Validation status + +The real ABC simulator has completed a 30-control-step CPU smoke test with camera +rendering. Two genuine ABC training episodes, totaling 1,870 frames, have also +been converted into a LeRobot dataset and read back successfully. These checks +exercise the environment and data path; they do not measure a learned policy. + +Local regression tests cover the execution contracts, evidence persistence, +reporting and command failure behavior. GPU inference for ABC-VLA and pi0.5, +fine-tuning, and a learned-policy performance comparison remain unrun. + +See [ADR-038](../architecture/ADR-038-abc-bimanual-episodes.md) for the process and +storage decision, [the reusable core](reusable-evaluation-core.md) for the shared +lifecycle, and [the standalone notebook](abc-jupyter-tutorial.md) for the original +ABC/XDOF learning exercise. diff --git a/docs/product/reusable-evaluation-core.md b/docs/product/reusable-evaluation-core.md index 0bc19db..be8d29a 100644 --- a/docs/product/reusable-evaluation-core.md +++ b/docs/product/reusable-evaluation-core.md @@ -156,11 +156,12 @@ environment identities must be supplied in configuration; ROVE does not independently resolve or verify them. Conservative comparison checks can reject a changed adapter even when its author believes the results comparable. -This is preparation for future integrations. It does not run ABC episodes, turn -XDOF demonstrations into executable scenes, deploy Azure workers, export to Fabric -or provide a generic LLM grader. Those integrations need their own tested adapters -and evidence contracts. The current [ABC import](abc-dataset-support.md) retains -its existing observation-import scope. +The core does not provision environments, turn XDOF demonstrations into executable +scenes, deploy Azure workers, export to Fabric or provide a generic LLM grader. +The experimental [ABC bimanual integration](abc-pi05-comparison.md) adds a specific +episode executor through this boundary, with its own runtime and evidence contract. +The existing [ABC import](abc-dataset-support.md) retains its observation-import +scope. Generic authoring remains distinct from that supported simulator preset. ## Implementation evidence diff --git a/docs/ux/2026-09-15-abc-bimanual-review.md b/docs/ux/2026-09-15-abc-bimanual-review.md new file mode 100644 index 0000000..7d3e7d0 --- /dev/null +++ b/docs/ux/2026-09-15-abc-bimanual-review.md @@ -0,0 +1,50 @@ +# ABC bimanual interface review + +## Design decisions + +The comparison uses ROVE's shared Foundry-inspired tokens, navigation, controls, +strategy progress lanes and trace inspector. The model list is a table with +selection, strategy identity, model type and readiness columns. The large trial +calculator updates as the user changes scene or model seeds. + +The task is intentionally fixed to the supported bottles task in this first +integration. The interface does not imply support for arbitrary robots or +customer scenes. Operator-controlled interpreter paths, checkpoints and runtime +configuration are managed through the CLI, not submitted by the browser. + +Success has a visible explanation before review. Review displays the exact +choice of goal reached during the episode or goal satisfied at the end. Model +generated text is not presented as measured simulator success or as available +AI improvement advice. + +## Accessibility and behavior checks + +- Native labels, inputs, tables, buttons and links support keyboard operation. +- Focus moves to review, running and results headings after transitions. +- Readiness and outcomes have text labels; color alone carries no meaning. +- Status updates use live regions, while action controls retain native states. +- Tables can scroll horizontally on narrow screens; form columns stack. +- Reduced-motion preferences disable the panel entrance animation. +- Model labels and error text use text nodes, not executable HTML. +- Review precedes every launch, and unready models cannot be launched. +- Inspecting evidence preserves the progress view instead of replacing it. + +## Validation scope + +Automated DOM tests exercise empty setup, unavailable model selection, safe label +rendering, the dynamic count, review without execution, edit invalidation, +departure warning, live progress, trace navigation, reload of campaign results +and rejection of changed configuration. API tests check bounded requests, +operator-path isolation, selected-model readiness, configuration confirmation, +shared campaign execution, early durable trial IDs and cancellation. + +Browser inspection checked the real missing-runtime page and a separate, +explicitly labelled UI fixture for the configured model table and review +transition. The fixture could not launch models or write evaluation records. +The review panel showed the count, criterion and distinct edit/run actions +without clipping at the inspected desktop viewport. + +These checks validate interface behavior and transport contracts. They do not +establish trained-policy performance, task success on real hardware, or a +completed human usability study. GPU execution remains a separate validation +step. diff --git a/docs/ux/abc-bimanual-user-journey.md b/docs/ux/abc-bimanual-user-journey.md new file mode 100644 index 0000000..02b43aa --- /dev/null +++ b/docs/ux/abc-bimanual-user-journey.md @@ -0,0 +1,65 @@ +# ABC bimanual comparison journey + +## Persona and goal + +The robotics developer wants to compare ABC-VLA and an adapted π0.5 on the same +two-arm task, inspect failures, and decide which checkpoint to improve. The +evaluation operator prepares the GPU runtime and checkpoints. These roles may be +the same person. Both need a repeatable CLI path as well as a desktop browser. + +## Starting point + +ROVE already records trials and campaigns. The ordinary image-based trial form +does not describe a simulator episode, a scene reset seed, or an adapted action +interface. Reusing that form without showing these distinctions would suggest +that one image and one action chunk establish task completion. + +## Journey + +```mermaid +flowchart LR + A[Trials or Campaigns] --> B[Compare bimanual models] + B --> C{Runtime configured?} + C -->|No| D[CLI setup guide and readiness check] + D --> C + C -->|Yes| E[Choose scenes, models and attempt seeds] + E --> F[Review frozen criterion and trial count] + F -->|Edit| E + F -->|Run| G[Shared strategy progress and live traces] + G --> H[Saved trial outcomes and campaign report] + H --> I[Inspect evidence or compare another checkpoint] +``` + +The initial page keeps task, model selection, success explanation and trial count +visible together. Review is a separate explicit gate. Selecting a model or seed +never starts inference. Changing configuration invalidates review. + +A standalone comparison uses one scene seed and one model seed. Each model +creates an ordinary saved trial. A campaign repeats the same scene cases with +the selected model seeds and creates the existing campaign report. No separate +result database is introduced. + +## What the user can distinguish + +- A scene seed chooses a fixed starting condition. A model seed identifies a + repeated attempt on that case. +- A strategy identifies a model checkpoint and its preprocessing. The simulator + and action budget are shared. +- Execution progress reports completed work. The simulator's configured task + criterion determines the outcome separately. +- A missing assessment is displayed as **Not assessed**, never as failed robot + performance. +- Inspecting a trial opens its existing trace view in another tab so the + comparison progress stays available. + +## Recovery + +Missing setup shows the operator's CLI path and setup guide. Unavailable models +remain visible with their reason and cannot be selected. A changed profile +requires a new review before launch. A reload can restore live campaign or trial +job progress through its URL; after a server restart, durable standalone trials +remain in the ordinary Trials browser even if the transient job is gone. + +Stopping a comparison awaits cancellation of the existing execution task. +Recorded evidence is preserved. Leaving an edited, unrun comparison warns that +the draft has not been saved. diff --git a/examples/abc-pi05/.gitignore b/examples/abc-pi05/.gitignore new file mode 100644 index 0000000..dd21475 --- /dev/null +++ b/examples/abc-pi05/.gitignore @@ -0,0 +1,3 @@ +.venv/ +.work/ +outputs/ diff --git a/examples/abc-pi05/README.md b/examples/abc-pi05/README.md new file mode 100644 index 0000000..049fd82 --- /dev/null +++ b/examples/abc-pi05/README.md @@ -0,0 +1,151 @@ +# ABC-VLA and adapted pi0.5 + +This example turns prepared ABC/XDOF demonstrations into a LeRobot dataset, +produces a pi0.5 training plan, and evaluates compatible checkpoints using ROVE's +existing Trials and Campaigns. Start with the +[comparison guide](../../docs/product/abc-pi05-comparison.md) for the user journey, +success criterion, reports and evidence boundaries. + +## 1. Prepare isolated runtimes + +Keep ROVE, ABC and LeRobot in separate Python environments. ABC and LeRobot have +different ML dependencies. Model execution and training require a Linux NVIDIA +GPU host; the ROVE application can run on that same host. Install ROVE there using +the repository's normal setup. Azure hosting is a future step. + +On a headless Linux GPU host, select MuJoCo's EGL renderer before starting ROVE +or its evaluation commands: + +```bash +export MUJOCO_GL=egl +``` + +The host needs NVIDIA drivers and working EGL libraries; a container also needs +access to the NVIDIA GPU and its graphics libraries. ROVE's child processes +inherit this setting. The dependency check does not prove that camera rendering +works: verify a short rendered simulator episode before launching a long job. + +Clone ABC and check out the source used by this integration: + +```bash +git clone https://github.com/amazon-far/abc.git +cd abc +git checkout d0e987b61b376f1a1b8777b19149f190b2ba6939 +uv sync +uv run prepare.py --sim-task put_plastic_bottles_in_bin +``` + +Follow the pinned ABC repository's preparation instructions for the +`vla_abc130k_200000_v2.pt` checkpoint, its metadata, prepared demonstrations and +tokenizer assets. That checkpoint and its prior training already include ABC task +data; this example cannot establish that its task was unseen during pretraining. + +From the ROVE repository, create the LeRobot environment: + +```bash +uv venv --python 3.12 /path/to/lerobot-env +uv pip install --python /path/to/lerobot-env/bin/python \ + -r examples/abc-pi05/requirements.txt +``` + +Install FFmpeg/FFprobe for video conversion. Download a local snapshot of +[`lerobot/pi05_base`](https://huggingface.co/lerobot/pi05_base) and its required +tokenizer assets after accepting any upstream access conditions. Use the exact +snapshot directory as `--base-checkpoint`; the training receipt records its +weight and configuration hashes. The requirements pin LeRobot source, but are +not a complete transitive dependency lock. Evaluation fingerprints the installed +runtime when you register it. + +Only register trusted local checkpoints. ABC's native PyTorch loader reads its +checkpoint format directly. Evaluation launches use offline Hugging Face settings: +missing checkpoints or tokenizer assets cause an error instead of a download. + +## 2. Convert training demonstrations + +The prepared cache is the directory containing native `train_sim`, `train_real`, +`val_sim` and `val_real` episode directories from ABC's preparation tooling. +Keep validation directories present so the converter can detect overlapping +episode IDs and content hashes. It rejects known validation splits, duplicates, +invalid camera layouts and misaligned frame counts. + +```bash +/path/to/lerobot-env/bin/python examples/abc-pi05/dataset.py \ + --cache /path/to/prepared-cache \ + --output /path/to/abc-bottles-training \ + --repo-id local/abc-bottles \ + --split train_sim \ + --task-name sim_put_the_plastic_bottles_in_the_bin +``` + +The output is a native LeRobot v3 dataset with videos, Parquet tables, training +statistics and a ROVE provenance manifest. Large camera files stay outside trial +JSON records. Current support is the prepared 30 Hz top/left/right camera format, +14D state and absolute joint targets. Raw or stereo XDOF recordings need an +explicit mapping before they can use this converter. + +## 3. Inspect and execute an adaptation plan + +```bash +/path/to/lerobot-env/bin/python examples/abc-pi05/train.py \ + --dataset /path/to/abc-bottles-training \ + --base-checkpoint /path/to/local-pi05-base-snapshot \ + --output /path/to/pi05-bottles-v1 \ + --steps 30000 --batch-size 1 +``` + +The command above prints the plan. Add `--execute` on the GPU host to train. +Start with a short `--steps 20` integration check before a substantial training +job. This short run does not establish useful specialization. The native trainer +saves the model and its processors at +`checkpoints/last/pretrained_model`, alongside a durable training receipt. + +The recipe trains the action expert while freezing the vision backbone, uses +absolute 14D actions, and saves normalization with the checkpoint. It disables +uploads, W&B, image augmentation and simulator evaluation inside the trainer. +ROVE runs held-out simulation episodes separately after adaptation. + +## 4. Register and compare + +```bash +rove bimanual init --output abc-profile.json +``` + +Edit the generated profile with the ABC source directory, both Python +interpreters, the ABC checkpoint and the adapted pi0.5 checkpoint directory. +Confirm `environment.prompt` and the success interpretation. The released ABC +evaluation uses `sim throw plastic bottles in bin`; the selected demonstrations +use `sim put the plastic bottles in the bin`. The converter preserves the training +prompt. Both models receive the same declared evaluation prompt, so choose it +explicitly and retain it in the frozen profile. Do not silently give each model +its own wording. Testing the alternate prompt is a separate comparison revision, +not a matched continuation with an unchanged task instruction. Then: + +```bash +rove bimanual setup --config abc-profile.json +rove bimanual doctor +rove bimanual trial --strategy abc-vla --strategy pi05 --scene 11 --seed 7 +rove bimanual run --name "Bottles comparison" \ + --strategy abc-vla --strategy pi05 \ + --scene 11 --scene 29 --scene 47 --seed 7 --seed 17 --seed 37 \ + --output bottles-report.html +``` + +This campaign creates 18 recorded trials. Open **ABC bimanual comparison** from +Trials or Campaigns for the equivalent UI. Use a new strategy ID for a revised +checkpoint, preserve the scene seeds and scoring settings, and compare reports. +Changing assets after registration blocks execution until the new profile is +registered. Completed scored failures are evaluation results; execution errors +make the command fail after preserving its report, which supports CI automation. + +## Validation achieved + +On 2026-09-15, the pinned LeRobot converter wrote and reopened two actual ABC +training-preview episodes: 1,870 frames, three RGB cameras, 14D state/actions and +saved quantile statistics. The training arguments passed native configuration +parsing. A real ABC/MuJoCo episode executed 30 control steps with three rendered +cameras, two replans and 1.02 simulated seconds, preserving reset and trajectory +evidence. It used a synthetic hold-position policy and correctly did not complete +the bottle-placement task. + +GPU training, ABC-VLA inference and adapted pi0.5 inference remain unvalidated. +The example supplies no learned-model ranking or physical robot performance claim. diff --git a/examples/abc-pi05/dataset.py b/examples/abc-pi05/dataset.py new file mode 100644 index 0000000..ad0d223 --- /dev/null +++ b/examples/abc-pi05/dataset.py @@ -0,0 +1,401 @@ +"""Convert pinned ABC prepared training episodes into a local LeRobot dataset. + +Usage: python dataset.py --cache /path/to/prepared-cache --output /path/to/new-dataset + --repo-id local/abc-bottles --split train_sim + --task-name sim_put_the_plastic_bottles_in_the_bin + +No downloads or uploads. Public validation episodes are never used for training. +Only the prepared 30 Hz, three-camera, 14D joint-position format is accepted. +""" + +from __future__ import annotations + +import argparse +import hashlib +import importlib.metadata +import json +import shutil +import subprocess # nosec B404 - local, argument-list ffmpeg/ffprobe calls +import tempfile +from dataclasses import dataclass +from fractions import Fraction +from pathlib import Path +from urllib.parse import unquote, urlparse + +ABC_REVISION = "d0e987b61b376f1a1b8777b19149f190b2ba6939" # pragma: allowlist secret +LEROBOT_REVISION = "89236ea0f4f81a81ca566081e20dd1ff5f823cbe" # pragma: allowlist secret +CAMERAS = ("top", "left", "right") +JOINTS = [f"{side}_j{i}" for side in ("left", "right") for i in range(1, 7)] +JOINTS = [*JOINTS[:6], "left_gripper", *JOINTS[6:], "right_gripper"] + + +def sha256(path): + digest = hashlib.sha256() + with Path(path).open("rb") as handle: + for block in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def write_json(path, value): + Path(path).write_text(json.dumps(value, indent=2, allow_nan=False) + "\n") + + +def task_prompt(value): + """Match the pinned ABC task_name_to_prompt whitespace/underscore convention.""" + return " ".join(value.replace("-", " ").replace("_", " ").split()) + + +def executable(name): + path = shutil.which(name) + if path is None: + raise RuntimeError(f"Install {name} for prepared-video conversion") + return str(Path(path).resolve()) + + +def verify_lerobot_revision(): + """Check the actual installed source, including a clean local validation clone.""" + distribution = importlib.metadata.distribution("lerobot") + origin = json.loads(distribution.read_text("direct_url.json") or "{}") + revision = origin.get("vcs_info", {}).get("commit_id") + url = urlparse(origin.get("url", "")) + if revision is None and url.scheme == "file": + path = Path(unquote(url.path)) + if (path / ".git").exists(): + result = subprocess.run( # nosec B603 - inspect the installed local source, no shell + [executable("git"), "-C", str(path), "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + ) + dirty = subprocess.run( # nosec B603 - read-only source provenance check + [executable("git"), "-C", str(path), "status", "--porcelain"], + check=True, + capture_output=True, + text=True, + ) + if not dirty.stdout.strip(): + revision = result.stdout.strip() + if revision != LEROBOT_REVISION: + raise ValueError("Install the pinned examples/abc-pi05/requirements.txt runtime") + return {"version": distribution.version, "source_revision": revision} + + +@dataclass(frozen=True) +class Episode: + path: Path + split: str + task: str + prompt: str + cameras: tuple[str, ...] + height: int + length: int + files: dict[str, str] + + +def read_episode(path: Path, split: str) -> Episode: + import numpy as np + + metadata = json.loads((path / "episode_metadata.json").read_text()) + task = metadata.get("task_name") + if not isinstance(task, str) or not task.strip(): + raise ValueError(f"{path.name}: missing task_name") + cameras = tuple(metadata.get("cameras", [])) + if len(cameras) != 3 or set(cameras) != set(CAMERAS): + raise ValueError(f"{path.name}: requires explicit top, left, right camera order") + resolutions = metadata.get("camera_resolutions", {}) + shapes = [resolutions.get(camera) for camera in cameras] + if any(shape != shapes[0] for shape in shapes) or shapes[0] not in ([224, 224], [224, 168]): + raise ValueError(f"{path.name}: only prepared 224x224 or 224x168 cameras are supported") + fps = metadata.get("fps") + if fps is None: + if metadata.get("alignment") != "fixed_clock_30hz_causal": + raise ValueError(f"{path.name}: missing 30 Hz alignment metadata") + tick = metadata.get("tick_ns") + if tick is None or abs(float(tick) - 1e9 / 30) > 1: + raise ValueError(f"{path.name}: invalid 30 Hz timestamp grid") + elif abs(float(fps) - 30) > 1e-6: + raise ValueError(f"{path.name}: only 30 Hz prepared episodes are supported") + binary = path / "states_actions.bin" + if not binary.is_file() or not binary.stat().st_size or binary.stat().st_size % (28 * 8): + raise ValueError(f"{path.name}: expected float64 rows with state14 + action14") + rows = np.memmap(binary, dtype=" 1)).any(): + raise ValueError(f"{path.name}: gripper state/targets must already be in [0,1]") + prompt = metadata.get("instruction") or task_prompt(task) + if not isinstance(prompt, str) or not prompt.strip(): + raise ValueError(f"{path.name}: missing task instruction") + # ABC's task_name_to_prompt replaces underscores. Keep explicit directives. + if task_prompt(prompt) == task_prompt(task): + prompt = task_prompt(task) + randomization = path / "randomization.json" + if randomization.exists(): + random_prompt = (json.loads(randomization.read_text()).get("metadata") or {}).get("prompt") + if random_prompt and task_prompt(random_prompt) != task_prompt(prompt): + raise ValueError( + f"{path.name}: randomization directive differs from training instruction" + ) + filenames = ["episode_metadata.json", "states_actions.bin", "combined_camera-images-rgb.mp4"] + if randomization.exists(): + filenames.append("randomization.json") + return Episode( + path, + split, + task, + prompt, + cameras, + shapes[0][1], + len(rows), + {name: sha256(path / name) for name in filenames}, + ) + + +def plan(cache: Path, splits: list[str], tasks: list[str]): + """Freeze selected training episode identities and reject known holdout overlap.""" + if not splits or set(splits) - {"train_real", "train_sim"} or len(set(splits)) != len(splits): + raise ValueError( + "Select each of train_real/train_sim at most once; validation splits are forbidden" + ) + if not tasks or any(not task.strip() for task in tasks): + raise ValueError("Select explicit source task names; do not silently mix all tasks") + heldout_ids, heldout_hashes = set(), set() + heldout = [] + for split in ("val_real", "val_sim"): + for binary in sorted((cache / split).glob("*/states_actions.bin")): + digest = sha256(binary) + heldout_ids.add(binary.parent.name) + heldout_hashes.add(digest) + heldout.append( + {"split": split, "episode_id": binary.parent.name, "states_actions_sha256": digest} + ) + episodes, seen_ids, seen_hashes = [], set(), set() + for split in splits: + if not (cache / split).is_dir(): + raise ValueError(f"Missing selected source split: {split}") + for metadata in sorted((cache / split).glob("*/episode_metadata.json")): + if json.loads(metadata.read_text()).get("task_name") not in tasks: + continue + ep = read_episode(metadata.parent, split) + digest = ep.files["states_actions.bin"] + if ep.path.name in heldout_ids or digest in heldout_hashes: + raise ValueError(f"{ep.path.name}: training episode overlaps public validation") + if ep.path.name in seen_ids or digest in seen_hashes: + raise ValueError(f"{ep.path.name}: duplicate training episode") + seen_ids.add(ep.path.name) + seen_hashes.add(digest) + episodes.append(ep) + missing = set(tasks) - {ep.task for ep in episodes} + if missing: + raise ValueError(f"No selected training episodes for tasks: {sorted(missing)}") + return episodes, heldout + + +def probe_video(ep: Episode): + output = subprocess.run( # nosec B603 - trusted local executable, no shell + [ + executable("ffprobe"), + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=width,height,avg_frame_rate", + "-of", + "json", + str(ep.path / "combined_camera-images-rgb.mp4"), + ], + check=True, + capture_output=True, + text=True, + ) + streams = json.loads(output.stdout).get("streams", []) + if ( + len(streams) != 1 + or streams[0].get("width") != 224 + or streams[0].get("height") != ep.height * 3 + ): + raise ValueError(f"{ep.path.name}: combined video does not match camera metadata") + if abs(float(Fraction(streams[0]["avg_frame_rate"])) - 30) > 1e-3: + raise ValueError(f"{ep.path.name}: combined video is not 30 Hz") + + +def split_frame(frame, ep): + """Preserve camera order; center-pad 168-high sim images like pi0.5 does.""" + import numpy as np + + if frame.dtype != np.uint8 or frame.shape != (ep.height * 3, 224, 3): + raise ValueError("Invalid decoded stacked RGB frame") + pad = (224 - ep.height) // 2 + result = {} + for index, camera in enumerate(ep.cameras): + image = frame[index * ep.height : (index + 1) * ep.height] + result[f"observation.images.{camera}"] = np.pad(image, ((pad, pad), (0, 0), (0, 0))) + return result + + +def frames(ep: Episode): + import numpy as np + + probe_video(ep) + frame_bytes = ep.height * 3 * 224 * 3 + with tempfile.TemporaryFile() as errors: + process = subprocess.Popen( # nosec B603 - local video decoding, no shell + [ + executable("ffmpeg"), + "-v", + "error", + "-i", + str(ep.path / "combined_camera-images-rgb.mp4"), + "-vsync", + "0", + "-f", + "rawvideo", + "-pix_fmt", + "rgb24", + "pipe:1", + ], + stdout=subprocess.PIPE, + stderr=errors, + ) + try: + for index in range(ep.length): + raw = process.stdout.read(frame_bytes) + if len(raw) != frame_bytes: + raise ValueError(f"{ep.path.name}: video ended before state row {index}") + yield np.frombuffer(raw, dtype=np.uint8).reshape(ep.height * 3, 224, 3) + if process.stdout.read(1): + raise ValueError(f"{ep.path.name}: video has more frames than state rows") + if process.wait() != 0: + errors.seek(0) + raise ValueError( + f"Video decode failed: {errors.read().decode(errors='replace')[-2000:]}" + ) + finally: + process.stdout.close() + if process.poll() is None: + process.terminate() + process.wait() + + +def features(): + return { + "observation.state": {"dtype": "float32", "shape": (14,), "names": JOINTS}, + "action": {"dtype": "float32", "shape": (14,), "names": JOINTS}, + **{ + f"observation.images.{camera}": { + "dtype": "video", + "shape": (224, 224, 3), + "names": ["height", "width", "channels"], + } + for camera in CAMERAS + }, + } + + +def convert(cache, output, repo_id, splits, tasks, *, dataset_class=None): + import numpy as np + + output = Path(output) + if output.exists(): + raise ValueError("Output already exists; choose a new versioned dataset directory") + episodes, heldout = plan(Path(cache), splits, tasks) + if dataset_class is None: + runtime = verify_lerobot_revision() + from lerobot.datasets.lerobot_dataset import LeRobotDataset + + dataset_class = LeRobotDataset + else: + runtime = {"implementation": "injected_test_writer"} + writer = dataset_class.create( + repo_id=repo_id, + root=output, + fps=30, + features=features(), + robot_type="abc_i2rt_yam_bimanual", + use_videos=True, + ) + manifest = { + "schema": "rove.abc-pi05.dataset/v1", + "status": "running", + "repo_id": repo_id, + "abc_revision": ABC_REVISION, + "lerobot_revision": LEROBOT_REVISION, + "lerobot_runtime": runtime, + "fps": 30, + "action_semantics": "absolute_joint_positions_radians_and_grippers_0_1", + "joint_names": JOINTS, + "camera_keys": list(CAMERAS), + "image_padding": "center_black_to_224_square", + "episodes": [], + "excluded_public_validation": heldout, + "validation_note": "Only locally available public validation files were checked. Pretraining exposure is not established by this audit.", + } + manifest_path = output / "rove_dataset_manifest.json" + write_json(manifest_path, manifest) + try: + for index, ep in enumerate(episodes): + rows = np.memmap(ep.path / "states_actions.bin", dtype=" 1_000_000: + raise ValueError("Portable checkpoint provenance exceeds 1 MB") + manifest_path.write_bytes(encoded + b"\n") + # Include the manifest in the outer receipt and ROVE's checkpoint fingerprint. + return {**checkpoint_files, PORTABLE_MANIFEST: sha256(manifest_path)} + + +def execute(plan): + import torch + + # Import only the dependency-light loader, without installing ROVE into this runtime. + sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src/rove/bimanual")) + from pi05_weights import install_strict_training_loader + + plan["lerobot_runtime"] = verify_lerobot_revision() + if not torch.cuda.is_available(): + raise RuntimeError("Training requires a CUDA GPU; the plan can be inspected without one") + output = Path(plan["output"]) + receipt = output.parent / f"{output.name}.training.json" + if output.exists() or receipt.exists(): + raise ValueError("This training output or receipt already exists; choose a new version") + output.parent.mkdir(parents=True, exist_ok=True) + plan["started_at"] = datetime.now(UTC).isoformat() + plan["status"] = "running" + write_json(receipt, plan) + saved_argv = sys.argv[:] + try: + install_strict_training_loader() + sys.argv = ["lerobot-train", *plan["native_training_arguments"]] + runpy.run_module("lerobot.scripts.lerobot_train", run_name="__main__") + checkpoint = output / "checkpoints/last/pretrained_model" + required = [ + "config.json", + "model.safetensors", + "policy_preprocessor.json", + "policy_postprocessor.json", + ] + if any(not (checkpoint / file).is_file() for file in required): + raise RuntimeError( + "Training returned without the expected model and saved processor checkpoint" + ) + plan["checkpoint"] = str(checkpoint.resolve()) + plan["checkpoint_files"] = write_checkpoint_provenance(checkpoint, plan) + plan["status"] = "completed" + except BaseException as exc: + plan["status"] = "interrupted" if isinstance(exc, KeyboardInterrupt) else "failed" + plan["error_type"] = type(exc).__name__ + raise + finally: + sys.argv = saved_argv + plan["finished_at"] = datetime.now(UTC).isoformat() + write_json(receipt, plan) + return plan + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--dataset", type=Path, required=True) + parser.add_argument("--base-checkpoint", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--steps", type=int, default=30000) + parser.add_argument("--batch-size", type=int, default=1) + parser.add_argument("--seed", type=int, default=42) + parser.add_argument("--execute", action="store_true") + args = parser.parse_args() + plan = training_plan( + args.dataset, + args.base_checkpoint, + args.output, + steps=args.steps, + batch_size=args.batch_size, + seed=args.seed, + ) + if args.execute: + execute(plan) + print(json.dumps(plan, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/frontend/benchmarks.html b/frontend/benchmarks.html index e04e7c8..b66eb8a 100644 --- a/frontend/benchmarks.html +++ b/frontend/benchmarks.html @@ -24,7 +24,7 @@ New campaign
-

Loading saved campaigns…

+

Loading saved campaigns…

Compare bimanual models
diff --git a/frontend/bimanual.css b/frontend/bimanual.css new file mode 100644 index 0000000..16eab95 --- /dev/null +++ b/frontend/bimanual.css @@ -0,0 +1,44 @@ +.bimanual-page { margin:0; background:var(--f-bg); color:var(--f-text); } +.bimanual-page #main { max-width:1200px; width:100%; margin:32px auto; padding:0 32px 48px; box-sizing:border-box; } +.bimanual-page p { color:var(--f-text-secondary); font-size:16px; line-height:1.55; margin:8px 0 16px; max-width:90ch; } +.bimanual-page .page-heading { display:flex; align-items:flex-start; gap:24px; justify-content:space-between; margin-bottom:24px; } +.bimanual-page .page-heading h1 { font-size:28px; line-height:1.3; margin:4px 0 8px; } +.bimanual-page .bimanual-eyebrow { color:var(--f-text-muted); font-size:13px; margin:0; } +.bimanual-page h2 { font-size:20px; line-height:1.4; margin:0 0 16px; font-weight:600; } +.bimanual-panel { border:1px solid var(--f-border); background:var(--f-surface); border-radius:var(--f-radius-panel); padding:24px; margin:24px 0; } +.bimanual-page fieldset { border:0; padding:0; margin:0; min-width:0; } +.bimanual-header-controls { display:grid; grid-template-columns:2fr 1fr; gap:24px; } +.bimanual-page label { display:flex; flex-direction:column; gap:8px; font-size:15px; font-weight:600; } +.bimanual-page label span { color:var(--f-text-muted); font-size:13px; line-height:1.5; font-weight:400; } +.bimanual-page input:not([type="checkbox"]), .bimanual-page select { width:100%; min-height:40px; box-sizing:border-box; } +.bimanual-page .bimanual-calculator { font-size:22px; font-weight:600; line-height:1.45; margin:24px 0; color:var(--f-text); } +.bimanual-page .bimanual-task { font-size:18px; font-weight:600; } +.bimanual-fields { display:grid; grid-template-columns:1fr 1fr 1fr; gap:24px; margin-top:24px; } +.bimanual-page table { width:100%; border-collapse:collapse; text-align:left; } +.bimanual-page th { color:var(--f-text-muted); font-size:13px; font-weight:600; background:var(--f-bg); padding:12px 16px; border-bottom:1px solid var(--f-border); } +.bimanual-page td { padding:16px; font-size:15px; vertical-align:top; border-bottom:1px solid var(--f-border); } +.bimanual-page td small { display:block; font-size:13px; margin-top:4px; max-width:60ch; } +.bimanual-page td input { width:18px; height:18px; margin:3px 0; } +.bimanual-page tr:has(input:checked) { background:var(--f-selected); } +.bimanual-actions { display:flex; align-items:center; flex-wrap:wrap; gap:12px; margin-top:20px; } +.bimanual-page .primary-link, .bimanual-page .secondary { min-height:40px; padding:8px 16px; box-sizing:border-box; white-space:nowrap; } +.bimanual-page button:disabled { opacity:.5; cursor:not-allowed; } +.bimanual-page pre { padding:16px; background:var(--f-bg); overflow:auto; border:1px solid var(--f-border); border-radius:4px; } +.bimanual-page li { font-size:15px; line-height:1.6; margin:8px 0; } +.bimanual-section-heading { display:flex; align-items:center; justify-content:space-between; gap:20px; margin-top:32px; } +.bimanual-facts { display:grid; grid-template-columns:180px 1fr; gap:12px 24px; font-size:16px; line-height:1.5; } +.bimanual-facts dt { color:var(--f-text-muted); } +.bimanual-facts dd { margin:0; overflow-wrap:anywhere; } +.bimanual-table-scroll { overflow:auto; } +.bimanual-page #pageStatus:empty, .bimanual-page #configurationIssues:empty, .bimanual-page #episodeActivity:empty { display:none; } +.bimanual-page #pageStatus[data-error="true"] { border-left:3px solid #eaa0a0; padding:12px 16px; background:var(--f-surface); } +.bimanual-page .bimanual-status { display:inline-block; border:1px solid var(--f-border); border-radius:4px; padding:2px 8px; font-size:13px; } +@media (max-width:760px) { + .bimanual-page #main { padding:0 20px 32px; margin-top:24px; } + .bimanual-page .page-heading { flex-direction:column; gap:8px; } + .bimanual-header-controls, .bimanual-fields { grid-template-columns:1fr; gap:16px; } + .bimanual-panel { padding:20px; } + .bimanual-facts { grid-template-columns:1fr; gap:4px; } + .bimanual-facts dd { margin-bottom:12px; } +} +@media (prefers-reduced-motion:no-preference) { .bimanual-panel { animation:bimanual-appear 160ms ease-out; } @keyframes bimanual-appear { from { opacity:0; transform:translateY(4px); } to { opacity:1; transform:translateY(0); } } } diff --git a/frontend/bimanual.html b/frontend/bimanual.html new file mode 100644 index 0000000..8e12016 --- /dev/null +++ b/frontend/bimanual.html @@ -0,0 +1,59 @@ + + + + + + ROVE · Compare bimanual models + + + + + + + + + + + + + +
+
+

ABC simulator · Two-arm manipulation

Compare bimanual models

See how ABC-VLA and an adapted π0.5 perform on the same task and starting scenes.

Back to trials
+

Checking the configured models…

+ + + + + +
+ + diff --git a/frontend/bimanual.js b/frontend/bimanual.js new file mode 100644 index 0000000..2c1f900 --- /dev/null +++ b/frontend/bimanual.js @@ -0,0 +1,211 @@ +/* ABC evaluations use the ordinary ROVE trial store, progress and report surfaces. */ +(function () { + "use strict"; + function parseSeeds(value) { + const parts = value.split(",").map(item => item.trim()); + if (!parts.length || parts.length > 100 || parts.some(item => !/^\d+$/.test(item))) throw Error("Enter up to 100 non-negative seed numbers, separated by commas."); + const seeds = parts.map(Number); + if (seeds.some(seed => !Number.isSafeInteger(seed) || seed > 2147483647) || new Set(seeds).size !== seeds.length) throw Error("Seed numbers must be unique and no greater than 2147483647."); + return seeds; + } + function countLabel(cases, strategies, attempts) { + const total = cases * strategies * attempts; + return `${cases} ${cases === 1 ? "case" : "cases"} × ${strategies} ${strategies === 1 ? "strategy" : "strategies"} × ${attempts} ${attempts === 1 ? "attempt" : "attempts"} = ${total} planned ${total === 1 ? "trial" : "trials"}`; + } + function criterionLabel(grading) { + const mode = grading?.success_mode || grading?.criterion || grading?.success || ""; + if (["final", "final_success", "goal_at_end"].includes(mode)) return "The simulator reports the goal satisfied at the end of the episode."; + if (["ever", "episode_success", "goal_reached"].includes(mode)) return "The simulator reports the goal reached at least once during the episode."; + return `Simulator task evaluator · ${mode || "the frozen campaign success criterion"}`; + } + if (typeof module !== "undefined" && module.exports) module.exports = {parseSeeds, countLabel, criterionLabel}; + if (typeof document === "undefined") return; + const el = id => document.getElementById(id); + if (!el("comparisonForm")) return; + const terminal = new Set(["completed", "error", "cancelled", "interrupted", "timeout", "runtime_changed", "evaluator_changed"]); + let profile = null, preview = null, reviewedRequest = null, active = null, timer = null, dirty = false, busy = false; + const events = new Map(); + function node(tag, text, className) { const value = document.createElement(tag); if (text != null) value.textContent = text; if (className) value.className = className; return value; } + function link(text, href, newTab = false) { const value = node("a", text, "secondary"); value.href = href; if (newTab) { value.target = "_blank"; value.rel = "noopener noreferrer"; } return value; } + function status(text, error = false) { el("pageStatus").textContent = text; el("pageStatus").dataset.error = String(error); } + async function api(url, options = {}) { + const response = await fetch(url, options.body ? {...options, headers: {"Content-Type": "application/json"}, body: JSON.stringify(options.body)} : options); + const value = await response.json(); + if (!response.ok) { + const detail = value.detail; + const error = Error(typeof detail === "string" ? detail : detail?.message ? `${detail.message}${detail.issues?.length ? `: ${detail.issues.join("; ")}` : ""}` : "Unable to complete the request. Check the configuration and try again."); + error.status = response.status; throw error; + } + return value; + } + function request() { + const reset = parseSeeds(el("resetSeeds").value), seeds = parseSeeds(el("policySeeds").value); + if (el("comparisonMode").value === "trial" && (reset.length !== 1 || seeds.length !== 1)) throw Error("A trial comparison uses one scene seed and one model seed. Choose Campaign for repeated comparisons."); + const strategies = [...el("strategyRows").querySelectorAll("input:checked")].map(input => input.value); + const maxSteps = Number(el("maxSteps").value); + if (!Number.isInteger(maxSteps) || maxSteps < 1 || maxSteps > 3540) throw Error("Choose between 1 and 3540 control steps."); + if (!strategies.length) throw Error("Select at least one ready model."); + if (reset.length * seeds.length * strategies.length > 2000) throw Error("Use at most 2,000 trials per comparison."); + if (!el("comparisonName").value.trim()) throw Error("Give this comparison a name."); + return {name: el("comparisonName").value.trim(), strategy_ids: strategies, reset_seeds: reset, policy_seeds: seeds, max_steps: maxSteps}; + } + function update() { + preview = null; reviewedRequest = null; + let cases = 0, seeds = 0; + try { cases = parseSeeds(el("resetSeeds").value).length; seeds = parseSeeds(el("policySeeds").value).length; } catch { /* Validation appears before review. */ } + const strategies = el("strategyRows").querySelectorAll("input:checked").length; + el("plannedCount").textContent = countLabel(cases, strategies, seeds); + try { request(); el("reviewButton").disabled = busy; } catch { el("reviewButton").disabled = true; } + } + function strategyTable(rows) { + el("strategyRows").replaceChildren(); + for (const row of rows) { + const tr = node("tr"), select = node("td"), input = node("input"); + input.type = "checkbox"; input.value = row.id; input.disabled = !row.ready; input.checked = Boolean(row.ready); input.setAttribute("aria-label", `Compare ${row.id}`); + select.append(input); const readiness = node("td"); readiness.append(node("span", row.ready ? "Ready" : "Setup needed", "bimanual-status")); + if (row.issues?.length) readiness.append(node("small", row.issues.join(" · "))); + const kind = row.kind === "pi05" || row.kind === "lerobot-pi05" ? "π0.5 · adapted checkpoint" : ["abc", "abc-vla", "abc_vla"].includes(row.kind) ? "ABC-VLA" : row.kind || "Configured model"; + const identity = node("td", row.label || row.id); if (row.label) identity.append(node("small", row.id)); + tr.append(select, identity, node("td", kind), readiness); el("strategyRows").append(tr); + } + } + async function load() { + busy = true; el("recheck").disabled = true; status("Checking the configured models…"); + try { + profile = await api("/api/bimanual/config"); strategyTable(profile.strategies || []); + el("setup").hidden = profile.configured && Boolean(profile.strategies?.length); + el("comparisonForm").hidden = !profile.configured; + el("configurationIssues").replaceChildren(...(profile.issues || []).map(issue => node("p", issue))); + status(profile.configured ? "" : "Runtime setup is required before a learned model comparison can run."); + } catch (error) { status(error.message, true); el("setup").hidden = false; } + finally { busy = false; el("recheck").disabled = false; update(); } + } + function showReview(value) { + el("comparisonForm").hidden = true; el("review").hidden = false; + el("reviewSummary").textContent = countLabel(value.cases, value.strategies.length, value.attempts_per_case); + const facts = [["Task", "Put plastic bottles in the bin"], ["Starting scenes", reviewedRequest.reset_seeds.join(", ")], ["Models", value.strategies.join(" · ")], ["Model seeds", reviewedRequest.policy_seeds.join(", ")], ["Action budget", `${value.max_steps} control steps per trial`], ["Success", criterionLabel(value.success_criterion)], ["Evidence", "Simulator outcome, executed trajectory and inference traces"]]; + el("reviewFacts").replaceChildren(...facts.flatMap(([key, val]) => [node("dt", key), node("dd", val)])); + el("reviewReadiness").textContent = value.ready ? "Configuration verified. Running will use the reviewed settings and save every trial." : value.issues.join(" · "); + el("launchButton").disabled = !value.ready; el("launchButton").textContent = el("comparisonMode").value === "campaign" ? "Run campaign" : "Run comparison"; + el("reviewHeading").focus(); + } + el("comparisonForm").addEventListener("submit", async event => { + event.preventDefault(); if (busy) return; + busy = true; el("reviewButton").disabled = true; status("Verifying model and simulator configuration…"); + try { reviewedRequest = request(); preview = await api("/api/bimanual/preview", {method: "POST", body: reviewedRequest}); showReview(preview); status(""); } + catch (error) { status(error.message, true); } + finally { busy = false; el("reviewButton").disabled = false; } + }); + el("editButton").addEventListener("click", () => { el("review").hidden = true; el("comparisonForm").hidden = false; update(); el("comparisonName").focus(); }); + el("configurationFields").addEventListener("input", () => { dirty = true; update(); }); + function modeContext(campaign) { + el("backToBrowser").href = campaign ? "/static/benchmarks.html" : "/static/history.html"; el("backToBrowser").textContent = campaign ? "Back to campaigns" : "Back to trials"; + window.RoveNavigation?.setActive(campaign ? "results" : "evaluate"); + } + el("comparisonMode").addEventListener("change", () => { + const campaign = el("comparisonMode").value === "campaign"; + el("resetSeeds").value = campaign ? "0, 1, 2" : "0"; el("policySeeds").value = campaign ? "0, 1, 2" : "0"; + modeContext(campaign); + dirty = true; update(); + }); + async function fetchEvents(id) { + const previous = events.get(id) || {events: [], offset: 0}; + try { + const page = await api(`/api/trials/${encodeURIComponent(id)}/events?limit=200&offset=${previous.offset}`); + previous.events.push(...page.events); previous.offset += page.events.length; + if (previous.events.length > 400) previous.events = previous.events.slice(-400); + previous.error = ""; + } catch { previous.error = "Live trace updates unavailable; retrying."; } + events.set(id, previous); return previous; + } + function activity(rows) { + const lines = []; + for (const row of rows.filter(item => item.execution === "running")) { + const page = events.get(row.trial_id), last = page?.events?.at(-1); + if (last) { const details = last.data || {}, steps = details.completed_steps ?? details.steps; lines.push(node("p", `${row.strategy_id}: ${last.name || last.stage || "Episode in progress"}${Number.isFinite(steps) ? ` · ${steps}${Number.isFinite(details.max_steps) ? ` / ${details.max_steps}` : ""} control steps` : ""}`)); } + if (page?.error) lines.push(node("p", page.error)); + } + el("episodeActivity").replaceChildren(...lines); + } + function resultTable(rows) { + el("resultRows").replaceChildren(); + for (const row of rows.slice(0, 50)) { + const tr = node("tr"), evidence = node("td"); + const outcome = row.outcome || row.result?.outcome; + const label = outcome === "pass" ? "Criterion met" : outcome === "fail" ? "Criterion not met" : "Not assessed"; + evidence.append(link("Inspect trial & traces", `/static/history.html?trial=${encodeURIComponent(row.trial_id)}#traces`, true)); + evidence.append(node("small", `Trial ${row.trial_id.slice(0, 8)}`)); + const caseName = /^bottles-scene-\d+$/.test(row.task_id || "") ? `Scene ${row.task_id.replace("bottles-scene-", "")}` : row.task_id || "Case unavailable"; + tr.append(node("td", row.strategy_id), node("td", caseName), node("td", Number.isInteger(row.seed) ? String(row.seed) : "Unavailable"), node("td", label), node("td", row.execution || "unknown"), evidence); el("resultRows").append(tr); + } + } + function complete(campaign, rows) { + dirty = false; el("cancelButton").hidden = true; el("results").hidden = false; + el("executionHeading").textContent = "Recorded execution"; + el("resultsSummary").textContent = campaign.status === "completed" ? "The comparison has finished. Outcomes below come from the simulator’s frozen success criterion. Inspect any trial for its measurements and trace." : "The comparison stopped before all work completed. Recorded trials remain available; incomplete attempts do not establish task success."; + if (rows.length > 50) el("resultsSummary").append(` Showing 50 of ${rows.length} recorded trials. The full report includes every trial.`); + el("resultActions").replaceChildren(); + if (active.mode === "campaign") { + el("resultActions").append(link("Open campaign report", `/api/campaigns/${encodeURIComponent(active.id)}/report?format=html`, true), link("Campaign history", `/static/campaign-history.html?campaign=${encodeURIComponent(active.id)}`), link("Export JSON", `/api/campaigns/${encodeURIComponent(active.id)}/report?format=json`)); + } else el("resultActions").append(link("All saved trials", "/static/history.html")); + resultTable(rows); el("resultsHeading").focus(); + } + async function poll() { + if (!active) return; + try { + let campaign, rows; + if (active.mode === "campaign") { + const [detail, attempts] = await Promise.all([api(`/api/campaigns/${encodeURIComponent(active.id)}`), api(`/api/campaigns/${encodeURIComponent(active.id)}/assessments`)]); + campaign = detail.campaign; rows = attempts.trials || []; + } else { + const job = await api(`/api/bimanual/trials/${encodeURIComponent(active.id)}`); + campaign = {status: job.status, spec: {tasks: job.tasks, strategies: job.strategy_ids, seeds: job.seeds}}; + rows = await Promise.all(Object.entries(job.trial_ids).map(async ([sid, id]) => { + const trial = await api(`/api/trials/${encodeURIComponent(id)}`); + return {trial_id: id, strategy_id: sid, task_id: job.tasks[0].id, seed: job.seeds[0], execution: trial.status, result: trial.result, outcome: trial.result?.outcome, started_at: trial.started_at || trial.created_at, finished_at: trial.finished_at}; + })); + } + await Promise.all(rows.filter(row => row.execution === "running").map(row => fetchEvents(row.trial_id))); + if (window.RoveCampaignProgress) window.RoveCampaignProgress.render(el("executionProgress"), window.RoveCampaignProgress.buildStrategyProgress(campaign, rows, Object.fromEntries(events)), id => window.open(`/static/history.html?trial=${encodeURIComponent(id)}#traces`, "_blank", "noopener,noreferrer")); + activity(rows); status(""); + if (terminal.has(campaign.status)) { complete(campaign, rows); return; } + } catch (error) { + if (error.status === 404) { status(error.message, true); el("cancelButton").hidden = true; el("executionHeading").textContent = "Comparison session unavailable"; return; } + status(`${error.message} Progress will retry. Saved trials remain available in Trials.`, true); + } + timer = setTimeout(poll, 2000); + } + el("launchButton").addEventListener("click", async () => { + if (!preview?.ready || busy) return; + busy = true; el("launchButton").disabled = true; status("Starting the reviewed comparison…"); + try { + const mode = el("comparisonMode").value; + const value = await api(`/api/bimanual/${mode === "campaign" ? "campaigns" : "trials"}`, {method: "POST", body: {...reviewedRequest, preview_hash: preview.preview_hash}}); + active = {mode, id: value.id}; dirty = false; el("review").hidden = true; el("execution").hidden = false; el("cancelButton").hidden = false; el("executionHeading").focus(); + history.replaceState(null, "", `/static/bimanual.html?${mode === "campaign" ? "campaign" : "job"}=${encodeURIComponent(value.id)}`); + await poll(); + } catch (error) { status(error.message, true); el("launchButton").disabled = false; } + finally { busy = false; } + }); + el("cancelButton").addEventListener("click", async () => { + if (!active) return; el("cancelButton").disabled = true; + try { await api(active.mode === "campaign" ? `/api/campaigns/${encodeURIComponent(active.id)}/cancel` : `/api/bimanual/trials/${encodeURIComponent(active.id)}/cancel`, {method: "POST"}); clearTimeout(timer); await poll(); } + catch (error) { status(error.message, true); } + finally { el("cancelButton").disabled = false; } + }); + el("newComparison").addEventListener("click", async () => { + clearTimeout(timer); active = null; events.clear(); el("execution").hidden = true; el("results").hidden = true; el("comparisonForm").hidden = false; el("executionHeading").textContent = "Running your comparison"; history.replaceState(null, "", `/static/bimanual.html${el("comparisonMode").value === "campaign" ? "?mode=campaign" : ""}`); + if (!profile) await load(); else update(); + }); + el("recheck").addEventListener("click", load); + window.addEventListener("beforeunload", event => { if (dirty) { event.preventDefault(); event.returnValue = ""; } }); + document.addEventListener("click", event => { const anchor = event.target.closest?.("a"); if (dirty && anchor && anchor.target !== "_blank" && (anchor.pathname !== location.pathname || anchor.search !== location.search) && !window.confirm("Leave this comparison? Your unrun changes have not been saved.")) event.preventDefault(); }); + const params = new URLSearchParams(location.search); + const campaignId = params.get("campaign"), jobId = params.get("job"); + if (campaignId || jobId) { + active = {mode: campaignId ? "campaign" : "trial", id: campaignId || jobId}; el("comparisonMode").value = active.mode; modeContext(Boolean(campaignId)); el("execution").hidden = false; poll(); + } else { + if (params.get("mode") === "campaign") { el("comparisonMode").value = "campaign"; el("comparisonMode").dispatchEvent(new Event("change")); dirty = false; } + load(); + } +})(); diff --git a/frontend/history.html b/frontend/history.html index be458c1..4831a6c 100644 --- a/frontend/history.html +++ b/frontend/history.html @@ -24,7 +24,7 @@
Export workspace for analysis

The ZIP includes all local cases, reviews, trial records and private assets. Active evaluations must finish first. SQLite remains the working database; the versioned tables can be transformed for Fabric or Delta Lake.

New trial
-

+

Compare bimanual models
    diff --git a/src/rove/benchmarks/api.py b/src/rove/benchmarks/api.py index 4ed215c..594a455 100644 --- a/src/rove/benchmarks/api.py +++ b/src/rove/benchmarks/api.py @@ -138,6 +138,7 @@ async def report(campaign_id: str, format: str = "html"): from rove.api.datasets import create_dataset_router from rove.api.insights import create_insights_router from rove.api.trajectory import create_trajectory_router + from rove.bimanual.api import create_bimanual_router combined = APIRouter(lifespan=lifespan) @@ -146,6 +147,7 @@ def ensure_launched(campaign_id: str): launch(campaign_id) combined.include_router(create_dataset_router(root, ensure_launched)) + combined.include_router(create_bimanual_router(root, ensure_launched)) combined.include_router(create_assistant_router(root, ensure_launched)) combined.include_router(create_insights_router(root)) combined.include_router(create_trajectory_router(root)) diff --git a/src/rove/benchmarks/report.py b/src/rove/benchmarks/report.py index b80cb39..726bafa 100644 --- a/src/rove/benchmarks/report.py +++ b/src/rove/benchmarks/report.py @@ -6,11 +6,148 @@ import html import io import json +import math +import re +from collections import defaultdict from datetime import datetime, timedelta from rove.benchmarks.evidence import summarize_evidence from rove.benchmarks.metrics import summarize +_EPISODE_MEASUREMENTS = { + "episode_success": "Goal reached during the episode", + "final_success": "Goal satisfied at episode end", + "episode_steps": "Executed steps", + "simulated_time_s": "Simulated episode duration", + "replans": "Policy predictions", + "inference_latency_mean_s": "Mean inference latency per episode", + "inference_latency_p95_s": "p95 inference latency per episode", + "inference_wall_s": "Total inference time per episode", +} + + +def episode_comparison(campaign: dict, trials: list[dict]) -> dict | None: + """Requested seeds alone do not establish that the actual reset states match.""" + spec = campaign["spec"] + if spec.get("execution") != "abc-bimanual": + return None + expected = { + (task["id"], strategy, seed) + for task in spec["tasks"] + for strategy in spec["strategies"] + for seed in spec["seeds"] + } + states, recorded = defaultdict(set), set() + for trial in trials: + identity = (trial.get("task_id"), trial.get("strategy_id"), trial.get("seed")) + output = (trial.get("result") or {}).get("output") or {} + if not isinstance(output, dict): + continue + digest = output.get("initial_state_digest") + if ( + identity in expected + and output.get("scope") == "abc_simulation" + and isinstance(digest, str) + and re.fullmatch(r"[a-f0-9]{64}", digest) + ): + states[identity[0]].add(digest) + recorded.add(identity) + mismatches = sorted(case for case, digests in states.items() if len(digests) > 1) + missing = len(expected - recorded) + status = "mismatched" if mismatches else "incomplete" if missing else "matched" + return { + "status": status, + "matched_initial_states": False if mismatches else None if missing else True, + "mismatched_cases": mismatches, + "recorded_initial_states": len(recorded), + "planned_trials": len(expected), + "missing_initial_states": missing, + "note": ( + "Initial simulator states differed; this is not a matched comparison. " + "Per-strategy outcomes remain available, but differences cannot be attributed to the policy alone." + if mismatches + else "Some initial simulator states are missing. Matching starting conditions are not yet established for all planned trials." + if missing + else "Recorded initial simulator states match within every case. This checks resets; it does not establish equal training exposure or deterministic GPU execution." + ), + } + + +def episode_measurement_section(data: dict) -> str: + """Compare supplied episode measurements, retaining units, quality and coverage.""" + spec = data["campaign"]["spec"] + if spec.get("execution", "robotics") == "robotics": + return "" + records = data.get("verification", {}).get("measurement_records", []) + keys = { + (r["name"], r["unit"], r["quality"]) + for r in records + if r.get("name") in _EPISODE_MEASUREMENTS and r.get("quality") != "unknown" + } + if not keys: + return "" + values = defaultdict(dict) + for record in records: + value = record.get("value") + key = (record.get("name"), record.get("unit"), record.get("quality")) + if ( + key in keys + and isinstance(value, int | float) + and not isinstance(value, bool) + and math.isfinite(value) + ): + # One measured value per planned episode. Duplicate evaluator records + # must not make a strategy appear to have more coverage than it has. + identity = (record["task"], record["seed"]) + group = values[*key, record["strategy"]] + if identity in group and group[identity] != value: + group[identity] = None + else: + group[identity] = value + planned = len(spec["tasks"]) * len(spec["seeds"]) + headings = "".join(f"{html.escape(sid)}" for sid in spec["strategies"]) + rows = [] + for name, unit, quality in sorted( + keys, key=lambda key: (list(_EPISODE_MEASUREMENTS).index(key[0]), key[1], key[2]) + ): + label = _EPISODE_MEASUREMENTS[name] + quality_label = ( + "observed in simulation" + if quality == "observed" and spec["execution"] == "abc-bimanual" + else quality + ) + cells = [] + for sid in spec["strategies"]: + known = [ + value for value in values[name, unit, quality, sid].values() if value is not None + ] + mean = sum(known) / len(known) if known else None + display = "Unavailable" if mean is None else f"{mean:.3f}".rstrip("0").rstrip(".") + if ( + name in {"episode_success", "final_success"} + and known + and all(0 <= value <= 1 for value in known) + ): + display = f"{mean:.1%}" + cells.append( + f'{html.escape(display)}
    ' + f"{len(known)} / {planned} episodes measured" + ) + rows.append( + f'{html.escape(label)}
    ' + f"{html.escape(unit)} · {html.escape(quality_label)}{''.join(cells)}" + ) + return ( + '

    Episode performance

    ' + '

    Compare what happened in the episodes with the inference time needed to control them. ' + "Values are averages across measured episodes; missing measurements are excluded and coverage is shown in every cell. " + "The average of episode p95 latencies is not a pooled p95 across all predictions.

    " + '
    ' + f"{headings}{''.join(rows)}
    Measurement
    " + "

    Simulated duration is not wall-clock cycle time. Inference measurements exclude model loading and environment setup; " + "the separate execution and assessment time includes that work. Goal-at-end and goal-reached measurements do not replace the frozen success criterion.

    " + ) + def public_campaign(campaign: dict) -> dict: # Endpoint configuration can contain credentials. Never put it in a report/API response. @@ -62,11 +199,14 @@ def report_data( **row, } ) + summary = summarize(campaign, trials) + if comparison := episode_comparison(campaign, trials): + summary["episode_comparison"] = comparison return { "schema_version": 2, "verification": summarize_evidence(campaign, trials), "campaign": public_campaign(campaign), - "summary": summarize(campaign, trials), + "summary": summary, "trials": trials, "trends": trends, "excluded_history_series": excluded, @@ -276,6 +416,10 @@ def escape(value): scope_label = "Task output assessment" source_label = "Independent grader assessment" scope_note = "Each outcome comes from the recorded grader and its evidence. Missing assessments remain unknown." + if spec["execution"] == "abc-bimanual": + scope_label = "Simulated episode outcome" + source_label = "ABC simulator assessment" + scope_note = "These outcomes measure the configured goal in ABC's simulator. They do not establish performance on a customer's physical robot." human = any( criterion.get("required") and criterion.get("assessment") == "human_review" for criterion in contract.get("criteria", []) @@ -619,7 +763,16 @@ def rows_table(rows, fields): portfolio = summary["portfolio"] campaign_url = ( f"/static/datasets.html?step=review&campaign={quote(str(campaign['id']), safe='')}" + if spec.get("execution", "robotics") == "robotics" + else "/static/benchmarks.html" ) + episode_section = episode_measurement_section(data) + comparison_notice = "" + if comparison := summary.get("episode_comparison"): + comparison_notice = ( + '

    Initial state comparison · ' + f"{escape(comparison['note'])}

    " + ) invalid_note = f"{invalid} invalid configured verdicts · {execution_issues} execution issues. These are execution diagnostics, separate from contract assessments." return f''' {escape(spec["name"])} · ROVE campaign report @@ -629,6 +782,7 @@ def rows_table(rows, fields):

    An evidence-backed view of your evaluated strategies.

    {escape(campaign["status"])}{escape(spec["suite_version"])}Revision {escape(spec["revision"])}{len(task_ids)} cases · {len(strategy_ids)} strategies · {len(spec["seeds"])} repeats

    {escape(campaign["evidence_kind"])} · {escape(campaign["grading_note"])}

    +{comparison_notice}
    {escape(source_label)}

    What did the trials establish?

    {scope_note}

    {escape(scope_label)}
    Assessed pass{passed}of {planned} planned trials
    @@ -638,6 +792,7 @@ def rows_table(rows, fields):
    Verdict coverage · {resolved} / {planned} trials resolved{coverage:.0%}

    {human_note}

    {invalid_note}

    Compare strategies

    {portfolio["tasks_solved"]} / {portfolio["tasks_total"]} cases succeeded at least once across the selected strategies, with {portfolio["attempt_budget_per_task"]} planned attempts per case. This is hindsight coverage, not a deployed selection policy.

    {outcome_chart}
    {"".join(strategy_cards)}
    +{episode_section} {robotics_section}

    Correctness and repeatability

    Each case has equal weight. Unknown outcomes stay visible as bounds; they cannot inflate a point estimate.

    {"".join(reliability)}
    How to read the curves and uncertainty

    Solid points require a verdict for every planned attempt. Dotted lines bound unresolved outcomes; they are not confidence intervals. Missing points at k greater than the repeat count mean insufficient trials.

    pass@k = 1 - C(n-c,k)/C(n,k); pass^k = C(c,k)/C(n,k), for n repeats and c successes. Errors, timeouts, interrupted and pending trials remain unknown.

    Task-level pass@1 intervals are 95% Wilson intervals assuming independent attempts. Correlated environments or provider changes weaken that assumption. A verifier's confidence is never used as reliability.

    diff --git a/src/rove/bimanual/__init__.py b/src/rove/bimanual/__init__.py new file mode 100644 index 0000000..0eb3959 --- /dev/null +++ b/src/rove/bimanual/__init__.py @@ -0,0 +1 @@ +"""ABC bimanual episodes using the shared evaluation lifecycle.""" diff --git a/src/rove/bimanual/abc_process.py b/src/rove/bimanual/abc_process.py new file mode 100644 index 0000000..c8df0f2 --- /dev/null +++ b/src/rove/bimanual/abc_process.py @@ -0,0 +1,135 @@ +"""Standalone ABC simulation runtime. Children inherit the ROVE worker process group.""" + +import contextlib +import json +import os +import subprocess # nosec B404 - trusted local model process; arguments never use a shell +import sys +from pathlib import Path + +from episode import canonical, run_episode + + +def emit(value): + print(canonical(value), file=sys.__stdout__, flush=True) + + +def run(request, directory): + import numpy as np + import torch + + environment, strategy = request["environment"], request["strategy"] + sys.path.insert(0, environment["source"]) + # External deployment state must not override fixed simulator initial joints. + os.environ.pop("DEPLOY_INIT_Q", None) + import abc_sim + + torch.manual_seed(request["seed"]) + np.random.seed(request["seed"]) + env = abc_sim.make_env( + task=environment["task"], + prompt=environment["prompt"], + render_cameras=True, + camera_backend="mujoco", + camera_height=168, + camera_width=224, + chunk_dim=environment["execute_steps"], + max_episode_steps=environment["max_steps"], + terminate_on_success=False, + physics_dt=0.002, + control_decimation=17, + ) + child = None + try: + emit({"type": "progress", "stage": "model", "phase": "started", "data": {}}) + if strategy["kind"] == "abc_vla": + from abc_minimal.policy import VLAInferencePolicy, VLAPolicyConfig + + policy = VLAInferencePolicy( + Path(strategy["checkpoint"]), + VLAPolicyConfig(prompt=environment["prompt"], camera_height=168, camera_width=224), + "cuda", + ) + if policy.action_dim != 14 or policy.camera_keys != ("top", "left", "right"): + raise ValueError("ABC checkpoint does not match the bimanual observation contract") + predict = policy.infer + else: + child_request = directory / "pi05-request.json" + child_request.write_text( + json.dumps( + { + "checkpoint": strategy["checkpoint"], + "seed": request["seed"], + "directory": str(directory), + } + ) + ) + child = subprocess.Popen( # nosec B603 -- registered local runtime, no shell + [ + strategy["python"], + str(Path(__file__).with_name("pi05_process.py")), + str(child_request), + ], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + ) + if json.loads(child.stdout.readline()).get("ready") is not True: + raise ValueError("pi0.5 checkpoint could not load") + + def predict(obs): + np.savez(directory / "observation.npz", state=obs["state"], **obs["images"]) + child.stdin.write(json.dumps({"prompt": obs["prompt"]}) + "\n") + child.stdin.flush() + if json.loads(child.stdout.readline()).get("done") is not True: + raise ValueError("pi0.5 inference failed") + return np.load(directory / "actions.npy", allow_pickle=False) + + emit({"type": "progress", "stage": "model", "phase": "completed", "data": {}}) + + def frame(obs, step): + from PIL import Image + + path = directory / f"frame-{step}.png" + pixels = np.concatenate( + [obs["images"][c].transpose(1, 2, 0) for c in ("top", "left", "right")], axis=1 + ) + Image.fromarray(pixels).save(path) + emit( + { + "type": "artifact", + "id": f"episode.frame.{step}", + "file": path.name, + "media_type": "image/png", + } + ) + + summary = run_episode( + env, + predict, + reset_seed=request["case"]["inputs"]["reset_seed"], + policy_seed=request["seed"], + prompt=environment["prompt"], + max_steps=environment["max_steps"], + execute_steps=environment["execute_steps"], + directory=directory, + emit=emit, + capture_frame=frame, + ) + emit({"type": "result", "summary": summary}) + finally: + env.close() + if child: + child.kill() + child.wait() + + +if __name__ == "__main__": + try: + request_path = Path(sys.argv[1]) + with contextlib.redirect_stdout(sys.stderr): + run(json.loads(request_path.read_text()), request_path.parent) + except Exception as error: + emit({"type": "error", "error_type": type(error).__name__}) + raise SystemExit(1) from None diff --git a/src/rove/bimanual/adapter.py b/src/rove/bimanual/adapter.py new file mode 100644 index 0000000..0231445 --- /dev/null +++ b/src/rove/bimanual/adapter.py @@ -0,0 +1,190 @@ +"""Built-in structured evaluation adapter for independently graded ABC episodes.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +import re +import tempfile +from contextlib import suppress +from pathlib import Path + +from rove.bimanual.config import HERE, Environment, Strategy, validate_profile +from rove.evaluation.context import EvaluationContext +from rove.evaluation.models import CandidateInput, EvaluationConfig +from rove.evaluation.results import EvaluatorResult, Measurement +from rove.trials.snapshots import canonical_json + + +class ABCBimanualEvaluation: + revision = "abc-pi05-episode-v1" + + def bind_context(self, context: EvaluationContext): + self.context = context + + async def predict(self, case: CandidateInput, strategy: dict, seed: int): + environment = Environment.model_validate(self.context.environment) + candidate = Strategy.model_validate(strategy) + if case.inputs.get("task") != environment.task or case.task != environment.prompt: + raise ValueError("Case and frozen simulator task/prompt differ") + reset_seed = case.inputs.get("reset_seed") + if type(reset_seed) is not int or not 0 <= reset_seed <= 2**32 - 1: + raise ValueError("Case must declare an unsigned scene seed") + config = EvaluationConfig( + environment=environment.model_dump(), strategies={"model": strategy} + ) + readiness = await asyncio.to_thread(validate_profile, config) + if not readiness["ready"]: + self.context.emit( + "preflight", "failed", data={"issues": readiness["strategies"][0]["issues"]} + ) + raise ValueError("Registered checkpoint or runtime is unavailable or changed") + self.context.emit("preflight", "completed") + with tempfile.TemporaryDirectory(prefix="rove-abc-") as temporary: + directory = Path(temporary) + request = { + "environment": environment.model_dump(), + "strategy": candidate.model_dump(), + "case": case.model_dump(), + "seed": seed, + } + request_path = directory / "request.json" + request_path.write_text(json.dumps(request)) + process = await asyncio.create_subprocess_exec( + environment.python, + str(HERE / "abc_process.py"), + str(request_path), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + limit=1_000_000, + env={**os.environ, "HF_HUB_OFFLINE": "1", "TRANSFORMERS_OFFLINE": "1"}, + ) + summary = None + try: + while line := await process.stdout.readline(): + event = json.loads(line) + if event["type"] == "progress": + self.context.emit( + event["stage"], + event["phase"], + data=event["data"], + span_id=event.get("span_id"), + ) + elif event["type"] == "artifact": + self._artifact(directory, event) + elif event["type"] == "result": + if summary is not None: + raise ValueError("Duplicate episode result") + summary = event["summary"] + elif event["type"] == "error": + self.context.emit( + "episode", + "failed", + data={ + "exception_type": event["error_type"], + "message": "ABC episode could not complete; check runtime and checkpoint compatibility", + }, + ) + else: + raise ValueError("Invalid episode runtime response") + if await process.wait() != 0 or summary is None: + raise ValueError("ABC episode failed before producing outcome evidence") + if summary.get("reset_seed") != reset_seed or summary.get("policy_seed") != seed: + raise ValueError("Episode runtime returned different scene or policy seeds") + if not (await asyncio.to_thread(validate_profile, config))["ready"]: + raise ValueError("Model or simulator changed during execution") + self.context.record_json("episode.summary", summary) + return summary + finally: + if process.returncode is None: + process.kill() + await process.wait() + # Preserve interrupted trajectory segments too, without grading them. + if "episode.trajectory" not in self.context.evidence_ids: + with suppress(OSError, ValueError): + self.context.record_file( + "episode.partial_trajectory", + directory / "trajectory.jsonl", + "application/x-ndjson", + kind="trajectory", + ) + + def _artifact(self, directory, event): + if not re.fullmatch(r"[A-Za-z0-9_.-]+", event["file"]): + raise ValueError("Runtime artifact must remain within its episode directory") + if not re.fullmatch(r"episode\.[A-Za-z0-9_.-]+", event["id"]): + raise ValueError("Runtime artifact must use the episode namespace") + path = directory / event["file"] + if path.resolve().parent != directory.resolve(): + raise ValueError("Runtime artifact escaped its episode directory") + kind = ( + "image" + if event["media_type"] == "image/png" + else ("trajectory" if event["id"].startswith("episode.trajectory") else "state") + ) + self.context.record_file(event["id"], path, event["media_type"], kind=kind) + + async def grade(self, case, output, reference, grading): + env = Environment.model_validate(self.context.environment) + mode = grading.get("success_mode", env.success_mode) + if mode != env.success_mode: + raise ValueError("Grading and simulator success mode differ") + refs = ["episode.summary", "episode.trajectory", "episode.initial_state"] + if set(refs) - self.context.evidence_ids: + raise ValueError("Measured episode evidence is missing") + if hashlib.sha256( + canonical_json(output).encode() + ).hexdigest() != self.context.evidence_digest("episode.summary"): + raise ValueError("Assessment input differs from the recorded episode summary") + if output.get("reset_seed") != case.inputs.get("reset_seed"): + raise ValueError("Episode evidence belongs to a different scene") + if output.get("scope") != "abc_simulation" or output.get("episode_steps") != env.max_steps: + raise ValueError("Simulator did not complete the declared budget") + if ( + type(output.get("task_success")) is not bool + or type(output.get("final_success")) is not bool + ): + raise ValueError("Simulator outcome was not measured") + success = output["task_success" if mode == "ever" else "final_success"] + units = { + "episode_steps": "step", + "simulated_time_s": "s", + "replans": "count", + "gripper_out_of_range_steps": "step", + "inference_latency_mean_s": "s", + "inference_latency_p95_s": "s", + "inference_wall_s": "s", + } + values = {key: output[key] for key in units} + values.update( + episode_success=float(output["task_success"]), + final_success=float(output["final_success"]), + ) + units.update(episode_success="fraction", final_success="fraction") + return EvaluatorResult( + verdict="pass" if success else "fail", + evidence_quality="observed", + reasoning=( + f"ABC's task evaluator {'observed' if success else 'did not observe'} task success " + f"({'at least once' if mode == 'ever' else 'at the final step'}) within {env.max_steps} simulated control steps." + ), + evidence_refs=refs, + measurements=[ + Measurement( + name=k, + value=v, + unit=units[k], + quality="observed", + evidence_refs=["episode.summary"], + ) + for k, v in values.items() + ], + details={ + "scope": "abc_simulation", + "success_mode": mode, + "physical_robot_success": "not_measured", + "collision_rate": "not_measured", + }, + ) diff --git a/src/rove/bimanual/api.py b/src/rove/bimanual/api.py new file mode 100644 index 0000000..a106fe0 --- /dev/null +++ b/src/rove/bimanual/api.py @@ -0,0 +1,244 @@ +"""Bounded browser controls for host-configured ABC evaluations. + +The browser selects registered strategies and episode budgets. Interpreter paths, +checkpoints and dataset provenance stay in the operator's local profile. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import secrets +from collections.abc import Callable +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Annotated + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, ConfigDict, Field, StrictInt, model_validator + +from rove.benchmarks.store import CampaignStore +from rove.bimanual.config import build_spec, read_profile, validate_profile +from rove.evaluation.executors import prepare_evaluation +from rove.evaluation.trials import run_evaluation_trials + +Seed = Annotated[StrictInt, Field(ge=0, le=2**31 - 1)] + + +class ComparisonRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + name: str = Field(min_length=1, max_length=160) + strategy_ids: list[Annotated[str, Field(min_length=1, max_length=120)]] = Field( + min_length=1, max_length=20 + ) + reset_seeds: list[Seed] = Field(min_length=1, max_length=100) + policy_seeds: list[Seed] = Field(min_length=1, max_length=100) + max_steps: Annotated[StrictInt, Field(ge=1, le=3540)] = 3540 + + @model_validator(mode="after") + def bounded_plan(self): + if not self.name.strip(): + raise ValueError("Give this comparison a name") + for values in (self.strategy_ids, self.reset_seeds, self.policy_seeds): + if len(values) != len(set(values)): + raise ValueError("Strategies and seeds must not contain duplicates") + if len(self.strategy_ids) * len(self.reset_seeds) * len(self.policy_seeds) > 2000: + raise ValueError("Use at most 2,000 trials per comparison") + return self + + +class LaunchRequest(ComparisonRequest): + preview_hash: str = Field(pattern=r"^[a-f0-9]{64}$") + + +def create_bimanual_router(root: Path, launch_campaign: Callable[[str], None]) -> APIRouter: + jobs: dict[str, dict] = {} + running: dict[str, asyncio.Task] = {} + + @asynccontextmanager + async def lifespan(_app): + yield + tasks = list(running.values()) + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + router = APIRouter(prefix="/api/bimanual", lifespan=lifespan) + + def profile(): + try: + return read_profile(root) + except (ValueError, OSError) as error: + raise HTTPException( + 409, "Configure a local ABC profile with the ROVE CLI first" + ) from error + + async def prepare_request(request: ComparisonRequest): + config = profile() + try: + spec, frozen = build_spec( + config, + name=request.name.strip(), + strategy_ids=request.strategy_ids, + reset_seeds=request.reset_seeds, + policy_seeds=request.policy_seeds, + max_steps=request.max_steps, + ) + readiness = await asyncio.to_thread(validate_profile, frozen, verify_assets=True) + except (ValueError, OSError) as error: + raise HTTPException(422, str(error)) from error + selected = {row["id"]: row for row in readiness.get("strategies", [])} + issues = list(readiness.get("issues", [])) + for sid in request.strategy_ids: + row = selected.get(sid) + if not row or not row.get("ready"): + issues.extend( + (row.get("issues") if row else None) or [f"Strategy {sid} is unavailable"] + ) + token = hashlib.sha256( + json.dumps( + {"spec": spec.model_dump(mode="json"), "config": frozen.model_dump(mode="json")}, + sort_keys=True, + separators=(",", ":"), + ).encode() + ).hexdigest() + preview = { + "preview_hash": token, + "name": spec.name, + "planned_trials": spec.planned_trials, + "cases": len(spec.tasks), + "strategies": request.strategy_ids, + "attempts_per_case": len(spec.seeds), + "max_steps": request.max_steps, + "ready": not issues, + "issues": list(dict.fromkeys(issues)), + "success_criterion": frozen.grading, + } + return spec, frozen, preview + + async def reviewed(request: LaunchRequest): + spec, frozen, preview = await prepare_request(request) + if not secrets.compare_digest(request.preview_hash, preview["preview_hash"]): + raise HTTPException( + 409, "Configuration changed. Review the comparison again before running" + ) + if not preview["ready"]: + raise HTTPException( + 409, {"message": "Complete setup before running", "issues": preview["issues"]} + ) + return spec, frozen + + @router.get("/config") + async def configuration(): + try: + config = read_profile(root) + readiness = await asyncio.to_thread(validate_profile, config, verify_assets=True) + except (ValueError, OSError): + return {"configured": False, "ready": False, "strategies": [], "issues": []} + return { + "configured": True, + "ready": readiness.get("ready", False), + "strategies": [ + {key: row[key] for key in ("id", "label", "kind", "ready", "issues") if key in row} + for row in readiness.get("strategies", []) + ], + "issues": readiness.get("issues", []), + "task": "Put plastic bottles in the bin", + "defaults": {"reset_seeds": [0, 1, 2], "policy_seeds": [0, 1, 2], "max_steps": 3540}, + } + + @router.post("/preview") + async def preview(request: ComparisonRequest): + return (await prepare_request(request))[2] + + @router.post("/campaigns", status_code=202) + async def campaign(request: LaunchRequest): + spec, config = await reviewed(request) + if running: + raise HTTPException(409, "Another standalone ABC comparison is running") + try: + payload = await asyncio.to_thread(prepare_evaluation, spec, config) + except ValueError as error: + raise HTTPException(422, str(error)) from error + campaign_id = CampaignStore(root).create(payload) + launch_campaign(campaign_id) + return {"id": campaign_id, "planned_trials": spec.planned_trials} + + async def execute(job_id: str, spec, config): + job = jobs[job_id] + + def on_created(strategy_id: str, trial_id: str): + job["trial_ids"][strategy_id] = trial_id + + try: + result = await run_evaluation_trials( + root=root, + case=spec.tasks[0], + config=config, + strategy_ids=spec.strategies, + execution=spec.execution, + seed=spec.seeds[0], + timeout_s=spec.timeout_s, + on_trial_created=on_created, + ) + job["trial_ids"].update(result["trial_ids"]) + job["status"] = "completed" + except asyncio.CancelledError: + job["status"] = "cancelled" + raise + except Exception: + job["status"] = "error" + job["message"] = "Comparison interrupted. Inspect any recorded trials for details." + finally: + running.pop(job_id, None) + + @router.post("/trials", status_code=202) + async def trials(request: LaunchRequest): + if len(request.reset_seeds) != 1 or len(request.policy_seeds) != 1: + raise HTTPException( + 422, "A standalone comparison uses one scene seed and one model seed" + ) + if running: + raise HTTPException(409, "Another standalone ABC comparison is running") + spec, config = await reviewed(request) + if running: + raise HTTPException(409, "Another standalone ABC comparison is running") + if len(jobs) >= 100: + jobs.pop(next(key for key in jobs if key not in running)) + job_id = secrets.token_hex(16) + jobs[job_id] = { + "id": job_id, + "status": "running", + "trial_ids": {}, + "strategy_ids": spec.strategies, + "tasks": [{"id": task.id, "task": task.task} for task in spec.tasks], + "seeds": spec.seeds, + } + running[job_id] = asyncio.create_task(execute(job_id, spec, config)) + return {**jobs[job_id], "planned_trials": spec.planned_trials} + + def job(job_id: str): + if job_id not in jobs: + raise HTTPException( + 404, "Comparison session not found. Recorded trials remain in Trials." + ) + return jobs[job_id] + + @router.get("/trials/{job_id}") + async def trial_job(job_id: str): + return job(job_id) + + @router.post("/trials/{job_id}/cancel") + async def cancel(job_id: str): + value = job(job_id) + if task := running.get(job_id): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + running.pop(job_id, None) + if value["status"] == "running": + value["status"] = "cancelled" + return value + + return router diff --git a/src/rove/bimanual/cli.py b/src/rove/bimanual/cli.py new file mode 100644 index 0000000..6681931 --- /dev/null +++ b/src/rove/bimanual/cli.py @@ -0,0 +1,191 @@ +"""Set up and run the ABC comparison through ordinary ROVE trials/campaigns.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +from pathlib import Path + +from rove.benchmarks.report import saved_report, to_html +from rove.benchmarks.runner import run_campaign +from rove.benchmarks.store import CampaignStore +from rove.bimanual.config import ( + Environment, + Strategy, + build_spec, + freeze_profile, + read_profile, + validate_profile, +) +from rove.evaluation.executors import prepare_evaluation +from rove.evaluation.models import EvaluationConfig +from rove.evaluation.trials import run_evaluation_trials +from rove.ui.cli_support import emit, read_document, run_cli + + +def main(argv): + parser = argparse.ArgumentParser( + prog="rove bimanual", description="ABC-VLA and adapted pi0.5 on the same bimanual task" + ) + commands = parser.add_subparsers(dest="action", required=True) + init = commands.add_parser("init", help="Write an editable local runtime profile") + init.add_argument("--output", type=Path, required=True) + setup = commands.add_parser( + "setup", help="Register installed checkpoints and fingerprint their runtimes" + ) + setup.add_argument("--config", type=Path, required=True) + commands.add_parser("doctor", help="Check registered assets; does not load model weights") + for name in ("plan", "trial", "run"): + cmd = commands.add_parser(name) + cmd.add_argument("--name", default="ABC bimanual comparison") + cmd.add_argument("--strategy", action="append", required=True) + cmd.add_argument( + "--scene", + action="append", + type=int, + required=True, + help="Fixed initial scene seed; repeat for more cases", + ) + cmd.add_argument( + "--seed", + action="append", + type=int, + required=True, + help="Policy sampling seed; repeat for reliability", + ) + cmd.add_argument("--max-steps", type=int, default=3540) + cmd.add_argument("--output", type=Path) + for cmd in commands.choices.values(): + cmd.add_argument("--store", type=Path, default=Path(".rove/benchmarks")) + args = parser.parse_args(argv) + + def action(): + if args.action == "init": + config = EvaluationConfig( + environment=Environment( + source="/path/to/abc", python="/path/to/abc/.venv/bin/python" + ).model_dump(), + strategies={ + "abc-vla": Strategy( + kind="abc_vla", + label="ABC-VLA", + checkpoint="/path/to/vla_abc130k_200000_v2.pt", + python="/path/to/abc/.venv/bin/python", + ).model_dump(), + "pi05": Strategy( + kind="pi05", + label="Adapted pi0.5", + checkpoint="/path/to/abc-adapted-pi05", + python="/path/to/lerobot/.venv/bin/python", + ).model_dump(), + }, + grading={"success_mode": "ever"}, + ) + with args.output.open("x") as stream: + stream.write(config.model_dump_json(indent=2) + "\n") + emit( + { + "profile": str(args.output.resolve()), + "next": "Edit paths, then rove bimanual setup --config PROFILE", + } + ) + return + if args.action == "setup": + config = freeze_profile(EvaluationConfig.model_validate(read_document(args.config))) + readiness = validate_profile(config) + if not readiness["ready"]: + emit(readiness) + raise ValueError( + "Profile is not ready; install the configured runtimes/checkpoints" + ) + args.store.mkdir(parents=True, exist_ok=True) + path = args.store / "bimanual.json" + temp = path.with_suffix(".tmp") + temp.write_text(config.model_dump_json(indent=2) + "\n") + temp.replace(path) + emit({"profile": str(path.resolve()), **readiness}) + return + config = read_profile(args.store) + if args.action == "doctor": + status = validate_profile(config) + emit(status) + if not status["ready"]: + raise ValueError("Bimanual runtime is not ready") + return + spec, config = build_spec( + config, + name=args.name, + strategy_ids=args.strategy, + reset_seeds=args.scene, + policy_seeds=args.seed, + max_steps=args.max_steps, + ) + readiness = validate_profile(config) + if args.action == "plan": + emit( + { + "spec": spec.model_dump(mode="json"), + "planned_trials": spec.planned_trials, + "readiness": readiness, + }, + args.output, + ) + return + if not readiness["ready"]: + emit(readiness) + raise ValueError("Selected model runtimes are not ready") + if args.action == "trial": + if len(spec.tasks) != 1 or len(spec.seeds) != 1: + raise ValueError( + "A trial comparison uses one scene and one policy seed; use run for a campaign" + ) + result = asyncio.run( + run_evaluation_trials( + root=args.store, + case=spec.tasks[0], + config=config, + strategy_ids=spec.strategies, + execution=spec.execution, + seed=spec.seeds[0], + timeout_s=spec.timeout_s, + ) + ) + emit(result, args.output) + if any(r["execution"] != "completed" for r in result["results"]): + raise RuntimeError( + "One or more model executions failed; inspect the retained traces" + ) + return + store = CampaignStore(args.store) + campaign_id = store.create(prepare_evaluation(spec, config)) + print( + json.dumps({"campaign_id": campaign_id, "planned_trials": spec.planned_trials}), + flush=True, + ) + asyncio.run(run_campaign(store, campaign_id)) + report = saved_report(args.store, campaign_id) + if args.output: + args.output.write_text(to_html(report)) + status = store.get(campaign_id)["status"] + emit( + { + "campaign_id": campaign_id, + "status": status, + "report_url": f"/api/campaigns/{campaign_id}/report?format=html", + } + ) + if ( + status != "completed" + or len(report["trials"]) != spec.planned_trials + or any( + row.get("execution") != "completed" or row.get("outcome") not in {"pass", "fail"} + for row in report["trials"] + ) + ): + raise RuntimeError( + "Campaign execution is incomplete or unassessed; the report and recorded trials " + "have been retained for inspection" + ) + + run_cli(parser, action) diff --git a/src/rove/bimanual/config.py b/src/rove/bimanual/config.py new file mode 100644 index 0000000..ed73dd4 --- /dev/null +++ b/src/rove/bimanual/config.py @@ -0,0 +1,247 @@ +"""Local model profiles and frozen comparison plans; no model downloads on launch.""" + +from __future__ import annotations + +import hashlib +import json +import subprocess # nosec B404 - local setup and fixed runtime probes only +from pathlib import Path +from typing import Literal + +from pydantic import Field + +from rove.benchmarks.models import CampaignSpec, fingerprint +from rove.evaluation.models import BoundedModel, EvaluationCase, EvaluationConfig + +ABC_REVISION = "d0e987b61b376f1a1b8777b19149f190b2ba6939" # pragma: allowlist secret +TASK = "put_plastic_bottles_in_bin" +PROMPT = "sim throw plastic bottles in bin" +HERE = Path(__file__).resolve().parent + + +class Environment(BoundedModel): + type: Literal["abc-bimanual"] = "abc-bimanual" + source: str = Field(min_length=1) + source_revision: Literal[ABC_REVISION] = ABC_REVISION + source_digest: str = "" + python: str = Field(min_length=1) + runtime_digest: str = "" + task: Literal["put_plastic_bottles_in_bin"] = TASK + prompt: str = Field(default=PROMPT, min_length=1, max_length=1000) + max_steps: int = Field(default=3540, ge=1, le=10000) + execute_steps: int = Field(default=15, ge=1, le=30) + camera_height: Literal[168] = 168 + camera_width: Literal[224] = 224 + physics_dt: Literal[0.002] = 0.002 + control_decimation: Literal[17] = 17 + success_mode: Literal["ever", "final"] = "ever" + + +class Strategy(BoundedModel): + kind: Literal["abc_vla", "pi05"] + label: str = Field(min_length=1, max_length=160) + checkpoint: str = Field(min_length=1) + checkpoint_digest: str = "" + python: str = Field(min_length=1) + runtime_digest: str = "" + device: Literal["cuda"] = "cuda" + + +def file_digest(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def tree_digest(path: Path) -> str: + if path.is_file(): + return file_digest(path) + if not path.is_dir(): + raise ValueError("Checkpoint/source directory is missing") + # Include processor stats, configuration, weights and model-specific assets. + entries = { + str(item.relative_to(path)): file_digest(item) + for item in sorted(path.rglob("*")) + if item.is_file() and "__pycache__" not in item.parts and ".git" not in item.parts + } + if not entries: + raise ValueError("Checkpoint/source directory is empty") + return fingerprint(entries) + + +def source_digest(source: str) -> str: + path = Path(source) + result = subprocess.run( # nosec B603 B607 -- trusted local repository; no shell + ["git", "-C", str(path), "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=True, + timeout=10, + ) + if result.stdout.strip() != ABC_REVISION: + raise ValueError("ABC checkout must use the documented pinned revision") + return fingerprint({name: tree_digest(path / name) for name in ("abc_sim", "abc_minimal")}) + + +def runtime_digest(python: str) -> str: + path = Path(python) + if not path.is_absolute() or not path.is_file(): + raise ValueError("Configure an absolute path to an installed Python interpreter") + result = subprocess.run( # nosec B603 -- explicitly configured local interpreter + [str(path), str(HERE / "runtime_probe.py")], + capture_output=True, + text=True, + check=True, + timeout=30, + ) + return fingerprint(json.loads(result.stdout)) + + +def read_profile(root: Path) -> EvaluationConfig: + path = root / "bimanual.json" + if path.stat().st_size > 1_000_000: + raise ValueError("Bimanual profile exceeds 1 MB") + return EvaluationConfig.model_validate_json(path.read_text()) + + +def runtime_health(python: str, kind: str, source: str) -> list[str]: + result = subprocess.run( # nosec B603 -- registered local runtime, fixed probe + [python, str(HERE / "runtime_probe.py"), "health", kind, source], + capture_output=True, + text=True, + check=True, + timeout=90, + ) + return json.loads(result.stdout)["issues"] + + +def freeze_profile(config: EvaluationConfig) -> EvaluationConfig: + """Explicit local setup fingerprints installed assets, not browser supplied paths.""" + env = Environment.model_validate(config.environment) + env.source = str(Path(env.source).expanduser().resolve()) + env.python = str(Path(env.python).expanduser().absolute()) + env.source_digest = source_digest(env.source) + env.runtime_digest = runtime_digest(env.python) + strategies = {} + for sid, value in config.strategies.items(): + strategy = Strategy.model_validate(value) + strategy.checkpoint = str(Path(strategy.checkpoint).expanduser().resolve()) + strategy.python = str(Path(strategy.python).expanduser().absolute()) + strategy.checkpoint_digest = tree_digest(Path(strategy.checkpoint)) + strategy.runtime_digest = runtime_digest(strategy.python) + strategies[sid] = strategy.model_dump(mode="json") + return EvaluationConfig( + environment=env.model_dump(mode="json"), + strategies=strategies, + grading={"success_mode": env.success_mode}, + ) + + +def validate_profile(config: EvaluationConfig, verify_assets: bool = True) -> dict: + env = Environment.model_validate(config.environment) + issues = [] + try: + if not env.source_digest or not env.runtime_digest: + raise ValueError("Register installed runtimes and checkpoints with rove bimanual setup") + if verify_assets and ( + source_digest(env.source) != env.source_digest + or runtime_digest(env.python) != env.runtime_digest + ): + raise ValueError("ABC source or runtime changed; register a new profile") + if verify_assets: + issues.extend(runtime_health(env.python, "abc_vla", env.source)) + except (OSError, ValueError, subprocess.SubprocessError) as error: + issues.append(str(error) if isinstance(error, ValueError) else "ABC runtime is unavailable") + rows = [] + for sid, value in config.strategies.items(): + strategy = Strategy.model_validate(value) + problems = [] + try: + checkpoint = Path(strategy.checkpoint) + if not strategy.checkpoint_digest or not strategy.runtime_digest: + raise ValueError("Checkpoint and runtime have not been registered") + if strategy.kind == "pi05": + for name in ("config.json", "model.safetensors"): + if not (checkpoint / name).is_file(): + raise ValueError("Use a locally adapted pi0.5 checkpoint with 14 actions") + model = json.loads((checkpoint / "config.json").read_text()) + if model.get("output_features", {}).get("action", {}).get("shape") != [14]: + raise ValueError("pi0.5 checkpoint must be trained for ABC's 14 actions") + elif not checkpoint.is_file(): + raise ValueError("ABC-VLA requires a local checkpoint file") + if strategy.kind == "abc_vla" and strategy.python != env.python: + raise ValueError("ABC-VLA must use the shared ABC interpreter") + if verify_assets and ( + tree_digest(checkpoint) != strategy.checkpoint_digest + or runtime_digest(strategy.python) != strategy.runtime_digest + ): + raise ValueError("Checkpoint or runtime changed; register a new profile") + if verify_assets and strategy.kind == "pi05": + problems.extend(runtime_health(strategy.python, "pi05", env.source)) + except (OSError, ValueError, subprocess.SubprocessError) as error: + problems.append( + str(error) if isinstance(error, ValueError) else "Model runtime is unavailable" + ) + rows.append( + { + "id": sid, + "label": strategy.label, + "kind": strategy.kind, + "checkpoint_revision": strategy.checkpoint_digest, + "ready": not problems and not issues, + "issues": problems + issues, + } + ) + return { + "ready": not issues and all(r["ready"] for r in rows), + "strategies": rows, + "issues": issues, + } + + +def build_spec( + config: EvaluationConfig, + *, + name: str, + strategy_ids: list[str], + reset_seeds: list[int], + policy_seeds: list[int], + max_steps: int, +) -> tuple[CampaignSpec, EvaluationConfig]: + env = Environment.model_validate(config.environment) + env.max_steps = max_steps + # Assignment is deliberately revalidated before freezing the public request. + env = Environment.model_validate(env.model_dump()) + if len(reset_seeds) != len(set(reset_seeds)) or any( + type(seed) is not int or not 0 <= seed <= 2**32 - 1 for seed in reset_seeds + ): + raise ValueError("Scene seeds must be unique unsigned 32-bit integers") + if any(sid not in config.strategies for sid in strategy_ids): + raise ValueError("Select registered strategies") + selected = EvaluationConfig( + strategies={sid: config.strategies[sid] for sid in strategy_ids}, + environment=env.model_dump(mode="json"), + grading={"success_mode": env.success_mode}, + ) + spec = CampaignSpec( + name=name, + suite_version=f"abc-{ABC_REVISION[:12]}", + revision="1", + execution="abc-bimanual", + grading="executor_assessment", + strategies=strategy_ids, + tasks=[ + EvaluationCase( + id=f"bottles-scene-{seed}", + task=env.prompt, + inputs={"task": env.task, "reset_seed": seed}, + ) + for seed in reset_seeds + ], + seeds=policy_seeds, + ks=[k for k in (1, 3, 5) if k <= len(policy_seeds)], + timeout_s=3600, + ) + return spec, selected diff --git a/src/rove/bimanual/episode.py b/src/rove/bimanual/episode.py new file mode 100644 index 0000000..4df6707 --- /dev/null +++ b/src/rove/bimanual/episode.py @@ -0,0 +1,206 @@ +"""Closed-loop episode contract. Also importable in ABC's isolated Python runtime.""" + +from __future__ import annotations + +import hashlib +import json +import time +from dataclasses import asdict, is_dataclass +from pathlib import Path + +import numpy as np + + +def plain(value): + if is_dataclass(value): + return plain(asdict(value)) + if isinstance(value, dict): + return {str(k): plain(v) for k, v in value.items()} + if isinstance(value, tuple | list): + return [plain(v) for v in value] + if isinstance(value, np.ndarray): + return value.tolist() + if isinstance(value, np.generic): + return value.item() + return value + + +def canonical(value): + return json.dumps(plain(value), sort_keys=True, separators=(",", ":"), allow_nan=False) + + +def valid_actions(actions, execute_steps): + values = np.asarray(actions, dtype=np.float32) + if values.ndim != 2 or values.shape[1] != 14 or not execute_steps <= len(values) <= 1000: + raise ValueError("Policy must return at least the execution horizon of 14D actions") + if not np.isfinite(values).all(): + raise ValueError("Policy returned non-finite actions") + # Arm targets are absolute radians, not the placeholder [-1,1] Gym box. + return values + + +def run_episode( + env, + predict, + *, + reset_seed, + policy_seed, + prompt, + max_steps, + execute_steps, + directory: Path, + emit, + capture_frame=None, +): + """Run fresh observations after each executed prefix; record every control step.""" + obs, reset_info = env.reset(seed=reset_seed, randomize=True) + initial = { + "state": plain(env.capture_state()), + "reset": plain(reset_info), + "requested_seed": reset_seed, + } + randomization = initial["reset"].get("randomization") + if not randomization or randomization.get("seed") != reset_seed: + raise ValueError("Simulator did not confirm the requested initial scene seed") + evaluator = env.evaluate_task() + if evaluator is None: + raise ValueError("This simulator task has no outcome evaluator") + if evaluator.scalar_success(): + raise ValueError("Case is already successful before the first model action") + initial["digest"] = hashlib.sha256(canonical(initial).encode()).hexdigest() + initial_path = directory / "initial.json" + initial_path.write_text(canonical(initial)) + emit( + { + "type": "artifact", + "id": "episode.initial_state", + "file": initial_path.name, + "media_type": "application/json", + } + ) + if capture_frame: + capture_frame(obs, 0) + steps, successes, latencies, gripper_out_of_range_steps = 0, [], [], 0 + final_info = {} + start_sim_time = float(env.capture_state()["time"]) + trajectory = directory / "trajectory.jsonl" + with trajectory.open("w") as log: + while steps < max_steps: + obs["prompt"] = prompt + emit( + { + "type": "progress", + "stage": "policy", + "phase": "started", + "span_id": f"policy-{len(latencies) + 1}", + "data": {"replan": len(latencies) + 1}, + } + ) + begin = time.perf_counter() + actions = valid_actions(predict(obs), execute_steps) + elapsed = time.perf_counter() - begin + latencies.append(elapsed) + emit( + { + "type": "progress", + "stage": "policy", + "phase": "completed", + "span_id": f"policy-{len(latencies)}", + "data": { + "replan": len(latencies), + "inference_s": elapsed, + "predicted_actions": len(actions), + "completed_steps": steps, + "max_steps": max_steps, + }, + } + ) + segment = [] + for action in actions[: min(execute_steps, max_steps - steps)]: + if (action[[6, 13]] < 0).any() or (action[[6, 13]] > 1).any(): + gripper_out_of_range_steps += 1 + state_before = obs["state"] + # Request images once per prefix, not 15 redundant render passes. + _, reward, terminated, truncated, info = env.step(action, render_obs=False) + steps += 1 + if type(info.get("task_success")) not in (bool, np.bool_): + raise ValueError("Simulator stopped supplying a measured task outcome") + successes.append(bool(info["task_success"])) + final_info = info + row = ( + canonical( + { + "step": steps, + "state_before": state_before, + "action": action, + "state_after": env.capture_state(), + "reward": reward, + "evaluation": info, + } + ) + + "\n" + ) + log.write(row) + segment.append(row) + # get_obs updates state and images at the replan boundary below. + obs["state"] = env.get_state() + if terminated or truncated: + break + log.flush() + # Commit each completed action prefix before the next model call. A + # hard process-group timeout must not erase the preceding trajectory. + part = directory / f"trajectory-{len(latencies)}.jsonl" + part.write_text("".join(segment)) + emit( + { + "type": "artifact", + "id": f"episode.trajectory.{len(latencies)}", + "file": part.name, + "media_type": "application/x-ndjson", + } + ) + obs = env.get_obs() + emit( + { + "type": "progress", + "stage": "episode", + "phase": "progress", + "data": { + "completed_steps": steps, + "max_steps": max_steps, + "task_success": successes[-1], + }, + } + ) + if capture_frame and (len(latencies) % 10 == 0 or steps >= max_steps): + capture_frame(obs, steps) + if terminated or truncated: + break + if steps < max_steps: + raise ValueError("Episode ended before its shared control budget") + summary = { + "task_success": any(successes), + "final_success": successes[-1], + "episode_steps": steps, + "simulated_time_s": float(env.capture_state()["time"]) - start_sim_time, + "replans": len(latencies), + "inference_latency_mean_s": float(np.mean(latencies)), + "gripper_out_of_range_steps": gripper_out_of_range_steps, + "inference_latency_p95_s": float(np.percentile(latencies, 95)), + "inference_wall_s": sum(latencies), + "reset_seed": reset_seed, + "policy_seed": policy_seed, + "initial_state_digest": initial["digest"], + "prompt": prompt, + "final_evaluation": plain(final_info), + "scope": "abc_simulation", + } + emit( + { + "type": "artifact", + "id": "episode.trajectory", + "file": trajectory.name, + "media_type": "application/x-ndjson", + } + ) + return summary diff --git a/src/rove/bimanual/pi05_process.py b/src/rove/bimanual/pi05_process.py new file mode 100644 index 0000000..5a71830 --- /dev/null +++ b/src/rove/bimanual/pi05_process.py @@ -0,0 +1,50 @@ +"""Persistent pi0.5 inference child; executed only by the ABC episode process.""" + +import contextlib +import json +import sys +from pathlib import Path + + +def serve(request): + import numpy as np + import torch + from pi05_weights import load_policy + + torch.manual_seed(request["seed"]) + np.random.seed(request["seed"]) + policy, preprocess, postprocess = load_policy(Path(request["checkpoint"]), "cuda") + policy.reset() + for processor in (preprocess, postprocess): + if hasattr(processor, "reset"): + processor.reset() + directory = Path(request["directory"]) + print('{"ready":true}', file=sys.__stdout__, flush=True) + for line in sys.stdin: + message = json.loads(line) + if message.get("stop"): + break + with np.load(directory / "observation.npz", allow_pickle=False) as arrays: + batch = { + "observation.state": torch.from_numpy(arrays["state"].copy()), + "task": message["prompt"], + } + for camera in ("top", "left", "right"): + batch[f"observation.images.{camera}"] = ( + torch.from_numpy(arrays[camera].copy()).float() / 255.0 + ) + with torch.inference_mode(): + chunk = policy.predict_action_chunk(preprocess(batch)) + action = postprocess(chunk)[0].detach().cpu().float().numpy() + np.save(directory / "actions.npy", action, allow_pickle=False) + print('{"done":true}', file=sys.__stdout__, flush=True) + + +if __name__ == "__main__": + try: + with contextlib.redirect_stdout(sys.stderr): + serve(json.loads(Path(sys.argv[1]).read_text())) + except Exception as error: + # Model exceptions can contain private tokens/paths. Parent exposes only type. + print(json.dumps({"error": type(error).__name__}), file=sys.__stdout__, flush=True) + raise SystemExit(1) from None diff --git a/src/rove/bimanual/pi05_weights.py b/src/rove/bimanual/pi05_weights.py new file mode 100644 index 0000000..b180e04 --- /dev/null +++ b/src/rove/bimanual/pi05_weights.py @@ -0,0 +1,80 @@ +"""Fail-closed loading for the pinned LeRobot pi0.5 implementation. + +The upstream loader catches weight-loading exceptions and returns a model. An +evaluation must never mistake that uninitialized model for a trained policy. +This uses the upstream key conversion with a strict local safetensors load. +""" + +from pathlib import Path + +CAMERA_FEATURES = ( + "observation.images.top", + "observation.images.left", + "observation.images.right", +) + + +def load_policy(checkpoint: Path, device: str): + """Load an ABC-adapted local checkpoint and its saved normalization pipelines.""" + from lerobot.configs.policies import PreTrainedConfig + from lerobot.policies.factory import make_pre_post_processors + + checkpoint = Path(checkpoint).resolve(strict=True) + config = PreTrainedConfig.from_pretrained(checkpoint, local_files_only=True) + if tuple(config.output_features["action"].shape) != (14,): + raise ValueError("pi0.5 checkpoint must be adapted to ABC's 14D bimanual actions") + if tuple(config.input_features["observation.state"].shape) != (14,): + raise ValueError("pi0.5 checkpoint must consume ABC's 14D bimanual state") + if set(config.image_features) != set(CAMERA_FEATURES): + raise ValueError("pi0.5 checkpoint must consume the ABC top, left and right cameras") + if config.use_relative_actions: + raise ValueError("This comparison recipe requires absolute joint-position actions") + config.device = device + config.compile_model = False + policy = load_local_pi05(checkpoint, config=config).to(device).eval() + preprocessor, postprocessor = make_pre_post_processors( + config, + pretrained_path=str(checkpoint), + preprocessor_overrides={"device_processor": {"device": device}}, + ) + return policy, preprocessor, postprocessor + + +def load_local_pi05(checkpoint, *, config=None, **kwargs): + from lerobot.configs.policies import PreTrainedConfig + from lerobot.policies.pi05.modeling_pi05 import PI05Policy + from safetensors.torch import load_file + + checkpoint = Path(checkpoint).resolve(strict=True) + if not checkpoint.is_dir() or not (checkpoint / "model.safetensors").is_file(): + raise ValueError("A local pi0.5 snapshot with model.safetensors is required") + config = config or PreTrainedConfig.from_pretrained(checkpoint, local_files_only=True) + if config.type != "pi05": + raise ValueError("The checkpoint must use policy type pi05") + if config.use_visual_memory or config.use_proprioceptive_memory: + raise ValueError("This first comparison requires a pi0.5 checkpoint without memory") + state_dict = load_file(str(checkpoint / "model.safetensors"), device="cpu") + if not state_dict: + raise ValueError("Empty pi0.5 checkpoint") + # The native constructor accepts dataset_stats/dataset_meta as extra kwargs. + policy = PI05Policy(config, **kwargs) + converted = policy._fix_pytorch_state_dict_keys(state_dict, config) + converted = { + key if key.startswith("model.") else f"model.{key}": value + for key, value in converted.items() + } + policy.load_state_dict(converted, strict=True) + policy.eval() + return policy + + +def install_strict_training_loader(): + """Use the same failure behavior inside LeRobot's ordinary training CLI.""" + from lerobot.policies.pi05.modeling_pi05 import PI05Policy + + def from_pretrained(cls, pretrained_name_or_path, *, config=None, **kwargs): + # Local snapshots pin weights; this runner does not download or resume. + kwargs.pop("revision", None) + return load_local_pi05(pretrained_name_or_path, config=config, **kwargs) + + PI05Policy.from_pretrained = classmethod(from_pretrained) diff --git a/src/rove/bimanual/runtime_probe.py b/src/rove/bimanual/runtime_probe.py new file mode 100644 index 0000000..42024af --- /dev/null +++ b/src/rove/bimanual/runtime_probe.py @@ -0,0 +1,66 @@ +"""Small dependency fingerprint, executable in either isolated interpreter.""" + +import contextlib +import hashlib +import importlib.metadata +import json +import os +import platform +import sys + + +def health(kind, source): + if platform.system() != "Linux": + return { + "issues": [ + "Learned-policy evaluation requires Linux with an NVIDIA GPU; local data inspection remains available" + ] + } + issues = [] + try: + import torch + + if not torch.cuda.is_available(): + issues.append( + "CUDA is unavailable in this interpreter; install its CUDA runtime and provide an NVIDIA GPU" + ) + if kind == "abc_vla": + sys.path.insert(0, source) + os.environ.pop("DEPLOY_INIT_Q", None) + import abc_sim + from abc_minimal.policy import VLAInferencePolicy # noqa: F401 + + env = abc_sim.make_env(task="put_plastic_bottles_in_bin", render_cameras=False) + env.close() + else: + from lerobot.policies.pi05.modeling_pi05 import PI05Policy # noqa: F401 + except Exception as error: + issues.append( + f"{kind} dependencies or simulator assets could not load ({type(error).__name__}); follow the isolated runtime setup" + ) + return {"issues": issues} + + +if __name__ == "__main__": + if sys.argv[1:2] == ["health"]: + with contextlib.redirect_stdout(sys.stderr): + status = health(sys.argv[2], sys.argv[3]) + print(json.dumps(status)) + raise SystemExit(0) + print( + json.dumps( + { + "python": platform.python_version(), + "platform": platform.platform(), + "packages": sorted( + ( + d.metadata["Name"], + d.version, + d.read_text("direct_url.json"), + hashlib.sha256((d.read_text("RECORD") or "").encode()).hexdigest(), + ) + for d in importlib.metadata.distributions() + ), + } + ) + ) diff --git a/src/rove/evaluation/context.py b/src/rove/evaluation/context.py new file mode 100644 index 0000000..d45f74e --- /dev/null +++ b/src/rove/evaluation/context.py @@ -0,0 +1,143 @@ +"""Optional durable progress and evidence access for trusted evaluation adapters. + +This is a host integration contract, not a sandbox. Candidate inputs and private +references remain separate; the context contains only declared environment data. +""" + +from __future__ import annotations + +import copy +import re +import time +from pathlib import Path + +from rove.trials.snapshots import canonical_json +from rove.trials.store import MAX_ASSET_BYTES, TrialStore + +_RESERVED = frozenset( + {"candidate.output", "case.reference", "assessment.output", "evaluation.error"} +) + + +class EvaluationContext: + def __init__(self, store: TrialStore, trial_id: str, environment: dict): + self._store = store + self.trial_id = trial_id + self.clock_id = f"worker:{trial_id}" + self._environment = copy.deepcopy(environment) + + @property + def environment(self) -> dict: + """A fresh copy prevents an adapter from changing the frozen declaration.""" + return copy.deepcopy(self._environment) + + @property + def evidence_ids(self) -> frozenset[str]: + """Only currently available, integrity-checked evidence may support grading.""" + return frozenset( + ref["id"] + for ref in self._store.evidence(self.trial_id) + if ref["availability"] == "available" + ) + + def evidence_digest(self, reference_id: str) -> str: + """Return the verified stored hash for comparing an assessment input to evidence.""" + for reference in self._store.evidence(self.trial_id): + if reference["id"] == reference_id and reference["availability"] == "available": + return reference["asset_sha256"] + raise ValueError("Evidence is missing or changed") + + def emit( + self, + stage: str, + phase: str, + *, + name: str | None = None, + data: dict | None = None, + span_id: str | None = None, + evidence_refs: list[str] | None = None, + ) -> None: + """Append before continuing so an interrupted episode retains its progress.""" + for value in (stage, phase): + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,79}", value): + raise ValueError("Progress stage and phase must be bounded identifiers") + refs = list(evidence_refs or []) + if refs and set(refs) - self.evidence_ids: + raise ValueError("Progress refers to evidence that was not recorded") + self._store.append_event( + self.trial_id, + { + "event_type": f"evaluation.{phase}", + "stage": stage, + "name": name or f"{stage.title()} {phase}", + "data": copy.deepcopy(data or {}), + "source": "rove.evaluation", + "source_clock_id": self.clock_id, + "time_unit": "ns", + "source_timestamp": time.monotonic_ns(), + "span_id": span_id or f"{self.trial_id}:{stage}", + "evidence_refs": refs, + }, + ) + + def record_bytes( + self, + reference_id: str, + data: bytes, + media_type: str, + *, + kind: str = "tool_output", + units: dict[str, str] | None = None, + clock_id: str | None = None, + frame: str | None = None, + ) -> dict: + """Save at most 64 MiB using existing content-addressed managed assets.""" + if reference_id in _RESERVED: + raise ValueError("Evidence ID is reserved for the evaluation lifecycle") + return self._record( + reference_id, + data, + media_type, + kind=kind, + units=units, + clock_id=clock_id, + frame=frame, + ) + + def record_json(self, reference_id: str, value, **metadata) -> dict: + return self.record_bytes( + reference_id, canonical_json(value).encode(), "application/json", **metadata + ) + + def record_file(self, reference_id: str, path: Path, media_type: str, **metadata) -> dict: + """Read a bounded artifact produced by the trusted adapter, not a manifest URL.""" + with Path(path).open("rb") as stream: + data = stream.read(MAX_ASSET_BYTES + 1) + if len(data) > MAX_ASSET_BYTES: + raise ValueError("Artifact exceeds 64 MiB; record bounded segments") + return self.record_bytes(reference_id, data, media_type, **metadata) + + def _record( + self, + reference_id: str, + data: bytes, + media_type: str, + *, + kind: str = "tool_output", + units: dict[str, str] | None = None, + clock_id: str | None = None, + frame: str | None = None, + ) -> dict: + asset = self._store.save_asset(data, media_type) + reference = { + "id": reference_id, + "asset_sha256": asset["sha256"], + "kind": kind, + "units": units or {}, + } + if clock_id is not None: + reference["clock_id"] = clock_id + if frame is not None: + reference["frame"] = frame + self._store.attach_evidence(self.trial_id, reference) + return {**reference, "asset": asset} diff --git a/src/rove/evaluation/registry.py b/src/rove/evaluation/registry.py index c4362b1..bcbbe77 100644 --- a/src/rove/evaluation/registry.py +++ b/src/rove/evaluation/registry.py @@ -15,12 +15,18 @@ def resolve(name: str) -> tuple[type[EvaluationAdapter], dict]: if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,79}", name) or name == "robotics": raise ValueError("Invalid evaluation executor name") entries = list(importlib.metadata.entry_points(group="rove.evaluations", name=name)) - if name == "text-example": - from rove.evaluation.examples import TextEvaluation - + if name in {"text-example", "abc-bimanual"}: if entries: - raise ValueError("Installed executor conflicts with the built-in text-example") - factory, distribution = TextEvaluation, "rove-eval" + raise ValueError(f"Installed executor conflicts with the built-in {name}") + if name == "text-example": + from rove.evaluation.examples import TextEvaluation + + factory = TextEvaluation + else: + from rove.bimanual.adapter import ABCBimanualEvaluation + + factory = ABCBimanualEvaluation + distribution = "rove-eval" else: if len(entries) != 1: raise ValueError(f"Install exactly one trusted rove.evaluations executor named {name}") @@ -33,11 +39,16 @@ def resolve(name: str) -> tuple[type[EvaluationAdapter], dict]: if not isinstance(revision, str) or not revision.strip() or len(revision) > 160: raise ValueError("Executor revision must contain 1-160 characters") sources = {} - for name_, value in ( + methods = [ ("adapter", factory), ("candidate", factory.predict), ("grader", factory.grade), - ): + ] + if hasattr(factory, "bind_context"): + if not callable(factory.bind_context): + raise ValueError("Optional bind_context must be callable") + methods.append(("context_binding", factory.bind_context)) + for name_, value in methods: path = inspect.getsourcefile(value) if not path: raise ValueError("Executor requires inspectable Python source for provenance") diff --git a/src/rove/evaluation/trials.py b/src/rove/evaluation/trials.py index 2962a33..ab38d15 100644 --- a/src/rove/evaluation/trials.py +++ b/src/rove/evaluation/trials.py @@ -22,6 +22,7 @@ async def run_evaluation_trials( execution: str, seed: int = 0, timeout_s: float = 120, + on_trial_created=None, ) -> dict: spec = CampaignSpec( name="Standalone evaluation", @@ -49,6 +50,8 @@ async def run_evaluation_trials( seed=seed, ) identities[sid] = trial_id + if on_trial_created is not None: + on_trial_created(sid, trial_id) started = time.monotonic() try: result = await run_attempt( diff --git a/src/rove/evaluation/worker.py b/src/rove/evaluation/worker.py index 07fe0f6..c362d02 100644 --- a/src/rove/evaluation/worker.py +++ b/src/rove/evaluation/worker.py @@ -4,13 +4,16 @@ import asyncio import copy +import inspect import json import sys import time +from contextlib import suppress from pathlib import Path from pydantic import JsonValue, TypeAdapter +from rove.evaluation.context import EvaluationContext from rove.evaluation.models import EvaluationCase from rove.evaluation.registry import resolve from rove.evaluation.results import EvaluatorResult @@ -22,88 +25,81 @@ async def execute(request: dict) -> dict: factory, identity = resolve(request["execution"]) if identity != request["executor_identity"]: raise ValueError("Evaluation executor changed before execution") - adapter = factory() case = EvaluationCase.model_validate(request["task"]) store, trial_id = TrialStore(Path(request["trial_root"])), request["trial_id"] config = request["config"] started = time.monotonic() - - def event(stage, phase): - store.append_event( - trial_id, - { - "event_type": f"evaluation.{phase}", - "stage": stage, - "name": f"{stage.title()} {phase}", - "source": "rove.evaluation", - "source_clock_id": f"worker:{trial_id}", - "time_unit": "ns", - "source_timestamp": time.monotonic_ns(), - "span_id": f"{trial_id}:{stage}", - }, + context = EvaluationContext(store, trial_id, config.get("environment", {})) + stage = "setup" + try: + adapter = factory() + if binder := getattr(adapter, "bind_context", None): + bound = binder(context) + if inspect.isawaitable(bound): + await bound + stage = "candidate" + context.emit(stage, "started") + output = await adapter.predict( + case.candidate(), + copy.deepcopy(config["strategies"][request["strategy_id"]]), + request["seed"], ) - - event("candidate", "started") - output = await adapter.predict( - case.candidate(), - copy.deepcopy(config["strategies"][request["strategy_id"]]), - request["seed"], - ) - output = TypeAdapter(JsonValue).validate_python(output) - encoded = canonical_json(output).encode() - if len(encoded) > 4_000_000: - raise ValueError("Candidate output exceeds 4 MB") - # Store raw output once as managed evidence; sanitize the public record below. - asset = store.save_asset(encoded, "application/json") - store.attach_evidence( - trial_id, - { - "id": "candidate.output", - "asset_sha256": asset["sha256"], - "kind": "tool_output", - "units": {}, - }, - ) - event("candidate", "completed") - reference_asset = store.save_asset(canonical_json(case.reference).encode(), "application/json") - store.attach_evidence( - trial_id, - { - "id": "case.reference", - "asset_sha256": reference_asset["sha256"], - "kind": "tool_output", - "units": {}, - }, - ) - event("assessment", "started") - # Fresh validated copies prevent candidate-side mutation from changing the grader inputs. - case = EvaluationCase.model_validate(request["task"]) - graded = await adapter.grade( - case.candidate(), copy.deepcopy(output), case.reference, copy.deepcopy(config["grading"]) - ) - assessment = EvaluatorResult.model_validate( - graded.model_dump(mode="json") if isinstance(graded, EvaluatorResult) else graded - ) - refs = set(assessment.evidence_refs) - refs.update(ref for m in assessment.measurements for ref in m.evidence_refs) - if refs - {"candidate.output", "case.reference"}: - raise ValueError("Assessment refers to evidence that was not recorded") - assessment_payload = assessment.model_dump(mode="json") - # Round-trip prohibits NaN in free-form details and validates bounded storage. - encoded_assessment = canonical_json(assessment_payload).encode() - if len(encoded_assessment) > 1_000_000: - raise ValueError("Assessment exceeds 1 MB") - assessment_asset = store.save_asset(encoded_assessment, "application/json") - store.attach_evidence( - trial_id, - { - "id": "assessment.output", - "asset_sha256": assessment_asset["sha256"], - "kind": "tool_output", - "units": {}, - }, - ) - event("assessment", "completed") + output = TypeAdapter(JsonValue).validate_python(output) + encoded = canonical_json(output).encode() + if len(encoded) > 4_000_000: + raise ValueError("Candidate output exceeds 4 MB") + # Store raw output once as managed evidence; sanitize the public record below. + context._record("candidate.output", encoded, "application/json") + context.emit(stage, "completed") + context._record( + "case.reference", canonical_json(case.reference).encode(), "application/json" + ) + stage = "assessment" + context.emit(stage, "started") + # Fresh copies prevent candidate mutation from changing the grader inputs. + case = EvaluationCase.model_validate(request["task"]) + graded = await adapter.grade( + case.candidate(), + copy.deepcopy(output), + case.reference, + copy.deepcopy(config["grading"]), + ) + assessment = EvaluatorResult.model_validate( + graded.model_dump(mode="json") if isinstance(graded, EvaluatorResult) else graded + ) + refs = set(assessment.evidence_refs) + refs.update(ref for m in assessment.measurements for ref in m.evidence_refs) + if refs - context.evidence_ids: + raise ValueError("Assessment refers to evidence that was not recorded") + assessment_payload = assessment.model_dump(mode="json") + # Round-trip prohibits NaN in free-form details and validates bounded storage. + encoded_assessment = canonical_json(assessment_payload).encode() + if len(encoded_assessment) > 1_000_000: + raise ValueError("Assessment exceeds 1 MB") + context._record("assessment.output", encoded_assessment, "application/json") + context.emit(stage, "completed") + except BaseException as error: + # Provider exceptions often contain URLs, credentials or customer inputs. + # Persist the stage and type only, while keeping already-written evidence. + with suppress(Exception): + context._record( + "evaluation.error", + canonical_json( + { + "stage": stage, + "exception_type": type(error).__name__, + "message": "Evaluation stage did not complete", + } + ).encode(), + "application/json", + ) + context.emit( + stage, + "cancelled" if isinstance(error, asyncio.CancelledError) else "failed", + data={"exception_type": type(error).__name__}, + evidence_refs=["evaluation.error"], + ) + raise return { "outcome": assessment.verdict, "execution": "completed", diff --git a/src/rove/ui/cli.py b/src/rove/ui/cli.py index a4b8b4e..bf1916c 100644 --- a/src/rove/ui/cli.py +++ b/src/rove/ui/cli.py @@ -8,6 +8,7 @@ def main(): commands = { + "bimanual": "rove.bimanual.cli", "trial": "rove.trials.cli", "campaign": "rove.benchmarks.workflow_cli", "baseline": "rove.benchmarks.baseline_cli", @@ -34,7 +35,7 @@ def main(): return parser = argparse.ArgumentParser( description="ROVE local robotics evaluation workbench", - epilog="Commands: trial, campaign, baseline, data, strategy, config, assistant, history, benchmark, exchange. Use COMMAND --help for automation options.", + epilog="Commands: trial, campaign, baseline, data, strategy, config, assistant, history, benchmark, exchange, bimanual. Use COMMAND --help for automation options.", ) parser.add_argument("command", nargs="?", choices=["serve"], default="serve") parser.add_argument("--port", type=int, default=5001) diff --git a/tests/bimanual_ui.cjs b/tests/bimanual_ui.cjs new file mode 100644 index 0000000..a678a7a --- /dev/null +++ b/tests/bimanual_ui.cjs @@ -0,0 +1,124 @@ +"use strict"; +const test = require("node:test"), assert = require("node:assert/strict"), fs = require("node:fs"), path = require("node:path"); +const {JSDOM} = require("jsdom"); +const {parseSeeds, countLabel, criterionLabel} = require("../frontend/bimanual.js"); +const root = path.resolve(__dirname, ".."); +const read = file => fs.readFileSync(path.join(root, "frontend", file), "utf8"); +const pause = () => new Promise(resolve => setTimeout(resolve, 30)); + +test("seed inputs retain distinct fixed scenes and repeated model attempts", () => { + assert.deepEqual(parseSeeds("0, 9, 17"), [0, 9, 17]); + for (const value of ["", "0,", "0,0", "-1", "1.3", "Infinity", "2147483648", "") + section = episode_measurement_section(episode_report) + assert "<script>" in section + assert "