[burnpanck]

Pull Request: Structured concurrency and scoped lifespans (HasLifespan)

This PR sharpens PyLabRobot's asynchronous execution semantics across device lifespans, background task ownership, cancellation and timeout boundaries, I/O framing, and temporary physical states (such as shaking or freedrive) by migrating the codebase to AnyIO structured concurrency (anyio>=4.7). Previously, imperative setup()/stop() calls and detached asyncio tasks/threads left cleanup, partial-initialization rollback, and mid-command cancellation under-specified—allowing failed startups to leak handles, timeouts to abandon in-flight hardware writes, and background readers to outlive their transports. Binding every resource and worker to a lexical scope (async with) makes hardware teardown, error recovery, and cancellation deterministic across both asyncio and trio.

TL;DR — Key Semantic Changes

1. Lexical device & transport lifespans (async with): Every resource-owning driver, legacy Machine, Visualizer, and pylabrobot.io transport inherits from HasLifespan. Entering async with device: guarantees the device is initialized on entry and inert after exit; failed partial entry or body exceptions unwind all acquired resources in reverse order.
2. Shielded safety teardown & exception chaining: Cleanup actions registered via AsyncExitStackWithShielding are all attempted in reverse order even if an earlier step fails, and async safety actions (e.g., stopping heaters/shakers/pumps before closing the transport) run under cancellation shielding. Cleanup failures propagate to the caller with any body/cancellation error preserved on __context__.
3. Interactive setup() / stop() compatibility bridge: await device.setup() and await device.stop() remain available as a deprecated, asyncio-only bridge for cross-cell notebook sessions (backed by an application-global manager task), but are forbidden inside library/driver code.
4. Level-triggered cancellation & commit boundaries: Deadlines use with anyio.fail_after(timeout):. Cancellation before a command is dispatched aborts cleanly; once a hardware write or stateful SDK call is dispatched, the driver/transport reconciles the finite reply or abort path before propagating cancellation so hardware is never left mid-transaction.
5. Scoped temporary operations (shaking, freedrive, objective_maintenance, acquisition): Long-lived physical states expose nested async with context managers inside an active device lifespan that reject conflicting imperative calls while active and guarantee shielded stop on scope exit.
6. Single ownership vs. borrowing (_enter_lifespan): Composite drivers implement _enter_lifespan(self, stack, **kwargs) to enter owned transports and task groups (await stack.enter_task_group()) while leaving borrowed shared interfaces (InhecoTECControlBox, VSpin, injected HTTP) untouched on teardown. Overriding setup, stop, __aenter__, or __aexit__ is rejected at class creation.
7. Sharpened I/O stream, boundary & connection contracts (pylabrobot.io): Transports distinguish per-call lifespans (IOBase), uninterrupted handles (ContinuousConnectionIO: Serial, HID), and explicit mid-lifespan reconnection (ReconnectableConnectionIO: USB, FTDI, Socket). read(size) has true byte-stream semantics (EndOfStream on EOF, unreturned bytes retained on cancellation), while read_to_boundary() / discard_to_boundary() handle packet- and report-framed protocols.
8. Dual-backend test harness (AnyioTestBase & MockIO): Async unit tests inherit from AnyioTestBase (running each test in a fresh lifespan under both asyncio and trio) and use loop-agnostic MockIO doubles instead of unittest.IsolatedAsyncioTestCase and AsyncMock.

---

Detailed User & Developer Reference

User-Facing Details

• Using devices in a scope:

  async with lh, plate_reader:
    await lh.move_plate(plate, plate_reader)
    async with plate_reader.shaking(speed=500):
      await anyio.sleep(10)
    data = await plate_reader.read_luminescence(focal_height=0)

• Non-reentrant lifespans: Entering an already-active HasLifespan instance concurrently or recursively raises RuntimeError.
• Scoped operation inventory:
  • shaking(...): Legacy Shaker / HeaterShaker, Incubator (Cytomat, Liconic, InhecoIncubatorShakerStack), BigBearOrbitalShaker, BioShake, InhecoThermoShake, BioTek plate readers (Cytation1, Cytation5, SynergyH1 + legacy backends), and SpectraMax plate readers (v1 + legacy).
  • freedrive(...): PreciseFlex, XArm6, and legacy ExperimentalSCARA.
  • objective_maintenance(...): ImageXpress Pico (v1 and legacy backend).
  • acquisition(): AravisCamera / CytationMicroscope and legacy CytationBackend / Cytation5Backend.
• EventBus integration: HasLifespan emits correlated .setup.started / .completed / .failed and .stop.started / .completed / .failed events via _lifespan_event_info(). Failed setup (including rollback of partial entry) emits only setup.failed; stop.* events fire only after setup has succeeded.

Developer-Facing Details

• Writing a driver lifespan (pylabrobot/concurrency.py):

  class MyDriver(HasLifespan):
    async def _enter_lifespan(self, stack: AsyncExitStackWithShielding) -> None:
      await super()._enter_lifespan(stack)
      await stack.enter_async_context(self._io.lifespan())
      tg = await stack.enter_task_group()
      stack.callback(tg.cancel_scope.cancel)
      tg.start_soon(self._reader_loop)
      stack.push_shielded_async_callback(self._safe_park, shield_timeout=5.0)
      await self._initialize()

• Additional concurrency primitives:
  • non_cancelling_task_group(): runs finite concurrent child tasks without cancelling siblings on ordinary child errors, raising SiblingTaskErrors(ExceptionGroup) after all children finish.
  • attempt_in_order(*actions): runs a sequence of cleanup callables, attempting all of them even if earlier ones fail and chaining exceptions via __context__.
• Mid-lifespan reconnection: Drivers that replace an owned transport while remaining active (BenchCel, HighResSampleStorage, ExperimentalTecanInfinite200ProBackend) keep a reusable nested AsyncExitStackWithShielding in _enter_lifespan and close/re-enter that nested stack on reconnect.
• Blocking SDK & transport calls: Blocking calls run in non-abandoning anyio.to_thread.run_sync workers (or a dedicated single-thread executor in HID to preserve HIDAPI thread affinity) so cancellation waits for in-flight native calls to complete before closing handles.
• Contributor guides & migration skill: See docs/contributor_guide/structured-concurrency.md, docs/contributor_guide/structured-concurrency-gotchas.md, and .agents/skills/convert-backend-structured-concurrency/SKILL.md.