fix: Track fire-and-forget flush tasks in CommandCoalescer (#65)

* fix: track fire-and-forget flush tasks in CommandCoalescer

Maintain a _pending_tasks set so that background flush tasks created by
asyncio.ensure_future are tracked. Tasks are added on creation and
discarded via done-callbacks on completion.

flush_all() now awaits any in-flight flush tasks before returning,
ensuring clean shutdown without unobserved task warnings.

Closes #53

* fix: snapshot _pending_tasks before gather and improve flush_all test

Address review feedback:
- Snapshot _pending_tasks into a list before asyncio.gather to avoid
  RuntimeError from set mutation during iteration by done-callbacks.
- Discard only the awaited snapshot instead of clear() to preserve
  tasks added concurrently.
- Rewrite test_flush_all_awaits_inflight_tasks to use a blocking
  send_fn gated by asyncio.Event, verifying flush_all genuinely
  awaits an in-flight background flush task.

Author: kclif9

src/actron_neo_api/actron.py

@@ -84,6 +84,7 @@ def __init__(
         self._state_manager = state_manager
         self._debounce = debounce_seconds
         self._batches: dict[str, _PendingBatch] = {}
+        self._pending_tasks: set[asyncio.Task[None]] = set()
 
     @property
     def debounce_seconds(self) -> float:
@@ -133,21 +134,35 @@ async def enqueue(self, serial_number: str, command: dict[str, Any]) -> None:
 
         def _schedule_flush(sn: str = serial_number) -> None:
             task = asyncio.ensure_future(self._flush(sn))
+            self._pending_tasks.add(task)
+            task.add_done_callback(self._pending_tasks.discard)
             task.add_done_callback(self._flush_task_done)
 
         batch.timer = loop.call_later(self._debounce, _schedule_flush)
 
         await future
 
     async def flush_all(self) -> None:
-        """Flush every pending batch immediately, cancelling debounce timers."""
+        """Flush every pending batch immediately, cancelling debounce timers.
+
+        Also awaits any in-flight flush tasks so that all pending work
+        completes before this method returns.
+        """
         serials = list(self._batches.keys())
         for serial in serials:
             batch = self._batches.get(serial)
             if batch and batch.timer:
                 batch.timer.cancel()
             await self._flush(serial)
 
+        # Await any in-flight flush tasks that were scheduled before this call.
+        # Snapshot first — done-callbacks mutate the set concurrently.
+        if self._pending_tasks:
+            pending = list(self._pending_tasks)
+            await asyncio.gather(*pending, return_exceptions=True)
+            for task in pending:
+                self._pending_tasks.discard(task)
+
     # -- internals --------------------------------------------------------------
 
     def _get_or_create_batch(self, serial_number: str) -> _PendingBatch:

tests/test_coalescer.py

@@ -300,6 +300,62 @@ async def test_stale_reads_dont_erase_prior_overrides(self) -> None:
         assert sent["UserAirconSettings.EnabledZones"] == [False, False, True, True]
 
 
+class TestCommandCoalescerTaskTracking:
+    """Tests for flush task tracking and cleanup."""
+
+    @pytest.mark.asyncio
+    async def test_pending_tasks_set_initialised(self) -> None:
+        """The _pending_tasks set is initialised empty."""
+        send_fn = AsyncMock()
+        sm = StateManager()
+        coalescer = CommandCoalescer(send_fn, sm, debounce_seconds=0.05)
+        assert isinstance(coalescer._pending_tasks, set)
+        assert len(coalescer._pending_tasks) == 0
+
+    @pytest.mark.asyncio
+    async def test_flush_task_tracked_and_discarded(self) -> None:
+        """Flush tasks are added to _pending_tasks and discarded on completion."""
+        send_fn = AsyncMock()
+        sm = _state_manager_with_zones("abc", [True, True])
+        coalescer = CommandCoalescer(send_fn, sm, debounce_seconds=0.05)
+
+        await coalescer.enqueue("abc", _make_zone_command([False, True]))
+
+        # Allow done-callbacks to execute
+        await asyncio.sleep(0)
+
+        # After the enqueue completes the flush has finished and the
+        # done-callback should have discarded the task from the set.
+        assert len(coalescer._pending_tasks) == 0
+
+    @pytest.mark.asyncio
+    async def test_flush_all_awaits_inflight_tasks(self) -> None:
+        """flush_all awaits a genuinely in-flight background flush task."""
+        gate = asyncio.Event()
+
+        async def _blocking_send(serial: str, cmd: dict[str, Any]) -> None:
+            await gate.wait()
+
+        sm = _state_manager_with_zones("abc", [True, True])
+        coalescer = CommandCoalescer(_blocking_send, sm, debounce_seconds=0.02)
+
+        # Start enqueue — it will block on _blocking_send until gate is set
+        enqueue_task = asyncio.create_task(
+            coalescer.enqueue("abc", _make_zone_command([False, True]))
+        )
+        # Let the debounce timer fire and create the background flush task
+        await asyncio.sleep(0.05)
+        assert len(coalescer._pending_tasks) > 0
+
+        # Release the gate so the flush can finish, then flush_all
+        gate.set()
+        await coalescer.flush_all()
+        await enqueue_task
+
+        # After flush_all, no pending tasks should remain
+        assert len(coalescer._pending_tasks) == 0
+
+
 class TestActronAirAPISendCommandCoalescing:
     """Integration tests for command coalescing through ActronAirAPI.send_command."""