backend/shared_play.py

"""Backend shared play service."""
from __future__ import annotations

from collections import defaultdict
from typing import Any
from sqlmodel import Session, select
from models import PromptDismissal, SimilarityEdge
from backend.constants import (
    BLOCKED_RATINGS,
    EDGE_TYPE_WEIGHTS,
    POSITIVE_RATINGS,
    STRONG_POSITIVE_RATINGS,
    SUPPRESSED_RATINGS,
)
from backend.scenarios import play_excluded_from_surfacing

_RELATED_EDGE_LIMIT = 36
_RELATED_TWO_HOP_LIMIT = 24
_RELATED_DIRECT_MIN = 0.30
_RELATED_TWO_HOP_MIN = 0.40
_RELATED_TWO_HOP_DECAY = 0.72


def _directions_compatible(self, left: list[str], right: list[str], kink: dict[str, Any]) -> bool:
    left_set = set(left)
    right_set = set(right)
    mode = self._direction_mode_for_kink(kink)
    if "together" in left_set and "together" in right_set:
        return True
    if "to_me" in left_set and "by_me" in right_set:
        return True
    if "by_me" in left_set and "to_me" in right_set:
        return True
    if mode == "shared" and ("together" in left_set or "together" in right_set):
        return True
    return False

def _directions_compatible_group(self, directions_list: list[list[str]], kink: dict[str, Any]) -> bool:
    if len(directions_list) < 2:
        return False
    direction_sets = [set(item) for item in directions_list]
    if all("together" in direction_set for direction_set in direction_sets):
        return True
    directed = [direction_set & {"to_me", "by_me"} for direction_set in direction_sets]
    if all(direction_set for direction_set in directed):
        has_receiver = any("to_me" in direction_set for direction_set in directed)
        has_giver = any("by_me" in direction_set for direction_set in directed)
        if has_receiver and has_giver:
            return True
    mode = self._direction_mode_for_kink(kink)
    if mode == "shared" and any("together" in direction_set for direction_set in direction_sets):
        return all(direction_set & {"together", "to_me", "by_me"} for direction_set in direction_sets)
    return False

def _safe_prompt_candidate(self, kink: dict[str, Any], support_count: int) -> bool:
    if play_excluded_from_surfacing(kink):
        return False
    if not kink.get("prompt_eligible"):
        return False
    penalty = self._discoverability_penalty(kink["cluster"])
    if penalty >= 0.9:
        return False
    if support_count >= 2 and kink["popularity"] >= 60:
        return True
    return kink["popularity"] >= 320

def _direct_shared_ids(self, left: dict[str, Any], right: dict[str, Any]) -> list[str]:
    detail_by_id = self._catalog()["detail_by_id"]
    direct_ids: list[str] = []
    shared_ids = set(left["plays"]) & set(right["plays"])
    for kink_id in sorted(shared_ids):
        left_state = left["plays"][kink_id]
        right_state = right["plays"][kink_id]
        kink = detail_by_id.get(kink_id)
        if not kink:
            continue
        if play_excluded_from_surfacing(kink):
            continue
        if (
            left_state["interest_state"] in POSITIVE_RATINGS
            and right_state["interest_state"] in POSITIVE_RATINGS
            and self._directions_compatible(left_state["directions"], right_state["directions"], kink)
        ):
            direct_ids.append(kink_id)
    return direct_ids

def _related_edge_strength(edge: dict[str, Any]) -> float:
    return min(float(edge["score"]) * float(EDGE_TYPE_WEIGHTS.get(edge.get("type", ""), 0.8)), 1.0)

def _similarity_edges_by_source(
    self,
    source_ids: set[str],
    *,
    per_source_limit: int = _RELATED_EDGE_LIMIT,
) -> dict[str, list[dict[str, Any]]]:
    if not source_ids:
        return {}
    with Session(self.engine) as session:
        rows = session.exec(
            select(SimilarityEdge)
            .where(SimilarityEdge.left_kink_id.in_(sorted(source_ids)))
            .order_by(SimilarityEdge.left_kink_id, SimilarityEdge.score.desc())
        ).all()
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in rows:
        items = grouped[row.left_kink_id]
        if len(items) >= per_source_limit:
            continue
        items.append({"id": row.right_kink_id, "score": row.score, "type": row.similarity_type})
    return dict(grouped)

def _positive_play_states(user: dict[str, Any]) -> dict[str, dict[str, Any]]:
    return {
        kink_id: state
        for kink_id, state in user.get("plays", {}).items()
        if state.get("interest_state") in POSITIVE_RATINGS
    }


def _positive_scenario_states(user: dict[str, Any]) -> dict[str, dict[str, Any]]:
    return {
        scenario_id: state
        for scenario_id, state in user.get("scenario_preferences", {}).items()
        if state.get("interest_state") in POSITIVE_RATINGS
    }


def _overlap_eligible_scenario_states(user: dict[str, Any]) -> dict[str, dict[str, Any]]:
    plays = user.get("plays", {})
    out: dict[str, dict[str, Any]] = {}
    for scenario_id, state in user.get("scenario_preferences", {}).items():
        if state.get("interest_state") not in STRONG_POSITIVE_RATINGS:
            continue
        parent_id = str(state.get("parent_kink_id", "") or "")
        parent_state = plays.get(parent_id, {}) if parent_id else {}
        if parent_state.get("interest_state") not in STRONG_POSITIVE_RATINGS:
            continue
        out[scenario_id] = {
            **state,
            "parent_kink_id": parent_id,
            "directions": state.get("directions", []) or parent_state.get("directions", []),
        }
    return out

def _related_direction_compatible(
    self,
    left_state: dict[str, Any],
    right_state: dict[str, Any],
    left_kink: dict[str, Any],
    right_kink: dict[str, Any],
) -> bool:
    left_dirs = left_state.get("directions", [])
    right_dirs = right_state.get("directions", [])
    return self._directions_compatible(left_dirs, right_dirs, left_kink) or self._directions_compatible(
        left_dirs,
        right_dirs,
        right_kink,
    )

def _related_match_entry(
    self,
    left_user: dict[str, Any],
    right_user: dict[str, Any],
    left_kink_id: str,
    right_kink_id: str,
    *,
    score: float,
    hops: int,
    path: list[str],
    edge_types: list[str],
) -> dict[str, Any] | None:
    detail_by_id = self._catalog()["detail_by_id"]
    left_kink = detail_by_id.get(left_kink_id)
    right_kink = detail_by_id.get(right_kink_id)
    if not left_kink or not right_kink:
        return None
    left_state = left_user["plays"][left_kink_id]
    right_state = right_user["plays"][right_kink_id]
    if not self._related_direction_compatible(left_state, right_state, left_kink, right_kink):
        return None
    bridge = None
    if len(path) == 3 and path[1] not in {left_kink_id, right_kink_id}:
        bridge = detail_by_id.get(path[1])
    reason = (
        f"{left_kink['name']} is graph-near {right_kink['name']}"
        if hops == 1
        else f"{left_kink['name']} connects to {right_kink['name']} via {bridge['name'] if bridge else 'a nearby kink'}"
    )
    return {
        "left_user_id": left_user["id"],
        "right_user_id": right_user["id"],
        "left_kink": left_kink,
        "right_kink": right_kink,
        "bridge_kink": bridge,
        "score": round(float(score), 4),
        "hops": hops,
        "path": path,
        "edge_types": edge_types,
        "reasons": [reason],
        "partner_reactions": {
            left_user["id"]: {
                "kink_id": left_kink_id,
                "interest_state": left_state["interest_state"],
                "directions": left_state.get("directions", []),
            },
            right_user["id"]: {
                "kink_id": right_kink_id,
                "interest_state": right_state["interest_state"],
                "directions": right_state.get("directions", []),
            },
        },
    }

def _related_positive_matches_for_pair(
    self,
    left: dict[str, Any],
    right: dict[str, Any],
    *,
    limit: int = 12,
) -> list[dict[str, Any]]:
    left_positive = _positive_play_states(left)
    right_positive = _positive_play_states(right)
    if not left_positive or not right_positive:
        return []
    exact_shared = set(left_positive) & set(right_positive)
    left_only = set(left_positive) - exact_shared
    right_only = set(right_positive) - exact_shared
    if not left_only or not right_only:
        return []

    first_edges = _similarity_edges_by_source(self, left_only | right_only, per_source_limit=_RELATED_EDGE_LIMIT)
    mid_ids = {
        edge["id"]
        for source in left_only | right_only
        for edge in first_edges.get(source, [])
        if edge["id"] not in exact_shared
    }
    second_edges = _similarity_edges_by_source(self, mid_ids, per_source_limit=_RELATED_TWO_HOP_LIMIT)
    best_by_pair: dict[tuple[str, str], dict[str, Any]] = {}

    def add_candidate(
        left_kink_id: str,
        right_kink_id: str,
        *,
        score: float,
        hops: int,
        path: list[str],
        edge_types: list[str],
    ) -> None:
        threshold = _RELATED_DIRECT_MIN if hops == 1 else _RELATED_TWO_HOP_MIN
        if score < threshold:
            return
        entry = _related_match_entry(
            self,
            left,
            right,
            left_kink_id,
            right_kink_id,
            score=score,
            hops=hops,
            path=path,
            edge_types=edge_types,
        )
        if not entry:
            return
        key = (left_kink_id, right_kink_id)
        current = best_by_pair.get(key)
        if current is None or entry["score"] > current["score"] or (entry["score"] == current["score"] and entry["hops"] < current["hops"]):
            best_by_pair[key] = entry

    for left_kink_id in left_only:
        for edge in first_edges.get(left_kink_id, []):
            strength = _related_edge_strength(edge)
            if edge["id"] in right_only:
                add_candidate(
                    left_kink_id,
                    edge["id"],
                    score=strength,
                    hops=1,
                    path=[left_kink_id, edge["id"]],
                    edge_types=[edge["type"]],
                )
            for edge2 in second_edges.get(edge["id"], []):
                if edge2["id"] not in right_only:
                    continue
                add_candidate(
                    left_kink_id,
                    edge2["id"],
                    score=strength * _related_edge_strength(edge2) * _RELATED_TWO_HOP_DECAY,
                    hops=2,
                    path=[left_kink_id, edge["id"], edge2["id"]],
                    edge_types=[edge["type"], edge2["type"]],
                )

    for right_kink_id in right_only:
        for edge in first_edges.get(right_kink_id, []):
            strength = _related_edge_strength(edge)
            if edge["id"] in left_only:
                add_candidate(
                    edge["id"],
                    right_kink_id,
                    score=strength,
                    hops=1,
                    path=[edge["id"], right_kink_id],
                    edge_types=[edge["type"]],
                )
            for edge2 in second_edges.get(edge["id"], []):
                if edge2["id"] not in left_only:
                    continue
                add_candidate(
                    edge2["id"],
                    right_kink_id,
                    score=strength * _related_edge_strength(edge2) * _RELATED_TWO_HOP_DECAY,
                    hops=2,
                    path=[edge2["id"], edge["id"], right_kink_id],
                    edge_types=[edge["type"], edge2["type"]],
                )

    related = sorted(
        best_by_pair.values(),
        key=lambda item: (
            int(item["hops"]),
            -float(item["score"]),
            item["left_kink"]["name"].lower(),
            item["right_kink"]["name"].lower(),
        ),
    )
    return related[:limit]

def _related_positive_matches_for_users(
    self,
    users: list[dict[str, Any]],
    *,
    limit: int = 12,
) -> list[dict[str, Any]]:
    if len(users) != 2:
        return []
    return self._related_positive_matches_for_pair(users[0], users[1], limit=limit)

def _pair_safe_similarity_floor(self, direct_count: int, support_count: int, popularity: float) -> bool:
    if support_count >= 2:
        return True
    if direct_count >= 3 and popularity >= 75:
        return True
    if direct_count >= 2 and popularity >= 180:
        return True
    return direct_count == 1 and popularity >= 450

def _direct_shared_ids_for_users(self, users: list[dict[str, Any]]) -> list[str]:
    detail_by_id = self._catalog()["detail_by_id"]
    if len(users) < 2:
        return []
    shared_ids = set(users[0]["plays"])
    for user in users[1:]:
        shared_ids &= set(user["plays"])
    direct_ids: list[str] = []
    for kink_id in sorted(shared_ids):
        states = [user["plays"][kink_id] for user in users]
        kink = detail_by_id.get(kink_id)
        if not kink:
            continue
        if play_excluded_from_surfacing(kink):
            continue
        if all(state["interest_state"] in POSITIVE_RATINGS for state in states) and self._directions_compatible_group(
            [state["directions"] for state in states],
            kink,
        ):
            direct_ids.append(kink_id)
    return direct_ids


def _scenario_direct_matches_for_users(
    self,
    users: list[dict[str, Any]],
    *,
    limit: int = 24,
) -> list[dict[str, Any]]:
    if len(users) < 2:
        return []
    detail_by_id = self._catalog()["detail_by_id"]
    positive_by_user = [_overlap_eligible_scenario_states(user) for user in users]
    if any(not prefs for prefs in positive_by_user):
        return []
    shared_ids = set(positive_by_user[0])
    for prefs in positive_by_user[1:]:
        shared_ids &= set(prefs)
    out: list[dict[str, Any]] = []
    for scenario_id in sorted(shared_ids):
        states = [prefs[scenario_id] for prefs in positive_by_user]
        scenario = detail_by_id.get(scenario_id)
        if not scenario:
            continue
        parent_ids = [state.get("parent_kink_id", "") for state in states]
        parent_id = next((pid for pid in parent_ids if pid), "")
        parent = detail_by_id.get(parent_id) if parent_id else None
        compatibility_kink = parent or scenario
        if not self._directions_compatible_group([state.get("directions", []) for state in states], compatibility_kink):
            continue
        out.append(
            {
                "scenario_kink": scenario,
                "parent_kink": parent,
                "score": round(1.0 + min(float(scenario.get("popularity", 0.0) or 0.0) / 100000.0, 0.2), 4),
                "partner_reactions": {
                    user["id"]: {
                        "parent_kink_id": positive_by_user[i][scenario_id].get("parent_kink_id", ""),
                        "interest_state": positive_by_user[i][scenario_id].get("interest_state", ""),
                        "directions": positive_by_user[i][scenario_id].get("directions", []),
                    }
                    for i, user in enumerate(users)
                },
            }
        )
    out.sort(
        key=lambda item: (
            -float(item.get("score", 0.0) or 0.0),
            (item.get("parent_kink") or {}).get("name", ""),
            item["scenario_kink"]["name"].lower(),
        )
    )
    return out[:limit]


def _scenario_related_matches_for_pair(
    self,
    left: dict[str, Any],
    right: dict[str, Any],
    *,
    limit: int = 24,
) -> list[dict[str, Any]]:
    detail_by_id = self._catalog()["detail_by_id"]
    left_positive = _overlap_eligible_scenario_states(left)
    right_positive = _overlap_eligible_scenario_states(right)
    if not left_positive or not right_positive:
        return []
    left_by_parent: dict[str, list[tuple[str, dict[str, Any]]]] = defaultdict(list)
    right_by_parent: dict[str, list[tuple[str, dict[str, Any]]]] = defaultdict(list)
    for sid, state in left_positive.items():
        parent_id = state.get("parent_kink_id", "")
        if parent_id:
            left_by_parent[parent_id].append((sid, state))
    for sid, state in right_positive.items():
        parent_id = state.get("parent_kink_id", "")
        if parent_id:
            right_by_parent[parent_id].append((sid, state))

    out: list[dict[str, Any]] = []
    for parent_id in sorted(set(left_by_parent) & set(right_by_parent)):
        parent = detail_by_id.get(parent_id)
        if not parent:
            continue
        for left_sid, left_state in left_by_parent[parent_id]:
            for right_sid, right_state in right_by_parent[parent_id]:
                if left_sid == right_sid:
                    continue
                left_scenario = detail_by_id.get(left_sid)
                right_scenario = detail_by_id.get(right_sid)
                if not left_scenario or not right_scenario:
                    continue
                if not self._directions_compatible(left_state.get("directions", []), right_state.get("directions", []), parent):
                    continue
                score = 0.7 + min(float(parent.get("popularity", 0.0) or 0.0) / 100000.0, 0.18)
                out.append(
                    {
                        "left_user_id": left["id"],
                        "right_user_id": right["id"],
                        "parent_kink": parent,
                        "left_scenario": left_scenario,
                        "right_scenario": right_scenario,
                        "score": round(score, 4),
                        "reasons": [f"Both are scenarios under {parent['name']}"],
                        "partner_reactions": {
                            left["id"]: {
                                "scenario_kink_id": left_sid,
                                "interest_state": left_state.get("interest_state", ""),
                                "directions": left_state.get("directions", []),
                            },
                            right["id"]: {
                                "scenario_kink_id": right_sid,
                                "interest_state": right_state.get("interest_state", ""),
                                "directions": right_state.get("directions", []),
                            },
                        },
                    }
                )
    out.sort(
        key=lambda item: (
            -float(item.get("score", 0.0) or 0.0),
            item["parent_kink"]["name"].lower(),
            item["left_scenario"]["name"].lower(),
            item["right_scenario"]["name"].lower(),
        )
    )
    return out[:limit]


def _scenario_related_matches_for_users(
    self,
    users: list[dict[str, Any]],
    *,
    limit: int = 24,
) -> list[dict[str, Any]]:
    if len(users) != 2:
        return []
    return self._scenario_related_matches_for_pair(users[0], users[1], limit=limit)

def _shared_play_list_for_users(self, users: list[dict[str, Any]], limit: int = 24) -> dict[str, list[dict[str, Any]]]:
    detail_by_id = self._catalog()["detail_by_id"]
    direct_ids = self._direct_shared_ids_for_users(users)
    candidate_scores: dict[str, float] = defaultdict(float)
    candidate_reasons: dict[str, set[str]] = defaultdict(set)
    candidate_support: dict[str, set[str]] = defaultdict(set)
    suppressed_ids = {
        kink_id
        for user in users
        for kink_id, state in user["plays"].items()
        if state["interest_state"] in SUPPRESSED_RATINGS | BLOCKED_RATINGS
    }
    for kink_id in direct_ids:
        source = detail_by_id.get(kink_id)
        if not source:
            continue
        for edge in self._edges_for_kink(kink_id, limit=12):
            if edge["id"] in direct_ids or any(edge["id"] in user["plays"] for user in users):
                continue
            target = detail_by_id.get(edge["id"])
            if not target or edge["id"] in suppressed_ids:
                continue
            if not target.get("shared_eligible"):
                continue
            score = edge["score"] - self._discoverability_penalty(target["cluster"])
            score += min(target["popularity"] / 60000.0, 0.3)
            if score <= 0:
                continue
            candidate_scores[target["id"]] += score
            candidate_support[target["id"]].add(kink_id)
            candidate_reasons[target["id"]].add(f"close to {source['name']}")

    direct_count = len(direct_ids)
    similar = []
    for kink_id, score in sorted(candidate_scores.items(), key=lambda item: item[1], reverse=True):
        if kink_id not in detail_by_id:
            continue
        kink = detail_by_id[kink_id]
        support_count = len(candidate_support.get(kink_id, set()))
        if not self._pair_safe_similarity_floor(direct_count, support_count, kink["popularity"]):
            continue
        similar.append(
            {
                "kink": kink,
                "score": score + min(support_count * 0.08, 0.24),
                "reasons": self._finalize_reasons(
                    candidate_reasons[kink_id],
                    fallback="close to what you both already like",
                ),
                "support_count": support_count,
            }
        )
        if len(similar) >= limit:
            break
    direct = []
    for kink_id in direct_ids:
        kink = detail_by_id.get(kink_id)
        if not kink:
            continue
        entry = dict(kink)
        entry["partner_reactions"] = {
            user["id"]: user["plays"][kink_id]["interest_state"]
            for user in users
            if kink_id in user.get("plays", {})
        }
        direct.append(entry)
    direct.sort(key=lambda item: (-item["popularity"], item["name"].lower()))
    related = self._related_positive_matches_for_users(users, limit=limit)
    scenario_matches = self._scenario_direct_matches_for_users(users, limit=limit)
    scenario_related = self._scenario_related_matches_for_users(users, limit=limit)
    return {
        "direct_matches": direct,
        "related_matches": related,
        "scenario_matches": scenario_matches,
        "scenario_related_matches": scenario_related,
        "similar_matches": similar,
    }

def _empty_shared_play_response() -> dict[str, Any]:
    return {
        "direct_matches": [],
        "related_matches": [],
        "scenario_matches": [],
        "scenario_related_matches": [],
        "similar_matches": [],
        "viewer_boundary_alerts": {"count": 0},
    }


def _viewer_boundary_alert_count(viewer: dict[str, Any], others: list[dict[str, Any]]) -> int:
    """Count the viewer's hard_no kinks that any other participant marks love/like.

    Returns a count only — kink identities are intentionally not exposed. The viewer
    is the only side that sees their own number; partners querying the same group
    get a different number computed against their own hard_nos.
    """
    viewer_id = viewer.get("id")
    viewer_hard_no_ids = {
        kink_id
        for kink_id, state in viewer.get("plays", {}).items()
        if state.get("interest_state") in BLOCKED_RATINGS
    }
    if not viewer_hard_no_ids:
        return 0
    conflicting: set[str] = set()
    for other in others:
        if other.get("id") == viewer_id:
            continue
        for kink_id, state in other.get("plays", {}).items():
            if (
                kink_id in viewer_hard_no_ids
                and state.get("interest_state") in STRONG_POSITIVE_RATINGS
            ):
                conflicting.add(kink_id)
    return len(conflicting)


def shared_play_list(self, user_id: str, partner_id: str, limit: int = 24) -> dict[str, Any]:
    left = self.get_user(user_id)
    right = self.get_user(partner_id)
    if not left or not right:
        return _empty_shared_play_response()
    result = self._shared_play_list_for_users([left, right], limit=limit)
    result["viewer_boundary_alerts"] = {
        "count": _viewer_boundary_alert_count(left, [right]),
    }
    return result

def shared_play_list_for_group(self, owner_user_id: str, group_id: str, limit: int = 24) -> dict[str, Any]:
    users = self._group_participants(owner_user_id, group_id)
    if len(users) < 2:
        return _empty_shared_play_response()
    result = self._shared_play_list_for_users(users, limit=limit)
    viewer = next((u for u in users if u.get("id") == owner_user_id), None)
    others = [u for u in users if u.get("id") != owner_user_id]
    count = _viewer_boundary_alert_count(viewer, others) if viewer else 0
    result["viewer_boundary_alerts"] = {"count": count}
    return result

def prompt_candidates(self, user_id: str, limit: int = 8, group_id: str | None = None) -> list[dict[str, Any]]:
    if group_id:
        return self.group_prompt_candidates(user_id, group_id, limit=limit)
    user = self.get_user(user_id)
    if not user or not user["partners"]:
        return []

    detail_by_id = self._catalog()["detail_by_id"]
    dismissed: set[str] = set()
    with Session(self.engine) as session:
        dismissed = {
            row.kink_id
            for row in session.exec(select(PromptDismissal).where(PromptDismissal.user_id == user_id)).all()
        }

    user_play_ids = set(user["plays"])
    blocked_ids = {
        kink_id
        for kink_id, state in user["plays"].items()
        if state["interest_state"] in SUPPRESSED_RATINGS | BLOCKED_RATINGS
    }
    candidate_scores: dict[str, float] = defaultdict(float)
    candidate_support: dict[str, set[str]] = defaultdict(set)
    candidate_reasons: dict[str, set[str]] = defaultdict(set)

    for partner_id in user["partners"]:
        partner = self.get_user(partner_id)
        if not partner:
            continue
        partner_positive = {
            kink_id: state
            for kink_id, state in partner["plays"].items()
            if state["interest_state"] in POSITIVE_RATINGS
        }
        direct_shared_ids = self._direct_shared_ids(user, partner)
        direct_neighbor_ids = {
            edge["id"]
            for direct_id in direct_shared_ids
            for edge in self._edges_for_kink(direct_id, limit=12)
        }

        for kink_id in direct_shared_ids:
            source = detail_by_id.get(kink_id)
            if not source:
                continue
            for edge in self._edges_for_kink(kink_id, limit=10):
                candidate_id = edge["id"]
                if candidate_id in user_play_ids or candidate_id in blocked_ids or candidate_id in dismissed:
                    continue
                target = detail_by_id.get(candidate_id)
                if not target:
                    continue
                candidate_scores[candidate_id] += edge["score"] + min(target["popularity"] / 80000.0, 0.18)
                candidate_support[candidate_id].add(kink_id)
                candidate_reasons[candidate_id].add(f"aligned with {source['name']}")

        for kink_id in partner_positive:
            if kink_id in user_play_ids or kink_id in blocked_ids or kink_id in dismissed:
                continue
            target = detail_by_id.get(kink_id)
            supported_by_shared_lane = kink_id in direct_neighbor_ids and target and target["popularity"] >= 80
            if not target or (not self._safe_prompt_candidate(target, 1) and not supported_by_shared_lane):
                continue
            candidate_scores[kink_id] += 0.55 + min(target["popularity"] / 70000.0, 0.22)
            candidate_support[kink_id].add(f"partner:{partner_id}")
            if supported_by_shared_lane:
                candidate_support[kink_id].add("shared-lane")
            candidate_reasons[kink_id].add("worth getting your take on")

    prompts = []
    for kink_id, score in sorted(candidate_scores.items(), key=lambda item: item[1], reverse=True):
        kink = detail_by_id.get(kink_id)
        if not kink:
            continue
        support_count = len(candidate_support.get(kink_id, set()))
        if not self._safe_prompt_candidate(kink, support_count):
            continue
        prompts.append(
            {
                "kink": kink,
                "score": score,
                "support_count": support_count,
                "reasons": sorted(candidate_reasons[kink_id]),
            }
        )
        if len(prompts) >= limit:
            break
    return prompts

def group_prompt_candidates(self, requester_user_id: str, group_id: str, limit: int = 8) -> list[dict[str, Any]]:
    requester = self.get_user(requester_user_id)
    users = self._group_participants(requester_user_id, group_id)
    if not requester or len(users) < 2:
        return []

    detail_by_id = self._catalog()["detail_by_id"]
    with Session(self.engine) as session:
        dismissed = {
            row.kink_id
            for row in session.exec(select(PromptDismissal).where(PromptDismissal.user_id == requester_user_id)).all()
        }

    requester_play_ids = set(requester["plays"])
    blocked_ids = {
        kink_id
        for kink_id, state in requester["plays"].items()
        if state["interest_state"] in SUPPRESSED_RATINGS | BLOCKED_RATINGS
    }
    direct_shared_ids = self._direct_shared_ids_for_users(users)
    shared_neighbor_ids = {
        edge["id"]
        for direct_id in direct_shared_ids
        for edge in self._edges_for_kink(direct_id, limit=12)
    }
    candidate_scores: dict[str, float] = defaultdict(float)
    candidate_support: dict[str, set[str]] = defaultdict(set)
    candidate_reasons: dict[str, set[str]] = defaultdict(set)

    for kink_id in direct_shared_ids:
        source = detail_by_id.get(kink_id)
        if not source:
            continue
        for edge in self._edges_for_kink(kink_id, limit=10):
            candidate_id = edge["id"]
            if candidate_id in requester_play_ids or candidate_id in blocked_ids or candidate_id in dismissed:
                continue
            target = detail_by_id.get(candidate_id)
            if not target:
                continue
            candidate_scores[candidate_id] += edge["score"] + min(target["popularity"] / 80000.0, 0.18)
            candidate_support[candidate_id].add(kink_id)
            candidate_reasons[candidate_id].add(f"aligned with {source['name']}")

    for partner in users:
        if partner["id"] == requester_user_id:
            continue
        partner_positive = {
            kink_id: state
            for kink_id, state in partner["plays"].items()
            if state["interest_state"] in POSITIVE_RATINGS
        }
        for kink_id in partner_positive:
            if kink_id in requester_play_ids or kink_id in blocked_ids or kink_id in dismissed:
                continue
            target = detail_by_id.get(kink_id)
            if not target:
                continue
            if kink_id in shared_neighbor_ids:
                continue
            candidate_scores[kink_id] += 0.55 + min(target["popularity"] / 70000.0, 0.22)
            candidate_support[kink_id].add(partner["id"])
            candidate_reasons[kink_id].add("worth getting your take on")

    prompts = []
    for kink_id, score in sorted(candidate_scores.items(), key=lambda item: item[1], reverse=True):
        kink = detail_by_id.get(kink_id)
        if not kink:
            continue
        support_count = len(candidate_support.get(kink_id, set()))
        if not self._safe_prompt_candidate(kink, support_count):
            continue
        prompts.append(
            {
                "kink": kink,
                "score": score,
                "support_count": support_count,
                "reasons": self._finalize_reasons(
                    candidate_reasons[kink_id],
                    fallback="worth getting your take on",
                ),
            }
        )
        if len(prompts) >= limit:
            break
    return prompts