Support bounded digital I/O for named signal skills

README.md

@@ -10,6 +10,13 @@ This package provides:
 - **`parol6-server`** CLI for standalone controller operation
 
 The controller speaks a msgpack-based UDP protocol and can run on the same machine or remotely.
+Every command datagram carries a 4-byte request id ahead of the msgpack body, and the
+OK / ERROR / RESPONSE reply echoes it, so a reply whose caller has already given up is
+dropped instead of answering the next request. An id of 0 asks for no reply, which is
+what streamed motion sends. Status broadcasts carry `PROTO_VERSION` in their second
+slot: a client reading a status from another version raises `ProtocolVersionError`
+naming both, rather than reporting the silence of a failed decode. Client and
+controller are released together — there is no compatibility window between versions.
 
 ---
 
@@ -451,3 +458,13 @@ The existing `set_tcp_offset(x, y, z)` clears user rotation and now returns its
 queued index for confirmation. `tcp_offset()` still reads three translations;
 `tcp_transform()` reads all six values. Both raise `TimeoutError` when no valid
 reply arrives instead of reporting a misleading zero correction.
+
+Digital I/O reads and writes accept an optional per-call `timeout` in seconds:
+`rbt.io(timeout=1.0)` returns `None` without a reply, while
+`rbt.write_io(0, 1, timeout=1.0)` raises `TimeoutError` if acceptance remains
+unconfirmed. The deadline includes transport setup and retries. Omitting it
+retains the configured client timeout. The same options work on the sync client.
+The client advertises `io.digital` for typed named-signal skills, which can be
+imported from `waldo_commander.skills`; mappings are `waldoctl.signals.DigitalSignal`
+values stored in a setup snapshot. Dry-run clients advertise `execution.preview`
+so those skills require explicit observation fixtures during preview.

parol6/client/async_client.py

@@ -5,6 +5,7 @@
 import asyncio
 import contextlib
 import logging
+import math
 import random
 import socket
 import struct
@@ -109,6 +110,8 @@
     ToolStatusResultStruct,
     ToolsCmd,
     WriteIOCmd,
+    MAX_REQ_ID,
+    ProtocolVersionError,
     decode_message,
     encode_command,
     encode_command_into,
@@ -235,7 +238,15 @@ def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None:
         if self._client._closed:
             return
         # Zero-allocation decode directly into shared buffer
-        if decode_status_bin_into(data, self._client._shared_status):
+        try:
+            fresh = decode_status_bin_into(data, self._client._shared_status)
+        except ProtocolVersionError as mismatch:
+            # Raising inside a datagram callback reaches nobody. Hold it for
+            # whoever reads status next, and wake them now.
+            self._client._proto_error = mismatch
+            self._client._status_event.set()
+            return
+        if fresh:
             self._client._status_generation += 1
             # Event.set() is synchronous, so it's safe to wake waiters from this callback
             self._client._status_event.set()
@@ -257,7 +268,11 @@ class AsyncRobotClient(_RobotClientABC):
 
     @property
     def skill_capabilities(self) -> frozenset[str]:
