From ac2674451e70b767a3e9ad662c80941cefbf6cc1 Mon Sep 17 00:00:00 2001 From: skywhite1024 <129768272+skywhite1024@users.noreply.github.com> Date: Thu, 27 Aug 2026 14:39:32 +0800 Subject: [PATCH] feat(gen-sim): add task contracts and scene authoring boundary --- .../embodichain.gen_sim.task_engine.rst | 30 + docs/source/api_reference/index.rst | 1 + docs/source/api_reference/public_api.rst | 52 + .../scene_engine/core/scene_edit_plan.py | 36 +- embodichain/gen_sim/scene_engine/errors.py | 23 + .../gen_sim/scene_engine/pipeline/__init__.py | 24 +- .../gen_sim/scene_engine/pipeline/api.py | 351 +++++ .../editing/scene_edit_asset_preparation.py | 33 +- embodichain/gen_sim/task_engine/__init__.py | 85 ++ embodichain/gen_sim/task_engine/contracts.py | 410 ++++++ .../gen_sim/task_engine/interpretation.py | 1200 +++++++++++++++++ embodichain/gen_sim/task_engine/ontology.py | 292 ++++ tests/gen_sim/__init__.py | 4 + .../gen_sim/scene_engine/test_pipeline_api.py | 244 ++++ tests/gen_sim/task_engine/__init__.py | 19 + .../task_engine/test_interpretation.py | 96 ++ 16 files changed, 2896 insertions(+), 4 deletions(-) create mode 100644 docs/source/api_reference/embodichain/embodichain.gen_sim.task_engine.rst create mode 100644 embodichain/gen_sim/scene_engine/errors.py create mode 100644 embodichain/gen_sim/scene_engine/pipeline/api.py create mode 100644 embodichain/gen_sim/task_engine/__init__.py create mode 100644 embodichain/gen_sim/task_engine/contracts.py create mode 100644 embodichain/gen_sim/task_engine/interpretation.py create mode 100644 embodichain/gen_sim/task_engine/ontology.py create mode 100644 tests/gen_sim/scene_engine/test_pipeline_api.py create mode 100644 tests/gen_sim/task_engine/__init__.py create mode 100644 tests/gen_sim/task_engine/test_interpretation.py diff --git a/docs/source/api_reference/embodichain/embodichain.gen_sim.task_engine.rst b/docs/source/api_reference/embodichain/embodichain.gen_sim.task_engine.rst new file mode 100644 index 000000000..b62262a93 --- /dev/null +++ b/docs/source/api_reference/embodichain/embodichain.gen_sim.task_engine.rst @@ -0,0 +1,30 @@ +embodichain.gen_sim.task_engine +================================ + +Task Engine owns scene-independent task interpretation, normalized semantic +contracts, and the E1-E9 task ontology. These APIs do not perform scene +grounding or simulator execution. + +Package exports +--------------- + +.. automodule:: embodichain.gen_sim.task_engine + :members: + +Contracts +--------- + +.. automodule:: embodichain.gen_sim.task_engine.contracts + :members: + +Interpretation +-------------- + +.. automodule:: embodichain.gen_sim.task_engine.interpretation + :members: + +Ontology +-------- + +.. automodule:: embodichain.gen_sim.task_engine.ontology + :members: diff --git a/docs/source/api_reference/index.rst b/docs/source/api_reference/index.rst index cc9e16c5f..be454024d 100644 --- a/docs/source/api_reference/index.rst +++ b/docs/source/api_reference/index.rst @@ -80,3 +80,4 @@ documentation. CI runs this same checker after style checks and before tests. :maxdepth: 1 public_api + embodichain/embodichain.gen_sim.task_engine diff --git a/docs/source/api_reference/public_api.rst b/docs/source/api_reference/public_api.rst index 9b1f941b1..483faf157 100644 --- a/docs/source/api_reference/public_api.rst +++ b/docs/source/api_reference/public_api.rst @@ -238,6 +238,58 @@ embodichain.gen_sim.scene_engine.core.scene_edit_plan SceneEditOperation SceneEditPlan +embodichain.gen_sim.scene_engine.errors +--------------------------------------- + +.. currentmodule:: embodichain.gen_sim.scene_engine.errors + +Scene service failures preserve typed preparation and materialization errors +across the Task Engine boundary. + +.. autosummary:: + + SceneServiceError + +embodichain.gen_sim.scene_engine.pipeline +----------------------------------------- + +.. currentmodule:: embodichain.gen_sim.scene_engine.pipeline + +The public authoring boundary separates deterministic scene analysis from +side-effecting materialization for generated and edited scenes. + +.. autosummary:: + + SCENE_BLUEPRINT_SCHEMA + SCENE_EDIT_BLUEPRINT_SCHEMA + SceneBlueprintPackage + SceneEditBlueprintPackage + SceneMaterialization + analyze_edit + analyze_image + materialize_blueprint + materialize_edit + +embodichain.gen_sim.scene_engine.pipeline.api +--------------------------------------------- + +.. currentmodule:: embodichain.gen_sim.scene_engine.pipeline.api + +Versioned blueprint artifacts and analyze/materialize operations provide the +implementation-level Scene Engine authoring contract. + +.. autosummary:: + + SCENE_BLUEPRINT_SCHEMA + SCENE_EDIT_BLUEPRINT_SCHEMA + SceneBlueprintPackage + SceneEditBlueprintPackage + SceneMaterialization + analyze_edit + analyze_image + materialize_blueprint + materialize_edit + embodichain.gen_sim.scene_engine.pipeline.editing.scene_edit_asset_preparation ------------------------------------------------------------------------------- diff --git a/embodichain/gen_sim/scene_engine/core/scene_edit_plan.py b/embodichain/gen_sim/scene_engine/core/scene_edit_plan.py index 0f78b9699..7dffaae2c 100644 --- a/embodichain/gen_sim/scene_engine/core/scene_edit_plan.py +++ b/embodichain/gen_sim/scene_engine/core/scene_edit_plan.py @@ -35,7 +35,24 @@ @dataclass(frozen=True) class SceneEditOperation: - """One normalized edit operation produced from an LLM edit draft.""" + """Describe one normalized add, move, or delete operation. + + Attributes: + op: Operation kind. Add creates a new object, move repositions an + existing object, and delete removes an existing object. + object_id: Existing object ID for move/delete, or the generated ID for + an added object. + target_id: Optional existing scene object used as the spatial target. + relation: Spatial relation between the edited object and ``target_id``. + table_region: Optional named tabletop region. It is valid only when the + target is the table and the relation is ``"on"``. + category: Semantic category required for an added object. + name: Human-readable name required for an added object. + description: Generation prompt and semantic description required for + an added object. + orientation_state: Optional standing or lying intent for an added + object. Move operations may only preserve the existing state. + """ op: SceneEditOperationType object_id: str | None = None @@ -64,7 +81,22 @@ def to_dict(self) -> dict[str, object]: @dataclass class SceneEditPlan: - """Validated operations against one immutable pre-edit scene state.""" + """Validate edit operations against one pre-edit scene state. + + Construction validates every object reference and rejects conflicting + operations without mutating the supplied scene or scene graph. + + Attributes: + scene: Scene state that exists before the edit is applied. + scene_graph: Pre-edit support and spatial-relation graph. Its node IDs + must match the scene object IDs. + operations: Normalized operations in application order. + + Raises: + ValueError: If scene IDs are inconsistent, an operation has invalid + fields or references, edits conflict, or a deletion would orphan a + support descendant. + """ scene: Scene scene_graph: SceneGraph diff --git a/embodichain/gen_sim/scene_engine/errors.py b/embodichain/gen_sim/scene_engine/errors.py new file mode 100644 index 000000000..dd23d2aa9 --- /dev/null +++ b/embodichain/gen_sim/scene_engine/errors.py @@ -0,0 +1,23 @@ +# ---------------------------------------------------------------------------- +# 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. +# ---------------------------------------------------------------------------- + +from __future__ import annotations + +__all__ = ["SceneServiceError"] + + +class SceneServiceError(RuntimeError): + """A transient or remote Scene Engine service failure.""" diff --git a/embodichain/gen_sim/scene_engine/pipeline/__init__.py b/embodichain/gen_sim/scene_engine/pipeline/__init__.py index 015c41510..ecf448d22 100644 --- a/embodichain/gen_sim/scene_engine/pipeline/__init__.py +++ b/embodichain/gen_sim/scene_engine/pipeline/__init__.py @@ -16,4 +16,26 @@ from __future__ import annotations -__all__: list[str] = [] +from .api import ( + SCENE_BLUEPRINT_SCHEMA, + SCENE_EDIT_BLUEPRINT_SCHEMA, + SceneBlueprintPackage, + SceneEditBlueprintPackage, + SceneMaterialization, + analyze_edit, + analyze_image, + materialize_blueprint, + materialize_edit, +) + +__all__ = [ + "SCENE_BLUEPRINT_SCHEMA", + "SCENE_EDIT_BLUEPRINT_SCHEMA", + "SceneBlueprintPackage", + "SceneEditBlueprintPackage", + "SceneMaterialization", + "analyze_edit", + "analyze_image", + "materialize_blueprint", + "materialize_edit", +] diff --git a/embodichain/gen_sim/scene_engine/pipeline/api.py b/embodichain/gen_sim/scene_engine/pipeline/api.py new file mode 100644 index 000000000..6f219615c --- /dev/null +++ b/embodichain/gen_sim/scene_engine/pipeline/api.py @@ -0,0 +1,351 @@ +# ---------------------------------------------------------------------------- +# 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. +# ---------------------------------------------------------------------------- + +"""Auditable stage boundaries for Scene Engine generation and editing.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass +import hashlib +import json +from pathlib import Path +from typing import Any, Final + +from embodichain.gen_sim.scene_engine.clients.geometry_generation import ( + GeometryGenerationClient, +) +from embodichain.gen_sim.scene_engine.clients.image_generation import ( + ImageGenerationClient, +) +from embodichain.gen_sim.scene_engine.clients.image_segmentation import ( + ImageSegmentationClient, +) +from embodichain.gen_sim.scene_engine.core.scene import Scene +from embodichain.gen_sim.scene_engine.core.scene_edit_plan import SceneEditPlan +from embodichain.gen_sim.scene_engine.core.scene_graph import SceneGraph +from embodichain.gen_sim.scene_engine.llms.openai_compatible_client import ( + OpenAICompatibleVLM, +) +from embodichain.gen_sim.scene_engine.pipeline.editing.scene_edit_asset_preparation import ( + prepare_scene_edit_assets, +) +from embodichain.gen_sim.scene_engine.pipeline.editing.scene_edit_layout_generation import ( + edit_layout, +) +from embodichain.gen_sim.scene_engine.pipeline.editing.scene_edit_understanding import ( + understand_scene_edit, +) +from embodichain.gen_sim.scene_engine.pipeline.generation.scene_generation import ( + generate_scene_and_refine, +) +from embodichain.gen_sim.scene_engine.pipeline.generation.scene_understanding import ( + understand_scene, +) +from embodichain.gen_sim.scene_engine.pipeline.utils.scene_exporter import SceneExporter +from embodichain.gen_sim.scene_engine.pipeline.utils.scene_importer import ( + SceneExportImporter, +) +from embodichain.utils.logger import log_info + +__all__ = [ + "SCENE_BLUEPRINT_SCHEMA", + "SCENE_EDIT_BLUEPRINT_SCHEMA", + "SceneBlueprintPackage", + "SceneEditBlueprintPackage", + "SceneMaterialization", + "analyze_edit", + "analyze_image", + "materialize_blueprint", + "materialize_edit", +] + +SCENE_BLUEPRINT_SCHEMA: Final = "embodichain.scene-blueprint/v1" +SCENE_EDIT_BLUEPRINT_SCHEMA: Final = "embodichain.scene-edit-blueprint/v1" + + +@dataclass(frozen=True) +class SceneBlueprintPackage: + """In-process scene semantics plus their persisted audit document.""" + + blueprint_id: str + image_path: Path + output_root: Path + manifest_path: Path + scene: Scene + scene_graph: SceneGraph + + +@dataclass(frozen=True) +class SceneEditBlueprintPackage: + """Validated edit intent before added assets and layout are materialized.""" + + blueprint_id: str + edit_prompt: str + output_root: Path + manifest_path: Path + scene_edit_plan: SceneEditPlan + updated_scene_graph: SceneGraph + + +@dataclass(frozen=True) +class SceneMaterialization: + """One exported materialized scene revision.""" + + scene: Scene + scene_graph: SceneGraph + output_root: Path + scene_config_path: Path + + +def analyze_image( + image_path: str | Path, + output_root: str | Path, + *, + vlm_client: OpenAICompatibleVLM | None = None, + image_segmentation_client: ImageSegmentationClient | None = None, +) -> SceneBlueprintPackage: + """Understand an image and persist the pre-generation semantic blueprint.""" + resolved_image = Path(image_path).expanduser().resolve() + resolved_output = Path(output_root).expanduser().resolve() + resolved_output.mkdir(parents=True, exist_ok=True) + effective_vlm = vlm_client or OpenAICompatibleVLM.from_dotenv() + segmentation = image_segmentation_client or ImageSegmentationClient.from_dotenv() + owns_segmentation = image_segmentation_client is None + log_info("Starting Scene Understanding") + try: + segmentation.check_health() + scene, scene_graph = understand_scene( + scene=Scene(), + image_path=resolved_image, + output_root=resolved_output, + vlm_client=effective_vlm, + image_segmentation_client=segmentation, + ) + finally: + if owns_segmentation: + segmentation.close() + log_info("Completed Scene Understanding") + + payload = { + "schema_version": SCENE_BLUEPRINT_SCHEMA, + "image_path": resolved_image.as_posix(), + "scene": scene.to_dict(), + "scene_graph": scene_graph.to_dict(), + "artifacts": _artifact_records(resolved_output / "scene_understanding"), + } + blueprint_id = _canonical_hash(payload) + document = {**payload, "blueprint_id": blueprint_id} + manifest_path = resolved_output / "scene_blueprint.json" + _write_json(manifest_path, document) + return SceneBlueprintPackage( + blueprint_id=blueprint_id, + image_path=resolved_image, + output_root=resolved_output, + manifest_path=manifest_path, + scene=scene, + scene_graph=scene_graph, + ) + + +def materialize_blueprint( + blueprint: SceneBlueprintPackage, + *, + vlm_client: OpenAICompatibleVLM | None = None, + geometry_generation_client: GeometryGenerationClient | None = None, + seed: int | None = None, +) -> SceneMaterialization: + """Generate assets and layout for one image-derived blueprint.""" + scene = deepcopy(blueprint.scene) + scene_graph = deepcopy(blueprint.scene_graph) + effective_vlm = vlm_client or OpenAICompatibleVLM.from_dotenv() + geometry = geometry_generation_client or GeometryGenerationClient.from_dotenv() + owns_geometry = geometry_generation_client is None + log_info("Starting Objects + Coarse Layout Generation") + try: + geometry.check_health() + scene = generate_scene_and_refine( + image_path=blueprint.image_path, + output_root=blueprint.output_root, + scene=scene, + scene_graph=scene_graph, + geometry_generation_client=geometry, + vlm_client=effective_vlm, + seed=seed, + ) + finally: + if owns_geometry: + geometry.close() + log_info("Completed Objects + Coarse Layout Generation") + return _export_materialization( + scene=scene, + scene_graph=scene_graph, + output_root=blueprint.output_root, + ) + + +def analyze_edit( + *, + output_root: str | Path, + edit_prompt: str, + vlm_client: OpenAICompatibleVLM | None = None, +) -> SceneEditBlueprintPackage: + """Interpret and persist one edit against an already generated scene.""" + resolved_output = Path(output_root).expanduser().resolve() + normalized_prompt = str(edit_prompt).strip() + if not normalized_prompt: + raise ValueError("Edit prompt must not be empty.") + scene, scene_graph = SceneExportImporter( + output_root=resolved_output + ).import_scene_and_graph() + effective_vlm = vlm_client or OpenAICompatibleVLM.from_dotenv() + log_info("Starting Edit Understanding") + scene_edit_plan, updated_scene_graph = understand_scene_edit( + scene=scene, + scene_graph=scene_graph, + edit_prompt=normalized_prompt, + vlm_client=effective_vlm, + ) + log_info("Completed Edit Understanding") + payload = { + "schema_version": SCENE_EDIT_BLUEPRINT_SCHEMA, + "edit_prompt": normalized_prompt, + "scene_edit_plan": scene_edit_plan.to_dict(), + "updated_scene_graph": updated_scene_graph.to_dict(), + } + blueprint_id = _canonical_hash(payload) + manifest_path = resolved_output / "scene_edit" / "scene_edit_blueprint.json" + _write_json(manifest_path, {**payload, "blueprint_id": blueprint_id}) + return SceneEditBlueprintPackage( + blueprint_id=blueprint_id, + edit_prompt=normalized_prompt, + output_root=resolved_output, + manifest_path=manifest_path, + scene_edit_plan=scene_edit_plan, + updated_scene_graph=updated_scene_graph, + ) + + +def materialize_edit( + blueprint: SceneEditBlueprintPackage, + *, + vlm_client: OpenAICompatibleVLM | None = None, + image_generation_client: ImageGenerationClient | None = None, + geometry_generation_client: GeometryGenerationClient | None = None, + image_segmentation_client: ImageSegmentationClient | None = None, + seed: int | None = None, +) -> SceneMaterialization: + """Generate added assets, apply layout edits, and export the new revision.""" + scene_edit_plan = deepcopy(blueprint.scene_edit_plan) + updated_scene_graph = deepcopy(blueprint.updated_scene_graph) + effective_vlm = vlm_client or OpenAICompatibleVLM.from_dotenv() + image_generation = image_generation_client or ImageGenerationClient.from_dotenv() + geometry = geometry_generation_client or GeometryGenerationClient.from_dotenv() + segmentation = image_segmentation_client or ImageSegmentationClient.from_dotenv() + owned_clients = ( + (image_generation, image_generation_client is None), + (geometry, geometry_generation_client is None), + (segmentation, image_segmentation_client is None), + ) + log_info("Starting Objects Preparation") + try: + for client, _ in owned_clients: + client.check_health() + added_assets = prepare_scene_edit_assets( + scene_edit_plan=scene_edit_plan, + output_root=blueprint.output_root, + image_generation_client=image_generation, + geometry_generation_client=geometry, + image_segmentation_client=segmentation, + vlm_client=effective_vlm, + seed=seed, + ) + finally: + for client, owned in owned_clients: + if owned: + client.close() + log_info("Completed Objects Preparation") + log_info("Starting Layout Generation") + scene = edit_layout( + scene=scene_edit_plan.scene, + scene_edit_plan=scene_edit_plan, + updated_scene_graph=updated_scene_graph, + added_assets=added_assets, + output_root=blueprint.output_root, + ) + log_info("Completed Layout Generation") + return _export_materialization( + scene=scene, + scene_graph=updated_scene_graph, + output_root=blueprint.output_root, + ) + + +def _export_materialization( + *, + scene: Scene, + scene_graph: SceneGraph, + output_root: Path, +) -> SceneMaterialization: + log_info("Starting Scene Export") + scene_config_path = SceneExporter( + scene=scene, + scene_graph=scene_graph, + output_root=output_root, + ).export() + log_info("Completed Scene Export") + return SceneMaterialization( + scene=scene, + scene_graph=scene_graph, + output_root=output_root, + scene_config_path=scene_config_path, + ) + + +def _artifact_records(root: Path) -> list[dict[str, Any]]: + if not root.is_dir(): + return [] + records = [] + for path in sorted(item for item in root.rglob("*") if item.is_file()): + records.append( + { + "path": path.resolve().as_posix(), + "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), + "size": path.stat().st_size, + } + ) + return records + + +def _canonical_hash(value: Any) -> str: + payload = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def _write_json(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text( + json.dumps(value, indent=2, ensure_ascii=False, allow_nan=False) + "\n", + encoding="utf-8", + ) + temporary.replace(path) diff --git a/embodichain/gen_sim/scene_engine/pipeline/editing/scene_edit_asset_preparation.py b/embodichain/gen_sim/scene_engine/pipeline/editing/scene_edit_asset_preparation.py index 220a91014..616d647e3 100644 --- a/embodichain/gen_sim/scene_engine/pipeline/editing/scene_edit_asset_preparation.py +++ b/embodichain/gen_sim/scene_engine/pipeline/editing/scene_edit_asset_preparation.py @@ -72,7 +72,38 @@ def prepare_scene_edit_assets( image_segmentation_client: ImageSegmentationClient, vlm_client: OpenAICompatibleVLM | None = None, ) -> list[SceneObject]: - """Prepare and return SimReady assets required by add operations.""" + """Generate canonical SimReady assets for a scene edit's add operations. + + Move-only and delete-only plans return immediately without modifying an + existing asset-preparation directory. For add operations, the function + generates and segments one image per object, creates coarse geometry, + processes it into SimReady geometry, and resets the returned objects to + identity edit-time poses. + + Args: + scene_edit_plan: Validated edit plan whose add operations define the + objects to generate. + output_root: Scene Engine output root. Intermediate artifacts are + written below ``scene_editing/asset_preparation``. + image_generation_client: Client used to render object images from the + operation descriptions. + geometry_generation_client: Client used to create coarse GLB geometry + from each generated image and mask. + image_segmentation_client: Client used to isolate the generated object + in each image. + vlm_client: Optional VLM used by SimReady processing to estimate asset + scale and orientation. + + Returns: + Added ``SceneObject`` assets in edit-plan order, or an empty list when + the plan contains no add operations. + + Raises: + ValueError: If add metadata or generated image, mask, and geometry + mappings are incomplete or inconsistent. + FileNotFoundError: If geometry generation does not produce an expected + GLB file. + """ # Prepare descriptions for all newly added objects. added_asset_descriptions = _collect_added_asset_descriptions(scene_edit_plan) # Skip asset generation when the edit plan only moves or deletes existing objects. diff --git a/embodichain/gen_sim/task_engine/__init__.py b/embodichain/gen_sim/task_engine/__init__.py new file mode 100644 index 000000000..bb142aaa1 --- /dev/null +++ b/embodichain/gen_sim/task_engine/__init__.py @@ -0,0 +1,85 @@ +# ---------------------------------------------------------------------------- +# 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. +# ---------------------------------------------------------------------------- + +"""Scene-independent task interpretation and protocol ownership.""" + +from __future__ import annotations + +from .contracts import ( + SCENE_REQUEST_SCHEMA, + SUCCESS_SPEC_SCHEMA, + TASK_CANDIDATE_SET_SCHEMA, + TASK_DRAFT_SCHEMA, + SceneRequest, + SuccessSpec, + TaskCandidate, + TaskCandidateSet, + TaskDraft, + canonical_hash, + validate_scene_request, + validate_success_spec, + validate_task_candidate, + validate_task_candidate_set, + validate_task_draft, +) +from .interpretation import ( + INSTRUCTION_INTENT_SCHEMA, + InstructionCaller, + InstructionDraftResult, + InstructionIntent, + interpret_instruction_draft, + validate_instruction_intent, +) +from .ontology import ( + RELATIONS, + TASK_CONTRACTS, + TERMINAL_BEHAVIORS, + TRANSPORT_DIRECTIONS, + TaskContract, + task_contract, + task_success_type, +) + +__all__ = [ + "INSTRUCTION_INTENT_SCHEMA", + "InstructionCaller", + "InstructionDraftResult", + "InstructionIntent", + "RELATIONS", + "SCENE_REQUEST_SCHEMA", + "SUCCESS_SPEC_SCHEMA", + "SceneRequest", + "SuccessSpec", + "TASK_CANDIDATE_SET_SCHEMA", + "TASK_CONTRACTS", + "TASK_DRAFT_SCHEMA", + "TERMINAL_BEHAVIORS", + "TRANSPORT_DIRECTIONS", + "TaskCandidate", + "TaskCandidateSet", + "TaskContract", + "TaskDraft", + "canonical_hash", + "interpret_instruction_draft", + "task_contract", + "task_success_type", + "validate_instruction_intent", + "validate_scene_request", + "validate_success_spec", + "validate_task_candidate", + "validate_task_candidate_set", + "validate_task_draft", +] diff --git a/embodichain/gen_sim/task_engine/contracts.py b/embodichain/gen_sim/task_engine/contracts.py new file mode 100644 index 000000000..237423e55 --- /dev/null +++ b/embodichain/gen_sim/task_engine/contracts.py @@ -0,0 +1,410 @@ +# ---------------------------------------------------------------------------- +# 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. +# ---------------------------------------------------------------------------- + +"""Strict, JSON-safe public contracts owned by Task Engine.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from copy import deepcopy +import hashlib +import json +from typing import Any, TypeAlias + +from .interpretation import validate_instruction_intent +from .ontology import TASK_CONTRACTS, task_success_type + +__all__ = [ + "SCENE_REQUEST_SCHEMA", + "SUCCESS_SPEC_SCHEMA", + "TASK_CANDIDATE_SET_SCHEMA", + "TASK_DRAFT_SCHEMA", + "SceneRequest", + "SuccessSpec", + "TaskCandidate", + "TaskCandidateSet", + "TaskDraft", + "canonical_hash", + "validate_scene_request", + "validate_success_spec", + "validate_task_candidate", + "validate_task_candidate_set", + "validate_task_draft", +] + +TASK_DRAFT_SCHEMA = "action_engine_task_draft_v1" +SCENE_REQUEST_SCHEMA = "action_engine_scene_request_v1" +SUCCESS_SPEC_SCHEMA = "action_engine_success_spec_v1" +TASK_CANDIDATE_SET_SCHEMA = "action_engine_task_candidate_set_v1" + +TaskDraft: TypeAlias = dict[str, Any] +SceneRequest: TypeAlias = dict[str, Any] +SuccessSpec: TypeAlias = dict[str, Any] +TaskCandidate: TypeAlias = dict[str, Any] +TaskCandidateSet: TypeAlias = dict[str, Any] + +_SUCCESS_TYPES = frozenset( + {contract.success_type for contract in TASK_CONTRACTS.values()} | {"semantic_goal"} +) +_DRAFT_KEYS = frozenset({"schema_version", "task_id", "instruction", "steps"}) +_SCENE_REQUEST_KEYS = frozenset({"schema_version", "task_id", "references"}) +_REFERENCE_KEYS = frozenset( + { + "reference_id", + "step_id", + "role", + "reference", + "quantifier", + "count", + "source_structure", + "affordances", + "initial_state", + "attributes", + } +) +_SUCCESS_KEYS = frozenset({"schema_version", "task_id", "op", "terms"}) +_SUCCESS_TERM_KEYS = frozenset({"step_id", "type"}) +_CANDIDATE_KEYS = frozenset( + { + "candidate_id", + "draft", + "scene_request", + "success_spec", + "semantic_hash", + "vote_count", + "attempts", + "normalizations", + } +) +_CANDIDATE_SET_KEYS = frozenset( + { + "schema_version", + "task_id", + "instruction", + "candidates", + "requested_candidate_count", + "valid_response_count", + "errors", + } +) + + +def canonical_hash(value: Any) -> str: + """Return the stable SHA-256 of one JSON-safe protocol value.""" + payload = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def validate_task_draft(value: Mapping[str, Any]) -> TaskDraft: + result = _mapping(value, "TaskDraft") + _keys(result, _DRAFT_KEYS, "TaskDraft") + _schema(result, TASK_DRAFT_SCHEMA, "TaskDraft") + result["task_id"] = _nonempty(result.get("task_id"), "TaskDraft.task_id") + result["instruction"] = _nonempty( + result.get("instruction"), "TaskDraft.instruction" + ) + intent = validate_instruction_intent({"steps": result.get("steps")}) + result["steps"] = intent["steps"] + return result + + +def validate_scene_request(value: Mapping[str, Any]) -> SceneRequest: + result = _mapping(value, "SceneRequest") + _keys(result, _SCENE_REQUEST_KEYS, "SceneRequest") + _schema(result, SCENE_REQUEST_SCHEMA, "SceneRequest") + task_id = _nonempty(result.get("task_id"), "SceneRequest.task_id") + references: list[dict[str, Any]] = [] + for index, raw in enumerate( + _sequence(result.get("references"), "SceneRequest.references") + ): + context = f"SceneRequest.references[{index}]" + reference = _mapping(raw, context) + _keys(reference, _REFERENCE_KEYS, context) + for key in ("reference_id", "step_id", "role", "reference", "source_structure"): + reference[key] = _nonempty(reference.get(key), f"{context}.{key}") + reference["role"] = _enum( + reference["role"], {"object", "target"}, f"{context}.role" + ) + reference["quantifier"] = _enum( + reference.get("quantifier"), + {"one", "all", "count"}, + f"{context}.quantifier", + ) + reference["count"] = _integer( + reference.get("count"), f"{context}.count", minimum=0 + ) + if reference["quantifier"] in {"one", "all"} and reference["count"] != 0: + raise ValueError( + f"{context} quantifier={reference['quantifier']} requires count=0." + ) + if reference["quantifier"] == "count" and reference["count"] < 1: + raise ValueError(f"{context} quantifier=count requires count>=1.") + reference["affordances"] = _strings( + reference.get("affordances"), f"{context}.affordances" + ) + reference["initial_state"] = _mapping( + reference.get("initial_state"), f"{context}.initial_state" + ) + reference["attributes"] = _mapping( + reference.get("attributes"), f"{context}.attributes" + ) + references.append(reference) + _unique([item["reference_id"] for item in references], "SceneRequest reference IDs") + result["task_id"] = task_id + result["references"] = references + _json_safe(result, "SceneRequest") + return result + + +def validate_success_spec( + value: Mapping[str, Any], + *, + draft: Mapping[str, Any] | None = None, +) -> SuccessSpec: + result = _mapping(value, "SuccessSpec") + _keys(result, _SUCCESS_KEYS, "SuccessSpec") + _schema(result, SUCCESS_SPEC_SCHEMA, "SuccessSpec") + task_id = _nonempty(result.get("task_id"), "SuccessSpec.task_id") + if result.get("op") != "all": + raise ValueError("SuccessSpec.op must be 'all'.") + terms: list[dict[str, str]] = [] + for index, raw in enumerate(_sequence(result.get("terms"), "SuccessSpec.terms")): + context = f"SuccessSpec.terms[{index}]" + term = _mapping(raw, context) + _keys(term, _SUCCESS_TERM_KEYS, context) + terms.append( + { + "step_id": _nonempty(term.get("step_id"), f"{context}.step_id"), + "type": _enum(term.get("type"), set(_SUCCESS_TYPES), f"{context}.type"), + } + ) + if not terms: + raise ValueError("SuccessSpec.terms must not be empty.") + _unique([term["step_id"] for term in terms], "SuccessSpec step IDs") + if draft is not None: + normalized_draft = validate_task_draft(draft) + if normalized_draft["task_id"] != task_id: + raise ValueError("SuccessSpec.task_id must match TaskDraft.task_id.") + expected = [ + { + "step_id": step["id"], + "type": task_success_type(step["task_type"], step), + } + for step in normalized_draft["steps"] + ] + if terms != expected: + raise ValueError( + "SuccessSpec terms must be ordered, complete, and derived from " + "task_success_type." + ) + result["task_id"] = task_id + result["terms"] = terms + return result + + +def validate_task_candidate(value: Mapping[str, Any]) -> TaskCandidate: + result = _mapping(value, "TaskCandidate") + _keys(result, _CANDIDATE_KEYS, "TaskCandidate") + result["candidate_id"] = _nonempty( + result.get("candidate_id"), "TaskCandidate.candidate_id" + ) + result["draft"] = validate_task_draft(result.get("draft")) + result["scene_request"] = validate_scene_request(result.get("scene_request")) + result["success_spec"] = validate_success_spec( + result.get("success_spec"), draft=result["draft"] + ) + for name in ("scene_request", "success_spec"): + if result[name]["task_id"] != result["draft"]["task_id"]: + raise ValueError(f"TaskCandidate {name}.task_id must match its draft.") + from .agent import derive_scene_request + + if result["scene_request"] != derive_scene_request(result["draft"]): + raise ValueError( + "TaskCandidate.scene_request must be derived exactly from its draft." + ) + result["semantic_hash"] = _digest( + result.get("semantic_hash"), "TaskCandidate.semantic_hash" + ) + if result["semantic_hash"] != canonical_hash(result["draft"]["steps"]): + raise ValueError( + "TaskCandidate.semantic_hash does not match its canonical steps." + ) + result["vote_count"] = _integer( + result.get("vote_count"), "TaskCandidate.vote_count", minimum=1 + ) + result["attempts"] = _integer( + result.get("attempts"), "TaskCandidate.attempts", minimum=1, maximum=2 + ) + result["normalizations"] = _mapping_sequence( + result.get("normalizations"), "TaskCandidate.normalizations" + ) + return result + + +def validate_task_candidate_set(value: Mapping[str, Any]) -> TaskCandidateSet: + result = _mapping(value, "TaskCandidateSet") + _keys(result, _CANDIDATE_SET_KEYS, "TaskCandidateSet") + _schema(result, TASK_CANDIDATE_SET_SCHEMA, "TaskCandidateSet") + task_id = _nonempty(result.get("task_id"), "TaskCandidateSet.task_id") + instruction = _nonempty(result.get("instruction"), "TaskCandidateSet.instruction") + requested = _integer( + result.get("requested_candidate_count"), + "TaskCandidateSet.requested_candidate_count", + minimum=1, + ) + valid = _integer( + result.get("valid_response_count"), + "TaskCandidateSet.valid_response_count", + minimum=1, + maximum=requested, + ) + candidates = [ + validate_task_candidate(item) + for item in _sequence(result.get("candidates"), "TaskCandidateSet.candidates") + ] + if not candidates: + raise ValueError("TaskCandidateSet.candidates must not be empty.") + _unique([item["candidate_id"] for item in candidates], "TaskCandidate IDs") + _unique( + [item["semantic_hash"] for item in candidates], "TaskCandidate semantic hashes" + ) + if sum(item["vote_count"] for item in candidates) != valid: + raise ValueError( + "TaskCandidate vote_count values must sum to valid_response_count." + ) + for candidate in candidates: + if ( + candidate["draft"]["task_id"] != task_id + or candidate["draft"]["instruction"] != instruction + ): + raise ValueError("Every TaskCandidate draft must match its candidate set.") + errors = _strings(result.get("errors"), "TaskCandidateSet.errors", allow_empty=True) + if valid + len(errors) != requested: + raise ValueError( + "Valid responses plus errors must equal requested_candidate_count." + ) + result.update( + { + "task_id": task_id, + "instruction": instruction, + "requested_candidate_count": requested, + "valid_response_count": valid, + "candidates": candidates, + "errors": errors, + } + ) + return result + + +def _mapping(value: Any, context: str) -> dict[str, Any]: + if not isinstance(value, Mapping): + raise ValueError(f"{context} must be a mapping.") + return deepcopy(dict(value)) + + +def _sequence(value: Any, context: str) -> list[Any]: + if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): + raise ValueError(f"{context} must be a list.") + return list(value) + + +def _keys( + value: Mapping[str, Any], expected: set[str] | frozenset[str], context: str +) -> None: + if set(value) != set(expected): + raise ValueError( + f"{context} requires exactly fields {sorted(expected)}; received {sorted(value)}." + ) + + +def _schema(value: Mapping[str, Any], expected: str, context: str) -> None: + if value.get("schema_version") != expected: + raise ValueError(f"{context}.schema_version must be {expected!r}.") + + +def _string(value: Any, context: str) -> str: + if not isinstance(value, str): + raise ValueError(f"{context} must be a string.") + return value.strip() + + +def _nonempty(value: Any, context: str) -> str: + result = _string(value, context) + if not result: + raise ValueError(f"{context} must not be empty.") + return result + + +def _enum(value: Any, choices: set[str], context: str) -> str: + result = _string(value, context) + if result not in choices: + raise ValueError(f"{context} must be one of {sorted(choices)}.") + return result + + +def _integer( + value: Any, context: str, *, minimum: int, maximum: int | None = None +) -> int: + if ( + not isinstance(value, int) + or isinstance(value, bool) + or value < minimum + or (maximum is not None and value > maximum) + ): + raise ValueError(f"{context} must be an integer in the allowed range.") + return value + + +def _strings(value: Any, context: str, *, allow_empty: bool = False) -> list[str]: + result = [_string(item, context) for item in _sequence(value, context)] + if not allow_empty and any(not item for item in result): + raise ValueError(f"{context} values must not be empty.") + if len(result) != len(set(result)): + raise ValueError(f"{context} values must be unique.") + return result + + +def _mapping_sequence(value: Any, context: str) -> list[dict[str, Any]]: + result = [_mapping(item, context) for item in _sequence(value, context)] + _json_safe(result, context) + return result + + +def _digest(value: Any, context: str) -> str: + result = _string(value, context) + if len(result) != 64 or any( + character not in "0123456789abcdef" for character in result + ): + raise ValueError(f"{context} must be a lowercase SHA-256 digest.") + return result + + +def _unique(values: Sequence[str], context: str) -> None: + if len(values) != len(set(values)): + raise ValueError(f"{context} must be unique.") + + +def _json_safe(value: Any, context: str) -> None: + try: + json.dumps(value, allow_nan=False) + except (TypeError, ValueError) as error: + raise ValueError(f"{context} must be finite and JSON serializable.") from error diff --git a/embodichain/gen_sim/task_engine/interpretation.py b/embodichain/gen_sim/task_engine/interpretation.py new file mode 100644 index 000000000..f1d5484ab --- /dev/null +++ b/embodichain/gen_sim/task_engine/interpretation.py @@ -0,0 +1,1200 @@ +# ---------------------------------------------------------------------------- +# 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. +# ---------------------------------------------------------------------------- + +"""Scene-independent structured interpretation for Task Engine.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping, Sequence +from copy import deepcopy +from dataclasses import dataclass +import json +import os +from pathlib import Path +import re +from time import perf_counter +from typing import Any, TypeAlias + +from .ontology import ( + RELATIONS, + TASK_CONTRACTS, + TERMINAL_BEHAVIORS, + TRANSPORT_DIRECTIONS, +) + +__all__ = [ + "INSTRUCTION_INTENT_SCHEMA", + "InstructionDraftResult", + "InstructionIntent", + "InstructionCaller", + "interpret_instruction_draft", + "validate_instruction_intent", +] + +InstructionCaller = Callable[..., Mapping[str, Any]] +InstructionIntent: TypeAlias = dict[str, Any] +TASK_TYPES = frozenset(TASK_CONTRACTS) + +_RELATIONS = RELATIONS +_ARMS = frozenset({"none", "auto", "left_arm", "right_arm"}) +_ORIENTATIONS = frozenset({"none", "preserve", "upright"}) +_TARGET_STATES = frozenset({"none", "open", "closed", "activated"}) +_LAYOUTS = frozenset({"none", "line"}) +_AXES = frozenset({"none", "world_x", "world_y"}) +_DIRECTIONS = TRANSPORT_DIRECTIONS +_TERMINAL_BEHAVIORS = TERMINAL_BEHAVIORS +_SELECTOR_KINDS = frozenset({"none", "scene_ref", "step_result"}) +_QUANTIFIERS = frozenset({"one", "all", "count"}) +_STEP_KEYS = frozenset( + { + "id", + "task_type", + "object", + "target", + "relation", + "required_arm", + "transfer_arm", + "receive_arm", + "orientation_goal", + "target_state", + "target_setting", + "layout", + "axis", + "direction", + "terminal_behavior", + "depends_on", + } +) +_INTENT_TASK_FIELD_REGISTRY = { + task_type: contract.applicable_intent_fields + for task_type, contract in TASK_CONTRACTS.items() +} +_INTENT_FIELD_DEFAULTS: dict[str, Any] = { + "target": None, + "relation": "none", + "required_arm": "none", + "transfer_arm": "none", + "receive_arm": "none", + "orientation_goal": "none", + "target_state": "none", + "target_setting": 0, + "layout": "none", + "axis": "none", + "direction": "none", + "terminal_behavior": "none", +} +_SELECTOR_KEYS = frozenset( + { + "kind", + "step_id", + "reference", + "quantifier", + "count", + } +) +_FORBIDDEN_FIELDS = frozenset( + { + "atomic_action", + "atomic_actions", + "atomicaction", + "coordinates", + "bbox", + "bboxes", + "grasp_pose", + "keypoint", + "keypoints", + "joint_positions", + "joints", + "pose", + "position", + "qpos", + "rotation", + "target_pose", + "translation", + "trajectory", + "waypoints", + } +) +# MiMo's OpenAI-compatible endpoint can spend the whole completion budget in +# hidden reasoning when the request leaves thinking enabled. A sparse final +# JSON object then looks like a schema failure to the deterministic verifier. +# Keep the budget bounded and turn reasoning off for the text interpretation +# call; the parser must return an auditable object rather than a thought trace. +_MIMO_MAX_COMPLETION_TOKENS = 4096 +_GEN_SIM_DIR = Path(__file__).resolve().parents[1] +_GEN_SIM_ENV_PATH = _GEN_SIM_DIR / ".env" +_GEN_CONFIG_PATH = _GEN_SIM_DIR / "simready_pipeline" / "configs" / "gen_config.json" + + +class _MissingRequiredTargetError(ValueError): + """Identify a validation failure that receives targeted repair guidance.""" + + +class _MissingRequiredObjectError(ValueError): + """Identify a missing manipulated-object selector for targeted repair.""" + + +@dataclass(frozen=True) +class InstructionDraftResult: + """One validated, scene-independent interpretation and its audit metadata.""" + + intent: InstructionIntent + model: str + attempts: int + latency_seconds: float + normalizations: tuple[dict[str, Any], ...] + + +# Object semantics remain open natural-language references until the dedicated +# scene-grounding phase resolves them. All other values are strict protocol +# enums; non-canonical model output is repaired by the model, never guessed by +# a local language alias table. + +_SELECTOR_SCHEMA = { + "type": "object", + "additionalProperties": False, + "required": sorted(_SELECTOR_KEYS), + "properties": { + "kind": {"type": "string", "enum": sorted(_SELECTOR_KINDS)}, + "step_id": {"type": "string"}, + "reference": {"type": "string"}, + "quantifier": {"type": "string", "enum": sorted(_QUANTIFIERS)}, + "count": {"type": "integer", "minimum": 0}, + }, +} + +_INTENT_OUTPUT_SCHEMA = { + "title": "ActionEngineInstructionIntent", + "type": "object", + "additionalProperties": False, + "required": ["steps"], + "properties": { + "steps": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "additionalProperties": False, + "required": sorted(_STEP_KEYS), + "properties": { + "id": {"type": "string"}, + "task_type": {"type": "string", "enum": sorted(TASK_TYPES)}, + "object": _SELECTOR_SCHEMA, + "target": _SELECTOR_SCHEMA, + "relation": {"type": "string", "enum": sorted(_RELATIONS)}, + "required_arm": {"type": "string", "enum": sorted(_ARMS)}, + "transfer_arm": {"type": "string", "enum": sorted(_ARMS)}, + "receive_arm": {"type": "string", "enum": sorted(_ARMS)}, + "orientation_goal": { + "type": "string", + "enum": sorted(_ORIENTATIONS), + }, + "target_state": { + "type": "string", + "enum": sorted(_TARGET_STATES), + }, + "target_setting": {"type": "integer"}, + "layout": {"type": "string", "enum": sorted(_LAYOUTS)}, + "axis": {"type": "string", "enum": sorted(_AXES)}, + "direction": { + "type": "string", + "enum": sorted(_DIRECTIONS), + }, + "terminal_behavior": { + "type": "string", + "enum": sorted(_TERMINAL_BEHAVIORS), + }, + "depends_on": { + "type": "array", + "items": {"type": "string"}, + }, + }, + }, + } + }, +} + +# Keep a read-only-by-convention public copy for callers that need to configure +# a structured client. The schema is an input contract, not a persisted task +# graph; ``validate_instruction_intent`` remains the authoritative verifier. +INSTRUCTION_INTENT_SCHEMA = deepcopy(_INTENT_OUTPUT_SCHEMA) + + +def interpret_instruction_draft( + instruction: str, + *, + model: str | None = None, + caller: InstructionCaller | None = None, +) -> InstructionDraftResult: + """Interpret one instruction without reading or grounding a scene.""" + instruction_text = str(instruction).strip() + if not instruction_text: + raise ValueError("instruction must be non-empty.") + prompt = _instruction_prompt(instruction_text) + invoke = caller or _default_instruction_caller + # An injected caller owns its transport and does not need provider config. + selected_model = model if caller is not None else _instruction_model(model) + if caller is None and selected_model is None: + raise ValueError( + "A text LLM model is required through --llm-model, " + "ACTION_ENGINE_LLM_MODEL, or OPENAI_MODEL." + ) + started = perf_counter() + first_error: Exception | None = None + for attempt in range(2): + current_prompt = prompt + if first_error is not None: + current_prompt += ( + "\n\nREPAIR OVERRIDE: the previous JSON was invalid. Return a corrected " + "JSON object only; do not repeat the sparse response. Every step " + "must contain all 16 step keys and every selector all 5 selector " + "keys. Keep semantic fields explicit: E4 requires transfer_arm " + "and receive_arm, and E1/E3 require target plus relation (unless " + "E1 layout=line). Use canonical defaults only for fields that do " + "not apply. Validation error: " + f"{first_error}\n" + "Copy this complete shape before filling values (shape only; do " + "not copy its values or step count):\n" + f"{json.dumps(_instruction_shape_example(), ensure_ascii=False, sort_keys=True)}\n" + "Selector kind rules:\n" + f"{_instruction_selector_rules()}" + f"{_instruction_repair_guidance(first_error)}" + ) + try: + response = invoke( + prompt=current_prompt, + schema=deepcopy(INSTRUCTION_INTENT_SCHEMA), + model=selected_model, + ) + normalized, normalizations = _normalize_instruction_intent_fields( + _coerce_instruction_response(response) + ) + intent = validate_instruction_intent(normalized) + return InstructionDraftResult( + intent=intent, + model=selected_model or "injected_caller", + attempts=attempt + 1, + latency_seconds=perf_counter() - started, + normalizations=tuple(normalizations), + ) + except (TypeError, ValueError) as error: + if attempt: + raise ValueError( + "Instruction intent failed validation after one repair: " f"{error}" + ) from error + first_error = error + raise AssertionError("unreachable") + + +def _normalize_instruction_intent_fields( + value: Mapping[str, Any], +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + """Canonicalize defaults and uniquely constrained cross-step continuity. + + The strict public validator deliberately remains unchanged. This pass is + confined to the LLM boundary, where weak JSON-mode providers sometimes + copy a meaningful value into an inapplicable slot such as E4.required_arm. + Required scene facts and ambiguous arm assignments are never inferred here + and still fail closed. + """ + result = deepcopy(dict(value)) + raw_steps = result.get("steps") + if not isinstance(raw_steps, list): + return result, [] + changes: list[dict[str, Any]] = [] + for index, raw_step in enumerate(raw_steps): + if not isinstance(raw_step, dict) or set(raw_step) != _STEP_KEYS: + continue + task_type = raw_step.get("task_type") + applicable = _INTENT_TASK_FIELD_REGISTRY.get(task_type) + if applicable is None: + continue + for field, configured_default in _INTENT_FIELD_DEFAULTS.items(): + field_applies = field in applicable + if task_type == "E1" and field in {"target", "relation"}: + field_applies = raw_step.get("layout") != "line" + if task_type == "E1" and field == "axis": + field_applies = raw_step.get("layout") == "line" + if field_applies: + continue + default = ( + _empty_selector() + if field == "target" and configured_default is None + else deepcopy(configured_default) + ) + if raw_step[field] == default: + continue + previous = deepcopy(raw_step[field]) + raw_step[field] = default + changes.append( + { + "path": f"steps[{index}].{field}", + "from": previous, + "to": deepcopy(default), + "reason": f"inapplicable_for_{task_type}", + } + ) + target = raw_step.get("target") + if ( + task_type == "E5" + and isinstance(target, Mapping) + and target.get("kind") == "none" + and raw_step.get("relation") == "none" + and raw_step.get("direction") == "none" + and raw_step.get("terminal_behavior") == "hold" + ): + raw_step["direction"] = "up" + changes.append( + { + "path": f"steps[{index}].direction", + "from": "none", + "to": "up", + "reason": "e5_hold_defaults_to_lift", + } + ) + if task_type == "E4" and raw_step.get("terminal_behavior") == "none": + terminal = ( + "place" + if isinstance(target, Mapping) and target.get("kind") != "none" + else "hold" + ) + raw_step["terminal_behavior"] = terminal + changes.append( + { + "path": f"steps[{index}].terminal_behavior", + "from": "none", + "to": terminal, + "reason": "e4_terminal_inferred_from_own_target", + } + ) + return result, changes + + +def _empty_selector() -> dict[str, Any]: + """Return the canonical selector value for an inapplicable target.""" + return { + "kind": "none", + "step_id": "", + "reference": "", + "quantifier": "one", + "count": 0, + } + + +def validate_instruction_intent(value: Mapping[str, Any]) -> dict[str, Any]: + """Validate the private, non-graph instruction interpretation contract.""" + if not isinstance(value, Mapping): + raise TypeError("Instruction intent must be a mapping.") + _reject_forbidden_fields(value) + if set(value) != {"steps"}: + raise ValueError("Instruction intent may contain only 'steps'.") + raw_steps = value.get("steps") + if not isinstance(raw_steps, Sequence) or isinstance(raw_steps, (str, bytes)): + raise ValueError("Instruction intent steps must be a list.") + if not raw_steps: + raise ValueError("Instruction intent steps must not be empty.") + steps = [] + ids: set[str] = set() + dependencies: dict[str, list[str]] = {} + for index, raw in enumerate(raw_steps): + context = f"InstructionIntent.steps[{index}]" + if not isinstance(raw, Mapping): + raise ValueError(f"{context} must be a mapping.") + if set(raw) != _STEP_KEYS: + raise ValueError( + f"{context} requires exactly fields {sorted(_STEP_KEYS)}; " + f"received {sorted(raw)}." + ) + step = deepcopy(dict(raw)) + step_id = _nonempty(step["id"], f"{context}.id") + if step_id in ids: + raise ValueError(f"Duplicate instruction step ID {step_id!r}.") + ids.add(step_id) + step["id"] = step_id + step["task_type"] = _choice( + step["task_type"], TASK_TYPES, f"{context}.task_type" + ) + step["object"] = _validate_selector(step["object"], f"{context}.object") + step["target"] = _validate_selector(step["target"], f"{context}.target") + step["relation"] = _canonical_relation(step["relation"], f"{context}.relation") + for key in ("required_arm", "transfer_arm", "receive_arm"): + step[key] = _canonical_arm(step[key], f"{context}.{key}") + step["orientation_goal"] = _canonical_orientation( + step["orientation_goal"], f"{context}.orientation_goal" + ) + step["target_state"] = _choice( + step["target_state"], _TARGET_STATES, f"{context}.target_state" + ) + if isinstance(step["target_setting"], bool) or not isinstance( + step["target_setting"], int + ): + raise ValueError(f"{context}.target_setting must be an integer.") + step["layout"] = _choice(step["layout"], _LAYOUTS, f"{context}.layout") + step["axis"] = _choice(step["axis"], _AXES, f"{context}.axis") + step["direction"] = _choice( + step["direction"], _DIRECTIONS, f"{context}.direction" + ) + step["terminal_behavior"] = _choice( + step["terminal_behavior"], + _TERMINAL_BEHAVIORS, + f"{context}.terminal_behavior", + ) + raw_depends = step["depends_on"] + if not isinstance(raw_depends, Sequence) or isinstance( + raw_depends, (str, bytes) + ): + raise ValueError(f"{context}.depends_on must be a list.") + step["depends_on"] = [ + _nonempty(item, f"{context}.depends_on") for item in raw_depends + ] + if step_id in step["depends_on"]: + raise ValueError(f"{context}.depends_on cannot contain its own ID.") + dependencies[step_id] = step["depends_on"] + _validate_task_fields(step, context) + steps.append(step) + positions = {str(step["id"]): index for index, step in enumerate(steps)} + for index, step in enumerate(steps): + for selector_name in ("object", "target"): + selector = step[selector_name] + if selector["kind"] != "step_result": + continue + reference = str(selector["step_id"]) + if reference not in positions: + raise ValueError( + f"Instruction step {step['id']!r} {selector_name} references " + f"unknown step {reference!r}." + ) + if positions[reference] >= index: + raise ValueError( + f"Instruction step {step['id']!r} {selector_name} must reference " + f"a preceding step, not {reference!r}." + ) + for step_id, depends_on in dependencies.items(): + unknown = set(depends_on) - ids + if unknown: + raise ValueError( + f"Instruction step {step_id!r} has unknown dependencies " + f"{sorted(unknown)}." + ) + _validate_dag(dependencies) + return {"steps": steps} + + +def _validate_selector(value: Any, context: str) -> dict[str, Any]: + if not isinstance(value, Mapping): + raise ValueError(f"{context} must be a mapping.") + if set(value) != _SELECTOR_KEYS: + raise ValueError( + f"{context} requires exactly fields {sorted(_SELECTOR_KEYS)}; " + f"received {sorted(value)}." + ) + selector = deepcopy(dict(value)) + selector["kind"] = _choice(selector["kind"], _SELECTOR_KINDS, f"{context}.kind") + selector["step_id"] = _selector_string(selector["step_id"], f"{context}.step_id") + selector["reference"] = _selector_string( + selector["reference"], f"{context}.reference" + ) + selector["quantifier"] = _canonical_quantifier( + selector["quantifier"], f"{context}.quantifier" + ) + if isinstance(selector["count"], bool) or not isinstance(selector["count"], int): + raise ValueError(f"{context}.count must be an integer.") + if selector["count"] < 0: + raise ValueError(f"{context}.count must be non-negative.") + kind = selector["kind"] + if kind == "scene_ref" and not selector["reference"]: + raise ValueError(f"{context} scene_ref requires a reference.") + if kind == "step_result": + if not selector["step_id"]: + raise ValueError(f"{context} step_result requires step_id.") + if selector["reference"]: + raise ValueError( + f"{context} step_result may identify only a prior step_id." + ) + if selector["quantifier"] != "one" or selector["count"] != 0: + raise ValueError( + f"{context} step_result requires quantifier=one and count=0." + ) + if kind == "scene_ref" and selector["step_id"]: + raise ValueError(f"{context} scene_ref cannot carry step_id.") + if kind == "none" and (selector["step_id"] or selector["reference"]): + raise ValueError(f"{context} kind=none cannot carry constraints.") + if kind == "none" and (selector["quantifier"] != "one" or selector["count"] != 0): + raise ValueError(f"{context} kind=none requires quantifier=one and count=0.") + if selector["quantifier"] == "one" and selector["count"] != 0: + raise ValueError(f"{context} quantifier=one requires count=0.") + if selector["quantifier"] == "all" and selector["count"] != 0: + raise ValueError(f"{context} quantifier=all requires count=0.") + if selector["quantifier"] == "count" and selector["count"] < 1: + raise ValueError(f"{context} quantifier=count requires count>=1.") + return selector + + +def _validate_task_fields(step: Mapping[str, Any], context: str) -> None: + task_type = str(step["task_type"]) + if step["object"]["kind"] == "none": + raise _MissingRequiredObjectError( + f"{context} {task_type} requires an object selector." + ) + target_kind = str(step["target"]["kind"]) + if task_type not in {"E1", "E3", "E4", "E5"} and step["relation"] != "none": + raise ValueError(f"{context} {task_type} does not accept relation.") + if task_type == "E3" and step["relation"] != "above": + raise ValueError(f"{context} E3 relation must be above.") + target_setting = int(step["target_setting"]) + if task_type != "E8" and target_setting != 0: + raise ValueError(f"{context} target_setting is only valid for E8.") + if task_type != "E1" and step["axis"] != "none": + raise ValueError(f"{context} axis is only valid for E1 line arrangement.") + if task_type == "E1" and step["layout"] != "line" and step["axis"] != "none": + raise ValueError(f"{context} axis is only valid for E1 line arrangement.") + if task_type not in {"E6", "E7", "E9"} and step["target_state"] != "none": + raise ValueError(f"{context} target_state is not valid for {task_type}.") + if task_type != "E4" and step["transfer_arm"] != "none": + raise ValueError(f"{context} transfer_arm is only valid for E4.") + if task_type != "E4" and step["receive_arm"] != "none": + raise ValueError(f"{context} receive_arm is only valid for E4.") + orientation_goal = str(step["orientation_goal"]) + if task_type == "E2" and orientation_goal != "upright": + raise ValueError(f"{context} E2 orientation_goal must be upright.") + if task_type not in {"E1", "E2", "E4"} and orientation_goal != "none": + raise ValueError( + f"{context} orientation_goal is only valid for E1, E2, and E4." + ) + if task_type == "E1" and step["layout"] == "line": + if target_kind != "none": + raise ValueError(f"{context} E1 line arrangement cannot carry a target.") + if step["relation"] != "none": + raise ValueError(f"{context} E1 line arrangement cannot carry a relation.") + elif task_type in {"E1", "E3"}: + if target_kind == "none": + raise _MissingRequiredTargetError( + f"{context} {task_type} requires a target selector." + ) + if step["relation"] == "none" and task_type == "E3": + raise ValueError(f"{context} {task_type} requires a symbolic relation.") + elif task_type == "E4": + terminal = str(step["terminal_behavior"]) + effective_terminal = ( + "place" if terminal == "none" and target_kind != "none" else terminal + ) + if effective_terminal == "none": + effective_terminal = "hold" + if effective_terminal not in _TERMINAL_BEHAVIORS - {"none"}: + raise ValueError(f"{context} E4 requires terminal_behavior hold/place.") + if effective_terminal == "place": + if target_kind == "none" or step["relation"] == "none": + raise ValueError( + f"{context} E4 terminal_behavior=place requires target and relation." + ) + elif target_kind != "none" or step["relation"] != "none": + raise ValueError( + f"{context} E4 terminal_behavior=hold cannot carry target or relation." + ) + elif task_type == "E5": + direction = str(step["direction"]) + terminal = str(step["terminal_behavior"]) + if terminal not in _TERMINAL_BEHAVIORS - {"none"}: + raise ValueError(f"{context} E5 requires terminal_behavior hold/place.") + if target_kind == "none": + if step["relation"] != "none": + raise ValueError(f"{context} E5 relation requires a target selector.") + if direction == "none" and terminal != "place": + raise ValueError( + f"{context} E5 requires a direction or target relation." + ) + else: + if step["relation"] == "none": + raise ValueError(f"{context} E5 target requires a relation.") + if direction != "none": + raise ValueError( + f"{context} E5 target relation cannot also carry direction." + ) + elif target_kind != "none": + raise ValueError(f"{context} {task_type} does not accept a target selector.") + if task_type not in {"E4", "E5"}: + if step["direction"] != "none": + raise ValueError(f"{context} direction is only valid for E5.") + if step["terminal_behavior"] != "none": + raise ValueError(f"{context} terminal_behavior is only valid for E5.") + if task_type == "E4": + transfer = str(step["transfer_arm"]) + receive = str(step["receive_arm"]) + if transfer not in {"left_arm", "right_arm"} or receive not in { + "left_arm", + "right_arm", + }: + raise ValueError(f"{context} E4 requires two explicit arms.") + if transfer == receive: + raise ValueError(f"{context} E4 transfer and receive arms must differ.") + if step["required_arm"] not in {"none", "auto"}: + raise ValueError( + f"{context} E4 uses transfer_arm/receive_arm, not required_arm." + ) + if task_type == "E5" and step["required_arm"] not in {"none", "auto"}: + raise ValueError(f"{context} E5 always uses both arms, not required_arm.") + if task_type == "E6" and step["target_state"] != "open": + raise ValueError(f"{context} E6 target_state must be open.") + if task_type == "E7" and step["target_state"] != "closed": + raise ValueError(f"{context} E7 target_state must be closed.") + if task_type == "E9" and step["target_state"] != "activated": + raise ValueError(f"{context} E9 target_state must be activated.") + if step["layout"] == "line" and task_type != "E1": + raise ValueError(f"{context} only E1 supports layout=line.") + + +def _instruction_prompt(instruction: str) -> str: + return ( + "Convert the user's explicit L1-L3 instruction into typed E1-E9 task " + "intent. Understand synonyms, ellipsis, and pronouns, but " + "do not invent missing objects. Use step_result for cross-step pronouns " + "and explicit references to the result of an earlier manipulation. Keep " + "an independently selected repeated noun as scene_ref; identical text " + "alone does not prove object identity. " + "Object directions are robot-relative; arm names are robot body sides. " + "Preserve each concrete object or target phrase from the instruction as " + "an open scene_ref.reference. Do not classify it or emit a scene UID. " + "Emit no AtomicAction, category label, affordance, coordinates, poses, " + "paths, or reasoning. Encode explicit ordering with depends_on; same-action set " + "members may remain independent. Use empty strings and 'none' for " + "inapplicable required fields. A request to retract the transfer arm " + "E4 owns the complete transfer. For a handover followed by placement in " + "the same user intent, emit one E4 with target, relation, and " + "terminal_behavior=place; do not emit a trailing E1. Use " + "terminal_behavior=hold only when the receiver should keep holding the " + "object. The exact output keys are steps -> id, " + "task_type, object, target, relation, required_arm, transfer_arm, " + "receive_arm, orientation_goal, target_state, target_setting, layout, " + "axis, direction, terminal_behavior, depends_on; each selector has kind, " + "step_id, reference, quantifier, count.\n\n" + "Use orientation_goal=none unless the instruction explicitly requests " + "upright orientation or preserving the original orientation. Spatial " + "placement and handover alone do not imply preserve. " + "Emptying, dumping, or pouring contents from one container into another " + "is exactly one E3 step: object selects the source container, target " + "selects the receiving container, and relation=above. Pickup and staging " + "are internal to that E3 step. " + "Opening or pulling out a drawer is E6 with object selecting that drawer " + "and target_state=open. Closing or pushing in a drawer is E7 with object " + "selecting that drawer and target_state=closed. " + f"Instruction:\n{instruction}\n\n" + f"E1-E9 catalog:\n{json.dumps(_intent_capability_catalog(), ensure_ascii=False, sort_keys=True)}\n\n" + "Shape-only complete JSON example (do not copy its step count or values; " + "copy every key, including keys whose value is none/empty/0):\n" + f"{json.dumps(_instruction_shape_example(), ensure_ascii=False, sort_keys=True)}\n\n" + "Selector kind rules (these are not extra output fields):\n" + f"{_instruction_selector_rules()}\n\n" + "For E5, use target+relation for moving an object relative to another " + "object, or direction for a small robot-relative move. A dual-arm pick, " + "lift, raise, or hold request without another target uses direction=up " + "and terminal_behavior=hold. Use hold unless the instruction explicitly " + "says to put/release the object. For pick " + "and release at the original location, use direction=none and place. A dual-arm " + "pick/move/transport request uses E5. Final checklist: every step " + "has all 16 step keys; every object and target " + "has all 5 selector keys. For an inapplicable field use the canonical " + "default shown in the example, never omit the field. E4 must explicitly " + "state transfer_arm, receive_arm, and terminal_behavior. E1/E3 must explicitly state target " + "and relation (except E1 layout=line)." + ) + + +def _instruction_shape_example() -> dict[str, Any]: + """Return a compact field-complete example for providers with weak schemas.""" + selector = { + "kind": "scene_ref", + "step_id": "", + "reference": "example object A", + "quantifier": "one", + "count": 0, + } + empty_selector = { + "kind": "none", + "step_id": "", + "reference": "", + "quantifier": "one", + "count": 0, + } + return { + "steps": [ + { + "id": "step_1", + "task_type": "E2", + "object": selector, + "target": empty_selector, + "relation": "none", + "required_arm": "auto", + "transfer_arm": "none", + "receive_arm": "none", + "orientation_goal": "upright", + "target_state": "none", + "target_setting": 0, + "layout": "none", + "axis": "none", + "direction": "none", + "terminal_behavior": "none", + "depends_on": [], + } + ] + } + + +def _instruction_selector_rules() -> str: + """Return the mutually exclusive selector encodings for model prompts.""" + step_result = { + "kind": "step_result", + "step_id": "step_1", + "reference": "", + "quantifier": "one", + "count": 0, + } + return ( + "- kind=none: step_id and reference are empty strings; " + "quantifier='one'; count=0.\n" + "- kind=scene_ref: step_id is empty and reference preserves the concrete " + "object phrase from the user's instruction. Repeated scene_ref text does " + "not establish cross-step identity.\n" + "- kind=step_result: use it only for a pronoun that means exactly one " + "object, or an explicit continuation of the result of an earlier " + "instruction step. Set step_id to that prior " + "step ID and set reference='', quantifier='one', count=0. Do not copy " + "the prior object's phrase into this selector. Replace step_1 in this " + f"complete shape with the actual prior step ID: {json.dumps(step_result, sort_keys=True)}\n" + "A step_result may identify only a prior step_id; it cannot carry any " + "other object constraint." + ) + + +def _instruction_repair_guidance(error: Exception) -> str: + """Add narrow semantic guidance for errors weak JSON-mode models repeat.""" + if "E4 transfer and receive arms must differ" in str(error): + return ( + "\nSame-arm handover repair rule: transfer_arm and receive_arm must " + "name different arms. Preserve the explicitly stated transfer arm. " + "When a later clause clearly continues with the handed object using " + "the other arm, use that arm as receive_arm. Resolve coreference from " + "the instruction semantics; identical scene_ref text alone does not " + "prove that two independently selected objects are the same.\n" + ) + if isinstance(error, _MissingRequiredObjectError): + return ( + "\nMissing-object repair rule: preserve the selected task_type and " + "set object to a scene_ref that preserves the explicit manipulated " + "object phrase from the instruction. For E6/E7 the drawer, door, or " + "other articulated part is the object selector; target remains none.\n" + ) + if not isinstance(error, _MissingRequiredTargetError): + return "" + if " E3 requires a target selector" in str(error): + return ( + "\nMissing-target repair rule for E3: keep task_type=E3. object is " + "the source container whose contents are poured, target is the " + "receiving container, and relation must be above. An explicit grab " + "is part of the same E3 task.\n" + ) + return ( + "\nMissing-target repair rule: for a non-line E1 placement, object is " + "the item being moved and target is the explicit reference object " + "after the spatial relation in the original instruction. For example, " + "in 'place it to the left of the striped pedestal', object is the earlier " + "step_result for 'it', while target selects the striped pedestal; target " + "must not use kind=none. Use target kind=step_result only when the " + "reference object itself is exactly the result of a prior step.\n" + ) + + +def _intent_capability_catalog() -> dict[str, dict[str, Any]]: + """Return the LLM's thin, import-safe E1-E9 capability view. + + Action Engine's online planning catalog also reports runtime availability + and therefore imports simulator action classes. Text interpretation only + needs symbolic E semantics and must remain testable before a simulator + backend is installed. + """ + return { + task_type: { + "semantics": contract.semantics, + "applicable_fields": sorted(_INTENT_TASK_FIELD_REGISTRY[task_type]), + } + for task_type, contract in TASK_CONTRACTS.items() + } + + +def _default_instruction_caller( + *, + prompt: str, + schema: Mapping[str, Any], + model: str | None, +) -> Mapping[str, Any]: + from langchain_core.messages import HumanMessage, SystemMessage + from langchain_openai import ChatOpenAI + + settings = _load_llm_settings(model=model) + kwargs: dict[str, Any] = { + "api_key": settings["api_key"], + "model": settings["model"], + "temperature": 0, + "http_socket_options": (), + } + for key in ("base_url", "default_query"): + if settings[key]: + kwargs[key] = settings[key] + if _is_mimo_compatible(settings): + # MiMo documents ``thinking`` as a provider extension carried in the + # OpenAI client's extra body. Disabling it is important here: hidden + # reasoning can consume the completion and leave only id/object/type. + kwargs.update( + { + "max_completion_tokens": _MIMO_MAX_COMPLETION_TOKENS, + "extra_body": {"thinking": {"type": "disabled"}}, + } + ) + client = ChatOpenAI(**kwargs) + # The full schema remains in the prompt and the local validator is still + # authoritative even when the provider only offers JSON mode. + structured = _structured_output_runnable( + client, + schema, + settings=settings, + ) + schema_prompt = ( + f"{prompt}\n\nReturn one JSON object conforming exactly to this JSON " + f"Schema:\n{json.dumps(schema, ensure_ascii=False, sort_keys=True)}" + ) + response = structured.invoke( + [ + SystemMessage( + content=( + "Return only the requested structured JSON response. Never " + "return reasoning, coordinates, or AtomicAction nodes." + ) + ), + HumanMessage(content=schema_prompt), + ] + ) + return _coerce_instruction_response(response) + + +def _instruction_model(explicit: str | None) -> str | None: + if isinstance(explicit, str) and explicit.strip(): + return explicit.strip() + # Keep model selection separate from credential loading. Reading the local + # dotenv file is side-effect free and gives generation the documented + # priority without leaking credentials into TaskSpec metadata. + for name in ("TASK_ENGINE_LLM_MODEL", "ACTION_ENGINE_LLM_MODEL", "OPENAI_MODEL"): + for source in ( + os.environ, + _load_local_env(), + ): + value = source.get(name) + if isinstance(value, str) and value.strip(): + return value.strip() + return None + + +def _load_local_env() -> dict[str, str]: + """Read Task Engine model configuration without mutating the environment.""" + return _load_env_file(_GEN_SIM_ENV_PATH) + + +def _is_mimo_compatible(settings: Mapping[str, Any]) -> bool: + model = str(settings.get("model", "")).casefold() + base_url = str(settings.get("base_url", "")).casefold() + return "mimo" in model or "xiaomimimo.com" in base_url + + +def _structured_output_runnable( + client: Any, + schema: Mapping[str, Any], + *, + settings: Mapping[str, Any], +) -> Any: + if not hasattr(client, "with_structured_output"): + return client + method = "json_mode" if _is_mimo_compatible(settings) else "json_schema" + try: + return client.with_structured_output(schema, method=method) + except (TypeError, ValueError): + if method == "json_mode" and hasattr(client, "bind"): + from langchain_core.output_parsers import JsonOutputParser + + return ( + client.bind(response_format={"type": "json_object"}) + | JsonOutputParser() + ) + return client.with_structured_output(schema) + + +def _load_llm_settings(*, model: str | None) -> dict[str, Any]: + local_env = _load_local_env() + config: dict[str, Any] = {} + if _GEN_CONFIG_PATH.is_file(): + raw = json.loads(_GEN_CONFIG_PATH.read_text(encoding="utf-8")) + if isinstance(raw, Mapping): + llm = raw.get("llm", {}) + if isinstance(llm, Mapping): + configured = llm.get("openai_compatible", {}) + if isinstance(configured, Mapping): + config = dict(configured) + api_key, base_url = _resolve_transport_settings(local_env, config) + selected_model = ( + (model.strip() if isinstance(model, str) else "") + or _first_env_value( + local_env, + "TASK_ENGINE_LLM_MODEL", + "ACTION_ENGINE_LLM_MODEL", + "OPENAI_MODEL", + "LLM_MODEL", + ) + or str(config.get("model", "")).strip() + ) + default_query = config.get("default_query", {}) or {} + if not api_key: + raise ValueError( + "OPENAI_API_KEY is required for Task Engine interpretation. Set it " + f"in the process environment or {_GEN_SIM_ENV_PATH}." + ) + if not selected_model: + raise ValueError( + "A text LLM model is required through model=, TASK_ENGINE_LLM_MODEL, " + f"OPENAI_MODEL, or {_GEN_CONFIG_PATH}." + ) + if not isinstance(default_query, Mapping): + raise ValueError("LLM default_query must be a mapping.") + return { + "api_key": api_key, + "model": selected_model, + "base_url": base_url, + "default_query": dict(default_query), + } + + +def _resolve_transport_settings( + local_env: Mapping[str, str], + config: Mapping[str, Any], +) -> tuple[str, str]: + """Resolve an API key and endpoint from one configuration source.""" + transports = ( + ( + _mapping_value(os.environ, "OPENAI_API_KEY"), + _mapping_value( + os.environ, + "OPENAI_BASE_URL", + "OPENAI_API_BASE", + "LLM_URL", + ), + ), + ( + _mapping_value(local_env, "OPENAI_API_KEY"), + _mapping_value( + local_env, + "OPENAI_BASE_URL", + "OPENAI_API_BASE", + "LLM_URL", + ), + ), + ( + _mapping_value(config, "api_key"), + _mapping_value(config, "base_url"), + ), + ) + for api_key, base_url in transports: + if api_key and base_url: + return api_key, base_url.rstrip("/") + for api_key, base_url in transports: + if api_key: + return api_key, base_url.rstrip("/") + return "", "" + + +def _mapping_value(source: Mapping[str, Any], *names: str) -> str: + for name in names: + value = source.get(name) + if isinstance(value, str) and value.strip(): + return value.strip() + return "" + + +def _load_env_file(path: Path) -> dict[str, str]: + if not path.is_file(): + return {} + values: dict[str, str] = {} + for line_number, raw_line in enumerate( + path.read_text(encoding="utf-8").splitlines(), start=1 + ): + line = raw_line.strip() + if not line or line.startswith("#"): + continue + if line.startswith("export "): + line = line[len("export ") :].lstrip() + if "=" not in line: + continue + key, raw_value = line.split("=", 1) + key = key.strip() + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", key): + raise ValueError(f"Invalid dotenv key at {path}:{line_number}.") + value = raw_value.strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: + value = value[1:-1] + elif " #" in value: + value = value.split(" #", 1)[0].rstrip() + values[key] = value + return values + + +def _first_env_value(local_env: Mapping[str, str], *names: str) -> str | None: + for source in (os.environ, local_env): + for name in names: + value = source.get(name) + if isinstance(value, str) and value.strip(): + return value.strip() + return None + + +def _choice(value: Any, allowed: set[str] | frozenset[str], context: str) -> str: + if not isinstance(value, str) or value not in allowed: + raise ValueError(f"{context} must be one of {sorted(allowed)}.") + return value + + +def _selector_string(value: Any, context: str) -> str: + if not isinstance(value, str): + raise ValueError(f"{context} must be a string.") + return value.strip() + + +def _canonical_quantifier(value: Any, context: str) -> str: + return _choice(value, _QUANTIFIERS, context) + + +def _canonical_arm(value: Any, context: str) -> str: + return _choice(value, _ARMS, context) + + +def _canonical_relation(value: Any, context: str) -> str: + return _choice(value, _RELATIONS, context) + + +def _canonical_orientation(value: Any, context: str) -> str: + return _choice(value, _ORIENTATIONS, context) + + +def _nonempty(value: Any, context: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{context} must be a non-empty string.") + return value.strip() + + +def _topological_steps(steps: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + """Return a stable topological ordering for validated intent steps.""" + by_id = {str(step["id"]): dict(step) for step in steps} + effective_dependencies: dict[str, tuple[str, ...]] = {} + for step_id, step in by_id.items(): + deps = list(str(dep) for dep in step["depends_on"]) + for selector in (step["object"], step["target"]): + if selector["kind"] == "step_result": + reference = str(selector["step_id"]) + if reference not in deps: + deps.append(reference) + effective_dependencies[step_id] = tuple(deps) + pending = set(by_id) + ordered: list[dict[str, Any]] = [] + original = [str(step["id"]) for step in steps] + while pending: + ready = [ + step_id + for step_id in original + if step_id in pending + and all(str(dep) not in pending for dep in effective_dependencies[step_id]) + ] + if not ready: + raise ValueError("Instruction intent dependencies contain a cycle.") + for step_id in ready: + ordered.append(by_id[step_id]) + pending.remove(step_id) + return ordered + + +def _coerce_instruction_response(response: Any) -> Mapping[str, Any]: + """Coerce common structured-client response wrappers without accepting prose.""" + if isinstance(response, Mapping): + return dict(response) + model_dump = getattr(response, "model_dump", None) + if callable(model_dump): + dumped = model_dump() + if isinstance(dumped, Mapping): + return dict(dumped) + content = getattr(response, "content", response) + if isinstance(content, Mapping): + return dict(content) + if isinstance(content, list): + content = "\n".join( + str(item.get("text", "")) + for item in content + if isinstance(item, Mapping) and item.get("type") == "text" + ) + if not isinstance(content, str): + raise ValueError( + f"Instruction model output has unsupported type {type(content).__name__}." + ) + text = content.strip() + if text.startswith("```"): + lines = text.splitlines() + if lines: + lines = lines[1:] + if lines and lines[-1].strip().startswith("```"): + lines = lines[:-1] + text = "\n".join(lines).strip() + try: + parsed = json.loads(text) + except json.JSONDecodeError as exc: + raise ValueError(f"Instruction model output is not valid JSON: {exc}") from exc + if not isinstance(parsed, Mapping): + raise ValueError("Instruction model output must decode to a JSON object.") + return dict(parsed) + + +def _validate_dag(dependencies: Mapping[str, Sequence[str]]) -> None: + visiting: set[str] = set() + visited: set[str] = set() + + def visit(node: str) -> None: + if node in visiting: + raise ValueError("Instruction intent dependencies contain a cycle.") + if node in visited: + return + visiting.add(node) + for dependency in dependencies[node]: + visit(str(dependency)) + visiting.remove(node) + visited.add(node) + + for node in dependencies: + visit(node) + + +def _reject_forbidden_fields(value: Any) -> None: + if isinstance(value, Mapping): + forbidden = _FORBIDDEN_FIELDS & {str(key).strip().lower() for key in value} + if forbidden: + raise ValueError( + f"Instruction intent contains forbidden fields {sorted(forbidden)}." + ) + for item in value.values(): + _reject_forbidden_fields(item) + elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)): + for item in value: + _reject_forbidden_fields(item) diff --git a/embodichain/gen_sim/task_engine/ontology.py b/embodichain/gen_sim/task_engine/ontology.py new file mode 100644 index 000000000..1f497b645 --- /dev/null +++ b/embodichain/gen_sim/task_engine/ontology.py @@ -0,0 +1,292 @@ +# ---------------------------------------------------------------------------- +# 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. +# ---------------------------------------------------------------------------- + +"""Scene-independent semantic ontology for the canonical E1-E9 tasks.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from types import MappingProxyType +from typing import Any + +__all__ = [ + "RELATIONS", + "TASK_CONTRACTS", + "TERMINAL_BEHAVIORS", + "TRANSPORT_DIRECTIONS", + "TaskContract", + "task_contract", + "task_success_type", +] + + +# These are protocol values consumed by executable planners. They are not a +# vocabulary for matching words in user instructions. +RELATIONS = frozenset( + { + "none", + "on", + "inside", + "above", + "left_of", + "right_of", + "front_of", + "behind", + "front_left_of", + "front_right_of", + "back_left_of", + "back_right_of", + } +) +TRANSPORT_DIRECTIONS = frozenset( + { + "none", + "world_x", + "world_y", + "front", + "back", + "left", + "right", + "front_left", + "front_right", + "back_left", + "back_right", + "up", + "down", + } +) +TERMINAL_BEHAVIORS = frozenset({"none", "hold", "place"}) +_RESOURCE_MODES = frozenset({"single_arm", "handover", "coordinated"}) + + +@dataclass(frozen=True, slots=True) +class TaskContract: + """One scene-independent semantic E-task contract.""" + + task_type: str + semantics: str + applicable_intent_fields: frozenset[str] + source_structure: str + required_affordances: frozenset[str] + success_type: str + scene_affordances: frozenset[str] + primary_role_field: str + resource_mode: str + moves_primary_object: bool + accepts_direct_payloads: bool + direct_payload_relations: frozenset[str] + accepts_incoming_hold: bool + terminal_success_types: tuple[tuple[str, str], ...] + + def __post_init__(self) -> None: + if not self.primary_role_field.endswith("_role"): + raise ValueError("primary_role_field must name one role parameter.") + if self.resource_mode not in _RESOURCE_MODES: + raise ValueError(f"Unknown task resource_mode {self.resource_mode!r}.") + if self.direct_payload_relations - RELATIONS: + raise ValueError("direct_payload_relations contain unknown relations.") + terminal_behaviors = [item[0] for item in self.terminal_success_types] + if len(terminal_behaviors) != len(set(terminal_behaviors)): + raise ValueError("terminal_success_types must use unique behaviors.") + if set(terminal_behaviors) - TERMINAL_BEHAVIORS: + raise ValueError("terminal_success_types contain unknown behaviors.") + + +def _contract( + task_type: str, + semantics: str, + applicable_intent_fields: frozenset[str], + source_structure: str, + required_affordances: frozenset[str], + success_type: str, + *, + scene_affordances: frozenset[str] | None = None, + primary_role_field: str = "object_role", + resource_mode: str = "single_arm", + moves_primary_object: bool = False, + accepts_direct_payloads: bool = False, + direct_payload_relations: frozenset[str] = frozenset(), + accepts_incoming_hold: bool = False, + terminal_success_types: tuple[tuple[str, str], ...] = (), +) -> TaskContract: + return TaskContract( + task_type=task_type, + semantics=semantics, + applicable_intent_fields=applicable_intent_fields, + source_structure=source_structure, + required_affordances=required_affordances, + success_type=success_type, + scene_affordances=scene_affordances or required_affordances, + primary_role_field=primary_role_field, + resource_mode=resource_mode, + moves_primary_object=moves_primary_object, + accepts_direct_payloads=accepts_direct_payloads, + direct_payload_relations=direct_payload_relations, + accepts_incoming_hold=accepts_incoming_hold, + terminal_success_types=terminal_success_types, + ) + + +TASK_CONTRACTS: Mapping[str, TaskContract] = MappingProxyType( + { + "E1": _contract( + "E1", + "Pick, move, and place one object at a symbolic relation.", + frozenset( + { + "target", + "relation", + "required_arm", + "orientation_goal", + "layout", + "axis", + } + ), + "rigid_object", + frozenset({"graspable", "placeable"}), + "semantic_goal", + moves_primary_object=True, + accepts_direct_payloads=True, + direct_payload_relations=frozenset({"on", "inside"}), + accepts_incoming_hold=True, + ), + "E2": _contract( + "E2", + "Make one fallen object upright and place it stably.", + frozenset({"required_arm", "orientation_goal"}), + "rigid_object", + frozenset({"graspable", "orientable"}), + "object_upright", + moves_primary_object=True, + accepts_incoming_hold=True, + ), + "E3": _contract( + "E3", + "Pick up a source container, execute a tilt-and-restore pour over " + "a fixed target container, then place and home.", + frozenset({"target", "relation", "required_arm"}), + "rigid_object", + frozenset({"graspable", "pourable"}), + "poured", + primary_role_field="source_role", + moves_primary_object=True, + accepts_incoming_hold=True, + ), + "E4": _contract( + "E4", + "Transfer one object between arms, then either leave the receiver " + "holding it safely or place it at a symbolic relation.", + frozenset( + { + "target", + "relation", + "transfer_arm", + "receive_arm", + "orientation_goal", + "terminal_behavior", + } + ), + "rigid_object", + frozenset({"graspable", "handover"}), + "handover_complete", + resource_mode="handover", + moves_primary_object=True, + accepts_incoming_hold=True, + terminal_success_types=( + ("hold", "handover_complete"), + ("place", "semantic_goal"), + ), + ), + "E5": _contract( + "E5", + "Use both arms to pick, move, and optionally release one shared rigid object.", + frozenset({"target", "relation", "direction", "terminal_behavior"}), + "rigid_object", + frozenset({"dual_graspable"}), + "held_by_both_grippers", + scene_affordances=frozenset({"dual_graspable", "rigid"}), + resource_mode="coordinated", + moves_primary_object=True, + accepts_direct_payloads=True, + terminal_success_types=( + ("hold", "held_by_both_grippers"), + ("place", "semantic_goal"), + ), + ), + "E6": _contract( + "E6", + "Pull an articulated part to its requested state.", + frozenset({"required_arm", "target_state"}), + "articulation", + frozenset({"pullable"}), + "articulation_joint_near", + scene_affordances=frozenset({"articulated", "pullable"}), + ), + "E7": _contract( + "E7", + "Push an articulated part to its requested state.", + frozenset({"required_arm", "target_state"}), + "articulation", + frozenset({"pushable"}), + "articulation_joint_near", + scene_affordances=frozenset({"articulated", "pushable"}), + ), + "E8": _contract( + "E8", + "Turn one knob to a requested setting.", + frozenset({"required_arm", "target_setting"}), + "articulation", + frozenset({"turnable"}), + "articulation_joint_near", + ), + "E9": _contract( + "E9", + "Press one button until its requested terminal state.", + frozenset({"required_arm", "target_state"}), + "articulation", + frozenset({"pressable"}), + "pressed", + ), + } +) + + +def task_contract(task_type: str) -> TaskContract: + """Return the canonical contract or reject an unknown E-task type.""" + try: + return TASK_CONTRACTS[str(task_type)] + except KeyError as exc: + raise ValueError(f"Unsupported task type {task_type!r}.") from exc + + +def task_success_type( + task_type: str, + params: Mapping[str, Any] | None = None, +) -> str: + """Resolve a TaskSpec success type, including E5's terminal behavior.""" + contract = task_contract(task_type) + if not contract.terminal_success_types: + return contract.success_type + terminal_behavior = str((params or {}).get("terminal_behavior", "hold")) + success_by_behavior = dict(contract.terminal_success_types) + try: + return success_by_behavior[terminal_behavior] + except KeyError as exc: + raise ValueError( + f"{contract.task_type} terminal_behavior must be one of " + f"{sorted(success_by_behavior)}." + ) from exc diff --git a/tests/gen_sim/__init__.py b/tests/gen_sim/__init__.py index 355d915ff..cdeead7b0 100644 --- a/tests/gen_sim/__init__.py +++ b/tests/gen_sim/__init__.py @@ -14,4 +14,8 @@ # limitations under the License. # ---------------------------------------------------------------------------- +"""Generative simulation tests.""" + from __future__ import annotations + +__all__: list[str] = [] diff --git a/tests/gen_sim/scene_engine/test_pipeline_api.py b/tests/gen_sim/scene_engine/test_pipeline_api.py new file mode 100644 index 000000000..9b10361f7 --- /dev/null +++ b/tests/gen_sim/scene_engine/test_pipeline_api.py @@ -0,0 +1,244 @@ +# ---------------------------------------------------------------------------- +# 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. +# ---------------------------------------------------------------------------- + +from __future__ import annotations + +from copy import deepcopy +import json +from pathlib import Path + +from embodichain.gen_sim.scene_engine.core.scene import Scene +from embodichain.gen_sim.scene_engine.core.scene_edit_plan import SceneEditPlan +from embodichain.gen_sim.scene_engine.core.scene_graph import ( + SceneGraph, + SceneGraphNode, +) +from embodichain.gen_sim.scene_engine.core.scene_object import SceneObject +from embodichain.gen_sim.scene_engine.pipeline import api + + +class _HealthyClient: + def __init__(self) -> None: + self.health_checks = 0 + + def check_health(self) -> None: + self.health_checks += 1 + + +def _materialization( + *, + scene: Scene, + scene_graph: SceneGraph, + output_root: Path, +) -> api.SceneMaterialization: + return api.SceneMaterialization( + scene=scene, + scene_graph=scene_graph, + output_root=output_root, + scene_config_path=output_root / "scene_export" / "scene_config.json", + ) + + +def _table_scene() -> tuple[Scene, SceneGraph]: + scene = Scene( + objects=[ + SceneObject( + id="table", + kind="table", + category="table", + name="table", + description="A work table.", + ) + ] + ) + graph = SceneGraph(nodes=[SceneGraphNode(object_id="table", parent_id=None)]) + return scene, graph + + +def test_analyze_image_persists_blueprint_and_artifact_hashes( + tmp_path: Path, + monkeypatch, +) -> None: + image_path = tmp_path / "input.png" + image_path.write_bytes(b"image") + scene, graph = _table_scene() + + def fake_understand_scene(**kwargs): + stage_root = Path(kwargs["output_root"]) / "scene_understanding" + stage_root.mkdir(parents=True) + (stage_root / "table-mask.png").write_bytes(b"mask") + return scene, graph + + monkeypatch.setattr(api, "understand_scene", fake_understand_scene) + segmentation = _HealthyClient() + package = api.analyze_image( + image_path, + tmp_path / "output", + vlm_client=object(), + image_segmentation_client=segmentation, + ) + document = json.loads(package.manifest_path.read_text(encoding="utf-8")) + + assert segmentation.health_checks == 1 + assert document["blueprint_id"] == package.blueprint_id + assert document["scene_graph"] == graph.to_dict() + assert document["artifacts"][0]["path"].endswith("table-mask.png") + assert len(document["artifacts"][0]["sha256"]) == 64 + + +def test_analyze_edit_persists_post_edit_blueprint( + tmp_path: Path, + monkeypatch, +) -> None: + scene, graph = _table_scene() + plan = SceneEditPlan(scene=scene, scene_graph=graph, operations=[]) + + class FakeImporter: + def __init__(self, *, output_root: Path) -> None: + self.output_root = output_root + + def import_scene_and_graph(self): + return scene, graph + + monkeypatch.setattr(api, "SceneExportImporter", FakeImporter) + monkeypatch.setattr( + api, + "understand_scene_edit", + lambda **_: (plan, graph), + ) + package = api.analyze_edit( + output_root=tmp_path, + edit_prompt="Keep the scene unchanged.", + vlm_client=object(), + ) + document = json.loads(package.manifest_path.read_text(encoding="utf-8")) + + assert document["blueprint_id"] == package.blueprint_id + assert document["scene_edit_plan"] == plan.to_dict() + assert document["updated_scene_graph"] == graph.to_dict() + + +def test_materialize_blueprint_does_not_mutate_audited_snapshot( + tmp_path: Path, + monkeypatch, +) -> None: + scene, graph = _table_scene() + manifest_path = tmp_path / "scene_blueprint.json" + manifest_path.write_text("audited blueprint\n", encoding="utf-8") + package = api.SceneBlueprintPackage( + blueprint_id="blueprint", + image_path=tmp_path / "input.png", + output_root=tmp_path, + manifest_path=manifest_path, + scene=scene, + scene_graph=graph, + ) + original_scene = deepcopy(scene.to_dict()) + original_graph = deepcopy(graph.to_dict()) + + def fake_generate_scene_and_refine(**kwargs): + assert kwargs["seed"] == 31 + assert kwargs["scene"] is not package.scene + assert kwargs["scene_graph"] is not package.scene_graph + kwargs["scene"].objects[0].name = "materialized table" + return kwargs["scene"] + + monkeypatch.setattr( + api, + "generate_scene_and_refine", + fake_generate_scene_and_refine, + ) + monkeypatch.setattr( + api, + "_export_materialization", + lambda *, scene, scene_graph, output_root: _materialization( + scene=scene, + scene_graph=scene_graph, + output_root=output_root, + ), + ) + + result = api.materialize_blueprint( + package, + vlm_client=object(), + geometry_generation_client=_HealthyClient(), + seed=31, + ) + + assert result.scene.objects[0].name == "materialized table" + assert package.scene.to_dict() == original_scene + assert package.scene_graph.to_dict() == original_graph + assert manifest_path.read_text(encoding="utf-8") == "audited blueprint\n" + + +def test_materialize_edit_does_not_mutate_audited_snapshot( + tmp_path: Path, + monkeypatch, +) -> None: + scene, graph = _table_scene() + plan = SceneEditPlan(scene=scene, scene_graph=graph, operations=[]) + manifest_path = tmp_path / "scene_edit_blueprint.json" + manifest_path.write_text("audited edit blueprint\n", encoding="utf-8") + package = api.SceneEditBlueprintPackage( + blueprint_id="edit-blueprint", + edit_prompt="Keep the scene unchanged.", + output_root=tmp_path, + manifest_path=manifest_path, + scene_edit_plan=plan, + updated_scene_graph=graph, + ) + original_plan = deepcopy(plan.to_dict()) + original_graph = deepcopy(graph.to_dict()) + + def fake_prepare_scene_edit_assets(**kwargs): + assert kwargs["seed"] == 32 + return [] + + monkeypatch.setattr( + api, "prepare_scene_edit_assets", fake_prepare_scene_edit_assets + ) + + def fake_edit_layout(**kwargs): + assert kwargs["scene_edit_plan"] is not package.scene_edit_plan + assert kwargs["updated_scene_graph"] is not package.updated_scene_graph + kwargs["scene"].objects[0].name = "edited table" + return kwargs["scene"] + + monkeypatch.setattr(api, "edit_layout", fake_edit_layout) + monkeypatch.setattr( + api, + "_export_materialization", + lambda *, scene, scene_graph, output_root: _materialization( + scene=scene, + scene_graph=scene_graph, + output_root=output_root, + ), + ) + clients = [_HealthyClient(), _HealthyClient(), _HealthyClient()] + + result = api.materialize_edit( + package, + vlm_client=object(), + image_generation_client=clients[0], + geometry_generation_client=clients[1], + image_segmentation_client=clients[2], + seed=32, + ) + + assert result.scene.objects[0].name == "edited table" + assert package.scene_edit_plan.to_dict() == original_plan + assert package.updated_scene_graph.to_dict() == original_graph + assert manifest_path.read_text(encoding="utf-8") == "audited edit blueprint\n" diff --git a/tests/gen_sim/task_engine/__init__.py b/tests/gen_sim/task_engine/__init__.py new file mode 100644 index 000000000..b201491d8 --- /dev/null +++ b/tests/gen_sim/task_engine/__init__.py @@ -0,0 +1,19 @@ +# ---------------------------------------------------------------------------- +# 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. +# ---------------------------------------------------------------------------- + +"""Tests for Task Engine semantics and orchestration.""" + +from __future__ import annotations diff --git a/tests/gen_sim/task_engine/test_interpretation.py b/tests/gen_sim/task_engine/test_interpretation.py new file mode 100644 index 000000000..009362e87 --- /dev/null +++ b/tests/gen_sim/task_engine/test_interpretation.py @@ -0,0 +1,96 @@ +# ---------------------------------------------------------------------------- +# 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. +# ---------------------------------------------------------------------------- + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from embodichain.gen_sim.task_engine import interpretation as interpretation_module + + +def _write_dotenv(path: Path) -> None: + path.write_text( + "\n".join( + ( + "OPENAI_API_KEY=dotenv-key", + "OPENAI_BASE_URL=https://dotenv.example/v1", + "OPENAI_MODEL=dotenv-model", + ) + ), + encoding="utf-8", + ) + + +def _clear_process_provider(monkeypatch: pytest.MonkeyPatch) -> None: + for name in ( + "OPENAI_API_KEY", + "OPENAI_BASE_URL", + "OPENAI_API_BASE", + "LLM_URL", + "TASK_ENGINE_LLM_MODEL", + "ACTION_ENGINE_LLM_MODEL", + "OPENAI_MODEL", + "LLM_MODEL", + ): + monkeypatch.delenv(name, raising=False) + + +def test_partial_process_transport_does_not_mix_with_dotenv( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + env_path = tmp_path / ".env" + _write_dotenv(env_path) + monkeypatch.setattr(interpretation_module, "_GEN_SIM_ENV_PATH", env_path) + monkeypatch.setattr( + interpretation_module, + "_GEN_CONFIG_PATH", + tmp_path / "missing.json", + ) + _clear_process_provider(monkeypatch) + monkeypatch.setenv("OPENAI_API_KEY", "unrelated-process-key") + + settings = interpretation_module._load_llm_settings(model=None) + + assert settings["api_key"] == "dotenv-key" + assert settings["base_url"] == "https://dotenv.example/v1" + assert settings["model"] == "dotenv-model" + + +def test_complete_process_transport_overrides_dotenv( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + env_path = tmp_path / ".env" + _write_dotenv(env_path) + monkeypatch.setattr(interpretation_module, "_GEN_SIM_ENV_PATH", env_path) + monkeypatch.setattr( + interpretation_module, + "_GEN_CONFIG_PATH", + tmp_path / "missing.json", + ) + _clear_process_provider(monkeypatch) + monkeypatch.setenv("OPENAI_API_KEY", "process-key") + monkeypatch.setenv("OPENAI_BASE_URL", "https://process.example/v1/") + monkeypatch.setenv("TASK_ENGINE_LLM_MODEL", "process-model") + + settings = interpretation_module._load_llm_settings(model=None) + + assert settings["api_key"] == "process-key" + assert settings["base_url"] == "https://process.example/v1" + assert settings["model"] == "process-model"