qwen_test_runner/vision/fuse.py

"""
fuse.py — the fusion engine: bind every captured signal into one FusedScene.

Consumes a `solids_digest` (compact JSON-able snapshot of a `Solids` — detection
boxes + scores + mask polygons + mask quality, continuous depth nearness, saliency
scores, OCR with confidence, style/class/symmetry/layout) plus the caption structs
(slot-registry JSON from the 9B structurer) and the raw captions, and emits the
fused relational representation:

  entities   — addressable instances (person_1, person_2, dog) with position grid,
               offset-from-center, continuous depth + rank, saliency + rank, mask,
               and STRATIFIED OWNED ATTRIBUTES (ownership decided by segmentation-
               polygon containment with confidence + margin thresholds)
  relations  — pairwise predicates + continuous dx/dy/distance/iou/depth-delta
  counts     — synonym-collapsed instance counts
  shared_basin — attributes NOT confidently assignable (never subjectively grouped),
               with per-entity likelihoods and the reason
  scene      — voted setting/style/mood + layout/symmetry/OCR/actions
  quality    — retained confidences + grounding accounting + overall_confidence

Pure numpy + stdlib + PIL (polygon rasterization) — torch-free, CPU-testable.
The ONLY GPU dependency is upstream: the optional `attr_boxes` in the digest come
from a second GroundingDINO pass over `phrases_for_grounding(...)` phrases.
"""

from __future__ import annotations

import re
from collections import Counter, defaultdict
from typing import Optional

import numpy as np

from . import derive
from .coords import CoordSpace
from .fuse_schema import (FusedScene, MASK_POLY_MAX_POINTS, MAX_ENTITIES,
                          MAX_RELATION_ENTITIES)
from .metrics import _depluralize, _seg_poly_points, _seg_rasterize, labels_match
from .specialists import box_to_space, poly_to_space
from .strata import _content_tokens, classify_stratum, is_groundable

# Containment rasterization grid (mask polygons are ≤64 points; 160² cells is
# ample resolution for an ownership FRACTION).
_GRID = 160

# Depth-relation threshold on the normalized nearness delta — same magnitude the
# spatial_relations engine uses on its normalized per-box depth deltas.
_DEPTH_REL_TOL = 0.15

# "near" relation threshold on centroid-distance / image-diagonal.
_NEAR_DIST = 0.25

# Positional-cue lexicon for caption-subject binding votes.
_POS_LEFT = frozenset({"left", "leftmost"})
_POS_RIGHT = frozenset({"right", "rightmost"})
_POS_FRONT = frozenset({"front", "foreground", "nearest", "closest", "nearer", "closer"})
_POS_BACK = frozenset({"behind", "background", "back", "farthest", "farther", "rear"})
_POS_TALL = frozenset({"tall", "taller", "tallest"})


# ═════════════════════════════════════════════════════════════════════════════
# Digest — the GPU→CPU handoff (also the durability/parquet payload)
# ═════════════════════════════════════════════════════════════════════════════

def solids_digest(s) -> dict:
    """Compact, JSON-able, deterministic snapshot of a Solids. Retains the signals
    the build_* task projections drop (mask quality, OCR conf, continuous nearness,
    the full saliency ranking, symmetry magnitudes)."""
    from .coords import BBox
    W, H = s.size
    nearness = (derive.depth_scalars(s.boxes, s.depth, s.depth_higher_is_nearer)
                if (s.depth is not None and s.boxes) else None)
    sal = derive.subject_scores(s.boxes, s.size, s.saliency) if s.boxes else []
    boxes = []
    for i, b in enumerate(s.boxes):
        mask = b.get("mask")
        poly = (derive.outline_polygon(mask, b["label"],
                                       max_points=MASK_POLY_MAX_POINTS)["outline"]
                if mask is not None else None)
        # GDINO emits unclamped boxes (border objects go past the frame) — clip
        # once at the digest boundary so all downstream geometry is in-range
        clipped = BBox(*[float(v) for v in b["box"]]).clip((W, H)).as_list()
        boxes.append({
            "label": str(b["label"]),
            "box": clipped,
            "score": float(b.get("score", 1.0)),
            "area_px": derive._area(clipped),
            "sal": float(sal[i]) if i < len(sal) else 0.0,
            "nearness": (round(float(nearness[i]), 4) if nearness is not None else None),
            "mask_poly": poly or None,
            "mask_quality": (float(b["mask_score"]) if b.get("mask_score") is not None
                             else None),
        })
    ocr = {"full_text": "", "lines": []}
    if s.ocr:
        ocr["full_text"] = str(s.ocr.get("full_text", ""))
        for ln in s.ocr.get("lines", []):
            q = ln.get("box")
            flat = ([min(max(float(v), 0.0), float(W if i % 2 == 0 else H))
                     for xy in q for i, v in enumerate(xy)] if q else None)
            ocr["lines"].append({"text": str(ln["text"]),
                                 "quad": flat,
                                 "conf": (float(ln["conf"]) if ln.get("conf") is not None
                                          else None)})
    attr_boxes = []
    for a in getattr(s, "attr_boxes", []):
        a = dict(a)
        a["box"] = BBox(*[float(v) for v in a["box"]]).clip((W, H)).as_list()
        attr_boxes.append(a)
    return {
        "size": [int(W), int(H)],
        "boxes": boxes,
        "attr_boxes": attr_boxes,
        "class_top": [{"label": str(c["label"]), "score": float(c["score"])}
                      for c in (s.class_top or [])],
        "style": s.style,
        "ocr": ocr,
        "symmetry": (derive.symmetry_scores(s.gray) if s.gray is not None else None),
        "layout": derive.layout_kind(s.boxes, s.size),
        "higher_is_nearer": bool(s.depth_higher_is_nearer),
    }


# ═════════════════════════════════════════════════════════════════════════════
# Caption-side collection + cross-source merge
# ═════════════════════════════════════════════════════════════════════════════

def _attr_key(text: str) -> frozenset:
    """Dedup key: depluralized content-token set (raw + depluralized forms so the
    crude depluralizer can't split 'dress'/'dres')."""
    toks = _content_tokens(text)
    return frozenset(t for tok in toks for t in (tok, _depluralize(tok)))


_HEAD_SPLIT_RE = re.compile(r"\b(?:in|on|at|with|of|to|wearing|holding)\b")


def _subject_head(name: str) -> str:
    """Head noun = last content token BEFORE the first post-modifier ("woman in
    red" → woman, "person on a bench" → person); falls back to the full-name head
    when the pre-modifier part has no content tokens."""
    pre = _HEAD_SPLIT_RE.split((name or "").lower(), 1)[0]
    toks = _content_tokens(pre)
    if toks:
        return toks[-1]
    toks = _content_tokens(name)
    return toks[-1] if toks else ""


def _collect_merged(caption_structs: dict) -> tuple:
    """caption_structs: {source: struct-or-None}. Returns (merged_attrs, actions,
    votes) where merged_attrs = [{text, key, sources, consensus, stratum,
    parents:{source: subject_name}}] (cross-source dedup: token-set equal-or-subset
    → canonical = longest text; provenance kept). Subjects are NEVER merged across
    sources by name — merging happens only through binding downstream."""
    sources = [k for k, v in caption_structs.items() if v]
    n_src = max(1, len(sources))

    raw_items = []
    actions = []
    votes = {"setting": Counter(), "style": Counter(), "mood": {}}
    for src in sources:
        st = caption_structs[src]
        for subj in (st.get("subjects") or []):
            name = str(subj.get("name") or "").strip()
            for att in (subj.get("attributes") or []):
                att = str(att).strip()
                if att:
                    raw_items.append({"text": att, "source": src, "subject": name})
        for act in (st.get("actions") or []):
            act = str(act).strip()
            if act:
                actions.append({"text": act, "source": src})
        if st.get("setting"):
            votes["setting"][str(st["setting"])] += 1
        if st.get("style"):
            votes["style"][str(st["style"])] += 1
        if st.get("mood"):
            votes["mood"][src] = str(st["mood"])

    # merge: iterate longest-token-set first so merged records are supersets
    raw_items.sort(key=lambda it: (-len(_attr_key(it["text"])), it["text"], it["source"]))
    merged = []
    for it in raw_items:
        key = _attr_key(it["text"])
        if not key:
            continue
        home = next((m for m in merged if key <= m["key"] or m["key"] <= key), None)
        if home is None:
            merged.append({"text": it["text"], "key": key, "sources": [it["source"]],
                           "parents": {it["source"]: it["subject"]}})
        else:
            home["key"] = home["key"] | key
            if len(it["text"]) > len(home["text"]):
                home["text"] = it["text"]
            if it["source"] not in home["sources"]:
                home["sources"].append(it["source"])
            home["parents"].setdefault(it["source"], it["subject"])
    for m in merged:
        m["sources"] = sorted(m["sources"])
        m["consensus"] = round(len(m["sources"]) / n_src, 4)
        m["stratum"] = classify_stratum(m["text"])

    # actions: same dedup, no parents
    actions.sort(key=lambda it: (-len(_attr_key(it["text"])), it["text"], it["source"]))
    merged_acts = []
    for it in actions:
        key = _attr_key(it["text"])
        if not key:
            continue
        home = next((m for m in merged_acts if key <= m["key"] or m["key"] <= key), None)
        if home is None:
            merged_acts.append({"text": it["text"], "key": key, "sources": [it["source"]]})
        else:
            home["key"] = home["key"] | key
            if len(it["text"]) > len(home["text"]):
                home["text"] = it["text"]
            if it["source"] not in home["sources"]:
                home["sources"].append(it["source"])
    for m in merged_acts:
        m["sources"] = sorted(m["sources"])
        m["consensus"] = round(len(m["sources"]) / n_src, 4)

    return merged, merged_acts, votes


