Fix single-acquisition test frame accounting (#1231)

* Document single-acquisition test reliability fix

* Plan single-acquisition test reliability fix

* Fix single-acquisition test frame accounting

Author: kevin-m-dean

docs/superpowers/plans/2026-07-21-single-acquisition-test-reliability.md

@@ -0,0 +1,135 @@
+# Single-Acquisition Test Reliability Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Prevent `test_single_acquisition` from mistaking coalesced display notifications for dropped camera frames and guarantee cleanup after failures.
+
+**Architecture:** Keep production acquisition behavior unchanged. Add a test-local pipe reader that returns the newest integer frame identifier observed before the `"stop"` sentinel, use it in the integration test, and move thread/pipe cleanup into `finally`.
+
+**Tech Stack:** Python 3.9, pytest 8, multiprocessing pipes, Navigate's `navigate` Conda environment.
+
+## Global Constraints
+
+- Modify test code only; do not alter camera acquisition or display-pipe behavior.
+- Preserve Python 3.9 compatibility.
+- Run repository-aware tests with `conda run -n navigate python -m pytest`.
+- Format modified Python with Black and lint it with Ruff.
+
+---
+
+### Task 1: Make single-acquisition frame accounting batch-safe
+
+**Files:**
+- Modify: `test/model/test_model.py:116-156`
+- Test: `test/model/test_model.py`
+
+**Interfaces:**
+- Consumes: pipe objects exposing `recv()` and messages containing integer frame identifiers followed by the `"stop"` sentinel.
+- Produces: `_receive_last_frame_id(show_img_pipe: Any, max_messages: int = 20) -> Optional[int]`.
+
+- [x] **Step 1: Write the failing split-batch regression**
+
+Add a fake readable pipe and a focused test before defining the helper:
+
+```python
+class FakeReadablePipe:
+    def __init__(self, messages):
+        self.messages = iter(messages)
+
+    def recv(self):
+        return next(self.messages)
+
+
+def test_receive_last_frame_id_handles_coalesced_batches():
+    show_img_pipe = FakeReadablePipe([0, 2, "stop"])
+
+    assert _receive_last_frame_id(show_img_pipe) == 2
+```
+
+- [x] **Step 2: Run the regression to verify RED**
+
+Run:
+
+```bash
+conda run -n navigate python -m pytest test/model/test_model.py::test_receive_last_frame_id_handles_coalesced_batches -q
+```
+
+Expected: FAIL with `NameError: name '_receive_last_frame_id' is not defined`.
+
+- [x] **Step 3: Implement the minimal pipe reader**
+
+Add `Any` and `Optional` to the imports and define:
+
+```python
+def _receive_last_frame_id(
+    show_img_pipe: Any, max_messages: int = 20
+) -> Optional[int]:
+    last_frame_id = None
+    for _ in range(max_messages):
+        message = show_img_pipe.recv()
+        if message == "stop":
+            break
+        if isinstance(message, int):
+            last_frame_id = message
+    return last_frame_id
+```
+
+- [x] **Step 4: Verify the focused regression is GREEN**
+
+Run:
+
+```bash
+conda run -n navigate python -m pytest test/model/test_model.py::test_receive_last_frame_id_handles_coalesced_batches -q
+```
+
+Expected: `1 passed`.
+
+- [x] **Step 5: Replace notification counting and guarantee cleanup**
+
+Update `test_single_acquisition` to compare the final frame identifier and always clean up:
+
+```python
+show_img_pipe = model.create_pipe("show_img_pipe")
+try:
+    model.run_command("acquire")
+    last_frame_id = _receive_last_frame_id(show_img_pipe)
+    assert last_frame_id == n_frames_expected - 1
+finally:
+    if model.data_thread is not None:
+        model.data_thread.join()
+    model.release_pipe("show_img_pipe")
+```
+
+- [x] **Step 6: Run focused acquisition verification**
+
+Run:
+
+```bash
+conda run -n navigate python -m pytest \
+  test/model/test_model.py::test_receive_last_frame_id_handles_coalesced_batches \
+  test/model/test_model.py::test_single_acquisition \
+  test/model/test_model.py::test_run_data_process_drains_pending_frames_after_signal_completion \
+  -q
+```
+
+Expected: `3 passed`.
+
+- [x] **Step 7: Format, lint, and inspect**
+
+Run:
+
+```bash
+conda run -n navigate black --check test/model/test_model.py
+conda run -n navigate ruff check test/model/test_model.py
+git diff --check
+git status --short --branch
+```
+
+Expected: Black and Ruff exit successfully, `git diff --check` emits no output, and only the planned files are modified.
+
+- [x] **Step 8: Commit the implementation**
+
+```bash
+git add test/model/test_model.py docs/superpowers/plans/2026-07-21-single-acquisition-test-reliability.md
+git commit -m "Fix single-acquisition test frame accounting"
+```

docs/superpowers/specs/2026-07-21-single-acquisition-test-reliability-design.md

@@ -0,0 +1,42 @@
+# Reliable Single-Acquisition Test Design
+
+## Context
+
+The Windows `develop` workflow failed in `test_single_acquisition` after the
+synthetic camera acquired three frames but the display pipe emitted only two
+notifications. `Model.run_data_process()` deliberately sends the newest frame
+identifier from each camera batch rather than every identifier in the batch.
+The test currently counts pipe notifications, so a valid batch sequence such
+as `[0]` followed by `[1, 2]` is miscounted as two frames.
+
+The test is marked for reruns and uses a module-scoped model. Its thread join
+and pipe release occur after the assertion, so a failed attempt skips cleanup
+and contaminates subsequent reruns.
+
+## Design
+
+Keep the production acquisition and display-pipe behavior unchanged. Update
+`test_single_acquisition` to track the most recent integer frame identifier
+received from the pipe and assert that it reaches the expected final frame
+identifier. This reflects the pipe's contract: intermediate display frames may
+be coalesced, but the newest acquired frame is delivered.
+
+Wrap acquisition, pipe reads, and assertions in `try/finally`. In `finally`,
+join the data thread when it exists and release `show_img_pipe`, ensuring that
+an assertion failure cannot leave shared module-scoped state behind.
+
+## Regression Coverage
+
+Add a small helper for interpreting display-pipe messages and test it with the
+split-batch notification sequence `0, 2, "stop"`. The regression must fail
+under notification counting and pass when progress is inferred from the final
+frame identifier.
+
+Run the updated single-acquisition test together with the existing batch-drain
+test in the project `navigate` Conda environment. Then run formatting and lint
+checks on the modified test file.
+
+## Scope
+
+This change modifies test code only. It does not alter camera acquisition,
+threading, display-pipe messages, or user-visible behavior.

test/model/test_model.py

@@ -34,6 +34,7 @@
 import pytest
 import os
 from types import SimpleNamespace
+from typing import Any, Iterable, Iterator, Optional
 from multiprocessing import Manager
 from unittest.mock import MagicMock, patch
 import multiprocessing
@@ -45,6 +46,31 @@
 IN_GITHUB_ACTIONS = os.getenv("GITHUB_ACTIONS") == "true"
 
 
+def _receive_last_frame_id(show_img_pipe: Any, max_messages: int = 20) -> Optional[int]:
+    last_frame_id = None
+    for _ in range(max_messages):
+        message = show_img_pipe.recv()
+        if message == "stop":
+            break
+        if isinstance(message, int):
+            last_frame_id = message
+    return last_frame_id
+
+
+class FakeReadablePipe:
+    def __init__(self, messages: Iterable[Any]) -> None:
+        self.messages: Iterator[Any] = iter(messages)
+
+    def recv(self) -> Any:
+        return next(self.messages)
+
+
+def test_receive_last_frame_id_handles_coalesced_batches() -> None:
+    show_img_pipe = FakeReadablePipe([0, 2, "stop"])
+
+    assert _receive_last_frame_id(show_img_pipe) == 2
+
+
 @pytest.fixture(scope="module")
 def model():
     from types import SimpleNamespace
@@ -124,36 +150,14 @@ def test_single_acquisition(model):
     )
 
     show_img_pipe = model.create_pipe("show_img_pipe")
