From 508891b24864ef202c40e0efe2597f6e0a3ddb62 Mon Sep 17 00:00:00 2001
From: ceej640 <42260127+Ceej640@users.noreply.github.com>
Date: Sat, 30 May 2026 22:10:32 -0400
Subject: [PATCH 1/2] Add hardware testing scaffold

---
 docs/testing-hardware.md                 |  47 ++++++
 gently/hardware/testing.py               | 206 +++++++++++++++++++++++
 pyproject.toml                           |   5 +
 tests/conftest.py                        |  45 +++++
 tests/hardware/__init__.py               |   1 +
 tests/hardware/test_live_device_layer.py |  24 +++
 tests/hardware/test_testing_helpers.py   |  69 ++++++++
 7 files changed, 397 insertions(+)
 create mode 100644 docs/testing-hardware.md
 create mode 100644 gently/hardware/testing.py
 create mode 100644 tests/hardware/__init__.py
 create mode 100644 tests/hardware/test_live_device_layer.py
 create mode 100644 tests/hardware/test_testing_helpers.py

diff --git a/docs/testing-hardware.md b/docs/testing-hardware.md
new file mode 100644
index 00000000..5188cb34
--- /dev/null
+++ b/docs/testing-hardware.md
@@ -0,0 +1,47 @@
+# Hardware Testing
+
+Gently hardware tests are split into two groups:
+
+- `hardware`: offline contract tests that use fakes/mocks and should run in CI.
+- `live_hardware`: opt-in checks that talk to a connected microscope/device layer.
+
+## Offline Hardware Tests
+
+Use the package-level helpers in `gently.hardware.testing` when testing tools or
+workflows that normally need a microscope client:
+
+```python
+from gently.hardware.testing import MockQueueServerClient
+
+client = MockQueueServerClient(stage_position=(10.0, 20.0))
+await client.move_to_position(100.0, 200.0)
+assert client.recorded_calls("move_to_position")
+```
+
+Run the offline hardware contracts with:
+
+```shell
+pytest tests/hardware -m hardware
+```
+
+## Live Hardware Tests
+
+Live tests are skipped unless explicitly enabled. Start the device layer first,
+then run:
+
+```shell
+pytest tests/hardware -m live_hardware --run-hardware --hardware-url http://127.0.0.1:60610
+```
+
+`--hardware-url` defaults to `GENTLY_HARDWARE_URL` when set, otherwise
+`http://127.0.0.1:60610`.
+
+## Adding New Hardware Tests
+
+Use `hardware` for deterministic tests that should pass without a microscope.
+Use `live_hardware` only when the test validates real hardware connectivity,
+device-layer state reporting, or behavior that cannot be represented by a mock.
+
+Live tests should assert health and invariants rather than move devices through
+large ranges. Keep destructive or sample-altering procedures manual unless they
+have explicit safety bounds and operator confirmation.
diff --git a/gently/hardware/testing.py b/gently/hardware/testing.py
new file mode 100644
index 00000000..40803787
--- /dev/null
+++ b/gently/hardware/testing.py
@@ -0,0 +1,206 @@
+"""Testing helpers for hardware-dependent code.
+
+The helpers in this module are intentionally small and dependency-free so tests
+can exercise agent/tool behavior without importing Micro-Manager or talking to
+physical devices.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict, deque
+from dataclasses import dataclass, field
+from typing import Any, AsyncIterator, Deque, Dict, Iterable, List, Mapping, Optional, Tuple
+
+
+@dataclass(frozen=True)
+class HardwareCondition:
+    """Result of checking whether a hardware capability is available."""
+
+    name: str
+    available: bool
+    details: Mapping[str, Any] = field(default_factory=dict)
+    error: Optional[str] = None
+
+    def require(self) -> None:
+        """Raise an AssertionError when the condition is not available."""
+        if not self.available:
+            message = f"Hardware condition unavailable: {self.name}"
+            if self.error:
+                message = f"{message}: {self.error}"
+            raise AssertionError(message)
+
+
+class MockQueueServerClient:
+    """Scriptable async fake for tests that normally need a device layer.
+
+    It implements the common methods used by Gently tools and records every
+    call. Tests can use :meth:`script_response` or :meth:`fail` to model device
+    success and failure paths deterministically.
+    """
+
+    def __init__(
+        self,
+        *,
+        stage_position: Tuple[float, float] = (0.0, 0.0),
+        piezo_position: float = 0.0,
+        has_sam: bool = True,
+        device_state: Optional[Mapping[str, Any]] = None,
+    ):
+        self.stage_position = stage_position
+        self.piezo_position = piezo_position
+        self.has_sam = has_sam
+        self.is_connected = False
+        self.calls: List[Dict[str, Any]] = []
+        self._responses: Dict[str, Deque[Any]] = defaultdict(deque)
+        self._failures: Dict[str, Exception] = {}
+        self._stream_states: Deque[Mapping[str, Any]] = deque()
+        self._device_state = dict(device_state or {})
+
+    def script_response(self, method: str, *responses: Any) -> None:
+        """Queue one or more responses for a method name."""
+        self._responses[method].extend(responses)
+
+    def script_stream(self, *states: Mapping[str, Any]) -> None:
+        """Queue device-state events for :meth:`stream_device_states`."""
+        self._stream_states.extend(states)
+
+    def fail(self, method: str, error: Exception) -> None:
+        """Make a method raise ``error`` until the failure is cleared."""
+        self._failures[method] = error
+
+    def clear_failure(self, method: str) -> None:
+        self._failures.pop(method, None)
+
+    def recorded_calls(self, method: Optional[str] = None) -> List[Dict[str, Any]]:
+        """Return all recorded calls, optionally filtered by method."""
+        if method is None:
+            return list(self.calls)
+        return [call for call in self.calls if call["method"] == method]
+
+    def _record(self, method: str, **payload: Any) -> None:
+        self.calls.append({"method": method, **payload})
+
+    def _next_response(self, method: str, default: Any) -> Any:
+        if method in self._failures:
+            raise self._failures[method]
+        if self._responses[method]:
+            response = self._responses[method].popleft()
+            if isinstance(response, Exception):
+                raise response
+            if callable(response):
+                return response()
+            return response
+        return default
+
+    async def connect(self) -> bool:
+        self._record("connect")
+        result = self._next_response("connect", True)
+        self.is_connected = bool(result)
+        return self.is_connected
+
+    async def disconnect(self) -> None:
+        self._record("disconnect")
+        self.is_connected = False
+
+    async def get_stage_position(self) -> Tuple[float, float]:
+        self._record("get_stage_position")
+        return self._next_response("get_stage_position", self.stage_position)
+
+    async def move_to_position(self, x: float, y: float) -> Dict[str, Any]:
+        self._record("move_to_position", x=x, y=y)
+        self.stage_position = (float(x), float(y))
+        return self._next_response(
+            "move_to_position",
+            {"success": True, "x": self.stage_position[0], "y": self.stage_position[1]},
+        )
+
+    async def get_piezo_position(self) -> float:
+        self._record("get_piezo_position")
+        return self._next_response("get_piezo_position", self.piezo_position)
+
+    async def get_device_state(self, refresh: bool = False) -> Dict[str, Any]:
+        self._record("get_device_state", refresh=refresh)
+        state = {
+            "available": True,
+            "positions": {
+                "stage": {"x": self.stage_position[0], "y": self.stage_position[1]},
+                "piezo": self.piezo_position,
+            },
+            **self._device_state,
+        }
+        return self._next_response("get_device_state", state)
+
+    async def stream_device_states(self, timeout: Optional[float] = None) -> AsyncIterator[Dict[str, Any]]:
+        self._record("stream_device_states", timeout=timeout)
+        if not self._stream_states:
+            yield await self.get_device_state(refresh=False)
+            return
+        while self._stream_states:
+            yield dict(self._stream_states.popleft())
+
+    async def capture_bottom_image(self, exposure_ms: float = 10.0) -> Dict[str, Any]:
+        self._record("capture_bottom_image", exposure_ms=exposure_ms)
+        return self._next_response(
+            "capture_bottom_image",
+            {"success": True, "image": [[0]], "exposure_ms": exposure_ms},
+        )
+
+    async def capture_for_marking(self, exposure_ms: float = 10.0) -> Dict[str, Any]:
+        self._record("capture_for_marking", exposure_ms=exposure_ms)
+        return self._next_response(
+            "capture_for_marking",
+            {"success": True, "image": [[0]], "stage_position": self.stage_position},
+        )
+
+    async def capture_lightsheet_image(self, **kwargs: Any) -> Dict[str, Any]:
+        self._record("capture_lightsheet_image", **kwargs)
+        return self._next_response(
+            "capture_lightsheet_image",
+            {"success": True, "image": [[0]], "shape": (1, 1), **kwargs},
+        )
+
+    async def acquire_volume(self, **kwargs: Any) -> Dict[str, Any]:
+        self._record("acquire_volume", **kwargs)
+        return self._next_response(
+            "acquire_volume",
+            {"success": True, "volume": None, "shape": (0,), **kwargs},
+        )
+
+    async def detect_embryos(self, **kwargs: Any) -> Dict[str, Any]:
+        self._record("detect_embryos", **kwargs)
+        return self._next_response("detect_embryos", {"success": True, "embryos": []})
+
+    async def set_led(self, state: str) -> Dict[str, Any]:
+        self._record("set_led", state=state)
+        return self._next_response("set_led", {"success": True, "state": state})
+
+    async def get_led_status(self) -> Dict[str, Any]:
+        self._record("get_led_status")
+        return self._next_response("get_led_status", {"state": "unknown"})
+
+    async def set_laser_power(self, wavelength: int, pct: float) -> Dict[str, Any]:
+        self._record("set_laser_power", wavelength=wavelength, pct=pct)
+        return self._next_response(
+            "set_laser_power",
+            {"success": True, "wavelength": wavelength, "power_pct": pct},
+        )
+
+    async def get_laser_power(self, wavelength: int) -> Dict[str, Any]:
+        self._record("get_laser_power", wavelength=wavelength)
+        return self._next_response(
+            "get_laser_power",
+            {"wavelength": wavelength, "power_pct": 0.0},
+        )
+
+    async def set_temperature(self, target_c: float) -> Dict[str, Any]:
+        self._record("set_temperature", target_c=target_c)
+        return self._next_response("set_temperature", {"success": True, "target_c": target_c})
+
+    async def get_temperature(self) -> Dict[str, Any]:
+        self._record("get_temperature")
+        return self._next_response("get_temperature", {"current_c": None, "target_c": None})
+
+
+def summarize_conditions(conditions: Iterable[HardwareCondition]) -> Dict[str, bool]:
+    """Return a compact availability map for a set of hardware conditions."""
+    return {condition.name: condition.available for condition in conditions}
diff --git a/pyproject.toml b/pyproject.toml
index 3f5a7ce7..821e985b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -62,3 +62,8 @@ exclude = ["tests*", "benchmarks*", "scripts*", "examples*"]
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 asyncio_mode = "auto"
+markers = [
+    "hardware: tests for hardware contracts using offline fakes or mocks",
+    "live_hardware: tests that require a connected physical microscope/device layer",
+    "device_layer: tests that exercise the device-layer HTTP API",
+]
diff --git a/tests/conftest.py b/tests/conftest.py
index be2fcb53..8529273e 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -2,12 +2,43 @@
 Shared fixtures for gently tests.
 """
 
