feat(eval): integrate LIBERO-PRO code-policy benchmark

CONTEXT.md

@@ -8,6 +8,26 @@ The language used to describe how DimOS measures task-performing behavior across
 A complete benchmark integration that owns its environment lifecycle, cases, scoring, aggregation, and native result semantics. Third-party evaluations retain their original harness rather than translating it into DimOS scoring primitives.
 _Avoid_: Runtime, universal scorer
 
+**Comparable LIBERO-PRO Evaluation**:
+An Evaluation that preserves LIBERO-PRO's tasks, observations, action semantics, initial states, native success, and reporting while declaring any policy-runtime deviations that affect direct score comparison.
+_Avoid_: Official LIBERO-PRO evaluation, protocol-identical score
+
+**CaP-X-Aligned LIBERO Policy Surface**:
+The non-privileged observations and robot capabilities available to code policies in CaP-X's reduced LIBERO environment, expressed through ordinary DimOS modules and messages. It includes calibrated RGB-D cameras, camera-to-robot transforms, robot state, perception, inverse kinematics, and executable arm and gripper motion, but excludes simulator object poses and native success state.
+_Avoid_: CaP-X privileged API, simulator oracle, identical CaP-X implementation
+
+**Unified Real-Time Policy**:
+A Policy Artifact that uses the same ordinary DimOS interface in simulation and on a real robot while the environment continues advancing independently of policy computation latency.
+_Avoid_: Benchmark-stepped policy, simulator-specific policy
+
+**Real-Time Horizon**:
+The fixed number of simulator control ticks available to a Unified Real-Time Policy after startup and settling complete. Every tick consumes the horizon whether or not the policy produces a new command.
+_Avoid_: Policy-call budget, wall-clock timeout
+
+**LIBERO-PRO Trial**:
+One scored policy episode against a selected LIBERO-PRO task and fresh initial state, producing the benchmark's native result and a side-by-side MP4 of the public camera observations. A trial validates the integration but is not by itself a comparable aggregate benchmark score.
+_Avoid_: LIBERO-PRO benchmark result, aggregate score
+
 **Evaluation Case**:
 A fixed task, environment, budget, and scoring definition. It does not select the behavior being evaluated.
 _Avoid_: Policy configuration, run
@@ -25,21 +45,45 @@ The callable produced during exploration and replayed without an agent during ev
 _Avoid_: Agent, exploration transcript, execution mode
 
 **Policy Artifact**:
-The serialized callable and human-readable source captured when `submit_policy(policy)` is called. One artifact is produced per benchmark task and reused across its held-out evaluation cases or seeds. REPL outputs and the agent transcript are separate exploration evidence.
+The serialized callable and human-readable source captured when `submit_policy(policy)` is called. One artifact is produced per benchmark task and reused across fresh evaluation episodes or seeds of that task. REPL outputs and the agent transcript are separate exploration evidence.
 _Avoid_: Exploration transcript, policy source
 
 **Exploration Stage**:
-The unscored stage in which an agent uses a persistent Python REPL and calls `submit_policy(policy)` to run complete debug trials in fresh environments and blueprints. Model latency does not consume the evaluation horizon.
+The unscored training stage in which an agent receives the selected benchmark task's exact instruction, uses a persistent Python REPL, and calls `submit_policy(policy)` to run complete debug trials of that task in fresh environments and blueprints. It is part of the evaluated code-as-policy system, while model latency remains outside the evaluation horizon.
 _Avoid_: Evaluation rollout, scoring
 
+**Task-Aligned Exploration**:
+An Exploration Stage whose debug trials and scored Evaluation Stage use the same benchmark suite, task identity, and task instruction. Episodes and initial states are fresh, but the policy is not asked to transfer to an undisclosed task.
+_Avoid_: Hidden-task evaluation, unperturbed training suite
+
+**Debug Trial**:
+One unscored policy attempt created by `submit_policy()`, owning a fresh complete DimOS blueprint and a fresh episode of the selected benchmark task. It returns a stopped `TrialRun` snapshot containing the outcome, logs, Memory2 recording, and artifacts.
+_Avoid_: Shared simulator reset, partial run, evaluation rollout
+
 **Evaluation Stage**:
