From 1150ca39828f9ce7cd8fe7e990d1c82ff0b2da89 Mon Sep 17 00:00:00 2001 From: liujiyuan Date: Mon, 3 Aug 2026 18:18:21 +0800 Subject: [PATCH 01/15] refactor(libero): split shared and task runtimes --- robots/libero/__init__.py | 253 ++++++++++++++++++++++++++++---------- 1 file changed, 185 insertions(+), 68 deletions(-) diff --git a/robots/libero/__init__.py b/robots/libero/__init__.py index a5339a07..67e1fe17 100644 --- a/robots/libero/__init__.py +++ b/robots/libero/__init__.py @@ -4,6 +4,8 @@ import argparse import os import sys +from collections.abc import Callable, Iterable +from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import TYPE_CHECKING, Any @@ -22,6 +24,36 @@ from rpent.utils.rpc import RpcClient +@dataclass(frozen=True, slots=True) +class LiberoSharedRuntime: + """Session-owned VLA/SAM3 clients and their local daemons. + + ``owned_daemons`` contains only subprocesses started by this process. + Clients connected through external endpoints therefore remain usable by + their external owner after :meth:`close`. + """ + + model: Any + sam3_client: Any + owned_daemons: tuple["ProcessDaemon", ...] = () + + def close(self) -> None: + """Stop every locally owned shared-service daemon, best effort.""" + _stop_owned_daemons(self.owned_daemons) + + +@dataclass(frozen=True, slots=True) +class LiberoTaskRuntime: + """TaskRun-owned LIBERO env client and optional local env daemon.""" + + env: Any + owned_daemons: tuple["ProcessDaemon", ...] = () + + def close(self) -> None: + """Stop the locally owned env daemon, if any.""" + _stop_owned_daemons(self.owned_daemons) + + def get_env_spec() -> EnvSpec: """Return the LIBERO env identity, prompt bundle, and runner hooks. @@ -136,34 +168,31 @@ def _subprocess_env(**extra: str) -> dict[str, str]: return env -def _init_runtime( +def init_task_runtime( args: argparse.Namespace, output_dir: Path, dashboard_events: DashboardEventSink, -) -> tuple[list[ProcessDaemon], dict[str, Any]]: - """Spawn env + vla + SAM3 daemons and build clients for LIBERO. +) -> LiberoTaskRuntime: + """Initialize one TaskRun-owned LIBERO environment. - Each server can be spawned or attached-to independently: pass an - endpoint to attach, or leave it unset to spawn a local subprocess. + A local env server is fresh for every call. When ``--env-endpoint`` is + supplied, the returned handle owns no daemon and closing it leaves the + external service running. - Heavy deps (rpc / vla / daemon / env_client) are imported lazily so - that a bare ``import robots.libero`` (for ``get_env_spec`` / - ``get_toolkit``) doesn't drag them in. + 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.sam3_client import Sam3Client from rpent.utils.socket_rpc import SocketRpcClient - from rpent.utils.vla_client import VLAClient - daemons: list[ProcessDaemon] = [] + 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 [] - # --- env_server -------------------------------------------------------- dashboard_events.emit(RuntimeStatusEvent("env", "starting")) try: env_daemon: ProcessDaemon | None = None @@ -192,21 +221,60 @@ def _init_runtime( log_path=str(Path(output_dir) / "env_server.log"), ) env_daemon.start() - daemons.append(env_daemon) + 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}" - ) + env_rpc = _external_rpc_client( + args.env_endpoint, + option="--env-endpoint", + parse_endpoint=parse_endpoint, + http_client_factory=HttpRpcClient, + socket_client_factory=SocketRpcClient, + ) + wait_for_ready(env_rpc) + 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, suppress_errors=True) dashboard_events.emit(RuntimeStatusEvent("env", "failed", error=exc)) raise + dashboard_events.emit(RuntimeStatusEvent("env", "ready")) + return LiberoTaskRuntime( + env=env, + owned_daemons=tuple(owned_daemons), + ) + + +def init_shared_runtime( + args: argparse.Namespace, + output_dir: Path, + dashboard_events: DashboardEventSink, +) -> LiberoSharedRuntime: + """Initialize Session-owned VLA and SAM3 services. + + Local services are started once per call and recorded in the returned + handle. 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")) @@ -229,19 +297,19 @@ def _init_runtime( log_path=str(Path(output_dir) / "vla_server.log"), ) vla_daemon.start() - daemons.append(vla_daemon) + 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}" - ) + vla_rpc = _external_rpc_client( + args.vla_endpoint, + option="--vla-endpoint", + parse_endpoint=parse_endpoint, + http_client_factory=HttpRpcClient, + socket_client_factory=SocketRpcClient, + ) + wait_for_ready(vla_rpc) except Exception as exc: + _stop_owned_daemons(owned_daemons, suppress_errors=True) dashboard_events.emit(RuntimeStatusEvent("vla", "failed", error=exc)) raise @@ -266,49 +334,98 @@ def _init_runtime( log_path=str(Path(output_dir) / "sam3_server.log"), ) sam3_daemon.start() - daemons.append(sam3_daemon) + 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}" - ) + sam3_rpc = _external_rpc_client( + args.sam3_endpoint, + option="--sam3-endpoint", + parse_endpoint=parse_endpoint, + http_client_factory=HttpRpcClient, + socket_client_factory=SocketRpcClient, + ) + wait_for_ready(sam3_rpc) + model = VLAClient(vla_rpc) + sam3_client = Sam3Client(sam3_rpc) except Exception as exc: + _stop_owned_daemons(owned_daemons, suppress_errors=True) dashboard_events.emit(RuntimeStatusEvent("sam3", "failed", error=exc)) raise - # All local daemons are running now, so they initialize concurrently while - # readiness is checked in a deterministic order. - for component, client, daemon in ( - ("env", env_rpc, env_daemon), - ("sam3", sam3_rpc, sam3_daemon), - ("vla", vla_rpc, vla_daemon), - ): - try: - wait_for_ready(client, daemon=daemon) - except Exception as exc: - for started_daemon in reversed(daemons): - started_daemon.stop() - dashboard_events.emit(RuntimeStatusEvent(component, "failed", error=exc)) - raise - dashboard_events.emit(RuntimeStatusEvent(component, "ready")) + return LiberoSharedRuntime( + model=model, + sam3_client=sam3_client, + owned_daemons=tuple(owned_daemons), + ) + + +def _init_runtime( + args: argparse.Namespace, + output_dir: Path, + dashboard_events: DashboardEventSink, +) -> tuple[list[ProcessDaemon], dict[str, Any]]: + """Compose task and shared runtimes for the existing one-shot runner. + The env is initialized before VLA and SAM3, preserving the established + one-shot startup order and return value. If a later shared service fails, + the already-started task env is stopped before the exception escapes. + """ + task_runtime: LiberoTaskRuntime | None = None + try: + task_runtime = init_task_runtime(args, output_dir, dashboard_events) + shared_runtime = init_shared_runtime(args, output_dir, dashboard_events) + except Exception: + if task_runtime is not None: + # Preserve the shared-runtime startup error while still cleaning + # the env. Cleanup errors must not replace the actionable cause. + _stop_owned_daemons( + task_runtime.owned_daemons, + suppress_errors=True, + ) + raise + + daemons = [ + *task_runtime.owned_daemons, + *shared_runtime.owned_daemons, + ] primitives_kwargs = { - "env": LiberoEnvClient( - env_rpc, - expected_meta={ - "suite": args.suite, - "task": args.task, - "seed": args.seed, - "max_episode_steps": args.max_episode_steps, - }, - ), - "model": VLAClient(vla_rpc), - "sam3_client": Sam3Client(sam3_rpc), + "env": task_runtime.env, + "model": shared_runtime.model, + "sam3_client": shared_runtime.sam3_client, } return daemons, primitives_kwargs + + +def _external_rpc_client( + endpoint: str, + *, + option: str, + parse_endpoint: Callable[[str], tuple[str, str, int]], + http_client_factory: Callable[[str], RpcClient], + socket_client_factory: Callable[[str, int], RpcClient], +) -> RpcClient: + """Build a non-owned RPC transport for one configured endpoint.""" + protocol, host, port = parse_endpoint(endpoint) + if protocol == "socket": + return socket_client_factory(host, port) + if protocol == "http": + return http_client_factory(f"http://{host}:{port}") + raise ValueError(f"{option} protocol must be socket or http, got {protocol!r}") + + +def _stop_owned_daemons( + daemons: Iterable[ProcessDaemon], + *, + suppress_errors: bool = False, +) -> None: + """Stop owned daemons in reverse order while attempting every stop.""" + errors: list[Exception] = [] + for daemon in reversed(tuple(daemons)): + try: + daemon.stop() + except Exception as exc: + errors.append(exc) + if errors and not suppress_errors: + raise RuntimeError( + f"failed to stop {len(errors)} LIBERO runtime daemon(s)" + ) from errors[0] From 844295139847bb8e0e700204983e310c6f31f063 Mon Sep 17 00:00:00 2001 From: liujiyuan Date: Mon, 3 Aug 2026 18:18:27 +0800 Subject: [PATCH 02/15] feat(dashboard): add sequential task session control --- rpent/cli/dashboard.py | 228 +++++++++++++ rpent/cli/main.py | 68 +--- rpent/dashboard/commands.py | 84 +++++ rpent/dashboard/events.py | 15 +- rpent/dashboard/index.html | 52 +-- rpent/dashboard/interaction.py | 15 +- rpent/dashboard/launcher.py | 87 ++--- rpent/dashboard/server.py | 77 ++--- rpent/dashboard/session.py | 63 ++++ rpent/dashboard/state.py | 452 ++++++++++++++++---------- rpent/dashboard/static/dashboard.css | 6 + rpent/dashboard/static/dashboard.js | 231 ++++++------- rpent/dashboard/static/interaction.js | 100 ++++-- rpent/planner/claude_code.py | 50 ++- 14 files changed, 975 insertions(+), 553 deletions(-) create mode 100644 rpent/cli/dashboard.py create mode 100644 rpent/dashboard/commands.py create mode 100644 rpent/dashboard/session.py diff --git a/rpent/cli/dashboard.py b/rpent/cli/dashboard.py new file mode 100644 index 00000000..9d30ec81 --- /dev/null +++ b/rpent/cli/dashboard.py @@ -0,0 +1,228 @@ +"""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 + +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 robots.libero import LiberoSharedRuntime + from rpent.dashboard.state import ClaimedTask, DashboardState + from rpent.envs.env_spec import EnvSpec + +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 robots.libero import init_shared_runtime + 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: 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: LiberoSharedRuntime, + session_root: Path, +) -> str | None: + """Execute one fresh Dashboard TaskRun against Session-owned services.""" + from robots.libero import init_task_runtime + + 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_runtime = None + toolkit = None + started = time.time() + try: + task_runtime = init_task_runtime(task_args, output_dir, state) + if not state.task_replacement_requested: + primitives_kwargs = { + "env": task_runtime.env, + "model": shared.model, + "sam3_client": shared.sam3_client, + } + 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}") + if task_runtime is not None: + try: + task_runtime.close() + 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..fdf936cd 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, @@ -289,7 +243,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 +283,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..3b6670ef --- /dev/null +++ b/rpent/dashboard/commands.py @@ -0,0 +1,84 @@ +"""Pure parsing for Dashboard-local control commands.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +LIBERO_SUITES = frozenset( + { + "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", + } +) + +_TASK_COMMAND = "/rpent-task" +_NON_NEGATIVE_INTEGER = re.compile(r"[0-9]+") +_TASK_COMMAND_USAGE = "/rpent-task " + + +@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..4c1a8d5f 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,23 +44,12 @@ 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 ) diff --git a/rpent/dashboard/index.html b/rpent/dashboard/index.html index 229ea0e3..11b3d2bc 100644 --- a/rpent/dashboard/index.html +++ b/rpent/dashboard/index.html @@ -9,70 +9,40 @@