feat(dual-franka): add manual continuous VLA diagnostics

robots/dual_franka/dual_franka_manual_call.py

@@ -67,7 +67,11 @@
 
 def _tool_spec_map(*, include_manual: bool = False) -> dict[str, dict[str, Any]]:
     specs = {
-        tool.name: {"name": tool.name, "input_schema": tool.input_schema}
+        tool.name: {
+            "name": tool.name,
+            "description": tool.description,
+            "input_schema": tool.input_schema,
+        }
         for tool in DualFrankaToolkit.declared_tools()
     }
     if include_manual:

rpent/robots/components/pi05_vla_client.py

@@ -236,6 +236,10 @@ def encode_obs(self, env_obs: dict) -> dict:
         """Dispatch to the embodiment's encoder from ``_ENCODE_OBS``."""
         return _ENCODE_OBS[self._embodiment](env_obs)
 
+    def status(self, *, timeout_s: float = 10) -> dict[str, Any]:
+        """Query read-only metadata from a server supporting ``vla.status``."""
+        return self._client.call("vla.status", timeout_s=timeout_s)
+
     # ---- inference ----
 
     def predict(self, env_obs: dict, options: dict | None = None) -> np.ndarray:

rpent/robots/components/pi05_vla_server.py

@@ -225,6 +225,17 @@ def __init__(
             cfg["model_path"],
         )
         self._model = get_model(cfg, torch_dtype=None).cuda().eval()
+        parameters = list(self._model.parameters())
+        self._status = {
+            "model_path": model_path,
+            "repo_id": repo_id,
+            "embodiment": embodiment,
+            "dtypes": sorted({str(p.dtype) for p in parameters}),
+            "device": str(parameters[0].device) if parameters else None,
+            "load_seconds": time.time() - t0,
+            "config": OmegaConf.to_container(cfg, resolve=True),
+        }
+        self._rpc["vla.status"] = lambda: self._status
         logger.info("model ready in %.1fs", time.time() - t0)
 
     def _register_rpc(self):

tests/e2e_tests/dual_franka/dual_franka_vla.py

@@ -104,10 +104,13 @@ def __init__(
         workspace: dict[str, Any],
         prompt: str,
         expected_steps: int,
+        *,
+        status: dict[str, Any] | None = None,
     ) -> None:
         self.env, self.model = env, model
         self.output, self.workspace = Path(output), workspace
         self.prompt = prompt
+        self.status = status
         self.index = 0
         self.execution_uncertain = False
         self.episode_done = False
@@ -181,9 +184,7 @@ def run_console(
     emit: Callable[[str], None] = print,
 ) -> None:
     """Read explicit commands without invoking an Agent."""
-    help_text = (
-        "prompt <instruction> | infer | step | run N (1..20 chunks) | reset | quit"
-    )
+    help_text = "status | prompt <instruction> | infer | step | run N (1..20 chunks) | reset | quit"
     emit(help_text)
     emit("infer 不下发预测动作；step 执行一个动作块。Ctrl-C 退出会话，不等同硬件急停。")
     while True:
@@ -199,6 +200,18 @@ def run_console(
                     raise ValueError("prompt cannot be empty")
                 session.prompt = arg.strip()
                 emit("Prompt: " + session.prompt)
+            elif command == "status" and not arg:
+                session.status = session.model.status(timeout_s=10)
+                emit(
+                    json.dumps(
+                        {
+                            **session.status,
+                            "prompt": session.prompt,
+                            "expected_action_steps": session.expected_steps,
+                        },
+                        ensure_ascii=False,
+                    )
+                )
             elif command == "help" and not arg:
                 emit(help_text)
             elif command == "reset" and not arg:
@@ -233,7 +246,7 @@ def run_console(
 
 def run_session(args: argparse.Namespace) -> int:
     """Own the diagnostic runtime and close it when the console exits."""
-    if args.expected_action_steps <= 0:
+    if args.expected_action_steps is not None and args.expected_action_steps <= 0:
         raise ValueError("expected action steps must be positive")
     spec = get_robot_spec()
     from robots.dual_franka.runtime_config import DEFAULT_CONFIG
@@ -261,10 +274,18 @@ def run_session(args: argparse.Namespace) -> int:
             args, output, NullDashboardEventSink(), {"env", "vla"}
         )
         model = kwargs["model"]
+        expected_steps = args.expected_action_steps
+        status = None
+        if expected_steps is None:
+            status = model.status(timeout_s=10)
+            expected_steps = int(status["config"]["openpi"]["action_chunk"])
+            if expected_steps <= 0:
+                raise ValueError("expected action steps must be positive")
         save_record(
             output / "deployment",
             {
-                "expected_action_steps": args.expected_action_steps,
+                "expected_action_steps": expected_steps,
+                "server": status,
                 "robot_config": data,
                 "camera_meta": kwargs["env"].get_camera_meta(),
             },
@@ -275,7 +296,8 @@ def run_session(args: argparse.Namespace) -> int:
             output,
             data["workspace"],
             args.instruction,
-            args.expected_action_steps,
+            expected_steps,
+            status=status,
         )
         logger.info("Records: %s", output)
         run_console(session)
@@ -298,8 +320,11 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         "--expected-action-steps",
         type=int,
-        default=20,
-        help="Expected prediction chunk length for action validation (default: 20)",
+        default=None,
+        help=(
+            "Override prediction chunk length; otherwise read vla.status "
+            "from the server. Use an explicit value with older servers."
+        ),
     )
     parser.add_argument("--output-dir", default=None)
     return parser