+import os
 import pytest
 from pathlib import Path
 
 from gently.core.event_bus import EventBus
 
 
+def pytest_addoption(parser):
+    """Command-line switches for opt-in hardware test runs."""
+    group = parser.getgroup("gently hardware")
+    group.addoption(
+        "--run-hardware",
+        action="store_true",
+        default=False,
+        help="run tests marked live_hardware against a connected device layer",
+    )
+    group.addoption(
+        "--hardware-url",
+        action="store",
+        default=os.environ.get("GENTLY_HARDWARE_URL", "http://127.0.0.1:60610"),
+        help="device-layer URL for tests marked live_hardware",
+    )
+
+
+def pytest_collection_modifyitems(config, items):
+    """Skip live hardware tests unless explicitly requested."""
+    if config.getoption("--run-hardware"):
+        return
+
+    skip_live_hardware = pytest.mark.skip(
+        reason="requires --run-hardware and a connected device layer"
+    )
+    for item in items:
+        if "live_hardware" in item.keywords:
+            item.add_marker(skip_live_hardware)
+
+
 @pytest.fixture
 def config_dir(tmp_path):
     """Temporary config directory for mesh/transfer state files."""
@@ -56,3 +87,17 @@ def file_context_store(tmp_path):
 def event_bus():
     """Fresh EventBus, isolated from global singleton."""
     return EventBus(history_size=50)
