"""Phase-5 consensus helpers: alliance detection, addressed-to classification, status-checks, and unaddressed-factor probing. All four are short JSON-shaped orchestrator calls layered on top of `json_calls.orchestrator_call`. """ from __future__ import annotations import logging import re from typing import Any from app.services.json_calls import orchestrator_call from app.services.prompts import ( ALLIANCE_DETECTION_PROMPT, ADDRESSED_TO_PROMPT, CONSENSUS_STATUS_PROMPT, UNADDRESSED_FACTOR_PROMPT, ) LOG = logging.getLogger(__name__) def _format_finalization_block( participants: list[Any], final_opinions: dict[str, str], ) -> str: lines: list[str] = [] for p in participants: text = final_opinions.get(p.participant_id, "(no final opinion)").strip() lines.append(f"--- {p.name} (id={p.participant_id}) ---") lines.append(text) lines.append("") return "\n".join(lines).strip() def _format_roster_block(participants: list[Any]) -> str: return "\n".join( f"- id: {p.participant_id} | name: {p.name}" for p in participants ) def _format_alliance_block(groups: list[dict[str, Any]]) -> str: lines: list[str] = [] for i, g in enumerate(groups): members = ", ".join(g.get("members") or []) lines.append(f"Group {i}: stance=\"{g.get('stance', '')}\" members=[{members}]") return "\n".join(lines) async def detect_alliances( *, orchestrator_model_id: str, question: str, participants: list[Any], final_opinions: dict[str, str], api_log: list[dict[str, Any]] | None = None, ) -> list[dict[str, Any]]: prompt = ALLIANCE_DETECTION_PROMPT.format( question=question, finalization_block=_format_finalization_block(participants, final_opinions), ) _raw, parsed = await orchestrator_call( orchestrator_model_id=orchestrator_model_id, user_prompt=prompt, label="alliances", api_log=api_log, max_tokens=1024, ) if isinstance(parsed, dict) and isinstance(parsed.get("groups"), list): groups = parsed["groups"] return _normalize_groups(groups, participants) # Fallback: every participant in their own group. return [ {"stance": "(unclassified)", "members": [p.participant_id]} for p in participants ] def _normalize_groups( groups: list[dict[str, Any]], participants: list[Any], ) -> list[dict[str, Any]]: """Make sure every participant id appears in exactly one group.""" valid_ids = {p.participant_id for p in participants} seen: set[str] = set() out: list[dict[str, Any]] = [] for g in groups: members = [m for m in (g.get("members") or []) if m in valid_ids and m not in seen] seen.update(members) if members: out.append({ "stance": g.get("stance", "(unspecified)"), "members": members, }) leftovers = [pid for pid in valid_ids if pid not in seen] for pid in leftovers: out.append({"stance": "(unclassified)", "members": [pid]}) return out def _heuristic_addressed_to( participants: list[Any], speaker_name: str, message: str, ) -> str | None: """Fast path when the addressee is obvious from the text. Returns a participant_id when confidence is high; otherwise None so the orchestrator LLM classifier runs. """ if not message or not participants: return None others = [p for p in participants if p.name != speaker_name] if len(others) == 1: return others[0].participant_id text = message.strip() lower = text.lower() # "@Name" or "Name:" at start of a sentence for p in others: name = p.name.strip() if not name: continue nl = name.lower() if re.search(rf"@{re.escape(nl)}\b", lower): return p.participant_id if re.search( rf"(^|[.!?\n]\s*){re.escape(nl)}\s*[:,]", lower, ): return p.participant_id # "I agree with Name" / "Name, I think" hits: list[str] = [] for p in others: name = p.name.strip() if not name: continue nl = re.escape(name.lower()) if re.search( rf"\b(?:agree with|disagree with|respond to|reply to|" rf"building on|thank you,?)\s+{nl}\b", lower, ): hits.append(p.participant_id) elif re.search(rf"\b{nl}\b[,:]?\s+(?:you|your|i think|i believe)\b", lower): hits.append(p.participant_id) elif re.search(rf"\b{nl}\b", lower) and len(name) >= 4: hits.append(p.participant_id) unique = list(dict.fromkeys(hits)) if len(unique) == 1: return unique[0] return None async def classify_addressed_to( *, orchestrator_model_id: str, participants: list[Any], speaker_name: str, message: str, api_log: list[dict[str, Any]] | None = None, ) -> str | None: heuristic = _heuristic_addressed_to(participants, speaker_name, message) if heuristic is not None: return heuristic prompt = ADDRESSED_TO_PROMPT.format( roster_block=_format_roster_block(participants), speaker=speaker_name, message=message, ) _raw, parsed = await orchestrator_call( orchestrator_model_id=orchestrator_model_id, user_prompt=prompt, label="addressed_to", api_log=api_log, max_tokens=128, ) if isinstance(parsed, dict): target = parsed.get("addressed_to") if target and any(p.participant_id == target for p in participants): return target return None async def assess_consensus_status( *, orchestrator_model_id: str, question: str, transcript: str, alliance_groups: list[dict[str, Any]], api_log: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: prompt = CONSENSUS_STATUS_PROMPT.format( question=question, transcript=transcript, alliance_block=_format_alliance_block(alliance_groups), ) _raw, parsed = await orchestrator_call( orchestrator_model_id=orchestrator_model_id, user_prompt=prompt, label="consensus_status", api_log=api_log, max_tokens=256, ) if isinstance(parsed, dict) and parsed.get("status") in {"majority", "productive", "unproductive"}: return parsed # Default: treat as productive so we keep iterating, but give it a # bounded number of attempts via the orchestrator-call cap. return {"status": "productive", "majority_group_index": None, "rationale": ""} async def find_unaddressed_factor( *, orchestrator_model_id: str, question: str, credential_summary_block: str, transcript: str, api_log: list[dict[str, Any]] | None = None, ) -> dict[str, Any] | None: prompt = UNADDRESSED_FACTOR_PROMPT.format( question=question, credential_summary=credential_summary_block, transcript=transcript, ) _raw, parsed = await orchestrator_call( orchestrator_model_id=orchestrator_model_id, user_prompt=prompt, label="unaddressed_factor", api_log=api_log, max_tokens=512, ) if isinstance(parsed, dict) and parsed.get("factor"): return parsed return None