# worker_identity.py — stage 5 character identity. FastAPI :8003. # no model of its own conceptually, but siglip2 loads in-process here (transformers, rocm torch) # after the orchestrator has opened a siglip2 session (the mutex guarantees it's the only GPU # resident). known characters + their reference embeddings come from the homesrv orchestrator. import os, uuid, json from fastapi import FastAPI from pydantic import BaseModel import numpy as np import requests import transport app = FastAPI() transport.install_logging(app, "identity") SHM = "/dev/shm" ORCH = os.environ.get("ORCHESTRATOR_URL", "http://192.168.1.104:9090") DEFAULT_THRESHOLD = 0.85 AMBIGUOUS_MARGIN = 0.05 # two known chars within this of each other -> flag low-confidence _siglip = None # (model, processor), lazy-loaded # F3 confirm-before-persist: unnamed provisional characters are held here (in memory, keyed by # session_id = one chapter) and only written to DB/S3 once seen >=2x. seen-once NPCs are never # persisted -> no DB row, no bucket crop. named chars skip the cache and persist immediately. # entry: {emb, crop(np), count, name, gender, appearance, occ:[(panel_id,local_id,conf)], cid} _pending: dict[str, list[dict]] = {} PENDING_PROMOTE_AT = 2 def _load_siglip(): global _siglip if _siglip is None: import torch from transformers import AutoModel, AutoProcessor name = "google/siglip2-so400m-patch16-384" model = AutoModel.from_pretrained(name).to("cuda").eval() _siglip = (model, AutoProcessor.from_pretrained(name), torch) return _siglip def embed_crop(img) -> np.ndarray: """siglip2 image embedding of a BGR/RGB crop (numpy HxWx3), L2-normalized.""" model, proc, torch = _load_siglip() inputs = proc(images=img, return_tensors="pt").to("cuda") with torch.no_grad(): out = model.get_image_features(**inputs) # transformers returns a ModelOutput here (not a bare tensor): use the attention-pooled # image embedding. fall back to mean-pooling patch tokens if a build lacks a pooler head. feat = getattr(out, "pooler_output", None) if feat is None: feat = getattr(out, "last_hidden_state", out) if hasattr(feat, "dim") and feat.dim() == 3: feat = feat.mean(dim=1) feat = feat.detach().cpu().numpy().reshape(-1) return feat / (np.linalg.norm(feat) + 1e-8) def cosine(a: np.ndarray, b: np.ndarray) -> float: return float(np.dot(a, b) / ((np.linalg.norm(a) * np.linalg.norm(b)) + 1e-8)) def gender_ok(a: str, b: str) -> bool: """two people can be the same only if their DECIDED genders agree. unknown on either side is a pass (don't over-block). the single biggest siglip cross-match error is a male crop scoring >0.85 against a female character (shared art style + panel context); this hard-blocks it.""" a, b = (a or "").strip().lower(), (b or "").strip().lower() return not (a in ("m", "f") and b in ("m", "f") and a != b) def match(emb: np.ndarray, known: list, threshold: float): """known: [{"character_id","embedding"(np)}]. returns (character_id|None, confidence, ambiguous).""" if not known: return None, 0.0, False scored = sorted(((cosine(emb, k["embedding"]), k["character_id"]) for k in known), reverse=True) best_conf, best_id = scored[0] ambiguous = len(scored) > 1 and (best_conf - scored[1][0]) < AMBIGUOUS_MARGIN if best_conf >= threshold: return best_id, best_conf, ambiguous return None, best_conf, ambiguous def shortlist(emb: np.ndarray, known: list, k: int = 5, gender: str | None = None) -> list: """TIER-2 evidence: the top-k gender-gated known characters by cosine, best first. Cosine is now a SHORTLISTER, not the decider — gemma /vision/resolve picks from this list. Each item carries the full row (name/gender/description) so the resolver can build a text character-sheet. pure.""" cands = [(cosine(emb, c["embedding"]), c) for c in known if gender_ok(gender, c.get("gender"))] cands.sort(key=lambda t: t[0], reverse=True) return [{**c, "cosine": round(s, 3)} for s, c in cands[:k]] def _crop_bbox(img, bbox): # [x1, y1, x2, y2] pixel corners. gemma answers on a 0-1000 normalized grid and `/vision` converts # to pixels before returning (`worker_vision._bbox_to_pixels`), so this reads real pixels. It did # not before 2026-08-12, which is why crops landed on balloons # (`decisions/identity-bbox.md#bbox-is-normalized`). x1, y1, x2, y2 = bbox return img[y1:y2, x1:x2] def _save_npy(emb: np.ndarray, uri: str): """upload an embedding in .npy format (np.load reads it back on the known-char side).""" tmp = f"{SHM}/emb_{uuid.uuid4().hex[:8]}.npy" np.save(tmp, emb.astype(np.float32)) transport.put(tmp, uri) os.remove(tmp) def _pending_match(pend: list, emb, threshold: float, gender: str | None = None): """index of the pending entry this embedding belongs to, or None (a new provisional). candidates of a conflicting decided gender are excluded. character_id=i keeps the returned index original. pure.""" cands = [{"character_id": i, "embedding": e["emb"]} for i, e in enumerate(pend) if gender_ok(gender, e.get("gender"))] idx, conf, _ = match(emb, cands, threshold) return idx, conf def _persist_char(manga_id, panel_id, local_id, crop, emb, name, gender, appearance) -> str: """upload crop + embedding to S3 and register the row via the orchestrator; return its id.""" import cv2 key = f"{manga_id}/characters/_new/{panel_id}_{local_id}" ref_img_uri, emb_uri = transport.CHAR_PNG_URI.format(key=key), transport.CHAR_NPY_URI.format(key=key) ref_png = f"{SHM}/crop_{uuid.uuid4().hex[:8]}.png" cv2.imwrite(ref_png, crop) transport.put(ref_png, ref_img_uri) os.remove(ref_png) _save_npy(emb, emb_uri) resp = requests.post(f"{ORCH}/characters/create", json={ "manga_id": manga_id, "name": (name or "").strip() or None, "gender": gender or "unknown", "description": appearance or {}, "ref_image_uri": ref_img_uri, "embedding_uri": emb_uri, }, timeout=30).json() _known_cache.pop(manga_id, None) return resp["character_id"] _known_cache: dict = {} # manga_id -> known list; invalidated in _persist_char def _load_known(manga_id: str) -> list: """orchestrator /characters/known -> roster rows with their embeddings loaded. A dangling embedding_uri (row persisted but .npy never landed / bucket wiped) is skipped, not fatal — every panel loads the full roster up front and one bad ref must not 500 the panel. Cached per manga_id for the run; _persist_char drops the entry when the roster changes.""" if manga_id in _known_cache: return _known_cache[manga_id] rows = requests.get(f"{ORCH}/characters/known", params={"manga_id": manga_id}, timeout=30).json() known = [] for c in rows: if not c.get("embedding_uri"): continue try: ref = transport.get(c["embedding_uri"], f"{SHM}/ref_{uuid.uuid4().hex[:8]}.npy") except Exception as e: print(f"[identity] skip {c['character_id']}: bad embedding_uri {c['embedding_uri']}: {e}", flush=True) continue refs = c.get("ref_image_uris") or [] if isinstance(refs, str): try: refs = json.loads(refs) except (ValueError, TypeError): refs = [] known.append({"character_id": c["character_id"], "embedding": np.load(ref), "gender": c.get("gender"), "name": c.get("name"), "species": c.get("species"), "description": c.get("description"), "reference_image_uris": refs}) os.remove(ref) _known_cache[manga_id] = known return known class IdentityInput(BaseModel): panel_uri: str panel_id: str = "" vision_characters: list = [] manga_id: str = "" session_id: str = "" k: int = 5 # shortlist width for gemma's tracklet decider (204: merged from /identity/candidates) @app.post("/identity/resolve") async def resolve(data: IdentityInput): """204: does the cosine assignment/persist pass AND builds gemma's shortlist in one crop/embed pass (was two separate endpoints, each re-cropping + re-embedding + reloading the roster per character). Cosine still makes the provisional assignment; the orchestrator's gemma tracklet phase (/vision/resolve) can still override it — shortlists are returned for every crop regardless of assignment outcome.""" import cv2 local = transport.get(data.panel_uri, f"{SHM}/ident_{uuid.uuid4().hex[:8]}.png") img = cv2.imread(local) # orchestrator /characters/known returns a plain list of character rows (embedding_uri per row). # ponytail: per-manga threshold isn't exposed by the orchestrator yet -> default; wire a # manga_config lookup here if tuning per title ever matters. threshold = DEFAULT_THRESHOLD known = _load_known(data.manga_id) # F3: only THIS session's pending cache is relevant (session == chapter); evict any others so a # long-lived worker doesn't leak past chapters' provisionals. for sid in [s for s in _pending if s != data.session_id]: _pending.pop(sid, None) pend = _pending.setdefault(data.session_id, []) assignments, backfill, new_chars, shortlists = [], [], [], [] for ch in data.vision_characters: # no detected face inside the box -> a background extra, a figure on a poster, or scenery # gemma called a person. Embedding it pollutes the registry and, once the boxes were pixels, # bound an extra to the lead at confidence 1.00. Abstain instead. `/vision` stamps this and # fails open, so a panel it could not gate arrives with has_face=True on every character # (decisions/identity-bbox.md#face-gates-enrollment). Absent key = an older vision blob. if ch.get("has_face") is False: continue crop = _crop_bbox(img, ch["bbox"]) if crop.size == 0: # degenerate/out-of-bounds bbox -> nothing to embed, skip continue emb = embed_crop(crop) # shortlist for gemma's decider, from the roster as it stood before this crop's own outcome. sl = shortlist(emb, known, data.k, ch.get("gender")) # the embedding ships with the crop. gemma's decider can answer "none of these", and the # orchestrator has to mint a character from that crop — which needs an embedding it cannot # compute (siglip is resident here, gemma is resident there, and the mutex forbids both). # Uploading it now is what removes the third siglip pass # (`decisions/identity-bbox.md#none-mints-an-anonymous-character`). key = f"{data.manga_id}/characters/_crops/{data.panel_id}_{ch['local_id']}" crop_uri, emb_uri = transport.CHAR_PNG_URI.format(key=key), transport.CHAR_NPY_URI.format(key=key) cp = f"{SHM}/cc_{uuid.uuid4().hex[:8]}.png"; cv2.imwrite(cp, crop) transport.put(cp, crop_uri); os.remove(cp) _save_npy(emb, emb_uri) shortlists.append({"local_id": ch["local_id"], "crop_uri": crop_uri, "emb_uri": emb_uri, "candidates": [{"character_id": c["character_id"], "name": c.get("name"), "gender": c.get("gender"), "species": c.get("species"), "appearance": c.get("description"), "cosine": c["cosine"], "reference_image_uris": c.get("reference_image_uris", [])} for c in sl]}) # gender gate: never match this crop to a known character of the opposite decided gender. g = ch.get("gender") cands = [k for k in known if gender_ok(g, k.get("gender"))] cid, conf, ambiguous = match(emb, cands, threshold) name = (ch.get("name") or "").strip() if cid is not None: # matched an already-persisted character assignments.append({"local_id": ch["local_id"], "character_id": cid, "confidence": round(conf, 3), "ambiguous": ambiguous}) continue if name: # named -> persist immediately (not an NPC) cid = _persist_char(data.manga_id, data.panel_id, ch["local_id"], crop, emb, name, ch.get("gender"), ch.get("appearance")) known.append({"character_id": cid, "embedding": emb, "gender": ch.get("gender")}) new_chars.append(cid) assignments.append({"local_id": ch["local_id"], "character_id": cid, "confidence": round(conf, 3), "ambiguous": ambiguous}) continue # unnamed + unknown -> confirm-before-persist. match against this chapter's pending cache. pcid, pconf = _pending_match(pend, emb, threshold, ch.get("gender")) if pcid is None: # first sighting: hold, do not persist or assign yet pend.append({"emb": emb, "crop": crop, "count": 1, "name": name, "gender": ch.get("gender"), "appearance": ch.get("appearance"), "occ": [(data.panel_id, ch["local_id"], round(conf, 3))], "cid": None}) continue e = pend[pcid] # seen before this chapter e["count"] += 1 e["occ"].append((data.panel_id, ch["local_id"], round(pconf, 3))) if e["cid"] is None and e["count"] >= PENDING_PROMOTE_AT: # promote -> persist + backfill e["cid"] = _persist_char(data.manga_id, e["occ"][0][0], e["occ"][0][1], e["crop"], e["emb"], e["name"], e["gender"], e["appearance"]) known.append({"character_id": e["cid"], "embedding": e["emb"], "gender": e.get("gender")}) new_chars.append(e["cid"]) for (bp, bl, bc) in e["occ"][:-1]: # earlier sightings that were deferred backfill.append({"panel_id": bp, "local_id": bl, "character_id": e["cid"], "confidence": bc}) if e["cid"]: assignments.append({"local_id": ch["local_id"], "character_id": e["cid"], "confidence": round(pconf, 3), "ambiguous": False}) os.remove(local) return {"panel_id": data.panel_id, "assignments": assignments, "backfill": backfill, "new_characters": new_chars, "shortlists": shortlists} @app.post("/unload") async def unload(): """free the resident siglip2 so the session manager can hand the GPU to the next model. the mutex can't reclaim in-process VRAM -- only the worker holding the model can.""" global _siglip was = _siglip is not None if was: torch = _siglip[2] _siglip = None import gc; gc.collect() torch.cuda.empty_cache() _pending.clear() # F3: drop any held provisionals with the model return {"ok": True, "unloaded": was} @app.get("/health") async def health(): return {"status": "ok"} if __name__ == "__main__": # self-check: cosine + match decision. deterministic vectors, no real model. v = lambda *xs: np.array(xs, dtype=np.float32) a, b = v(1, 0, 0), v(1, 0, 0) assert abs(cosine(a, b) - 1.0) < 1e-6 known = [{"character_id": "c1", "embedding": v(1, 0, 0)}, {"character_id": "c2", "embedding": v(0, 1, 0)}] cid, conf, amb = match(v(0.99, 0.01, 0), known, 0.85) assert cid == "c1" and conf > 0.85 and not amb, (cid, conf, amb) cid, conf, amb = match(v(0, 0, 1), known, 0.85) # nothing close -> new assert cid is None # ambiguous: equidistant-ish between c1 and c2 _, _, amb = match(v(0.71, 0.70, 0), known, 0.5) assert amb is True # gender gate: opposite decided genders never match; unknown on either side passes. assert gender_ok("m", "m") and gender_ok("m", "unknown") and gender_ok("", "f") assert not gender_ok("m", "f") and not gender_ok("f", "m") # a male crop must NOT match a female known char even at high cosine (the Choi Haeseon bug). kn = [{"character_id": "female_char", "embedding": v(1, 0, 0), "gender": "f"}] cands = [k for k in kn if gender_ok("m", k.get("gender"))] assert match(v(1, 0, 0), cands, 0.85)[0] is None # gated out -> new male char, not the female id # pending gate: a female provisional is skipped for a male crop even if embeddings are identical. pend_g = [{"emb": v(1, 0, 0), "count": 1, "gender": "f"}] assert _pending_match(pend_g, v(1, 0, 0), 0.85, "m")[0] is None assert _pending_match(pend_g, v(1, 0, 0), 0.85, "f")[0] == 0 # F3 confirm-before-persist: first sighting is new (held, not persisted); a matching second # sighting hits the same pending entry -> promotes. pend = [] i, _ = _pending_match(pend, v(1, 0, 0), 0.85) assert i is None # nothing pending yet -> new provisional pend.append({"emb": v(1, 0, 0), "count": 1}) i, _ = _pending_match(pend, v(0.99, 0.02, 0), 0.85) assert i == 0 # second sighting matches the held entry i, _ = _pending_match(pend, v(0, 0, 1), 0.85) assert i is None # unrelated crop -> its own new provisional # tier-2 shortlist: top-k by cosine, gender-gated, best first; carries the row for the sheet. kn = [{"character_id": "c1", "embedding": v(1, 0, 0), "gender": "m", "name": "Gojo"}, {"character_id": "c2", "embedding": v(0, 1, 0), "gender": "f", "name": "Choi"}, {"character_id": "c3", "embedding": v(0.9, 0.1, 0), "gender": "m", "name": "Nanami"}] sl = shortlist(v(1, 0, 0), kn, k=2, gender="m") assert [c["character_id"] for c in sl] == ["c1", "c3"] # female c2 gated out; c1 beats c3 assert sl[0]["cosine"] >= sl[1]["cosine"] and sl[0]["name"] == "Gojo" print("worker_identity self-check ok")