fix: bound and gate the crop evidence lane

The crop lane asked the phone for a full-resolution crop of a facing the
frame could not decide. Four defects made that unsafe or useless:

* A capability was half-applied. `crop: true` without a usable
  `max_crop_bytes` enabled the lane on the client's say-so, leaving the two
  sides believing different limits. A capability is now believed whole or
  refused with BAD_MESSAGE, and the limit must be an integer in 1..1048576
  (bool is an int in Python and is not a limit).
* A small JPEG could claim enormous dimensions. The header is now read
  without decoding and anything over 4 megapixels is refused before the
  decode allocates it.
* A request was single-use, so an expired one stranded a facing only the
  crop lane could decide. The lane now spends the same max_attempts budget
  the frame path spends, refilled on method change and on settled-label
  drift, and a phone that never answers still cannot loop.
* A candidate could be asked about from a box the frame path would refuse:
  clipped by the frame edge, carried over from an earlier frame, or measured
  at a score the tracker does not trust. Both lanes now share one edge gate,
  and only a track whose box this frame actually measured is asked about.

Drift and method change also cleared `identity` but left `confirmed = True`,
which permanently blocked both the crop lane and label inheritance for that
track; both now clear it, without which the budget reset was dead code.

Blur and glare are refused as whole-capture faults, while `too_far` is
exempt on purpose: a distant shelf is the one frame a native-resolution crop
exists to decide.

recognization/code/app.py

@@ -94,6 +94,37 @@ MAX_FRAME_BYTES = 4 * 1024 * 1024
 DEFAULT_MAX_SESSIONS = 4
 WS_IDLE_TIMEOUT_S = 120.0
 CATALOG_VERSION_FALLBACK = "unknown"
+MAX_CROP_PIXELS = 4_000_000
+
+
+def _jpeg_dimensions(data: bytes) -> Optional[Tuple[int, int]]:
+    """Read JPEG SOF dimensions without allocating a decoded image."""
+    if not data.startswith(b"\xff\xd8"):
+        return None
+    offset = 2
+    while offset + 4 <= len(data):
+        if data[offset] != 0xff:
+            return None
+        while offset < len(data) and data[offset] == 0xff:
+            offset += 1
+        if offset >= len(data):
+            return None
+        marker = data[offset]
+        offset += 1
+        if marker in (0xd8, 0x01) or 0xd0 <= marker <= 0xd7:
+            continue
+        if marker in (0xd9, 0xda) or offset + 2 > len(data):
+            return None
+        length = int.from_bytes(data[offset:offset + 2], "big")
+        if length < 2 or offset + length > len(data):
+            return None
+        if marker in (0xc0, 0xc1, 0xc2, 0xc3, 0xc5, 0xc6, 0xc7, 0xc9, 0xca, 0xcb, 0xcd, 0xce, 0xcf):
+            if length < 7:
+                return None
+            return (int.from_bytes(data[offset + 5:offset + 7], "big"),
+                    int.from_bytes(data[offset + 3:offset + 5], "big"))
+        offset += length
+    return None
 
 
 # ---------------------------------------------------------------------------
@@ -821,13 +852,20 @@ def create_app(
             """A client that can answer with a full-resolution crop says so here.
 
             Until this arrives the session keeps the protocol-v1 behaviour: no
-            result carries a crop request, so a v1 client is never asked.
+            result carries a crop request, so a v1 client is never asked. A
+            capability is believed whole or not at all: a message that does not
+            name the lane with a limit the server can honour is refused and the
+            lane stays off, rather than half-applied on the client's say-so.
             """
-            try:
+            if message.get("crop") is not True or "max_crop_bytes" not in message:
-                capabilities = session.enable_crops(message.get("max_crop_bytes"))
-            except (TypeError, ValueError):
                 await send({"type": "error", "code": "BAD_MESSAGE",
-                            "message": "bad max_crop_bytes"})
+                            "message": "crop must be true and state max_crop_bytes"})
+                return
+            try:
+                capabilities = session.enable_crops(message["max_crop_bytes"])
+            except ValueError:
+                await send({"type": "error", "code": "BAD_MESSAGE",
+                            "message": "max_crop_bytes must be 1..%d" % MAX_CROP_BYTES})
                 return
             await send({"type": "capabilities_ack", "capabilities": capabilities})
 
