perf(orchestrator): parallel phases, streaming, early table summaries

- Parallel participant turns in Phases 1/2/4 via orchestrator_speed.run_roster_ai_turns_parallel
- SSE token streaming on consensus + targeted follow-up turns (message_stream_start / message_delta)
- Parallel ballot collection in majority / ranked-choice / Robert's Rules votes
- Heuristic-first addressed_to classifier skips most LLM calls
- Compact rolling transcript for orchestrator judge prompts (cached per session)
- Fast orchestrator model (gemini-2.0-flash) for lightweight classifier calls
- Lazy contribution summaries kicked off after Phase 4 so Table View loads instantly
- Conditional Phase-3 credential refresh only when follow-ups are needed
- max_tokens 4x multiplier limited to thinking models

Co-authored-by: Cursor <cursoragent@cursor.com>

COMMIT_MSG.txt

@@ -0,0 +1,11 @@
+perf(orchestrator): parallel phases, streaming, early table summaries
+
+- Parallel participant turns in Phases 1/2/4 via orchestrator_speed.run_roster_ai_turns_parallel
+- SSE token streaming on consensus + targeted follow-up turns (message_stream_start / message_delta)
+- Parallel ballot collection in majority / ranked-choice / Robert's Rules votes
+- Heuristic-first addressed_to classifier skips most LLM calls
+- Compact rolling transcript for orchestrator judge prompts (cached per session)
+- Fast orchestrator model (gemini-2.0-flash) for lightweight classifier calls
+- Lazy contribution summaries kicked off after Phase 4 so Table View loads instantly
+- Conditional Phase-3 credential refresh only when follow-ups are needed
+- max_tokens 4x multiplier limited to thinking models

backend/app/api/chat.py

@@ -985,6 +985,13 @@ async def api_table_view(session_id: str):
     if not session:
         raise HTTPException(404, "Session not found")
 
+    from app.services.orchestrator import ensure_contribution_summaries
+
+    try:
+        await ensure_contribution_summaries(session)
+    except Exception as exc:
+        LOG.warning("Failed to build contribution summaries: %s", exc)
+
     rows = []
     for p in session.participants:
         first = (session.initial_opinions or {}).get(p.participant_id, "")

backend/app/clients/llm_router.py

@@ -45,11 +45,19 @@ async def chat_completion(
     temperature: float = 0.7,
     max_tokens: int = 1024,
     timeout: float | None = None,
+    on_text_delta: Any | None = None,
 ) -> dict[str, Any]:
     """Unified LLM call that routes Neon models through HANA and others through OpenAI-compat."""
     if resolved.get("is_neon"):
         return await _call_hana(resolved, messages, temperature, max_tokens)
 
+    # Streaming uses a single request path (no model racing).
+    if on_text_delta is not None:
+        return await _plain_openai(
+            resolved, messages, temperature, max_tokens, timeout,
+            on_text_delta=on_text_delta,
+        )
+
     from app.config import settings
     if settings.speed_priority:
         return await _racing_openai(resolved, messages, temperature, max_tokens, timeout)