-The measured stage in which the Policy Artifact executes without an agent against a reset or held-out environment. The native benchmark owns its real-time or step horizon and privileged scoring.
+The measured stage in which the Policy Artifact executes without an agent in a fresh episode of the same suite and task disclosed during Task-Aligned Exploration. The task instruction is unchanged, the episode initial state is fresh, and the evaluation owns the declared real-time horizon while native privileged scoring remains benchmark-owned and unavailable to the policy.
 _Avoid_: Agent session, policy generation
 
 **Policy Environment**:
-The capabilities exposed by the fresh policy-only DimOS blueprint while a policy runs. Simulated, live, and replay-backed evaluations all pass the policy a connected `Dimos` application; completed trials additionally expose their Memory2 recording read-only through `TrialRun`.
+The capabilities exposed by the fresh policy-only DimOS blueprint while a policy runs. Simulated, live, and replay-backed evaluations all pass the policy a connected `Dimos` application whose `app.memory` property is the active read-only Memory2 store. Completed trials expose the same recording through `TrialRun`.
 _Avoid_: Agent tools, scorer context
 
+**RGB-D Observation**:
+A timestamp-aligned bundle of one public color image, metric depth image, exact camera calibration, and world-to-optical transform selected from Memory2. It is sensor evidence chosen by the policy, not a persistent scene model or simulator snapshot.
+_Avoid_: Scene façade, latest-camera RPC, simulator observation
+
+**Grounded Segmentation**:
+An ordinary on-demand DimOS perception capability that accepts an explicit timestamped image and text descriptions and returns typed image masks. It has no camera stream subscription and can operate identically on live, simulated, replayed, or post-trial images.
+_Avoid_: LIBERO perception, implicit latest frame, privileged object mask
+
+**Policy Interface**:
+The non-privileged robot observations, state, commands, and operational health that a simulator or real robot connection contributes to the complete Policy Environment.
+_Avoid_: Normal API, evaluator interface
+
+**Evaluation Control Interface**:
+The evaluator-only capabilities for environment configuration, reset, initial-state selection, clock control, native terminal state, and scoring. It is not part of the Policy Environment.
+_Avoid_: Privileged API, robot interface
+
 **Evaluation Oracle**:
 The evaluator-only source of truth used to score a policy attempt. In simulation it contains privileged state, such as true poses and object identities, that the Policy Environment cannot access.
 _Avoid_: Runtime memory, perception output

dimos/agents/code_policy_core.py

@@ -144,9 +144,10 @@ def _bootstrap_source(environment: SubmissionEnvironment | LiveDimosEnvironment)
 from dimos.memory2.store.sqlite import SqliteStore
 from dimos.porcelain.dimos import Dimos
 
