test: 覆盖暂停恢复时复位请求的合并与新目标分配

resume(reset=True) 与暂停中未领取的请求共享目标，领取后另分配；变异验证两项断言有效。SDK 窄套件 103 项通过。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CpCfJF6SvonMKmPZhzKZHb

tests/test_pipeline.py

@@ -8,7 +8,7 @@ import time
 import numpy as np
 import pytest
 
-from orisys import FrameSnapshot, ProducedFrame, StreamPipeline
+from orisys import FrameSnapshot, PauseLevel, ProducedFrame, StreamPipeline
 
 
 class FakeSensor:
@@ -651,3 +651,126 @@ def test_legacy_sensor_with_only_reset_is_reset_on_the_flow_stage():
     assert sensor.reset_calls == 1
     assert sensor.reset_threads == ["orisys-pipeline-flow"]
     assert sensor.reset_at_frame == [0], "the fallback reset must precede compute_flow"
+
+
+class PauseResetFakeSensor(FakeSensor):
+    """A split-reset fake whose flow stage parks at chosen frame numbers.
+
+    Parking inside ``compute_flow`` is what makes a pause deterministic: the
+    flow stage has already decided the current frame's version and cannot claim
+    a new request until the gate is released.
+    """
+
+    def __init__(self, *, gates=(1,)) -> None:
+        super().__init__()
+        self._gate_calls = {int(gate) for gate in gates}
+        self.gate_reached = {call: threading.Event() for call in self._gate_calls}
+        self.release_gate = {call: threading.Event() for call in self._gate_calls}
+        self.hhd_calls = 0
+        self.image_reset_threads: list[str] = []
+        self.force_reset_threads: list[str] = []
+        self.force_reset_at_hhd_calls: list[int] = []
+
+    def compute_flow(self, img=None, **kwargs) -> None:
+        FakeSensor.compute_flow(self, img, **kwargs)
+        call = self._flow_calls
+        if call in self._gate_calls:
+            self.gate_reached[call].set()
+            self.release_gate[call].wait(timeout=5.0)
+
+    def compute_hhd(self, flow=None, **kwargs) -> None:
+        self.hhd_calls += 1
+        FakeSensor.compute_hhd(self, flow, **kwargs)
+
+    def reset_image_state(self) -> None:
+        self.image_reset_threads.append(threading.current_thread().name)
+
+    def reset_force_state(self) -> None:
+        self.force_reset_threads.append(threading.current_thread().name)
+        self.force_reset_at_hhd_calls.append(self.hhd_calls)
+
+    def reset(self) -> None:
+        # Mirrors Sensor.reset(): both halves immediately, on the caller's thread.
+        self.reset_calls += 1
+        self.reset_at_frame.append(self._flow_calls)
+        self.reset_image_state()
+        self.reset_force_state()
+
+
+def test_resume_coalesces_with_a_reset_request_queued_while_paused():
+    """A request queued while flow cannot claim it shares the resume's target."""
+    sensor = PauseResetFakeSensor(gates=(1,))
+    frames: list[FrameSnapshot] = []
+    pipe = StreamPipeline(
+        sensor, make_producer(sensor, 8), on_frame=frames.append, depth=2
+    )
+    pipe.start()
+    try:
+        assert sensor.gate_reached[1].wait(timeout=5.0), "flow never reached its gate"
+        pipe.pause(PauseLevel.CAPTURE_ONLY)
+        assert pipe.tracker_reset_seq == 0
+        request = pipe.request_reset()
+        assert request == 1
+        pipe.resume(reset=True)
+        # The flow stage is still parked, so resume's own request coalesces.
+        assert pipe.request_reset() == request, (
+            "resume allocated a new target although the flow stage had not claimed "
+            "the queued request"
+        )
+        sensor.release_gate[1].set()
+        assert _wait_until(
+            lambda: any(snap.tracker_reset_seq >= request for snap in frames)
+        ), "the pipeline never published the resumed reference version"
+        versions = [snap.tracker_reset_seq for snap in frames]
+        assert 0 in versions, versions
+        assert max(versions) == request, versions
+        assert sensor.image_reset_threads == ["orisys-pipeline-flow"]
+        assert sensor.force_reset_threads == ["orisys-pipeline-hhd"]
+    finally:
+        sensor.release_gate[1].set()
+        pipe.stop()
+
+
+def test_resume_after_a_claimed_reset_gets_a_new_target():
+    """Once flow has claimed a request, resume's reset must not merge into it."""
+    sensor = PauseResetFakeSensor(gates=(1, 3))
+    frames: list[FrameSnapshot] = []
+    pipe = StreamPipeline(
+        sensor, make_producer(sensor, 10), on_frame=frames.append, depth=2
+    )
+    pipe.start()
+    try:
+        assert sensor.gate_reached[1].wait(timeout=5.0), "flow never reached gate 1"
+        first = pipe.request_reset()
+        assert first == 1
+        sensor.release_gate[1].set()
+        assert _wait_until(
+            lambda: any(snap.tracker_reset_seq >= first for snap in frames)
+        ), "the first requested version was never published"
+        assert sensor.gate_reached[3].wait(timeout=5.0), "flow never reached gate 3"
+        assert pipe.tracker_reset_seq == first
+
+        pipe.pause(PauseLevel.CAPTURE_ONLY)
+        second = pipe.request_reset()
+        assert second == first + 1
+        pipe.resume(reset=True)
+        assert pipe.request_reset() == second, (
+            "resume merged into a target the flow stage had already claimed"
+        )
+        sensor.release_gate[3].set()
+        assert _wait_until(
+            lambda: any(snap.tracker_reset_seq >= second for snap in frames)
+        ), "the resumed version was never published"
+        versions = [snap.tracker_reset_seq for snap in frames]
+        assert {first, second}.issubset(set(versions)), versions
+        assert max(versions) == second, versions
+        # Each new reference version resets the image and the force state once.
+        assert len(sensor.image_reset_threads) == 2, sensor.image_reset_threads
+        assert len(sensor.force_reset_threads) == 2, sensor.force_reset_threads
+        assert sensor.force_reset_at_hhd_calls == sorted(
+            sensor.force_reset_at_hhd_calls
+        ), sensor.force_reset_at_hhd_calls
+    finally:
+        sensor.release_gate[1].set()
+        sensor.release_gate[3].set()
+        pipe.stop()