Add shared contracts for Atomic Skill benchmark suites

scripts/benchmark/motion_generation/README.md

@@ -101,3 +101,103 @@ scenes. Use `--no-headless` instead to open the live simulator viewer.
   against non-target appliance links or other environment geometry
 - Dual-arm actions and multi-action chains
 - Latency-budget Pareto sweeps, confidence intervals, subprocess isolation
+
+
+## Atomic Skill contract foundation (#669)
+
+The first migration layer provides shared contracts while preserving existing
+skill scoring. It does not yet implement official embodiment component loading,
+new physical success criteria, segment/contact sampling, domain generation, or
+domain-level aggregation. The three existing report tables remain unchanged.
+
+### Domain provenance
+
+Add `domain` alongside `scenario` in an existing track. This YAML fragment
+shows the new fields; retain the track's existing `config`:
+
+```yaml
+tracks:
+  - id: atomic-nominal
+    scenario: atomic_task
+    domain:
+      id: nominal
+      version: v1
+      kind: nominal
+```
+
+`domain` accepts exactly `id`, `version`, and `kind`. The kind is `nominal`,
+`robustness`, or `held_out`; IDs and versions must be nonempty strings. This
+field declares provenance only: it does not perturb scenes, select held-out
+objects, or establish training/evaluation separation. Use separate tracks for
+separate populations until domain generators are added. Existing suites without
+this declaration retain `domain: null`; randomized suites are not silently
+classified as nominal.
+
+The runner snapshots this identity onto generated cases and carries it into all
+associated lifecycle/trial records. `case_manifest.json` now uses
+`case_schema_version: 3` and contains the optional `domain` on each case. Each
+track/batch population is written before any planner evaluates that population;
+later populations extend the manifest. Planner configuration hashes stay in
+trial records, separate from planner-independent case content.
+
+### Provider and evaluator interfaces
+
+`atomic_skill.py` owns `AtomicSkillCaseProvider`; the previous import from
+`scenarios.atomic_task` remains available. Existing providers continue to use
+`task_result()`. A migrated provider overrides `physical_evaluator()` to return
+an implementation of `contracts.PhysicalStageEvaluator`:
+
+```python
+def evaluate(
+    self,
+    case: BenchmarkCase,
+    observation: ExecutionObservation,
+    motion_outcomes: tuple[CaseOutcome, ...],
+) -> tuple[PhysicalEvaluation, ...]:
+    ...
+```
+
+The evaluator receives frozen task inputs, measured replay observations, and
+motion-validation results. It receives neither the scenario/engine nor a
+`CompiledTrajectory` or `projected_context`. `ExecutionObservation` currently
+contains the existing measured replay summary. Missing optional measurements
+mean unavailable evidence; the follow-up evaluation work must extend observation
+collection before claiming grasp/release/retention or intermediate-stage success.
+
+Return one `PhysicalEvaluation` per environment, each containing the same ordered
+stage IDs. `StageOutcome.status` distinguishes `passed`, `failed`, `not_reached`,
+and `not_applicable`. Failed stages require a failure code; measurements retain
+finite numeric values with units in their names. After a failed or unreached
+stage, subsequent stages cannot be reached. Duplicate/missing environment rows,
+duplicate stage IDs, and inconsistent stage order are evaluator errors, recorded
+by the existing runner as `metric_evaluation_error`.
+
+Task success requires at least one applicable stage and every applicable stage
+to pass. An all-inapplicable or incomplete result cannot establish success.
+Physical success also remains gated by motion validity and execution success.
+Raw outcomes retain `stages` and `failure_stage`; legacy outcomes have empty
+stages. Stage/domain aggregation and reporting are separate follow-up work.
+
+### Follow-up ownership
+
+- **Embodiment and case integration:** build against `AtomicSkillCaseProvider`,
+  reuse official component loading, and keep case generation independent of
+  planner outcomes. Do not duplicate the runner lifecycle or artifact writers.
+- **Physical evaluation:** extend `ExecutionObservation` with measured segment
+  and contact evidence, implement evaluators and counterexample tests, then opt
+  providers into `physical_evaluator()` individually.
+- **Framework and statistics:** add versioned domain generators and aggregation
+  after agreeing on capability eligibility, stage denominators, and retention.
+
+Contract tests can run without the simulator engine:
+
+```bash
+python -m pytest -q tests/benchmark/motion_generation/test_contracts.py
+```
+
+Runner/evaluator integration and existing benchmark regressions use the normal
+EmbodiChain environment:
+
+```bash
+python -m pytest -q tests/benchmark/motion_generation
+```

scripts/benchmark/motion_generation/artifacts.py

@@ -22,6 +22,7 @@
 import json
 import platform
 import subprocess
+from dataclasses import asdict
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
@@ -137,6 +138,7 @@ def _case_to_dict(case: BenchmarkCase) -> dict[str, Any]:
         ),
         "target_waypoints": case.target_waypoints.detach().cpu().tolist(),
         "case_parameters": _to_json_value(case.case_parameters),