-        return super().skill_capabilities | {"backend.parol6", "tool.gripper"}
+        return super().skill_capabilities | {
+            "backend.parol6",
+            "tool.gripper",
+            "io.digital",
+        }
 
     def __init__(
         self,
@@ -294,6 +309,9 @@ def __init__(
         # Single shared buffer with event-based notification
         self._status_transport: asyncio.DatagramTransport | None = None
         self._status_sock: socket.socket | None = None
+        self._proto_error: ProtocolVersionError | None = None
+        #: Correlates each reply with its request; 0 means "no reply wanted".
+        self._next_req_id = 1
         self._shared_status: StatusBuffer = StatusBuffer()
         self._status_generation: int = 0
         self._status_event: asyncio.Event = asyncio.Event()
@@ -526,6 +544,7 @@ async def stream_status_shared(self) -> AsyncIterator[StatusBuffer]:
         last_gen = 0
 
         while not self._closed:
+            self._check_protocol()
             # Clear before waiting - only affects future waits, not current waiters
             self._status_event.clear()
 
@@ -543,6 +562,16 @@ async def stream_status_shared(self) -> AsyncIterator[StatusBuffer]:
                 last_gen = self._status_generation
                 yield self._shared_status
 
+    def _request_id(self) -> int:
+        """The next request id, wrapping past the wire's 32-bit field."""
+        req_id = self._next_req_id
+        self._next_req_id = req_id + 1 if req_id < MAX_REQ_ID else 1
+        return req_id
+
+    def _check_protocol(self) -> None:
+        if self._proto_error is not None:
+            raise self._proto_error
+
     async def _send(self, cmd: msgspec.Struct) -> int:
         """
         Send a binary command based on AckPolicy.
@@ -560,16 +589,22 @@ async def _send(self, cmd: msgspec.Struct) -> int:
 
         # System commands need stable bytes across the await, so encode a fresh buffer
         if cmd_type in SYSTEM_CMD_TYPES:
+            req_id = self._request_id()
             try:
-                await self._request_ok_raw(encode_command(cmd), self.timeout)
+                await self._request_ok_raw(
+                    encode_command(cmd, req_id), self.timeout, req_id
+                )
                 return 1
             except TimeoutError:
                 return 0
 
         if cmd_type not in QUERY_CMD_TYPES:
             if self._ack_policy.requires_ack(cmd_type):
+                req_id = self._request_id()
                 try:
-                    ok = await self._request_ok_raw(encode_command(cmd), self.timeout)
+                    ok = await self._request_ok_raw(
+                        encode_command(cmd, req_id), self.timeout, req_id
+                    )
                     self._last_command_index = ok.index
                     return ok.index if ok.index is not None else 0
                 except TimeoutError:
@@ -583,14 +618,18 @@ async def _send(self, cmd: msgspec.Struct) -> int:
         self._transport.sendto(self._tx_buf)
         return 1
 
-    async def _request(self, cmd: msgspec.Struct) -> Response | None:
+    async def _request(
+        self, cmd: msgspec.Struct, timeout: float | None = None
+    ) -> Response | None:
         """Send a query command and wait for a typed response.
 
         Drains the receive queue until a ResponseMsg is found or timeout.
         Non-ResponseMsg datagrams (e.g. status broadcasts) are discarded.
 
         Args:
             cmd: Typed command struct
+            timeout: Per-call deadline; when given, the query is sent once
+                with no retries so the deadline is the caller's total wait.
 
         Returns:
             Typed Response struct, or None on timeout.
@@ -600,12 +639,15 @@ async def _request(self, cmd: msgspec.Struct) -> Response | None:
         """
         await self._ensure_endpoint()
         assert self._transport is not None
-        data = encode_command(cmd)
-        for attempt in range(self.retries + 1):
+        wait = self.timeout if timeout is None else timeout
+        attempts = self.retries + 1 if timeout is None else 1
+        for attempt in range(attempts):
+            req_id = self._request_id()
+            data = encode_command(cmd, req_id)
             try:
                 async with self._req_lock:
                     self._transport.sendto(data)
-                    end_time = time.monotonic() + self.timeout
+                    end_time = time.monotonic() + wait
                     while time.monotonic() < end_time:
                         try:
                             resp_data, _ = await asyncio.wait_for(
@@ -614,6 +656,12 @@ async def _request(self, cmd: msgspec.Struct) -> Response | None:
                             )
                             try:
                                 parsed = decode_message(resp_data)
+                                if parsed.req_id != req_id:
+                                    # A reply to a request whose caller has
+                                    # given up. Answering this one with it
+                                    # would leave every later query a reply
+                                    # behind.
+                                    continue
                                 if isinstance(parsed, ResponseMsg):
                                     return parsed.result
                                 if isinstance(parsed, ErrorMsg):
@@ -632,18 +680,20 @@ async def _request(self, cmd: msgspec.Struct) -> Response | None:
                 pass
             except Exception:
                 break
-            if attempt < self.retries:
+            if attempt < attempts - 1:
                 backoff = min(0.5, 0.05 * (2**attempt)) + random.uniform(0, 0.05)
                 await asyncio.sleep(backoff)
         return None
 
-    async def _request_ok_raw(self, data: bytes, timeout: float) -> OkMsg:
+    async def _request_ok_raw(self, data: bytes, timeout: float, req_id: int) -> OkMsg:
         """
-        Send pre-encoded binary command and wait for 'OK' or 'ERROR' reply.
+        Send pre-encoded binary command and wait for the 'OK' or 'ERROR' reply
+        carrying *req_id*; replies to abandoned requests are discarded.
 
         Args:
-            data: Pre-encoded msgpack bytes
+            data: Pre-encoded command datagram, id header included
             timeout: Timeout in seconds.
+            req_id: The id *data* carries, echoed by the reply.
 
         Returns OkMsg on OK; raises RuntimeError on ERROR, TimeoutError on timeout.
         """
@@ -661,9 +711,9 @@ async def _request_ok_raw(self, data: bytes, timeout: float) -> OkMsg:
                     )
                     try:
                         match decode_message(resp_data):
-                            case OkMsg() as ok:
+                            case OkMsg(reply_id) as ok if reply_id == req_id:
                                 return ok
-                            case ErrorMsg(message):
+                            case ErrorMsg(reply_id, message) if reply_id == req_id:
                                 raise MotionError(RobotError.from_wire(message))
                     except msgspec.ValidationError:
                         pass  # Ignore non-matching datagrams
@@ -850,15 +900,27 @@ async def angles(self) -> list[float] | None:
         resp = await self._request(AnglesCmd())
         return resp.angles if isinstance(resp, AnglesResultStruct) else None
 
-    async def io(self) -> list[int] | None:
+    async def io(self, *, timeout: float | None = None) -> list[int] | None:
         """Digital I/O status [in1, in2, out1, out2, estop].
 
+        ``timeout`` bounds setup, retries, and the reply; None uses client defaults.
+
         Category: Query
 
         Example:
             io = rbt.io()
         """
-        resp = await self._request(IOCmd())
+        if timeout is not None and (
+            isinstance(timeout, bool) or not math.isfinite(timeout) or timeout <= 0
+        ):
+            raise ValueError("I/O timeout must be positive and finite")
+        # The outer deadline also bounds endpoint setup and its retries on an
+        # absent peer; the inner one keeps the query to a single attempt.
+        try:
+            async with asyncio.timeout(timeout):
+                resp = await self._request(IOCmd(), timeout=timeout)
+        except TimeoutError:
+            return None
         return resp.io if isinstance(resp, IOResultStruct) else None
 
     async def joint_speeds(self) -> list[float] | None:
@@ -1373,6 +1435,7 @@ async def wait_status(
         end_time = time.monotonic() + timeout
 
         while time.monotonic() < end_time and not self._closed:
+            self._check_protocol()
             self._status_event.clear()
 
             # Check if we already have new data
@@ -1851,14 +1914,19 @@ async def jog_l(
 
     # --------------- IO / Gripper / Utility ---------------
 
-    async def write_io(self, index: int, value: int) -> int:
+    async def write_io(
+        self, index: int, value: int, *, timeout: float | None = None
+    ) -> int:
         """Set digital output by logical index (0 = first output pin).
 
         The firmware I/O byte layout is ``[in0, in1, out0, out1, estop, ...]``
         so logical output index 0 maps to bit position 2.
 
         Returns the command index (≥ 0) on success, -1 on failure.
 
+        ``timeout`` bounds command acceptance. TimeoutError leaves application
+        unconfirmed; None uses the client defaults.
+
         Category: I/O
 
         Example:
@@ -1870,8 +1938,15 @@ async def write_io(self, index: int, value: int) -> int:
             raise ValueError("I/O value must be 0 or 1")
         # Firmware bit layout: [in0, in1, out0, out1, estop, ...]
         firmware_index = index + 2
-        result = await self._send(WriteIOCmd(port_index=firmware_index, value=value))
-        return result
+        if timeout is not None and (
+            isinstance(timeout, bool) or not math.isfinite(timeout) or timeout <= 0
+        ):
+            raise ValueError("I/O timeout must be positive and finite")
+        async with asyncio.timeout(timeout):
+            result = await self._send(
+                WriteIOCmd(port_index=firmware_index, value=value)
+            )
+            return result
 
     async def delay(self, seconds: float) -> int:
         """Insert a non-blocking delay in the motion queue.

parol6/client/dry_run_client.py

@@ -44,6 +44,7 @@
     SelectToolCmd,
     SetTcpOffsetCmd,
     SetTcpTransformCmd,
+    WriteIOCmd,
     TeleportCmd,
     ToolActionCmd,
 )
@@ -550,7 +551,14 @@ def _simulate_cartesian_jog(self, cmd: JogLCommand) -> DryRunResult | None:
     @property
     def skill_capabilities(self) -> frozenset[str]:
         return frozenset(
-            {"motion.joint", "motion.linear", "tool.gripper", "backend.parol6"}
+            {
+                "motion.joint",
+                "motion.linear",
+                "tool.gripper",
+                "backend.parol6",
+                "io.digital",
+                "execution.preview",
+            }
         )
 
     def angles(self) -> list[float]:
@@ -605,6 +613,16 @@ def servo_j(
             return self._dispatch(build_cmd("servo_j_pose", pose, **kwargs))
         return self._dispatch(build_cmd("servo_j", angles or [], **kwargs))
 
+    def write_io(self, index: int, value: int, *, timeout: float | None = None) -> int:
+        if type(index) is not int or index not in (0, 1):
+            raise ValueError("Output index must be 0 or 1")
+        if type(value) not in (int, bool) or value not in (0, 1):
+            raise ValueError("Digital output must be 0 or 1")
+        result = self._dispatch(WriteIOCmd(port_index=index + 2, value=int(value)))
+        if result is not None and result.error is not None:
+            raise RuntimeError(str(result.error))
+        return 0
+
     def jog_j(
         self,
         joint: int = -1,

parol6/client/sync_client.py

@@ -287,13 +287,13 @@ def angles(self) -> list[float] | None:
         """
         return _run(self._inner.angles())
 
-    def io(self) -> list[int] | None:
+    def io(self, *, timeout: float | None = None) -> list[int] | None:
         """Digital I/O status.
 
         Returns:
             List of 5 integers [in1, in2, out1, out2, estop], or None on timeout.
         """
-        return _run(self._inner.io())
+        return _run(self._inner.io(timeout=timeout))
 
     def joint_speeds(self) -> list[float] | None:
         """Current joint speeds in steps per second.
@@ -851,9 +851,9 @@ def checkpoint(self, label: str) -> int:
     def wait_checkpoint(self, label: str, timeout: float = 30.0) -> bool:
         return _run(self._inner.wait_checkpoint(label, timeout=timeout))
 
-    def write_io(self, index: int, value: int) -> int:
+    def write_io(self, index: int, value: int, *, timeout: float | None = None) -> int:
         """Set digital output by logical index (0 = first output pin)."""
-        return _run(self._inner.write_io(index, value))
+        return _run(self._inner.write_io(index, value, timeout=timeout))
 
     def delay(self, seconds: float) -> int:
         """Insert a non-blocking delay in the motion queue."""

parol6/commands/base.py

@@ -11,7 +11,7 @@
 import numpy as np
 
 from parol6.config import TRACE
-from parol6.protocol.wire import CmdType, Command, CommandCode, QueryType
+from parol6.protocol.wire import CmdType, Command, CommandCode, QueryType, Response
 from parol6.server.state import ControllerState
 from parol6.utils.error_catalog import RobotError, extract_robot_error, make_error
 from parol6.utils.error_codes import ErrorCode
@@ -254,8 +254,8 @@ class QueryCommand(CommandBase[P]):
     QUERY_TYPE: ClassVar[QueryType]
 
     @abstractmethod
-    def compute(self, state: ControllerState) -> bytes:
-        """Compute the query result, pack it, and return response bytes."""
+    def compute(self, state: ControllerState) -> Response:
+        """The query's typed result; the controller packs it with the request id."""
         ...
 
     def execute_step(self, state: ControllerState) -> ExecutionStatusCode: