product-recognition-demo/recognization/code/test_live.py

"""
Contract tests for the live session.

Everything runs against a fake engine, so these tests are fast and deterministic.
They defend the behaviours a live shelf view depends on: labels that appear
without blocking frames, labels that never silently change, restarted ids that
are re-verified rather than inherited, and backpressure that always keeps the
newest frame.
"""

from __future__ import annotations

import os
import asyncio
import sys
import threading
import time
import unittest

current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
    sys.path.insert(0, current_dir)

import cv2  # noqa: E402
import numpy as np  # noqa: E402

from engine import RecognitionEngine  # noqa: E402
from live import MAX_CROP_BYTES, LiveSession, LiveSettings  # noqa: E402


class FakeDetector:
    weights_path = "fake.pt"

    def __init__(self):
        self.boxes = [{"id": 0, "box": [100, 100, 200, 260], "confidence": 0.9, "class_id": 0,
                       "width": 100, "height": 160, "center": [150, 180]}]

    def detect(self, image, conf_threshold=0.25, iou_threshold=0.45):
        return {"status": "success", "boxes": [dict(b) for b in self.boxes]}

    def detect_empty_slots(self, image_shape, boxes, min_gap_ratio=0.75):
        return []


class FakeMatcher:
    def __init__(self, sku="sku_a", score=40.0, delay=0.0):
        self.sku = sku
        self.score = score
        self.delay = delay
        self.calls = 0
        self.lock = threading.Lock()

    def score_crop(self, crop, method="hybrid"):
        with self.lock:
            self.calls += 1
        if self.delay:
            time.sleep(self.delay)
        # The runner-up stays far below the winner so the decision is about the
        # configured score, not about ambiguity.
        return {self.sku: self.score, "other": 0.1}, {}

    def score_crop_fast(self, crop, shortlist=5):
        return self.score_crop(crop)


class FakeGallery:
    """The gallery's own accept line, which the live aggregate must clear."""

    head_threshold = 0.40
    threshold = 0.55

    def __init__(self, head=True):
        self._head = head

    def head_loaded(self):
        return self._head


class CosineEngine(RecognitionEngine):
    """A gallery-scale engine: score_crop only returns scores the gallery accepted."""

    def __init__(self, *args, head=True, **kwargs):
        super().__init__(*args, **kwargs)
        self._fake_gallery = FakeGallery(head=head)

    @property
    def identity_scale(self):
        return "cosine"

    def identity_gallery(self):
        return self._fake_gallery

    def score_crop(self, crop, method=None):
        scores, _ = self.matcher.score_crop(crop, method or "hybrid")
        return scores