tests/manual/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2026 The RPent Authors.
+#
+# 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
+#
+#     https://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.
+
+"""Explicitly invoked manual diagnostics; never automatic hardware tests."""

tests/manual/dual_franka_full_vla.py

@@ -0,0 +1,217 @@
+# Copyright 2026 The RPent Authors.
+#
+# 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
+#
+#     https://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.
+
+"""Manual-only continuous VLA rollout; never registered with the agent toolkit."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import signal
+import threading
+import time
+from contextlib import ExitStack
+from datetime import datetime
+from pathlib import Path
+
+import yaml
+
+from robots.dual_franka.env_client import DualFrankaEnvClient
+from robots.dual_franka.runtime_config import DEFAULT_CONFIG
+from robots.dual_franka.tasks import get_dual_franka_task
+from rpent.robots.components.pi05_vla_client import Pi05VLAClient
+from rpent.utils.rpc import make_rpc_client
+from tests.e2e_tests.dual_franka.dual_franka_vla import (
+    validate_actions,
+    validate_observation,
+)
+
+
+def run_full_vla(env, model, *, prompt, workspace, expected_steps, stopped, emit):
+    """Run indefinitely, checking cancellation before each single-action RPC.
+
+    An in-flight inference/RPC is allowed to return; its remaining actions are
+    discarded on cancellation. No reset, recovery or gripper command is added.
+    """
+    if not prompt.strip():
+        raise ValueError("policy instruction must not be empty")
+    chunks = steps = 0
+    started = time.monotonic()
+    summary = {"status": "stopped", "reason": "operator_stop"}
+    try:
+        while not stopped():
+            observation = dict(env.get_observation())
+            validate_observation(observation)
+            if stopped():
+                break
+            observation["task_descriptions"] = prompt
+            prediction = model.predict(observation, options={"mode": "eval"})
+            if stopped():
+                break
+            actions = validate_actions(prediction, workspace, expected_steps)
+            chunks += 1
+            emit({"kind": "prediction", "chunk": chunks, "actions": actions.tolist()})
+            for index, action in enumerate(actions):
+                if stopped():
+                    return summary
+                # Do not queue a whole prediction chunk on the robot server.
+                result = env.chunk_step(action[None, :])
+                steps += 1
+                emit(
+                    {
+                        "kind": "action",
+                        "chunk": chunks,
+                        "index": index,
+                        "steps": steps,
+                        "terminated": bool(result.get("terminated")),
+                        "truncated": bool(result.get("truncated")),
+                    }
+                )
+                if result.get("error") or result.get("ok") is False:
+                    raise RuntimeError(f"environment action failed: {result}")
+                if result.get("terminated") or result.get("truncated"):
+                    summary.update(
+                        status="environment_stopped", reason="terminated_or_truncated"
+                    )
+                    return summary
+            emit(
+                {
+                    "kind": "chunk_complete",
+                    "chunks": chunks,
+                    "steps": steps,
+                    "elapsed_s": time.monotonic() - started,
+                }
+            )
+        return summary
+    except BaseException as exc:
+        summary.update(status="failed", reason=f"{type(exc).__name__}: {exc}")
+        raise
+    finally:
+        summary.update(chunks=chunks, steps=steps, elapsed_s=time.monotonic() - started)
+        emit({"kind": "finish", **summary})
+
+
+# Manual-only entry point; not registered among the agent's native tools.
+MANUAL_SKILLS = {"vla_full_rollout": run_full_vla}
+
+
+def build_parser():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--skill", choices=MANUAL_SKILLS, default="vla_full_rollout")
+    parser.add_argument("--env-endpoint", default="http://127.0.0.1:6002")
+    parser.add_argument("--vla-endpoint", default="http://127.0.0.1:6000")
+    parser.add_argument("--task-id", type=int, default=6)
+    parser.add_argument(
+        "--instruction", help="Override task-configured policy instruction"
+    )
+    parser.add_argument("--robot-config", type=Path, default=DEFAULT_CONFIG)
+    parser.add_argument("--output-dir", type=Path)
+    parser.add_argument("--stop-file", type=Path)
+    return parser
+
+
+def main():
+    args = build_parser().parse_args()
+    output = args.output_dir or Path("logs") / datetime.now().strftime(
+        "%Y%m%d-%H%M%S-manual-full-vla"
+    )
+    output = output.resolve()
+    # Never overwrite an earlier experiment's logs or remove its stop signal.
+    output.mkdir(parents=True, exist_ok=False)
+    stop_file = (args.stop_file or output / "STOP").resolve()
+    if stop_file.exists():
+        raise RuntimeError(f"stop file already exists: {stop_file}; use a new path")
+    prompt = args.instruction or get_dual_franka_task(args.task_id).vla_instruction
+    if not prompt:
+        raise ValueError("selected task has no VLA instruction")
+    config = yaml.safe_load(args.robot_config.read_text())
+    event = threading.Event()
+    signals = []
+
+    def request_stop(signum, _frame):
+        signals.append(signal.Signals(signum).name)
+        event.set()
+
+    previous = {
+        s: signal.signal(s, request_stop) for s in (signal.SIGINT, signal.SIGTERM)
+    }
+    print(f"Logs: {output}\nStop file: {stop_file}", flush=True)
+    print(
+        "No reset. No duration/chunk/semantic-boundary limit. Ctrl+C or create STOP to stop.",
+        flush=True,
+    )
+    print(
+        "An in-flight RPC may finish; this is NOT a hardware emergency stop.",
+        flush=True,
+    )
+    try:
+        with ExitStack() as stack:
+            env_rpc = make_rpc_client(args.env_endpoint)
+            stack.callback(env_rpc.close)
+            model_rpc = make_rpc_client(args.vla_endpoint)
+            stack.callback(model_rpc.close)
+            env = DualFrankaEnvClient(env_rpc, reset_on_connect=False)
+            model = Pi05VLAClient(model_rpc, embodiment="dual_franka")
+            status = model.status(timeout_s=10)
+            (output / "deployment.json").write_text(
+                json.dumps(
+                    {
+                        "skill": args.skill,
+                        "task_id": args.task_id,
+                        "prompt": prompt,
+                        "model": status,
+                        "robot_config": str(args.robot_config.resolve()),
+                        "stop_file": str(stop_file),
+                        "reset_on_connect": False,
+                    },
+                    indent=2,
+                    ensure_ascii=False,
+                )
+            )
+            with (output / "events.jsonl").open("x", buffering=1) as log:
+
+                def emit(record):
+                    log.write(
+                        json.dumps(
+                            {"timestamp": time.time(), **record}, ensure_ascii=False
+                        )
+                        + "\n"
+                    )
+                    if record["kind"] == "finish":
+                        (output / "outcome.json").write_text(
+                            json.dumps({**record, "signals": signals}, indent=2)
+                        )
+                    if record["kind"] in {"chunk_complete", "finish"}:
+                        print(json.dumps(record, ensure_ascii=False), flush=True)
+
+                result = MANUAL_SKILLS[args.skill](
+                    env,
+                    model,
+                    prompt=prompt,
+                    workspace=config["workspace"],
+                    expected_steps=int(status["config"]["openpi"]["action_chunk"]),
+                    stopped=lambda: event.is_set() or stop_file.exists(),
+                    emit=emit,
+                )
+                (output / "outcome.json").write_text(
+                    json.dumps({**result, "signals": signals}, indent=2)
+                )
+                return 0 if result["status"] == "stopped" else 1
+    finally:
+        for signum, handler in previous.items():
+            signal.signal(signum, handler)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/unit_tests/robots/dual_franka/test_dual_franka_extension.py

@@ -151,3 +151,21 @@ def test_dual_franka_exploration_prompt_is_opt_in():
     assert "request_scene_reset" in explore_prompt
     assert "request_operator_verdict" in explore_prompt
     assert "/tmp/memory/_internal/inbox/dual_franka_t4" in explore_user_prompt
+
+
+def test_manual_schemas_preserve_native_tool_declarations():
+    from robots.dual_franka.dual_franka_manual_call import (
+        _registered_tool_names,
+        _schema_payload,
+    )
+    from robots.dual_franka.toolkit import DualFrankaToolkit
+
+    tools = DualFrankaToolkit.declared_tools()
+    assert _registered_tool_names() == {tool.name for tool in tools}
+    for tool in tools:
+        assert _schema_payload(tool.name) == {
+            "primitive": tool.name,
+            "description": tool.description,
+            "input_schema": tool.input_schema,
+        }
+    assert _schema_payload("reset")["description"].startswith("Manual-only")