feat(task-program): configure planner from execution policy

agent_context/topics/task-programs/configuration.md

@@ -72,8 +72,10 @@ Ownership is explicit rather than a generic deep merge:
 - `configs/components/embodiments/*.yaml` owns simulation robot construction,
   the sensor suite, and an optional `skill_profile` containing logical
   resources/endpoints, command presets, and embodiment-specific services;
-- `configs/components/execution_policies/*.yaml` owns motion, tracking,
-  recovery, runner, and effect-assurance policy.
+- `configs/components/execution_policies/*.yaml` owns planner
+  selection/configuration, motion, tracking, recovery, runner, and
+  effect-assurance policy. Planner collision objects are attached from the
+  selected live scene and are not serialized in the policy.
 
 The reference embodiment `skill_profile.contract_id` and `profile_id` values
 are unversioned. Versioned Gym, task-integration, or scene-registry IDs are

embodichain/lab/task_program/integrations/_configured_composition.py

@@ -56,6 +56,7 @@
         "force_refresh",
     }
 )
+_PLANNER_TYPES = frozenset({"curobo", "neural", "toppra", "trapezoidal"})
 
 
 @dataclass(frozen=True, slots=True)
@@ -68,6 +69,7 @@ class _ConfiguredTaskProgramDeployment:
     selection: TaskProgramIntegrationCfg
     integration: _ConfiguredTaskProgramIntegration
     scene_binding: dict[str, object]
+    planner_config: dict[str, object] | None
 
 
 def _component_path(
@@ -113,6 +115,33 @@ def _load_yaml_component(path: Path, *, field_name: str) -> dict[str, object]:
     return _owned_mapping(load_config(path), path=field_name)
 
 
+def _decode_planner_config(value: object, *, path: str) -> dict[str, object]:
+    """Decode an executable-free planner selection from an execution policy."""
+    config = _mapping(
+        value,
+        path=path,
+        required=frozenset({"type"}),
+        optional=frozenset({"config"}),
+    )
+    planner_type = _identifier(config["type"], path=f"{path}.type")
+    if planner_type not in _PLANNER_TYPES:
+        raise ValueError(
+            f"{path}.type must be one of {sorted(_PLANNER_TYPES)}, "
+            f"got {planner_type!r}."
+        )
+    options = config.get("config", {})
+    if not isinstance(options, Mapping):
+        raise TypeError(f"{path}.config must be a mapping.")
+    planner_config = {
+        "type": planner_type,
+        "config": deepcopy(dict(options)),
+    }
+    from .simulation.environment import _planner_cfg_from_config
+
+    _planner_cfg_from_config(planner_config, robot_uid="__configured_policy__")
+    return planner_config
+
+
 def _merge_runtime_services(
     skill_profile_services: object,
     task_services: object,
@@ -270,9 +299,23 @@ def _resolve_task_program_components(
                 "effect_assurance",
             }
         ),
-        optional=frozenset({"required_planner"}),
+        optional=frozenset({"required_planner", "planner"}),
     )
     _identifier(policy["policy_id"], path="execution policy.policy_id")
+    if "planner" in policy:
+        policy["planner"] = _decode_planner_config(
+            policy["planner"],
+            path="execution policy.planner",
+        )
+        required_planner = policy.get("required_planner")
+        if (
+            required_planner is not None
+            and required_planner != policy["planner"]["type"]
+        ):
+            raise ValueError(
+                "execution policy.required_planner must match "
+                "execution policy.planner.type when both are declared."
+            )
     return program_path, integration, policy
 
 
@@ -347,7 +390,9 @@ def _compose_integration_payload(
         "effect_assurance": deepcopy(policy["effect_assurance"]),
         "effect_monitors": deepcopy(profile["effect_monitors"]),
     }
-    if "required_planner" in policy:
+    if "planner" in policy:
+        preset["required_planner"] = deepcopy(policy["planner"]["type"])
+    elif "required_planner" in policy:
         preset["required_planner"] = deepcopy(policy["required_planner"])
 
     scene_payload = {
@@ -416,7 +461,11 @@ def _load_configured_task_program_deployment(
         skill_profile=selected_skill_profile,
         scene=scene_binding,
     )
-    integration = _decode_configured_task_program_integration(payload)
+    planner_config = deepcopy(policy.get("planner"))
+    integration = _decode_configured_task_program_integration(
+        payload,
+        planner_config=planner_config,
+    )
     integration_id = _identifier(
         task["integration_id"],
         path="task integration.integration_id",
@@ -440,4 +489,5 @@ def _load_configured_task_program_deployment(
         selection=selection,
         integration=integration,
         scene_binding=deepcopy(dict(scene_binding)),
+        planner_config=planner_config,
     )

embodichain/lab/task_program/integrations/configured.py

@@ -2185,8 +2185,17 @@ class _ConfiguredTaskProgramIntegration:
 
 def _decode_configured_task_program_integration(
     value: object,
+    *,
+    planner_config: Mapping[str, object] | None = None,
 ) -> _ConfiguredTaskProgramIntegration:
-    """Decode one composable, callable-free Task Program integration."""
+    """Decode one composable, callable-free Task Program integration.
+
+    Args:
+        value: Provider-free scene, robot-profile, and runtime-service payload.
+        planner_config: Optional execution-policy planner declaration retained
+            for the live simulation adapter. It is deliberately kept outside
+            the semantic integration payload.
+    """
     path = "integration"
     config = _mapping(
         value,
@@ -2225,6 +2234,7 @@ def _decode_configured_task_program_integration(
     fingerprint_payload = {
         "registration": registration.fingerprint,
         "grasp_pose_generators": grasp_fingerprint,
+        "planner_config": deepcopy(planner_config),
     }
     integration_fingerprint = hashlib.sha256(
         json.dumps(
@@ -2238,6 +2248,7 @@ def _decode_configured_task_program_integration(
     delegate = SimulationTaskProgramAdapterFactory(
         registration,
         grasp_pose_generator_factories=dict(services.grasp_pose_generators),
+        planner_config=planner_config,
     )
     adapter_factory = _ConfiguredTaskProgramAdapterFactory(
         delegate=delegate,