Make the catalog hash honest, share the text helpers, drop the duplicate labels

`catalog_hash` hashed the CSV and the view *names*, so replacing
`sku_l09.jpg` with a different product's packshot left the hash unchanged and
a report could claim the same catalog while every l09 decision in it had
moved. It hashes each view's bytes now.

`identity_lanes` carried its own `normalize_text` and `bigrams`, near-copies
of the ones in `rebuild_catalog` that had already drifted (the lane's table
folded ユ to itself, the catalog's did not). The lane imports the catalog's
now, so one read scores the same either way.

`evidence/image6_v2.json` was byte-identical to `eval/labels/image6.json`; the
evidence scripts and the decision document point at the canonical file and the
copy is gone. `score_crop`'s docstring now states the contract its callers
already depend on: the dict's insertion order is the lane's ranking, and
re-sorting by value names a different SKU than the lane chose.

Tests: the lane is shared by the classify pool, so a concurrent-ranking test
pins that one BFMatcher does not cross crops under eight threads.
This commit is contained in:
tngo 2026-09-29 00:48:03 +09:00
parent fc49ba60e6
commit b6c87c45ea
36 changed files with 89 additions and 1524 deletions

View file

@ -564,6 +564,51 @@ class DecisionPolicyTests(unittest.TestCase):
self.assertEqual(decision["identity_level"], "rejected")
class LocalLaneConcurrencyTests(unittest.TestCase):
"""The lane is shared by the classify pool, so `rank` must be thread-safe.
One `cv2.BFMatcher` is reused across calls. The classify pool runs up to
eight workers, so a lane that mutated matcher state per call would return
another crop's answer under load - the failure would look like a
recognition error, not a crash.
"""
def test_concurrent_ranking_matches_solo_ranking(self):
import threading
import cv2
from identity_lanes import LocalLane
stock_dir = os.path.join(os.path.dirname(current_dir), "stock-lopia-clean")
if not os.path.isdir(stock_dir):
self.skipTest("no catalog checked out")
image_path = os.path.join(os.path.dirname(current_dir), "add", "image6.jpg")
if not os.path.isfile(image_path):
self.skipTest("no evaluation photo checked out")
lane = LocalLane(stock_dir)
image = cv2.imread(image_path)
crops = [image[700:1000, 600 + 40 * i:800 + 40 * i] for i in range(8)]
solo = lane.rank(crops[0])
results = {}
errors = []
def work(index: int) -> None:
try:
for _ in range(6):
results[index] = lane.rank(crops[index])
except Exception as error: # noqa: BLE001 - reported below
errors.append(repr(error))
threads = [threading.Thread(target=work, args=(i,)) for i in range(8)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
self.assertEqual(errors, [])
self.assertEqual(results[0], solo, "a shared matcher must not cross crops")
class CatalogImageTests(unittest.TestCase):
def test_catalog_image_path_is_resolved_through_inventory(self):
engine, _, _, _ = make_engine()