"""Discussion Manager agent. Runs once before each main-bot turn. Reads the discussion plan, the manager's own past verdicts, the recent conversation, and the participant's current message. Outputs a JSON verdict that tells the main bot what kind of move to make this turn. Prompt lives in `prompts/manager_prompt.md` and is loaded via `core.config_loader.MANAGER_PROMPT` at module import time. To change the manager's behaviour, edit that markdown file; do not touch this module. """ import json import re import time from dataclasses import dataclass, asdict from core.config_loader import client as _default_client, MODEL, MANAGER_PROMPT from core import plan_parsing VALID_TURN_TYPES = { "opening", "engaging-default", "transition-due", "needs-clarification", "plan-complete", } @dataclass class ManagerVerdict: """One manager classification. Serialized into the session log so the analyst can reproduce the bot's per-turn behaviour from the verdict trajectory alone.""" live_subprobe_id: int advanced_this_turn: bool turn_type: str directive: str reason: str user_msg_excerpt: str = "" # ~120-char excerpt of the participant # message that triggered this verdict. # Lets the manager see, on the NEXT turn, # what the participant actually said when # each past verdict was made (not just # the verdict's own "reason" field). question_set: list = None # NEW: ordered list of {text, status} items # for the CURRENT live sub-probe. Manager # maintains across turns. Bot reads the # first item with status="pending" and asks it. latency_ms: int = 0 def to_dict(self): d = asdict(self) if d.get("question_set") is None: d["question_set"] = [] return d _JSON_FENCE_RE = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL | re.IGNORECASE) _BARE_JSON_RE = re.compile(r"(\{[\s\S]*\})", re.DOTALL) def _parse_json(raw): """Extract the first JSON object from the model output. Tolerant of markdown fences and surrounding prose.""" raw = (raw or "").strip() if not raw: return {} m = _JSON_FENCE_RE.search(raw) if m: try: return json.loads(m.group(1)) except json.JSONDecodeError: pass m = _BARE_JSON_RE.search(raw) if m: try: return json.loads(m.group(1)) except json.JSONDecodeError: pass return {} def _format_manager_history(history): """Render the manager_history list as a multi-line block for the prompt. Each row includes: classification + the participant message excerpt + the directive issued + the reason + the question_set state at the end of that turn (so the next manager turn can read what was pending / asked and evolve correctly).""" if not history: return "(empty; this is the first manager turn of the session)" rows = [] for i, v in enumerate(history, 1): umsg = (v.get("user_msg_excerpt") or "").strip() or "(n/a)" directive = (v.get("directive") or "").strip() if len(directive) > 180: directive = directive[:180].rstrip() + "…" reason = (v.get("reason") or "").strip() or "(no reason)" qset = v.get("question_set") or [] if qset: qset_str = "\n".join( f" [{j}] ({item.get('status','?')}) {item.get('text','')}" for j, item in enumerate(qset) ) else: qset_str = " (none)" rows.append( f"- Turn {i}: live_subprobe_id={v.get('live_subprobe_id')} " f"turn_type={v.get('turn_type')} " f"advanced={v.get('advanced_this_turn')}\n" f" participant said: {umsg}\n" f" you instructed bot: {directive}\n" f" reason: {reason}\n" f" question_set at end of this turn:\n{qset_str}" ) return "\n".join(rows) def _format_recent_turns(turns, k=None): """Render participant + bot exchanges from the chat history. The history is in Gradio chatbot format (list of {role, content} dicts). The very first 'assistant' message (the case text) is skipped. `k=None` (default) renders ALL turns in the session. Pass an integer to cap to the last K turns; the prior policy was k=12 but the new manager needs the full dialogue flow to judge path-(A) gap-satisfaction vs path-(B) participant-stuck. """ if not turns: return "(no prior turns)" skip_first_assistant = True cleaned = [] for m in turns: role = m.get("role", "?") if skip_first_assistant and role == "assistant": skip_first_assistant = False continue content = m.get("content", "") if isinstance(content, list): content = " ".join( b.get("text", "") if isinstance(b, dict) else str(b) for b in content ) cleaned.append((role, str(content).strip())) sliced = cleaned if k is None else cleaned[-k:] return "\n".join(f"[{role}] {content}" for role, content in sliced) or "(no prior turns)" def classify(discussion_plan_subprobes, manager_history, recent_turns, current_msg, initial_argument="", client=None): """Run the manager LLM call. Returns a ManagerVerdict. - `discussion_plan_subprobes`: parsed sub-probe list from plan_parsing - `manager_history`: list of prior verdict dicts (most recent last) - `recent_turns`: Gradio chat history list (full); ALL turns are passed to the manager so it can judge transition paths from the complete flow - `current_msg`: the participant's message this turn - `initial_argument`: the participant's full initial argument (the essay they submitted at Phase 1). Always passed as a separate block so the manager can ALWAYS see the participant's overall stance. """ plan_block = plan_parsing.format_for_prompt(discussion_plan_subprobes) history_block = _format_manager_history(manager_history) turns_block = _format_recent_turns(recent_turns, k=None) initial_block = (initial_argument or "").strip() or "(unavailable)" user_msg = ( "INITIAL_ARGUMENT (participant's full original essay; the discussion " "plan was generated from this; ALWAYS visible regardless of how many " "turns have passed):\n" f"{initial_block}\n\n" "DISCUSSION_PLAN:\n" f"{plan_block}\n\n" "MANAGER_HISTORY:\n" f"{history_block}\n\n" "RECENT_TURNS (all messages in this session, oldest first):\n" f"{turns_block}\n\n" "CURRENT_MSG:\n" f"{current_msg}\n\n" "Output your JSON verdict now. No prose around it, JSON only." ) api = client if client is not None else _default_client t0 = time.perf_counter() response = api.responses.create( model=MODEL, reasoning={"effort": "low"}, text={"verbosity": "low"}, input=[ {"role": "developer", "content": MANAGER_PROMPT}, {"role": "user", "content": user_msg}, ], ) latency_ms = int((time.perf_counter() - t0) * 1000) raw = response.output_text or "" data = _parse_json(raw) # Validate fields with fallbacks. prev_id = manager_history[-1]["live_subprobe_id"] if manager_history else None default_id = ( prev_id if prev_id is not None else (discussion_plan_subprobes[0]["id"] if discussion_plan_subprobes else 1) ) live_id = data.get("live_subprobe_id", default_id) if not isinstance(live_id, int): try: live_id = int(live_id) except (ValueError, TypeError): live_id = default_id turn_type = data.get("turn_type", "") if turn_type not in VALID_TURN_TYPES: turn_type = "engaging-default" # advanced_this_turn must match the actual id change, regardless of what # the model reported. This is the one piece of state we hard-verify. advanced = (prev_id is not None) and (live_id != prev_id) # Directive is a natural-language paragraph (no labelled fields). directive = (data.get("directive") or "").strip() reason = (data.get("reason") or "").strip() or "(no reason given)" # question_set: ordered list of {text, status} items. Sanitize: keep # only well-formed items with a non-empty text and a valid status. raw_qset = data.get("question_set") or [] question_set = [] if isinstance(raw_qset, list): for item in raw_qset: if not isinstance(item, dict): continue text = (item.get("text") or "").strip() status = (item.get("status") or "").strip().lower() if text and status in ("asked", "pending"): question_set.append({"text": text, "status": status}) return ManagerVerdict( live_subprobe_id=live_id, advanced_this_turn=advanced, turn_type=turn_type, directive=directive, reason=reason, user_msg_excerpt=(current_msg or "").strip()[:160], question_set=question_set, latency_ms=latency_ms, )