class LocalScaleEngine(RecognitionEngine):
    """The shipped default scale, with the lanes' evidence injected.

    `local` is the branch that was wrong twice - it re-sorted the lane's own
    order by value, and its early return on an empty local observation made the
    text lane's "answer on its own" job unreachable - so it is tested here
    rather than only through the real lane, which the fake matcher cannot drive.
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.lane_answers = []
        self.ocr_calls = 0

    @property
    def identity_scale(self):
        return "local"

    def score_crop_lanes(self, crop, method=None, skip_ocr=False):
        answer = self.lane_answers.pop(0) if self.lane_answers else ({}, [], "")
        if skip_ocr:
            # The real engine re-runs only the local lane and returns no text;
            # the live path reuses the reading it cached on the track.
            return answer[0], [], ""
        self.ocr_calls += 1
        return answer


class FakeInventory:
    """Stateful stand-in for InventoryManager, so stock assertions are real."""

    def __init__(self):
        self.products = {
            "sku_a": {"sku_id": "sku_a", "product_name": "コクうま カロリーカット65%", "price": 150.0,
                      "number_in": 10, "number_out": 1, "number_on": 9,
                      "receipt_date": "", "expiration_date": "", "local_image_path": None},
            "other": {"sku_id": "other", "product_name": "アマニ油マヨネーズ", "price": 150.0,
                      "receipt_date": "", "expiration_date": "", "local_image_path": None},
        }

    def list_all_skus(self):
        return list(self.products)

    def get_product(self, sku_id):
        return self.products.get(sku_id)

    def update_stock(self, sku_id, delta):
        product = self.products.get(sku_id)
        if product is None:
            return False
        product["number_on"] = max(0, product["number_on"] + delta)
        if delta < 0:
            product["number_out"] += abs(delta)
        else:
            product["number_in"] += delta
        return True


def make_session(matcher=None, settings=None, engine_cls=RecognitionEngine):
    detector = FakeDetector()
    matcher = matcher or FakeMatcher()
    # These fixtures inject a fake matcher, so they exercise the `matcher`
    # backend explicitly: the shipped default is the `local` lane, which reads
    # real reference images and would ignore the fake.
    engine = engine_cls(detector, matcher, FakeInventory(),
                        settings=__import__("engine").EngineSettings(
                            classify_workers=2, identity_backend="matcher"))
    session = LiveSession("test", engine, settings=settings or LiveSettings(method="hybrid"))
    session.refresh_catalog()
    return session, engine, detector, matcher


def frame_bytes(width=640, height=480):
    ok, buffer = cv2.imencode(".jpg", np.zeros((height, width, 3), dtype="uint8"))
    return buffer.tobytes()


def process_payload(session, frame_id, payload):
    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:
        if predicate():
            return True
        time.sleep(0.01)
    return False


class LabelLifecycleTests(unittest.TestCase):
    def test_first_frame_is_detected_then_labelled(self):
        session, engine, _, matcher = make_session()
        first = process(session, 1)
        self.assertEqual(first["tracks"][0]["identity"]["state"], "unconfirmed")
        self.assertIsNone(first["tracks"][0]["identity"]["sku_id"])

        process(session, 2)  # second frame: eligible for classification
        self.assertTrue(wait_for(lambda: matcher.calls > 0), "classification should be scheduled")
        self.assertTrue(wait_for(lambda: any(
            track["identity"]["state"] == "recognized" for track in process(session, 3)["tracks"]
        )), "the product should be labelled once classified")
        engine.shutdown()

    def test_a_weak_match_is_shown_provisional_not_accepted(self):
        """Coverage-first: a below-gate candidate is shown, but marked provisional.

        The old contract left this box blank. Blank is now the worse outcome:
        the operator sees a guess that the next attempt may upgrade, and the
        wire says which it is, so a client can draw it differently.
        """
        session, engine, _, _ = make_session(FakeMatcher(score=0.4))  # below the accept threshold
        for frame_id in range(1, 8):
            process(session, frame_id)
        result = process(session, 9)
        track = result["tracks"][0]
        self.assertEqual(track["identity"]["sku_id"], "sku_a",
                         "a weak match must still be shown")
        self.assertEqual(track["identity"]["identity_level"], "provisional",
                         "a weak match must not claim to be accepted")
        engine.shutdown()

    def test_a_provisional_label_is_never_blanked(self):
        """Once shown, a guess stays: the budget running out must not take it back."""
        session, engine, _, _ = make_session(FakeMatcher(score=0.4))
        for frame_id in range(1, 12):
            process(session, frame_id)
        result = process(session, 13)
        track = result["tracks"][0]
        self.assertEqual(track["identity"]["sku_id"], "sku_a",
                         "a spent budget must not blank a label already on screen")
        engine.shutdown()

    def test_restarted_id_is_re_verified_not_inherited(self):
        session, engine, detector, matcher = make_session()
        for frame_id in range(1, 5):
            process(session, frame_id)
        self.assertTrue(wait_for(lambda: process(session, 5)["tracks"][0]["identity"]["state"]
                                 == "recognized"))
        first_relabels = process(session, 6)["tracks"][0]["relabels"]

        detector.boxes = []           # the product leaves the frame
        for frame_id in range(7, 14):
            process(session, frame_id)
        detector.boxes = [{"id": 0, "box": [100, 100, 200, 260], "confidence": 0.9, "class_id": 0,
                           "width": 100, "height": 160, "center": [150, 180]}]
        result = process(session, 14)
        track = result["tracks"][0]
        self.assertGreaterEqual(track["relabels"], first_relabels + 1,
                                "a restarted id must count as re-verified")
        self.assertGreaterEqual(track["restarts"], 1)
        engine.shutdown()

    def test_inherited_label_is_flagged(self):
        """A fresh track covering a labelled product inherits the label, and says so."""
        session, engine, detector, matcher = make_session()
        for frame_id in range(1, 5):
            process(session, frame_id)
        self.assertTrue(wait_for(lambda: process(session, 5)["tracks"][0]["identity"]["state"]
                                 == "recognized"))
        # same place, new detection id => the tracker treats it as a new object
        detector.boxes = [{"id": 0, "box": [102, 100, 202, 260], "confidence": 0.9, "class_id": 0,
                           "width": 100, "height": 160, "center": [152, 180]}]
        result = process(session, 6)
        self.assertTrue(result["tracks"][0]["inherited"] or result["tracks"][0]["identity"]["state"]
                        == "recognized")
        engine.shutdown()

    def test_a_pan_does_not_let_a_neighbour_inherit(self):
        """A label must not slide onto whatever moved into its screen position.

        The memory box is stored in the coordinates of the frame it was
        measured on. Under a pan the shelf slides beneath that box, so the
        product now occupying the old screen position is a different one.
        Comparing the raw box hands it the label at IoU 1.0; compensating by
        the shelf movement drops the overlap to nothing.
        """
        session, engine, detector, matcher = make_session()
        for frame_id in range(1, 5):
            process(session, frame_id)
        self.assertTrue(wait_for(lambda: process(session, 5)["tracks"][0]["identity"]["state"]
                                 == "recognized"))
        # The product leaves, so its track ages out and the next detection at
        # that screen position is a new object rather than the same one.
        detector.boxes = []
        for frame_id in range(6, 10):
            process(session, frame_id)
        # The shelf slides 90 px: the labelled product is now 90 px to the
        # left, and the screen position it used to occupy holds its neighbour.
        session._frame_origin = [90.0, 0.0]
        detector.boxes = [{"id": 0, "box": [100, 100, 200, 260], "confidence": 0.9, "class_id": 0,
                           "width": 100, "height": 160, "center": [150, 180]}]
        result = process(session, 10)
        row = result["tracks"][0]
        self.assertFalse(row["inherited"],
                         "the product that slid into the old box is not the labelled one")
        engine.shutdown()


class CosineScaleTests(unittest.TestCase):
    """A gallery-scale score is gated by the gallery, not re-gated by the live path."""

    def test_a_head_scale_match_is_labelled(self):
        """0.45 clears the head's own 0.40 line but not COSINE_ACCEPT (0.60).

        score_crop only returns scores the gallery already accepted, so the live
        aggregate must not re-apply the cosine accept line: that refused every
        head match the gallery had accepted, which is the live regression.
        """
        session, engine, _, _ = make_session(FakeMatcher(score=0.45), engine_cls=CosineEngine)
        for frame_id in range(1, 5):
            process(session, frame_id)
        self.assertTrue(wait_for(lambda: any(
            track["identity"]["state"] == "recognized" for track in process(session, 5)["tracks"]
        )), "a gallery-accepted head match must be labelled")
        engine.shutdown()

    def test_a_close_cosine_pair_is_shown_provisional(self):
        """The margin gate still applies to *acceptance*: two near-equal scores
        must not be named as settled. Coverage-first shows the top one anyway,
        marked provisional, because a blank box is the worse outcome."""
        session, engine, _, _ = make_session(FakeMatcher(score=0.45), engine_cls=CosineEngine)
        # The matcher contract is (scores, details); a bare dict unpacks into two
        # strings and raises inside the future callback, where the failure is
        # swallowed and the test passes for the wrong reason.
        engine.matcher.score_crop = lambda crop, method=None: ({"sku_a": 0.45, "other": 0.44}, {})
        for frame_id in range(1, 8):
            process(session, frame_id)
        result = process(session, 9)
        for track in result["tracks"]:
            self.assertEqual(track["identity"]["identity_level"], "provisional",
                             "a cosine pair inside the margin must not be accepted")
        engine.shutdown()

    def test_a_rejected_crop_cannot_dilute_an_accepted_one(self):
        """aggregate_votes divides by the whole window, so a miss must not lower the line.

        A rejected crop first, then an accepted 0.45: the mean is 0.225. Gating the
        aggregate at zero would *accept* that, at a score the gallery calls unknown.
        Coverage-first still shows it, but the level must stay provisional: the
        dilution must not be promoted to a settled label.
        """
        session, engine, _, _ = make_session(FakeMatcher(score=0.45), engine_cls=CosineEngine)
        answers = iter([{}, {"sku_a": 0.45}])
        engine.matcher.score_crop = lambda crop, method=None: (next(answers, {}), {})
        # The retry needs retry_gap frames before the second attempt is eligible.
        for frame_id in range(1, 12):
            process(session, frame_id)
        result = process(session, 13)
        for track in result["tracks"]:
            self.assertEqual(track["identity"]["identity_level"], "provisional",
                             "a diluted mean must not be accepted below the gallery's line")
        engine.shutdown()


class LocalScaleTests(unittest.TestCase):
    """The shipped default scale, which the fake matcher cannot reach.

    `make_session` builds a `matcher`-backend engine, so `scale == "local"` in
    `_on_classified` never runs in the rest of this file - and that branch is
    the one that was wrong twice. These tests drive it with injected lane
    evidence, so a regression in the local decision path fails here.
    """

    def test_a_text_only_read_still_labels_the_track(self):
        """The text lane answers on its own when the local lane has nothing.

        The early return on an empty local observation made this unreachable:
        a SKU with no enrolled image can only ever be named by its text, so
        dropping that path silently removed the lane's fill-in job.
        """
        session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
        engine.lane_answers = [({}, [("sku_a", 0.9)], "コクうま 65%")]
        result = process(session, 1)
        self.assertTrue(
            wait_for(lambda: any(
                track["identity"]["sku_id"] == "sku_a" for track in process(session, 2)["tracks"]
            )),
            "a text-only read must label the track",
        )
        engine.shutdown()

    def test_a_text_veto_blocks_a_local_accept(self):
        """A local match whose package text names another row must not be shown.

        The sibling shares the artwork, so the geometry is real and the local
        score is high; only the words differ. The text here is below its own
        accept line, so it vetoes without filling in: the local answer must not
        reach the wire at all.

        `scores` is a dict keyed by SKU, which is what `score_crop` returns and
        what `_on_classified` iterates; a list here raises inside the future
        callback, where the failure is swallowed and the test passes for the
        wrong reason.
        """
        session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
        engine.lane_answers = [({"sku_a": 5.0}, [("other", 0.3)], "other マヨネーズ")] * 4
        for frame_id in range(1, 14):
            result = process(session, frame_id)
            for track in result["tracks"]:
                self.assertNotEqual(
                    track["identity"]["sku_id"], "sku_a",
                    "a vetoed local answer must not be labelled",
                )
        engine.shutdown()

    def test_a_size_only_read_cannot_name_a_product(self):
        """A crop reading "400g" matches every 400g product's size line.

        Size and number tokens are not identity text: without this gate the
        text lane answered a distractor crop reading only "400g" as a product.
        """
        session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
        engine.lane_answers = [({}, [("sku_a", 0.9)], "400g")] * 4
        for frame_id in range(1, 14):
            for track in process(session, frame_id)["tracks"]:
                self.assertIsNone(
                    track["identity"]["sku_id"],
                    "a size-only read must not name a product",
                )
        engine.shutdown()

    def test_ocr_runs_once_per_track(self):
        """Vision on every attempt is what made coverage pacing-dependent.

        A crop costs 5 ms without OCR and 25 ms with it, so re-reading the same
        package on every attempt cut throughput from 789 to 60 crops/s at 8
        workers. The reading is cached on the track instead, and the local lane
        keeps running on every attempt.
        """
        session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
        # Below the text accept line, so the track stays undecided and retries:
        # the retry is what would re-run OCR without the cache.
        engine.lane_answers = [({}, [("sku_a", 0.3)], "コクうま 65%")] * 6
        for frame_id in range(1, 14):
            process(session, frame_id)
        self.assertEqual(engine.ocr_calls, 1, "OCR must run once per track")
        engine.shutdown()

    def test_no_ocr_never_reads_a_crop(self):
        """The session's `ocr` switch turns the text lane off for every crop.

        The text lane is the slowest thing the server does per crop, so the
        phone can ask for a session that runs on geometry alone. Only the text
        lane goes away: the local lane still names the track, because the whole
        point of the setting is a faster session, not a quieter one.
        """
        session, engine, _, _ = make_session(
            engine_cls=LocalScaleEngine,
            settings=LiveSettings(method="hybrid", ocr=False),
        )
        engine.lane_answers = [({"sku_a": 5.0}, [("sku_a", 0.9)], "コクうま 65%")] * 6
        labelled = False
        for frame_id in range(1, 14):
            for track in process(session, frame_id)["tracks"]:
                if track["identity"]["sku_id"] == "sku_a":
                    labelled = True
        self.assertEqual(engine.ocr_calls, 0, "no crop may be read with the text lane off")
        self.assertTrue(labelled, "the local lane must still name the track")
        engine.shutdown()

    def test_the_ocr_switch_is_reported_back_to_the_client(self):
        """The client reads the session's own value, not the one it asked for."""
        session, engine, _, _ = make_session()
        self.assertTrue(session.config()["ocr"], "the text lane is on by default")
        session.configure(ocr=False)
        self.assertFalse(session.config()["ocr"], "the switch must reach the config")
        session.configure(ocr=True)
        self.assertTrue(session.config()["ocr"], "and back on again")
        engine.shutdown()

    def test_a_size_only_read_is_not_cached(self):
        """An unreadable crop must not veto the product for the track's life.

        The cache used to keep any non-empty read, so a first crop that read
        only "400g" was replayed on every later attempt: the track could never
        be named even after a clean crop arrived, because the cached read
        replaced it. Only a read that carries identity text is cached now, so
        the next attempt gets a fresh OCR.
        """
        session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
        engine.lane_answers = [
            ({}, [("sku_a", 0.9)], "400g"),
            ({}, [("sku_a", 0.9)], "コクうま 65%"),
        ] + [({}, [("sku_a", 0.9)], "コクうま 65%")] * 6
        labelled = False
        for frame_id in range(1, 14):
            for track in process(session, frame_id)["tracks"]:
                if track["identity"]["sku_id"] == "sku_a":
                    labelled = True
        self.assertTrue(labelled, "a later readable crop must still name the track")
        self.assertGreater(engine.ocr_calls, 1, "the size-only read must not be cached")
        engine.shutdown()

    def test_a_failing_lane_is_counted_and_logged(self):
        """A lane that raises is a failure, not a crop that matched nothing.

        `_on_classified` substituted an empty answer for the exception, so a
        run in which `score_crop_lanes` failed on every crop produced tracks,
        zero labels and a clean log - indistinguishable from a real coverage
        drop. The counter and the log line are what tell the two apart, and
        the session has to keep serving frames either way.
        """
        session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)

        def boom(crop, method=None, skip_ocr=False):
            raise RuntimeError("lane unavailable")

        engine.score_crop_lanes = boom
        named = None
        with self.assertLogs("dorayaki.live", level="ERROR") as logs:
            for frame_id in range(1, 14):
                tracks = process(session, frame_id)["tracks"]
                self.assertTrue(tracks, "a failing lane must not take the session down")
                named = named or next(
                    (t["identity"]["sku_id"] for t in tracks if t["identity"]["sku_id"]), None)
            self.assertTrue(wait_for(lambda: bool(logs.output)),
                            "the exception must be logged, not swallowed")
        self.assertGreaterEqual(session.snapshot()["stats"]["classify_failures"], 1,
                               "a lane failure must be counted in the session stats")
        self.assertIsNone(named, "a failed lane must not name a product")
        engine.shutdown()


class TagPassIntegrationTests(unittest.TestCase):
    """The tag pass must reach the live result without blocking frames."""

    def make_tagged_session(self):
        from ocr import FakeOcr, OcrLine

        def reader(crop):
            height, width = crop.shape[:2]
            return [
                OcrLine("コーヒー", 0.8, [int(width * 0.05), int(height * 0.05), int(width * 0.4), int(height * 0.3)]),
                OcrLine("150", 0.9, [int(width * 0.05), int(height * 0.4), int(width * 0.35), int(height * 0.8)]),
                # 150 x 1.08 = 162.00, so the tag's arithmetic corroborates the price.
                OcrLine("162.00", 0.6, [int(width * 0.55), int(height * 0.5), int(width * 0.8), int(height * 0.75)]),
            ]

        session, engine, detector, matcher = make_session()
        # Rail bands come from product rows, so the shelf needs a row of facings.
        detector.boxes = [
            {"id": index, "box": [100 + index * 120, 100, 200 + index * 120, 260],
             "confidence": 0.9, "class_id": 0, "width": 100, "height": 160,
             "center": [150 + index * 120, 180]}
            for index in range(3)
        ]
        engine.set_ocr_engine(FakeOcr(reader=reader))
        self.assertTrue(session.enable_tags())
        return session, engine, detector

    def test_track_gains_a_price_tag_and_a_reconciliation(self):
        import time as _time

        session, engine, detector = self.make_tagged_session()
        result = None
        for frame_id in range(1, 12):
            result = process(session, frame_id)
            _time.sleep(0.05)  # the tag pass runs on its own worker
            if any(
                (track.get("price_tag") or {}).get("state") == "readable"
                for track in result["tracks"]
            ):
                break
        track = result["tracks"][0]
        self.assertIn("price_tag", track, "the tag pass should attach to the facing above it")
        self.assertEqual(track["price_tag"]["price"], 150)
        self.assertIn("reconciliation", track)
        self.assertEqual(result["summary"]["price_tags"]["readable"], 1)
        engine.shutdown()

    def test_frames_are_not_blocked_by_ocr(self):
        """A slow OCR engine must not slow the frame loop: frames keep coming."""
        import time as _time

        from ocr import FakeOcr, OcrLine

        def slow_reader(crop):
            _time.sleep(0.5)
            height, width = crop.shape[:2]
            return [OcrLine("150", 0.9, [10, 10, int(width * 0.3), int(height * 0.6)])]

        session, engine, detector, _ = make_session()
        detector.boxes = [
            {"id": index, "box": [100 + index * 120, 100, 200 + index * 120, 260],
             "confidence": 0.9, "class_id": 0, "width": 100, "height": 160,
             "center": [150 + index * 120, 180]}
            for index in range(3)
        ]
        engine.set_ocr_engine(FakeOcr(reader=slow_reader))
        self.assertTrue(session.enable_tags())
        started = _time.perf_counter()
        for frame_id in range(1, 4):
            process(session, frame_id)
        elapsed = _time.perf_counter() - started
        self.assertLess(elapsed, 1.0, "three frames must not wait for a 0.5 s OCR pass")
        engine.shutdown()

    def test_config_can_turn_tags_off(self):
        session, engine, _ = self.make_tagged_session()
        config = session.configure(tags=False)
        self.assertFalse(config["tags"])
        self.assertTrue(config["tags_available"])
        result = process(session, 1)
        self.assertNotIn("price_tag", result["tracks"][0])
        engine.shutdown()

    def test_a_panned_shelf_reusing_the_screen_row_is_planned(self):
        """The same screen row on a shelf that has moved is a new rail."""
        session, engine, _ = self.make_tagged_session()
        session._tags.settings.pass_interval_frames = 0
        session._tags.settings.settle_observations = 1
        session._tags.settings.settle_passes = 1

        class Panned:
            def __init__(self):
                self.calls = 0

            def register(self, image):
                self.calls += 1
                # Content moved up 400 px, so the frame origin moves down the shelf.
                return (0.0, -400.0) if self.calls == 2 else (0.0, 0.0)

        session._shift_map = Panned()
        process(session, 1)
        time.sleep(0.05)
        before = session._tags.stats()["passes_run"]
        self.assertGreaterEqual(before, 1)
        process(session, 2)
        time.sleep(0.1)
        self.assertGreaterEqual(session._frame_origin[1], 400.0)
        self.assertGreater(
            session._tags.stats()["passes_run"], before,
            "a shelf 400 px away reuses screen row 0 and must be read again",
        )
        engine.shutdown()

    def test_overlapping_frames_keep_their_own_origin(self):
        """A later worker must not become the previous-frame baseline or steal origin."""
        session, engine, _ = self.make_tagged_session()

        class Ordered:
            def __init__(self):
                self.inside = 0
                self.max_inside = 0
                self.calls = 0
                self._inner = threading.Lock()

            def register(self, image):
                with self._inner:
                    self.inside += 1
                    self.max_inside = max(self.max_inside, self.inside)
                    self.calls += 1
                    call = self.calls
                time.sleep(0.03)
                with self._inner:
                    self.inside -= 1
                return (0.0, -400.0)

        session._shift_map = Ordered()
        errors = []

        def run(frame_id):
            try:
                session.process_frame(frame_id, frame_bytes())
            except Exception as exc:
                errors.append(exc)

        early = threading.Thread(target=run, args=(1,))
        late = threading.Thread(target=run, args=(2,))
        early.start()
        late.start()
        early.join(2.0)
        late.join(2.0)
        self.assertEqual(errors, [])
        self.assertFalse(early.is_alive() or late.is_alive())
        self.assertEqual(session._shift_map.calls, 2)
        self.assertEqual(session._shift_map.max_inside, 1)
        self.assertEqual(session._frame_origin[1], 800.0)
        engine.shutdown()

    def test_cached_tag_projects_at_this_frame_origin(self):
        """A panned frame must not attach a tag that only overlaps the previous origin."""
        from tags import TagRecord

        session, engine, _ = self.make_tagged_session()
        record = TagRecord(
            tag_id="cached", box=[150, 700, 190, 760], band_index=0, name="cached", price=150,
            flags=["tax_consistent"],
        )
        record.settled = True
        session._tags._tags[record.tag_id] = record

        class Panned:
            def __init__(self):
                self.calls = 0

            def register(self, image):
                self.calls += 1
                return (0.0, -400.0) if self.calls == 2 else (0.0, 0.0)

        session._shift_map = Panned()
        first = session.process_frame(1, frame_bytes())
        self.assertFalse(any(track.get("price_tag", {}).get("id") for track in first["tracks"]))
        second = session.process_frame(2, frame_bytes())
        bound = [track["price_tag"] for track in second["tracks"] if track.get("price_tag", {}).get("id")]
        self.assertTrue(bound)
        self.assertGreaterEqual(bound[0]["box"][1], 200)
        self.assertLess(bound[0]["box"][1], 480)
        engine.shutdown()


