feat: identity decision contract and session-local observation model

Keeps the model's uncertainty out of the UI: decide_identity() separates raw
matcher scores into accepted / ambiguous / unknown / rejected, and only an
accepted result can name a SKU. Zero evidence is rejected, never a product.

observe.py adds frame-local observations and one-zone scan sessions (draft,
finalize, cancel, rescan) with replay, staleness and backpressure rules.
pipeline.process_click() now reports the identity level instead of failing
when the match is not accepted.

Incorporates earlier uncommitted session work in this repository
(identity.py, observe.py, yolo_adapter.py, test_observe.py, and the pipeline
hunk those needed); yolo_adapter keeps its decoded frame per thread so
concurrent sessions cannot swap images.

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

recognization/code/identity.py

@@ -0,0 +1,106 @@
+"""Identity ranking for catalog scores.
+
+Separates raw matcher scores from accept/unknown/ambiguous/reject so a
+zero-score crop cannot become a SKU. Thresholds are conservative defaults
+for the 21-SKU closed set; they are not calibrated accuracy claims.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Sequence, Tuple
+
+IDENTITY_ACCEPTED = "accepted"
+IDENTITY_AMBIGUOUS = "ambiguous"
+IDENTITY_UNKNOWN = "unknown"
+IDENTITY_REJECTED = "rejected"
+
+# ponytail: closed-set defaults. Recalibrate when held-out scores exist.
+DEFAULT_MIN_SCORE = 1.0
+DEFAULT_MIN_MARGIN = 0.15
+DEFAULT_MIN_CONFIDENCE = 0.55
+
+
+def relative_confidence(top_score: float, runner_up_score: float) -> float:
+    """Existing matcher formula: top / (top + 0.5 * runner)."""
+    return round(min(1.0, top_score / max(1.0, (top_score + runner_up_score * 0.5))), 4)
+
+
+def decide_identity(
+    scores: Dict[str, float],
+    *,
+    min_score: float = DEFAULT_MIN_SCORE,
+    min_margin: float = DEFAULT_MIN_MARGIN,
+    min_confidence: float = DEFAULT_MIN_CONFIDENCE,
+    details: Optional[Dict[str, Any]] = None,
+    method: str = "hybrid",
+) -> Dict[str, Any]:
+    """Rank catalog scores into accepted / ambiguous / unknown / rejected.
+
+    Zero or missing evidence is rejected. A weak but unique top score is
+    unknown. Two close scores are ambiguous. Only a unique, above-threshold
+    score is accepted.
+    """
+    if not scores:
+        return {
+            "status": IDENTITY_REJECTED,
+            "method": method,
+            "best_sku": None,
+            "score": 0.0,
+            "confidence": 0.0,
+            "identity_level": IDENTITY_REJECTED,
+            "ranked_candidates": [],
+            "details": {},
+            "message": "No templates available",
+        }
+
+    sorted_candidates: List[Tuple[str, float]] = sorted(
+        scores.items(), key=lambda x: x[1], reverse=True
+    )
+    top_sku, top_score = sorted_candidates[0]
+    runner_up_sku, runner_up_score = (
+        sorted_candidates[1] if len(sorted_candidates) > 1 else (None, 0.0)
+    )
+    confidence = relative_confidence(top_score, runner_up_score)
+    ranked = [{"sku_id": s, "score": round(sc, 2)} for s, sc in sorted_candidates[:5]]
+    top_details = (details or {}).get(top_sku, {})
+
+    if top_score <= 0.0:
+        level = IDENTITY_REJECTED
+        message = "No matching evidence"
+        best = None
+    elif (
+        runner_up_sku is not None
+        and top_score > 0.0
+        and (top_score - runner_up_score) < min_margin * max(top_score, 1.0)
+    ):
+        level = IDENTITY_AMBIGUOUS
+        message = f"Close scores: {top_sku} vs {runner_up_sku}"
+        best = None
+    elif top_score < min_score or confidence < min_confidence:
+        level = IDENTITY_UNKNOWN
+        message = "Top score below accept threshold"
+        best = None
+    else:
+        level = IDENTITY_ACCEPTED
+        message = "Accepted unique catalog match"
+        best = top_sku
+
+    return {
+        "status": "success" if level == IDENTITY_ACCEPTED else level,
+        "method": method,
+        "best_sku": best,
+        "score": round(top_score, 2),
+        "confidence": confidence,
+        "identity_level": level,
+        "ranked_candidates": ranked,
+        "details": top_details,
+        "message": message,
+    }
+
+
+def identity_from_ranked(
+    ranked_candidates: Sequence[Dict[str, Any]],
+    **kwargs: Any,
+) -> Dict[str, Any]:
+    scores = {str(row["sku_id"]): float(row["score"]) for row in ranked_candidates}
+    return decide_identity(scores, **kwargs)

recognization/code/observe.py

@@ -0,0 +1,474 @@
+"""Session-local perception and one-zone assisted scan.
+
+Production path: observe(frame, catalog_version) returns an immutable
+frame-local observation. Scan sessions own draft → finalized/cancelled
+state. Detector/matcher stay replaceable adapters; tests inject fakes.
+
+ponytail: in-memory sessions only. Durable store when a successful
+experiment needs one.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import threading
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+from identity import (
+    IDENTITY_ACCEPTED,
+    IDENTITY_AMBIGUOUS,
+    IDENTITY_REJECTED,
+    IDENTITY_UNKNOWN,
+    decide_identity,
+)
+
+COORDINATE_CONVENTION = "pixel-xyxy-origin-top-left"
+
+SESSION_DRAFT = "draft"
+SESSION_FINALIZED = "finalized"
+SESSION_CANCELLED = "cancelled"
+
+CAPTURE_STILL = "still"
+CAPTURE_SAMPLED = "sampled"
+
+DEFAULT_MAX_IN_FLIGHT = 1
+DEFAULT_MAX_QUEUED = 2
+DEFAULT_SESSION_TTL_S = 30 * 60
+
+
+class ObserveError(ValueError):
+    """Caller or session contract violation."""
+
+
+def content_fingerprint(frame_bytes: bytes) -> str:
+    return hashlib.sha256(frame_bytes).hexdigest()
+
+
+def _now() -> float:
+    return time.time()
+
+
+@dataclass
+class FrameRequest:
+    frame_id: str
+    frame_bytes: bytes
+    sequence: int
+    captured_at: float
+    session_id: str
+    zone_id: str
+    catalog_version: str
+    capture_kind: str = CAPTURE_SAMPLED
+    width: Optional[int] = None
+    height: Optional[int] = None
+    orientation: str = "upright"
+
+
+@dataclass
+class Observation:
+    frame_id: str
+    content_fingerprint: str
+    sequence: int
+    captured_at: float
+    observed_at: float
+    session_id: str
+    zone_id: str
+    catalog_version: str
+    capture_kind: str
+    width: int
+    height: int
+    orientation: str
+    coordinate_convention: str
+    boxes: List[Dict[str, Any]]
+    identities: List[Dict[str, Any]]
+    empty_slots: List[Dict[str, Any]]
+    model_version: str
+    matcher_method: str
+    feature_extractor: str
+    quality: Dict[str, Any]
+    selected: bool = False
+
+    def as_dict(self) -> Dict[str, Any]:
+        return {
+            "frame_id": self.frame_id,
+            "content_fingerprint": self.content_fingerprint,
+            "sequence": self.sequence,
+            "captured_at": self.captured_at,
+            "observed_at": self.observed_at,
+            "session_id": self.session_id,
+            "zone_id": self.zone_id,
+            "catalog_version": self.catalog_version,
+            "capture_kind": self.capture_kind,
+            "width": self.width,
+            "height": self.height,
+            "orientation": self.orientation,
+            "coordinate_convention": self.coordinate_convention,
+            "boxes": list(self.boxes),
+            "identities": list(self.identities),
+            "empty_slots": list(self.empty_slots),
+            "model_version": self.model_version,
+            "matcher_method": self.matcher_method,
+            "feature_extractor": self.feature_extractor,
+            "quality": dict(self.quality),
+            "selected": self.selected,
+        }
+
+
+class PerceptionAdapter:
+    """Replaceable detector + matcher. Default is a testable fake."""
+
+    def __init__(
+        self,
+        detect_fn: Optional[Callable[[bytes], Dict[str, Any]]] = None,
+        classify_fn: Optional[Callable[[Dict[str, Any]], Dict[str, float]]] = None,
+        model_version: str = "fake-detector",
+        matcher_method: str = "hybrid",
+        feature_extractor: str = "recorded",
+    ):
+        self._detect_fn = detect_fn
+        self._classify_fn = classify_fn
+        self.model_version = model_version
+        self.matcher_method = matcher_method
+        self.feature_extractor = feature_extractor
+
+    def detect(self, frame_bytes: bytes) -> Dict[str, Any]:
+        if self._detect_fn is not None:
+            return self._detect_fn(frame_bytes)
+        return {
+            "width": 1,
+            "height": 1,
+            "boxes": [],
+            "empty_slots": [],
+            "quality": {"coverage": 0.0, "limitation": "no-detector"},
+        }
+
+    def classify_box(self, box: Dict[str, Any]) -> Dict[str, float]:
+        if self._classify_fn is not None:
+            return self._classify_fn(box)
+        return {}
+
+
+class PerceptionService:
+    """Frame-local observe(). Shared adapter cache is allowed; results are not."""
+
+    def __init__(self, adapter: Optional[PerceptionAdapter] = None):
+        self.adapter = adapter or PerceptionAdapter()
+
+    def observe(self, request: FrameRequest, catalog_version: str) -> Observation:
+        if request.catalog_version != catalog_version:
+            raise ObserveError("catalog_version mismatch")
+        fingerprint = content_fingerprint(request.frame_bytes)
+        detected = self.adapter.detect(request.frame_bytes)
+        width = int(request.width or detected.get("width") or 0)
+        height = int(request.height or detected.get("height") or 0)
+        boxes = list(detected.get("boxes") or [])
+        empty_slots = list(detected.get("empty_slots") or [])
+        identities = []
+        for box in boxes:
+            scores = self.adapter.classify_box(box)
+            ranked = decide_identity(scores, method=self.adapter.matcher_method)
+            identities.append(
+                {
+                    "box_id": box.get("id"),
+                    "identity_level": ranked["identity_level"],
+                    "best_sku": ranked["best_sku"],
+                    "score": ranked["score"],
+                    "confidence": ranked["confidence"],
+                    "ranked_candidates": ranked["ranked_candidates"],
+                    "message": ranked["message"],
+                }
+            )
+        quality = dict(detected.get("quality") or {})
+        if not boxes:
+            quality.setdefault("limitation", "no-visible-objects")
+        return Observation(
+            frame_id=request.frame_id,
+            content_fingerprint=fingerprint,
+            sequence=request.sequence,
+            captured_at=request.captured_at,
+            observed_at=_now(),
+            session_id=request.session_id,
+            zone_id=request.zone_id,
+            catalog_version=catalog_version,
+            capture_kind=request.capture_kind,
+            width=width,
+            height=height,
+            orientation=request.orientation,
+            coordinate_convention=COORDINATE_CONVENTION,
+            boxes=boxes,
+            identities=identities,
+            empty_slots=empty_slots,
+            model_version=self.adapter.model_version,
+            matcher_method=self.adapter.matcher_method,
+            feature_extractor=self.adapter.feature_extractor,
+            quality=quality,
+        )
+
+
+@dataclass
+class ScanSession:
+    session_id: str
+    zone_id: str
+    catalog_version: str
+    capture_kind: str
+    created_at: float
+    expires_at: float
+    status: str = SESSION_DRAFT
+    selected_frame_id: Optional[str] = None
+    observations: Dict[str, Observation] = field(default_factory=dict)
+    in_flight: int = 0
+    queued: int = 0
+    dropped_stale: int = 0
+    review: Optional[Dict[str, Any]] = None
+    finalized_at: Optional[float] = None
+    cancelled_at: Optional[float] = None
+    revision: int = 0
+    lock: threading.Lock = field(default_factory=threading.Lock, repr=False, compare=False)
+
+    def snapshot(self) -> Dict[str, Any]:
+        selected = self.observations.get(self.selected_frame_id) if self.selected_frame_id else None
+        return {
+            "session_id": self.session_id,
+            "zone_id": self.zone_id,
+            "catalog_version": self.catalog_version,
+            "capture_kind": self.capture_kind,
+            "status": self.status,
+            "created_at": self.created_at,
+            "expires_at": self.expires_at,
+            "selected_frame_id": self.selected_frame_id,
+            "selected_observation": selected.as_dict() if selected else None,
+            "observation_count": len(self.observations),
+            "in_flight": self.in_flight,
+            "queued": self.queued,
+            "dropped_stale": self.dropped_stale,
+            "review": dict(self.review) if self.review else None,
+            "finalized_at": self.finalized_at,
+            "cancelled_at": self.cancelled_at,
+            "revision": self.revision,
+        }
+
+
+def _usable_score(observation: Observation) -> Tuple[float, float, int]:
+    """Prefer more accepted identities, then fewer unknowns, then newer sequence."""
+    accepted = sum(1 for i in observation.identities if i["identity_level"] == IDENTITY_ACCEPTED)
+    unknown = sum(
+        1
+        for i in observation.identities
+        if i["identity_level"] in (IDENTITY_UNKNOWN, IDENTITY_AMBIGUOUS, IDENTITY_REJECTED)
+    )
+    return (float(accepted), -float(unknown), int(observation.sequence))
+
+
+class ScanService:
+    """One-zone assisted scan: sampled frames, review, finalize, rescan."""
+
+    def __init__(
+        self,
+        perception: Optional[PerceptionService] = None,
+        *,
+        max_in_flight: int = DEFAULT_MAX_IN_FLIGHT,
+        max_queued: int = DEFAULT_MAX_QUEUED,
+        ttl_s: float = DEFAULT_SESSION_TTL_S,
+        clock: Callable[[], float] = _now,
+    ):
+        self.perception = perception or PerceptionService()
+        self.max_in_flight = max_in_flight
+        self.max_queued = max_queued
+        self.ttl_s = ttl_s
+        self._clock = clock
+        self._sessions: Dict[str, ScanSession] = {}
+        self._global_lock = threading.Lock()
+
+    def start_session(
+        self,
+        zone_id: str,
+        catalog_version: str,
+        *,
+        capture_kind: str = CAPTURE_SAMPLED,
+        session_id: Optional[str] = None,
+    ) -> ScanSession:
+        now = self._clock()
+        session = ScanSession(
+            session_id=session_id or str(uuid.uuid4()),
+            zone_id=zone_id,
+            catalog_version=catalog_version,
+            capture_kind=capture_kind,
+            created_at=now,
+            expires_at=now + self.ttl_s,
+        )
+        with self._global_lock:
+            self._sessions[session.session_id] = session
+        return session
+
+    def get_session(self, session_id: str) -> ScanSession:
+        with self._global_lock:
+            session = self._sessions.get(session_id)
+        if session is None:
+            raise ObserveError(f"unknown session {session_id}")
+        if self._clock() >= session.expires_at and session.status == SESSION_DRAFT:
+            raise ObserveError("session expired")
+        return session
+
+    def observe_frame(self, request: FrameRequest) -> Observation:
+        session = self.get_session(request.session_id)
+        if request.zone_id != session.zone_id:
+            raise ObserveError("zone mismatch")
+        if request.catalog_version != session.catalog_version:
+            raise ObserveError("catalog_version mismatch")
+        if session.status != SESSION_DRAFT:
+            raise ObserveError(f"session is {session.status}")
+
+        fingerprint = content_fingerprint(request.frame_bytes)
+        with session.lock:
+            existing = session.observations.get(request.frame_id)
+            if existing is not None:
+                if existing.content_fingerprint != fingerprint:
+                    raise ObserveError("frame_id reused with different content")
+                return existing
+            if session.in_flight >= self.max_in_flight:
+                if session.queued >= self.max_queued:
+                    session.dropped_stale += 1
+                    raise ObserveError("backpressure")
+                session.queued += 1
+                queued = True
+            else:
+                session.in_flight += 1
+                queued = False
+
+        try:
+            if queued:
+                with session.lock:
+                    session.queued -= 1
+                    if session.in_flight >= self.max_in_flight:
+                        session.dropped_stale += 1
+                        raise ObserveError("backpressure")
+                    session.in_flight += 1
+            observation = self.perception.observe(request, session.catalog_version)
+        finally:
+            with session.lock:
+                if session.in_flight > 0:
+                    session.in_flight -= 1
+
+        with session.lock:
+            if session.status != SESSION_DRAFT:
+                raise ObserveError(f"session is {session.status}")
+            replay = session.observations.get(request.frame_id)
+            if replay is not None:
+                if replay.content_fingerprint != fingerprint:
+                    raise ObserveError("frame_id reused with different content")
+                return replay
+            selected = session.observations.get(session.selected_frame_id) if session.selected_frame_id else None
+            if selected is not None and request.sequence < selected.sequence:
+                session.observations[request.frame_id] = observation
+                session.dropped_stale += 1
+                return observation
+            session.observations[request.frame_id] = observation
+            if selected is None or _usable_score(observation) >= _usable_score(selected):
+                if selected is not None:
+                    selected.selected = False
+                observation.selected = True
+                session.selected_frame_id = observation.frame_id
+            return observation
+
+    def review(
+        self,
+        session_id: str,
+        *,
+        actor: str,
+        reason: str,
+        note: Optional[str] = None,
+        previous_value: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        session = self.get_session(session_id)
+        with session.lock:
+            if session.status != SESSION_DRAFT:
+                raise ObserveError(f"session is {session.status}")
+            session.review = {
+                "actor": actor,
+                "reason": reason,
+                "note": note,
+                "previous_value": previous_value,
+                "reviewed_at": self._clock(),
+            }
+            session.revision += 1
+            return session.snapshot()
+
+    def finalize(self, session_id: str) -> Dict[str, Any]:
+        session = self.get_session(session_id)
+        with session.lock:
+            if session.status != SESSION_DRAFT:
+                raise ObserveError(f"session is {session.status}")
+            session.status = SESSION_FINALIZED
+            session.finalized_at = self._clock()
+            return session.snapshot()
+
+    def cancel(self, session_id: str) -> Dict[str, Any]:
+        session = self.get_session(session_id)
+        with session.lock:
+            if session.status == SESSION_FINALIZED:
+                raise ObserveError("finalized sessions are immutable")
+            session.status = SESSION_CANCELLED
+            session.cancelled_at = self._clock()
+            return session.snapshot()
+
+    def rescan(self, session_id: str) -> ScanSession:
+        parent = self.get_session(session_id)
+        if parent.status == SESSION_DRAFT:
+            raise ObserveError("cancel or finalize before rescan")
+        child = self.start_session(
+            parent.zone_id,
+            parent.catalog_version,
+            capture_kind=parent.capture_kind,
+        )
+        child.revision = parent.revision + 1
+        child.review = {
+            "actor": "system",
+            "reason": "rescan",
+            "previous_session_id": parent.session_id,
+            "reviewed_at": self._clock(),
+        }
+        return child
+
+    def compare_live_vs_still(
+        self, sampled_session_id: str, still_session_id: str
+    ) -> Dict[str, Any]:
+        sampled = self.get_session(sampled_session_id)
+        still = self.get_session(still_session_id)
+        sampled_obs = sampled.observations.get(sampled.selected_frame_id) if sampled.selected_frame_id else None
+        still_obs = still.observations.get(still.selected_frame_id) if still.selected_frame_id else None
+
+        def counts(obs: Optional[Observation]) -> Dict[str, int]:
+            if obs is None:
+                return {"accepted": 0, "unknown": 0, "ambiguous": 0, "rejected": 0, "boxes": 0}
+            levels = [i["identity_level"] for i in obs.identities]
+            return {
+                "accepted": levels.count(IDENTITY_ACCEPTED),
+                "unknown": levels.count(IDENTITY_UNKNOWN),
+                "ambiguous": levels.count(IDENTITY_AMBIGUOUS),
+                "rejected": levels.count(IDENTITY_REJECTED),
+                "boxes": len(obs.boxes),
+            }
+
+        return {
+            "sampled": {
+                "session_id": sampled.session_id,
+                "capture_kind": sampled.capture_kind,
+                "observation_count": len(sampled.observations),
+                "dropped_stale": sampled.dropped_stale,
+                "selected": counts(sampled_obs),
+            },
+            "still": {
+                "session_id": still.session_id,
+                "capture_kind": still.capture_kind,
+                "observation_count": len(still.observations),
+                "dropped_stale": still.dropped_stale,
+                "selected": counts(still_obs),
+            },
+            "duplicate_frames_prevented": any(
+                len({o.content_fingerprint for o in s.observations.values()}) < len(s.observations)
+                for s in (sampled, still)
+            ),
+            "note": "Comparator of session mechanics, not measured shelf accuracy or latency.",
+        }

recognization/code/pipeline.py

@@ -116,34 +116,31 @@ class ShelfRecognitionPipeline:
 
         # Classify with selected method
         cls_result = self.matcher.classify_crop(crop, method=method)
-        if cls_result.get("status") != "success":
+        identity_level = cls_result.get("identity_level", "rejected")
-            return {
+        best_sku = cls_result.get("best_sku")
-                "status": "classification_failed",
+        product_info = self.inventory.get_product(best_sku) if best_sku else None
-                "target_box": target_box,
-                "message": cls_result.get("message", "Feature matching failed.")
-            }
 
-        best_sku = cls_result["best_sku"]
+        vis_match = None
-        product_info = self.inventory.get_product(best_sku)
+        if best_sku:
+            vis_match = self.matcher.generate_match_visualization(crop, best_sku, method=method)
 
-        # Generate match visualization
-        vis_match = self.matcher.generate_match_visualization(crop, best_sku, method=method)
-
-        # Generate annotated shelf image highlighting the selected item
         annotated_shelf = self.detector.draw_detections(
             self.current_image, self.current_detections, selected_box_id=target_box["id"]
         )
 
+        click_status = "success" if identity_level == "accepted" else identity_level
         return {
-            "status": "success",
+            "status": click_status,
             "clicked_point": (x, y),
             "target_box": target_box,
             "classification": {
                 "method": method,
                 "matched_sku": best_sku,
+                "identity_level": identity_level,
                 "confidence": cls_result["confidence"],
                 "score": cls_result["score"],
                 "top_candidates": cls_result["ranked_candidates"],
+                "message": cls_result.get("message"),
             },
             "inventory": product_info,
             "crop_image": crop,
@@ -172,9 +169,10 @@ if __name__ == "__main__":
         res = pipeline.process_click(cx, cy, method=args.method)
         print("\n--- Recognition & Inventory Result ---")
         print(f"Status: {res['status']}")
-        if res['status'] == 'success':
+        cls = res.get("classification") or {}
-            prod = res['inventory']
+        print(f"Identity: {cls.get('identity_level')} SKU={cls.get('matched_sku')} score={cls.get('score')}")
-            print(f"Matched SKU: {res['classification']['matched_sku']} (Score: {res['classification']['score']})")
+        prod = res.get("inventory")
+        if prod:
             print(f"Product Name: {prod['product_name']}")
             print(f"Price: {prod['price']} JPY")
             print(f"In Stock: {prod['number_on']} units (Received: {prod['number_in']}, Sold: {prod['number_out']})")

recognization/code/test_observe.py

@@ -0,0 +1,319 @@
+"""Contract tests for identity ranking and session-local observe().
+
+No OpenCV, YOLO, or network. These tests defend replay, isolation,
+stale frames, backpressure, unknown/reject, and live-vs-still mechanics.
+They do not measure shelf accuracy or latency.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import threading
+import unittest
+
+current_dir = os.path.dirname(os.path.abspath(__file__))
+if current_dir not in sys.path:
+    sys.path.insert(0, current_dir)
+
+from identity import (
+    IDENTITY_ACCEPTED,
+    IDENTITY_AMBIGUOUS,
+    IDENTITY_REJECTED,
+    IDENTITY_UNKNOWN,
+    decide_identity,
+)
+from observe import (
+    CAPTURE_SAMPLED,
+    CAPTURE_STILL,
+    FrameRequest,
+    ObserveError,
+    PerceptionAdapter,
+    PerceptionService,
+    ScanService,
+)
+
+
+def _req(session, frame_id, payload, sequence, captured_at=1.0, kind=CAPTURE_SAMPLED):
+    return FrameRequest(
+        frame_id=frame_id,
+        frame_bytes=payload,
+        sequence=sequence,
+        captured_at=captured_at,
+        session_id=session.session_id,
+        zone_id=session.zone_id,
+        catalog_version=session.catalog_version,
+        capture_kind=kind,
+    )
+
+
+class IdentityRankingTests(unittest.TestCase):
+    def test_zero_scores_are_rejected_not_a_sku(self):
+        ranked = decide_identity({"sku_a": 0.0, "sku_b": 0.0})
+        self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED)
+        self.assertIsNone(ranked["best_sku"])
+        self.assertEqual(ranked["status"], IDENTITY_REJECTED)
+
+    def test_empty_scores_are_rejected(self):
+        ranked = decide_identity({})
+        self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED)
+        self.assertIsNone(ranked["best_sku"])
+
+    def test_weak_unique_score_is_unknown(self):
+        ranked = decide_identity({"sku_a": 0.4, "sku_b": 0.0})
+        self.assertEqual(ranked["identity_level"], IDENTITY_UNKNOWN)
+        self.assertIsNone(ranked["best_sku"])
+
+    def test_close_scores_are_ambiguous(self):
+        ranked = decide_identity({"sku_a": 20.0, "sku_b": 19.5})
+        self.assertEqual(ranked["identity_level"], IDENTITY_AMBIGUOUS)
+        self.assertIsNone(ranked["best_sku"])
+
+    def test_unique_strong_score_is_accepted(self):
+        ranked = decide_identity({"sku_a": 40.0, "sku_b": 5.0})
+        self.assertEqual(ranked["identity_level"], IDENTITY_ACCEPTED)
+        self.assertEqual(ranked["best_sku"], "sku_a")
+        self.assertEqual(ranked["status"], "success")
+
+
+class SessionObserveTests(unittest.TestCase):
+    def setUp(self):
+        def detect(frame_bytes: bytes):
+            sku = frame_bytes.decode("ascii", errors="ignore")
+            box = {"id": 0, "box": [0, 0, 10, 10], "sku_hint": sku}
+            return {
+                "width": 10,
+                "height": 10,
+                "boxes": [box],
+                "empty_slots": [],
+                "quality": {"coverage": 1.0},
+            }
+
+        def classify(box):
+            hint = box.get("sku_hint", "")
+            if hint.startswith("unknown"):
+                return {"sku_a": 0.0, "sku_b": 0.0}
+            if hint.startswith("ambig"):
+                return {"sku_a": 20.0, "sku_b": 19.8}
+            if hint.startswith("weak"):
+                return {"sku_a": 0.4, "sku_b": 0.0}
+            if hint.startswith("sku_b"):
+                return {"sku_a": 2.0, "sku_b": 40.0}
+            return {"sku_a": 40.0, "sku_b": 2.0}
+
+        adapter = PerceptionAdapter(
+            detect_fn=detect,
+            classify_fn=classify,
+            model_version="fake",
+            matcher_method="hybrid",
+            feature_extractor="fake",
+        )
+        self.scans = ScanService(PerceptionService(adapter), max_in_flight=1, max_queued=1)
+
+    def test_replay_same_frame_id_and_content_is_idempotent(self):
+        session = self.scans.start_session("zone-1", "stock-v1")
+        first = self.scans.observe_frame(_req(session, "f1", b"sku_a", 1))
+        second = self.scans.observe_frame(_req(session, "f1", b"sku_a", 1))
+        self.assertIs(first, second)
+        self.assertEqual(len(session.observations), 1)
+        self.assertEqual(first.identities[0]["best_sku"], "sku_a")
+
+    def test_same_frame_id_different_content_is_rejected(self):
+        session = self.scans.start_session("zone-1", "stock-v1")
+        self.scans.observe_frame(_req(session, "f1", b"sku_a", 1))
+        with self.assertRaises(ObserveError):
+            self.scans.observe_frame(_req(session, "f1", b"sku_b", 1))
+
+    def test_sessions_do_not_share_observations(self):
+        a = self.scans.start_session("zone-1", "stock-v1")
+        b = self.scans.start_session("zone-1", "stock-v1")
+        self.scans.observe_frame(_req(a, "f1", b"sku_a", 1))
+        self.scans.observe_frame(_req(b, "f1", b"sku_b", 1))
+        self.assertEqual(a.observations["f1"].identities[0]["best_sku"], "sku_a")
+        self.assertEqual(b.observations["f1"].identities[0]["best_sku"], "sku_b")
+        self.assertEqual(a.selected_frame_id, "f1")
+        self.assertEqual(b.selected_frame_id, "f1")
+        self.assertIsNot(a.observations["f1"], b.observations["f1"])
+
+    def test_stale_sequence_cannot_overwrite_selected_state(self):
+        session = self.scans.start_session("zone-1", "stock-v1")
+        newer = self.scans.observe_frame(_req(session, "f2", b"sku_a", 2))
+        older = self.scans.observe_frame(_req(session, "f1", b"sku_b", 1))
+        self.assertEqual(session.selected_frame_id, "f2")
+        self.assertTrue(newer.selected)
+        self.assertFalse(older.selected)
+        self.assertGreaterEqual(session.dropped_stale, 1)
+        self.assertIn("f1", session.observations)
+
+    def test_better_later_frame_replaces_selected_best(self):
+        session = self.scans.start_session("zone-1", "stock-v1")
+        weak = self.scans.observe_frame(_req(session, "f1", b"weak", 1))
+        strong = self.scans.observe_frame(_req(session, "f2", b"sku_a", 2))
+        self.assertFalse(weak.selected)
+        self.assertTrue(strong.selected)
+        self.assertEqual(session.selected_frame_id, "f2")
+
+    def test_unknown_zero_score_is_not_selected_as_sku(self):
+        session = self.scans.start_session("zone-1", "stock-v1")
+        obs = self.scans.observe_frame(_req(session, "f1", b"unknown", 1))
+        self.assertEqual(obs.identities[0]["identity_level"], IDENTITY_REJECTED)
+        self.assertIsNone(obs.identities[0]["best_sku"])
+
+    def test_backpressure_rejects_when_queue_full(self):
+        started = threading.Event()
+        release = threading.Event()
+
+        def slow_detect(frame_bytes: bytes):
+            started.set()
+            release.wait(timeout=2)
+            return {
+                "width": 1,
+                "height": 1,
+                "boxes": [],
+                "empty_slots": [],
+                "quality": {},
+            }
+
+        scans = ScanService(
+            PerceptionService(PerceptionAdapter(detect_fn=slow_detect)),
+            max_in_flight=1,
+            max_queued=0,
+        )
+        session = scans.start_session("zone-1", "stock-v1")
+        errors = []
+
+        def run():
+            try:
+                scans.observe_frame(_req(session, "slow", b"sku_a", 1))
+            except ObserveError as exc:
+                errors.append(str(exc))
+
+        worker = threading.Thread(target=run)
+        worker.start()
+        self.assertTrue(started.wait(timeout=2))
+        with self.assertRaises(ObserveError) as ctx:
+            scans.observe_frame(_req(session, "other", b"sku_b", 2))
+        self.assertIn("backpressure", str(ctx.exception))
+        release.set()
+        worker.join(timeout=2)
+        self.assertTrue(session.dropped_stale >= 1)
+
+    def test_finalize_is_immutable_and_rescan_creates_revision(self):
+        session = self.scans.start_session("zone-1", "stock-v1")
+        self.scans.observe_frame(_req(session, "f1", b"sku_a", 1))
+        self.scans.review(session.session_id, actor="op", reason="looks-right")
+        snap = self.scans.finalize(session.session_id)
+        self.assertEqual(snap["status"], "finalized")
+        with self.assertRaises(ObserveError):
+            self.scans.observe_frame(_req(session, "f2", b"sku_b", 2))
+        with self.assertRaises(ObserveError):
+            self.scans.cancel(session.session_id)
+        child = self.scans.rescan(session.session_id)
+        self.assertEqual(child.status, "draft")
+        self.assertEqual(child.revision, session.revision + 1)
+        self.assertEqual(child.review["previous_session_id"], session.session_id)
+        self.assertEqual(len(child.observations), 0)
+
+    def test_live_vs_still_comparator_does_not_sum_frames(self):
+        sampled = self.scans.start_session("zone-1", "stock-v1", capture_kind=CAPTURE_SAMPLED)
+        still = self.scans.start_session("zone-1", "stock-v1", capture_kind=CAPTURE_STILL)
+        self.scans.observe_frame(_req(sampled, "s1", b"sku_a", 1, kind=CAPTURE_SAMPLED))
+        self.scans.observe_frame(_req(sampled, "s2", b"sku_a", 2, kind=CAPTURE_SAMPLED))
+        self.scans.observe_frame(_req(still, "still1", b"sku_a", 1, kind=CAPTURE_STILL))
+        comparison = self.scans.compare_live_vs_still(sampled.session_id, still.session_id)
+        self.assertEqual(comparison["sampled"]["observation_count"], 2)
+        self.assertEqual(comparison["still"]["observation_count"], 1)
+        self.assertEqual(comparison["sampled"]["selected"]["accepted"], 1)
+        self.assertEqual(comparison["still"]["selected"]["accepted"], 1)
+        self.assertEqual(sampled.selected_frame_id, "s2")
+        self.assertIn("not measured shelf accuracy", comparison["note"])
+
+    def test_expired_session_rejects_new_frames(self):
+        clock = {"t": 100.0}
+
+        def now():
+            return clock["t"]
+
+        scans = ScanService(self.scans.perception, ttl_s=10, clock=now)
+        session = scans.start_session("zone-1", "stock-v1")
+        clock["t"] = 120.0
+        with self.assertRaises(ObserveError):
+            scans.observe_frame(_req(session, "f1", b"sku_a", 1))
+
+
+class SessionHttpTests(unittest.TestCase):
+    def setUp(self):
+        from fastapi import FastAPI
+        from fastapi.testclient import TestClient
+        from observe import ScanService, PerceptionService, PerceptionAdapter, CAPTURE_SAMPLED
+
+        def detect(frame_bytes: bytes):
+            return {
+                "width": 8,
+                "height": 8,
+                "boxes": [{"id": 0, "box": [0, 0, 8, 8], "sku_hint": frame_bytes.decode()}],
+                "empty_slots": [],
+                "quality": {},
+            }
+
+        def classify(box):
+            hint = box.get("sku_hint", "")
+            if hint == "sku_b":
+                return {"sku_a": 2.0, "sku_b": 40.0}
+            return {"sku_a": 40.0, "sku_b": 2.0}
+
+        self.scans = ScanService(
+            PerceptionService(PerceptionAdapter(detect_fn=detect, classify_fn=classify))
+        )
+        app = FastAPI()
+
+        @app.post("/api/sessions")
+        def start(payload: dict):
+            session = self.scans.start_session(
+                payload.get("zone_id", "zone-1"),
+                payload.get("catalog_version", "stock-v1"),
+                capture_kind=payload.get("capture_kind", CAPTURE_SAMPLED),
+            )
+            return {"status": "success", "session": session.snapshot()}
+
+        @app.post("/api/sessions/{session_id}/frames")
+        def frames(session_id: str, payload: dict):
+            import base64
+            session = self.scans.get_session(session_id)
+            request = FrameRequest(
+                frame_id=payload["frame_id"],
+                frame_bytes=base64.b64decode(payload["frame_bytes_b64"]),
+                sequence=int(payload["sequence"]),
+                captured_at=1.0,
+                session_id=session.session_id,
+                zone_id=session.zone_id,
+                catalog_version=session.catalog_version,
+                capture_kind=session.capture_kind,
+            )
+            obs = self.scans.observe_frame(request)
+            return {"status": "success", "observation": obs.as_dict(), "session": session.snapshot()}
+
+        self.client = TestClient(app)
+
+    def test_http_sessions_are_isolated(self):
+        import base64
+        a = self.client.post("/api/sessions", json={"zone_id": "zone-1", "catalog_version": "stock-v1"}).json()
+        b = self.client.post("/api/sessions", json={"zone_id": "zone-1", "catalog_version": "stock-v1"}).json()
+        aid = a["session"]["session_id"]
+        bid = b["session"]["session_id"]
+        oa = self.client.post(
+            f"/api/sessions/{aid}/frames",
+            json={"frame_id": "f1", "sequence": 1, "frame_bytes_b64": base64.b64encode(b"sku_a").decode()},
+        ).json()
+        ob = self.client.post(
+            f"/api/sessions/{bid}/frames",
+            json={"frame_id": "f1", "sequence": 1, "frame_bytes_b64": base64.b64encode(b"sku_b").decode()},
+        ).json()
+        self.assertEqual(oa["observation"]["identities"][0]["best_sku"], "sku_a")
+        self.assertEqual(ob["observation"]["identities"][0]["best_sku"], "sku_b")
+        self.assertNotEqual(aid, bid)
+
+
+
+if __name__ == "__main__":
+    unittest.main()

recognization/code/yolo_adapter.py

@@ -0,0 +1,64 @@
+"""Optional YOLO + matcher adapter for PerceptionService.
+
+Imported only when OpenCV/YOLO are available. Tests use PerceptionAdapter
+fakes instead. This does not keep process-global active images.
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any, Dict, List
+
+import cv2
+import numpy as np
+
+from observe import PerceptionAdapter
+
+
+class YoloMatcherAdapter(PerceptionAdapter):
+    def __init__(self, detector: Any, matcher: Any, method: str = "hybrid"):
+        super().__init__(
+            model_version=getattr(detector, "weights_path", "yolo-local"),
+            matcher_method=method,
+            feature_extractor="surf-or-sift+orb+brisk",
+        )
+        self.detector = detector
+        self.matcher = matcher
+        # observe() calls detect() then classify_box() on the same thread; keep
+        # the decoded frame per thread so concurrent sessions cannot swap images.
+        self._thread_local = threading.local()
+
+    def detect(self, frame_bytes: bytes) -> Dict[str, Any]:
+        array = np.frombuffer(frame_bytes, dtype=np.uint8)
+        image = cv2.imdecode(array, cv2.IMREAD_COLOR)
+        if image is None:
+            return {
+                "width": 0,
+                "height": 0,
+                "boxes": [],
+                "empty_slots": [],
+                "quality": {"limitation": "undecodable-frame"},
+            }
+        self._thread_local.image = image
+        h, w = image.shape[:2]
+        det = self.detector.detect(image)
+        boxes: List[Dict[str, Any]] = det.get("boxes", [])
+        empty = self.detector.detect_empty_slots(image.shape, boxes)
+        return {
+            "width": w,
+            "height": h,
+            "boxes": boxes,
+            "empty_slots": empty,
+            "quality": {"decoded": True},
+        }
+
+    def classify_box(self, box: Dict[str, Any]) -> Dict[str, float]:
+        image = getattr(self._thread_local, "image", None)
+        if image is None:
+            return {}
+        crop = self.detector.crop_roi(image, box["box"])
+        ranked = self.matcher.classify_crop(crop, method=self.matcher_method)
+        return {
+            row["sku_id"]: float(row["score"])
+            for row in ranked.get("ranked_candidates", [])
+        }