Return flow sample locations with an active flag and a fixed colour ramp

The selector returned only the arrows it would draw, so a view could not draw
anything at a quiet sample and the below-threshold branch was unreachable. It
now returns every candidate location plus an `active` flag, both callers filter
on it, and the magnitude colour ramp is fixed so one magnitude is one colour
across frames. Verified by pixel parity against HEAD for the 2D canvas and the
3D overlay, and by the shared and studio suites.

shared/src/orisys_app_shared/acquisition/arrows.py

@@ -8,6 +8,7 @@ from functools import lru_cache
 import cv2
 import numpy as np
 
+from orisys_app_shared.scene3d.contracts import diagnostic
 from orisys_app_shared.theme import color
 
 ARROW_BG_BGR = (0x16, 0x13, 0x12)
@@ -22,6 +23,32 @@ FLOW_IDLE_HEX = "#8A93A6"
 #: BGR grey used for depth samples that carry no value.
 NO_DEPTH_BGR = (0x8A, 0x8A, 0x8A)
 
+#: The fixed flow-magnitude colour range, in **uv pixels of displacement**.
+#:
+#: The colour of an arrow is a statement about its magnitude, so it has to be the same
+#: statement from one frame to the next: a per-frame range makes the same magnitude a
+#: different colour when the frame's maximum changes, which is how two frames can look
+#: alike while showing different motion. ``0`` is still (no motion) and the default
+#: upper end is ``2.0`` px of displacement — above the ``flow_grid_spacing * 0.8`` cap
+#: the drawn length is already past on a 10 px finger grid — so the ramp saturates only
+#: where the arrow itself does. A pane that offers the operator a colour range owns this
+#: value and passes it as ``magnitude_range`` to :func:`draw_arrow_display`.
+FLOW_MAGNITUDE_RANGE: tuple[float, float] = (0.0, 2.0)
+
+#: Whether an inactive sample is drawn as a quiet dot. Off: the live 2D canvas draws
+#: arrows and nothing else, so splitting activity out of the selector changed no pixel.
+DRAW_IDLE_GRID_DOTS = False
+
+#: Machine-readable code for "this frame could not be drawn": the caller gets the base
+#: image back and must surface the code instead of presenting it as an idle frame. The
+#: code already exists in the contract's severity table, so no new locale key is needed
+#: — see :data:`ARROW_RENDER_FAILED_MESSAGE` for what a caller shows a person.
+ARROW_RENDER_FAILED = "field_unavailable"
+
+#: Operator-facing text for :data:`ARROW_RENDER_FAILED`. This module is not a locale
+#: source: a UI that shows the failure should map the code to its own translated string.
+ARROW_RENDER_FAILED_MESSAGE = "flow arrows could not be drawn for this frame"
+
 _LUT_STEPS = 256
 _RAMP_MIN_STOPS = 2
 _RAMP_MAX_STOPS = 8
@@ -172,6 +199,63 @@ def _ensure_bgr(img: np.ndarray) -> np.ndarray:
     return img
 
 