class FramePipelineTests(unittest.TestCase):
    def test_latest_frame_wins_and_drops_are_counted(self):
        session, engine, _, _ = make_session()
        session.submit_frame(1, frame_bytes())
        session.submit_frame(2, frame_bytes())
        session.submit_frame(3, frame_bytes())
        pending = session.take_pending(timeout=1.0)
        self.assertEqual(pending[0], 3, "the newest frame must be the one processed")
        stats = session.snapshot()["stats"]
        self.assertEqual(stats["frames_received"], 3)
        self.assertEqual(stats["frames_dropped"], 2)
        engine.shutdown()

    def test_undecodable_frame_raises(self):
        session, engine, _, _ = make_session()
        with self.assertRaises(Exception):
            session.process_frame(1, b"not an image")
        engine.shutdown()

    def test_result_payload_shape(self):
        session, engine, _, _ = make_session()
        result = process(session, 1)
        for key in ("type", "frame_id", "seq", "width", "height", "tracks", "empty_slots",
                    "quality", "summary", "timing_ms", "stats"):
            self.assertIn(key, result)
        self.assertEqual(result["type"], "result")
        self.assertIn("unconfirmed", result["summary"])
        track = result["tracks"][0]
        for key in ("track_id", "box", "raw_box", "confidence", "first_seen_seq", "relabels",
                    "restarts", "identity"):
            self.assertIn(key, track)
        engine.shutdown()

    def test_quality_hint_reports_glare(self):
        session, engine, _, _ = make_session()
        ok, buffer = cv2.imencode(".jpg", np.full((480, 640, 3), 255, dtype="uint8"))
        session.submit_frame(1, buffer.tobytes())
        result = session.process_frame(*session.take_pending(timeout=1.0))
        self.assertTrue(result["quality"]["glare"] > 0.5)
        self.assertEqual(result["quality"]["hint"], "glare")
        engine.shutdown()


class ClassificationBudgetTests(unittest.TestCase):
    def test_per_session_inflight_is_capped(self):
        matcher = FakeMatcher(delay=0.3)
        session, engine, detector, _ = make_session(
            matcher, LiveSettings(method="hybrid", max_inflight_per_session=2)
        )
        detector.boxes = [
            {"id": index, "box": [10 + index * 120, 10, 110 + index * 120, 200],
             "confidence": 0.9, "class_id": 0, "width": 100, "height": 190,
             "center": [60 + index * 120, 105]}
            for index in range(6)
        ]
        process(session, 1)
        process(session, 2)
        self.assertLessEqual(matcher.calls, 2, "only the cap may be in flight at once")
        engine.shutdown()

    def test_method_change_clears_labels_and_ignores_late_results(self):
        session, engine, _, matcher = make_session(FakeMatcher(delay=0.25))
        process(session, 1)
        process(session, 2)
        session.configure(method="orb")
        self.assertEqual(session.config()["method"], "orb")
        result = process(session, 3)
        for track in result["tracks"]:
            self.assertNotEqual(track["identity"]["state"], "recognized",
                                "a late result from the old method must not label a track")
        engine.shutdown()

    def test_inspect_reports_track_not_in_view(self):
        session, engine, _, _ = make_session()
        process(session, 1)
        with self.assertRaises(KeyError):
            session.inspect(999)


    def test_close_stops_the_loop(self):
        session, engine, _, _ = make_session()
        session.close()
        self.assertTrue(session.closed)
        self.assertIsNone(session.take_pending(timeout=0.05))

    def test_async_take_keeps_latest_and_wakes_on_close(self):
        """The server waiter must not miss a frame that beat registration, and close must wake it."""
        session, engine, _, _ = make_session()

        async def scenario():
            session.submit_frame(1, frame_bytes())
            session.submit_frame(2, frame_bytes())
            first = await asyncio.wait_for(session.take_pending_async(), timeout=1.0)
            self.assertEqual(first[0], 2)

            waiting = asyncio.create_task(session.take_pending_async())
            await asyncio.sleep(0)
            self.assertFalse(waiting.done(), "an empty queue must not complete the waiter")
            session.reset()
            session.submit_frame(3, frame_bytes())
            self.assertEqual((await asyncio.wait_for(waiting, timeout=1.0))[0], 3)

            cancelled = asyncio.create_task(session.take_pending_async())
            await asyncio.sleep(0)
            cancelled.cancel()
            with self.assertRaises(asyncio.CancelledError):
                await cancelled
            session.submit_frame(4, frame_bytes())
            self.assertEqual((await asyncio.wait_for(session.take_pending_async(), timeout=1.0))[0], 4)

            closing = asyncio.create_task(session.take_pending_async())
            await asyncio.sleep(0)
            session.close()
            self.assertIsNone(await asyncio.wait_for(closing, timeout=1.0))
            self.assertFalse(session.submit_frame(5, frame_bytes()))

        asyncio.run(scenario())
        engine.shutdown()