@@ -855,6 +893,11 @@ def create_app(
                 await send({"type": "crop_error", "request_id": request_id,
                             "code": "CROP_BAD_MESSAGE", "message": "jpeg is not valid base64"})
                 return
+            dimensions = _jpeg_dimensions(crop_bytes)
+            if dimensions is not None and dimensions[0] * dimensions[1] > MAX_CROP_PIXELS:
+                await send({"type": "crop_error", "request_id": request_id,
+                            "code": "CROP_TOO_LARGE", "message": "crop dimensions exceed the limit"})
+                return
             reply = await loop.run_in_executor(
                 None, session.submit_crop, request_id, message.get("track_id"),
                 message.get("frame_id"), crop_bytes,

recognization/code/live.py

@@ -25,7 +25,7 @@ from concurrent.futures import Future, ThreadPoolExecutor
 from dataclasses import dataclass, field
 LOGGER = logging.getLogger("dorayaki.live")
 
-from typing import Any, Dict, List, Optional, Sequence, Tuple
+from typing import Any, Dict, List, Optional, Sequence, Set, Tuple
 
 current_dir = os.path.dirname(os.path.abspath(__file__))
 if current_dir not in sys.path:
@@ -66,6 +66,11 @@ SOURCE_INHERITED = "inherited"
 # A crop the client returns is base64 inside a text control message, so it is
 # bounded on its own rather than riding the binary frame path.
 MAX_CROP_BYTES = 1024 * 1024
+# Frame faults a full-resolution crop cannot undo. Blur and glare are optical
+# faults of the whole capture, so the phone would photograph the same bad
+# scene. "too_far" is exempt on purpose: a distant shelf is the one frame a
+# native-resolution crop exists to decide.
+CROP_BLIND_QUALITY = ("blur", "glare")
 
 
 @dataclass
@@ -135,9 +140,11 @@ class TrackState:
     # or pruned then re-created). Restarting resets the label, so it is always a
     # re-verification rather than an unexplained label change.
     restarts: int = 0
-    # A crop is asked for at most once per object. A recycled id gets a fresh
+    # How many full-resolution crops have been asked for on this object. It is
-    # TrackState, so the new object may be asked again.
+    # the budget the frame path spends on attempts, so a phone that never
-    crop_requested: bool = False
+    # answers cannot hold a small facing in the lane for ever. A recycled id
+    # gets a fresh TrackState, so a new object starts with a fresh budget.
+    crop_attempts: int = 0
     # Which evidence produced the newest observation: frame or crop.
     last_source: str = SOURCE_FRAME
 
@@ -162,6 +169,21 @@ def _iou(a: Sequence[int], b: Sequence[int]) -> float:
     return inter / union if union > 0 else 0.0
 
 
+def _clear_of_edge(box: Sequence[int], frame_shape: Sequence[int], margin_ratio: float) -> bool:
+    """True when a box sits clear of the frame edge.
+
+    A facing the frame clips is only part of a product, so neither lane may
+    spend a decision on it. Both lanes ask this question, so a candidate the
+    frame path would refuse cannot be asked about through the crop lane
+    instead.
+    """
+    frame_h, frame_w = frame_shape[:2]
+    margin_x = int(frame_w * margin_ratio)
+    margin_y = int(frame_h * margin_ratio)
+    x1, y1, x2, y2 = box
+    return x1 > margin_x and y1 > margin_y and x2 < frame_w - margin_x and y2 < frame_h - margin_y
+
+
 def _crop_off_product(requested: Sequence[int], current: Sequence[int]) -> bool:
     """True when a track no longer covers the box a crop was taken from.
 
@@ -175,6 +197,21 @@ def _crop_off_product(requested: Sequence[int], current: Sequence[int]) -> bool:
     return _iou(current, requested) < DRIFT_IOU
 
 
+def _validate_crop_limit(value: Any) -> int:
+    """The crop limit a client stated, or ValueError.
+
+    The limit is a memory bound, so a value the server cannot honour is refused
+    rather than clamped: quietly handing a client that asked for 512 B a 1 KiB
+    floor, or one that asked for 4 GiB a 1 MiB ceiling, leaves the two sides
+    believing different limits. bool is an int in Python and is not a limit.
+    """
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError("max_crop_bytes must be an integer")
+    if value <= 0 or value > MAX_CROP_BYTES:
+        raise ValueError("max_crop_bytes must be 1..%d" % MAX_CROP_BYTES)
+    return value
+
+
 class _CropRejected(Exception):
     """A returned crop that may not be used, carrying the reply to send back."""
 
@@ -376,6 +413,10 @@ class LiveSession:
                 self._generation += 1
                 for track in self._tracks.values():
                     track.state = STATE_DETECTED
+                    # Nothing has decided this track under the new method yet,
+                    # so it is unconfirmed again rather than a settled track
+                    # that happens to carry no label.
+                    track.confirmed = False
                     track.identity = None
                     track.candidates = []
                     track.score_sums = {}
@@ -383,6 +424,10 @@ class LiveSession:
                     track.attempts = 0
                     track.in_flight = 0
                     track.last_attempt_seq = 0
+                    # A crop asked about under the old method is evidence measured
+                    # under settings the session no longer holds, so the new method
+                    # gets a budget of its own.
+                    track.crop_attempts = 0
                     track.last_attempt_area = 0
                 # Outstanding crops were minted against the old method and the
                 # old generation; answering one now would be evidence measured
@@ -452,12 +497,18 @@ class LiveSession:
             self._process_times.append(tracked_at)
 
             by_id = {box["id"]: box for box in boxes}
+            # Ids whose box this frame actually measured. A track the tracker
+            # is carrying without a detection keeps its last measured box, and
+            # that box must not be handed to the phone as if it described the
+            # frame the request points at.
+            measured: Set[int] = set()
             active: Dict[int, TrackState] = {}
             for track in tracks:
                 state = self._tracks.get(track.track_id)
                 raw_box = None
                 confidence = track.score
-                if track.det_index is not None and track.det_index in by_id:
+                measured_now = track.det_index is not None and track.det_index in by_id
+                if measured_now:
                     detection = by_id[track.det_index]
                     raw_box = list(detection["box"])
                     confidence = float(detection["confidence"])
@@ -513,6 +564,10 @@ class LiveSession:
                 if raw_box and not low_score:
                     state.raw_box = raw_box
                     state.confidence = confidence
+                    if measured_now:
+                        # A detection in this frame's list, at a score the tracker
+                        # trusts: the box describes the frame the request names.
+                        measured.add(track.track_id)
                 elif not state.raw_box and raw_box:
                     state.raw_box = raw_box
                     state.confidence = confidence
@@ -556,7 +611,7 @@ class LiveSession:
             # releasing it. add_done_callback can run the callback inline when the
             # future already finished, and the callback takes this same lock.
             submissions = self._plan_classification(image, seq)
-            crop_requests = self._plan_crop_requests(seq, int(frame_id))
+            crop_requests = self._plan_crop_requests(image.shape, seq, int(frame_id), measured)
 
             summary = self._summary_locked()
             stats = self._stats_locked()
@@ -669,9 +724,19 @@ class LiveSession:
         if track.drift_frames < 2:
             return
         track.drift_frames = 0
+        # A crop asked about the product this box has now left says nothing
+        # about the neighbour in its place, so the answer goes with the label
+        # and the new product is asked about on a budget of its own.
+        for request_id in [rid for rid, rec in self._crop_requests.items()
+                           if rec["track_id"] == track.track_id]:
+            self._crops_rejected += 1
+            self._release_crop_slot(self._crop_requests.pop(request_id))
         self._relabel_count[track.track_id] = self._relabel_count.get(track.track_id, 0) + 1
         track.relabels = self._relabel_count[track.track_id]
         track.state = STATE_DETECTED
+        # The label went with the product, so the track is undecided again:
+        # a settled track that carries no label can never be asked about.
+        track.confirmed = False
         track.identity = None
         track.candidates = []
         track.score_sums = {}
@@ -681,6 +746,7 @@ class LiveSession:
         track.last_attempt_seq = 0
         track.last_attempt_area = 0
         track.label_box = []
+        track.crop_attempts = 0
 
     # ------------------------------------------------------------------ classification
     def _plan_classification(self, image: Any, seq: int) -> List[Tuple[TrackState, Any, int]]:
@@ -690,7 +756,6 @@ class LiveSession:
         after the lock is released.
         """
         frame_h, frame_w = image.shape[:2]
-        margin_x = int(frame_w * self.settings.edge_margin_ratio)
         margin_y = int(frame_h * self.settings.edge_margin_ratio)
         inflight = sum(track.in_flight for track in self._tracks.values())
         capacity = max(0, self.settings.max_inflight_per_session - inflight)
@@ -711,7 +776,7 @@ class LiveSession:
             x1, y1, x2, y2 = track.raw_box
             if min(x2 - x1, y2 - y1) < self.settings.min_crop_px:
                 continue
-            if x1 <= margin_x or y1 <= margin_y or x2 >= frame_w - margin_x or y2 >= frame_h - margin_y:
+            if not _clear_of_edge(track.raw_box, image.shape, self.settings.edge_margin_ratio):
                 continue
             if track.attempts >= self.settings.max_attempts:
                 continue
@@ -775,24 +840,27 @@ class LiveSession:
 
     def enable_crops(self, max_crop_bytes: Optional[int] = None) -> Dict[str, Any]:
         """Turn the lane on for a client that advertised it. The client states the
-        largest crop it will produce; the server keeps the lower of the two."""
+        largest crop it will produce; a limit that is not one is refused and the
+        lane stays where it was, so a bad statement enables nothing."""
+        stated = None if max_crop_bytes is None else _validate_crop_limit(max_crop_bytes)
         with self._lock:
             self.settings.crops = True
-            if max_crop_bytes is not None:
+            if stated is not None:
-                try:
+                self.settings.max_crop_bytes = stated
-                    stated = int(max_crop_bytes)
-                except (TypeError, ValueError):
-                    stated = 0
-                self.settings.max_crop_bytes = max(1024, min(stated, MAX_CROP_BYTES))
             return self.capabilities()
 
-    def _plan_crop_requests(self, seq: int, frame_id: int) -> Dict[int, Dict[str, Any]]:
+    def _plan_crop_requests(
+        self, frame_shape: Tuple[int, int], seq: int, frame_id: int, measured: Set[int]
+    ) -> Dict[int, Dict[str, Any]]:
         """Ask the phone for a fuller crop of tracks the frame crop could not use.
 
         The box is in the pixel space of `frame_id`, the frame the client just
         sent, so the phone can answer from the buffer behind that exact frame
-        rather than from whatever is on screen now. One request per track object,
+        rather than from whatever is on screen now. That is also why the box
-        at most `max_crop_inflight` outstanding, and only for a client that
+        has to be one this frame measured and has to sit clear of the edge: the
+        phone crops by these numbers, so a carried-over or clipped box buys a
+        photograph of the wrong thing. At most `max_attempts` requests per
+        object, `max_crop_inflight` outstanding, and only for a client that
         advertised the lane.
         """
         self._expire_crop_requests(seq)
@@ -801,16 +869,27 @@ class LiveSession:
         room = self.settings.max_crop_inflight - len(self._crop_requests)
         if room <= 0:
             return {}
+        if (self._last_quality or {}).get("hint") in CROP_BLIND_QUALITY:
+            # A blurred or glared frame photographs the same bad scene at the
+            # camera's own resolution, so there is nothing to ask the phone for.
+            return {}
         planned: Dict[int, Dict[str, Any]] = {}
         for track in self._tracks.values():
             if len(planned) >= room:
                 break
-            if track.crop_requested or track.in_flight > 0:
+            if track.crop_attempts >= self.settings.max_attempts or track.in_flight > 0:
                 continue
             if track.state != STATE_DETECTED or track.confirmed:
                 continue
             if track.hits < self.settings.min_hits or len(track.raw_box) != 4:
                 continue
+            if track.track_id not in measured:
+                # The box is a carry-over: either no detection matched this
+                # frame, or the one that did was too low a score to move it.
+                # Either way it points at where the product used to be.
+                continue
+            if not _clear_of_edge(track.raw_box, frame_shape, self.settings.edge_margin_ratio):
+                continue
             x1, y1, x2, y2 = track.raw_box
             # The frame crop is too small to say anything about this product. At
             # the camera's own resolution it may still be worth a decision, and
@@ -832,7 +911,7 @@ class LiveSession:
                 "seq": seq,
                 "method": self.settings.method,
             }
-            track.crop_requested = True
+            track.crop_attempts += 1
             # Hold the track while the phone answers, so the frame path does not
             # spend a second attempt on the same box at the same time.
             track.in_flight += 1
@@ -899,14 +978,22 @@ class LiveSession:
         generation = record["generation"]
         method = record["method"]
         object_generation = record["object_generation"]
+        with self._lock:
+            # Decoding is the slow part and the session keeps running through it,
+            # so the binding checked before the decode is re-checked here. The
+            # check and the submit share one critical section: an ack has to mean
+            # this crop was measured against the track object it answers.
+            stale = self._crop_stale_reason_locked(record)
+            if stale is not None:
+                self._release_crop_slot(record)
+                self._crops_rejected += 1
+                return _crop_error("CROP_STALE", stale)
             try:
                 future = self._executor.submit(self.engine.score_crop, crop, method)
             except RuntimeError:  # executor already shut down
-            with self._lock:
                 self._release_crop_slot(record)
                 self._crops_rejected += 1
                 return _crop_error("CROP_UNAVAILABLE", "classifier is shutting down")
-        with self._lock:
             self._crops_accepted += 1
         future.add_done_callback(
             lambda fut, tid=track_id_out, gen=generation, m=method, started=submitted_at,
@@ -975,6 +1062,26 @@ class LiveSession:
             # the track is by the time the answer lands.
             return record, list(record["box"]), payload
 
+    def _crop_stale_reason_locked(self, record: Dict[str, Any]) -> Optional[str]:
+        """Why a decoded crop can no longer speak for its track, or None.
+
+        Run after the decode, under the lock that submits the classification.
+        A reset or a method change, a recycled track id, a pan that carried the
+        track off the photographed product, or another attempt that already
+        settled the track can all land while the JPEG is being decoded, and an
+        ack is a promise that this crop decided that track. Lock must be held.
+        """
+        if record["generation"] != self._generation:
+            return "the session changed while the crop was decoded"
+        track = self._tracks.get(record["track_id"])
+        if track is None or track.object_generation != record["object_generation"]:
+            return "the track id was recycled while the crop was decoded"
+        if _crop_off_product(record["box"], track.raw_box):
+            return "the track moved off the product while the crop was decoded"
+        if track.state != STATE_DETECTED or track.confirmed:
+            return "the track is no longer waiting to be decided"
+        return None
+
     def _on_classified(
         self,
         future: Future,
@@ -1000,6 +1107,12 @@ class LiveSession:
                 track.in_flight = max(0, track.in_flight - 1)
                 return
             track.in_flight = max(0, track.in_flight - 1)
+            if track.state != STATE_DETECTED or track.confirmed:
+                # Other evidence settled this track while the answer sat in the
+                # classifier - a label inherited from a labelled neighbour, say.
+                # A late answer must not repaint a label the client is already
+                # showing, so the track keeps the one it settled on.
+                return
             if source == SOURCE_CROP and _crop_off_product(submitted_box, track.raw_box):
                 # The answer was measured on the box the track has since left.
                 # Dropping it leaves the track to be decided from a later frame

recognization/code/test_api.py

@@ -366,8 +366,35 @@ class WebSocketProtocolTests(unittest.TestCase):
             self.assertEqual(ack["type"], "capabilities_ack")
             self.assertEqual(ack["capabilities"], {"crop": True, "max_crop_bytes": 65536})
 
+    def test_a_capability_the_server_cannot_honour_is_refused(self):
+        """A half-stated or malformed capability turns nothing on."""
+        for message in (
+            {"type": "capabilities", "max_crop_bytes": 65536},
+            {"type": "capabilities", "crop": False, "max_crop_bytes": 65536},
+            {"type": "capabilities", "crop": "true", "max_crop_bytes": 65536},
+            {"type": "capabilities", "crop": True},
+            {"type": "capabilities", "crop": True, "max_crop_bytes": 0},
+            {"type": "capabilities", "crop": True, "max_crop_bytes": -4096},
+            {"type": "capabilities", "crop": True, "max_crop_bytes": True},
+            {"type": "capabilities", "crop": True, "max_crop_bytes": "65536"},
+            {"type": "capabilities", "crop": True, "max_crop_bytes": 64 * 1024 * 1024},
+        ):
+            with self.subTest(capability=message):
+                with self.crop_app().websocket_connect("/v1/stream") as socket:
+                    socket.receive_json()  # hello
+                    socket.send_json(message)
+                    self.assertEqual(socket.receive_json()["code"], "BAD_MESSAGE")
+                    # A refused capability leaves this session on protocol v1: the
+                    # facing it cannot report is never asked about.
+                    for frame_id in (1, 2, 3):
+                        socket.send_bytes(frame_payload(frame_id))
+                        result = socket.receive_json()
+                    self.assertEqual(result["stats"]["crops"]["requested"], 0)
+                    for track in result["tracks"]:
+                        self.assertNotIn("crop_request", track)
+
     def test_a_v1_client_is_never_asked_for_a_crop(self):
-        with self.client.websocket_connect("/v1/stream") as socket:
+        with self.crop_app().websocket_connect("/v1/stream") as socket:
             socket.receive_json()  # hello
             for frame_id in (1, 2, 3):
                 socket.send_bytes(frame_payload(frame_id))
@@ -379,7 +406,7 @@ class WebSocketProtocolTests(unittest.TestCase):
     def test_a_crop_round_trip_labels_the_track(self):
         with self.crop_app().websocket_connect("/v1/stream") as socket:
             socket.receive_json()  # hello
-            socket.send_json({"type": "capabilities", "crop": True})
+            socket.send_json({"type": "capabilities", "crop": True, "max_crop_bytes": 131072})
             self.assertEqual(socket.receive_json()["type"], "capabilities_ack")
             socket.send_bytes(frame_payload(1))
             socket.receive_json()
@@ -406,7 +433,7 @@ class WebSocketProtocolTests(unittest.TestCase):
     def test_a_crop_that_does_not_match_its_request_is_refused(self):
         with self.crop_app().websocket_connect("/v1/stream") as socket:
             socket.receive_json()  # hello
-            socket.send_json({"type": "capabilities", "crop": True})
+            socket.send_json({"type": "capabilities", "crop": True, "max_crop_bytes": 131072})
             socket.receive_json()
             socket.send_bytes(frame_payload(1))
             socket.receive_json()
@@ -429,7 +456,7 @@ class WebSocketProtocolTests(unittest.TestCase):
     def test_a_crop_that_is_not_an_image_is_refused(self):
         with self.crop_app().websocket_connect("/v1/stream") as socket:
             socket.receive_json()  # hello
-            socket.send_json({"type": "capabilities", "crop": True})
+            socket.send_json({"type": "capabilities", "crop": True, "max_crop_bytes": 131072})
             socket.receive_json()
             socket.send_bytes(frame_payload(1))
             socket.receive_json()
@@ -445,6 +472,28 @@ class WebSocketProtocolTests(unittest.TestCase):
             })
             self.assertEqual(socket.receive_json()["code"], "CROP_UNDECODABLE")
 
+    def test_a_small_jpeg_claiming_huge_pixels_is_refused_before_decode(self):
+        with self.crop_app().websocket_connect("/v1/stream") as socket:
+            socket.receive_json()
+            socket.send_json({"type": "capabilities", "crop": True, "max_crop_bytes": 131072})
+            socket.receive_json()
+            socket.send_bytes(frame_payload(1))
+            socket.receive_json()
+            socket.send_bytes(frame_payload(2))
+            request = socket.receive_json()["tracks"][0]["crop_request"]
+            jpeg = bytearray(jpeg_bytes(80, 120))
+            sof = jpeg.index(b"\xff\xc0")
+            jpeg[sof + 5:sof + 9] = (16384).to_bytes(2, "big") * 2
+            socket.send_json({
+                "type": "crop", "request_id": request["request_id"],
+                "track_id": request["track_id"], "frame_id": request["frame_id"],
+                "jpeg": base64.b64encode(jpeg).decode("ascii"),
+            })
+            refused = socket.receive_json()
+            self.assertEqual(refused["code"], "CROP_TOO_LARGE")
+            socket.send_json({"type": "ping", "t": 1})
+            self.assertEqual(socket.receive_json(), {"type": "pong", "t": 1})
+
     def test_a_crop_without_an_image_is_refused(self):
         with self.client.websocket_connect("/v1/stream") as socket:
             socket.receive_json()  # hello

recognization/code/test_live.py

@@ -25,7 +25,7 @@ import cv2  # noqa: E402
 import numpy as np  # noqa: E402
 
 from engine import RecognitionEngine  # noqa: E402
-from live import LiveSession, LiveSettings  # noqa: E402
+from live import MAX_CROP_BYTES, LiveSession, LiveSettings  # noqa: E402
 
 
 class FakeDetector:
@@ -109,14 +109,23 @@ def frame_bytes(width=640, height=480):
     return buffer.tobytes()
 
 
-def process(session, frame_id=1):
+def process_payload(session, frame_id, payload):
-    payload = frame_bytes()
     session.submit_frame(frame_id, payload)
     pending = session.take_pending(timeout=1.0)
     assert pending is not None, "the session dropped a frame it should have kept"
     return session.process_frame(*pending)
 
 
+def process(session, frame_id=1):
+    return process_payload(session, frame_id, frame_bytes())
+
+
+def blurred_frame_bytes(width=640, height=480, value=120):
+    """A flat grey capture: the whole frame is out of focus, not just a facing."""
+    ok, buffer = cv2.imencode(".jpg", np.full((height, width, 3), value, dtype="uint8"))
+    return buffer.tobytes()
+
+
 def wait_for(predicate, timeout=2.0):
     deadline = time.time() + timeout
     while time.time() < deadline:
@@ -556,6 +565,34 @@ class GatedMatcher(FakeMatcher):
         return super().score_crop(crop, method)
 
 
+class SecondAnswerWaits(FakeMatcher):
+    """Answers the first crop at once, holds the second one, and disagrees.
+
+    Two facings that overlap this much are the only way a track can be handed
+    a settled label while its own answer is still in the classifier, so the
+    test answers the first crop, lets the track settle, and only then hands
+    over a second crop whose answer names a different product.
+    """
+
+    def __init__(self, sku="sku_a", other="other"):
+        super().__init__(sku=sku)
+        self.gate = threading.Event()
+        self.first_answered = threading.Event()
+        self.disagreeing = other
+        self._seen = 0
+
+    def score_crop(self, crop, method="hybrid"):
+        with self.lock:
+            self._seen += 1
+            turn = self._seen
+            self.calls += 1
+        if turn == 1:
+            self.first_answered.set()
+            return {self.sku: 40.0, "other": 0.1}, {}
+        self.gate.wait(2.0)
+        return {self.disagreeing: 40.0, self.sku: 0.1}, {}
+
+
 class CropLaneTests(unittest.TestCase):
     """A full-resolution crop is extra evidence for one track, nothing else."""
 
@@ -764,6 +801,216 @@ class CropLaneTests(unittest.TestCase):
         self.assertEqual(session.capabilities()["crop"], True)
         engine.shutdown()
 
+    def test_a_refused_limit_never_turns_the_lane_on(self):
+        session, engine, _ = make_crop_session()
+        session.settings.crops = False  # a lane no client has advertised yet
+        for bad in (True, 0, -1, 4096.0, "4096", MAX_CROP_BYTES + 1):
+            with self.subTest(limit=bad):
+                with self.assertRaises(ValueError):
+                    session.enable_crops(bad)
+                self.assertFalse(session.capabilities()["crop"],
+                                 "a refused limit must leave the lane off")
+        self.assertEqual(session.enable_crops(4096)["max_crop_bytes"], 4096)
+        engine.shutdown()
+
+    def test_a_reset_while_the_crop_decodes_is_refused(self):
+        """An ack promises this crop decided that track; a reset says it cannot."""
+        matcher = FakeMatcher()
+        session, engine, _ = make_crop_session(matcher=matcher)
+        process(session, 1)
+        request = requested_crop(process(session, 2))
+        self.assertIsNotNone(request)
+
+        decoding = threading.Event()
+        resumed = threading.Event()
+        decode = engine.decode_image
+
+        def gated_decode(payload):
+            decoding.set()
+            resumed.wait(2.0)
+            return decode(payload)
+
+        engine.decode_image = gated_decode
+        replies = {}
+
+        def answer():
+            replies["reply"] = session.submit_crop(
+                request["request_id"], request["track_id"], request["frame_id"], frame_bytes()
+            )
+
+        worker = threading.Thread(target=answer)
+        worker.start()
+        self.assertTrue(decoding.wait(2.0), "the crop must be decoding before the reset")
+        session.reset()
+        resumed.set()
+        worker.join(5.0)
+        self.assertIn("reply", replies, "the answer never came back")
+        self.assertEqual(replies["reply"]["code"], "CROP_STALE")
+        self.assertEqual(session._crops_accepted, 0)
+        self.assertEqual(matcher.calls, 0, "a refused crop never reaches the classifier")
+        engine.shutdown()
+
+    def test_a_blurred_frame_is_not_asked_about(self):
+        """More pixels of an out-of-focus scene are not more evidence."""
+        session, engine, _ = make_crop_session()
+        process(session, 1)
+        result = process_payload(session, 2, blurred_frame_bytes())
+        self.assertEqual(result["quality"]["hint"], "blur",
+                         "the frame under test must be the blurry one")
+        self.assertIsNone(requested_crop(result))
+        self.assertEqual(result["stats"]["crops"]["requested"], 0)
+        engine.shutdown()
+
+    def test_a_distant_shelf_is_still_asked_about(self):
+        """Too far to read at the frame's own resolution is the case a crop exists for."""
+        session, engine, detector = make_crop_session(
+            settings=LiveSettings(method="hybrid", min_crop_px=60)
+        )
+        detector.boxes[0].update({
+            "box": [300, 380, 336, 430], "width": 36, "height": 50, "center": [318, 405],
+        })
+        process(session, 1)
+        result = process(session, 2)
+        self.assertEqual(result["quality"]["hint"], "too_far",
+                         "the frame under test must be the distant one")
+        self.assertIsNotNone(requested_crop(result))
+        engine.shutdown()
+
+    def test_a_facing_the_frame_clips_is_not_asked_about(self):
+        session, engine, detector = make_crop_session()
+        clipped = {"box": [0, 100, 36, 220], "width": 36, "height": 120, "center": [18, 160]}
+        detector.boxes[0].update(clipped)
+        process(session, 1)
+        result = process(session, 2)
+        self.assertIsNone(requested_crop(result), "half a facing is not worth the phone's pixels")
+        self.assertEqual(result["stats"]["crops"]["requested"], 0)
+        # The same facing, walked clear of the edge, is asked about: only the
+        # clip kept it out of the lane.
+        asked = None
+        for frame_id, x in enumerate([12, 24, 36], start=3):
+            detector.boxes[0].update({"box": [x, 100, x + 36, 220], "center": [x + 18, 160]})
+            result = process(session, frame_id)
+            asked = asked or requested_crop(result)
+            if asked is not None:
+                break
+        self.assertIsNotNone(asked, "a whole facing is worth the phone's pixels")
+        engine.shutdown()
+
+    def test_a_box_this_frame_did_not_measure_is_not_asked_about(self):
+        """The phone crops by the box it is sent, so a stale box crops the wrong thing."""
+        session, engine, detector = make_crop_session()
+        process(session, 1)
+        detector.boxes[0]["confidence"] = 0.15  # too low for the tracker to trust
+        result = process(session, 2)
+        self.assertIsNone(result["quality"]["hint"], "only the detection is at fault here")
+        self.assertIsNone(requested_crop(result))
+        detector.boxes[0]["confidence"] = 0.9
+        self.assertIsNotNone(requested_crop(process(session, 3)),
+                            "the same facing is asked about once a frame measures it again")
+        engine.shutdown()
+
+    def test_a_method_change_lets_the_small_facing_ask_again(self):
+        """A crop minted under the old method is not an attempt under the new one."""
+        session, engine, _ = make_crop_session()
+        process(session, 1)
+        first = requested_crop(process(session, 2))
+        self.assertIsNotNone(first)
+        session.configure(method="orb")
+        second = requested_crop(process(session, 3))
+        self.assertIsNotNone(second, "the new method gets a budget of its own")
+        self.assertNotEqual(second["request_id"], first["request_id"])
+        self.assertEqual(process(session, 4)["stats"]["crops"]["requested"], 2)
+        late = session.submit_crop(first["request_id"], first["track_id"],
+                                   first["frame_id"], frame_bytes())
+        self.assertEqual(late["code"], "CROP_UNKNOWN_REQUEST",
+                         "the outstanding request was minted under settings the session left")
+        engine.shutdown()
+
+    def test_an_expired_request_leaves_exactly_one_more_ask(self):
+        session, engine, _ = make_crop_session(settings=LiveSettings(crop_ttl_frames=1))
+        process(session, 1)
+        first = requested_crop(process(session, 2))
+        self.assertIsNotNone(first)
+        process(session, 3)
+        # The one-frame budget ran out on frame 4, so the facing gets one more
+        # ask rather than being left with a request nobody will ever answer.
+        second = requested_crop(process(session, 4))
+        self.assertIsNotNone(second, "an expiry must not strand a facing only the crop can decide")
+        self.assertNotEqual(second["request_id"], first["request_id"])
+        process(session, 5)
+        process(session, 6)
+        result = process(session, 7)
+        self.assertIsNone(requested_crop(result), "the budget is bounded, not a loop")
+        self.assertEqual(result["stats"]["crops"]["requested"], 2)
+        engine.shutdown()
+
+    def test_a_drifted_label_lets_the_next_product_be_asked_about(self):
+        """The facing in the frame is a different product, so it gets its own ask."""
+        session, engine, detector = make_crop_session()
+        panned(detector, 100)
+        process(session, 1)
+        request = requested_crop(process(session, 2))
+        session.submit_crop(request["request_id"], request["track_id"],
+                            request["frame_id"], frame_bytes())
+        self.assertTrue(wait_for(lambda: session._tracks[request["track_id"]].state == "recognized"))
+        track_id = request["track_id"]
+        asked_again = None
+        for frame_id, x in enumerate([118, 136, 154, 172, 190, 208], start=3):
+            panned(detector, x)
+            result = process(session, frame_id)
+            self.assertIn(track_id, session._tracks, "the id must survive the pan")
+            again = requested_crop(result)
+            if again is not None and again["request_id"] != request["request_id"]:
+                asked_again = again
+                break
+        self.assertIsNotNone(asked_again,
+                             "a drifted label must hand the new product its own crop ask")
+        self.assertGreater(asked_again["box"][0], request["box"][0],
+                           "the new ask must be about where the product is now")
+        engine.shutdown()
+
+    def test_a_late_crop_cannot_repaint_a_settled_label(self):
+        """A track that took a neighbour's label keeps it; a later crop does not overrule it."""
+        matcher = SecondAnswerWaits()
+        session, engine, detector = make_crop_session(
+            matcher=matcher, settings=LiveSettings(method="hybrid", min_crop_px=60)
+        )
+        # Two narrow facings that overlap by more than a label can be trusted
+        # across: that is the only shape in which one inherits the other's.
+        detector.boxes = [
+            {"id": 0, "box": [100, 100, 150, 340], "confidence": 0.9, "class_id": 0,
+             "width": 50, "height": 240, "center": [125, 220]},
+            {"id": 1, "box": [110, 100, 160, 340], "confidence": 0.9, "class_id": 0,
+             "width": 50, "height": 240, "center": [135, 220]},
+        ]
+        process(session, 1)
+        requests = {row["track_id"]: row["crop_request"]
+                    for row in process(session, 2)["tracks"] if "crop_request" in row}
+        self.assertEqual(len(requests), 2, "both facings are too small for the frame crop")
+        settled_id, waiting_id = list(requests)
+
+        session.submit_crop(requests[settled_id]["request_id"], settled_id,
+                            requests[settled_id]["frame_id"], frame_bytes())
+        self.assertTrue(matcher.first_answered.wait(2.0), "the first crop must be decided")
+        session.submit_crop(requests[waiting_id]["request_id"], waiting_id,
+                            requests[waiting_id]["frame_id"], frame_bytes())
+
+        inherited = process(session, 3)
+        row = next(t for t in inherited["tracks"] if t["track_id"] == waiting_id)
+        self.assertTrue(row["inherited"], "the facing takes the label its neighbour settled")
+        self.assertEqual(row["identity"]["sku_id"], "sku_a")
+
+        matcher.gate.set()
+        self.assertTrue(wait_for(lambda: session._tracks[waiting_id].in_flight == 0))
+        settled = next(t for t in process(session, 4)["tracks"] if t["track_id"] == waiting_id)
+        self.assertEqual(settled["identity"]["sku_id"], "sku_a",
+                         "a crop that arrives late must not overrule a settled label")
+        self.assertTrue(settled["inherited"])
+        self.assertEqual(settled["identity"]["source"], "inherited")
+        self.assertEqual(session._tracks[waiting_id].attempts, 0,
+                         "an answer that could not decide is not an attempt")
+        engine.shutdown()
+
 
 if __name__ == "__main__":
     unittest.main()

recognization/docs/API.md

@@ -272,24 +272,30 @@ classification, not recognized; subsequent results carry the updated identity an
 `identity.source` (`frame`, `crop`, or `inherited`). `stats.crops` has `requested`,
 `accepted`, and `rejected` counts.
 
-The hello limit is 1 MiB of **decoded crop bytes**. The optional
+The hello limit is 1 MiB of **decoded crop bytes**. The `max_crop_bytes` in
-`max_crop_bytes` in `capabilities` states the client's limit; the acknowledged
+`capabilities` states the client's limit and must be an integer in 1..1048576; a
-session limit is clamped to 1 KiB–1 MiB (default 128 KiB when omitted). Use the
+statement that is not one is refused with `BAD_MESSAGE` and the lane stays where
+it was, so both sides keep one limit (128 KiB when the key is `null`). Use the
 acknowledged limit, not only hello's ceiling. At most four requests are outstanding
-per session and each track object is requested at most once. A request expires
+per session, and each track object is asked about at most `max_attempts` times
+(the same budget the frame path spends, 2 by default); the budget refills when the
+method changes or when a settled label drifts off the object. A request expires
 after more than six processed frames; clients keep a bounded recent-frame buffer
 and may miss a crop if its capture has already been evicted. A late response
-cannot label a recycled track. Changing method or resetting invalidates requests.
+cannot label a recycled track. Changing method, resetting, or dropping a drifted
+label invalidates that object's outstanding requests.
 
 Crop failures are recoverable text messages with `type: "crop_error"`, `code`,
 and `message`. Pre-decode failures also echo `request_id`; session-level refusals
 do not currently echo it. Codes: `CROP_BAD_MESSAGE` (missing/invalid base64 JPEG
-or missing IDs), `CROP_TOO_LARGE` (empty or oversized decoded crop, or excessive
+or missing IDs), `CROP_TOO_LARGE` (empty or oversized decoded crop, excessive
-encoded length), `CROP_UNKNOWN_REQUEST` (no outstanding request, including one
+encoded length, or a JPEG whose header claims more than 4 megapixels),
-already expired), `CROP_MISMATCH` (track/frame differs), `CROP_STALE` (late or
+`CROP_UNKNOWN_REQUEST` (no outstanding request, including one already expired),
+`CROP_MISMATCH` (track/frame differs), `CROP_STALE` (late or
 invalidated track/session), `CROP_UNDECODABLE` (unreadable image), and
 `CROP_UNAVAILABLE` (classifier shutting down). A request is single-use, including
 when its response is rejected; continue streaming frames rather than retrying it.
+The server may ask again for the same object while its budget lasts.
 
 **Backpressure:** the server keeps one pending frame per session and processes the
 newest, counting the ones it replaced in `frames_dropped`. A client should keep one