-
-    model.run_command("acquire")
-
-    # If three channel acquisitions happen quickly, the synthetic camera may return
-    # something like [0, 1, 2] in a single call, and only one pipe message is sent
-    # for that batch.
-    image_id = show_img_pipe.recv()
-    if isinstance(image_id, list):
-        n_framed_received = len(image_id)
-    else:
-        n_framed_received = image_id + 1
-
-    # maximum of 20 iterations to avoid infinite loop in case of failure
-    max_iters = 20
-    while image_id != "stop" and max_iters > 0:
-
-        image_id = show_img_pipe.recv()
-        if image_id == "stop":
-            break
-        if isinstance(image_id, list):
-            n_framed_received += len(image_id)
-        else:
-            n_framed_received += 1
-
-        # decrement iteration counter
-        max_iters -= 1
-
-    assert n_framed_received == n_frames_expected
-    model.data_thread.join()
-    model.release_pipe("show_img_pipe")
+    try:
+        model.run_command("acquire")
+        last_frame_id = _receive_last_frame_id(show_img_pipe)
+        assert last_frame_id == n_frames_expected - 1
+    finally:
+        if model.data_thread is not None:
+            model.data_thread.join()
+        model.release_pipe("show_img_pipe")
 
 
 def test_run_acquisition_marks_finite_signal_completion_for_data_thread():