test: defend the live session and the server API

test_live: a product is unconfirmed, then labelled, without blocking frames; a
weak match never names a SKU and ends as needs_confirmation; a restarted id is
re-verified rather than inheriting its old label; inherited labels are flagged;
latest-frame-wins accounting; payload shape; glare hint; per-session
classification cap; a method change clears labels and drops late results.

test_api: health, catalog shape (no local paths), template-image 404s including
traversal-looking ids, stock validation, recognize over raw bytes and multipart,
415/422/400 with one error shape, classify/inspect roundtrip, IMAGE_EXPIRED,
token auth on REST and the socket, the WebSocket protocol (hello, frames,
config/reset/ping, recoverable BAD_FRAME and UNDECODABLE_IMAGE, inspect and
TRACK_NOT_IN_VIEW), the session limit, and the legacy session route.

Server fixes the tests forced out: FastAPI validation errors now use the same
{"error": {...}} envelope as every other failure, and the legacy session routes
accept a missing JSON body instead of returning 500.

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

recognization/code/app.py

@@ -51,6 +51,7 @@ from fastapi import (  # noqa: E402
     WebSocket,
     WebSocketDisconnect,
 )
+from fastapi.exceptions import RequestValidationError  # noqa: E402
 from fastapi.middleware.cors import CORSMiddleware  # noqa: E402
 from fastapi.responses import FileResponse, JSONResponse, Response  # noqa: E402
 from fastapi.staticfiles import StaticFiles  # noqa: E402
@@ -562,7 +563,8 @@ def create_app(
 
     # ---------------------------------------------------------------- legacy session routes
     @app.post("/api/sessions")
-    async def legacy_start_session(payload: Dict[str, Any]) -> Dict[str, Any]:
+    async def legacy_start_session(payload: Dict[str, Any] = None) -> Dict[str, Any]:
+        payload = payload or {}
         session = scans.start_session(
             zone_id=str(payload.get("zone_id") or "zone-1"),
             catalog_version=str(payload.get("catalog_version") or "stock-v1"),
@@ -571,9 +573,10 @@ def create_app(
         return {"status": "success", "session": session.snapshot()}
 
     @app.post("/api/sessions/{session_id}/frames")
-    async def legacy_observe_frame(session_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+    async def legacy_observe_frame(session_id: str, payload: Dict[str, Any] = None) -> Dict[str, Any]:
         from observe import FrameRequest, ObserveError
 
+        payload = payload or {}
         try:
             session = scans.get_session(session_id)
             raw = payload.get("frame_bytes_b64") or payload.get("image_base64") or ""
@@ -608,9 +611,11 @@ def create_app(
             raise _error("NOT_FOUND", str(exc), 404)
 
     @app.post("/api/sessions/{session_id}/review")
-    async def legacy_review(session_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+    async def legacy_review(session_id: str, payload: Dict[str, Any] = None) -> Dict[str, Any]:
         from observe import ObserveError
 
+        payload = payload or {}
+
         try:
             snapshot = scans.review(
                 session_id,
@@ -650,9 +655,10 @@ def create_app(
             raise _error("CONFLICT", str(exc), 409)
 
     @app.post("/api/compare_live_vs_still")
-    async def legacy_compare(payload: Dict[str, Any]) -> Dict[str, Any]:
+    async def legacy_compare(payload: Dict[str, Any] = None) -> Dict[str, Any]:
         from observe import ObserveError
 
+        payload = payload or {}
         try:
             comparison = scans.compare_live_vs_still(
                 str(payload.get("sampled_session_id")), str(payload.get("still_session_id"))
@@ -662,6 +668,16 @@ def create_app(
             raise _error("CONFLICT", str(exc), 409)
 
     # ---------------------------------------------------------------- errors
+    @app.exception_handler(RequestValidationError)
+    async def validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
+        first = (exc.errors() or [{}])[0]
+        location = ".".join(str(part) for part in first.get("loc", []) if part != "body")
+        message = first.get("msg", "invalid request")
+        detail = f"{location}: {message}" if location else message
+        return JSONResponse(
+            status_code=422, content={"error": {"code": "VALIDATION_ERROR", "message": detail}}
+        )
+
     @app.exception_handler(HTTPException)
     async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
         detail = exc.detail

recognization/code/test_api.py

@@ -0,0 +1,333 @@
+"""
+Contract tests for the recognition server API.
+
+A fake engine stands in for the model, so these run without weights. They defend
+the wire contracts a client depends on: response shapes, the shared error shape,
+auth, limits, and the WebSocket protocol (hello, frames, control messages,
+recoverable errors).
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import os
+import sys
+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 app import ServerSettings, create_app  # noqa: E402
+from engine import EngineSettings, RecognitionEngine  # noqa: E402
+from test_live import FakeDetector, FakeInventory, FakeMatcher  # noqa: E402
+
+from fastapi.testclient import TestClient  # noqa: E402
+
+
+def make_scans(engine):
+    """Legacy session model wired to the fake engine (the real factory is exercised
+    against the real model in test_real_model)."""
+    from observe import PerceptionAdapter, PerceptionService, ScanService
+
+    def detect(frame_bytes):
+        image = engine.decode_image(frame_bytes)
+        boxes = engine.detect(image)
+        return {"width": image.shape[1], "height": image.shape[0], "boxes": boxes,
+                "empty_slots": [], "quality": {"decoded": True}}
+
+    def classify(box):
+        scores = engine.score_crop(engine.crop(engine._last_image, box["box"]))
+        return scores
+
+    # keep the frame for the classify callback
+    def detect_with_frame(frame_bytes):
+        image = engine.decode_image(frame_bytes)
+        engine._last_image = image
+        boxes = engine.detect(image)
+        return {"width": image.shape[1], "height": image.shape[0], "boxes": boxes,
+                "empty_slots": [], "quality": {"decoded": True}}
+
+    adapter = PerceptionAdapter(detect_fn=detect_with_frame, classify_fn=classify,
+                                model_version="fake", matcher_method="hybrid")
+    return ScanService(PerceptionService(adapter))
+
+
+def make_app(*, token=None, max_sessions=4):
+    engine = RecognitionEngine(
+        FakeDetector(), FakeMatcher(), FakeInventory(),
+        settings=EngineSettings(classify_workers=2),
+    )
+    settings = ServerSettings(token=token, max_sessions=max_sessions)
+    app = create_app(engine, settings, web_dir="/nonexistent", scans=make_scans(engine))
+    return TestClient(app), engine
+
+
+def jpeg_bytes(width=640, height=480):
+    ok, buffer = cv2.imencode(".jpg", np.zeros((height, width, 3), dtype="uint8"))
+    return buffer.tobytes()
+
+
+def frame_payload(frame_id, payload=None):
+    payload = payload if payload is not None else jpeg_bytes()
+    return frame_id.to_bytes(4, "big") + payload
+
+
+class RestContractTests(unittest.TestCase):
+    def setUp(self):
+        self.client, self.engine = make_app()
+
+    def tearDown(self):
+        self.engine.shutdown()
+
+    def test_health_needs_no_token_and_reports_model(self):
+        response = self.client.get("/v1/health")
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertEqual(body["status"], "ok")
+        self.assertIn("version", body["model"])
+        self.assertEqual(body["catalog"]["sku_count"], 2)
+
+    def test_catalog_shape_has_no_local_paths(self):
+        body = self.client.get("/v1/catalog").json()
+        product = body["products"][0]
+        self.assertEqual(set(product), {
+            "sku_id", "product_name", "price", "number_in", "number_out", "number_on",
+            "receipt_date", "expiration_date", "image_url",
+        })
+
+    def test_catalog_image_404_for_unknown_and_traversal(self):
+        response = self.client.get("/v1/catalog/nope/image")
+        self.assertEqual(response.status_code, 404)
+        self.assertEqual(response.json()["error"]["code"], "NOT_FOUND")
+        # A traversal-looking id is rejected before it can reach the filesystem
+        # (either by routing or by the inventory lookup).
+        for sku in ("..%2F..%2Fstock.csv", "../../etc/passwd", "sku_a%2F..%2F..%2Fstock.csv"):
+            response = self.client.get(f"/v1/catalog/{sku}/image")
+            self.assertIn(response.status_code, (400, 404), sku)
+
+    def test_stock_update_validates_and_returns_product(self):
+        before = self.client.get("/v1/catalog").json()["products"]
+        before_on = next(p["number_on"] for p in before if p["sku_id"] == "sku_a")
+        response = self.client.post("/v1/catalog/sku_a/stock", json={"delta": -1})
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()["product"]["number_on"], before_on - 1)
+        self.assertEqual(self.client.post("/v1/catalog/sku_a/stock", json={"delta": 0}).status_code, 422)
+        self.assertEqual(self.client.post("/v1/catalog/sku_a/stock", json={"delta": 500}).status_code, 422)
+        self.assertEqual(self.client.post("/v1/catalog/ghost/stock", json={"delta": -1}).status_code, 404)
+
+    def test_recognize_accepts_raw_bytes(self):
+        response = self.client.post(
+            "/v1/recognize?classify=true", content=jpeg_bytes(),
+            headers={"Content-Type": "image/jpeg"},
+        )
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertTrue(body["image_id"].startswith("img_"))
+        self.assertEqual(body["coordinate_convention"], "pixel-xyxy-origin-top-left")
+        self.assertEqual(len(body["detections"]), 1)
+        self.assertIn(body["detections"][0]["identity"]["state"], ("recognized", "unconfirmed"))
+        for key in ("width", "height", "catalog_version", "model_version", "summary", "timing_ms"):
+            self.assertIn(key, body)
+
+    def test_recognize_accepts_multipart(self):
+        response = self.client.post(
+            "/v1/recognize?classify=false",
+            files={"file": ("shelf.jpg", jpeg_bytes(), "image/jpeg")},
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()["detections"][0]["identity"]["state"], "detected")
+
+    def test_recognize_rejects_bad_input(self):
+        self.assertEqual(self.client.post("/v1/recognize", content=b"").status_code, 400)
+        response = self.client.post(
+            "/v1/recognize", content=b"not an image", headers={"Content-Type": "image/jpeg"}
+        )
+        self.assertEqual(response.status_code, 422)
+        self.assertEqual(response.json()["error"]["code"], "UNDECODABLE_IMAGE")
+        response = self.client.post(
+            "/v1/recognize", content=b"{}", headers={"Content-Type": "application/json"}
+        )
+        self.assertEqual(response.status_code, 415)
+        self.assertEqual(response.json()["error"]["code"], "UNSUPPORTED_MEDIA_TYPE")
+
+    def test_classify_and_inspect_roundtrip(self):
+        recognized = self.client.post(
+            "/v1/recognize?classify=false", content=jpeg_bytes(),
+            headers={"Content-Type": "image/jpeg"},
+        ).json()
+        image_id = recognized["image_id"]
+        box = recognized["detections"][0]["box"]
+
+        classified = self.client.post(
+            f"/v1/images/{image_id}/classify", json={"method": "hybrid"}
+        )
+        self.assertEqual(classified.status_code, 200)
+        self.assertEqual(len(classified.json()["identities"]), 1)
+
+        inspected = self.client.post(
+            f"/v1/images/{image_id}/inspect", json={"box": box, "method": "hybrid"}
+        )
+        self.assertEqual(inspected.status_code, 200)
+        body = inspected.json()
+        self.assertIn("identity", body)
+        self.assertTrue(body["crop_jpeg_b64"].startswith("data:image/jpeg;base64,"))
+
+    def test_unknown_method_is_a_400(self):
+        response = self.client.post(
+            "/v1/recognize?method=nope", content=jpeg_bytes(),
+            headers={"Content-Type": "image/jpeg"},
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(response.json()["error"]["code"], "BAD_REQUEST")
+
+    def test_expired_image_id(self):
+        response = self.client.post("/v1/images/img_missing/classify", json={"method": "hybrid"})
+        self.assertEqual(response.status_code, 404)
+        self.assertEqual(response.json()["error"]["code"], "IMAGE_EXPIRED")
+
+    def test_validation_errors_use_the_shared_shape(self):
+        response = self.client.post("/v1/images/img_x/inspect", json={"box": [1, 2, 3]})
+        self.assertEqual(response.status_code, 422)
+        self.assertIn("error", response.json())
+
+
+class AuthTests(unittest.TestCase):
+    def test_token_is_required_when_configured(self):
+        client, engine = make_app(token="s3cret")
+        try:
+            self.assertEqual(client.get("/v1/health").status_code, 200)
+            self.assertEqual(client.get("/v1/catalog").status_code, 401)
+            self.assertEqual(client.get("/v1/catalog", headers={"Authorization": "Bearer wrong"}).status_code, 401)
+            self.assertEqual(client.get("/v1/catalog", headers={"Authorization": "Bearer s3cret"}).status_code, 200)
+        finally:
+            engine.shutdown()
+
+
+class WebSocketProtocolTests(unittest.TestCase):
+    def setUp(self):
+        self.client, self.engine = make_app()
+
+    def tearDown(self):
+        self.engine.shutdown()
+
+    def test_hello_frame_result_and_control_messages(self):
+        with self.client.websocket_connect("/v1/stream") as socket:
+            hello = socket.receive_json()
+            self.assertEqual(hello["type"], "hello")
+            self.assertEqual(hello["protocol"], 1)
+            self.assertIn("session_id", hello)
+            self.assertIn("config", hello)
+            self.assertEqual(hello["limits"]["max_frame_bytes"], 4 * 1024 * 1024)
+
+            socket.send_json({"type": "ping", "t": 7})
+            self.assertEqual(socket.receive_json(), {"type": "pong", "t": 7})
+
+            socket.send_json({"type": "config", "method": "orb"})
+            ack = socket.receive_json()
+            self.assertEqual(ack["type"], "config_ack")
+            self.assertEqual(ack["config"]["method"], "orb")
+
+            socket.send_bytes(frame_payload(11))
+            result = socket.receive_json()
+            self.assertEqual(result["type"], "result")
+            self.assertEqual(result["frame_id"], 11)
+            self.assertEqual(result["seq"], 1)
+            self.assertIn("tracks", result)
+            self.assertIn("summary", result)
+            self.assertIn("quality", result)
+
+            socket.send_json({"type": "reset"})
+            self.assertEqual(socket.receive_json()["type"], "reset_ack")
+
+            socket.send_json({"type": "not-a-thing"})
+            error = socket.receive_json()
+            self.assertEqual(error["code"], "BAD_MESSAGE")
+
+    def test_short_frame_is_recoverable(self):
+        with self.client.websocket_connect("/v1/stream") as socket:
+            socket.receive_json()  # hello
+            socket.send_bytes(b"\x00\x01")
+            error = socket.receive_json()
+            self.assertEqual(error["code"], "BAD_FRAME")
+            # the connection is still usable
+            socket.send_bytes(frame_payload(1))
+            self.assertEqual(socket.receive_json()["type"], "result")
+
+    def test_undecodable_frame_is_reported_and_keeps_the_socket(self):
+        with self.client.websocket_connect("/v1/stream") as socket:
+            socket.receive_json()
+            socket.send_bytes(frame_payload(5, b"not an image"))
+            error = socket.receive_json()
+            self.assertEqual(error["code"], "UNDECODABLE_IMAGE")
+            self.assertEqual(error["frame_id"], 5)
+            socket.send_bytes(frame_payload(6))
+            self.assertEqual(socket.receive_json()["frame_id"], 6)
+
+    def test_inspect_over_the_socket(self):
+        with self.client.websocket_connect("/v1/stream") as socket:
+            socket.receive_json()
+            socket.send_bytes(frame_payload(1))
+            result = socket.receive_json()
+            track_id = result["tracks"][0]["track_id"]
+            socket.send_json({"type": "inspect", "track_id": track_id, "request_id": "r1"})
+            reply = socket.receive_json()
+            self.assertEqual(reply["type"], "inspect_result")
+            self.assertEqual(reply["request_id"], "r1")
+            self.assertIn("identity", reply)
+
+            socket.send_json({"type": "inspect", "track_id": 999999, "request_id": "r2"})
+            missing = socket.receive_json()
+            self.assertEqual(missing["code"], "TRACK_NOT_IN_VIEW")
+
+    def test_token_is_enforced_on_the_socket(self):
+        client, engine = make_app(token="s3cret")
+        try:
+            with self.assertRaises(Exception):
+                with client.websocket_connect("/v1/stream") as socket:
+                    socket.receive_json()
+            with client.websocket_connect("/v1/stream?token=s3cret") as socket:
+                self.assertEqual(socket.receive_json()["type"], "hello")
+        finally:
+            engine.shutdown()
+
+    def test_session_limit_closes_extra_clients(self):
+        client, engine = make_app(max_sessions=1)
+        try:
+            with client.websocket_connect("/v1/stream") as first:
+                first.receive_json()
+                with self.assertRaises(Exception):
+                    with client.websocket_connect("/v1/stream") as second:
+                        second.receive_json()
+        finally:
+            engine.shutdown()
+
+
+class LegacyRouteTests(unittest.TestCase):
+    def test_legacy_sessions_still_respond(self):
+        """The earlier session model stays mounted; it is backed by the engine now."""
+        client, engine = make_app()
+        try:
+            started = client.post("/api/sessions", json={"zone_id": "z1", "catalog_version": "v1"})
+            self.assertEqual(started.status_code, 200)
+            session_id = started.json()["session"]["session_id"]
+            frames = client.post(
+                f"/api/sessions/{session_id}/frames",
+                json={
+                    "frame_id": "f1",
+                    "sequence": 1,
+                    "frame_bytes_b64": base64.b64encode(jpeg_bytes()).decode(),
+                },
+            )
+            self.assertEqual(frames.status_code, 200)
+            self.assertIn("observation", frames.json())
+        finally:
+            engine.shutdown()
+
+
+if __name__ == "__main__":
+    unittest.main()

recognization/code/test_live.py

@@ -0,0 +1,284 @@
+"""
+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 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 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 FakeInventory:
+    """Stateful stand-in for InventoryManager, so stock assertions are real."""
+
+    def __init__(self):
+        self.products = {
+            "sku_a": {"sku_id": "sku_a", "product_name": "Product sku_a", "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": "Product other", "price": 150.0,
+                      "number_in": 10, "number_out": 1, "number_on": 9,
+                      "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):
+    detector = FakeDetector()
+    matcher = matcher or FakeMatcher()
+    engine = RecognitionEngine(detector, matcher, FakeInventory(),
+                               settings=__import__("engine").EngineSettings(classify_workers=2))
+    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(session, frame_id=1):
+    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 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_label_is_never_shown_without_acceptance(self):
+        session, engine, _, _ = make_session(FakeMatcher(score=0.4))  # below the accept threshold
+        for frame_id in range(1, 8):
+            result = process(session, frame_id)
+            for track in result["tracks"]:
+                self.assertIsNone(track["identity"]["sku_id"],
+                                  "a weak match must never name a SKU")
+        engine.shutdown()
+
+    def test_weak_match_ends_as_needs_confirmation(self):
+        session, engine, _, _ = make_session(FakeMatcher(score=0.4))
+        for frame_id in range(1, 8):
+            process(session, frame_id)
+        result = process(session, 8)
+        self.assertTrue(wait_for(lambda: process(session, 9)["tracks"][0]["identity"]["state"]
+                                 == "needs_confirmation", timeout=3.0))
+        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()
+
+
+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)
+        engine.shutdown()
+
+    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))
+        engine.shutdown()
+
+
+if __name__ == "__main__":
+    unittest.main()