+        "domain": None if case.domain is None else asdict(case.domain),
         "validity_evidence": {
             "method": (
                 "reference_qpos_fk"
@@ -168,7 +170,7 @@ def write_case_manifest(path: str | Path, cases: list[BenchmarkCase]) -> Path:
     return write_json(
         path,
         {
-            "case_schema_version": 2,
+            "case_schema_version": 3,
             "cases": [_case_to_dict(case) for case in cases],
         },
     )

scripts/benchmark/motion_generation/atomic_skill.py

@@ -0,0 +1,105 @@
+# ----------------------------------------------------------------------------
+# Copyright (c) 2021-2026 DexForce Technology Co., Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ----------------------------------------------------------------------------
+
+"""Case construction contract for Atomic Skill benchmark integrations."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Mapping
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import torch
+    from embodichain.lab.sim.atomic_actions import ActionInvocation, TaskState
+    from embodichain.lab.sim.atomic_actions.plans import CompiledTrajectory
+    from .config import SuiteCfg, TrackCfg
+    from .contracts import ExecutionObservation, PhysicalStageEvaluator
+    from .models import BenchmarkCase, CaseOutcome
+    from .planners.base import PlannerAdapter
+    from .scenarios.atomic_task import AtomicTaskScenario
+
+__all__ = ["AtomicSkillCaseProvider"]
+
+
+class AtomicSkillCaseProvider(ABC):
+    """Generate and ground one Atomic Action without planner-specific logic."""
+
+    skill_id: str
+    requires_gripper = False
+
+    @abstractmethod
+    def generate_case(
+        self,
+        scenario: "AtomicTaskScenario",
+        suite: SuiteCfg,
+        track: TrackCfg,
+        config: Mapping[str, object],
+        *,
+        seed: int,
+        batch_size: int,
+    ) -> BenchmarkCase:
+        """Generate one frozen case and independent IK validity evidence."""
+
+    @abstractmethod
+    def build_invocation(
+        self,
+        scenario: "AtomicTaskScenario",
+        case: BenchmarkCase,
+        adapter: "PlannerAdapter",
+    ) -> ActionInvocation:
+        """Ground the case into one planner-independent action invocation."""
+
+    def object_id(self, case: BenchmarkCase) -> str | None:
+        """Return the manipulated object identifier when one exists."""
+        return case.object_id
+
+    def lift_segment_start(self, compiled: CompiledTrajectory) -> int | None:
+        """Return the first lift waypoint that should release object dynamics."""
+        return None
+
+    def articulation_effect_segment(self, case: BenchmarkCase) -> str | None:
+        """Return the segment whose live articulation displacement is scored."""
+        del case
+        return None
+
+    def initial_task_state(
+        self, scenario: "AtomicTaskScenario", case: BenchmarkCase
+    ) -> TaskState | None:
+        """Return an optional symbolic precondition for isolated action testing."""
+        del scenario, case
+        return None
+
+    def physical_evaluator(self) -> PhysicalStageEvaluator | None:
+        """Return a measured-stage evaluator, or use legacy task_result.
+
+        Returns:
+            An evaluator that receives no compiled planning state.
+        """
+        return None
+
+    def task_result(
+        self,
+        scenario: "AtomicTaskScenario",
+        case: BenchmarkCase,
+        compiled: CompiledTrajectory,
+        observation: ExecutionObservation,
+        motion_outcomes: tuple[CaseOutcome, ...],
+    ) -> tuple[torch.Tensor, str]:
+        """Return legacy task success until this provider supplies an evaluator."""
+        raise NotImplementedError(
+            "Provide physical_evaluator() or legacy task_result()."
+        )

scripts/benchmark/motion_generation/config.py

@@ -30,10 +30,12 @@
 
 from embodichain.utils import configclass
 
+from .contracts import EvaluationDomain
 from .models import AlgorithmRole
 
 __all__ = [
     "AtomicPoseRandomizationCfg",
+    "DomainCfg",
     "BENCHMARK_ROOT",
     "FreeSpaceTrackCfg",
     "PlannerSpecCfg",
@@ -133,6 +135,27 @@ class FreeSpaceTrackCfg:
     seeds: list[int] = [11]
 
 
+@configclass
+class DomainCfg:
+    """Explicit provenance for one track's existing case population.
+
+    This declaration labels cases; it does not generate perturbations or
+    establish that a population was held out during model development.
+    """
+
+    id: str = ""
+    version: str = ""
+    kind: str = ""
+
+    def to_identity(self) -> EvaluationDomain:
+        """Validate and snapshot domain provenance for a frozen case.
+
+        Returns:
+            Immutable domain identity, independent of planner configuration.
+        """
+        return EvaluationDomain(id=self.id, version=self.version, kind=self.kind)
+
+
 @configclass
 class TrackCfg:
     """One enabled benchmark track and its scenario provider."""
@@ -141,6 +164,7 @@ class TrackCfg:
     scenario: str = ""
     enabled: bool = True
     config: dict[str, Any] = {}
+    domain: DomainCfg | None = None
 
 
 @configclass
@@ -221,6 +245,8 @@ def validate_benchmark(self) -> None:
         if not self.enabled_tracks():
             raise ValueError("At least one track must be enabled.")
         for track in self.tracks:
+            if track.domain is not None:
+                track.domain.to_identity()
             if not track.id or not track.scenario:
                 raise ValueError("Every track must define a non-empty id and scenario.")
         if self.protocol.warmup_trials < 0:
@@ -277,7 +303,15 @@ def _resolve_tracks_and_free_space(
     else:
         if not isinstance(tracks_data, list):
             raise TypeError("tracks must be a list of track mappings.")
-        tracks = [TrackCfg(**item) for item in tracks_data]
+        tracks = []
+        for item in tracks_data:
+            values = dict(item)
+            domain = values.get("domain")
+            if domain is not None:
+                if not isinstance(domain, Mapping):
+                    raise TypeError("track.domain must be a mapping.")
+                values["domain"] = DomainCfg(**domain)
+            tracks.append(TrackCfg(**values))
         for track in tracks:
             if track.scenario == _FREE_SPACE_SCENARIO and track.config:
                 free_space_data = {**free_space_data, **dict(track.config)}