-memory = SqliteStore(path={environment.recording_path!r}, must_exist=True, read_only=True)
-memory.start()
-app = Dimos.connect()
+app = Dimos.connect(
+    memory=SqliteStore(path={environment.recording_path!r}, must_exist=True, read_only=True)
+)
+app.memory.start()
 """
 
 

dimos/agents/test_code_policy_core.py

@@ -111,8 +111,10 @@ def test_live_repl_bootstraps_public_runtime_without_credentials(
     source = _bootstrap_source(environment)
     kernel_environment = _kernel_environment(config)
 
-    assert "memory = SqliteStore" in source
-    assert "app = Dimos.connect()" in source
+    assert "app = Dimos.connect(" in source
+    assert "memory=SqliteStore" in source
+    assert "\nmemory =" not in source
+    assert "app.memory.start()" in source
     assert "submit_policy" not in source
     assert "OPENAI_API_KEY" not in kernel_environment
     assert kernel_environment["ORDINARY_SETTING"] == "retained"

dimos/benchmark/evaluation/protocol.py

@@ -46,7 +46,7 @@ class PolicyArtifact:
 
 @dataclass(frozen=True)
 class PolicyExecution:
-    status: Literal["completed", "policy_error", "timed_out", "infrastructure_error"]
+    status: Literal["completed", "policy_error", "stopped", "infrastructure_error"]
     duration_seconds: float
     error: str | None = None
 
@@ -165,12 +165,22 @@ def explore(
         max_submissions: int = 5,
     ) -> ExplorationOutcome: ...
 
-    def execute(
+    def prepare(
         self,
         policy: PolicyArtifact,
         *,
-        timeout_s: float,
-    ) -> PolicyExecution: ...
+        memory_path: Path,
+        startup_timeout_s: float,
+    ) -> PolicyExecutionHandle: ...
+
+
+@runtime_checkable
+class PolicyExecutionHandle(Protocol):
+    """A loaded policy waiting behind an evaluator-owned start gate."""
+
+    def start(self) -> None: ...
+
+    def finish(self, *, grace_s: float = 1.0) -> PolicyExecution: ...
 
 
 @dataclass(frozen=True)

dimos/benchmark/evaluation/registry.py

@@ -30,6 +30,7 @@
 ENTRY_POINT_GROUP = "dimos.evaluations"
 LOCAL_NAME_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
 BUILTIN_EVALUATIONS: dict[str, str] = {
+    "libero-pro": "dimos.benchmark.libero_pro.evaluation:libero_pro",
     "vlnce-r2r": "dimos.benchmark.vlnce_r2r.evaluation:vlnce_r2r",
 }
 

dimos/benchmark/evaluation/runtime.py

@@ -20,7 +20,6 @@
 import json
 import multiprocessing
 from pathlib import Path
-import queue
 import shutil
 import time
 from typing import Any, Literal
@@ -34,6 +33,7 @@
     ExplorationOutcome,
     PolicyArtifact,
     PolicyExecution,
+    PolicyExecutionHandle,
     TrialRun,
 )
 
@@ -195,52 +195,41 @@ def explore(
             server.stop()
             shutil.rmtree(path / "working", ignore_errors=True)
 
-    def execute(self, policy: PolicyArtifact, *, timeout_s: float) -> PolicyExecution:
-        if timeout_s <= 0:
-            raise ValueError("timeout_s must be positive")
-        started = time.monotonic()
+    def prepare(
+        self,
+        policy: PolicyArtifact,
+        *,
+        memory_path: Path,
+        startup_timeout_s: float,
+    ) -> PolicyExecutionHandle:
+        if startup_timeout_s <= 0:
+            raise ValueError("startup_timeout_s must be positive")
         context = multiprocessing.get_context("spawn")
-        result_queue = context.Queue(maxsize=1)
+        messages, worker_messages = context.Pipe(duplex=False)
+        start_event = context.Event()
         process = context.Process(
             target=_execute_policy_worker,
-            args=(str(policy.serialized_path), result_queue),
+            args=(str(policy.serialized_path), str(memory_path), worker_messages, start_event),
             daemon=True,
         )
         try:
             process.start()
-            process.join(timeout_s)
+            worker_messages.close()
+            if not messages.poll(startup_timeout_s):
+                _stop_process(process)
+                raise TimeoutError(f"policy startup timed out after {startup_timeout_s:g}s")
+            message, error = messages.recv()
+            if message != "ready":
+                _stop_process(process)
+                messages.close()
+                raise RuntimeError(error or "policy worker failed before readiness")
+            return _PolicyExecutionProcess(process, messages, start_event)
+        except BaseException:
             if process.is_alive():
-                process.terminate()
-                process.join(5)
-                if process.is_alive():
-                    process.kill()
-                    process.join()
-                return PolicyExecution(
-                    status="timed_out",
-                    duration_seconds=time.monotonic() - started,
-                    error=f"policy timed out after {timeout_s:g}s",
-                )
-            try:
-                status, error = result_queue.get_nowait()
-            except queue.Empty:
-                return PolicyExecution(
-                    status="infrastructure_error",
-                    duration_seconds=time.monotonic() - started,
-                    error=f"policy worker exited with code {process.exitcode} without a result",
-                )
-            return PolicyExecution(
-                status=status,
-                duration_seconds=time.monotonic() - started,
-                error=error,
-            )
-        except Exception as exc:
-            return PolicyExecution(
-                status="infrastructure_error",
-                duration_seconds=time.monotonic() - started,
-                error=f"{type(exc).__name__}: {exc}",
-            )
-        finally:
-            result_queue.close()
+                _stop_process(process)
+            messages.close()
+            worker_messages.close()
+            raise
 
     def _record_prompt(
         self,
@@ -401,36 +390,108 @@ def submit(self, source: str, serialized: bytes) -> TrialRun:
         return trial
 
 
-def _execute_policy_worker(serialized_path: str, result_queue: Any) -> None:
+def _execute_policy_worker(
+    serialized_path: str,
+    memory_path: str,
+    messages: Any,
+    start_event: Any,
+) -> None:
     try:
-        import cloudpickle
+        try:
+            import cloudpickle
 
-        with Path(serialized_path).open("rb") as handle:
-            policy = cloudpickle.load(handle)
-    except Exception as exc:
-        result_queue.put(
-            ("infrastructure_error", f"policy load failed: {type(exc).__name__}: {exc}")
-        )
-        return
-    try:
-        from dimos.porcelain.dimos import Dimos
+            with Path(serialized_path).open("rb") as handle:
+                policy = cloudpickle.load(handle)
+        except Exception as exc:
+            messages.send(
+                ("infrastructure_error", f"policy load failed: {type(exc).__name__}: {exc}")
+            )
+            return
+        try:
+            from dimos.memory2.store.sqlite import SqliteStore
+            from dimos.porcelain.dimos import Dimos
 
-        app = Dimos.connect()
-    except Exception as exc:
-        result_queue.put(
-            ("infrastructure_error", f"DimOS connection failed: {type(exc).__name__}: {exc}")
+            memory = SqliteStore(path=memory_path, must_exist=True, read_only=True)
+            memory.start()
+            app = Dimos.connect(memory=memory)
+        except Exception as exc:
+            messages.send(
+                ("infrastructure_error", f"DimOS connection failed: {type(exc).__name__}: {exc}")
+            )
+            return
+        messages.send(("ready", None))
+        start_event.wait()
+        try:
+            result = policy(app)
+            if result is not None:
+                raise TypeError("policy(app) must return None")
+        except Exception as exc:
+            messages.send(("policy_error", f"{type(exc).__name__}: {exc}"))
+        else:
+            messages.send(("completed", None))
+        finally:
+            app.stop()
+    finally:
+        messages.close()
+
+
+class _PolicyExecutionProcess:
+    def __init__(self, process: Any, messages: Any, start_event: Any) -> None:
+        self._process = process
+        self._messages = messages
+        self._start_event = start_event
+        self._started_at: float | None = None
+        self._finished: PolicyExecution | None = None
+
+    def start(self) -> None:
+        if self._started_at is not None:
+            raise RuntimeError("policy execution already started")
+        self._started_at = time.monotonic()
+        self._start_event.set()
+
+    def finish(self, *, grace_s: float = 1.0) -> PolicyExecution:
+        if grace_s < 0:
+            raise ValueError("grace_s must be non-negative")
+        if self._finished is not None:
+            return self._finished
+        if self._started_at is None:
+            raise RuntimeError("policy execution has not started")
+        if self._messages.poll(grace_s):
+            try:
+                status, error = self._messages.recv()
+            except EOFError:
+                status = "infrastructure_error"
+                error = "policy worker closed its result channel without a result"
+            self._process.join(grace_s)
+            if self._process.is_alive():
+                _stop_process(self._process)
+        else:
+            stopped = self._process.is_alive()
+            if stopped:
+                _stop_process(self._process)
+                status, error = "stopped", None
+            else:
+                self._process.join()
+                status = "infrastructure_error"
+                error = f"policy worker exited with code {self._process.exitcode} without a result"
+        self._messages.close()
+        self._finished = PolicyExecution(
+            status=status,
+            duration_seconds=time.monotonic() - self._started_at,
+            error=error,
         )
+        return self._finished
+
+
+def _stop_process(process: Any) -> None:
+    if not process.is_alive():
+        process.join()
         return
-    try:
-        result = policy(app)
-        if result is not None:
-            raise TypeError("policy(app) must return None")
-    except Exception as exc:
-        result_queue.put(("policy_error", f"{type(exc).__name__}: {exc}"))
-    else:
-        result_queue.put(("completed", None))
-    finally:
-        app.stop()
+    process.terminate()
+    process.join(5)
+    if process.is_alive():
+        process.kill()
+        process.join()
 
 
 def _assemble_user_message(evaluation_protocol: str, task_input: str) -> str: