Add queued execution speed and explicit pause/resume

.github/workflows/tests.yml

@@ -144,6 +144,12 @@ jobs:
           PYTHONUTF8: '1'
         run: |
           pytest
+      - name: Preserve test report
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: junit-${{ matrix.os }}-python-${{ matrix.python-version }}
+          path: test-results.xml
       # The examples are the scripts a user copies, and `--examples` is opt-in,
       # so nothing was running them: four had rotted into refusals on their
       # own happy path. They run each script as a subprocess against the
@@ -152,4 +158,10 @@ jobs:
         env:
           PYTHONUNBUFFERED: '1'
           PYTHONUTF8: '1'
-        run: pytest tests/test_examples.py --examples
+        run: pytest tests/test_examples.py --examples --junitxml=example-results.xml
+      - name: Preserve example report
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: examples-${{ matrix.os }}-python-${{ matrix.python-version }}
+          path: example-results.xml

README.md

@@ -310,6 +310,32 @@ Speed and accel are fractions of maximum (0.0–1.0), not percentages.
 
 For Cartesian moves, joint limits stay at 100% as hard bounds—the speed fraction only affects the Cartesian velocity constraint.
 
+### Queued execution speed and pause
+
+`set_execution_speed(scale)` selects 10–100% of an already planned trajectory's
+speed. The command's `speed`, `accel` and `duration` still define the original
+plan. Jog and streamed servo commands retain their own timing.
+
+Override transitions use a separate rate ramp and acceleration checks. The
+nominal motion profile's jerk ceiling is not guaranteed during a transition.
+
+Use `pause()` to retain the queue and decelerate queued motion to a hold, and
+`resume()` to continue at the selected scale. Changing speed while paused keeps
+the pause. The speed setter rejects zero. These controls return 1 when their
+request is confirmed, or 0 when confirmation times out.
+
+Fresh `execution_speed()` readback exposes `target_scale`, `applied_scale` and
+`resume_scale`. Its `paused` property confirms the applied scale reached zero;
+the pause request can be acknowledged while still decelerating. Queued delays
+retain their remaining time while paused; positive speed changes do not retime
+delays, tool actuators or homing routines already in progress.
+
+Standalone `wait_command()` keeps its wall-clock timeout and returns false if
+completion is unconfirmed. Blocking motion calls raise `TimeoutError` in that
+case. A timed-out wait leaves the motion queued; `stop()` cancels it. Planning
+preview retimes trajectories and reports paused queued operations as
+`UnresolvedPreview` instead of claiming completion.
+
 ## Command system
 
 Jog and servo commands (JogJ, JogL, ServoJ, ServoL) automatically use the streaming fast-path — the server de-duplicates stale inputs, reduces ACK chatter, and reuses the active command. Use jog/servo for UI-driven motion or teleoperation; use planned moves (MoveJ, MoveL, etc.) for discrete motions and queued programs.

examples/draw_circle.py

@@ -75,7 +75,7 @@ def circle_pt(cx, cz, angle_deg):
         z = z_min + t * (z_max - z_min)
         x = RADIUS * math.cos(t * 3 * 2 * math.pi)
         spline.append([x, CIRCLE_Y, z] + ORIENTATION)
-    rbt.move_s(spline, speed=SPEED, wait=True)
+    rbt.move_s(spline, speed=SPEED, wait=True, timeout=60)
 
     rbt.home(wait=True)
     print("Done!")

parol6/ack_policy.py

@@ -14,6 +14,8 @@
     CmdType.WRITE_IO,
     CmdType.SET_SHAPES,
     CmdType.SET_STATUS_RATE,
+    CmdType.SET_EXECUTION_SPEED,
+    CmdType.PAUSE,
 }
 
 # Query command types (use request/response, not ACK)
@@ -38,6 +40,7 @@
     CmdType.TCP_TRANSFORM,
     CmdType.SHAPES,
     CmdType.STATUS_RATE,
+    CmdType.EXECUTION_SPEED,
 }
 
 # Streaming commands are fire-and-forget (no ACK needed)

parol6/client/async_client.py

@@ -27,6 +27,8 @@
 )
 from waldoctl.tools import ToolSpec
 
+from waldoctl.execution import ExecutionSpeed, validate_execution_scale
+
 from .. import config as cfg
 from ..ack_policy import QUERY_CMD_TYPES, SYSTEM_CMD_TYPES, AckPolicy
 from ..utils.error_catalog import RobotError
@@ -43,6 +45,10 @@
     DelayCmd,
     EnablementResultStruct,
     ErrorCmd,
+    ExecutionSpeedCmd,
+    ExecutionSpeedResultStruct,
+    SetExecutionSpeedCmd,
+    PauseCmd,
     ErrorResultStruct,
     ErrorMsg,
     IOCmd,
@@ -270,6 +276,7 @@ class AsyncRobotClient(_RobotClientABC):
     def skill_capabilities(self) -> frozenset[str]:
         return super().skill_capabilities | {
             "backend.parol6",
+            "execution.speed",
             "tool.gripper",
             "io.digital",
         }
