fix: MQTT threading model and event emitter correctness

[eman]

Fourth batch from the comprehensive repo evaluation: moves MQTT message handling off the AWS CRT network thread and fixes event emitter races. Two of these bugs were verified with minimal reproductions during the evaluation.

Bugs fixed

• All message handling ran on the AWS CRT network thread (mqtt/subscriptions.py): the raw awscrt callback did JSON decode, pydantic model_validate, and invoked user callbacks directly on the SDK's network thread. Consequences:

  • a slow/blocking user callback (e.g. the CLI monitor's synchronous CSV write) stalled all MQTT message processing
  • user callbacks ran on an undocumented SDK thread where asyncio operations are unsafe
  • the handler registry could be mutated on the event loop (subscribe/unsubscribe/resubscribe_all) while the CRT thread iterated it → RuntimeError: dictionary changed size during iteration, dropping the message — most likely during the reconnect window when queued broker messages are delivered

  The CRT callback is now trivial: it marshals the raw payload onto the event loop via the existing schedule_coroutine mechanism; parse + dispatch run there, iterating snapshots of the registries. Message ordering is preserved (call_soon_threadsafe is FIFO from a single CRT thread).

• One raising handler aborted delivery to the rest: dispatch caught only (TypeError, AttributeError, KeyError); a callback raising ValueError (or anything else) skipped the remaining handlers. Failures are now logged per handler and isolated (matching the documented EventEmitter resilience rationale).

• --unit-system silently ignored for MQTT data (unit_system.py): the preference was a ContextVar set in the caller's task. Tasks scheduled from CRT threads get a fresh context, so DeviceStatus computed fields fell back to the device's native unit — nwp-cli --unit-system metric monitor wrote Fahrenheit values labeled °C into logs and CSVs. The preference is now process-wide (module global), visible from every task and thread. Per-task unit isolation was a documented footgun nobody could actually rely on for real-time data.

• Once-listeners fired more than once (events.py, empirically verified): emit() removed once-listeners after invoking them inside the try, so a raising callback stayed registered forever, and two overlapping emits both snapshotted the list before either removed. Removal now happens synchronously before invocation, in the same event-loop slice as the snapshot.

• wait_for() leaked its listener on cancellation (events.py, empirically verified): only TimeoutError triggered cleanup; a cancelled waiter left a once-listener that would later fire and set a result on a dead future. Cleanup moved to finally.

• wait_for() docstring examples broke user code (events.py, mqtt_events.py): examples showed args, _ = await wait_for(...) but it returns just the args tuple.

Tests

New tests/test_threading_model.py (13 tests): message from a foreign thread dispatches on the loop thread with parsed payload; registry mutation during dispatch is safe; raising handler doesn't block others; invalid JSON logged not raised; unit system visible in tasks scheduled from foreign threads and in plain threads; raising once-listener removed; concurrent emits single-fire; wait_for cleanup on cancellation/timeout/success.

Validation

• pytest: 521 passed (2 consecutive runs)
• ruff check / ruff format --check: clean
• pyright src/nwp500: 0 errors
• mypy src/nwp500: no issues

[Copilot]

Pull request overview

This PR improves the reliability and correctness of the library’s MQTT and eventing layers by ensuring message parsing/dispatch occurs on the asyncio event loop (not the AWS CRT network thread), hardening callback isolation, and fixing EventEmitter once/wait-for race/leak behaviors. It also adjusts unit-system configuration semantics so real-time MQTT-driven model validation consistently honors the configured preference across threads/tasks.

Changes:

• Move MQTT message parsing + handler dispatch off the AWS CRT callback thread and onto the event loop, with snapshot-based registry iteration and per-handler error isolation.
• Fix EventEmitter once-listener and wait_for() cleanup races/leaks; update doc/examples to match actual wait_for() return shape.
• Replace unit-system ContextVar storage with a process-wide setting; add regression tests covering threading/ordering and event emitter behavior.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
tests/test_threading_model.py	Adds regression coverage for MQTT threading dispatch, handler isolation, unit-system visibility across threads, and EventEmitter once/wait_for fixes.
src/nwp500/unit_system.py	Changes unit-system preference storage from context-local to process-wide to work correctly with cross-thread MQTT scheduling.
src/nwp500/mqtt/subscriptions.py	Makes the AWS CRT message callback trivial and marshals parse/dispatch onto the event loop; dispatch iterates registry snapshots and isolates handler failures.
src/nwp500/mqtt_events.py	Fixes wait_for() example usage to match its actual return value (args tuple only).
src/nwp500/events.py	Removes once-listeners before invocation to prevent double-fire/retention on exceptions; ensures wait_for() always unregisters listeners via finally.
CHANGELOG.rst	Documents the fixed threading, event emitter, unit-system, and documentation issues under Unreleased.