+
+
+@pytest.fixture
+def hardware_url(pytestconfig):
+    """Device-layer URL used by opt-in live hardware tests."""
+    return pytestconfig.getoption("--hardware-url")
+
+
+@pytest.fixture
+def live_hardware_url(pytestconfig, hardware_url):
+    """Return the configured hardware URL, skipping unless live tests are enabled."""
+    if not pytestconfig.getoption("--run-hardware"):
+        pytest.skip("requires --run-hardware and a connected device layer")
+    return hardware_url
diff --git a/tests/hardware/__init__.py b/tests/hardware/__init__.py
new file mode 100644
index 00000000..13cc9672
--- /dev/null
+++ b/tests/hardware/__init__.py
@@ -0,0 +1 @@
+"""Hardware test helpers and contracts."""
diff --git a/tests/hardware/test_live_device_layer.py b/tests/hardware/test_live_device_layer.py
new file mode 100644
index 00000000..1bf77c1f
--- /dev/null
+++ b/tests/hardware/test_live_device_layer.py
@@ -0,0 +1,24 @@
+import pytest
+
+
+pytestmark = [
+    pytest.mark.live_hardware,
+    pytest.mark.device_layer,
+    pytest.mark.asyncio,
+]
+
+
+async def test_live_device_layer_reports_state(live_hardware_url):
+    from gently.hardware.dispim.client import DiSPIMMicroscope
+
+    client = DiSPIMMicroscope(http_url=live_hardware_url)
+    try:
+        connected = await client.connect()
+        assert connected, f"could not connect to device layer at {live_hardware_url}"
+
+        state = await client.get_device_state(refresh=True)
+
+        assert isinstance(state, dict)
+        assert state, "device layer returned an empty state payload"
+    finally:
+        await client.disconnect()
diff --git a/tests/hardware/test_testing_helpers.py b/tests/hardware/test_testing_helpers.py
new file mode 100644
index 00000000..261b962c
--- /dev/null
+++ b/tests/hardware/test_testing_helpers.py
@@ -0,0 +1,69 @@
+import pytest
+
+from gently.hardware.testing import (
+    HardwareCondition,
+    MockQueueServerClient,
+    summarize_conditions,
+)
+
+
+pytestmark = pytest.mark.hardware
+
+
+@pytest.mark.asyncio
+async def test_mock_client_records_stage_moves():
+    client = MockQueueServerClient(stage_position=(10.0, 20.0))
+
+    await client.move_to_position(1200.5, -25.0)
+
+    assert await client.get_stage_position() == (1200.5, -25.0)
+    assert client.recorded_calls("move_to_position") == [
+        {"method": "move_to_position", "x": 1200.5, "y": -25.0}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_mock_client_scripts_failures():
+    client = MockQueueServerClient()
+    client.fail("acquire_volume", RuntimeError("camera timeout"))
+
+    with pytest.raises(RuntimeError, match="camera timeout"):
+        await client.acquire_volume(num_slices=5)
+
+    assert client.recorded_calls("acquire_volume")[0]["num_slices"] == 5
+
+
+@pytest.mark.asyncio
+async def test_mock_client_streams_scripted_device_states():
+    client = MockQueueServerClient()
+    client.script_stream(
+        {"positions": {"stage": {"x": 1.0, "y": 2.0}}},
+        {"positions": {"stage": {"x": 3.0, "y": 4.0}}},
+    )
+
+    states = []
+    async for state in client.stream_device_states(timeout=0.1):
+        states.append(state)
+
+    assert [state["positions"]["stage"]["x"] for state in states] == [1.0, 3.0]
+    assert client.recorded_calls("stream_device_states")[0]["timeout"] == 0.1
+
+
+def test_hardware_condition_require_raises_with_context():
+    condition = HardwareCondition(
+        name="device-layer",
+        available=False,
+        error="not reachable",
+    )
+
+    with pytest.raises(AssertionError, match="device-layer: not reachable"):
+        condition.require()
+
+
+def test_summarize_conditions_reports_availability():
+    conditions = [
+        HardwareCondition("stage", True),
+        HardwareCondition("camera", False, error="not configured"),
+    ]
+
+    assert summarize_conditions(conditions) == {"stage": True, "camera": False}

From af216c2551896b55567187e2a541c2d242b2a9c8 Mon Sep 17 00:00:00 2001
From: Johnson <johnsonc5@hhmi.org>
Date: Mon, 1 Jun 2026 00:45:05 -0400
Subject: [PATCH 2/2] Document hardware simulation coverage layers

---
 docs/testing-hardware.md | 19 +++++++++++++++++++
 1 file changed, 19 insertions(+)

diff --git a/docs/testing-hardware.md b/docs/testing-hardware.md
index 5188cb34..2cd65fea 100644
--- a/docs/testing-hardware.md
+++ b/docs/testing-hardware.md
@@ -45,3 +45,22 @@ device-layer state reporting, or behavior that cannot be represented by a mock.
 Live tests should assert health and invariants rather than move devices through
 large ranges. Keep destructive or sample-altering procedures manual unless they
 have explicit safety bounds and operator confirmation.
+
+## Simulation Coverage Matrix
+
+Before adding a simulator, decide what risk the test is meant to catch. Gently
+needs several fidelity layers rather than one generic simulated microscope:
+
+| Layer | Purpose | Typical coverage | Default suite |
+| --- | --- | --- | --- |
+| Device command contract | Check tool/device API semantics and safety gates | command payloads, range checks, error propagation, state shape | yes |
+| Hardware digital twin | Check stateful device behavior without live hardware | queue timing, stage/camera/laser/temperature state, retries | opt-in or CI service |
+| Optical/perception simulator | Check whether perception/control logic handles images | sample rendering, focus, drift, noise, segmentation outputs | targeted |
+| Sample dynamics simulator | Check scientific state over time | development, motion, perturbation, damage/exposure effects | targeted |
+| End-to-end rehearsal | Check the full experiment loop | plan, acquire, perceive, decide, recover, export | opt-in |
+
+The helpers in `gently.hardware.testing` cover the device command contract
+layer. They are intentionally small fakes, not an optical or biological
+simulation. A richer simulator should declare which layer it belongs to, which
+failure modes it models, and which real-world behaviors it deliberately leaves
+out.