+def magnitude_to_ramp_index(
+    magnitude: np.ndarray,
+    *,
+    value_range: tuple[float, float] | None = None,
+    per_frame: bool = False,
+) -> np.ndarray:
+    """Map magnitudes to 0..255 colour-ramp indices under a **stated** range.
+
+    ``value_range`` defaults to :data:`FLOW_MAGNITUDE_RANGE`, so the same magnitude maps
+    to the same colour in every frame. Bounds left as ``None`` fall back to the default's
+    own bound, which is what makes a caller able to pin only the top of the range.
+
+    ``per_frame`` is the old behaviour — re-normalise against this frame's own min and
+    max — and exists only so a caller can ask for it explicitly. It is not the default:
+    it is what made colour incomparable between frames.
+    """
+    values = np.asarray(magnitude, dtype=np.float64).reshape(-1)
+    lo_default, hi_default = FLOW_MAGNITUDE_RANGE
+    lo, hi = (lo_default, hi_default) if value_range is None else value_range
+    lo = lo_default if lo is None else float(lo)
+    hi = hi_default if hi is None else float(hi)
+    if per_frame:
+        lo = float(values.min()) if values.size else 0.0
+        hi = float(values.max()) if values.size else 0.0
+    if not (hi > lo):
+        # A range of zero width says nothing about position within it.
+        return np.zeros(values.shape, dtype=np.uint8)
+    scaled = (values - lo) / (hi - lo)
+    return np.clip(scaled * 255.0, 0.0, 255.0).astype(np.uint8)
+
+
+def _flow_result(image: np.ndarray, diagnostic=None) -> np.ndarray:
+    """``image`` as the result of :func:`draw_arrow_display`, carrying ``diagnostic``.
+
+    An ``np.ndarray`` subclass with one extra attribute, so every existing caller keeps
+    its array — a shape, a dtype, a buffer for Qt, a pixel comparison — and only a caller
+    that wants to know whether the draw succeeded has to look. The attribute is named
+    after the code the rest of the app already routes: ``frame_diagnostic`` is ``None``
+    when the frame was drawn.
+
+    The caller must **state** the failure: the returned pixels are the base image, which
+    is also a legitimate idle frame, so a caller that shows it without checking the
+    attribute is showing a render failure as if the device had measured nothing. The
+    panel's job is ``code -> localized text``; this module deliberately carries no locale
+    string (see :data:`ARROW_RENDER_FAILED_MESSAGE`).
+    """
+    out = image.view(_DrawResult)
+    out.frame_diagnostic = diagnostic
+    return out
+
+
+class _DrawResult(np.ndarray):
+    """An image plus ``frame_diagnostic`` (``None`` when the draw succeeded)."""
+
+    frame_diagnostic = None
+
+
 def _policy_from_style(
     draw_style: str,
     *,
@@ -208,33 +292,71 @@ def _paint_shared_flow_arrows(
     policy,
     colormap: np.ndarray | None,
     below_threshold_color: tuple[int, int, int],
+    magnitude_range: tuple[float, float] | None = None,
+    per_frame_colour: bool = False,
 ) -> np.ndarray:
-    """Draw arrows from :func:`select_flow_samples` so 2D matches 3D sample points."""
+    """Draw arrows from :func:`select_flow_samples` so 2D matches 3D sample points.
+
+    The selector returns *candidate locations* and an ``active`` flag per sample; the
+    drawing choice is this function's own. Arrows are drawn for active samples, so what
+    reaches the canvas is what reached it before the contract split.
+
+    An inactive sample is drawn as a quiet dot in ``below_threshold_color`` only when
+    :data:`DRAW_IDLE_GRID_DOTS` asks for it, and this canvas does not: a live finger frame
+    at 60 Hz would otherwise gain a grid of dots under the arrows. Flipping that constant
+    is what the split made possible — the dots used to depend on a below-threshold branch
+    that the selector's own filtering had already made unreachable.
+
+    ``magnitude_range`` is the **fixed** colour range in uv pixels of displacement
+    (:data:`FLOW_MAGNITUDE_RANGE`); ``per_frame_colour`` opts in to the old
+    re-normalisation of every frame against its own maximum, which makes one magnitude
+    a different colour from one frame to the next.
+    """
     from orisys_app_shared.scene3d.visual_policy import select_flow_samples
 
     image = canvas.copy()
     samples = select_flow_samples(flow, valid_mask, policy)
     if samples.rows.size == 0:
         return image
-    mag = samples.magnitude
-    mag_range = float(mag.max() - mag.min()) if mag.size else 0.0
-    if mag_range > 0:
-        norm = ((mag - mag.min()) / mag_range * 255.0).astype(np.uint8)
-    else:
-        norm = np.zeros(mag.shape, dtype=np.uint8)
+    active = np.asarray(samples.active, dtype=bool)
+    if DRAW_IDLE_GRID_DOTS:
+        for i in np.flatnonzero(~active):
+            cv2.circle(
+                image,
+                (int(samples.cols[i]), int(samples.rows[i])),
+                radius=1,
+                color=below_threshold_color,
+                thickness=-1,
+            )
+    if not np.any(active):
+        return image
     cmap = cv2.COLORMAP_JET if colormap is None else colormap
+    # The range is taken over the samples this canvas actually draws. In per-frame mode
+    # that is what the old selector handed the painter (it had already dropped the
+    # inactive candidates, so the frame's span was theirs); a fixed range does not care
+    # either way, because its bounds come from the caller.
+    magnitudes = np.asarray(samples.magnitude[active], dtype=np.float64)
+    norm = magnitude_to_ramp_index(
+        magnitudes,
+        value_range=magnitude_range,
+        per_frame=per_frame_colour,
+    )
     colors = cv2.applyColorMap(norm.reshape(-1, 1), cmap)
-    for i in range(samples.rows.size):
-        pt1 = (int(samples.cols[i]), int(samples.rows[i]))
-        pt2 = (
-            int(round(samples.cols[i] + samples.dx[i])),
-            int(round(samples.rows[i] + samples.dy[i])),
+    rows = samples.rows[active]
+    cols = samples.cols[active]
+    dx = samples.dx[active]
+    dy = samples.dy[active]
+    for i in range(rows.size):
+        pt1 = (int(cols[i]), int(rows[i]))
+        pt2 = (int(round(cols[i] + dx[i])), int(round(rows[i] + dy[i])))
+        cv2.arrowedLine(
+            image,
+            pt1,
+            pt2,
+            tuple(int(c) for c in colors[i, 0]),
+            thickness=1,
+            tipLength=0.2,
         )
-        color = tuple(int(c) for c in colors[i, 0])
-        if mag[i] < policy.flow_on_px:
-            cv2.circle(image, pt1, radius=1, color=below_threshold_color, thickness=-1)
-        else:
-            cv2.arrowedLine(image, pt1, pt2, color, thickness=1, tipLength=0.2)
     return image
 
 
@@ -255,6 +377,8 @@ def draw_arrow_display(
     flow_grid_spacing: int | None = None,
     max_arrows: int | None = None,
     policy=None,
+    magnitude_range: tuple[float, float] | None = None,
+    per_frame_colour: bool = False,
 ) -> np.ndarray:
     """Draw flow arrows on a solid canvas or over the processed frame.
 
@@ -273,10 +397,24 @@ def draw_arrow_display(
     Passing it is what makes the two figures trigger at the same strength; the
     finger-style defaults below are only the fallback for a caller that has no pane
     (a tool, a test). Box style keeps the SDK path either way.
+
+    ``magnitude_range`` fixes the colour scale, in the same unit as ``flow_on_px``
+    (uv pixels of displacement); the default is :data:`FLOW_MAGNITUDE_RANGE` and a
+    settings model that owns the range passes its own here. ``per_frame_colour`` opts
+    back in to normalising every frame against its own maximum, which makes one
+    magnitude a different colour from one frame to the next.
+
+    Returns the drawn image, which carries ``frame_diagnostic``: ``None`` when the frame
+    was drawn, and otherwise a :class:`~orisys_app_shared.scene3d.contracts.Diagnostic`
+    whose ``code`` is :data:`ARROW_RENDER_FAILED`. **The failure must be surfaced**: the
+    pixels are the base image, exactly as they are for a frame with no flow, so a view
+    that shows it without reading the attribute reports "nothing measured" for "nothing
+    could be drawn". A ``flow`` of ``None`` is a legitimate empty frame and leaves the
+    attribute ``None``.
     """
     base = _ensure_bgr(processed_frame)
     if flow is None:
-        return base
+        return _flow_result(base)
     finger = str(draw_style).lower().startswith("f")
     draw_kw = dict(_FINGER_ARROW_DRAW_KW if finger else _ARROW_DRAW_KW)
     if colormap is not None:
@@ -324,20 +462,36 @@ def draw_arrow_display(
                     max_arrows=max_arrows,
                 )
             )
-            return _paint_shared_flow_arrows(
-                canvas,
-                np.asarray(flow),
-                valid_mask=valid_mask,
-                policy=selected,
-                colormap=colormap,
-                below_threshold_color=tuple(draw_kw["below_threshold_color"]),
+            return _flow_result(
+                _paint_shared_flow_arrows(
+                    canvas,
+                    np.asarray(flow),
+                    valid_mask=valid_mask,
+                    policy=selected,
+                    colormap=colormap,
+                    below_threshold_color=tuple(draw_kw["below_threshold_color"]),
+                    magnitude_range=magnitude_range,
+                    per_frame_colour=per_frame_colour,
+                )
             )
 
         import orisys
 
-        return orisys.util.draw_arrows(canvas, flow, valid_mask=valid_mask, **draw_kw)
-    except Exception:
-        return base
+        return _flow_result(
+            orisys.util.draw_arrows(canvas, flow, valid_mask=valid_mask, **draw_kw)
+        )
+    except Exception as exc:
+        # The frame is *not* idle: it failed to draw. The image alone cannot say so, so
+        # the code travels on it. ``type(exc).__name__`` only — an exception message can
+        # carry paths or field data, and diagnostics travel to the UI and into logs.
+        return _flow_result(
+            base,
+            diagnostic(
+                ARROW_RENDER_FAILED,
+                f"{ARROW_RENDER_FAILED_MESSAGE}: {type(exc).__name__}",
+                detail=type(exc).__name__,
+            ),
+        )
 
 
 def depth_map_to_bgr(

shared/src/orisys_app_shared/scene3d/composer.py

@@ -272,8 +272,11 @@ class SceneComposer:
                     frame.flow_valid_mask,
                     policy,
                 )
+                # The selector returns every candidate location plus an ``active`` flag;
+                # this figure draws the active ones, which is what it drew before the
+                # split (the selector used to filter and this figure drew the lot).
                 flow_arrows, extra = build_flow_overlay(
-                    samples,
+                    samples.active_only(),
                     mapping,
                     style=policy,
                     normal_offset_m=mesh_span_from_mapping(mapping) * 0.002,

shared/src/orisys_app_shared/scene3d/visual_policy.py

@@ -20,6 +20,17 @@ from orisys_app_shared.scene3d.contracts import (
     freeze,
 )
 
+#: Canvas pixels of displacement per unit of the SDK flow field.
+#:
+#: The flow field's magnitude is documented in its own units; the arrow length and
+#: :attr:`FlowSamples.magnitude` are in **canvas pixels of displacement**, so this is
+#: the conversion between the two. It is one factor in one place, which is what keeps
+#: the threshold (:attr:`VisualPolicy.flow_on_px`) and the drawn length in the same
+#: unit: change it here and the sample grid, the threshold and the arrows all move
+#: together. The convention was already documented and pinned by
+#: ``shared/tests/test_image_backend.py``; the arithmetic is unchanged.
+DISPLACEMENT_PX_PER_FLOW = 2.0
+
 
 @dataclass(frozen=True)
 class VisualPolicy:
@@ -87,13 +98,52 @@ def policy_for_settings(settings: DisplaySettings) -> VisualPolicy:
 
 @dataclass(frozen=True)
 class FlowSamples:
-    """Shared UV sample points for 2D and 3D flow arrows."""
+    """Shared UV sample points for 2D and 3D flow arrows.
+
+    **Location and activity are separate.** Every returned row is a candidate sample
+    location: mask-clipped, inside the edge margin, at the policy's grid and density —
+    *not* a verdict about how strong the motion there is. ``active`` carries that
+    verdict, one boolean per sample, computed from the thresholds (and, when the caller
+    passes ``previously_on``, from hysteresis). A caller that wants the old
+    "only the arrows I would draw" list calls :meth:`active_only`; a caller that
+    wants to draw something else at a quiet location — a grid dot, a lighter glyph —
+    now has the location to draw it at, which active-only samples made impossible.
+
+    The two views used to disagree about the band between ``flow_off_px`` and
+    ``flow_on_px`` precisely because the verdict was fused into the location list.
+    """
 
     rows: np.ndarray  # [K] int
     cols: np.ndarray  # [K] int
     dx: np.ndarray  # [K] float — tip offset in UV pixels
     dy: np.ndarray  # [K] float
-    magnitude: np.ndarray  # [K] float — |2*flow|
+    magnitude: np.ndarray  # [K] float — |DISPLACEMENT_PX_PER_FLOW * flow|, UV pixels
+    active: np.ndarray  # [K] bool — magnitude cleared this sample's threshold
+
+    @property
+    def active_count(self) -> int:
+        """How many candidates are active; ``0`` for an empty selection."""
+        return int(np.count_nonzero(self.active))
+
+    def active_only(self) -> FlowSamples:
+        """This selection with the inactive candidates dropped.
+
+        The old contract, as a caller's explicit choice rather than the selector's
+        hidden one: a figure that draws (or maps) an arrow per sample calls this, and
+        keeps drawing exactly what it drew before locations and activity were split.
+        """
+        keep = np.asarray(self.active, dtype=bool)
+        if bool(np.all(keep)):
+            return self
+        return FlowSamples(
+            rows=self.rows[keep],
+            cols=self.cols[keep],
+            dx=self.dx[keep],
+            dy=self.dy[keep],
+            magnitude=self.magnitude[keep],
+            active=keep[keep],
+        )
+
 
 
 def build_domain_mask(
@@ -152,11 +202,22 @@ def select_flow_samples(
     *,
     previously_on: np.ndarray | None = None,
 ) -> FlowSamples:
-    """Sample optical flow on a UV grid with hysteresis and mask clipping.
+    """Every candidate UV sample location, each flagged active or not.
 
-    ``previously_on`` is a boolean grid of the same shape as ``flow`` (or flat
-    over the candidate points) used for on/off hysteresis; pass ``None`` to use
-    ``flow_on_px`` only (stateless).
+    The returned points are the policy's own grid, clipped by the edge margin and by
+    ``mask`` (at the sample and at its tip), capped at ``max_arrows`` by magnitude —
+    and they carry a magnitude, which is what decides their order under that cap.
+    "Active" is a **separate** array: ``magnitude >= threshold``, with the threshold
+    chosen per sample by ``previously_on`` when it is given.
+
+    ``previously_on`` is a boolean grid of the same shape as ``flow`` (or flat over
+    the candidate points) used for on/off hysteresis: a sample that was on last time
+    keeps a lower floor (``flow_off_px``), everything else must clear
+    ``flow_on_px``. Pass ``None`` for a stateless selection. This is the one place
+    hysteresis is applied, so a stateful selection can only ever be shared by both
+    views or by neither — which is why it is off by default and why the old
+    active-only contract could not be shared: it baked one view's state into a
+    function the other view also called.
     """
     policy = policy.normalised()
     flow = np.asarray(flow, dtype=np.float64)
@@ -173,7 +234,7 @@ def select_flow_samples(
     rr = rr.reshape(-1)
     cc = cc.reshape(-1)
 
-    disp = 2.0 * flow[rr, cc, :]
+    disp = DISPLACEMENT_PX_PER_FLOW * flow[rr, cc, :]
     mag = np.linalg.norm(disp, axis=1)
     arrow_scale = policy.flow_arrow_scale
     dx = disp[:, 0] / arrow_scale
@@ -202,7 +263,6 @@ def select_flow_samples(
         & (cc < w - edge)
         & (rr >= edge)
         & (rr < h - edge)
-        & (mag >= threshold)
         & np.isfinite(mag)
     )
     if mask is not None:
@@ -220,20 +280,32 @@ def select_flow_samples(
     if not np.any(valid):
         empty = np.empty(0, dtype=np.int32)
         empty_f = np.empty(0, dtype=np.float64)
-        return FlowSamples(rows=empty, cols=empty, dx=empty_f, dy=empty_f, magnitude=empty_f)
+        return FlowSamples(
+            rows=empty,
+            cols=empty,
+            dx=empty_f,
+            dy=empty_f,
+            magnitude=empty_f,
+            active=np.empty(0, dtype=bool),
+        )
 
     idx = np.flatnonzero(valid)
     if idx.size > policy.max_arrows:
-        # Keep the strongest magnitudes when the grid is dense.
+        # Keep the strongest magnitudes when the grid is dense. The cap is a location
+        # budget, so it is spent before the activity verdict: a caller that draws the
+        # inactive candidates too draws the strongest ``max_arrows`` of them, not a
+        # different set depending on where the threshold sits.
         order = np.argsort(mag[idx])[::-1][: policy.max_arrows]
         idx = idx[order]
 
+    selected_mag = mag[idx].astype(np.float64, copy=False)
     return FlowSamples(
         rows=rr[idx].astype(np.int32, copy=False),
         cols=cc[idx].astype(np.int32, copy=False),
         dx=dx[idx].astype(np.float64, copy=False),
         dy=dy[idx].astype(np.float64, copy=False),
-        magnitude=mag[idx].astype(np.float64, copy=False),
+        magnitude=selected_mag,
+        active=np.asarray(selected_mag >= threshold, dtype=bool),
     )
 
 

shared/tests/test_image_backend.py

@@ -148,6 +148,115 @@ def test_the_2d_canvas_draws_from_the_policy_it_is_given():
     assert len(np.unique(direct.reshape(-1, 3), axis=0)) == 1
 
 
+def test_the_2d_canvas_colours_one_magnitude_the_same_in_every_frame():
+    """The colour range is fixed, so two frames are comparable by colour.
+
+    The canvas used to normalise each frame against its own maximum: the same magnitude
+    was one colour in a frame with a fast sample and another in a frame without one, so
+    two frames could look alike while showing different motion. ``magnitude_range``
+    defaults to :data:`arrows.FLOW_MAGNITUDE_RANGE`, and the old re-normalisation is
+    available only when a caller asks for it.
+    """
+    from orisys_app_shared.acquisition.arrows import magnitude_to_ramp_index
+
+    # One probe magnitude in two frames whose maxima differ by a factor of four.
+    weak = magnitude_to_ramp_index(np.array([0.5, 1.0]))
+    strong = magnitude_to_ramp_index(np.array([0.5, 4.0]))
+    assert weak[0] == strong[0], (
+        "the same 0.5 px magnitude took two different ramp positions because the frame's "
+        f"maximum changed: {weak[0]} against {strong[0]}"
+    )
+    # The range is the documented default, not one frame's own span.
+    assert magnitude_to_ramp_index(np.array([0.0, 2.0])).tolist() == [0, 255]
+    # The per-frame mode is still available, and is what made the colours drift.
+    assert magnitude_to_ramp_index(np.array([0.5, 4.0]), per_frame=True)[0] != weak[0]
+    # A caller may pin the range it wants instead: 1.0 of a 0..4.0 range is a quarter up.
+    assert magnitude_to_ramp_index(np.array([1.0]), value_range=(0.0, 4.0))[0] == 63
+
+
+def test_the_2d_canvas_draws_the_same_magnitude_in_the_same_colour():
+    """The painted canvas obeys the fixed range through :func:`draw_arrow_display`.
+
+    The helper's contract is one thing; this is the pixel an operator looks at. Two
+    frames carry the same magnitude at the same sample and different maxima, and the
+    sample's pixel is the same colour in both.
+    """
+    from orisys_app_shared.acquisition.arrows import ARROW_BG_BGR, flow_colormap
+    from orisys_app_shared.scene3d.visual_policy import VisualPolicy
+
+    h, w = 12, 12
+    policy = VisualPolicy(flow_on_px=0.4, flow_grid_spacing=1, flow_edge_margin=0)
+
+    def canvas(probe: float, strongest: float) -> np.ndarray:
+        flow = np.zeros((h, w, 2), dtype=np.float64)
+        flow[8, 8] = (probe, 0.0)
+        flow[2, 8] = (strongest, 0.0)
+        return draw_arrow_display(
+            np.zeros((h, w, 3), dtype=np.uint8),
+            flow,
+            draw_style="finger",
+            policy=policy,
+            colormap=flow_colormap("dark"),
+        )
+
+    def sample_colour(drawn: np.ndarray, row: int, col: int) -> tuple[int, ...]:
+        """The colour this sample was drawn in.
+
+        The canvas is one flat colour and this sample is drawn at the centre of the
+        patch, so any pixel that is not the canvas is the drawn sample.
+        """
+        margin = 2
+        patch = drawn[row - margin : row + margin + 1, col - margin : col + margin + 1]
+        background = np.array(ARROW_BG_BGR, dtype=np.uint8)
+        drawn_pixels = patch[np.any(patch != background, axis=-1)]
+        assert drawn_pixels.size, f"nothing was drawn at {(row, col)}"
+        return tuple(int(v) for v in drawn_pixels[0])
+
+    quiet = canvas(0.25, 0.5)  # probe |2*flow| = 0.5 px, frame maximum 1.0 px
+    loud = canvas(0.25, 4.0)  # same probe, frame maximum 8.0 px
+    assert sample_colour(quiet, 8, 8) == sample_colour(loud, 8, 8), (
+        "one magnitude took two colours because another sample moved faster"
+    )
+
+
+def test_a_failed_draw_is_not_an_idle_frame():
+    """The base image comes back with a code on it, because the pixels cannot say so.
+
+    A render failure and a frame with nothing to draw produced the same picture, so
+    "the arrows are missing" was indistinguishable from "the device measured nothing".
+    The code travels on the returned image; the caller surfaces it (this backend's own
+    panel wiring is not part of this change).
+    """
+    from orisys_app_shared.acquisition.arrows import ARROW_RENDER_FAILED
+
+    base = np.zeros((12, 12, 3), dtype=np.uint8)
+    base[:, :] = (0x20, 0x30, 0x40)
+    # ``select_flow_samples`` rejects a field that is not [H,W,2], which is a draw
+    # failure rather than an empty measurement.
+    failed = draw_arrow_display(base, np.zeros((12, 12), dtype=np.float32), draw_style="finger")
+    assert np.array_equal(failed, base), "a failed draw must still hand back the image"
+    assert failed.frame_diagnostic is not None
+    assert failed.frame_diagnostic.code == ARROW_RENDER_FAILED
+    assert failed.frame_diagnostic.severity == "warning"
+    # The exception's own message is not carried: diagnostics travel to the UI and logs.
+    assert "Traceback" not in failed.frame_diagnostic.message
+
+    # A frame with no flow at all is a legitimate empty frame, not a failure.
+    idle = draw_arrow_display(base, None, draw_style="finger")
+    assert idle.frame_diagnostic is None
+    assert np.array_equal(idle, base)
+
+    # A drawn frame says nothing, and stays an ordinary array for every other caller.
+    drawn = draw_arrow_display(
+        base,
+        np.full((12, 12, 2), 0.5, dtype=np.float32),
+        draw_style="finger",
+    )
+    assert drawn.frame_diagnostic is None
+    assert isinstance(drawn, np.ndarray)
+    assert drawn.shape == base.shape and drawn.dtype == np.uint8
+
+
 def test_the_backend_renders_what_the_primitives_rendered_before():
     """Live went through these two calls; the backend must not change a pixel."""
     image = _image()

shared/tests/test_visual_policy.py

@@ -44,6 +44,13 @@ def test_select_flow_samples_uses_shared_grid_and_mask() -> None:
 
 
 def test_select_flow_samples_hysteresis_keeps_weaker_motion() -> None:
+    """The band between ``flow_off_px`` and ``flow_on_px`` decides through ``active``.
+
+    The location is a candidate either way — the band is a statement about activity, not
+    about where the policy samples — so what changes with ``previously_on`` is the flag.
+    This is the single place the two views' thresholds are applied: a caller that draws
+    ``active_only()`` draws what it drew when the selector filtered for it.
+    """
     h, w = 30, 30
     flow = np.zeros((h, w, 2), dtype=np.float64)
     # |2*flow| = 0.7: below on (0.8) but above off (0.5)
@@ -58,8 +65,36 @@ def test_select_flow_samples_hysteresis_keeps_weaker_motion() -> None:
     previously[15, 15] = True
     off = select_flow_samples(flow, None, policy, previously_on=None)
     on = select_flow_samples(flow, None, policy, previously_on=previously)
-    assert off.rows.size == 0
-    assert on.rows.size == 1
+    # Same candidate locations: the grid does not depend on the verdict.
+    assert off.rows.tolist() == on.rows.tolist()
+    assert off.cols.tolist() == on.cols.tolist()
+    assert (15, 15) in set(zip(on.rows.tolist(), on.cols.tolist()))
+    # Only the verdict does.
+    assert on.active_count == 1
+    assert off.active_count == 0
+    assert on.active_only().rows.tolist() == [15]
+    assert on.active_only().cols.tolist() == [15]
+    assert off.active_only().rows.size == 0
+
+
+def test_the_location_cap_is_a_location_budget() -> None:
+    """``max_arrows`` limits candidate locations, not arrows.
+
+    The cap has to be spent before the activity verdict, otherwise the set of locations
+    would depend on the threshold and a caller drawing inactive candidates too would draw
+    a different set from the one a caller drawing arrows does. The strongest candidates
+    are kept, and every inactive candidate is one no arrow was ever drawn for.
+    """
+    h, w = 40, 40
+    flow = np.zeros((h, w, 2), dtype=np.float64)
+    flow[15, 15] = (5.0, 0.0)  # far above the threshold
+    policy = VisualPolicy(flow_on_px=0.4, flow_grid_spacing=1, flow_edge_margin=0, max_arrows=4)
+    samples = select_flow_samples(flow, np.ones((h, w), dtype=bool), policy)
+    assert samples.rows.size == 4
+    assert samples.active_count == 1
+    assert (15, 15) in set(zip(samples.rows.tolist(), samples.cols.tolist())), (
+        "the cap dropped the only moving sample"
+    )
 
 
 def test_paint_domain_background_colours() -> None: