"""CCAI orchestrator: six-phase state machine driving a multi-participant group discussion to a consensus (or to a documented failure-to-consense). Phase outline (matches the build plan): 1. Initial Opinions (independent, no peeking) 1.5. Credential Summary (built concurrently during Phase 1) 2. Critique x 2 rounds (full history visible) 3. Status Assessment (max 3 iterations of targeted follow-ups) 4. Opinion Finalization 5. Consensus Gathering (alliance-aware, addressed-to aware) 6. Closure (majority report, or unaddressed-factor probe + retry, or failure report) Two failsafes pause the loop until the user clicks "Continue": - Participant-message cap: 60, then +20. - Orchestrator-call cap: 100, then +50. Every LLM response runs through `app.utils.sanitize.strip_thinking` on its way into history, into the orchestrator's prompts, and into the summarizer. """ from __future__ import annotations import asyncio import json import logging import time import uuid from dataclasses import asdict from typing import Any, AsyncIterator from app.clients.llm_router import chat_completion from app.config import settings from app.services import context_budget, human_io from app.services.consensus import ( assess_consensus_status, classify_addressed_to, detect_alliances, find_unaddressed_factor, ) from app.services.context_budget import ( ContextSummary, DEFAULT_REPLY_BUDGET, KEEP_RECENT_MESSAGES, build_compressed_transcript_block, cap_max_tokens_for_window, context_window_for, estimate_messages_tokens, replace_embedded_transcript, run_summarize, select_summarizer_model_id, should_summarize, ) from app.services.credential import ( assemble_credential_summary_list, build_credential_for_participant, credentials_to_block, ) from app.services.json_calls import orchestrator_call from app.services.orchestrator_speed import ( _AiTurnResult, _AiTurnSpec, compact_transcript_for_orchestrator, orchestrator_fast_model_id, run_initial_opinions_roster, run_roster_ai_turns_parallel, ) from app.services.resilience import run_resilient_turn from app.services.models import ( DEFAULT_MAX_PARTICIPANTS, MAX_MAX_PARTICIPANTS, MIN_MAX_PARTICIPANTS, Participant, Phase, Session, ) from app.services.prompts import ( CONSENSUS_ALLIED_PROMPT, CONSENSUS_SOLO_PROMPT, CONTRIBUTION_SUMMARY_PROMPT, CRITIQUE_PROMPT, FINALIZATION_PROMPT, INITIAL_OPINION_PROMPT, MAJORITY_REPORT_PROMPT, NO_CONSENSUS_REPORT_PROMPT, NO_REASONING_DIRECTIVE, PARTICIPANT_BASE_DIRECTIVE, STATUS_ASSESSMENT_PROMPT, TARGETED_FOLLOWUP_PROMPT, TARGETED_FOLLOWUP_FROM_PARTICIPANT_PROMPT, ) from app.utils.sanitize import strip_thinking LOG = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Session registry # --------------------------------------------------------------------------- _sessions: dict[str, Session] = {} def get_session(sid: str) -> Session | None: return _sessions.get(sid) def create_session() -> Session: s = Session() _sessions[s.session_id] = s return s # --------------------------------------------------------------------------- # SSE helpers # --------------------------------------------------------------------------- def _sse(event: str, data: dict[str, Any]) -> str: return f"event: {event}\ndata: {json.dumps(data)}\n\n" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _active_participants(session: Session) -> list[Participant]: return [p for p in session.participants if p.enabled] def _orchestrator_model_id(session: Session) -> str: return session.orchestrator_model_id or settings.orchestrator_model def _summarizer_model_id(session: Session) -> str: return select_summarizer_model_id( session.summarizer_model_id, session.orchestrator_model_id, ) def _format_history( messages: list[dict[str, Any]], *, include_orchestrator: bool = True, ) -> str: lines: list[str] = [] for m in messages: if m.get("role") == "orchestrator" and not include_orchestrator: continue speaker = m.get("speaker_name") or m.get("speaker_id") or "(anon)" if m.get("role") == "orchestrator": speaker = "Orchestrator" lines.append(f"{speaker}: {m.get('text', '')}") return "\n".join(lines) def _participant_roster_string( speaker: Participant, participants: list[Participant], ) -> str: others = [p.name for p in participants if p.participant_id != speaker.participant_id] return ", ".join(others) if others else "(no other participants)" # Per-prompt cap on how much of a long pending-thread message we quote back # to the speaker. The full message is still in the transcript above; this # block is a "what specifically is owed to you" reminder, not a re-render. _PENDING_TRUNCATE_CHARS = 600 def _pending_addressed_for( session: Session, speaker: Participant, ) -> list[tuple[str, str, str]]: """Return (asker_id, asker_name, message_text) for participant messages that addressed `speaker` since `speaker`'s last own-message turn. Used to (1) inject an "Open threads directed at you" block into per- speaker prompt templates and (2) populate the `replying_to` field on the speaker's outgoing message so the frontend can render a "Replying to X, Y" pill above the bubble. """ last_own_idx = -1 for i, m in enumerate(session.messages): if ( m.get("role") == "participant" and m.get("speaker_id") == speaker.participant_id ): last_own_idx = i pending: list[tuple[str, str, str]] = [] for m in session.messages[last_own_idx + 1:]: if m.get("role") != "participant": continue if m.get("addressed_to") != speaker.participant_id: continue asker_id = m.get("speaker_id") or "unknown" asker_name = m.get("speaker_name") or "another participant" text = (m.get("text") or "").strip() if not text: continue pending.append((asker_id, asker_name, text)) return pending def _format_pending_block( pending: list[tuple[str, str, str]], ) -> str: """Render the open-threads section that gets interpolated into per- speaker prompt templates. Always non-empty so templates read naturally; we explicitly print "(none)" when there are no open threads. """ if not pending: return ( "Open threads directed at you since your last turn: (none).\n\n" ) lines = ["Open threads directed at you since your last turn:"] for _asker_id, asker_name, text in pending: snippet = text if len(snippet) > _PENDING_TRUNCATE_CHARS: snippet = snippet[:_PENDING_TRUNCATE_CHARS].rstrip() + "..." lines.append(f' - {asker_name} said to you: "{snippet}"') return "\n".join(lines) + "\n\n" def _replying_to_ids(pending: list[tuple[str, str, str]]) -> list[str]: """Stable, de-duplicated list of asker participant_ids extracted from a pending-thread list. Used to populate the message's `replying_to` field so the frontend can render the "Replying to X, Y" pill. """ seen: set[str] = set() out: list[str] = [] for asker_id, _name, _text in pending: if asker_id in seen: continue seen.add(asker_id) out.append(asker_id) return out # --------------------------------------------------------------------------- # Failsafe checks # --------------------------------------------------------------------------- def _participant_msg_cap_hit(session: Session) -> bool: return session.total_participant_messages >= session.participant_message_cap def _orchestrator_cap_hit(session: Session) -> bool: return session.orchestrator_call_count >= session.orchestrator_call_cap def _bump_orchestrator_count(session: Session) -> None: session.orchestrator_call_count += 1 async def _wait_for_continue( session: Session, reason: str, ) -> AsyncIterator[str]: """Pause the state machine until the user clicks Continue. Increment values come from `session.limits`, which the user can tune via the settings menu. Defaults match the historical PARTICIPANT_MESSAGE_PAUSE_INC / ORCHESTRATOR_CALL_PAUSE_INC. """ session.paused_for_continue = True session.pause_reason = reason if reason == "messages": bump_inc = session.limits.participant_message_pause_inc msg = ( f"Conversation paused after {session.total_participant_messages} " "participant messages. Click Continue to allow another " f"{bump_inc} messages." ) evt = "failsafe_pause" else: bump_inc = session.limits.orchestrator_call_pause_inc msg = ( f"Conversation paused after {session.orchestrator_call_count} " "orchestrator calls. Click Continue to allow another " f"{bump_inc} orchestrator calls." ) evt = "orchestrator_cap_pause" yield _sse(evt, { "reason": reason, "message": msg, "participant_messages": session.total_participant_messages, "orchestrator_calls": session.orchestrator_call_count, }) # Block until pending_continue is flipped by the API layer. while session.paused_for_continue and not session.pending_continue: await asyncio.sleep(0.25) session.pending_continue = False session.paused_for_continue = False if reason == "messages": session.participant_message_cap += bump_inc else: session.orchestrator_call_cap += bump_inc session.pause_reason = None for chunk in _orchestrator_banner_sse(session, "Resuming conversation..."): yield chunk # --------------------------------------------------------------------------- # Human-participant turn # --------------------------------------------------------------------------- async def _wait_for_human_text( session: Session, participant: Participant, *, phase: Phase, addressed_to: str | None = None, asker_id: str | None = None, asker_name: str | None = None, prompt_context: str | None = None, ) -> AsyncIterator[str]: """Pause the orchestrator until the human types a response (or skips). Yields a `human_turn_needed` SSE event with the metadata the frontend needs to render the input slot and the lower-screen "waiting for your input" cue, then polls the human_io slot until the API layer's POST /human-response sets it, then yields a `human_turn_cleared` event so the frontend can dismiss the cue. The actual response text + skipped flag are NOT returned from this generator (async gens can't return values cleanly). The caller reads them via `human_io.slot_for(session.session_id)` AFTER the iteration completes: slot.response_text (str) slot.skipped (bool) slot.started_at (float) - subtract from now() for elapsed slot.pending_snapshot (list) - pending threads at turn-start Caller is expected to reset_slot after consuming the result. """ started = time.time() pending = _pending_addressed_for(session, participant) slot = human_io.slot_for(session.session_id) slot.event.clear() slot.response_text = "" slot.skipped = False slot.started_at = started slot.pending_snapshot = pending awaiting = { "speaker_id": participant.participant_id, "speaker_name": participant.name, "phase": phase.value, "addressed_to": addressed_to, "asker_id": asker_id, "asker_name": asker_name, "prompt_context": prompt_context, } session.awaiting_human = awaiting session.paused_for_continue = True session.pause_reason = "human_turn" yield _sse("human_turn_needed", awaiting) try: # Poll with the same 0.25s cadence as _wait_for_continue so # SSE-stream cancellation propagates promptly to the user # clicking Stop. while not slot.event.is_set(): await asyncio.sleep(0.25) finally: session.paused_for_continue = False session.pause_reason = None session.awaiting_human = None yield _sse("human_turn_cleared", { "speaker_id": participant.participant_id, }) async def _do_human_turn( session: Session, participant: Participant, *, phase: Phase, actives: list[Participant], addressed_to_target: str | None = None, asker_id: str | None = None, asker_name: str | None = None, prompt_context: str | None = None, classify_addressed: bool = False, track_initial_opinion: bool = False, track_final_opinion: bool = False, addressed_state: dict[str, Any] | None = None, ) -> AsyncIterator[str]: """End-to-end human turn: emit human_turn_needed, await response, emit human_turn_cleared, then either record a skip note or append a participant message (with addressed-to classification when asked). Yields SSE chunks throughout, then runs the failsafe-pause check. `addressed_state`, when provided, is a caller-owned dict that gets mutated with {"last_addressed": } after the turn so the consensus phase can update its routing variable without a return value sneaking out of the generator. """ async for chunk in _wait_for_human_text( session, participant, phase=phase, addressed_to=addressed_to_target, asker_id=asker_id, asker_name=asker_name, prompt_context=prompt_context, ): yield chunk slot = human_io.slot_for(session.session_id) text = (slot.response_text or "").strip() skipped = slot.skipped elapsed = max(0.0, time.time() - slot.started_at) pending = list(slot.pending_snapshot or []) human_io.reset_slot(session.session_id) if skipped or not text: note = _add_orchestrator_message( session, f"{participant.name} declined to comment this turn.", kind="status", ) yield _sse("orchestrator", _msg_payload(note)) if addressed_state is not None: addressed_state["last_addressed"] = None return addressed: str | None = None if classify_addressed: addressed = await classify_addressed_to( orchestrator_model_id=orchestrator_fast_model_id(session), participants=actives, speaker_name=participant.name, message=text, api_log=session.api_log, ) _bump_orchestrator_count(session) msg = _add_participant_message( session, participant, text, phase=phase, elapsed=elapsed, addressed_to=addressed, replying_to=_replying_to_ids(pending), ) if track_initial_opinion: session.initial_opinions[participant.participant_id] = text if track_final_opinion: session.final_opinions[participant.participant_id] = text if addressed_state is not None: addressed_state["last_addressed"] = addressed yield _sse("message", _msg_payload(msg)) if _participant_msg_cap_hit(session): async for chunk in _wait_for_continue(session, "messages"): yield chunk if _orchestrator_cap_hit(session): async for chunk in _wait_for_continue(session, "orchestrator"): yield chunk # --------------------------------------------------------------------------- # Participant turn (with context budgeting + summarize-on-demand) # --------------------------------------------------------------------------- async def _maybe_summarize_for_participant( session: Session, participant: Participant, api_messages: list[dict[str, Any]], ) -> None: """If this participant's input estimate exceeds the threshold, run a summarize call against the configured summarizer model and update `participant.summary` in place.""" needs_sum, _trim, _budget = should_summarize( participant.model_id, api_messages, participant.summary, ) if not needs_sum: return # Build a transcript that excludes orchestrator status banners (those # don't add information value to a summary) but keeps everything the # participant has said and heard. summarizable_msgs = [ m for m in session.messages if m.get("role") != "orchestrator_status" ] if not summarizable_msgs: return transcript = _format_history(summarizable_msgs, include_orchestrator=False) if not transcript.strip(): return summarizer_id = _summarizer_model_id(session) summary_text = await run_summarize(summarizer_id, transcript) # The summarizer counts as an orchestrator-side call for cap purposes. session.orchestrator_call_count += 1 if summary_text: participant.summary.summary_text = summary_text participant.summary.summarized_through_idx = len(session.messages) - 1 async def _call_participant( *, session: Session, participant: Participant, user_prompt: str, label: str, max_tokens: int = 600, timeout: float = 45.0, stream_events: list[str] | None = None, stream_message_id: str | None = None, ) -> tuple[str, float, bool, str]: """Run one participant turn. Returns ``(text, elapsed_seconds, ok, error_kind)``. ``error_kind`` is ``""`` on success. On failure it's one of: * ``"transient"`` — HTTP 5xx, 429, timeout, connection error. The same model is worth retrying. * ``"permanent"`` — auth, invalid request, content filter, model gone. Retrying the same model won't help. * ``"empty"`` — call returned a 200 with an empty body. Treated as transient by the resilience layer (retry once before substituting). * ``"unknown"`` — orchestrator-side exception we couldn't classify. The state-machine handles auto-disable on repeated failure; the resilience layer (`services.resilience.run_resilient_turn`) handles in-turn retry / alternate / substitution under speed-priority. """ others = _participant_roster_string(participant, _active_participants(session)) base_directive = PARTICIPANT_BASE_DIRECTIVE.format( n_participants=len(_active_participants(session)), other_participants=others, ) system_text = ( f"{participant.role_prompt}\n\n{base_directive}\n\n{NO_REASONING_DIRECTIVE}" ) api_messages: list[dict[str, Any]] = [ {"role": "system", "content": system_text}, {"role": "user", "content": user_prompt}, ] await _maybe_summarize_for_participant(session, participant, api_messages) needs_sum, needs_trim, input_budget = should_summarize( participant.model_id, api_messages, participant.summary, ) # CCAI embeds the transcript inside the user prompt. When over budget, # swap that block for summary + recent tail (AskJerry pattern). if needs_sum or needs_trim: recent_transcript = _format_history( session.messages[-KEEP_RECENT_MESSAGES:], include_orchestrator=False, ) compressed_block = build_compressed_transcript_block( participant.summary, recent_transcript, ) if compressed_block: new_prompt = replace_embedded_transcript(user_prompt, compressed_block) user_prompt = new_prompt api_messages[1] = {"role": "user", "content": user_prompt} elif len(user_prompt) > input_budget * 4: # Last-resort: keep prompt head + tail if no transcript header matched. keep = max(512, input_budget * 2) user_prompt = ( user_prompt[:keep] + "\n\n[…middle truncated for context…]\n\n" + user_prompt[-keep:] ) api_messages[1] = {"role": "user", "content": user_prompt} max_tokens = cap_max_tokens_for_window( participant.model_id, api_messages, max_tokens, ) resolved = { "model_id": participant.model_id, "base_url": participant.base_url, "api_key": participant.api_key, "is_neon": participant.is_neon, "hana_model_id": participant.hana_model_id, "persona_name": participant.persona_name, "neon_direct_vllm": participant.neon_direct_vllm, "vllm_base_url": participant.vllm_base_url, "vllm_api_key": participant.vllm_api_key, } log_entry: dict[str, Any] = { "timestamp": time.time(), "label": f"participant:{participant.participant_id}:{label}", "model": participant.model_id, "request": {"messages": api_messages, "max_tokens": max_tokens}, } msg_id = stream_message_id or str(uuid.uuid4()) on_text_delta_cb = None if stream_events is not None: stream_events.append(_sse("message_stream_start", { "message_id": msg_id, "speaker_id": participant.participant_id, "speaker_name": participant.name, "kind": participant.kind, "phase": session.phase.value, "model_id": participant.model_id, "model_display": participant.display_name, })) def on_text_delta_cb(piece: str) -> None: stream_events.append(_sse("message_delta", { "message_id": msg_id, "delta": piece, })) try: result = await chat_completion( resolved=resolved, messages=api_messages, temperature=0.7, max_tokens=max_tokens, timeout=timeout, on_text_delta=on_text_delta_cb, ) except Exception as exc: LOG.exception("Participant %s call failed: %s", participant.participant_id, exc) log_entry["response"] = {"error": str(exc)} session.api_log.append(log_entry) participant.consecutive_failures += 1 return "", 0.0, False, "unknown" log_entry["response"] = result session.api_log.append(log_entry) if result.get("error"): participant.consecutive_failures += 1 return ( "", result.get("elapsed_seconds", 0), False, result.get("error_kind") or "permanent", ) text = strip_thinking(result.get("response", "")) elapsed = float(result.get("elapsed_seconds", 0) or 0) if not text.strip(): # 200 OK but the model returned nothing usable. Worth one # retry / substitute attempt before we surface participant_error. participant.consecutive_failures += 1 return "", elapsed, False, "empty" participant.consecutive_failures = 0 return text, elapsed, True, "" def _add_participant_message( session: Session, participant: Participant, text: str, *, phase: Phase, elapsed: float, addressed_to: str | None = None, replying_to: list[str] | None = None, message_id: str | None = None, ) -> dict[str, Any]: msg = { "message_id": message_id or str(uuid.uuid4()), "speaker_id": participant.participant_id, "speaker_name": participant.name, "role": "participant", # `kind` lets the frontend distinguish a human participant's # message ("human") from LLM messages ("neon" | "extra" | # "expert") so the green left-edge accent can be applied # independently of the rotating color palette. "kind": participant.kind, "text": text, "phase": phase.value, "timestamp": time.time(), "elapsed_seconds": round(elapsed, 2), "addressed_to": addressed_to, # `replying_to` mirrors the pending-thread list we showed the # speaker at turn-start: ordered, de-duplicated participant_ids of # everyone whose questions this turn was supposed to address. # Empty list when there were no open threads. The frontend renders # this as a "Replying to X, Y" pill above the bubble. "replying_to": list(replying_to) if replying_to else [], "model_id": participant.model_id, "model_display": participant.display_name, } session.messages.append(msg) session.total_participant_messages += 1 return msg def _add_orchestrator_message( session: Session, text: str, *, kind: str, extra: dict[str, Any] | None = None, ) -> dict[str, Any]: msg = { "speaker_id": "orchestrator", "speaker_name": "Orchestrator", "role": "orchestrator", "kind": kind, # "status" | "factor" | "majority_report" | "no_consensus_report" "text": text, "phase": session.phase.value, "timestamp": time.time(), } if extra: msg.update(extra) session.messages.append(msg) return msg def _msg_payload(msg: dict[str, Any]) -> dict[str, Any]: """Public payload for a message event over SSE.""" return msg def _orchestrator_banner_sse( session: Session, text: str, *, kind: str = "status", extra: dict[str, Any] | None = None, ) -> list[str]: """Append an orchestrator line to the transcript and emit chat + status SSE.""" msg = _add_orchestrator_message(session, text, kind=kind, extra=extra) return [ _sse("orchestrator", _msg_payload(msg)), _sse("status", {"message": text}), ] def _participant_turn_failure_sse( session: Session, participant: Participant, ) -> list[str]: """Emit participant_error and auto-disable banner when threshold hit.""" out = [ _sse("participant_error", { "participant_id": participant.participant_id, "name": participant.name, "phase": session.phase.value, }), ] if participant.consecutive_failures >= session.limits.auto_disable_failures: participant.enabled = False out.extend(_orchestrator_banner_sse( session, f"{participant.name} auto-disabled after " f"{session.limits.auto_disable_failures} consecutive failures.", )) return out # --------------------------------------------------------------------------- # Credential summary (concurrent with Phase 1) # --------------------------------------------------------------------------- async def _credential_build_runner( session: Session, participant: Participant, initial_opinion: str, ) -> None: """Background task: one participant's credential row.""" try: cred = await build_credential_for_participant( orchestrator_model_id=_orchestrator_model_id(session), question=session.question, participant=participant, initial_opinion=initial_opinion, api_log=session.api_log, ) session.credential_entries_by_pid[participant.participant_id] = cred session.credential_model_by_pid[participant.participant_id] = ( participant.model_id ) _bump_orchestrator_count(session) except Exception as exc: LOG.exception( "Credential build failed for %s: %s", participant.participant_id, exc, ) def _schedule_phase1_credential_build( session: Session, participant: Participant, initial_opinion: str, ) -> None: """Start (or restart) a background credential build for one AI participant.""" if participant.kind == "human": return if not (initial_opinion or "").strip(): return pid = participant.participant_id existing = session.credential_build_tasks.get(pid) if existing is not None and not existing.done(): existing.cancel() session.credential_build_tasks[pid] = asyncio.create_task( _credential_build_runner(session, participant, initial_opinion), name=f"credential:{pid}", ) def _sync_credential_summary_from_entries(session: Session) -> None: session.credential_summary = assemble_credential_summary_list( participants=_active_participants(session), credential_entries_by_pid=session.credential_entries_by_pid, human_credential=session.human_credential, ) async def _await_phase1_credential_tasks(session: Session) -> None: """Wait for any in-flight per-participant credential builds.""" tasks = [ t for t in session.credential_build_tasks.values() if t is not None and not t.done() ] if tasks: await asyncio.gather(*tasks, return_exceptions=True) _sync_credential_summary_from_entries(session) async def _rebuild_participant_credential_on_model_change( session: Session, participant: Participant, ) -> bool: """Rebuild one credential row when the backing LLM model changes.""" if participant.kind == "human": return False pid = participant.participant_id opinion = (session.initial_opinions or {}).get(pid, "") if not opinion.strip(): return False prior_model = session.credential_model_by_pid.get(pid) if not prior_model or prior_model == participant.model_id: return False cred = await build_credential_for_participant( orchestrator_model_id=_orchestrator_model_id(session), question=session.question, participant=participant, initial_opinion=opinion, api_log=session.api_log, ) _bump_orchestrator_count(session) session.credential_entries_by_pid[pid] = cred session.credential_model_by_pid[pid] = participant.model_id _sync_credential_summary_from_entries(session) return True # --------------------------------------------------------------------------- # Phase implementations # --------------------------------------------------------------------------- async def _phase_initial_opinions(session: Session) -> AsyncIterator[str]: session.phase = Phase.INITIAL_OPINIONS for chunk in _orchestrator_banner_sse( session, "Phase 1: collecting independent first opinions...", ): yield chunk actives = _active_participants(session) async def _human_initial(p: Participant) -> AsyncIterator[str]: async for chunk in _do_human_turn( session, p, phase=session.phase, actives=actives, track_initial_opinion=True, prompt_context=( "Share your initial opinion on the question. " "You're speaking BEFORE seeing the other participants." ), ): yield chunk async def _post_initial(result: _AiTurnResult) -> dict[str, Any]: speaker = result.turn.speaker session.initial_opinions[speaker.participant_id] = result.turn.text _schedule_phase1_credential_build( session, speaker, result.turn.text, ) return {} def _build_initial_spec(p: Participant) -> _AiTurnSpec | None: if p.kind == "human": return None return _AiTurnSpec( participant=p, user_prompt=INITIAL_OPINION_PROMPT.format(question=session.question), label="initial_opinion", max_tokens=700, ) async for chunk in run_initial_opinions_roster( session, actives, build_spec=_build_initial_spec, call_participant=_call_participant, on_human_turn=_human_initial, post_process=_post_initial, ): yield chunk # Credential rows were built in parallel as each opinion landed; # only wait here if any background task is still finishing. await _await_phase1_credential_tasks(session) yield _sse("credentials_updated", { "stage": "built", "credentials": session.credential_summary, }) _CRITIQUE_PHASES = { 1: Phase.CRITIQUE_ROUND_1, 2: Phase.CRITIQUE_ROUND_2, 3: Phase.CRITIQUE_ROUND_3, 4: Phase.CRITIQUE_ROUND_4, } def _critique_phase_for(round_number: int) -> Phase: """Map a critique round number to the matching Phase enum value. Falls back to CRITIQUE_ROUND_2 for unknown numbers - the API layer clamps `critique_rounds` to the bounds, so this fallback is purely defensive.""" return _CRITIQUE_PHASES.get(round_number, Phase.CRITIQUE_ROUND_2) async def _phase_critique(session: Session, round_number: int) -> AsyncIterator[str]: session.phase = _critique_phase_for(round_number) round_total = session.limits.critique_rounds for chunk in _orchestrator_banner_sse( session, f"Phase 2: critique round {round_number} of {round_total}...", ): yield chunk cred_block = credentials_to_block(session.credential_summary) actives = _active_participants(session) # Freeze transcript + pending threads at round start so parallel # turns see the same context and reply metadata stays consistent. transcript_snapshot = _format_history(session.messages) pending_snapshot = { p.participant_id: _pending_addressed_for(session, p) for p in actives if p.kind != "human" } async def _human_critique(p: Participant) -> AsyncIterator[str]: async for chunk in _do_human_turn( session, p, phase=session.phase, actives=actives, classify_addressed=True, prompt_context=( f"Critique round {round_number} of {round_total}. " "Push back on, agree with, or build on what others " "have said. Address other participants by name." ), ): yield chunk def _build_critique_spec(p: Participant) -> _AiTurnSpec | None: if p.kind == "human": return None pending = pending_snapshot.get(p.participant_id, []) pending_block = _format_pending_block(pending) prompt = CRITIQUE_PROMPT.format( round_number=round_number, round_total=round_total, question=session.question, credential_summary=cred_block, transcript=transcript_snapshot, pending_block=pending_block, ) return _AiTurnSpec( participant=p, user_prompt=prompt, label=f"critique_round_{round_number}", max_tokens=700, ) async def _post_critique(result: _AiTurnResult) -> dict[str, Any]: speaker = result.turn.speaker addressed = await classify_addressed_to( orchestrator_model_id=orchestrator_fast_model_id(session), participants=_active_participants(session), speaker_name=speaker.name, message=result.turn.text, api_log=session.api_log, ) _bump_orchestrator_count(session) return { "addressed_to": addressed, "replying_to": _replying_to_ids(result.pending), } async for chunk in run_roster_ai_turns_parallel( session, actives, phase=session.phase, build_spec=_build_critique_spec, call_participant=_call_participant, on_human_turn=_human_critique, post_process=_post_critique, ): yield chunk async def _phase_status_assessment(session: Session) -> AsyncIterator[str]: session.phase = Phase.STATUS_ASSESSMENT for chunk in _orchestrator_banner_sse( session, "Phase 3: assessing whether more questions are needed...", ): yield chunk cred_block = credentials_to_block(session.credential_summary) for iteration in range(session.limits.status_assessment_max): session.status_assessment_iterations = iteration + 1 transcript = await compact_transcript_for_orchestrator( session, orchestrator_model_id=orchestrator_fast_model_id(session), ) prompt = STATUS_ASSESSMENT_PROMPT.format( question=session.question, credential_summary=cred_block, transcript=transcript, ) _raw, parsed = await orchestrator_call( orchestrator_model_id=_orchestrator_model_id(session), user_prompt=prompt, label=f"status_assessment_{iteration + 1}", api_log=session.api_log, max_tokens=512, ) _bump_orchestrator_count(session) opinions_solidified = bool( isinstance(parsed, dict) and parsed.get("opinions_solidified") ) open_qs: list[dict[str, Any]] = [] if isinstance(parsed, dict): open_qs = parsed.get("open_questions") or [] if opinions_solidified or not open_qs: msg = _add_orchestrator_message( session, "Opinions appear solidified - moving to finalization.", kind="status", ) yield _sse("orchestrator", _msg_payload(msg)) return # Otherwise run targeted follow-ups active_ids = {p.participant_id for p in _active_participants(session)} for oq in open_qs: pid = oq.get("participant_id") question_text = (oq.get("question") or "").strip() if not pid or pid not in active_ids or not question_text: continue target = next(p for p in session.participants if p.participant_id == pid) # Decide synthesized vs verbatim. Source of truth is # asker_participant_id - if it resolves to a real, *different*, # active participant we treat the question as verbatim from # them. Otherwise we treat it as orchestrator-synthesized. asker_id = (oq.get("asker_participant_id") or "").strip() or None asker: Participant | None = None if asker_id and asker_id in active_ids and asker_id != pid: asker = next( p for p in session.participants if p.participant_id == asker_id ) if asker is not None: announce = ( f"{asker.name} raised a question earlier, to " f"{target.name}: \"{question_text}\"" ) else: announce = ( f"I have a follow-up question for {target.name}: " f"\"{question_text}\"" ) announce_msg = _add_orchestrator_message(session, announce, kind="status") yield _sse("orchestrator", _msg_payload(announce_msg)) if target.kind == "human": async for chunk in _do_human_turn( session, target, phase=session.phase, actives=_active_participants(session), asker_id=(asker.participant_id if asker else None), asker_name=(asker.name if asker else None), prompt_context=question_text, ): yield chunk continue transcript = _format_history(session.messages) if asker is not None: prompt2 = TARGETED_FOLLOWUP_FROM_PARTICIPANT_PROMPT.format( transcript=transcript, credential_summary=cred_block, targeted_question=question_text, asker_name=asker.name, ) else: prompt2 = TARGETED_FOLLOWUP_PROMPT.format( transcript=transcript, credential_summary=cred_block, targeted_question=question_text, ) stream_events: list[str] = [] stream_msg_id = str(uuid.uuid4()) turn = await run_resilient_turn( session=session, participant=target, user_prompt=prompt2, label="targeted_followup", max_tokens=600, call_participant=_call_participant, stream_events=stream_events, stream_message_id=stream_msg_id, ) for ev in stream_events: yield ev for ev in turn.sse_events: yield ev if not turn.ok: for chunk in _participant_turn_failure_sse(session, target): yield chunk continue speaker = turn.speaker text, elapsed = turn.text, turn.elapsed # When the orchestrator is relaying a verbatim question from # another participant, mark this turn as replying to that # asker so the frontend can render the "Replying to X" pill. replying_to = [asker.participant_id] if asker is not None else [] msg = _add_participant_message( session, speaker, text, phase=session.phase, elapsed=elapsed, replying_to=replying_to, message_id=stream_msg_id, ) yield _sse("message", _msg_payload(msg)) if _participant_msg_cap_hit(session): async for chunk in _wait_for_continue(session, "messages"): yield chunk if _orchestrator_cap_hit(session): async for chunk in _wait_for_continue(session, "orchestrator"): yield chunk msg = _add_orchestrator_message( session, "Moving to finalization.", kind="status", ) yield _sse("orchestrator", _msg_payload(msg)) async def _phase_finalization(session: Session) -> AsyncIterator[str]: session.phase = Phase.FINALIZATION for chunk in _orchestrator_banner_sse(session, "Phase 4: opinion finalization..."): yield chunk cred_block = credentials_to_block(session.credential_summary) actives = _active_participants(session) transcript_snapshot = _format_history(session.messages) pending_snapshot = { p.participant_id: _pending_addressed_for(session, p) for p in actives if p.kind != "human" } async def _human_final(p: Participant) -> AsyncIterator[str]: async for chunk in _do_human_turn( session, p, phase=session.phase, actives=actives, track_final_opinion=True, prompt_context=( "Phase 4: state your final opinion on the question, " "incorporating whatever you've learned in the discussion." ), ): yield chunk def _build_final_spec(p: Participant) -> _AiTurnSpec | None: if p.kind == "human": return None pending = pending_snapshot.get(p.participant_id, []) pending_block = _format_pending_block(pending) prompt = FINALIZATION_PROMPT.format( question=session.question, credential_summary=cred_block, transcript=transcript_snapshot, pending_block=pending_block, ) return _AiTurnSpec( participant=p, user_prompt=prompt, label="finalization", max_tokens=600, ) async def _post_final(result: _AiTurnResult) -> dict[str, Any]: session.final_opinions[result.participant.participant_id] = result.turn.text return {"replying_to": _replying_to_ids(result.pending)} async for chunk in run_roster_ai_turns_parallel( session, actives, phase=session.phase, build_spec=_build_final_spec, call_participant=_call_participant, on_human_turn=_human_final, post_process=_post_final, ): yield chunk async def _phase_consensus(session: Session) -> AsyncIterator[str]: session.phase = Phase.CONSENSUS for chunk in _orchestrator_banner_sse(session, "Phase 5: consensus gathering..."): yield chunk cred_block = credentials_to_block(session.credential_summary) actives = _active_participants(session) # Initial alliance detection from the finalization-phase opinions groups = await detect_alliances( orchestrator_model_id=_orchestrator_model_id(session), question=session.question, participants=actives, final_opinions=session.final_opinions, api_log=session.api_log, ) _bump_orchestrator_count(session) session.alliance_groups = groups # Render alliance group members using the same display names shown # in the sidebar (Participant.name), not raw participant_ids. id_to_name = {p.participant_id: p.name for p in actives} alliance_prefix = ( "Updated alliance groups detected: " if session.consensus_attempts > 0 else "Alliance groups detected: " ) announce = alliance_prefix + "; ".join( f"\"{g.get('stance', '')}\" -> [" + ", ".join( id_to_name.get(m, m) for m in (g.get("members") or []) ) + "]" for g in groups ) msg = _add_orchestrator_message(session, announce, kind="status") yield _sse("orchestrator", _msg_payload(msg)) # Round-robin among active participants, but yield to the addressed-to # target whenever the previous message named one explicitly. To keep # two participants from monopolizing the floor with an A->B->A->B # loop, we cap consecutive addressed-to routings at the configured # `dyad_cap`. After that many in a row, we force a round-robin pick. queue: list[Participant] = list(actives) last_addressed: str | None = None dyad_run: int = 0 dyad_cap = session.limits.dyad_cap # Hard backstop on this phase: if we make a lot of consensus turns # without resolving, exit and let closure handle it. The orchestrator- # call cap will usually hit before this, but it's a clean upper bound. max_consensus_turns = ( session.limits.consensus_turns_per_participant * len(actives) ) consensus_turns = 0 while consensus_turns < max_consensus_turns: consensus_turns += 1 actives = _active_participants(session) if len(actives) < 2: break queue = [p for p in queue if p.enabled] # Pick speaker. Prefer the addressed-to target (dyadic exchange) # only while we're under the consecutive-routing cap. Once the # cap is hit, force a round-robin pick so a third voice can join. if last_addressed and dyad_run < dyad_cap: speaker = next( (p for p in actives if p.participant_id == last_addressed), None, ) if speaker is None: speaker = queue[0] if queue else actives[0] dyad_run = 0 else: queue = [p for p in queue if p.participant_id != speaker.participant_id] dyad_run += 1 last_addressed = None else: if not queue: queue = list(actives) speaker = queue.pop(0) dyad_run = 0 last_addressed = None if speaker.kind == "human": addressed_state: dict[str, Any] = {} async for chunk in _do_human_turn( session, speaker, phase=session.phase, actives=actives, classify_addressed=True, addressed_state=addressed_state, prompt_context=( "Phase 5: weigh in on whether you agree, disagree, " "or want to refine. Address other participants by " "name when you're responding to something specific " "they said." ), ): yield chunk # Propagate addressed_to so dyad routing also works when the # last speaker was the human. last_addressed = addressed_state.get("last_addressed") # Status check every full round (every len(actives) turns). # Replicated here because the LLM-path code below also does # it, and we need it on the human path too. if consensus_turns % max(1, len(actives)) == 0: terminal = await _consensus_status_terminal_sse(session, actives) if terminal: yield terminal return continue # Decide allied vs solo prompt speaker_group, other_groups = _find_speaker_group(speaker, session.alliance_groups) prompt = _build_consensus_prompt( session, speaker, speaker_group, other_groups, actives, cred_block, ) # Snapshot pending threads BEFORE the call so the outgoing # message records who this turn was supposed to be replying to. pending = _pending_addressed_for(session, speaker) stream_events: list[str] = [] stream_msg_id = str(uuid.uuid4()) turn = await run_resilient_turn( session=session, participant=speaker, user_prompt=prompt, label="consensus", max_tokens=700, call_participant=_call_participant, stream_events=stream_events, stream_message_id=stream_msg_id, ) for ev in stream_events: yield ev for ev in turn.sse_events: yield ev if not turn.ok: for chunk in _participant_turn_failure_sse(session, speaker): yield chunk continue # Consensus phase doesn't swap participants, only LLMs behind # them, so turn.speaker is the same instance as `speaker`. Use # turn.speaker to stay consistent with other phases. speaker = turn.speaker text, elapsed = turn.text, turn.elapsed addressed = await classify_addressed_to( orchestrator_model_id=orchestrator_fast_model_id(session), participants=actives, speaker_name=speaker.name, message=text, api_log=session.api_log, ) _bump_orchestrator_count(session) last_addressed = addressed msg = _add_participant_message( session, speaker, text, phase=session.phase, elapsed=elapsed, addressed_to=addressed, replying_to=_replying_to_ids(pending), message_id=stream_msg_id, ) yield _sse("message", _msg_payload(msg)) if _participant_msg_cap_hit(session): async for chunk in _wait_for_continue(session, "messages"): yield chunk if _orchestrator_cap_hit(session): async for chunk in _wait_for_continue(session, "orchestrator"): yield chunk # Status check every full round (every len(actives) turns) if consensus_turns % max(1, len(actives)) == 0: terminal = await _consensus_status_terminal_sse(session, actives) if terminal: yield terminal return async def _consensus_status_terminal_sse( session: Session, actives: list[Participant], ) -> str | None: """Run a consensus status check. Returns an SSE chunk when the phase should end (majority or unproductive), else None.""" transcript = await compact_transcript_for_orchestrator( session, orchestrator_model_id=orchestrator_fast_model_id(session), ) status = await assess_consensus_status( orchestrator_model_id=orchestrator_fast_model_id(session), question=session.question, transcript=transcript, alliance_groups=session.alliance_groups, api_log=session.api_log, ) _bump_orchestrator_count(session) if status.get("status") == "majority": session.alliance_groups = await _refresh_alliance_groups(session, actives) msg = _add_orchestrator_message( session, f"Majority reached. {status.get('rationale', '')}".strip(), kind="status", ) return _sse("orchestrator", _msg_payload(msg)) if status.get("status") == "unproductive": msg = _add_orchestrator_message( session, f"Conversation no longer productive. {status.get('rationale', '')}".strip(), kind="status", ) return _sse("orchestrator", _msg_payload(msg)) return None async def _refresh_alliance_groups( session: Session, actives: list[Participant], ) -> list[dict[str, Any]]: """Re-cluster after the consensus phase, treating the latest round of consensus statements as each participant's current stance.""" latest_by_id: dict[str, str] = {} for m in session.messages: if m.get("role") != "participant": continue if m.get("phase") != Phase.CONSENSUS.value: continue latest_by_id[m["speaker_id"]] = m["text"] # Fall back to finalization opinions for any participant who didn't # speak in the consensus phase yet. merged: dict[str, str] = dict(session.final_opinions) merged.update(latest_by_id) groups = await detect_alliances( orchestrator_model_id=_orchestrator_model_id(session), question=session.question, participants=actives, final_opinions=merged, api_log=session.api_log, ) _bump_orchestrator_count(session) return groups def _find_speaker_group( speaker: Participant, groups: list[dict[str, Any]], ) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]: speaker_group: dict[str, Any] | None = None others: list[dict[str, Any]] = [] for g in groups: if speaker.participant_id in (g.get("members") or []): speaker_group = g else: others.append(g) return speaker_group, others def _build_consensus_prompt( session: Session, speaker: Participant, speaker_group: dict[str, Any] | None, other_groups: list[dict[str, Any]], actives: list[Participant], cred_block: str, ) -> str: transcript = _format_history(session.messages) pending_block = _format_pending_block( _pending_addressed_for(session, speaker) ) # NOTE: The "speaker is whoever was addressed last" routing happens in # _phase_consensus, NOT here. This function only renders the prompt the # speaker receives. Whatever needs answering shows up in pending_block, # so a single unified allied/solo template handles both targeted and # broadcast turns - the prompt's "FIRST address open threads" rule # naturally focuses the speaker on whoever was just talking to them. if speaker_group and len(speaker_group.get("members") or []) > 1: members = ", ".join( p.name for p in actives if p.participant_id in (speaker_group.get("members") or []) and p.participant_id != speaker.participant_id ) or "(no co-allies named)" return CONSENSUS_ALLIED_PROMPT.format( alliance_members=members, alliance_stance=speaker_group.get("stance", "(unspecified)"), question=session.question, credential_summary=cred_block, transcript=transcript, pending_block=pending_block, ) other_groups_block = "\n".join( f" - \"{g.get('stance', '')}\" supported by " + ", ".join( p.name for p in actives if p.participant_id in (g.get("members") or []) ) for g in other_groups ) or "(no other groups)" return CONSENSUS_SOLO_PROMPT.format( your_stance=(speaker_group or {}).get("stance", "(unspecified)"), other_groups_block=other_groups_block, question=session.question, credential_summary=cred_block, transcript=transcript, pending_block=pending_block, ) # --------------------------------------------------------------------------- # Closure # --------------------------------------------------------------------------- async def _phase_closure(session: Session) -> AsyncIterator[str]: session.phase = Phase.CLOSURE for chunk in _orchestrator_banner_sse(session, "Phase 6: closure..."): yield chunk cred_block = credentials_to_block(session.credential_summary) transcript = await compact_transcript_for_orchestrator( session, orchestrator_model_id=orchestrator_fast_model_id(session), ) status = await assess_consensus_status( orchestrator_model_id=orchestrator_fast_model_id(session), question=session.question, transcript=transcript, alliance_groups=session.alliance_groups, api_log=session.api_log, ) _bump_orchestrator_count(session) actives = _active_participants(session) if status.get("status") == "majority": idx = status.get("majority_group_index") majority_group = None if isinstance(idx, int) and 0 <= idx < len(session.alliance_groups): majority_group = session.alliance_groups[idx] else: # Fallback: largest group wins if session.alliance_groups: majority_group = max( session.alliance_groups, key=lambda g: len(g.get("members") or []), ) if majority_group: members_names = [ p.name for p in actives if p.participant_id in (majority_group.get("members") or []) ] stance = majority_group.get("stance", "") prompt = MAJORITY_REPORT_PROMPT.format( question=session.question, credential_summary=cred_block, majority_members=", ".join(members_names), majority_stance=stance, transcript=transcript, ) raw, _ = await orchestrator_call( orchestrator_model_id=_orchestrator_model_id(session), user_prompt=prompt, label="majority_report", api_log=session.api_log, expect_json=False, max_tokens=900, temperature=0.3, ) _bump_orchestrator_count(session) session.final_report = { "kind": "majority", "text": raw, "majority_members": members_names, "majority_stance": stance, "alliance_groups": session.alliance_groups, } msg = _add_orchestrator_message( session, raw, kind="majority_report", extra={"majority_members": members_names, "majority_stance": stance}, ) yield _sse("orchestrator", _msg_payload(msg)) return # Not productive / no majority. We may surface an unaddressed # factor and re-run consensus up to `stall_recovery_attempts` times # before giving up and emitting the no-consensus report. if session.consensus_attempts < session.limits.stall_recovery_attempts: session.consensus_attempts += 1 factor = await find_unaddressed_factor( orchestrator_model_id=_orchestrator_model_id(session), question=session.question, credential_summary_block=cred_block, transcript=transcript, api_log=session.api_log, ) _bump_orchestrator_count(session) if factor and factor.get("factor"): announce = ( f"The discussion has stalled. The orchestrator surfaces a new " f"factor for the group to consider: {factor['factor']}" ) msg = _add_orchestrator_message( session, announce, kind="factor", extra={"expected_to_shift": factor.get("expected_to_shift") or []}, ) yield _sse("orchestrator", _msg_payload(msg)) # Re-run the consensus phase once more async for chunk in _phase_consensus(session): yield chunk async for chunk in _phase_closure(session): yield chunk return # Failed twice (or no factor surfaced) -> emit no-consensus report prompt = NO_CONSENSUS_REPORT_PROMPT.format( question=session.question, credential_summary=cred_block, alliance_block="\n".join( f" - \"{g.get('stance', '')}\": " + ", ".join( p.name for p in actives if p.participant_id in (g.get("members") or []) ) for g in session.alliance_groups ), transcript=transcript, ) raw, _ = await orchestrator_call( orchestrator_model_id=_orchestrator_model_id(session), user_prompt=prompt, label="no_consensus_report", api_log=session.api_log, expect_json=False, max_tokens=900, temperature=0.3, ) _bump_orchestrator_count(session) session.final_report = { "kind": "no_consensus", "text": raw, "alliance_groups": session.alliance_groups, } msg = _add_orchestrator_message(session, raw, kind="no_consensus_report") yield _sse("orchestrator", _msg_payload(msg)) # --------------------------------------------------------------------------- # Public driver # --------------------------------------------------------------------------- async def run_conversation(session: Session) -> AsyncIterator[str]: """Drive the full conversation, yielding SSE chunks. The flow is structure → decision: the chosen ConversationStructure runs its phases, then hands a DecisionInput to the chosen DecisionMethod which runs the decision phase(s). Both are resolved from `session.conversation_structure_id` / `session.decision_method_id` (defaults: collaborative + consensus, which preserves the original CCAI behavior). """ # Lazy import so the conversation package can import orchestrator # helpers without a circular module load. from app.services.conversation import get_structure, get_decision actives = _active_participants(session) if len(actives) < 2: yield _sse("error", { "message": "Need at least 2 active participants to start.", }) yield _sse("done", {}) return if len(actives) > session.max_participants: # Defense in depth - the API layer should have already enforced this. for extra in actives[session.max_participants:]: extra.enabled = False structure_cls = get_structure(session.conversation_structure_id) decision_cls = get_decision(session.decision_method_id) structure = structure_cls(session) try: async for chunk in structure.run(): yield chunk # Kick off contribution summaries in the background just before # the decision phase. The Table View blocks on this task only if # the user opens it before it finishes - usually it'll be done # by then, so the table loads instantly. _start_contribution_summary_task(session) decision_input = structure.build_decision_input() decision = decision_cls(session, decision_input) async for chunk in decision.run(): yield chunk except Exception as exc: LOG.exception("Conversation crashed: %s", exc) yield _sse("error", {"message": f"Internal error: {exc}"}) finally: session.finished = True session.phase = Phase.FINISHED # Drop the human-input slot (if any) so its asyncio.Event # doesn't outlive the session in the module-level registry. human_io.drop_session(session.session_id) yield _sse("system", {"text": "End of Chat", "phase": session.phase.value}) yield _sse("done", {}) def _start_contribution_summary_task(session: Session) -> None: """Schedule the contribution-summary build as a background task. Idempotent: if a task is already in flight (or completed) we don't start another one. Errors in the background task are swallowed and logged - the Table View endpoint will fall back to a synchronous build if needed. """ if session.contribution_summary_task is not None: return if any((session.contribution_summaries or {}).values()): return async def _runner() -> None: try: await _build_contribution_summaries(session) except Exception as exc: # noqa: BLE001 LOG.warning( "Background contribution_summaries failed for %s: %s", session.session_id, exc, ) try: session.contribution_summary_task = asyncio.create_task(_runner()) except RuntimeError: session.contribution_summary_task = None async def ensure_contribution_summaries(session: Session) -> None: """Block on contribution summaries for the Table View. Order of preference: 1. Cached - return immediately. 2. Background task in flight - await it. 3. Nothing started - build synchronously. """ if any((session.contribution_summaries or {}).values()): return task = session.contribution_summary_task if task is not None and not task.done(): try: await task except Exception as exc: # noqa: BLE001 LOG.warning("contribution_summary_task await failed: %s", exc) if any((session.contribution_summaries or {}).values()): return await _build_contribution_summaries(session) async def _build_contribution_summaries(session: Session) -> None: actives = _active_participants(session) roster = "\n".join( f"- id: {p.participant_id} | name: {p.name}" for p in actives ) transcript = _format_history(session.messages) prompt = CONTRIBUTION_SUMMARY_PROMPT.format( roster_block=roster, transcript=transcript, ) _raw, parsed = await orchestrator_call( orchestrator_model_id=_orchestrator_model_id(session), user_prompt=prompt, label="contribution_summaries", api_log=session.api_log, max_tokens=900, ) session.orchestrator_call_count += 1 if isinstance(parsed, dict) and isinstance(parsed.get("contributions"), list): for c in parsed["contributions"]: pid = c.get("participant_id") summary = (c.get("summary") or "").strip() if pid and summary: session.contribution_summaries[pid] = summary