class SmallBoxDetector(FakeDetector):
    """A facing too small for the frame crop to decide, so only a crop can."""

    def __init__(self, box=None):
        super().__init__()
        self.boxes = [{
            "id": 0, "box": list(box or [100, 100, 130, 200]), "confidence": 0.9,
            "class_id": 0, "width": 30, "height": 100, "center": [115, 150],
        }]


def make_crop_session(matcher=None, settings=None, detector=None):
    detector = detector or SmallBoxDetector()
    engine = RecognitionEngine(
        detector, matcher or FakeMatcher(), FakeInventory(),
        settings=__import__("engine").EngineSettings(
            classify_workers=2, identity_backend="matcher"),
    )
    session = LiveSession("crop", engine, settings=settings or LiveSettings(method="hybrid"))
    session.refresh_catalog()
    session.enable_crops()
    return session, engine, detector


def requested_crop(result):
    for track in result["tracks"]:
        if "crop_request" in track:
            return track["crop_request"]
    return None


def panned(detector, x):
    """Put the fake facing at x, the way a camera panning a shelf does.

    The facing stays too narrow for the frame path, so only the crop lane can
    decide it, and it stays under crop_below_px so a crop is asked about.
    """
    detector.boxes[0].update({
        "box": [x, 100, x + 36, 220], "width": 36, "height": 120, "center": [x + 18, 160],
    })