@@ -1009,6 +1016,93 @@ async def reset_loop_stats(self) -> int:
         """
         return await self._send(ResetLoopStatsCmd())
 
+    async def execution_speed(self, *, timeout: float = 3.0) -> ExecutionSpeed:
+        """Read fresh requested, applied, and retained execution scales.
+
+        Category: Query
+
+        Example:
+            speed = rbt.execution_speed()
+        """
+        self._validate_execution_timeout(timeout)
+        async with asyncio.timeout(timeout):
+            response = await self._request(ExecutionSpeedCmd())
+            if not isinstance(response, ExecutionSpeedResultStruct):
+                raise ConnectionError("Controller execution speed is unavailable")
+            return ExecutionSpeed(
+                response.target_scale, response.applied_scale, response.resume_scale
+            )
+
+    @staticmethod
+    def _validate_execution_timeout(timeout: float) -> None:
+        if isinstance(timeout, bool) or not math.isfinite(timeout) or timeout <= 0:
+            raise ValueError("Execution control timeout must be positive and finite")
+
+    async def _request_execution_state(
+        self, *, timeout: float, scale: float | None = None, paused: bool = False
+    ) -> int:
+        self._validate_execution_timeout(timeout)
+        try:
+            async with asyncio.timeout(timeout):
+                command = (
+                    SetExecutionSpeedCmd(scale)
+                    if scale is not None
+                    else PauseCmd(paused)
+                )
+                if await self._send(command) <= 0:
+                    return 0
+                while True:
+                    state = await self.execution_speed(timeout=timeout)
+                    confirmed = (
+                        state.resume_scale == scale
+                        if scale is not None
+                        else (state.target_scale == 0) == paused
+                    )
+                    if confirmed:
+                        return 1
+                    await asyncio.sleep(0.01)
+        except (TimeoutError, ConnectionError):
+            # 0 is "unconfirmed": the command may or may not have been applied.
+            # A readback whose reply was lost inside the confirmation window is
+            # exactly that, and raising instead told the caller the controller
+            # was unreachable when it had acked the command a moment earlier.
+            return 0
+
+    async def set_execution_speed(self, scale: float, *, timeout: float = 3.0) -> int:
+        """Select 10–100% of planned queued-motion speed, preserving pause.
+
+        Category: Control
+
+        Example:
+            rbt.set_execution_speed(0.5)
+        """
+        return await self._request_execution_state(
+            scale=validate_execution_scale(scale), timeout=timeout
+        )
+
+    async def pause(self, *, timeout: float = 3.0) -> int:
+        """Request a controlled hold, retaining queued trajectory progress.
+
+        A confirmed request returns 1. Read ``execution_speed().paused``
+        to confirm the hold. Standalone Python completion timeouts continue.
+
+        Category: Control
+
+        Example:
+            rbt.pause()
+        """
+        return await self._request_execution_state(paused=True, timeout=timeout)
+
+    async def resume(self, *, timeout: float = 3.0) -> int:
+        """Resume the retained queue at its selected positive speed.
+
+        Category: Control
+
+        Example:
+            rbt.resume()
+        """
+        return await self._request_execution_state(paused=False, timeout=timeout)
+
     async def set_status_rate(self, hz: float) -> int:
         """Set the rate the controller broadcasts status at.
 
@@ -1567,8 +1661,8 @@ async def move_j(
                     rel=rel,
                 )
             )