@@ -63,6 +71,7 @@ async def _plain_openai(
     temperature: float,
     max_tokens: int,
     timeout: float | None,
+    on_text_delta: Any | None = None,
 ) -> dict[str, Any]:
     return await openai_chat_completion(
         base_url=resolved["base_url"],
@@ -72,6 +81,7 @@ async def _plain_openai(
         temperature=temperature,
         max_tokens=max_tokens,
         timeout=timeout,
+        on_text_delta=on_text_delta,
     )
 
 

backend/app/clients/openai_compat.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import logging
 import time
 from typing import Any
@@ -19,6 +20,11 @@ _MAX_COMPLETION_TOKEN_MODELS = {
 }
 _NO_TEMPERATURE_MODELS = {"o1", "o1-mini", "o1-preview", "o3", "o3-mini", "o4-mini"}
 
+# Models that emit long internal reasoning traces; allow a higher completion
+# cap. Flash/mini/instruct models use the requested max_tokens as-is.
+_THINKING_MULTIPLIER_MODEL_PREFIXES = _MAX_COMPLETION_TOKEN_MODELS
+_THINKING_NAME_HINTS = ("thinking", "reasoning", "-think", "/think")
+
 # Reserve a few tokens for chat-template framing the server tacks on (the
 # vLLM server and OpenAI both add a small overhead per request that our
 # input estimate doesn't account for).
@@ -68,6 +74,13 @@ def _estimate_input_tokens(messages: list[dict[str, str]]) -> int:
     return total
 
 
+def _model_wants_thinking_multiplier(model: str) -> bool:
+    mid = (model or "").lower()
+    if any(mid.startswith(prefix) for prefix in _THINKING_MULTIPLIER_MODEL_PREFIXES):
+        return True
+    return any(hint in mid for hint in _THINKING_NAME_HINTS)
+
+
 def _resolve_effective_max(
     model: str, requested: int, messages: list[dict[str, str]],
 ) -> tuple[int, int, int]:
@@ -86,7 +99,8 @@ def _resolve_effective_max(
     window = context_window_for(model)
     input_estimate = _estimate_input_tokens(messages)
     headroom = max(_MIN_OUTPUT_TOKENS, window - input_estimate - _INPUT_SAFETY_MARGIN)
-    effective_max = max(_MIN_OUTPUT_TOKENS, min(requested * 4, headroom))
+    multiplier = 4 if _model_wants_thinking_multiplier(model) else 1
+    effective_max = max(_MIN_OUTPUT_TOKENS, min(requested * multiplier, headroom))
     return effective_max, input_estimate, window
 
 
@@ -110,6 +124,7 @@ async def openai_chat_completion(
     temperature: float = 0.7,
     max_tokens: int = 1024,
     timeout: float | None = None,
+    on_text_delta: Any | None = None,
 ) -> dict[str, Any]:
     """Send a chat completion request to any OpenAI-compatible endpoint."""
     url = f"{base_url.rstrip('/')}/chat/completions"
@@ -123,11 +138,12 @@ async def openai_chat_completion(
     effective_max, input_estimate, window = _resolve_effective_max(
         model, max_tokens, messages,
     )
-    if effective_max < max_tokens * 4:
+    mult = 4 if _model_wants_thinking_multiplier(model) else 1
+    if effective_max < max_tokens * mult:
         LOG.info(
-            "Capped max_tokens for %s: requested %d (x4=%d), input ~=%d, "
+            "Capped max_tokens for %s: requested %d (x%d=%d), input ~=%d, "
             "window=%d, sending %d",
-            model, max_tokens, max_tokens * 4, input_estimate, window,
+            model, max_tokens, mult, max_tokens * mult, input_estimate, window,
             effective_max,
         )
     effective_timeout = max(timeout * 2, 120) if timeout else timeout
@@ -146,6 +162,59 @@ async def openai_chat_completion(
     req_timeout = httpx.Timeout(effective_timeout) if effective_timeout else None
     client = _get_client()
     t0 = time.time()
+
+    if on_text_delta is not None:
+        body["stream"] = True
+        try:
+            parts: list[str] = []
+            async with client.stream(
+                "POST", url, json=body, headers=headers, timeout=req_timeout,
+            ) as resp:
+                if resp.status_code >= 400:
+                    detail = (await resp.aread()).decode("utf-8", errors="replace")[:300]
+                    return {
+                        "response": f"[Error {resp.status_code}]: {detail}",
+                        "elapsed_seconds": round(time.time() - t0, 2),
+                        "model": model,
+                        "error": True,
+                        "error_kind": _classify_http_status(resp.status_code),
+                        "error_status": resp.status_code,
+                    }
+                async for line in resp.aiter_lines():
+                    if not line or not line.startswith("data:"):
+                        continue
+                    payload = line[5:].strip()
+                    if payload == "[DONE]":
+                        break
+                    try:
+                        chunk = json.loads(payload)
+                    except json.JSONDecodeError:
+                        continue
+                    choices = chunk.get("choices") or []
+                    if not choices:
+                        continue
+                    delta = choices[0].get("delta") or {}
+                    piece = delta.get("content") or ""
+                    if piece:
+                        parts.append(piece)
+                        on_text_delta(piece)
+            text = strip_thinking("".join(parts))
+            return {
+                "response": text.strip(),
+                "elapsed_seconds": round(time.time() - t0, 2),
+                "model": model,
+                "finish_reason": "stop",
+            }
+        except Exception as exc:
+            LOG.exception("OpenAI-compat stream failed: %s", exc)
+            return {
+                "response": f"[Error]: {exc}",
+                "elapsed_seconds": round(time.time() - t0, 2),
+                "model": model,
+                "error": True,
+                "error_kind": _classify_exception(exc),
+            }
+
     for attempt in range(2):
         try:
             resp = await client.post(url, json=body, headers=headers, timeout=req_timeout)

backend/app/config.py

@@ -37,6 +37,9 @@ class Settings(BaseSettings):
     mistral_api_key: str = ""
 
     orchestrator_model: str = "gpt-4o-mini"
+    # Lightweight model for addressed-to / status classifiers. Falls back
+    # to orchestrator_model when unset or unresolvable.
+    orchestrator_fast_model: str = "gemini-2.0-flash"
     speed_priority: bool = False
 
     cors_origins: str = "http://localhost:3000,http://localhost:3001,http://localhost:3002"

backend/app/services/consensus.py

@@ -7,6 +7,7 @@ All four are short JSON-shaped orchestrator calls layered on top of
 from __future__ import annotations
 
 import logging
+import re
 from typing import Any
 
 from app.services.json_calls import orchestrator_call
@@ -99,6 +100,64 @@ def _normalize_groups(
     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,
@@ -107,6 +166,10 @@ async def classify_addressed_to(
     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,

backend/app/services/conversation/decisions/majority.py

@@ -28,6 +28,7 @@ from typing import Any, AsyncIterator
 from app.services.conversation.decisions.base import DecisionMethod
 from app.services.conversation.voting import (
     cast_vote_single,
+    gather_votes_parallel,
     cast_vote_yesno,
     extract_candidate_options,
     tally_single_votes,
@@ -86,10 +87,13 @@ class MajorityRulesDecision(DecisionMethod):
         yield _sse("orchestrator", _msg_payload(msg))
 
         ballots: list[dict[str, Any]] = []
-        for p in voters:
-            result = await cast_vote_yesno(
-                session=session, participant=p, motion=motion,
-            )
+        for p, result in await gather_votes_parallel(
+            voters,
+            cast_vote_yesno,
+            session=session,
+            default_mode="yesno",
+            motion=motion,
+        ):
             ballot = {
                 "voter_id": p.participant_id,
                 "voter_name": p.name,
@@ -162,11 +166,14 @@ class MajorityRulesDecision(DecisionMethod):
         yield _sse("orchestrator", _msg_payload(msg))
 
         ballots: list[dict[str, Any]] = []
-        for p in voters:
-            result = await cast_vote_single(
-                session=session, participant=p, question=di.question,
-                options=options,
-            )
+        for p, result in await gather_votes_parallel(
+            voters,
+            cast_vote_single,
+            session=session,
+            default_mode="single",
+            question=di.question,
+            options=options,
+        ):
             ballot = {
                 "voter_id": p.participant_id,
                 "voter_name": p.name,

backend/app/services/conversation/decisions/ranked_choice.py

@@ -26,6 +26,7 @@ from app.services.conversation.decisions.base import DecisionMethod
 from app.services.conversation.voting import (
     cast_vote_ranking,
     extract_candidate_options,
+    gather_votes_parallel,
     run_irv,
 )
 from app.services.models import Phase
@@ -94,11 +95,14 @@ class RankedChoiceDecision(DecisionMethod):
 
         ballots: list[dict[str, Any]] = []
         rankings_only: list[list[int]] = []
-        for p in voters:
-            result = await cast_vote_ranking(
-                session=session, participant=p,
-                question=di.question, options=options,
-            )
+        for p, result in await gather_votes_parallel(
+            voters,
+            cast_vote_ranking,
+            session=session,
+            default_mode="rank",
+            question=di.question,
+            options=options,
+        ):
             ranking = result.get("ranking") or []
             ballots.append({
                 "voter_id": p.participant_id,

backend/app/services/conversation/decisions/roberts_rules_vote.py

@@ -19,6 +19,7 @@ from typing import Any, AsyncIterator
 from app.services.conversation.decisions.base import DecisionMethod
 from app.services.conversation.voting import (
     cast_vote_yesno,
+    gather_votes_parallel,
     tally_yesno_votes,
 )
 from app.services.json_calls import orchestrator_call
@@ -114,10 +115,13 @@ class RobertsRulesVote(DecisionMethod):
         yield _sse("orchestrator", _msg_payload(msg))
 
         ballots: list[dict[str, Any]] = []
-        for p in voters:
-            result = await cast_vote_yesno(
-                session=session, participant=p, motion=motion,
-            )
+        for p, result in await gather_votes_parallel(
+            voters,
+            cast_vote_yesno,
+            session=session,
+            default_mode="yesno",
+            motion=motion,
+        ):
             ballot = {
                 "voter_id": p.participant_id,
                 "voter_name": p.name,

backend/app/services/conversation/voting.py

@@ -19,10 +19,11 @@ method classes can compose them however they want.
 """
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 import time
-from typing import Any
+from typing import Any, Awaitable, Callable
 
 from app.clients.llm_router import chat_completion
 from app.services.json_calls import (
@@ -194,6 +195,36 @@ async def extract_candidate_options(
     return options
 
 
+async def gather_votes_parallel(
+    voters: list[Participant],
+    cast_fn: Callable[..., Awaitable[dict[str, Any]]],
+    *,
+    session: Session,
+    default_mode: str = "single",
+    **cast_kwargs: Any,
+) -> list[tuple[Participant, dict[str, Any]]]:
+    """Run ballot calls concurrently; roster order is preserved."""
+    if not voters:
+        return []
+
+    async def _one(p: Participant) -> tuple[Participant, dict[str, Any]]:
+        result = await cast_fn(session=session, participant=p, **cast_kwargs)
+        return p, result
+
+    gathered = await asyncio.gather(
+        *[_one(p) for p in voters],
+        return_exceptions=True,
+    )
+    out: list[tuple[Participant, dict[str, Any]]] = []
+    for p, item in zip(voters, gathered):
+        if isinstance(item, BaseException):
+            LOG.exception("Parallel vote failed for %s: %s", p.participant_id, item)
+            out.append((p, _vote_default(default_mode)))
+        else:
+            out.append(item)
+    return out
+
+
 async def cast_vote_single(
     *,
     session: Session,

backend/app/services/models.py

@@ -407,3 +407,15 @@ class Session:
     # on RR output.
     main_motion: str | None = None
     proposed_motions: list[dict[str, Any]] = field(default_factory=list)
+
+    # Rolling summary for orchestrator judge prompts when the transcript
+    # grows past the compact-transcript char budget (see orchestrator_speed).
+    orchestrator_context_summary: str = ""
+    orchestrator_context_through_idx: int = -1
+
+    # Background task that builds per-participant contribution summaries
+    # for the Table View. Kicked off by run_conversation just before the
+    # decision phase so that by the time the user opens Table View the
+    # work is already done. `Any` rather than `asyncio.Task` to avoid the
+    # import-time dependency on a running loop.
+    contribution_summary_task: Any = None

backend/app/services/orchestrator.py

@@ -26,6 +26,7 @@ import asyncio
 import json
 import logging
 import time
+import uuid
 from dataclasses import asdict
 from typing import Any, AsyncIterator
 
@@ -55,6 +56,13 @@ from app.services.credential import (
     refresh_credential_summary,
 )
 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_roster_ai_turns_parallel,
+)
 from app.services.resilience import run_resilient_turn
 from app.services.models import (
     DEFAULT_MAX_PARTICIPANTS,
@@ -421,7 +429,7 @@ async def _do_human_turn(
     addressed: str | None = None
     if classify_addressed:
         addressed = await classify_addressed_to(
-            orchestrator_model_id=_orchestrator_model_id(session),
+            orchestrator_model_id=orchestrator_fast_model_id(session),
             participants=actives,
             speaker_name=participant.name,
             message=text,
@@ -501,6 +509,8 @@ async def _call_participant(
     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.
 
@@ -563,6 +573,25 @@ async def _call_participant(
         "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,
@@ -570,6 +599,7 @@ async def _call_participant(
             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)
@@ -611,8 +641,10 @@ def _add_participant_message(
     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",
@@ -676,55 +708,42 @@ async def _phase_initial_opinions(session: Session) -> AsyncIterator[str]:
     yield _sse("status", {"message": "Phase 1: collecting independent first opinions..."})
 
     actives = _active_participants(session)
-    for p in actives:
+
+    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]:
+        session.initial_opinions[result.participant.participant_id] = result.turn.text
+        return {}
+
+    def _build_initial_spec(p: Participant) -> _AiTurnSpec | None:
         if p.kind == "human":
-            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
-            continue
-        # Phase 1 deliberately uses a *bare* prompt (no transcript) so each
-        # participant's first opinion is independent of the others.
-        prompt = INITIAL_OPINION_PROMPT.format(question=session.question)
-        turn = await run_resilient_turn(
-            session=session, participant=p,
-            user_prompt=prompt,
+            return None
+        return _AiTurnSpec(
+            participant=p,
+            user_prompt=INITIAL_OPINION_PROMPT.format(question=session.question),
             label="initial_opinion",
             max_tokens=700,
-            call_participant=_call_participant,
         )
-        for ev in turn.sse_events:
-            yield ev
-        if not turn.ok:
-            yield _sse("participant_error", {
-                "participant_id": p.participant_id,
-                "name": p.name,
-                "phase": session.phase.value,
-            })
-            if p.consecutive_failures >= session.limits.auto_disable_failures:
-                p.enabled = False
-                yield _sse("status", {
-                    "message": (
-                        f"{p.name} auto-disabled after "
-                        f"{session.limits.auto_disable_failures} failures."
-                    ),
-                })
-            continue
-        speaker = turn.speaker
-        msg = _add_participant_message(
-            session, speaker, turn.text, phase=session.phase, elapsed=turn.elapsed,
-        )
-        session.initial_opinions[speaker.participant_id] = turn.text
-        yield _sse("message", _msg_payload(msg))
 
-        if _participant_msg_cap_hit(session):
-            async for chunk in _wait_for_continue(session, "messages"):
-                yield chunk
+    async for chunk in run_roster_ai_turns_parallel(
+        session,
+        actives,
+        phase=session.phase,
+        build_spec=_build_initial_spec,
+        call_participant=_call_participant,
+        on_human_turn=_human_initial,
+        post_process=_post_initial,
+    ):
+        yield chunk
 
     yield _sse("status", {"message": "Building Credential Summary..."})
     creds = await build_credential_summary(
@@ -770,83 +789,72 @@ async def _phase_critique(session: Session, round_number: int) -> AsyncIterator[
     })
     cred_block = credentials_to_block(session.credential_summary)
     actives = _active_participants(session)
-    for p in actives:
+    # 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":
-            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
-            continue
-        transcript = _format_history(session.messages)
-        # Snapshot pending threads BEFORE the call so we can both render
-        # them in the prompt and stamp them onto the outgoing message as
-        # `replying_to` (frontend pill).
-        pending = _pending_addressed_for(session, p)
+            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,
+            transcript=transcript_snapshot,
             pending_block=pending_block,
         )
-        turn = await run_resilient_turn(
-            session=session, participant=p,
+        return _AiTurnSpec(
+            participant=p,
             user_prompt=prompt,
             label=f"critique_round_{round_number}",
             max_tokens=700,
-            call_participant=_call_participant,
         )
-        for ev in turn.sse_events:
-            yield ev
-        if not turn.ok:
-            yield _sse("participant_error", {
-                "participant_id": p.participant_id, "name": p.name,
-                "phase": session.phase.value,
-            })
-            if p.consecutive_failures >= session.limits.auto_disable_failures:
-                p.enabled = False
-                yield _sse("status", {
-                    "message": (
-                        f"{p.name} auto-disabled after "
-                        f"{session.limits.auto_disable_failures} failures."
-                    ),
-                })
-            continue
-        speaker = turn.speaker
-        text, elapsed = turn.text, turn.elapsed
 
-        # Detect addressed_to so the consensus phase's targeted-response
-        # logic can also reuse it - cheap classification call.
+    async def _post_critique(result: _AiTurnResult) -> dict[str, Any]:
+        speaker = result.turn.speaker
         addressed = await classify_addressed_to(
-            orchestrator_model_id=_orchestrator_model_id(session),
+            orchestrator_model_id=orchestrator_fast_model_id(session),
             participants=_active_participants(session),
             speaker_name=speaker.name,
-            message=text,
+            message=result.turn.text,
             api_log=session.api_log,
         )
         _bump_orchestrator_count(session)
+        return {
+            "addressed_to": addressed,
+            "replying_to": _replying_to_ids(result.pending),
+        }
 
-        msg = _add_participant_message(
-            session, speaker, text, phase=session.phase, elapsed=elapsed,
-            addressed_to=addressed,
-            replying_to=_replying_to_ids(pending),
-        )
-        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
+    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]:
@@ -854,33 +862,18 @@ async def _phase_status_assessment(session: Session) -> AsyncIterator[str]:
     yield _sse("status", {"message": "Phase 3: assessing whether more questions are needed..."})
 
     cred_block = credentials_to_block(session.credential_summary)
-
-    # Refresh Credential Summary once after Phase 2 critique - participants
-    # have revealed a lot more about themselves through critique.
-    transcript = _format_history(session.messages)
-    refreshed = await refresh_credential_summary(
-        orchestrator_model_id=_orchestrator_model_id(session),
-        question=session.question,
-        participants=_active_participants(session),
-        existing=session.credential_summary,
-        critique_transcript=transcript,
-        api_log=session.api_log,
-    )
-    _bump_orchestrator_count(session)
-    if refreshed != session.credential_summary:
-        session.credential_summary = refreshed
-        yield _sse("credentials_updated", {
-            "stage": "refreshed",
-            "credentials": session.credential_summary,
-        })
-    cred_block = credentials_to_block(session.credential_summary)
+    cred_refreshed = False
 
     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=_format_history(session.messages),
+            transcript=transcript,
         )
         _raw, parsed = await orchestrator_call(
             orchestrator_model_id=_orchestrator_model_id(session),
@@ -907,6 +900,27 @@ async def _phase_status_assessment(session: Session) -> AsyncIterator[str]:
             yield _sse("orchestrator", _msg_payload(msg))
             return
 
+        # Refresh credentials only when follow-ups are actually needed.
+        if not cred_refreshed:
+            cred_refreshed = True
+            full_transcript = _format_history(session.messages)
+            refreshed = await refresh_credential_summary(
+                orchestrator_model_id=_orchestrator_model_id(session),
+                question=session.question,
+                participants=_active_participants(session),
+                existing=session.credential_summary,
+                critique_transcript=full_transcript,
+                api_log=session.api_log,
+            )
+            _bump_orchestrator_count(session)
+            if refreshed != session.credential_summary:
+                session.credential_summary = refreshed
+                yield _sse("credentials_updated", {
+                    "stage": "refreshed",
+                    "credentials": session.credential_summary,
+                })
+            cred_block = credentials_to_block(session.credential_summary)
+
         # Otherwise run targeted follow-ups
         active_ids = {p.participant_id for p in _active_participants(session)}
         for oq in open_qs:
@@ -966,13 +980,19 @@ async def _phase_status_assessment(session: Session) -> AsyncIterator[str]:
                     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:
@@ -990,6 +1010,7 @@ async def _phase_status_assessment(session: Session) -> AsyncIterator[str]:
             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))
 
@@ -1014,54 +1035,56 @@ async def _phase_finalization(session: Session) -> AsyncIterator[str]:
 
     cred_block = credentials_to_block(session.credential_summary)
     actives = _active_participants(session)
-    for p in actives:
+    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":
-            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
-            continue
-        transcript = _format_history(session.messages)
-        pending = _pending_addressed_for(session, p)
+            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,
+            transcript=transcript_snapshot,
             pending_block=pending_block,
         )
-        turn = await run_resilient_turn(
-            session=session, participant=p,
+        return _AiTurnSpec(
+            participant=p,
             user_prompt=prompt,
             label="finalization",
             max_tokens=600,
-            call_participant=_call_participant,
-        )
-        for ev in turn.sse_events:
-            yield ev
-        if not turn.ok:
-            yield _sse("participant_error", {
-                "participant_id": p.participant_id, "name": p.name,
-                "phase": session.phase.value,
-            })
-            continue
-        speaker = turn.speaker
-        text, elapsed = turn.text, turn.elapsed
-        session.final_opinions[speaker.participant_id] = text
-        msg = _add_participant_message(
-            session, speaker, text, phase=session.phase, elapsed=elapsed,
-            replying_to=_replying_to_ids(pending),
         )
-        yield _sse("message", _msg_payload(msg))
 
-        if _participant_msg_cap_hit(session):
-            async for chunk in _wait_for_continue(session, "messages"):
-                yield chunk
+    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]:
@@ -1160,31 +1183,9 @@ async def _phase_consensus(session: Session) -> AsyncIterator[str]:
             # 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:
-                transcript = _format_history(session.messages)
-                status = await assess_consensus_status(
-                    orchestrator_model_id=_orchestrator_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",
-                    )
-                    yield _sse("orchestrator", _msg_payload(msg))
-                    return
-                if status.get("status") == "unproductive":
-                    msg = _add_orchestrator_message(
-                        session,
-                        f"Conversation no longer productive. {status.get('rationale', '')}".strip(),
-                        kind="status",
-                    )
-                    yield _sse("orchestrator", _msg_payload(msg))
+                terminal = await _consensus_status_terminal_sse(session, actives)
+                if terminal:
+                    yield terminal
                     return
             continue
 
@@ -1199,13 +1200,19 @@ async def _phase_consensus(session: Session) -> AsyncIterator[str]:
         # 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:
@@ -1221,7 +1228,7 @@ async def _phase_consensus(session: Session) -> AsyncIterator[str]:
         text, elapsed = turn.text, turn.elapsed
 
         addressed = await classify_addressed_to(
-            orchestrator_model_id=_orchestrator_model_id(session),
+            orchestrator_model_id=orchestrator_fast_model_id(session),
             participants=actives,
             speaker_name=speaker.name,
             message=text,
@@ -1234,6 +1241,7 @@ async def _phase_consensus(session: Session) -> AsyncIterator[str]:
             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))
 
@@ -1246,33 +1254,46 @@ async def _phase_consensus(session: Session) -> AsyncIterator[str]:
 
         # Status check every full round (every len(actives) turns)
         if consensus_turns % max(1, len(actives)) == 0:
-            transcript = _format_history(session.messages)
-            status = await assess_consensus_status(
-                orchestrator_model_id=_orchestrator_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",
-                )
-                yield _sse("orchestrator", _msg_payload(msg))
-                return
-            if status.get("status") == "unproductive":
-                msg = _add_orchestrator_message(
-                    session,
-                    f"Conversation no longer productive. {status.get('rationale', '')}".strip(),
-                    kind="status",
-                )
-                yield _sse("orchestrator", _msg_payload(msg))
+            terminal = await _consensus_status_terminal_sse(session, actives)
+            if terminal:
+                yield terminal
                 return
-            # else: productive - keep going
+
+
+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(
@@ -1376,10 +1397,13 @@ async def _phase_closure(session: Session) -> AsyncIterator[str]:
     yield _sse("status", {"message": "Phase 6: closure..."})
 
     cred_block = credentials_to_block(session.credential_summary)
-    transcript = _format_history(session.messages)
+    transcript = await compact_transcript_for_orchestrator(
+        session,
+        orchestrator_model_id=orchestrator_fast_model_id(session),
+    )
 
     status = await assess_consensus_status(
-        orchestrator_model_id=_orchestrator_model_id(session),
+        orchestrator_model_id=orchestrator_fast_model_id(session),
         question=session.question,
         transcript=transcript,
         alliance_groups=session.alliance_groups,
@@ -1538,6 +1562,12 @@ async def run_conversation(session: Session) -> AsyncIterator[str]:
         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():
@@ -1552,16 +1582,59 @@ async def run_conversation(session: Session) -> AsyncIterator[str]:
         # doesn't outlive the session in the module-level registry.
         human_io.drop_session(session.session_id)
 
-    # Build per-participant contribution summaries for the table view.
-    try:
-        await _build_contribution_summaries(session)
-    except Exception as exc:
-        LOG.warning("Failed to build contribution summaries: %s", exc)
-
     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(

backend/app/services/orchestrator_speed.py

@@ -0,0 +1,303 @@
+"""Speed-oriented orchestration helpers.
+
+Parallel participant turns, compact orchestrator context, and fast-
+model routing for lightweight classifier calls. Keeps the same
+visible message count while shortening wall-clock time.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+from dataclasses import dataclass, field
+from typing import Any, AsyncIterator, Awaitable, Callable, TYPE_CHECKING
+
+from app.services.resilience import ResilientTurnResult, run_resilient_turn
+
+if TYPE_CHECKING:
+    from app.services.models import Participant, Phase, Session
+
+LOG = logging.getLogger(__name__)
+
+# Rough char budget for orchestrator-side prompts that would otherwise
+# resend the entire transcript every few turns.
+ORCHESTRATOR_TRANSCRIPT_CHAR_BUDGET = 14_000
+_RECENT_TAIL_MESSAGES = 24
+
+CallParticipantFn = Callable[..., Awaitable[tuple[str, float, bool, str]]]
+
+
+@dataclass
+class _AiTurnSpec:
+    participant: "Participant"
+    user_prompt: str
+    label: str
+    max_tokens: int
+
+
+@dataclass
+class _AiTurnResult:
+    participant: "Participant"
+    turn: ResilientTurnResult
+    pending: list[tuple[str, str, str]] = field(default_factory=list)
+
+
+def orchestrator_fast_model_id(session: "Session") -> str:
+    """Model for lightweight orchestrator classifiers (addressed-to, status)."""
+    from app.config import settings
+    from app.services.orchestrator import _orchestrator_model_id
+
+    fast = (getattr(settings, "orchestrator_fast_model", None) or "").strip()
+    if fast and settings.resolve_model(fast):
+        return fast
+    return _orchestrator_model_id(session)
+
+
+def _format_history_for_orchestrator(
+    messages: list[dict[str, Any]],
+    *,
+    include_orchestrator: bool = True,
+) -> str:
+    from app.services.orchestrator import _format_history
+
+    return _format_history(messages, include_orchestrator=include_orchestrator)
+
+
+async def compact_transcript_for_orchestrator(
+    session: "Session",
+    *,
+    orchestrator_model_id: str,
+) -> str:
+    """Return a transcript block sized for orchestrator judge prompts.
+
+    Uses a rolling summary + recent tail when the full history exceeds
+    the char budget. Summaries are built lazily (one extra orchestrator
+    call) and cached on the session.
+    """
+    from app.services.json_calls import orchestrator_call
+    from app.services.orchestrator import _bump_orchestrator_count
+
+    messages = session.messages
+    full = _format_history_for_orchestrator(messages)
+    if len(full) <= ORCHESTRATOR_TRANSCRIPT_CHAR_BUDGET:
+        return full
+
+    tail = messages[-_RECENT_TAIL_MESSAGES:]
+    tail_text = _format_history_for_orchestrator(tail)
+    through = len(messages) - len(tail)
+    if (
+        session.orchestrator_context_summary
+        and session.orchestrator_context_through_idx >= through - 2
+    ):
+        return (
+            "[Earlier discussion summary]\n"
+            f"{session.orchestrator_context_summary}\n\n"
+            "[Recent messages]\n"
+            f"{tail_text}"
+        )
+
+    prompt = (
+        "Summarize the following group discussion for an orchestrator that "
+        "will judge consensus and open questions. Preserve names, stances, "
+        "and unresolved disagreements. Be concise (under 400 words).\n\n"
+        f"{full}"
+    )
+    raw, _ = await orchestrator_call(
+        orchestrator_model_id=orchestrator_model_id,
+        user_prompt=prompt,
+        label="orchestrator_transcript_summary",
+        api_log=session.api_log,
+        expect_json=False,
+        max_tokens=700,
+        temperature=0.2,
+    )
+    _bump_orchestrator_count(session)
+    summary = (raw or "").strip() or full[-ORCHESTRATOR_TRANSCRIPT_CHAR_BUDGET:]
+    session.orchestrator_context_summary = summary
+    session.orchestrator_context_through_idx = through
+    return (
+        "[Earlier discussion summary]\n"
+        f"{summary}\n\n"
+        "[Recent messages]\n"
+        f"{tail_text}"
+    )
+
+
+async def _execute_ai_turn(
+    session: "Session",
+    spec: _AiTurnSpec,
+    call_participant: CallParticipantFn,
+) -> _AiTurnResult:
+    from app.services.orchestrator import _pending_addressed_for
+
+    pending = _pending_addressed_for(session, spec.participant)
+    turn = await run_resilient_turn(
+        session=session,
+        participant=spec.participant,
+        user_prompt=spec.user_prompt,
+        label=spec.label,
+        max_tokens=spec.max_tokens,
+        call_participant=call_participant,
+    )
+    return _AiTurnResult(
+        participant=spec.participant,
+        turn=turn,
+        pending=pending,
+    )
+
+
+PostProcessFn = Callable[
+    [_AiTurnResult],
+    Awaitable[dict[str, Any] | None],
+]
+
+
+async def run_roster_ai_turns_parallel(
+    session: "Session",
+    actives: list["Participant"],
+    *,
+    phase: "Phase",
+    build_spec: Callable[["Participant"], _AiTurnSpec | None],
+    call_participant: CallParticipantFn,
+    on_human_turn: Callable[
+        ["Participant"],
+        AsyncIterator[str],
+    ],
+    post_process: PostProcessFn | None = None,
+) -> AsyncIterator[str]:
+    """Run participant turns: humans sequentially, AI in parallel batches.
+
+    Walks `actives` in roster order. Consecutive AI participants are
+    executed with ``asyncio.gather``; results are applied in roster
+    order so the message log stays deterministic. Humans are awaited
+    one at a time via ``on_human_turn``.
+
+    Yields orchestrator SSE strings (status, message, errors, etc.).
+    """
+    from app.services.orchestrator import (
+        _msg_payload,
+        _participant_msg_cap_hit,
+        _sse,
+        _wait_for_continue,
+    )
+
+    ai_batch: list[_AiTurnSpec] = []
+
+    async def flush_ai_batch() -> AsyncIterator[str]:
+        nonlocal ai_batch
+        if not ai_batch:
+            return
+        specs = ai_batch
+        ai_batch = []
+        results = await asyncio.gather(
+            *[
+                _execute_ai_turn(session, spec, call_participant)
+                for spec in specs
+            ],
+            return_exceptions=True,
+        )
+        for spec, item in zip(specs, results):
+            if isinstance(item, BaseException):
+                LOG.exception(
+                    "Parallel turn failed for %s: %s",
+                    spec.participant.participant_id,
+                    item,
+                )
+                yield _sse("participant_error", {
+                    "participant_id": spec.participant.participant_id,
+                    "name": spec.participant.name,
+                    "phase": phase.value,
+                })
+                continue
+            extra: dict[str, Any] | None = None
+            if post_process is not None:
+                extra = await post_process(item) or {}
+            async for chunk in _emit_ai_turn_result(
+                session, item, phase=phase, extra=extra,
+            ):
+                yield chunk
+            if _participant_msg_cap_hit(session):
+                async for chunk in _wait_for_continue(session, "messages"):
+                    yield chunk
+
+    for p in actives:
+        if p.kind == "human":
+            async for chunk in flush_ai_batch():
+                yield chunk
+            async for chunk in on_human_turn(p):
+                yield chunk
+            continue
+
+        spec = build_spec(p)
+        if spec is None:
+            continue
+        ai_batch.append(spec)
+
+    async for chunk in flush_ai_batch():
+        yield chunk
+
+
+async def _emit_ai_turn_result(
+    session: "Session",
+    result: _AiTurnResult,
+    *,
+    phase: "Phase",
+    extra: dict[str, Any] | None = None,
+) -> AsyncIterator[str]:
+    """Apply a completed AI turn to the session and yield SSE."""
+    from app.services.orchestrator import (
+        _add_participant_message,
+        _msg_payload,
+        _orchestrator_cap_hit,
+        _replying_to_ids,
+        _sse,
+        _wait_for_continue,
+    )
+
+    p = result.participant
+    turn = result.turn
+    for ev in turn.sse_events:
+        yield ev
+    if not turn.ok:
+        yield _sse("participant_error", {
+            "participant_id": p.participant_id,
+            "name": p.name,
+            "phase": phase.value,
+        })
+        if p.consecutive_failures >= session.limits.auto_disable_failures:
+            p.enabled = False
+            yield _sse("status", {
+                "message": (
+                    f"{p.name} auto-disabled after "
+                    f"{session.limits.auto_disable_failures} failures."
+                ),
+            })
+        return
+
+    speaker = turn.speaker
+    meta = extra or {}
+    msg = _add_participant_message(
+        session,
+        speaker,
+        turn.text,
+        phase=phase,
+        elapsed=turn.elapsed,
+        addressed_to=meta.get("addressed_to"),
+        replying_to=meta.get(
+            "replying_to",
+            _replying_to_ids(result.pending),
+        ),
+        message_id=meta.get("message_id"),
+    )
+    yield _sse("message", _msg_payload(msg))
+
+    if _orchestrator_cap_hit(session):
+        async for chunk in _wait_for_continue(session, "orchestrator"):
+            yield chunk
+
+
+async def run_parallel_coroutines(
+    coros: list[Awaitable[Any]],
+) -> list[Any]:
+    """Gather with exception isolation (failed tasks become exceptions)."""
+    return await asyncio.gather(*coros, return_exceptions=True)

backend/app/services/resilience.py

@@ -329,6 +329,8 @@ async def run_resilient_turn(
     label: str,
     max_tokens: int,
     call_participant: CallParticipantFn,
+    stream_events: list[str] | None = None,
+    stream_message_id: str | None = None,
 ) -> ResilientTurnResult:
     """Phase-aware wrapper around a single participant LLM turn.
 
@@ -356,6 +358,8 @@ async def run_resilient_turn(
         user_prompt=user_prompt,
         label=label,
         max_tokens=max_tokens,
+        stream_events=stream_events,
+        stream_message_id=stream_message_id,
     )
 
     if ok and text.strip():

frontend/src/App.js

@@ -753,9 +753,38 @@ export default function App() {
             setSessionParticipants(data.participants || []);
           },
           onMessage: (data) => {
-            setMessages(prev => [...prev, data]);
+            setMessages(prev => {
+              const mid = data?.message_id;
+              if (!mid) return [...prev, data];
+              const idx = prev.findIndex(m => m.message_id === mid);
+              if (idx >= 0) {
+                const next = [...prev];
+                next[idx] = { ...next[idx], ...data, streaming: false };
+                return next;
+              }
+              return [...prev, data];
+            });
             setStatusText('Conversation in progress...');
           },
+          onMessageStreamStart: (data) => {
+            setMessages(prev => [...prev, {
+              ...data,
+              role: 'participant',
+              text: '',
+              streaming: true,
+              timestamp: Date.now() / 1000,
+            }]);
+          },
+          onMessageDelta: (data) => {
+            const mid = data?.message_id;
+            const delta = data?.delta || '';
+            if (!mid || !delta) return;
+            setMessages(prev => prev.map(m => (
+              m.message_id === mid
+                ? { ...m, text: `${m.text || ''}${delta}` }
+                : m
+            )));
+          },
           onOrchestrator: (data) => {
             // Orchestrator events with kind == "status" but no text are
             // status banners; bubble them into a message-style entry so

frontend/src/utils/api.js

@@ -117,6 +117,8 @@ function eventHandlerKey(eventType) {
   switch (eventType) {
     case 'session': return 'onSession';
     case 'message': return 'onMessage';
+    case 'message_stream_start': return 'onMessageStreamStart';
+    case 'message_delta': return 'onMessageDelta';
     case 'orchestrator': return 'onOrchestrator';
     case 'system': return 'onSystem';
     case 'status': return 'onStatus';