feat(host): thread-safe send/query (io lock) + py.typed (0.2.1)

marlin_host/__init__.py

@@ -44,4 +44,4 @@
     "reset_line_number",
 ]
 
-__version__ = "0.2.0"
+__version__ = "0.2.1"

marlin_host/host.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import threading
 import time
 from collections.abc import Callable, Iterable, Iterator, Mapping
 from dataclasses import dataclass
@@ -133,6 +134,11 @@ def __init__(
         self._max_resends = max_resends
         self._connect_probes = connect_probes
         self._on_action = on_action
+        # Serialises transport access so send()/query()/stream() are mutually
+        # thread-safe — e.g. a worker streaming while another thread issues a
+        # manual command. emergency_stop() deliberately bypasses this (M112 must
+        # preempt, not wait behind an in-flight line).
+        self._io_lock = threading.Lock()
         self._line_number = 0
         self._halted = False
         self._connected = False
@@ -292,8 +298,9 @@ def send(self, command: str) -> MarlinResponse:
         """
         if self._halted:
             raise HaltError("controller is halted; a reset is required")
-        self._write(command)
-        return self._await_terminal(command)
+        with self._io_lock:
+            self._write(command)
+            return self._await_terminal(command)
 
     def query(self, command: str) -> list[MarlinResponse]:
         """Send a reporting command and return its intermediate response lines.
@@ -304,8 +311,9 @@ def query(self, command: str) -> list[MarlinResponse]:
         if self._halted:
             raise HaltError("controller is halted; a reset is required")
         collected: list[MarlinResponse] = []
-        self._write(command)
-        self._await_terminal(command, collected)
+        with self._io_lock:
+            self._write(command)
+            self._await_terminal(command, collected)
         return collected
 
     def capabilities(self) -> Profile:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "marlin-host"
-version = "0.2.0"
+version = "0.2.1"
 description = "Host-side Python library for the Marlin firmware serial protocol."
 readme = "README.md"
 requires-python = ">=3.11"

tests/test_host.py

@@ -451,3 +451,39 @@ def test_position_reads_the_m114_report() -> None:
 def test_position_empty_when_no_report_line() -> None:
     host = MarlinHost(FakeTransport(responder=lambda _line: ["ok"]))
     assert host.position() == {}
+
+
+def test_send_and_query_are_mutually_exclusive_across_threads() -> None:
+    # The io lock must serialise transport access so a worker streaming and
+    # another thread's manual command never interleave on the wire.
+    import threading
+    import time as _time
+
+    ops: list[str] = []
+
+    class _SlowTransport:
+        def write_line(self, line: str) -> None:
+            ops.append(f"w:{line}")
+
+        def read_line(self, timeout: float | None = None) -> str | None:
+            ops.append("r0")
+            _time.sleep(0.05)
+            ops.append("r1")
+            return "ok"
+
+        def close(self) -> None:
+            pass
+
+    host = MarlinHost(_SlowTransport())
+    worker = threading.Thread(target=lambda: host.send("A"))
+    worker.start()
+    _time.sleep(0.01)  # let A enter the locked region first
+    host.send("B")
+    worker.join()
+
+    # Each send is [w:X, r0, r1]; the lock keeps the two triples contiguous.
+    # Without it, the 0.05s read overlap would interleave them.
+    assert len(ops) == 6
+    assert ops[1:3] == ["r0", "r1"]
+    assert ops[4:6] == ["r0", "r1"]
+    assert ops[0].startswith("w:") and ops[3].startswith("w:")