Support bounded digital I/O for named signal skills

README.md

@@ -451,3 +451,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
@@ -257,7 +258,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,
@@ -583,14 +588,43 @@ 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:
+    def _drop_stale_replies(self) -> None:
+        """Discard replies already queued when a new request is about to go out.
+
+        Nothing awaits them: their caller's deadline expired mid-flight. The
+        wire carries no request id, so handing one to the next request would
+        answer it with the wrong struct and leave every later query one reply
+        behind.
+        """
+        kept = []
+        while True:
+            try:
+                item = self._rx_queue.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+            try:
+                stale = isinstance(
+                    decode_message(item[0]), (ResponseMsg, OkMsg, ErrorMsg)
+                )
+            except msgspec.DecodeError:
+                stale = False
+            if not stale:
+                kept.append(item)
+        for item in kept:
+            self._rx_queue.put_nowait(item)
+
+    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.
@@ -601,11 +635,14 @@ 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):
             try:
                 async with self._req_lock:
+                    self._drop_stale_replies()
                     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(
@@ -632,7 +669,7 @@ 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
@@ -652,6 +689,7 @@ async def _request_ok_raw(self, data: bytes, timeout: float) -> OkMsg:
 
         end_time = time.monotonic() + timeout
         async with self._req_lock:
+            self._drop_stale_replies()
             self._transport.sendto(data)
             while time.monotonic() < end_time:
                 try:
@@ -850,15 +888,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:
@@ -1847,14 +1897,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:
@@ -1866,8 +1921,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

@@ -43,6 +43,7 @@
     SelectToolCmd,
     SetTcpOffsetCmd,
     SetTcpTransformCmd,
+    WriteIOCmd,
     TeleportCmd,
     ToolActionCmd,
 )
@@ -546,7 +547,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]:
@@ -601,6 +609,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 delay(self, seconds: float = 0.0) -> None:
         pass
 

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."""

pyproject.toml

@@ -46,7 +46,7 @@ dependencies = [
     "psutil>=5.9",
     "msgspec>=0.18",
     "ormsgpack>=1.4.0",
-    "waldoctl @ git+https://github.com/Jepson2k/waldoctl.git@v0.15.0",
+    "waldoctl @ git+https://github.com/Jepson2k/waldoctl.git@v0.16.0",
 ]
 
 [tool.setuptools.packages.find]

tests/integration/test_digital_io.py

@@ -0,0 +1,82 @@
+"""Digital I/O uses logical output indices and per-call reply deadlines."""
+
+import asyncio
+import socket
+import time
+
+import pytest
+from waldoctl.skills import skill
+
+from parol6 import AsyncRobotClient
+
+
+def test_digital_io_readback_and_missing_peer_deadlines(client, server_proc):
+    before = client.io(timeout=2)
+    assert before is not None
+    try:
+        assert client.write_io(0, 1 - before[2], timeout=2) >= 0
+        assert client.wait_status(lambda s: s.io[2] == 1 - before[2], timeout=2)
+        assert client.io(timeout=2)[2] == 1 - before[2]
+    finally:
+        client.write_io(0, before[2], timeout=2)
+
+    async def missing_peer():
+        @skill(id="test.io_deadline", version="1.0.0")
+        async def query(rbt):
+            return await rbt.io(timeout=0.05)
+
+        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as silent:
+            silent.bind(("127.0.0.1", 0))
+            async with AsyncRobotClient(
+                port=silent.getsockname()[1], timeout=5, retries=3
+            ) as absent:
+                start = time.monotonic()
+                assert await query.async_call(absent) is None
+                assert time.monotonic() - start < 1.0, (
+                    "query ignored its per-call deadline"
+                )
+                start = time.monotonic()
+                with pytest.raises(TimeoutError):
+                    await absent.write_io(0, 1, timeout=0.05)
+                assert time.monotonic() - start < 1.0, (
+                    "write ignored its per-call deadline"
+                )
+                for invalid in (0, -1, float("nan"), float("inf"), True):
+                    with pytest.raises(ValueError):
+                        await absent.io(timeout=invalid)
+                    with pytest.raises(ValueError):
+                        await absent.write_io(0, 1, timeout=invalid)
+
+    asyncio.run(missing_peer())
+
+
+def test_late_replies_never_answer_the_next_request(ports, server_proc):
+    """A reply that lands after its caller's deadline expired is not served
+    to the next request: the wire carries no request ids, so a stale reply
+    would answer the wrong query and leave every later one a reply behind."""
+    from parol6.protocol.wire import IOResultStruct, pack_ok, pack_response
+
+    async def scenario():
+        async with AsyncRobotClient(
+            host=ports.server_ip, port=ports.server_port, timeout=5.0
+        ) as rbt:
+            await rbt._ensure_endpoint()
+            peer = (ports.server_ip, ports.server_port)
+            rbt._rx_queue.put_nowait(
+                (pack_response(IOResultStruct(io=[0, 0, 0, 0, 1])), peer)
+            )
+            pose = await rbt.pose()
+            assert pose is not None and len(pose) == 6
+            assert await rbt.angles() is not None
+            rbt._rx_queue.put_nowait((pack_ok(), peer))
+            index = await rbt.delay(0.1)
+            assert index >= 1, "a stale index-less OK must not stand in for the ack"
+            assert await rbt.wait_command(index, timeout=5)
+            # A deadline that lapses mid-flight: the reply lands afterwards
+            # and must be dropped before the next query goes out.
+            assert await rbt.io(timeout=1e-4) is None
+            await asyncio.sleep(0.1)
+            pose = await rbt.pose()
+            assert pose is not None and len(pose) == 6
+
+    asyncio.run(scenario())