def phrases_for_grounding(caption_structs: dict) -> list:
    """The canonical phrases the GPU grounding pass should box — merged attribute
    texts whose stratum is GROUNDABLE, emitted stripped-lowercase (ground_phrases
    lowercases anyway; matching its normalization keeps the downstream
    phrase↔attribute lookup exact)."""
    merged, _, _ = _collect_merged(caption_structs)
    return sorted({m["text"].strip().lower() for m in merged
                   if is_groundable(m["stratum"])})


# ═════════════════════════════════════════════════════════════════════════════
# Geometry: entities, containment, relations
# ═════════════════════════════════════════════════════════════════════════════

def _grid_cell(cx: float, cy: float, W: float, H: float) -> str:
    col = "left" if cx < W / 3 else ("center" if cx < 2 * W / 3 else "right")
    row = "upper" if cy < H / 3 else ("middle" if cy < 2 * H / 3 else "lower")
    return f"{row} {col}"


def _build_entities(digest: dict, dedup_iou: float) -> list:
    """Dedup detector double-boxes, cap by saliency, order left-to-right, and
    assign _uniq_labels ids. Returns internal entity dicts (pixel space)."""
    boxes = [dict(b) for b in digest["boxes"]]
    kept = []
    for b in sorted(boxes, key=lambda b: (-b["score"], b["box"][0])):
        if any(derive._iou(b["box"], k["box"]) >= dedup_iou
               and labels_match(b["label"], k["label"]) for k in kept):
            continue
        kept.append(b)
    kept.sort(key=lambda b: -b["sal"])
    kept = kept[:MAX_ENTITIES]
    for rank, b in enumerate(kept, 1):
        b["sal_rank"] = rank
    kept.sort(key=lambda b: (0.5 * (b["box"][0] + b["box"][2]), b["box"][1]))
    ids = derive._uniq_labels([b["label"] for b in kept])
    for b, eid in zip(kept, ids):
        b["id"] = eid
    if any(b["nearness"] is not None for b in kept):
        by_near = sorted([b for b in kept if b["nearness"] is not None],
                         key=lambda b: -b["nearness"])
        for rank, b in enumerate(by_near, 1):
            b["depth_rank"] = rank
    return kept


def _entity_grid_mask(ent: dict, size, cache: dict):
    """Rasterized mask polygon on the containment grid (cached per entity)."""
    eid = ent["id"]
    if eid in cache:
        return cache[eid]
    W, H = size
    m = None
    if ent.get("mask_poly"):
        pts = _seg_poly_points(ent["mask_poly"])
        m = _seg_rasterize(pts, _GRID, _GRID / max(1.0, W), _GRID / max(1.0, H))
    cache[eid] = m
    return m


def _own_frac(attr_box, ent: dict, size, cache: dict) -> float:
    """|attr_box ∩ entity mask| / |attr_box| on the grid; box-fraction fallback
    when the entity has no mask polygon ("box_containment")."""
    W, H = size
    m = _entity_grid_mask(ent, size, cache)
    x1 = int(np.clip(attr_box[0] / W * _GRID, 0, _GRID))
    y1 = int(np.clip(attr_box[1] / H * _GRID, 0, _GRID))
    x2 = int(np.clip(np.ceil(attr_box[2] / W * _GRID), 0, _GRID))
    y2 = int(np.clip(np.ceil(attr_box[3] / H * _GRID), 0, _GRID))
    if x2 <= x1 or y2 <= y1:
        return 0.0
    if m is not None:
        return float(m[y1:y2, x1:x2].sum()) / float((x2 - x1) * (y2 - y1))
    # box fallback: inter / area(attr_box)
    b = ent["box"]
    ix1, iy1 = max(attr_box[0], b[0]), max(attr_box[1], b[1])
    ix2, iy2 = min(attr_box[2], b[2]), min(attr_box[3], b[3])
    inter = max(0.0, ix2 - ix1) * max(0.0, iy2 - iy1)
    a = derive._area(attr_box)
    return inter / a if a > 0 else 0.0