-        if wait and index >= 0:
-            await self.wait_command(index, timeout=timeout)
+        if wait and index >= 0 and not await self.wait_command(index, timeout=timeout):
+            raise TimeoutError(f"Command {index} did not complete within {timeout}s")
         return index
 
     async def move_l(
@@ -1614,8 +1708,8 @@ async def move_l(
             rel=rel,
         )
         index = await self._send(cmd)
-        if wait and index >= 0:
-            await self.wait_command(index, timeout=timeout)
+        if wait and index >= 0 and not await self.wait_command(index, timeout=timeout):
+            raise TimeoutError(f"Command {index} did not complete within {timeout}s")
         return index
 
     async def move_c(
@@ -1661,8 +1755,8 @@ async def move_c(
             r=r,
         )
         index = await self._send(cmd)
-        if wait and index >= 0:
-            await self.wait_command(index, timeout=timeout)
+        if wait and index >= 0 and not await self.wait_command(index, timeout=timeout):
+            raise TimeoutError(f"Command {index} did not complete within {timeout}s")
         return index
 
     async def move_s(
@@ -1702,8 +1796,8 @@ async def move_s(
             accel=accel,
         )
         index = await self._send(cmd)
-        if wait and index >= 0:
-            await self.wait_command(index, timeout=timeout)
+        if wait and index >= 0 and not await self.wait_command(index, timeout=timeout):
+            raise TimeoutError(f"Command {index} did not complete within {timeout}s")
         return index
 
     async def move_p(
@@ -1743,8 +1837,8 @@ async def move_p(
             accel=accel,
         )
         index = await self._send(cmd)
-        if wait and index >= 0:
-            await self.wait_command(index, timeout=timeout)
+        if wait and index >= 0 and not await self.wait_command(index, timeout=timeout):
+            raise TimeoutError(f"Command {index} did not complete within {timeout}s")
         return index
 
     async def checkpoint(self, label: str) -> int:
@@ -1994,6 +2088,10 @@ async def tool_action(
             params=params or [],
         )
         result = await self._send(cmd)
-        if wait and result >= 0:
-            await self.wait_command(result, timeout=timeout)
+        if (
+            wait
+            and result >= 0
+            and not await self.wait_command(result, timeout=timeout)
+        ):
+            raise TimeoutError(f"Command {result} did not complete within {timeout}s")
         return result

parol6/client/dry_run_client.py

@@ -13,6 +13,8 @@
 from typing import Any
 
 import numpy as np
+from waldoctl.execution import ExecutionSpeed, validate_execution_scale
+from waldoctl.skills import UnresolvedPreview
 
 import parol6.PAROL6_ROBOT as PAROL6_ROBOT
 from ..commands.base import MotionCommand
@@ -181,8 +183,7 @@ class DryRunRobotClient:
     simulated separately since the planner doesn't handle streaming.
 
     Most methods are auto-dispatched via __getattr__ using CMD_MAP.
-    Explicit methods exist only for angles/pose (read from state)
-    and delay (no-op).
+    Execution controls change the planning clock; observations read local state.
     """
 
     def __init__(
@@ -251,6 +252,8 @@ def tcp_transform(self) -> list[float]:
 
     def flush(self) -> list[DryRunResult]:
         """Flush pending blend buffer. Call after script completion."""
+        if self._planner._blend_buffer:
+            self._require_running()
         segments = self._planner.flush()
         self._state.Position_in[:] = self._planner.state.Position_in
         results: list[DryRunResult] = []
@@ -283,6 +286,13 @@ def _snap_to_angles(self, angles_deg: list[float]) -> DryRunResult:
 
     def _dispatch(self, params: Any) -> DryRunResult | None:
         """Route a command struct through the trajectory planner."""
+        cmd_cls = self._registry.get_command_for_struct(type(params))
+        if (
+            cmd_cls is not None
+            and issubclass(cmd_cls, MotionCommand)
+            and not cmd_cls.streamable
+        ):
+            self._require_running()
         if isinstance(params, HomeCmd):
             if params.calibrate or not self._planner.state.Homed_in[:6].all():
                 return self._snap_to_angles(HOME_ANGLES_DEG)
@@ -312,7 +322,6 @@ def _dispatch(self, params: Any) -> DryRunResult | None:
         # Detect jog/servo commands — planner doesn't handle streaming.
         # Other non-trajectory MotionCommands (SelectTool, Home) fall through
         # to the planner which handles them as inline segments.
-        cmd_cls = self._registry.get_command_for_struct(type(params))
         if cmd_cls is not None and issubclass(cmd_cls, (JogJCommand, JogLCommand)):
             self._planner.flush()
             self._state.Position_in[:] = self._planner.state.Position_in
@@ -360,7 +369,7 @@ def _trajectory_segment_to_result(self, seg: TrajectorySegment) -> DryRunResult:
         for i in range(len(sampled)):
             steps_to_rad(sampled[i], radians[i])
 
-        return _build_result(radians, seg.duration)
+        return _build_result(radians, seg.duration / self._state.execution_speed)
 
     def _error_segment_to_result(self, seg: ErrorSegment) -> DryRunResult:
         """Convert an ErrorSegment to a DryRunResult with per-pose validity."""
@@ -558,6 +567,7 @@ def skill_capabilities(self) -> frozenset[str]:
                 "backend.parol6",
                 "io.digital",
                 "execution.preview",
+                "execution.speed",
             }
         )
 
@@ -623,6 +633,30 @@ def write_io(self, index: int, value: int, *, timeout: float | None = None) -> i
             raise RuntimeError(str(result.error))
         return 0
 
+    def _require_running(self) -> None:
+        if self._state.execution_paused:
+            raise UnresolvedPreview(
+                "Queued execution is paused; preview needs an explicit resume "
+                "before it can predict completion"
+            )
+
+    def set_execution_speed(self, scale: float, *, timeout: float = 3.0) -> int:
+        self._state.execution_speed = validate_execution_scale(scale)
+        return 1
+
+    def execution_speed(self, *, timeout: float = 3.0) -> ExecutionSpeed:
+        scale = self._state.execution_speed
+        applied = 0.0 if self._state.execution_paused else scale
+        return ExecutionSpeed(applied, applied, scale)
+
+    def pause(self, *, timeout: float = 3.0) -> int:
+        self._state.execution_paused = True
+        return 1
+
+    def resume(self, *, timeout: float = 3.0) -> int:
+        self._state.execution_paused = False
+        return 1
+
     def jog_j(
         self,
         joint: int = -1,
@@ -670,7 +704,7 @@ def jog_l(
         )
 
     def delay(self, seconds: float = 0.0) -> None:
-        pass
+        self._require_running()
 
     def wait_motion(self, **kwargs: Any) -> None:
         self.flush()