diff --git a/robots/libero/__init__.py b/robots/libero/__init__.py index a5339a07..7bf9323e 100644 --- a/robots/libero/__init__.py +++ b/robots/libero/__init__.py @@ -36,6 +36,8 @@ def get_env_spec() -> EnvSpec: ), add_cli_args=_add_cli_args, parse_config=_parse_config, + init_shared_runtime=init_shared_runtime, + init_task_runtime=init_task_runtime, init_runtime=_init_runtime, ) @@ -45,7 +47,6 @@ def get_toolkit( primitives_kwargs: dict[str, Any], dashboard_events: DashboardEventSink, video_path: str | None = None, - save_action_videos: bool = False, ): """Return the LIBERO toolkit (common tools + LIBERO primitives).""" from robots.libero.toolkit import LiberoToolkit @@ -54,7 +55,6 @@ def get_toolkit( primitives_kwargs=primitives_kwargs, dashboard_events=dashboard_events, video_path=video_path, - save_action_videos=save_action_videos, ) @@ -136,6 +136,212 @@ def _subprocess_env(**extra: str) -> dict[str, str]: return env +def init_task_runtime( + args: argparse.Namespace, + output_dir: Path, + dashboard_events: DashboardEventSink, +) -> tuple[list[ProcessDaemon], dict[str, Any]]: + """Initialize one TaskRun-owned LIBERO environment. + + A local env server is fresh for every call. When ``--env-endpoint`` is + supplied, the returned daemon list is empty so the external service stays + running. + + Heavy runtime dependencies stay lazy so importing :mod:`robots.libero` + for its descriptor or toolkit does not load RPC/model packages. + """ + from robots.libero.env_client import LiberoEnvClient + from rpent.utils.config import get_libero_type + from rpent.utils.daemon import ProcessDaemon, pick_free_port + from rpent.utils.http_rpc import HttpRpcClient + from rpent.utils.rpc import parse_endpoint, wait_for_ready + from rpent.utils.socket_rpc import SocketRpcClient + + owned_daemons: list[ProcessDaemon] = [] + libero_type = args.libero_type or get_libero_type() + cuda_args = ["--cuda-device", str(args.cuda_device)] if args.cuda_device is not None else [] + + dashboard_events.emit(RuntimeStatusEvent("env", "starting")) + try: + env_daemon: ProcessDaemon | None = None + if args.env_endpoint is None: + host, port = "127.0.0.1", pick_free_port() + env_daemon = ProcessDaemon( + name="env_server", + cmd=[ + sys.executable, + str(get_repo_root() / "robots" / "libero" / "env_server.py"), + "--suite", args.suite, + "--task", str(args.task), + "--seed", str(args.seed), + "--max-episode-steps", str(args.max_episode_steps), + "--transport", "http", + "--host", host, + "--port", str(port), + "--parent-watch", + *cuda_args, + ], + env=_subprocess_env( + LIBERO_TYPE=libero_type, + MUJOCO_GL="egl", + ROBOT_PLATFORM="LIBERO", + ), + log_path=str(Path(output_dir) / "env_server.log"), + ) + env_daemon.start() + owned_daemons.append(env_daemon) + env_rpc: RpcClient = HttpRpcClient(f"http://{host}:{port}") + else: + protocol, host, port = parse_endpoint(args.env_endpoint) + if protocol == "socket": + env_rpc = SocketRpcClient(host, port) + elif protocol == "http": + env_rpc = HttpRpcClient(f"http://{host}:{port}") + else: + raise ValueError( + f"--env-endpoint protocol must be socket or http, got {protocol!r}" + ) + wait_for_ready(env_rpc, daemon=env_daemon) + env = LiberoEnvClient( + env_rpc, + expected_meta={ + "suite": args.suite, + "task": args.task, + "seed": args.seed, + "max_episode_steps": args.max_episode_steps, + }, + ) + except Exception as exc: + _stop_owned_daemons(owned_daemons) + dashboard_events.emit(RuntimeStatusEvent("env", "failed", error=exc)) + raise + dashboard_events.emit(RuntimeStatusEvent("env", "ready")) + return owned_daemons, {"env": env} + + +def init_shared_runtime( + args: argparse.Namespace, + output_dir: Path, + dashboard_events: DashboardEventSink, +) -> tuple[list[ProcessDaemon], dict[str, Any]]: + """Initialize Session-owned VLA and SAM3 services. + + The returned list contains only locally started services. External + endpoints are connected to but never become owned. + """ + from rpent.utils.daemon import ProcessDaemon, pick_free_port + from rpent.utils.http_rpc import HttpRpcClient + from rpent.utils.rpc import parse_endpoint, wait_for_ready + from rpent.utils.sam3_client import Sam3Client + from rpent.utils.socket_rpc import SocketRpcClient + from rpent.utils.vla_client import VLAClient + + owned_daemons: list[ProcessDaemon] = [] + cuda_args = ( + ["--cuda-device", str(args.cuda_device)] + if args.cuda_device is not None + else [] + ) + + # --- vla_server -------------------------------------------------------- + dashboard_events.emit(RuntimeStatusEvent("vla", "starting")) + try: + vla_daemon: ProcessDaemon | None = None + if args.vla_endpoint is None: + host, port = "127.0.0.1", pick_free_port() + vla_daemon = ProcessDaemon( + name="vla_server", + cmd=[ + sys.executable, + str(get_repo_root() / "robots" / "libero" / "vla_server.py"), + "--transport", "http", + "--host", host, + "--port", str(port), + "--parent-watch", + *cuda_args, + ], + env=_subprocess_env(), + log_path=str(Path(output_dir) / "vla_server.log"), + ) + vla_daemon.start() + owned_daemons.append(vla_daemon) + vla_rpc: RpcClient = HttpRpcClient(f"http://{host}:{port}") + else: + protocol, host, port = parse_endpoint(args.vla_endpoint) + if protocol == "socket": + vla_rpc = SocketRpcClient(host, port) + elif protocol == "http": + vla_rpc = HttpRpcClient(f"http://{host}:{port}") + else: + raise ValueError( + f"--vla-endpoint protocol must be socket or http, got {protocol!r}" + ) + except Exception as exc: + _stop_owned_daemons(owned_daemons) + dashboard_events.emit(RuntimeStatusEvent("vla", "failed", error=exc)) + raise + + # --- sam3_server ------------------------------------------------------- + dashboard_events.emit(RuntimeStatusEvent("sam3", "starting")) + try: + sam3_daemon: ProcessDaemon | None = None + if args.sam3_endpoint is None: + host, port = "127.0.0.1", pick_free_port() + sam3_daemon = ProcessDaemon( + name="sam3_server", + cmd=[ + sys.executable, + str(get_repo_root() / "robots" / "libero" / "sam3_server.py"), + "--transport", "http", + "--host", host, + "--port", str(port), + "--parent-watch", + *cuda_args, + ], + env=_subprocess_env(), + log_path=str(Path(output_dir) / "sam3_server.log"), + ) + sam3_daemon.start() + owned_daemons.append(sam3_daemon) + sam3_rpc: RpcClient = HttpRpcClient(f"http://{host}:{port}") + else: + protocol, host, port = parse_endpoint(args.sam3_endpoint) + if protocol == "socket": + sam3_rpc = SocketRpcClient(host, port) + elif protocol == "http": + sam3_rpc = HttpRpcClient(f"http://{host}:{port}") + else: + raise ValueError( + f"--sam3-endpoint protocol must be socket or http, got {protocol!r}" + ) + except Exception as exc: + _stop_owned_daemons(owned_daemons) + dashboard_events.emit(RuntimeStatusEvent("sam3", "failed", error=exc)) + raise + + # Start both local services before waiting so heavyweight initialization + # continues concurrently, matching the one-shot runtime behavior. + for component, client, daemon in ( + ("sam3", sam3_rpc, sam3_daemon), + ("vla", vla_rpc, vla_daemon), + ): + try: + wait_for_ready(client, daemon=daemon) + except Exception as exc: + _stop_owned_daemons(owned_daemons) + dashboard_events.emit(RuntimeStatusEvent(component, "failed", error=exc)) + raise + dashboard_events.emit(RuntimeStatusEvent(component, "ready")) + + model = VLAClient(vla_rpc) + sam3_client = Sam3Client(sam3_rpc) + + return owned_daemons, { + "model": model, + "sam3_client": sam3_client, + } + + def _init_runtime( args: argparse.Namespace, output_dir: Path, @@ -312,3 +518,12 @@ def _init_runtime( "sam3_client": Sam3Client(sam3_rpc), } return daemons, primitives_kwargs + + +def _stop_owned_daemons(daemons: list[ProcessDaemon]) -> None: + """Stop owned daemons in reverse order without masking startup errors.""" + for daemon in reversed(daemons): + try: + daemon.stop() + except Exception: + pass diff --git a/robots/libero/toolkit.py b/robots/libero/toolkit.py index f62521a4..475c1e74 100644 --- a/robots/libero/toolkit.py +++ b/robots/libero/toolkit.py @@ -12,7 +12,7 @@ from robots.libero import tools as libero_tools from rpent.dashboard.events import DashboardEventSink, ToolResultEvent -from rpent.tools.toolkit import Toolkit +from rpent.tools.toolkit import ToolCancelled, Toolkit from rpent.utils.logging import get_logger, get_output_dir @@ -29,12 +29,10 @@ def __init__( primitives_kwargs: dict[str, Any], dashboard_events: DashboardEventSink, video_path: str | None = None, - save_action_videos: bool = False, ) -> None: super().__init__(dashboard_events=dashboard_events) self._next_step: int = 0 self._video_path: str | None = video_path - self._save_action_videos = save_action_videos self.init_primitives_clean(primitives_kwargs=primitives_kwargs) self._register_libero_tools() @@ -65,7 +63,15 @@ def _step(self, name: str, **kwargs) -> dict: command = {"action": name, **kwargs} t0 = time.time() start_frame = self._primitives.recorded_frame_count() - result = getattr(self._primitives, name)(**kwargs) + try: + result = getattr(self._primitives, name)(**kwargs) + self.raise_if_cancelled() + except ToolCancelled as exc: + result = { + "error": str(exc), + "code": "tool_cancelled", + "interrupted": True, + } elapsed = round(time.time() - t0, 2) if isinstance(result, dict): @@ -76,7 +82,7 @@ def _step(self, name: str, **kwargs) -> dict: self._next_step += 1 step_idx = self._next_step output_dir = get_output_dir() - if self._save_action_videos: + if self._dashboard_events.enabled: video_dir = libero_tools.artifact_path(output_dir, "action_videos") video_path = video_dir / f"step_{step_idx:02d}_{name}.mp4" try: @@ -93,6 +99,8 @@ def _step(self, name: str, **kwargs) -> dict: ) out = libero_tools.view_driver_state(step_idx) out["agent_elapsed_s"] = elapsed + if result_dict.get("interrupted"): + out.update(result_dict) return out def init_primitives_clean( @@ -115,7 +123,10 @@ def init_primitives_clean( if target.exists(): target.unlink() - primitives = libero_tools.LiberoPrimitives(**primitives_kwargs) + primitives = libero_tools.LiberoPrimitives( + check_cancelled=self.raise_if_cancelled, + **primitives_kwargs, + ) primitives.reset() primitives.start_recording() libero_tools.dump_state(primitives, str(out_dir), step_idx=0, log=None) diff --git a/robots/libero/tools.py b/robots/libero/tools.py index c6e7ab62..44145227 100644 --- a/robots/libero/tools.py +++ b/robots/libero/tools.py @@ -3,6 +3,7 @@ import json import os +from collections.abc import Callable from pathlib import Path from typing import Any @@ -104,10 +105,12 @@ def __init__( env: LiberoEnvClient, model: VLAClient, sam3_client: Sam3Client, + check_cancelled: Callable[[], None], ): self.env = env self.model = model self._sam3_client = sam3_client + self._check_cancelled = check_cancelled self._last_obs = None self._last_obs_eef_pos = None self._last_obs_eef_z = None @@ -162,7 +165,8 @@ def reset(self): return self._last_obs, info def _step_env(self, action) -> None: - """Execute one action and update the cached observation and video.""" + """Execute one env action between cancellation checkpoints.""" + self._check_cancelled() obs, _r, _t, _tr, _i = self.env.step(action) self.set_obs(obs) if self._recording: @@ -170,31 +174,30 @@ def _step_env(self, action) -> None: def _vlm_chunk(self, instruction: str): """One model forward + ``chunk_size`` env steps. Overrides prompt.""" - # Stash & override task_descriptions (one prompt). - original_td = self._last_obs.get("task_descriptions") - self._last_obs["task_descriptions"] = instruction - self._last_obs.setdefault("extra_view_images", None) - - actions, _ = self.model.predict_action_batch(self._last_obs, mode="eval") - # actions: [chunk_size, action_dim] The whole chunk - # runs in a single env.chunk_step RPC; the env owns the per-step - # loop server-side. - if not self._recording: - chunk_obs, _r, _t, _tr, _i = self.env.chunk_step(actions) - obs = chunk_obs[-1] if self.env.return_all_frames else chunk_obs - else: - chunk_obs, _r, _t, _tr, _i = self.env.chunk_step( - actions, return_all_frames=True - ) - for obs in chunk_obs: - self.record_frame(obs) - obs = chunk_obs[-1] - self.set_obs(obs) - # Restore original task_descriptions on the obs dict for fairness - # with future steps (no leaked state if caller switches primitives). - if original_td is not None: - self._last_obs["task_descriptions"] = original_td - return self._last_obs + self._check_cancelled() + original_task = self._last_obs.get("task_descriptions") + try: + self._last_obs["task_descriptions"] = instruction + self._last_obs.setdefault("extra_view_images", None) + + actions, _ = self.model.predict_action_batch(self._last_obs, mode="eval") + self._check_cancelled() + + if not self._recording: + chunk_obs, _r, _t, _tr, _i = self.env.chunk_step(actions) + obs = chunk_obs[-1] if self.env.return_all_frames else chunk_obs + else: + chunk_obs, _r, _t, _tr, _i = self.env.chunk_step( + actions, return_all_frames=True + ) + for obs in chunk_obs: + self.record_frame(obs) + obs = chunk_obs[-1] + self.set_obs(obs) + return self._last_obs + finally: + if original_task is not None: + self._last_obs["task_descriptions"] = original_task def pi0_pick( self, diff --git a/rpent/cli/dashboard.py b/rpent/cli/dashboard.py new file mode 100644 index 00000000..a367820b --- /dev/null +++ b/rpent/cli/dashboard.py @@ -0,0 +1,232 @@ +"""CLI orchestration for one long-lived Dashboard Session.""" + +from __future__ import annotations + +import argparse +import copy +import json +import shlex +import sys +import threading +import time +from datetime import datetime +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from rpent.cli.main import _serialize_messages +from rpent.dashboard.events import RunStartedEvent +from rpent.envs import get_toolkit +from rpent.planner.base import build_planner +from rpent.utils.logging import get_logger, init_output_dir +from rpent.utils.resources import ensure_resources + +if TYPE_CHECKING: + from rpent.dashboard.state import ClaimedTask, DashboardState + from rpent.envs.env_spec import EnvSpec + from rpent.utils.daemon import ProcessDaemon + +logger = get_logger("agent") + + +def run_dashboard_session( + args: argparse.Namespace, + env_spec: EnvSpec, + *, + parser: argparse.ArgumentParser, +) -> int: + """Run one long-lived Dashboard Session with sequential fresh TaskRuns.""" + from rpent.dashboard.launcher import apply_to_args, defaults_from_args + from rpent.dashboard.server import DashboardServer + from rpent.dashboard.session import DashboardSessionController + from rpent.dashboard.state import DashboardState + from rpent.utils.config import get_repo_root + + dashboard_server = DashboardServer( + host=args.dashboard_host, + port=args.dashboard_port, + language=args.dashboard_language, + ) + dashboard_url = dashboard_server.start() + print( + f"Dashboard: {dashboard_url}. Open it, adjust the Session config, " + "and click Start Session.", + flush=True, + ) + launch_config = dashboard_server.wait_for_launch( + defaults=defaults_from_args(args) + ) + apply_to_args(args, launch_config) + + if args.env_endpoint is not None: + parser.error( + "Dashboard task control cannot use --env-endpoint because each " + "TaskRun requires a fresh owned env_server" + ) + + if args.output_dir is None: + timestamp = datetime.now().strftime("%Y%m%d-%H:%M:%S") + session_root = get_repo_root() / "logs" / f"{timestamp}_dashboard_session" + else: + session_root = Path(args.output_dir) + session_root = init_output_dir(session_root, verbose=args.verbose) + logger.info("Dashboard: %s", dashboard_url) + logger.info("launcher Session config applied: %s", launch_config) + logger.info("physical agent cmd: %s", shlex.join([sys.executable, *sys.argv])) + + ensure_resources(args.env_name) + state = DashboardState( + run_id=f"dashboard-session/{session_root.name}", + output_dir=session_root, + ) + dashboard_server.register(state) + + controller = DashboardSessionController( + state=state, + start_shared=lambda: env_spec.init_shared_runtime( + args, + session_root, + state, + ), + run_task=lambda claimed, shared: _run_dashboard_task( + args=args, + env_spec=env_spec, + state=state, + claimed=claimed, + shared=shared, + session_root=session_root, + ), + ) + try: + controller.run() + if state.session_state == "fatal": + logger.error( + "Dashboard Session is fatal. Still serving at %s; " + "press Ctrl+C to stop.", + dashboard_url, + ) + threading.Event().wait() + except KeyboardInterrupt: + state.request_shutdown() + return 0 + + +def _run_dashboard_task( + *, + args: argparse.Namespace, + env_spec: EnvSpec, + state: DashboardState, + claimed: ClaimedTask, + shared_primitives_kwargs: dict[str, Any], + session_root: Path, +) -> str | None: + """Execute one fresh Dashboard TaskRun against Session-owned services.""" + task_args = copy.copy(args) + task_args.suite = claimed.command.suite + task_args.task = claimed.command.task + task_args.seed = claimed.command.seed + task_args.output_dir = str(claimed.output_dir) + run_config = env_spec.parse_config(task_args) + output_dir = init_output_dir(run_config.output_dir, verbose=args.verbose) + + recipe_tag = run_config.recipe_tag + finish_result = None + messages: list[dict] = [] + stats: dict = {} + agent_error: str | None = None + task_daemons: list[ProcessDaemon] = [] + toolkit = None + started = time.time() + try: + task_daemons, task_primitives_kwargs = env_spec.init_task_runtime( + task_args, + output_dir, + state, + ) + if not state.task_replacement_requested: + primitives_kwargs = { + **task_primitives_kwargs, + **shared_primitives_kwargs, + } + toolkit = get_toolkit( + args.env_name, + primitives_kwargs=primitives_kwargs, + video_path=str(output_dir / "episode.mp4"), + dashboard_events=state, + ) + planner = build_planner( + args.planner, + output_dir=output_dir, + recipe_tag=recipe_tag, + env_name=args.env_name, + base_url=args.base_url, + model=args.model, + max_tokens=args.max_tokens, + planner_timeout_s=args.planner_timeout_s, + claude_code_max_budget_usd=args.claude_code_max_budget_usd, + dashboard_events=state, + no_images=args.no_images, + ) + + prompt_vars = {**run_config.prompt_vars, "output_dir": output_dir} + system_prompt = env_spec.prompts.render( + "system", + variables=prompt_vars, + ) + user_message = env_spec.prompts.render( + "user", + variables=prompt_vars, + ) + if not state.task_replacement_requested: + state.emit(RunStartedEvent()) + result = planner.solve( + system_prompt=system_prompt, + user_message=user_message, + toolkit=toolkit, + max_turns=args.max_turns, + dashboard_interaction=state, + ) + finish_result = result.finish_result + messages = result.messages + stats = result.stats + agent_error = result.error + except Exception as exc: + logger.error("EXCEPTION in Dashboard TaskRun %04d: %s", claimed.number, exc) + agent_error = str(exc) + finally: + cleanup_errors: list[str] = [] + if toolkit is not None: + try: + toolkit.close() + recipe_path = toolkit.write_recipe(recipe_tag) + logger.info("recipe: %s", recipe_path) + except Exception as exc: + cleanup_errors.append(f"Toolkit cleanup failed: {exc}") + for daemon in reversed(task_daemons): + try: + daemon.stop() + except Exception as exc: + cleanup_errors.append(f"env cleanup failed: {exc}") + if cleanup_errors: + cleanup_error = "; ".join(cleanup_errors) + if agent_error is None: + agent_error = cleanup_error + else: + logger.warning("%s", cleanup_error) + + transcript_path = output_dir / f"transcript_{run_config.recipe_tag}.json" + record = { + **run_config.task_desc, + "model": args.model, + "elapsed_s": round(time.time() - started, 1), + "finish": finish_result, + "stats": stats, + "messages": _serialize_messages(messages), + } + try: + with open(transcript_path, "a") as transcript_file: + json.dump(record, transcript_file, indent=2, default=str) + except Exception as exc: + logger.warning("failed to write TaskRun transcript %s: %s", transcript_path, exc) + init_output_dir(session_root, verbose=args.verbose) + + return agent_error diff --git a/rpent/cli/main.py b/rpent/cli/main.py index 37c2478b..2596793d 100644 --- a/rpent/cli/main.py +++ b/rpent/cli/main.py @@ -25,7 +25,6 @@ import queue import shlex import sys -import threading import time from collections.abc import Callable from pathlib import Path @@ -35,12 +34,9 @@ start_interactive_reader, ) from rpent.dashboard.events import ( - DashboardEventSink, NullDashboardEventSink, - RunFinishedEvent, RunStartedEvent, ) -from rpent.dashboard.interaction import DashboardInteractionPort from rpent.envs import get_env_spec, get_toolkit from rpent.planner.base import build_planner from rpent.utils.logging import get_logger, init_output_dir @@ -150,34 +146,10 @@ def main() -> int: args = parser.parse_args() if args.dashboard and args.interactive: parser.error("--dashboard and --interactive cannot be used together") - - # With --dashboard, open the launcher first: serve the start screen, then - # block until the user clicks Run and overlay their choices onto args. - # parse_config runs afterwards so validation + derivation see the final - # config. - dashboard_server = None - dashboard_url = None - launch_config = None if args.dashboard: - from rpent.dashboard.launcher import apply_to_args, defaults_from_args - from rpent.dashboard.server import DashboardServer + from rpent.cli.dashboard import run_dashboard_session - dashboard_server = DashboardServer( - host=args.dashboard_host, port=args.dashboard_port, - language=args.dashboard_language, - ) - dashboard_url = dashboard_server.start() - # The run directory is not final until the launcher form is submitted, so - # print the pre-launch URL without initializing the run.log file handler. - print( - f"Dashboard: {dashboard_url}. " - "Open it, adjust the run config, and click Run to start.", - flush=True, - ) - launch_config = dashboard_server.wait_for_launch( - defaults=defaults_from_args(args) - ) - apply_to_args(args, launch_config) + return run_dashboard_session(args, env_spec, parser=parser) run_config = env_spec.parse_config(args) recipe_tag = run_config.recipe_tag @@ -189,29 +161,11 @@ def main() -> int: # mkdir + logging wiring (env-side already picked the path). output_dir = init_output_dir(output_dir, verbose=args.verbose) - # Now that output_dir is fixed, repeat launcher details into this run's log. - if dashboard_url is not None: - logger.info("Dashboard: %s", dashboard_url) - if launch_config is not None: - logger.info("launcher config applied: %s", launch_config) logger.info("physical agent cmd: %s", shlex.join([sys.executable, *sys.argv])) ensure_resources(env_name) - # --- dashboard state --------------------------------------------------- - dashboard_events: DashboardEventSink = NullDashboardEventSink() - dashboard_interaction: DashboardInteractionPort | None = None - if dashboard_server is not None: - from rpent.dashboard.state import DashboardState - - state = DashboardState.from_run_config(run_config) - if args.planner == "claude_code": - state.enable_interaction(session_id=state.run_id) - dashboard_interaction = state - # Server is already serving the launcher; register the run so the - # frontend can switch from the start screen to the live monitor. - dashboard_server.register(state) - dashboard_events = state + dashboard_events = NullDashboardEventSink() planner = build_planner( args.planner, @@ -267,7 +221,6 @@ def main() -> int: primitives_kwargs=primitives_kwargs, video_path=str(Path(output_dir) / "episode.mp4"), dashboard_events=dashboard_events, - save_action_videos=dashboard_server is not None, ) # --- agent loop -------------------------------------------------------- @@ -289,7 +242,6 @@ def main() -> int: toolkit=toolkit, max_turns=args.max_turns, input_queue=input_queue, - dashboard_interaction=dashboard_interaction, ) finish_result = result.finish_result messages = result.messages @@ -330,21 +282,6 @@ def main() -> int: if agent_error: logger.error("error: %s", agent_error) - dashboard_events.emit( - RunFinishedEvent( - state="failed" if agent_error else "succeeded", - error=agent_error, - ) - ) - if dashboard_server is not None: - logger.info( - "Run finished. Dashboard still serving at %s. Press Ctrl+C to stop.", - dashboard_url, - ) - try: - threading.Event().wait() - except KeyboardInterrupt: - pass return 0 diff --git a/rpent/dashboard/commands.py b/rpent/dashboard/commands.py new file mode 100644 index 00000000..c05b0878 --- /dev/null +++ b/rpent/dashboard/commands.py @@ -0,0 +1,87 @@ +"""Pure parsing for Dashboard-local control commands.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +LIBERO_SUITE_NAMES = ( + "libero_spatial", + "libero_object", + "libero_goal", + "libero_90", + "libero_object_task", + "libero_object_swap", + "libero_object_lan", + "libero_goal_task", + "libero_goal_swap", + "libero_goal_lan", + "libero_spatial_task", + "libero_spatial_swap", + "libero_spatial_lan", + "libero_10", + "libero_10_task", + "libero_10_swap", + "libero_10_lan", +) +LIBERO_SUITES = frozenset(LIBERO_SUITE_NAMES) + +TASK_COMMAND = "/rpent-task" +_NON_NEGATIVE_INTEGER = re.compile(r"[0-9]+") +_TASK_COMMAND_USAGE = f"{TASK_COMMAND} " + + +@dataclass(frozen=True, slots=True) +class TaskCommand: + """One validated request to create a fresh Dashboard TaskRun.""" + + suite: str + task: int + seed: int + + +class DashboardCommandError(ValueError): + """Raised when Dashboard command input is invalid or unsupported.""" + + +def parse_dashboard_command(text: str) -> TaskCommand | None: + """Parse a local task command or return ``None`` for ordinary text. + + Every input whose first token starts with ``/rpent-`` is reserved for the + Dashboard, so unsupported command names are rejected locally. + """ + + if not isinstance(text, str): + raise TypeError("Dashboard input must be a string") + + tokens = text.split() + if not tokens: + return None + + command_name = tokens[0] + if command_name.startswith("/rpent-"): + if command_name != TASK_COMMAND: + raise DashboardCommandError( + f"unknown Dashboard command: {command_name}" + ) + elif command_name.lower() != TASK_COMMAND: + return None + + if len(tokens) != 4 or command_name != TASK_COMMAND: + raise DashboardCommandError(f"expected {_TASK_COMMAND_USAGE}") + + _, suite, task_text, seed_text = tokens + if suite not in LIBERO_SUITES: + raise DashboardCommandError(f"unsupported LIBERO suite: {suite}") + + task = _parse_non_negative_integer("task", task_text) + seed = _parse_non_negative_integer("seed", seed_text) + return TaskCommand(suite=suite, task=task, seed=seed) + + +def _parse_non_negative_integer(name: str, value: str) -> int: + if _NON_NEGATIVE_INTEGER.fullmatch(value) is None: + raise DashboardCommandError( + f"{name} must be a non-negative integer, got {value!r}" + ) + return int(value) diff --git a/rpent/dashboard/events.py b/rpent/dashboard/events.py index 6276403e..5784624d 100644 --- a/rpent/dashboard/events.py +++ b/rpent/dashboard/events.py @@ -3,9 +3,7 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, Literal, Protocol, TypeAlias - -TerminalRunState: TypeAlias = Literal["succeeded", "failed", "cancelled"] +from typing import Any, Protocol, TypeAlias @dataclass(frozen=True, slots=True) @@ -46,29 +44,23 @@ class RunStartedEvent: """Mark startup complete and the agent run active.""" -@dataclass(frozen=True, slots=True) -class RunFinishedEvent: - """Mark the run terminal with execution and task outcomes kept separate.""" - - terminated: bool | None = None - state: TerminalRunState = "succeeded" - reason: str | None = None - error: BaseException | str | None = None - - DashboardEvent: TypeAlias = ( TranscriptEvent | UsageEvent | RuntimeStatusEvent | ToolResultEvent | RunStartedEvent - | RunFinishedEvent ) class DashboardEventSink(Protocol): """Consumer used by planners, toolkits, and environment runtimes.""" + @property + def enabled(self) -> bool: + """Whether Dashboard-only projections and artifacts are needed.""" + ... + def emit(self, event: DashboardEvent) -> None: """Consume one Dashboard event.""" ... @@ -78,5 +70,9 @@ def emit(self, event: DashboardEvent) -> None: class NullDashboardEventSink: """No-op sink used when the Dashboard is disabled.""" + @property + def enabled(self) -> bool: + return False + def emit(self, event: DashboardEvent) -> None: return None diff --git a/rpent/dashboard/index.html b/rpent/dashboard/index.html index 0f22f4da..26f60562 100644 --- a/rpent/dashboard/index.html +++ b/rpent/dashboard/index.html @@ -9,74 +9,40 @@