def _pair_predicates(a: dict, b: dict) -> list:
    """a→b predicates, same semantics as derive.spatial_relations (dominant axis,
    containment first, depth via nearness delta) — pinned by a consistency test."""
    preds = []
    if derive._contains(b["box"], a["box"]):
        preds.append("inside")
    else:
        ca, cb = derive._centroid(a["box"]), derive._centroid(b["box"])
        dx, dy = cb[0] - ca[0], cb[1] - ca[1]
        if abs(dx) >= abs(dy):
            preds.append("left_of" if dx > 0 else "right_of")
        else:
            preds.append("above" if dy > 0 else "below")
    if a["nearness"] is not None and b["nearness"] is not None:
        d = a["nearness"] - b["nearness"]
        if abs(d) >= _DEPTH_REL_TOL:
            preds.append("in_front_of" if d > 0 else "behind")
    return preds


# ═════════════════════════════════════════════════════════════════════════════
# The fusion
# ═════════════════════════════════════════════════════════════════════════════

def fuse(digest: dict, caption_structs: dict, raw_captions: Optional[dict] = None,
         *, t_own: float = 0.60, t_margin: float = 0.25, dedup_iou: float = 0.75,
         coord_space: CoordSpace = CoordSpace.NORM_0_1000) -> dict:
    """→ FusedScene as a schema-validated dict. Deterministic: same inputs →
    byte-identical json.dumps. See the module docstring for the shape and the
    ownership cascade; t_own / t_margin are the assignment thresholds (an attribute
    below them lands in shared_basin with per-entity likelihoods — never guessed)."""
    W, H = digest["size"]
    size = (float(W), float(H))
    raw_captions = raw_captions or {}

    ents = _build_entities(digest, dedup_iou)
    by_id = {e["id"]: e for e in ents}
    grid_cache: dict = {}

    # entity output records (attributes attached during the cascade)
    ent_out = {}
    for e in ents:
        cx, cy = derive._centroid(e["box"])
        ent_out[e["id"]] = {
            "id": e["id"], "label": e["label"], "detection_score": round(e["score"], 4),
            "box": box_to_space(e["box"], coord_space, size),
            "centroid": poly_to_space([cx, cy], coord_space, size),
            "area_frac": round(e["area_px"] / (W * H + 1e-9), 4),
            "position": {"grid": _grid_cell(cx, cy, W, H),
                         "offset_from_center": [round((cx - W / 2) / (W / 2), 4),
                                                round((cy - H / 2) / (H / 2), 4)]},
            "depth": ({"nearness": round(e["nearness"], 4), "rank": e["depth_rank"]}
                      if e.get("nearness") is not None and e.get("depth_rank") else None),
            "saliency": {"score": round(e["sal"], 4), "rank": e["sal_rank"]},
            "is_primary": e["sal_rank"] == 1,
            "mask": ({"polygon": poly_to_space(e["mask_poly"], coord_space, size),
                      "quality": (round(e["mask_quality"], 4)
                                  if e.get("mask_quality") is not None else None)}
                     if e.get("mask_poly") else None),
            "caption_bindings": [],
            "attributes": [],
        }

    counts = Counter()
    for e in ents:
        counts["person" if labels_match(e["label"], "person") else e["label"]] += 1
    people = counts.get("person", 0)

    merged, merged_acts, votes = _collect_merged(caption_structs)
    n_sources = sum(1 for v in caption_structs.values() if v)

    # grounding lookup: canonical phrase -> [attr box records]. ground_phrases
    # lowercases its input phrases, so BOTH sides normalize to strip().lower()
    # (an uppercase structurer attribute must not silently lose its grounding).
    grounded_by_phrase = defaultdict(list)
    for a in digest.get("attr_boxes", []):
        grounded_by_phrase[str(a["phrase"]).strip().lower()].append(a)
    for recs in grounded_by_phrase.values():
        recs.sort(key=lambda r: -r["score"])

    def _candidates(m) -> tuple:
        """(candidates, head_ok) — head_ok is False only when a subject head EXISTS
        and matched no entity (fallback-to-all is then a guess, not evidence).
        Pose/action attributes fall back to PERSON entities only — verbs apply to
        agents, not to a baseball glove."""
        cands, any_head = [], False
        for src, subj in sorted(m["parents"].items()):
            head = _subject_head(subj)
            any_head = any_head or bool(head)
            for e in ents:
                if head and labels_match(head, e["label"]) and e not in cands:
                    cands.append(e)
        if cands:
            return cands, True
        if m.get("stratum") in ("pose", "action"):
            persons = [e for e in ents if labels_match(e["label"], "person")]
            if persons:
                return persons, not any_head
        return list(ents), not any_head

    basin, scene_attrs, assigned_attrs = [], [], []
    unresolved = []           # (merged, candidates) awaiting the binding pass
    subj_votes = defaultdict(lambda: defaultdict(float))   # (src, subject) -> {eid: score}
    subj_nvotes = defaultdict(int)

    def _attach(eid, m, conf, method, margin=None, gbox=None, gscore=None):
        rec = {"text": m["text"], "stratum": m["stratum"], "sources": m["sources"],
               "consensus": m["consensus"], "grounded": gbox is not None,
               "box": box_to_space(gbox, coord_space, size) if gbox else None,
               "grounding_score": round(gscore, 4) if gscore is not None else None,
               "ownership": {"confidence": round(conf, 4),
                             "margin": round(margin, 4) if margin is not None else None,
                             "method": method},
               "region_on_owner": None}
        if gbox is not None:
            o = by_id[eid]
            ocx, ocy = derive._centroid(o["box"])
            acx, acy = derive._centroid(gbox)
            hw = max(1.0, (o["box"][2] - o["box"][0]) / 2)
            hh = max(1.0, (o["box"][3] - o["box"][1]) / 2)
            rel_y, rel_x = (acy - ocy) / hh, (acx - ocx) / hw
            rec["region_on_owner"] = {
                "vertical": "upper" if rel_y < -1 / 3 else ("lower" if rel_y > 1 / 3 else "middle"),
                "horizontal": "left" if rel_x < -1 / 3 else ("right" if rel_x > 1 / 3 else "center"),
                "offset": [round(rel_x, 4), round(rel_y, 4)]}
        ent_out[eid]["attributes"].append(rec)
        assigned_attrs.append((m, eid, conf))
        for src, subj in m["parents"].items():
            subj_votes[(src, subj)][eid] += conf
            subj_nvotes[(src, subj)] += 1

    def _to_basin(m, reason, gbox=None, fracs=None):
        cands = [{"entity_id": e["id"], "likelihood": round(f, 4)}
                 for e, f in (fracs or []) if f >= 0.15]
        cands.sort(key=lambda c: -c["likelihood"])
        basin.append({"text": m["text"], "stratum": m["stratum"], "sources": m["sources"],
                      "consensus": m["consensus"], "reason": reason,
                      "grounded": gbox is not None,
                      "box": box_to_space(gbox, coord_space, size) if gbox else None,
                      "candidates": cands})

    # ── pass A: scene routing, single-candidate fast path, grounded assignment ──
    n_grounded_phrases = 0
    for m in merged:
        if m["stratum"] == "scene_level":
            scene_attrs.append({"text": m["text"], "stratum": m["stratum"],
                                "sources": m["sources"]})
            continue
        gboxes = (grounded_by_phrase.get(m["text"].strip().lower(), [])
                  if is_groundable(m["stratum"]) else [])
        if gboxes:
            n_grounded_phrases += 1
        cands, head_ok = _candidates(m)

        if len(cands) == 1 and head_ok:
            e = cands[0]
            if gboxes:
                f = _own_frac(gboxes[0]["box"], e, size, grid_cache)
                if f < 0.2:      # caption mentions something visibly NOT on this entity
                    _to_basin(m, "low_margin", gboxes[0]["box"], [(e, f)])
                    continue
                _attach(e["id"], m, 0.9, "single_entity",
                        gbox=gboxes[0]["box"], gscore=gboxes[0]["score"])
            else:
                _attach(e["id"], m, 0.9, "single_entity")
            continue

        if gboxes:
            if not cands:    # zero entities survived detection — grounded but unownable
                _to_basin(m, "low_margin", gboxes[0]["box"])
                continue
            top = gboxes[0]["score"]
            accepted = [g for g in gboxes if g["score"] >= 0.75 * top]
            taken_eids = set()
            any_assigned = False
            best_fracs = None
            for g in accepted:
                fracs = sorted(((e, _own_frac(g["box"], e, size, grid_cache))
                                for e in cands), key=lambda ef: -ef[1])
                if best_fracs is None:
                    best_fracs = (g, fracs)
                f1 = fracs[0][1]
                f2 = fracs[1][1] if len(fracs) > 1 else 0.0
                winner = fracs[0][0]
                if winner["id"] in taken_eids:
                    continue
                method = ("mask_containment"
                          if _entity_grid_mask(winner, size, grid_cache) is not None
                          else "box_containment")
                if f1 >= t_own and (f1 - f2) >= t_margin:
                    taken_eids.add(winner["id"])
                    any_assigned = True
                    _attach(winner["id"], m, f1, method, margin=f1 - f2,
                            gbox=g["box"], gscore=g["score"])
            if not any_assigned:
                g, fracs = best_fracs
                _to_basin(m, "low_margin", g["box"], fracs)
            continue

        unresolved.append((m, cands))

    # ── binding: caption subjects ↔ entities (votes from grounded assignments
    #    + positional cues in subject names and raw captions) ───────────────────
    def _positional_vote(text: str, cands: list, votes_out: dict):
        if not cands:
            return 0
        toks = set(_content_tokens(text))
        if toks & _POS_LEFT:
            e = min(cands, key=lambda e: derive._centroid(e["box"])[0])
            votes_out[e["id"]] += 0.5
            return 1
        if toks & _POS_RIGHT:
            e = max(cands, key=lambda e: derive._centroid(e["box"])[0])
            votes_out[e["id"]] += 0.5
            return 1
        if toks & _POS_FRONT and any(e.get("depth_rank") for e in cands):
            e = min((e for e in cands if e.get("depth_rank")), key=lambda e: e["depth_rank"])
            votes_out[e["id"]] += 0.5
            return 1
        if toks & _POS_BACK and any(e.get("depth_rank") for e in cands):
            e = max((e for e in cands if e.get("depth_rank")), key=lambda e: e["depth_rank"])
            votes_out[e["id"]] += 0.5
            return 1
        if toks & _POS_TALL:
            e = max(cands, key=lambda e: e["box"][3] - e["box"][1])
            votes_out[e["id"]] += 0.5
            return 1
        return 0

    bindings = {}       # (src, subject) -> (eid, bind_conf)
    all_subjects = {(src, subj) for m in merged for src, subj in m["parents"].items()}
    for (src, subj) in sorted(all_subjects):
        head = _subject_head(subj)
        cands = [e for e in ents if head and labels_match(head, e["label"])] or list(ents)
        v = dict(subj_votes.get((src, subj), {}))
        v = defaultdict(float, v)
        nv = subj_nvotes.get((src, subj), 0)
        pos_n = _positional_vote(subj, cands, v)
        raw = raw_captions.get(src, "")
        if raw and head:
            # "<positional> [word] <head>"  — tight adjacency, so a positional word
            # in a NEIGHBORING clause can't vote for this subject
            for mtc in re.finditer(rf"\b(\w+)\s+(?:\w+\s+)?{re.escape(head)}\b", raw.lower()):
                pos_n += _positional_vote(mtc.group(1), cands, v)
            # "<head> ... on the <positional>" — reject windows crossing an "and"
            # (clause boundary: "a woman AND a man on the right")
            for mtc in re.finditer(rf"\b{re.escape(head)}\b([\w\s,]{{0,24}}?)\bon the (\w+)",
                                   raw.lower()):
                if " and " in f" {mtc.group(1)} ":
                    continue
                pos_n += _positional_vote(mtc.group(2), cands, v)
        # bind on >=2 containment votes, OR any explicit positional cue (the caption
        # author's own disambiguation — stronger evidence than one weak containment)
        if not v or (nv < 2 and pos_n < 1):
            continue
        total = sum(v.values())
        eid, top = max(sorted(v.items()), key=lambda kv: kv[1])
        bind_conf = top / total if total > 0 else 0.0
        if bind_conf >= 0.6:
            bindings[(src, subj)] = (eid, bind_conf)
            ent_out[eid]["caption_bindings"].append(
                {"source": src, "subject_name": subj, "confidence": round(bind_conf, 4)})

    # ── pass B: unresolved attributes inherit their subject's binding ───────────
    for m, cands in unresolved:
        # collapse per entity (max conf) with DETERMINISTIC iteration order —
        # set iteration over tuples is process-hash-dependent
        by_eid: dict = {}
        for src, subj in sorted(m["parents"].items()):
            if (src, subj) in bindings:
                eid, conf = bindings[(src, subj)]
                by_eid[eid] = max(by_eid.get(eid, 0.0), conf)
        if len(by_eid) == 1:
            eid, bind_conf = next(iter(by_eid.items()))
            _attach(eid, m, bind_conf * 0.6, "caption_binding")
        elif len(by_eid) > 1:
            _to_basin(m, "ambiguous_binding",
                      fracs=sorted(((by_id[eid], conf) for eid, conf in by_eid.items()),
                                   key=lambda ef: (-ef[1], ef[0]["id"])))
        else:
            reason = ("no_grounding_multi_entity" if is_groundable(m["stratum"])
                      else "abstract_unbound")
            n_c = max(1, len(cands))
            _to_basin(m, reason, fracs=[(e, 1.0 / n_c) for e in cands])

    # ── actions: one person → attach as stratum "action"; else scene-level ─────
    # (actions are NOT part of the attribute-routing identity
    # assigned + basin + scene_level == phrases_total — separate accumulator)
    scene_actions = []
    action_confs = []
    person_ents = [e for e in ents if labels_match(e["label"], "person")]
    for m in merged_acts:
        if len(person_ents) == 1:
            e = person_ents[0]
            ent_out[e["id"]]["attributes"].append(
                {"text": m["text"], "stratum": "action", "sources": m["sources"],
                 "consensus": m["consensus"], "grounded": False, "box": None,
                 "grounding_score": None,
                 "ownership": {"confidence": 0.9, "margin": None,
                               "method": "single_entity"},
                 "region_on_owner": None})
            action_confs.append(0.9)
        else:
            scene_actions.append({"text": m["text"], "stratum": "action",
                                  "sources": m["sources"]})

    # ── relations among the top-K entities by saliency ──────────────────────────
    rel_ents = sorted(ents, key=lambda e: e["sal_rank"])[:MAX_RELATION_ENTITIES]
    rel_ents = sorted(rel_ents, key=lambda e: [x["id"] for x in ents].index(e["id"]))
    diag = (W ** 2 + H ** 2) ** 0.5 or 1.0
    relations = []
    for i in range(len(rel_ents)):
        for j in range(i + 1, len(rel_ents)):
            a, b = rel_ents[i], rel_ents[j]
            # containment is orientation-independent: put the INNER entity first so
            # "inside" always reads a-inside-b regardless of left-to-right id order
            if (derive._contains(a["box"], b["box"])
                    and not derive._contains(b["box"], a["box"])):
                a, b = b, a
            ca, cb = derive._centroid(a["box"]), derive._centroid(b["box"])
            depth_delta = (round(a["nearness"] - b["nearness"], 4)
                           if a["nearness"] is not None and b["nearness"] is not None
                           else None)
            relations.append({
                "a": a["id"], "b": b["id"],
                "predicates": _pair_predicates(a, b),
                "dx": round((cb[0] - ca[0]) / W, 4), "dy": round((cb[1] - ca[1]) / H, 4),
                "distance": round(((cb[0] - ca[0]) ** 2 + (cb[1] - ca[1]) ** 2) ** 0.5 / diag, 4),
                "iou": round(derive._iou(a["box"], b["box"]), 4),
                "depth_delta": depth_delta,
                "confidence": round(min(a["score"], b["score"]), 4),
            })

    # ── scene block ─────────────────────────────────────────────────────────────
    set_votes = votes["setting"]
    setting_val = None
    if set_votes:
        ranked = set_votes.most_common()
        setting_val = ("unknown" if len(ranked) > 1 and ranked[0][1] == ranked[1][1]
                       else ranked[0][0])
    style_votes = votes["style"]
    style_val = digest.get("style") or (style_votes.most_common(1)[0][0]
                                        if style_votes else None)
    mood_per_source = votes["mood"]
    mood_val = None
    if mood_per_source:
        mood_counts = Counter(mood_per_source.values())
        mood_val = mood_counts.most_common(1)[0][0]
    sym = digest.get("symmetry")
    sym_axis = "none"
    if sym:
        v, h = sym["lr"] >= 0.80, sym["tb"] >= 0.80
        sym_axis = "radial" if (v and h) else ("vertical" if v else ("horizontal" if h else "none"))
    ocr_lines = []
    for ln in digest.get("ocr", {}).get("lines", []):
        q = ln.get("quad")
        box = None
        if q:
            xs, ys = q[0::2], q[1::2]
            box = box_to_space([min(xs), min(ys), max(xs), max(ys)], coord_space, size)
        ocr_lines.append({"text": ln["text"], "box": box, "conf": ln.get("conf")})

    scene = {
        "setting": {"value": setting_val, "votes": dict(sorted(set_votes.items()))},
        "style": {"value": style_val, "caption_votes": dict(sorted(style_votes.items())),
                  "specialist": digest.get("style")},
        "mood": {"value": mood_val, "per_source": dict(sorted(mood_per_source.items()))},
        "layout": digest.get("layout", "unknown"),
        "symmetry": {"axis": sym_axis,
                     "lr": round(sym["lr"], 4) if sym else None,
                     "tb": round(sym["tb"], 4) if sym else None},
        "actions": scene_actions,
        "scene_attributes": scene_attrs,
        "ocr": {"full_text": digest.get("ocr", {}).get("full_text", ""), "lines": ocr_lines},
        "class_top": digest.get("class_top", []),
    }

    # ── quality + accounting ────────────────────────────────────────────────────
    n_groundable = sum(1 for m in merged if is_groundable(m["stratum"]))
    n_scene = len(scene_attrs)
    n_ungroundable = sum(1 for m in merged
                         if not is_groundable(m["stratum"]) and m["stratum"] != "scene_level")
    n_assigned_attrs = len({id(m) for m, _, _ in assigned_attrs})
    mask_qualities = [e["mask_quality"] for e in ents if e.get("mask_quality") is not None]
    ocr_confs = [l["conf"] for l in ocr_lines if l.get("conf") is not None]
    det_mean = float(np.mean([e["score"] for e in ents])) if ents else 0.0
    own_confs = [c for _, _, c in assigned_attrs] + action_confs
    n_routed = n_assigned_attrs + len(basin)
    overall = round(
        0.5 * (float(np.mean(own_confs)) if own_confs else 0.0)
        + 0.3 * det_mean
        + 0.2 * (n_assigned_attrs / n_routed if n_routed else 0.0), 4)

    out = {
        "coord_space": str(coord_space.value if hasattr(coord_space, "value") else coord_space),
        "image_size": [int(W), int(H)],
        "counts": {"total_entities": len(ents), "people": people,
                   "by_label": dict(sorted(counts.items()))},
        "entities": [ent_out[e["id"]] for e in ents],
        "relations": relations,
        "shared_basin": basin,
        "scene": scene,
        "quality": {
            "n_caption_sources": n_sources,
            "detection_score_mean": round(det_mean, 4),
            "mask_quality_mean": (round(float(np.mean(mask_qualities)), 4)
                                  if mask_qualities else None),
            "ocr_conf_mean": (round(float(np.mean(ocr_confs)), 4) if ocr_confs else None),
            "grounding": {"phrases_total": len(merged),
                          "phrases_grounded": n_grounded_phrases,
                          "assigned": n_assigned_attrs,
                          "basin": len(basin),
                          "scene_level": n_scene,
                          "ungroundable": n_ungroundable},
            "overall_confidence": overall,
        },
    }
    # schema-validate + normalize field order (byte-determinism of json.dumps)
    return FusedScene.model_validate(out).model_dump()