class GatedMatcher(FakeMatcher):
    """Holds every classification open until the test lets it finish."""

    def __init__(self):
        super().__init__()
        self.gate = threading.Event()

    def score_crop(self, crop, method="hybrid"):
        self.gate.wait(2.0)
        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."""

    def test_a_protocol_v1_client_is_never_asked(self):
        session, engine, _ = make_crop_session()
        session.enable_crops  # noqa: B018 - documents that crops start disabled
        session.settings.crops = False
        process(session, 1)
        result = process(session, 2)
        self.assertIsNone(requested_crop(result))
        self.assertEqual(result["stats"]["crops"]["requested"], 0)
        engine.shutdown()

    def test_accepted_crop_labels_the_track_it_was_asked_about(self):
        session, engine, _ = make_crop_session()
        process(session, 1)
        request = requested_crop(process(session, 2))
        self.assertIsNotNone(request, "a box the frame crop cannot use must be asked about")
        self.assertEqual(request["frame_id"], 2)

        reply = session.submit_crop(request["request_id"], request["track_id"],
                                    request["frame_id"], frame_bytes())
        self.assertEqual(reply["type"], "crop_ack")
        self.assertTrue(wait_for(lambda: process(session, 3)["stats"]["crops"]["accepted"] == 1))
        engine.shutdown()

    def test_the_label_says_it_was_measured_on_a_crop(self):
        session, engine, _ = make_crop_session()
        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: "crop_request" not in process(session, 3)["tracks"][0]))
        labelled = [t for t in process(session, 4)["tracks"] if t["identity"]["state"] == "recognized"]
        self.assertTrue(labelled, "the crop should settle the track the frame could not")
        self.assertEqual(labelled[0]["identity"]["sku_id"], "sku_a")
        self.assertEqual(labelled[0]["identity"]["source"], "crop")
        self.assertEqual(labelled[0]["identity"]["attempts"], 1)
        engine.shutdown()

    def test_a_crop_can_arrive_after_the_next_frame(self):
        session, engine, _ = make_crop_session()
        process(session, 1)
        request = requested_crop(process(session, 2))
        process(session, 3)
        reply = session.submit_crop(request["request_id"], request["track_id"],
                                    request["frame_id"], frame_bytes())
        self.assertEqual(reply["type"], "crop_ack")
        self.assertTrue(wait_for(lambda: session._crops_accepted == 1))
        engine.shutdown()

    def test_a_delayed_crop_is_accepted_while_the_track_stays_on_its_product(self):
        """Movement alone is not staleness: the crop still shows the same product."""
        session, engine, detector = make_crop_session()
        panned(detector, 100)
        process(session, 1)
        panned(detector, 106)
        request = requested_crop(process(session, 2))
        self.assertIsNotNone(request, "a facing the frame cannot use is asked about")
        # The phone answers a frame late, on a track that has panned a little.
        panned(detector, 112)
        moved = process(session, 3)
        self.assertEqual(moved["tracks"][0]["track_id"], request["track_id"])
        reply = session.submit_crop(request["request_id"], request["track_id"],
                                    request["frame_id"], frame_bytes())
        self.assertEqual(reply["type"], "crop_ack")
        self.assertTrue(wait_for(lambda: process(session, 4)["stats"]["crops"]["accepted"] == 1))
        labelled = [t for t in process(session, 5)["tracks"] if t["identity"]["state"] == "recognized"]
        self.assertEqual(labelled[0]["identity"]["sku_id"], "sku_a")
        engine.shutdown()

    def test_a_crop_for_a_track_that_slid_onto_its_neighbour_is_refused(self):
        """The crop photographs one product; a track that walked off it answers nothing."""
        session, engine, detector = make_crop_session()
        panned(detector, 100)
        process(session, 1)
        panned(detector, 118)
        request = requested_crop(process(session, 2))
        self.assertIsNotNone(request)
        # The pan carries the track onto the next product before the phone answers.
        panned(detector, 136)
        moved = process(session, 3)
        self.assertEqual(moved["tracks"][0]["track_id"], request["track_id"],
                         "the id must survive the pan, or this refusal proves nothing")
        reply = session.submit_crop(request["request_id"], request["track_id"],
                                    request["frame_id"], frame_bytes())
        self.assertEqual(reply["code"], "CROP_STALE")
        self.assertEqual(session._tracks[request["track_id"]].in_flight, 0,
                         "a refused crop gives the track's busy slot back")
        result = process(session, 4)
        self.assertEqual(result["stats"]["crops"]["accepted"], 0)
        self.assertEqual(result["stats"]["crops"]["rejected"], 1)
        self.assertNotEqual(result["tracks"][0]["identity"]["state"], "recognized")
        engine.shutdown()

    def test_a_crop_answered_after_the_track_left_its_product_does_not_label_it(self):
        """A classification in flight cannot settle a track that has since moved on."""
        matcher = GatedMatcher()
        session, engine, detector = make_crop_session(matcher=matcher)
        panned(detector, 100)
        process(session, 1)
        panned(detector, 118)
        request = requested_crop(process(session, 2))
        self.assertIsNotNone(request)
        reply = session.submit_crop(request["request_id"], request["track_id"],
                                    request["frame_id"], frame_bytes())
        self.assertEqual(reply["type"], "crop_ack", "the track is still on the product here")
        # The answer is still in the classifier when the pan takes the track off it.
        panned(detector, 136)
        process(session, 3)
        matcher.gate.set()
        track = session._tracks[request["track_id"]]
        self.assertTrue(wait_for(lambda: session._tracks[request["track_id"]].in_flight == 0))
        self.assertEqual(track.attempts, 0, "a crop of a product left behind is not an attempt")
        self.assertIsNone(track.identity, "the neighbour must not wear the old product's label")
        engine.shutdown()

    def test_a_track_is_asked_once(self):
        session, engine, _ = make_crop_session()
        process(session, 1)
        process(session, 2)
        result = process(session, 3)
        self.assertIsNone(requested_crop(result), "one crop per track object")
        self.assertEqual(result["stats"]["crops"]["requested"], 1)
        engine.shutdown()

    def test_a_crop_for_another_frame_is_refused(self):
        session, engine, _ = make_crop_session()
        process(session, 1)
        request = requested_crop(process(session, 2))
        reply = session.submit_crop(request["request_id"], request["track_id"], 999, frame_bytes())
        self.assertEqual(reply["code"], "CROP_MISMATCH")
        result = process(session, 3)
        self.assertNotEqual(result["tracks"][0]["identity"]["state"], "recognized")
        self.assertEqual(result["stats"]["crops"]["rejected"], 1)
        self.assertEqual(result["stats"]["crops"]["accepted"], 0)
        engine.shutdown()

    def test_a_crop_for_another_track_is_refused(self):
        session, engine, _ = make_crop_session()
        process(session, 1)
        request = requested_crop(process(session, 2))
        reply = session.submit_crop(request["request_id"], request["track_id"] + 41,
                                    request["frame_id"], frame_bytes())
        self.assertEqual(reply["code"], "CROP_MISMATCH")
        self.assertEqual(process(session, 3)["stats"]["crops"]["accepted"], 0)
        engine.shutdown()

    def test_an_oversized_crop_is_refused(self):
        session, engine, _ = make_crop_session(settings=LiveSettings(max_crop_bytes=2048))
        session.enable_crops(2048)
        process(session, 1)
        request = requested_crop(process(session, 2))
        reply = session.submit_crop(request["request_id"], request["track_id"],
                                    request["frame_id"], b"\xff\xd8" + b"\x00" * 4096)
        self.assertEqual(reply["code"], "CROP_TOO_LARGE")
        self.assertEqual(process(session, 3)["stats"]["crops"]["accepted"], 0)
        engine.shutdown()

    def test_an_undecodable_crop_is_refused(self):
        session, engine, _ = make_crop_session()
        process(session, 1)
        request = requested_crop(process(session, 2))
        reply = session.submit_crop(request["request_id"], request["track_id"],
                                    request["frame_id"], b"not an image")
        self.assertEqual(reply["code"], "CROP_UNDECODABLE")
        self.assertEqual(process(session, 3)["stats"]["crops"]["accepted"], 0)
        engine.shutdown()

    def test_a_crop_cannot_be_replayed(self):
        session, engine, _ = make_crop_session()
        process(session, 1)
        request = requested_crop(process(session, 2))
        first = session.submit_crop(request["request_id"], request["track_id"],
                                    request["frame_id"], frame_bytes())
        self.assertEqual(first["type"], "crop_ack")
        again = session.submit_crop(request["request_id"], request["track_id"],
                                    request["frame_id"], frame_bytes())
        self.assertEqual(again["code"], "CROP_UNKNOWN_REQUEST")
        engine.shutdown()

    def test_an_unknown_request_is_refused(self):
        session, engine, _ = make_crop_session()
        process(session, 1)
        reply = session.submit_crop("c-nope", 1, 1, frame_bytes())
        self.assertEqual(reply["code"], "CROP_UNKNOWN_REQUEST")
        engine.shutdown()

    def test_an_unanswered_request_expires_and_frees_the_track(self):
        session, engine, _ = make_crop_session(settings=LiveSettings(crop_ttl_frames=1))
        process(session, 1)
        request = requested_crop(process(session, 2))
        process(session, 3)
        # The request was made on frame 2 with a one-frame budget, so by frame 4
        # it is out of date and the track it held is free again.
        result = process(session, 4)
        self.assertEqual(result["stats"]["crops"]["rejected"], 1, "an unanswered request is counted")
        late = session.submit_crop(request["request_id"], request["track_id"],
                                   request["frame_id"], frame_bytes())
        self.assertEqual(late["code"], "CROP_UNKNOWN_REQUEST")
        engine.shutdown()

    def test_the_client_states_the_smaller_crop_limit(self):
        session, engine, _ = make_crop_session()
        self.assertEqual(session.enable_crops(4096)["max_crop_bytes"], 4096)
        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()