# ═════════════════════════════════════════════════════════════════════════════
# semantic_association — the 12th deterministic task (VLM→INTEGRATE reclass)
# ═════════════════════════════════════════════════════════════════════════════

def build_semantic_association(scene: dict, max_items: int = 32) -> dict:
    """FusedScene → the EXISTING registry shape {associations:[{a,relation,b}]}
    (enum: left_of/right_of/near/is_a/related_to). Deterministic: geometry gives
    left_of/right_of/near; caption bindings give is_a (bound subject head vs the
    detector label, e.g. woman is_a person)."""
    out, seen = [], set()

    def _emit(a, rel, b):
        t = (a, rel, b)
        if t not in seen and len(out) < max_items:
            seen.add(t)
            out.append({"a": a, "relation": rel, "b": b})

    for r in scene.get("relations", []):
        for p in r.get("predicates", []):
            if p in ("left_of", "right_of"):
                _emit(r["a"], p, r["b"])
        if r.get("distance") is not None and r["distance"] <= _NEAR_DIST:
            _emit(r["a"], "near", r["b"])
    for e in scene.get("entities", []):
        for cb in e.get("caption_bindings", []):
            head = _subject_head(cb["subject_name"])
            if head and head != e["label"] and labels_match(head, e["label"]):
                _emit(head, "is_a", e["label"])
    return {"associations": out}